ssd remove problem fix

mooncake-store/include/client_service.h

@@ -463,6 +463,9 @@ class Client {
     [[nodiscard]] tl::expected<std::vector<RemoveTaskItem>, ErrorCode>
     RemoveObjectHeartbeat(const UUID& client_id);
 
+    tl::expected<void, ErrorCode> AckRemoveObjectHeartbeat(
+        const UUID& client_id, const std::vector<RemoveTaskItem>& tasks);
+
     /**
      * @brief Stage a PROCESSING MEMORY replica for an existing key during
      * L2->L1 promotion. Returns the new replica's descriptor that the caller

mooncake-store/include/file_storage.h

@@ -54,8 +54,9 @@ class FileStorage {
     // Forward explicit-delete tombstone to the storage backend.
     // For BucketStorageBackend: marks tombstone + enables GC.
     // For other backends: no-op (default in StorageBackendInterface).
-    void MarkRemoved(const std::string& key);
-    void BatchMarkRemoved(const std::vector<std::string>& keys);
+    tl::expected<void, ErrorCode> MarkRemoved(const std::string& key);
+    tl::expected<void, ErrorCode> BatchMarkRemoved(
+        const std::vector<std::string>& keys);
 
    private:
     friend class FileStorageTest;

mooncake-store/include/master_client.h

@@ -493,14 +493,11 @@ class MasterClient {
     [[nodiscard]] tl::expected<std::vector<PromotionTaskItem>, ErrorCode>
     PromotionObjectHeartbeat(const UUID& client_id);
 
-    /**
-     * @brief Drain the removed_keys queue from master. Returns {tenant_id,
-     * key} pairs that were removed via Remove/BatchRemove and had LOCAL_DISK
-     * replicas on this client. The caller should MarkRemoved each key to
-     * trigger SSD tombstone + GC compaction.
-     */
+    /** Fetch pending remove tasks without removing them from the queue. */
     [[nodiscard]] tl::expected<std::vector<RemoveTaskItem>, ErrorCode>
     RemoveObjectHeartbeat(const UUID& client_id);
+    tl::expected<void, ErrorCode> AckRemoveObjectHeartbeat(
+        const UUID& client_id, const std::vector<RemoveTaskItem>& tasks);
 
     /**
      * @brief Stage a PROCESSING MEMORY replica for an existing key during

mooncake-store/include/master_service.h

@@ -727,16 +727,12 @@ class MasterService {
     auto PromotionObjectHeartbeat(const UUID& client_id)
         -> tl::expected<std::vector<PromotionTaskItem>, ErrorCode>;
 
-    /**
-     * @brief Drain the removed_keys queue for a client.
-     *
-     * Returns the list of {tenant_id, key} pairs that were removed via
-     * Remove/BatchRemove and had LOCAL_DISK replicas on this client. The
-     * client should call MarkRemoved on each key to trigger SSD tombstone
-     * marking + GC compaction.
-     */
+    /** Fetch pending remove tasks without removing them from the queue. */
     auto RemoveObjectHeartbeat(const UUID& client_id)
         -> tl::expected<std::vector<RemoveTaskItem>, ErrorCode>;
+    auto AckRemoveObjectHeartbeat(
+        const UUID& client_id, const std::vector<RemoveTaskItem>& tasks)
+        -> tl::expected<void, ErrorCode>;
 
     /**
      * @brief Stage a PROCESSING MEMORY replica for an existing key. Allocates
@@ -1544,6 +1540,8 @@ class MasterService {
     void FinalizeRemovedReplicasAfterDurable(
         const OpLogEntry& durable_entry,
         const std::vector<ReplicaID>& replica_ids, QuotaEraseMode quota_mode);
+    void EnqueueRemoveTasks(const std::vector<UUID>& holder_ids,
+                            const RemoveTaskItem& task);
     void FinalizeMetadataEraseAfterDurable(const OpLogEntry& durable_entry,
                                            QuotaEraseMode quota_mode);
     void FinalizeExpiredProcessingReplicasAfterDurable(

mooncake-store/include/rpc_service.h

@@ -219,13 +219,10 @@ class WrappedMasterService {
 
     tl::expected<bool, ErrorCode> PollRemoveAll(const UUID& client_id);
 
-    tl::expected<std::vector<RemoveTaskItem>, ErrorCode>
-    RemoveObjectHeartbeat(const UUID& client_id);
-
-    tl::expected<std::vector<RemoveTaskItem>, ErrorCode>
-    RemoveObjectHeartbeat(const UUID& client_id);
-
-    tl::expected<bool, ErrorCode> PollRemoveAll(const UUID& client_id);
+    tl::expected<std::vector<RemoveTaskItem>, ErrorCode> RemoveObjectHeartbeat(
+        const UUID& client_id);
+    tl::expected<void, ErrorCode> AckRemoveObjectHeartbeat(
+        const UUID& client_id, const std::vector<RemoveTaskItem>& tasks);
     tl::expected<void, ErrorCode> ReportSsdCapacity(
         const UUID& client_id, int64_t ssd_total_capacity_bytes);
 

mooncake-store/include/storage_backend.h

@@ -42,6 +42,10 @@ struct BucketMetadata {
     std::vector<std::string> keys;
     std::vector<BucketObjectMetadata> metadatas;
 
+    // Persisted tombstones. Init filters these keys before rebuilding the
+    // in-memory object index.
+    std::vector<std::string> tombstones;
+
     // Runtime-only fields (not serialized) for safe deletion support
     // Tracks number of in-flight reads to enable safe bucket deletion
     mutable std::atomic<int32_t> inflight_reads_{0};
@@ -55,6 +59,10 @@ struct BucketMetadata {
     // Runtime-only (not serialized): true while a GC compaction is in flight
     // for this bucket, preventing re-entrant compaction.
     mutable std::atomic<bool> compacting_{false};
+    // Runtime-only version of the bucket's deletion state. Compaction captures
+    // this value with its read snapshot and validates it before publishing a
+    // new bucket, so a concurrent deletion cannot publish stale data.
+    uint64_t generation_{0};
 
     // Default constructor
     BucketMetadata() = default;
@@ -65,21 +73,25 @@ struct BucketMetadata {
           data_size(other.data_size),
           keys(other.keys),
           metadatas(other.metadatas),
+          tombstones(other.tombstones),
           inflight_reads_(0),
           last_access_ns_(0),
           deleted_bytes_(0),
-          compacting_(false) {}
+          compacting_(false),
+          generation_(0) {}
 
     // Move constructor
     BucketMetadata(BucketMetadata&& other) noexcept
         : meta_size(other.meta_size),
           data_size(other.data_size),
           keys(std::move(other.keys)),
           metadatas(std::move(other.metadatas)),
+          tombstones(std::move(other.tombstones)),
           inflight_reads_(0),
           last_access_ns_(0),
           deleted_bytes_(0),
-          compacting_(false) {}
+          compacting_(false),
+          generation_(0) {}
 
     // Copy assignment
     BucketMetadata& operator=(const BucketMetadata& other) {
@@ -88,6 +100,7 @@ struct BucketMetadata {
             data_size = other.data_size;
             keys = other.keys;
             metadatas = other.metadatas;
+            tombstones = other.tombstones;
             // Don't copy runtime state
         }
         return *this;
@@ -100,12 +113,13 @@ struct BucketMetadata {
             data_size = other.data_size;
             keys = std::move(other.keys);
             metadatas = std::move(other.metadatas);
+            tombstones = std::move(other.tombstones);
             // Don't move runtime state
         }
         return *this;
     }
 };
-YLT_REFL(BucketMetadata, data_size, keys, metadatas);
+YLT_REFL(BucketMetadata, data_size, keys, metadatas, tombstones);
 
 /**
  * @brief RAII guard for tracking in-flight bucket reads.
@@ -445,11 +459,16 @@ class StorageBackendInterface {
     // Default no-op: only BucketStorageBackend implements tombstone + GC.
     // File-per-key and other backends inherit the no-op (do not delete files).
     // Safe to call for keys not present in local storage (idempotent).
-    virtual void MarkRemoved(const std::string& /* key */) {}
+    virtual tl::expected<void, ErrorCode> MarkRemoved(
+        const std::string& /* key */) {
+        return {};
+    }
 
     // Batch variant: mark multiple keys as removed in one lock acquisition.
