[Store] Allow DFS-only replica placement

docs/source/api-reference/cpp/mooncake-store.md

@@ -64,7 +64,7 @@ struct ReplicateConfig {
 };
 ```
 
-`dfs_replica_num` may currently be `0` or `1`. When it is `1`, `replica_num >= 1` is required, so DFS-only placement is not supported. DFS replicas currently support only the `default` tenant and require the master and client DFS backends to be configured with the same shared root and shard layout. See the {ref}`DFS deployment documentation <dfs-storage>` for setup and lifecycle limitations. Native C++ clients must initialize a `DistributedStorageBackend` and attach it with `SetDfsStorageBackend()` before issuing DFS reads or writes. DFS descriptors are carried by each `PutStart`, `UpsertStart`, or query response; there is no client-side descriptor cache. The Python/RealClient setup path attaches the backend through `FileStorage`.
+`dfs_replica_num` may currently be `0` or `1`. DFS-only placement (`replica_num = 0`, `nof_replica_num = 0`, `dfs_replica_num = 1`) is supported; the DFS replica is then the object's only copy, so DFS eviction or removal deletes the object. DFS replicas currently support only the `default` tenant and require the master and client DFS backends to be configured with the same shared root and shard layout. See the {ref}`DFS deployment documentation <dfs-storage>` for setup and lifecycle limitations. Native C++ clients must initialize a `DistributedStorageBackend` and attach it with `SetDfsStorageBackend()` before issuing DFS reads or writes. DFS descriptors are carried by each `PutStart`, `UpsertStart`, or query response; there is no client-side descriptor cache. The Python/RealClient setup path attaches the backend through `FileStorage`.
 
 Soft pinning starts when the first replica becomes readable and has a fixed
 lifetime: reads do not extend it. `PRESERVE` keeps the committed deadline on an

docs/source/api-reference/python/mooncake-store.md

@@ -654,9 +654,10 @@ config.nof_replica_num = 1
 **Default:** `0`
 **Status:** **Work in progress; development and evaluation only.**
 **Description:** Requests an additional replica in the configured shared
-distributed filesystem. The supported values are currently `0` and `1`. When
-set to `1`, `replica_num` must be at least `1`, so DFS-only placement is not
-supported. DFS replicas currently support only the `default` tenant.
+distributed filesystem. The supported values are currently `0` and `1`.
+DFS-only placement (`replica_num = 0` with `dfs_replica_num = 1`) is supported;
+the DFS replica is then the object's only copy, so DFS eviction or removal
+deletes the object. DFS replicas currently support only the `default` tenant.
 
 ```python
 config = ReplicateConfig()

docs/source/deployment/mooncake-store-deployment-guide.md

