[Bug]: [SSD] DSV4 enable SSD offload，stress testing, repeat offloading the same batch key, lead to OBJECT_ALREADY_EXISTS/persist failed/INVALID_KEY

mooncake-store/include/file_storage.h

@@ -77,6 +77,26 @@ class FileStorage {
     tl::expected<void, ErrorCode> OffloadObjects(
         const std::vector<OffloadTaskItem>& offloading_objects);
 
+    /**
+     * @brief Classifies a BatchOffload error as affecting only the current
+     * bucket rather than the whole offload cycle.
+     *
+     * Such an error means the bucket's keys simply cannot be persisted this
+     * round; OffloadObjects reports them back to the master as failed (so their
+     * offloading tasks and source-replica refcounts are released) and continues
+     * with the remaining buckets, instead of aborting the entire cycle.
+     *
+     *   - INVALID_READ: source data for these keys could not be read/staged.
+     *   - OBJECT_ALREADY_EXISTS: the key(s) were already offloaded, or are
+     *     being offloaded concurrently. The bucket backend rejects the whole
+     *     bucket atomically by design (see BucketStorageBackend::BatchOffload
+     *     and its PrepareEviction duplicate guard). Treating this as fatal
+     *     aborted the cycle, leaked offloading tasks, and left master/SSD
+     *     metadata inconsistent, which surfaced as spurious INVALID_KEY on the
+     *     read path (issue #2827).
+     */
+    static bool IsPerBucketSoftOffloadError(ErrorCode error);
+
     /**
      * @brief Performs a heartbeat operation for the FileStorage component.
      * 1. Sends object status (e.g., access frequency, size) to the master via

mooncake-store/src/file_storage.cpp

@@ -440,6 +440,11 @@ tl::expected<FileStorage::BatchGetResult, ErrorCode> FileStorage::BatchGet(
     return batch_result;
 }
 
+bool FileStorage::IsPerBucketSoftOffloadError(ErrorCode error) {
+    return error == ErrorCode::INVALID_READ ||
+           error == ErrorCode::OBJECT_ALREADY_EXISTS;
+}
+
 tl::expected<void, ErrorCode> FileStorage::OffloadObjects(
     const std::vector<OffloadTaskItem>& offloading_objects) {
     if (offloading_objects.empty()) {
@@ -507,6 +512,10 @@ tl::expected<void, ErrorCode> FileStorage::OffloadObjects(
     // orphaned offloading_tasks and release source replica refcounts.
     std::vector<OffloadTaskItem> failed_tasks;
     std::unordered_set<std::string> all_bucket_keys;
+    // Set when a whole-cycle error aborts the bucket loop early. We still fall
+    // through to the NACK flush below before returning it, so no drained key is
+    // left waiting on the TTL reaper.
+    std::optional<ErrorCode> abort_error;
 
     for (const auto& keys : buckets_keys) {
         for (const auto& k : keys) all_bucket_keys.insert(k);
@@ -584,6 +593,14 @@ tl::expected<void, ErrorCode> FileStorage::OffloadObjects(
             }
         }
 
+        // If every object in this bucket failed D2H staging, host_batch_object
+        // is empty (those keys are already in failed_tasks). Skip BatchOffload,
+        // which rejects an empty map as INVALID_KEY and would otherwise trip the
+        // whole-cycle abort below for a bucket that has nothing left to persist.
+        if (host_batch_object.empty()) {
+            continue;
+        }
+
         auto offload_start = std::chrono::steady_clock::now();
         auto bucket_complete_handler =
             [this, offload_start, complete_handler](
@@ -623,18 +640,29 @@ tl::expected<void, ErrorCode> FileStorage::OffloadObjects(
         if (!offload_res) {
             LOG(ERROR) << "Failed to store objects with error: "
                        << offload_res.error();
+            // This bucket did not persist, so report its keys back to the
+            // master as failed regardless of whether we continue or abort.
+            // Doing it here (rather than only on the soft path) keeps their
+            // offloading tasks and source-replica refcounts from leaking until
+            // the put_start_release_timeout_sec_ TTL reaper fires.
+            for (const auto& [key, _] : host_batch_object) {
+                failed_tasks.push_back(task_by_storage_key.at(key));
+            }
             if (offload_res.error() == ErrorCode::KEYS_ULTRA_LIMIT) {
+                // Disk is over the key-count limit: stop offloading entirely.
                 MutexLocker locker(&offloading_mutex_);
                 enable_offloading_ = false;
-                return tl::make_unexpected(offload_res.error());
             }
-            if (offload_res.error() == ErrorCode::INVALID_READ) {
-                for (const auto& [key, _] : host_batch_object) {
-                    failed_tasks.push_back(task_by_storage_key.at(key));
-                }
-            } else {
-                return tl::make_unexpected(offload_res.error());
+            if (!IsPerBucketSoftOffloadError(offload_res.error())) {
+                // Whole-cycle error (KEYS_ULTRA_LIMIT or any hard failure):
+                // stop processing further buckets, but fall through to the NACK
+                // flush below so every drained key is released. Unvisited
+                // buckets are not yet in all_bucket_keys, so the sweep NACKs
+                // them too; this bucket's keys were just pushed above.
+                abort_error = offload_res.error();
+                break;
             }
+            // Soft per-bucket error: keep processing the remaining buckets.
         }
     }
 
@@ -661,6 +689,9 @@ tl::expected<void, ErrorCode> FileStorage::OffloadObjects(
         }
     }
 
+    if (abort_error) {
+        return tl::make_unexpected(*abort_error);
+    }
     return {};
 }
 

mooncake-store/tests/file_storage_test.cpp

@@ -108,6 +108,12 @@ class FileStorageTest : public ::testing::Test {
         return bucket_backend->UngroupedOffloadingObjectsSize();
     }
 
+    // Static funnel to the private FileStorage::IsPerBucketSoftOffloadError.
+    // FileStorageTest is friended; TEST_F-generated subclasses are not.
+    static bool CallIsPerBucketSoftOffloadError(ErrorCode error) {
+        return FileStorage::IsPerBucketSoftOffloadError(error);
+    }
+
     void AssertHeartbeatEvictsAllKeys(
         FileStorage& fileStorage, const std::vector<std::string>& expected_keys,
         const std::unordered_map<std::string, std::vector<Slice>>&
@@ -521,6 +527,25 @@ TEST_F(FileStorageTest, NotifyEvictedDiskReplicasUsesTenantScopedKeys) {
     }
 }
 
+// Regression test for issue #2827: under concurrent/repeat offload of the same
+// keys, the bucket backend rejects a whole bucket atomically with
+// OBJECT_ALREADY_EXISTS (see BucketStorageBackend duplicate-key tests). That
+// error must be treated as a per-bucket soft failure so OffloadObjects reports
+// the keys back to the master and continues, rather than aborting the whole
+// offload cycle and leaving master/SSD metadata inconsistent (which surfaced as
+// spurious INVALID_KEY on the read path). It stays alongside INVALID_READ,
+// while genuinely fatal errors (e.g. KEYS_ULTRA_LIMIT, INTERNAL_ERROR) do not.
+TEST_F(FileStorageTest, DuplicateOffloadErrorIsPerBucketSoftError) {
+    EXPECT_TRUE(
+        CallIsPerBucketSoftOffloadError(ErrorCode::OBJECT_ALREADY_EXISTS));
+    EXPECT_TRUE(CallIsPerBucketSoftOffloadError(ErrorCode::INVALID_READ));
+
+    EXPECT_FALSE(CallIsPerBucketSoftOffloadError(ErrorCode::KEYS_ULTRA_LIMIT));
+    EXPECT_FALSE(CallIsPerBucketSoftOffloadError(ErrorCode::INTERNAL_ERROR));
+    EXPECT_FALSE(CallIsPerBucketSoftOffloadError(ErrorCode::INVALID_KEY));
+    EXPECT_FALSE(CallIsPerBucketSoftOffloadError(ErrorCode::OK));
+}
+
 TEST_F(FileStorageTest, InvalidIntValueUsesDefault) {
     SetEnv("MOONCAKE_OFFLOAD_BUCKET_KEYS_LIMIT", "abc");
     SetEnv("MOONCAKE_OFFLOAD_TOTAL_SIZE_LIMIT_BYTES", "sdfsdf");