-    virtual void BatchMarkRemoved(
-        const std::vector<std::string>& /* keys */) {}
+    virtual tl::expected<void, ErrorCode> BatchMarkRemoved(
+        const std::vector<std::string>& /* keys */) {
+        return {};
+    }
     // Remove all persisted objects from disk. Called during RemoveAll to
     // clean up physical SSD files alongside master metadata deletion.
     virtual void RemoveAll() {}
@@ -1047,8 +1066,10 @@ class BucketStorageBackend : public StorageBackendInterface {
     // Removes key from object_bucket_map_ (immediately invisible to
     // BatchLoad/IsExist) and bumps bucket deleted_bytes_.
     // Idempotent: no-op if key not in local storage.
-    void MarkRemoved(const std::string& key) override;
-    void BatchMarkRemoved(const std::vector<std::string>& keys) override;
+    tl::expected<void, ErrorCode> MarkRemoved(
+        const std::string& key) override;
+    tl::expected<void, ErrorCode> BatchMarkRemoved(
+        const std::vector<std::string>& keys) override;
 
     // Compact a single bucket: copy-on-write live keys to a new bucket,
     // atomically swap mappings, delete old bucket file after reads drain.
@@ -1073,13 +1094,6 @@ class BucketStorageBackend : public StorageBackendInterface {
 
    private:
     // --- Background GC ---
-    // Select the best GC candidate bucket: deleted_bytes_>0, not compacting,
-    // highest deleted_ratio, then coldest last_access_ns_.
-    // Must be called with mutex_ held (exclusive).
-    // Returns buckets_.end() if no candidate.
-    std::map<int64_t, std::shared_ptr<BucketMetadata>>::iterator
-    SelectGCCandidate();
-
     // Background GC thread entry point.
     void GCThreadFunc();
 

mooncake-store/src/client_service.cpp

@@ -3510,6 +3510,11 @@ Client::RemoveObjectHeartbeat(const UUID& client_id) {
     return master_client_.RemoveObjectHeartbeat(client_id);
 }
 
+tl::expected<void, ErrorCode> Client::AckRemoveObjectHeartbeat(
+    const UUID& client_id, const std::vector<RemoveTaskItem>& tasks) {
+    return master_client_.AckRemoveObjectHeartbeat(client_id, tasks);
+}
+
 tl::expected<PromotionAllocStartResponse, ErrorCode>
 Client::PromotionAllocStart(
     const std::string& key, uint64_t size,

mooncake-store/src/file_storage.cpp

@@ -815,12 +815,14 @@ tl::expected<bool, ErrorCode> FileStorage::IsEnableOffloading() {
     return enable_offloading;
 }
 
-void FileStorage::MarkRemoved(const std::string& key) {
-    storage_backend_->MarkRemoved(key);
+tl::expected<void, ErrorCode> FileStorage::MarkRemoved(
+    const std::string& key) {
+    return storage_backend_->MarkRemoved(key);
 }
 
-void FileStorage::BatchMarkRemoved(const std::vector<std::string>& keys) {
-    storage_backend_->BatchMarkRemoved(keys);
+tl::expected<void, ErrorCode> FileStorage::BatchMarkRemoved(
+    const std::vector<std::string>& keys) {
+    return storage_backend_->BatchMarkRemoved(keys);
 }
 
 tl::expected<void, ErrorCode> FileStorage::Heartbeat() {
@@ -857,13 +859,26 @@ tl::expected<void, ErrorCode> FileStorage::Heartbeat() {
         auto remove_result =
             client_->RemoveObjectHeartbeat(client_->getClientId());
         if (remove_result) {
+            bool all_marked = true;
             for (const auto& item : remove_result.value()) {
                 auto storage_key =
-                    MakeTenantScopedStorageKey(item.tenant_id, item.key);
-                storage_backend_->MarkRemoved(storage_key);
+                    TenantId(item.tenant_id).MakeScopedKey(item.key);
+                auto mark_result = storage_backend_->MarkRemoved(storage_key);
+                if (!mark_result) {
+                    all_marked = false;
+                    LOG(ERROR) << "Failed to persist remove tombstone: "
+                               << mark_result.error();
+                    break;
+                }
             }
-            if (!remove_result.value().empty()) {
-                VLOG(1) << "RemoveObjectHeartbeat drained "
+            if (all_marked && !remove_result.value().empty()) {
+                auto ack_result = client_->AckRemoveObjectHeartbeat(
+                    client_->getClientId(), remove_result.value());
+                if (!ack_result) {
+                    LOG(ERROR) << "Failed to ACK remove tasks: "
+                               << ack_result.error();
+                }
+                VLOG(1) << "RemoveObjectHeartbeat processed "
                         << remove_result.value().size()
                         << " removed key(s) from master";
             }

mooncake-store/src/master_client.cpp

@@ -254,6 +254,11 @@ struct RpcNameTraits<&WrappedMasterService::RemoveObjectHeartbeat> {
     static constexpr const char* value = "RemoveObjectHeartbeat";
 };
 
+template <>
+struct RpcNameTraits<&WrappedMasterService::AckRemoveObjectHeartbeat> {
+    static constexpr const char* value = "AckRemoveObjectHeartbeat";
+};
+
 template <>
 struct RpcNameTraits<&WrappedMasterService::PromotionAllocStart> {
     static constexpr const char* value = "PromotionAllocStart";
@@ -1208,6 +1213,15 @@ MasterClient::RemoveObjectHeartbeat(const UUID& client_id) {
                       std::vector<RemoveTaskItem>>(client_id);
 }
 
+tl::expected<void, ErrorCode> MasterClient::AckRemoveObjectHeartbeat(
+    const UUID& client_id, const std::vector<RemoveTaskItem>& tasks) {
+    ScopedVLogTimer timer(1, "MasterClient::AckRemoveObjectHeartbeat");
+    timer.LogRequest("client_id=", client_id.first, ":", client_id.second,
+                     " tasks=", tasks.size());
+    return invoke_rpc<&WrappedMasterService::AckRemoveObjectHeartbeat, void>(
+        client_id, tasks);
+}
+
 tl::expected<PromotionAllocStartResponse, ErrorCode>
 MasterClient::PromotionAllocStart(
     const UUID& client_id, const std::string& key, uint64_t size,