@@ -1025,9 +1025,9 @@ config.dfs_replica_num = 1
 store.put("key", b"value", config)
 ```
 
-`dfs_replica_num` may currently be `0` or `1`. A DFS replica must be requested
-with at least one memory replica (`replica_num >= 1`), so DFS-only placement is
-not supported.
+`dfs_replica_num` may currently be `0` or `1`. DFS-only placement is supported
+by setting `replica_num = 0` and `nof_replica_num = 0`; the object then has no
+memory replica, and DFS eviction or removal deletes it.
 
 Allocation first tries the key's hash-selected DFS shard, then tries other
 ready shards if that shard has no suitable extent. This lets new shards accept
@@ -1049,7 +1049,9 @@ failed key is revoked without downgrading successful keys.
 For a same-size `Upsert`, if either the existing object or the new request has
 a DFS replica, the requested memory, NoF, and DFS replica counts must match the
 existing topology. A different-size update releases the old placement and
-allocates a new topology.
+allocates a new topology. For example, after the memory replica of a
+memory-plus-DFS object is evicted, a same-size `Upsert` must request the
+DFS-only topology.
 
 On reads, the master returns the readable replica list through the normal query
 path, and the client selects the first complete replica. If it selects DFS, any
@@ -1059,8 +1061,7 @@ reads for that descriptor.
 #### Current limitations
 
 - Only the `default` tenant is supported.
-- `dfs_replica_num` must be `0` or `1`, and `replica_num >= 1` is required when
-  it is enabled.
+- `dfs_replica_num` must be `0` or `1`.
 - C and Rust clients cannot currently request or access descriptor-based DFS:
   their replication configuration does not expose `dfs_replica_num`, and their
   setup API cannot initialize the distributed `FileStorage` backend. Use the

mooncake-store/src/master_service.cpp

@@ -5522,11 +5522,10 @@ auto MasterService::PutStart(const UUID& client_id, const std::string& key,
                    << ", key_size=" << key.size() << ", error=invalid_params";
         return tl::make_unexpected(ErrorCode::INVALID_PARAMS);
     }
-    if (config.dfs_replica_num > 1 ||
-        (config.dfs_replica_num > 0 && config.replica_num == 0)) {
-        LOG(ERROR) << "key=" << key << ", replica_num=" << config.replica_num
+    if (config.dfs_replica_num > 1) {
+        LOG(ERROR) << "key=" << key
                    << ", dfs_replica_num=" << config.dfs_replica_num
-                   << ", error=invalid_dfs_replica_config";
+                   << ", error=dfs_replica_num_must_be_0_or_1";
         return tl::make_unexpected(ErrorCode::INVALID_PARAMS);
     }
     if (config.dfs_replica_num > 0 && !object_id.tenant_id.IsDefault()) {
@@ -6204,11 +6203,10 @@ auto MasterService::UpsertStart(const UUID& client_id, const std::string& key,
                    << ", key_size=" << key.size() << ", error=invalid_params";
         return tl::make_unexpected(ErrorCode::INVALID_PARAMS);
     }
-    if (config.dfs_replica_num > 1 ||
-        (config.dfs_replica_num > 0 && config.replica_num == 0)) {
-        LOG(ERROR) << "key=" << key << ", replica_num=" << config.replica_num
+    if (config.dfs_replica_num > 1) {
+        LOG(ERROR) << "key=" << key
                    << ", dfs_replica_num=" << config.dfs_replica_num
-                   << ", error=invalid_dfs_replica_config";
+                   << ", error=dfs_replica_num_must_be_0_or_1";
         return tl::make_unexpected(ErrorCode::INVALID_PARAMS);
     }
     if (config.dfs_replica_num > 0 && !object_id.tenant_id.IsDefault()) {

mooncake-store/tests/dfs_immutable_bucket_client_test.cpp

@@ -242,6 +242,60 @@ TEST_F(DfsImmutableBucketClientTest, ExhaustionEvictsWholeBucketAndRetries) {
     EXPECT_EQ(third_query->replicas.size(), 2u);
 }
 
+TEST_F(DfsImmutableBucketClientTest, DfsOnlyObjectsLiveAndDieWithTheirBucket) {
+    ReplicateConfig dfs_only;
+    dfs_only.replica_num = 0;
+    dfs_only.dfs_replica_num = 1;
+    auto expect_value = [this](const std::string& key,
+                               const std::string& expected) {
+        auto query = writer_->Query(key);
+        ASSERT_TRUE(query) << key;
+        ASSERT_EQ(query->replicas.size(), 1u) << key;
+        ASSERT_TRUE(query->replicas[0].is_dfs_replica()) << key;
+        // The fixture's zero lease TTL makes Client::Get report LEASE_EXPIRED,
+        // so read the descriptor through the backend like the other tests.
+        std::string output(expected.size(), '\0');
+        auto reads =
+            backend_->BatchRead({{key,
+                                  query->replicas[0].get_dfs_descriptor(),
+                                  {{output.data(), output.size()}}}});
+        ASSERT_TRUE(reads[0]) << key;
+        EXPECT_EQ(output, expected) << key;
+    };
+
+    std::string first(4096, '1');
+    std::string second(4096, '2');
+    std::string third(4096, '3');
+    auto first_slices = Slices(first);
+    auto second_slices = Slices(second);
+    auto third_slices = Slices(third);
+    ASSERT_TRUE(writer_->Put("first", first_slices, dfs_only));
+    ASSERT_TRUE(writer_->Put("second", second_slices, dfs_only));
+    // Both buckets are full: the third write forces a whole-bucket eviction.
+    // With no memory replica, the evicted object disappears entirely.
+    ASSERT_TRUE(writer_->Put("third", third_slices, dfs_only));
+    auto first_query = writer_->Query("first");
+    ASSERT_FALSE(first_query);
+    EXPECT_EQ(first_query.error(), ErrorCode::OBJECT_NOT_FOUND);
+    expect_value("third", third);
+
+    adapter_->FailNextWrite();
+    std::string failed(4096, 'F');
+    auto failed_slices = Slices(failed);
+    auto failed_put = writer_->Put("failed", failed_slices, dfs_only);
+    ASSERT_FALSE(failed_put);
+    EXPECT_EQ(failed_put.error(), ErrorCode::FILE_WRITE_FAIL);
+    auto missing = writer_->Query("failed");
+    ASSERT_FALSE(missing);
+    EXPECT_EQ(missing.error(), ErrorCode::OBJECT_NOT_FOUND);
+
+    // Same-size upsert reallocates into a new bucket.
+    std::string replacement(4096, 'R');
+    auto replacement_slices = Slices(replacement);
+    ASSERT_TRUE(writer_->Upsert("third", replacement_slices, dfs_only));
+    expect_value("third", replacement);
+}
+
 TEST_F(DfsImmutableBucketLruClientTest, BatchQueryRefreshesDfsBucketLru) {
     std::string first(4096, '1');
     std::string second(4096, '2');

mooncake-store/tests/dfs_sync_client_test.cpp

@@ -127,6 +127,34 @@ class DfsSyncClientTest : public ::testing::Test {
         return config;
     }
 
+    static ReplicateConfig DfsOnlyConfig() {
+        ReplicateConfig config;
+        config.replica_num = 0;
+        config.dfs_replica_num = 1;
+        return config;
+    }
+
+    // Reads through Query + Client::Get and checks that DFS is the only
+    // replica.
+    void ExpectDfsOnlyValue(const std::string& key,
+                            const std::string& expected) {
+        auto query = writer_->Query(key);
+        ASSERT_TRUE(query.has_value()) << key;
+        last_read_lease_deadline_ = query->lease_timeout;
+        ASSERT_EQ(query->replicas.size(), 1) << key;
+        ASSERT_TRUE(query->replicas[0].is_dfs_replica()) << key;
+        std::string output(expected.size(), '\0');
+        std::vector<Slice> slices{{output.data(), output.size()}};
+        ASSERT_TRUE(writer_->Get(key, *query, slices).has_value()) << key;
+        EXPECT_EQ(output, expected) << key;
+    }
+
+    void ExpectNotFound(const std::string& key) {
+        auto query = writer_->Query(key);
+        ASSERT_FALSE(query.has_value()) << key;
+        EXPECT_EQ(query.error(), ErrorCode::OBJECT_NOT_FOUND) << key;
+    }
+
     tl::expected<QueryResult, ErrorCode> QueryDfsOnly(const std::string& key) {
         auto query = writer_->Query(key);
         if (!query) {
@@ -564,4 +592,98 @@ TEST_F(DfsSyncClientTest, DfsMetricsAppearInSerializedOutput) {
     EXPECT_NE(summary->find("=== DFS Metrics Summary ==="), std::string::npos);
 }
 
+TEST_F(DfsSyncClientTest, DfsOnlyPutGetRemoveAndReput) {
+    const std::string key = "dfs_only_put";
+    std::string value(4096, 'a');
+    std::vector<Slice> slices{{value.data(), value.size()}};
+    ASSERT_TRUE(writer_->Put(key, slices, DfsOnlyConfig()).has_value());
+    ExpectDfsOnlyValue(key, value);
+
+    WaitForReadLeaseExpiry();
+    ASSERT_TRUE(writer_->Remove(key).has_value());
+    ExpectNotFound(key);
+
+    std::string larger(8192, 'b');
+    std::vector<Slice> larger_slices{{larger.data(), larger.size()}};
+    ASSERT_TRUE(writer_->Put(key, larger_slices, DfsOnlyConfig()).has_value());
+    ExpectDfsOnlyValue(key, larger);
+}
+
+TEST_F(DfsSyncClientTest, DfsOnlyBatchFailureOnlyRevokesFailedKey) {
+    std::vector<std::string> keys{"dfs_only_ok", "dfs_only_fail"};
+    std::vector<std::string> values{std::string(4096, 'c'),
+                                    std::string(4096, 'd')};
+    auto slices = MakeSlices(values);
+    adapter_->FailWriteCall(adapter_->WriteCalls() + 2);
+
+    auto results = writer_->BatchPut(keys, slices, DfsOnlyConfig());
+    ASSERT_EQ(results.size(), keys.size());
+    EXPECT_TRUE(results[0].has_value());
+    ASSERT_FALSE(results[1].has_value());
+    EXPECT_EQ(results[1].error(), ErrorCode::FILE_WRITE_FAIL);
+    ExpectDfsOnlyValue(keys[0], values[0]);
+    ExpectNotFound(keys[1]);
+
+    // The revoked key and its extent can be reused.
+    std::vector<Slice> retry{{values[1].data(), values[1].size()}};
+    ASSERT_TRUE(writer_->Put(keys[1], retry, DfsOnlyConfig()).has_value());
+    ExpectDfsOnlyValue(keys[1], values[1]);
+}
+
+TEST_F(DfsSyncClientTest, DfsOnlyUpsertSameSizeResizeAndBatch) {
+    const std::string key = "dfs_only_upsert";
+    std::string initial(4096, 'e');
+    std::vector<Slice> initial_slices{{initial.data(), initial.size()}};
+    ASSERT_TRUE(
+        writer_->Upsert(key, initial_slices, DfsOnlyConfig()).has_value());
+    ExpectDfsOnlyValue(key, initial);
+
+    WaitForReadLeaseExpiry();
+    std::string same_size(4096, 'f');
+    std::vector<Slice> same_slices{{same_size.data(), same_size.size()}};
+    ASSERT_TRUE(writer_->Upsert(key, same_slices, DfsOnlyConfig()).has_value());
+    ExpectDfsOnlyValue(key, same_size);
+
+    WaitForReadLeaseExpiry();
+    std::string resized(12288, 'g');
+    std::vector<Slice> resized_slices{{resized.data(), resized.size()}};
+    ASSERT_TRUE(
+        writer_->Upsert(key, resized_slices, DfsOnlyConfig()).has_value());
+    ExpectDfsOnlyValue(key, resized);
+
+    WaitForReadLeaseExpiry();
+    std::vector<std::string> keys{key, "dfs_only_upsert_new"};
+    std::vector<std::string> values{std::string(4096, 'h'),
+                                    std::string(4096, 'i')};
+    auto batch_slices = MakeSlices(values);
+    auto results = writer_->BatchUpsert(keys, batch_slices, DfsOnlyConfig());
+    ASSERT_EQ(results.size(), keys.size());
+    ASSERT_TRUE(results[0].has_value());
+    ASSERT_TRUE(results[1].has_value());
+    ExpectDfsOnlyValue(keys[0], values[0]);
+    ExpectDfsOnlyValue(keys[1], values[1]);
+}
+
+TEST_F(DfsSyncClientTest, DfsOnlyUpsertMatchesMemoryLostTopology) {
+    const std::string key = "dfs_only_memory_lost";
+    std::string initial(4096, 'j');
+    std::vector<Slice> initial_slices{{initial.data(), initial.size()}};
+    ASSERT_TRUE(writer_->Put(key, initial_slices, DfsConfig()).has_value());
+    ASSERT_TRUE(provider_->UnmountSegment(segment_, segment_size_).has_value());
+    ExpectDfsOnlyValue(key, initial);
+
+    WaitForReadLeaseExpiry();
+    std::string updated(4096, 'k');
+    std::vector<Slice> updated_slices{{updated.data(), updated.size()}};
+    auto mismatched = writer_->Upsert(key, updated_slices, DfsConfig());
+    ASSERT_FALSE(mismatched.has_value());
+    EXPECT_EQ(mismatched.error(), ErrorCode::INVALID_PARAMS);
+    ExpectDfsOnlyValue(key, initial);
+
+    WaitForReadLeaseExpiry();
+    ASSERT_TRUE(
+        writer_->Upsert(key, updated_slices, DfsOnlyConfig()).has_value());
+    ExpectDfsOnlyValue(key, updated);
+}
+
 }  // namespace mooncake::test