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

[pjdurden]

Issue #2827 — SSD offload duplicate-key storm (OBJECT_ALREADY_EXISTS →

persist failed → INVALID_KEY)

Upstream issue: #2827

Symptom (DeepSeek‑V4, PD‑disaggregated, SSD offload enabled, 32‑concurrency
stress): the same batch KV keys are offloaded repeatedly, the P node logs
Duplicate key detected in BatchOffload … Returning OBJECT_ALREADY_EXISTS,
whole‑batch persist failed, and the D node then gets floods of
Key not found … BatchQuerySlices/BatchLoad … INVALID_KEY. A single request
never triggers it — only concurrent/repeat offload of the same key does.

1. Root cause

Under concurrency, two offload flows on the same node can target an overlapping
KV key (shared prefix block → identical block hash). The bucket storage backend
is, by design, a strict single‑writer‑per‑key store that commits a bucket
atomically:

• BucketStorageBackend::BatchOffload first calls PrepareEviction
  (mooncake-store/src/storage_backend.cpp), which reserves every key of the
  bucket in pending_write_keys_ and returns OBJECT_ALREADY_EXISTS for the
  entire bucket if any key is already committed (object_bucket_map_),
  already reserved (pending_write_keys_), or being evicted
  (pending_eviction_keys_).
• A secondary commit‑time pre‑check does the same.

This atomic‑bucket rejection is intentional and is locked in by existing tests
(storage_backend_test.cpp: BucketStorageBackend_DuplicateKeyRejected,
BucketStorageBackend_DuplicateBatchPartialRejection — the latter asserts a
batch [keyA(dup), keyB(new)] rejects the whole batch and keyB is not
written).

The actual bug is in the caller, FileStorage::OffloadObjects
(mooncake-store/src/file_storage.cpp). Its per‑bucket error handling treated
only INVALID_READ as a recoverable, per‑bucket condition and treated
everything else — including OBJECT_ALREADY_EXISTS — as fatal:

if (offload_res.error() == ErrorCode::INVALID_READ) {
    for (…) failed_tasks.push_back(…);        // report keys, keep going
} else {
    return tl::make_unexpected(offload_res.error());   // abort whole cycle
}

So a single duplicate key would return out of OffloadObjects, which:

1. Skips the remaining buckets in this offload cycle.
2. Skips the failed_tasks NACK reporting at the end of the function
   (the early return jumps over it). Those keys were drained from the
   master's offloading queue but are never resolved.
3. Leaves the master's per‑key offloading_tasks dangling and never releases
   the source‑replica refcounts (dec_refcnt), so master and local‑SSD
   metadata drift apart. Keys the master believes are already offloaded are
   not actually on this node's SSD.

When the D node later reads those keys, the master redirects the read to the P
node's SSD, object_bucket_map_ has no entry → BatchLoad/BatchQuerySlices
return INVALID_KEY — exactly the flood in the report.

2. The fix and why

Treat OBJECT_ALREADY_EXISTS from BatchOffload the same way INVALID_READ
is already treated: a per‑bucket soft failure, not a fatal one. The
bucket's keys are pushed into failed_tasks and the loop continues to the next
bucket; at the end of OffloadObjects those keys are reported back to the
master with the existing data_size == -1 NACK sentinel.

This is the minimal change that resolves the reported failure chain, and it is
aligned with the issue's own primary recommendation (#1: handle
OBJECT_ALREADY_EXISTS idempotently, do not fail the whole batch). It does
not touch the backend's deliberate atomic‑bucket contract.

Why it is correct and safe:

• The master's NACK handler (MasterService::NotifyOffloadSuccess, the
  metadata.data_size < 0 branch) only erases the key's offloading_task and
  dec_refcnts the source replica if a task is present; it never removes
  or overwrites an existing valid replica. So NACK'ing a key that another flow
  legitimately committed is a harmless no‑op on that valid replica, and it
  correctly releases this flow's dangling task/refcount. This holds under any
  ordering of the concurrent success/NACK RPCs.
• No more ghost LOCAL_DISK replicas → no more spurious INVALID_KEY on the
  read path.
• Remaining buckets in the cycle are still processed; every drained key is
  resolved on the master (success or NACK), so tasks/refcounts no longer leak
  (which otherwise also blocks future writes via OBJECT_HAS_REPLICATION_TASK).

The classification is factored into a small, documented, testable predicate
FileStorage::IsPerBucketSoftOffloadError(ErrorCode). Genuinely global errors
(KEYS_ULTRA_LIMIT, INTERNAL_ERROR, BUCKET_NOT_FOUND, …) still abort the
cycle as before.

Trade‑off: on the rare colliding bucket, the non‑duplicate keys in that bucket
are not persisted this round (they are NACK'd and can be re‑offloaded or
recomputed later). This is not a regression — the previous behaviour
discarded the whole bucket too, and additionally aborted the entire cycle and
leaked master state. Correctness (no data loss on the read path) is preserved
because a NACK never deletes a valid replica.

3. Files changed

• mooncake-store/include/file_storage.h — declare the private static
  predicate IsPerBucketSoftOffloadError(ErrorCode) with rationale doc.
• mooncake-store/src/file_storage.cpp — define the predicate and use it in
  OffloadObjects' per‑bucket error handling so OBJECT_ALREADY_EXISTS is
  handled alongside INVALID_READ (report keys as failed, continue) instead of
  aborting the cycle.
• mooncake-store/tests/file_storage_test.cpp — add
  DuplicateOffloadErrorIsPerBucketSoftError (with a static funnel, mirroring
  the repo's existing friend‑access pattern) asserting the classification:
  OBJECT_ALREADY_EXISTS and INVALID_READ are per‑bucket soft errors, while
  KEYS_ULTRA_LIMIT / INTERNAL_ERROR / INVALID_KEY / OK are not.

4. Risk / uncertainty

• Low risk. The change reuses the already‑tested failed_tasks → NACK path
  and adds one error code to a classification. The backend contract and all its
  duplicate‑key tests are untouched.
• The concurrency origin of the duplicate (which specific interleaving
  produces OBJECT_ALREADY_EXISTS at runtime — pending_write_keys_
  reservation vs. pending_eviction_keys_ vs. commit‑time race) was analysed
  statically but not reproduced live; the fix is deliberately defensive and
  correct regardless of which interleaving triggers it.
• The added test is a focused unit test of the decision logic rather than a
  full concurrent end‑to‑end reproduction. A deterministic end‑to‑end test
  would need to force the transient backend state (in‑flight eviction /
  concurrent writer) and heavy InProcMaster + real MEM‑replica setup; that
  was judged disproportionate and flaky for a one‑condition caller fix.

5. How I verified it

• Static tracing of the full failure chain across
  storage_backend.cpp (BatchOffload / PrepareEviction /
  RollbackCommittedBucket / eviction), file_storage.cpp
  (OffloadObjects / BatchQuerySegmentSlices / NotifyEvictedDiskReplicas),
  and master_service.cpp (NotifyOffloadSuccess NACK path,
  OffloadObjectHeartbeat) — confirming (a) where the whole‑bucket
  OBJECT_ALREADY_EXISTS originates, (b) that the early return skips the
  failed_tasks NACK, and (c) that the master NACK path never removes a valid
  replica and dec_refcnts at most once per task.
• Confirmed the backend behaviour is intentional via the existing
  storage_backend_test.cpp duplicate‑key tests, so the fix belongs in the
  caller.
• Added DuplicateOffloadErrorIsPerBucketSoftError covering the changed
  classification.
• Style/format: verified all edited lines are within the repo's
  ColumnLimit: 80 (Google‑based .clang-format) and match surrounding
  4‑space indentation and naming.
• Not run: the C++ test binaries were not compiled/executed in this
  environment (no configured build; RDMA/uring/etc. toolchain not provisioned),
  so runtime verification is limited to the reasoning above. git commit/
  push/gh were intentionally not run.

[gemini-code-assist]

Code Review

This pull request introduces the IsPerBucketSoftOffloadError helper to classify certain errors, such as OBJECT_ALREADY_EXISTS and INVALID_READ, as soft, per-bucket failures during object offloading. This prevents aborting the entire offload cycle and leaking tasks when duplicate offloads occur. However, a critical leak path remains: if a soft error occurs and populates failed_tasks, but a subsequent bucket encounters a hard error, the early return will bypass the NACK reporting block, causing the accumulated failed_tasks to leak.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

While accumulating failed tasks in failed_tasks for soft errors is a great improvement, there is still a critical leak path. If a soft error occurs in an earlier bucket (populating failed_tasks), but a subsequent bucket encounters a hard error (such as KEYS_ULTRA_LIMIT or any other unhandled error), the function will execute an early return tl::make_unexpected(...) (e.g., on line 634 or 645). This early return bypasses the NACK reporting block at the end of the function, causing all previously accumulated failed_tasks to be leaked on the master (dangling tasks and unreleased replica refcounts).\n\nTo completely prevent this, we should ensure that any early return path also reports the accumulated failed_tasks (and ideally any remaining unprocessed tasks in the cycle) to the master before returning, or use an RAII cleanup guard to automate the reporting of any tasks that were not successfully completed.

[Copilot]

Pull request overview

Fixes a concurrency-triggered SSD offload failure mode where BucketStorageBackend::BatchOffload can legitimately return OBJECT_ALREADY_EXISTS for an entire bucket, but FileStorage::OffloadObjects previously treated that as fatal—aborting the cycle and skipping the “failed tasks” NACK reporting back to the master, which could leave master/SSD metadata inconsistent and surface as INVALID_KEY on later reads.

Changes:

• Introduce FileStorage::IsPerBucketSoftOffloadError(ErrorCode) to classify per-bucket “soft” failures.
• Update FileStorage::OffloadObjects to treat OBJECT_ALREADY_EXISTS the same way as INVALID_READ: record the bucket’s keys as failed and continue processing remaining buckets (instead of aborting the whole cycle).
• Add a unit test verifying the error classification behavior for relevant error codes.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated no comments.

File	Description
mooncake-store/include/file_storage.h	Declares and documents the new private helper predicate for per-bucket soft offload errors.
mooncake-store/src/file_storage.cpp	Implements the predicate and uses it to handle OBJECT_ALREADY_EXISTS without aborting the full offload cycle.
mooncake-store/tests/file_storage_test.cpp	Adds a regression/unit test ensuring OBJECT_ALREADY_EXISTS is classified as a per-bucket soft offload error.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[LujhCoconut]

Thanks for this PR. Overall this fix is solid: both OBJECT_ALREADY_EXISTS producers return before complete_handler with local state cleaned up, so the whole-bucket NACK is consistent with the master side (idempotent, never touches existing replicas).

The leak Gemini flagged is pre-existing and bounded by the TTL reaper (put_start_release_timeout_sec_, 600s default), but it's cheap to close here: turn the fatal-path returns into record-error + break — since all_bucket_keys is populated incrementally, unvisited buckets' keys automatically fall through the existing "not in any bucket" sweep (file_storage.cpp:643-647) and get NACKed. ~±30 lines, single function.

Two more non-blocking suggestions, both cheap enough to ride along:

1. Empty-bucket hard abort: the empty guard checks batch_object (file_storage.cpp:546) but BatchOffload receives host_batch_object. If every object in a bucket fails D2H staging, BatchOffload gets an empty map and returns INVALID_KEY (storage_backend.cpp:1623) → hard abort, even though those keys are already in failed_tasks. One line after the staging loop: if (host_batch_object.empty()) continue;

2. Flush before the KEYS_ULTRA_LIMIT return: after enable_offloading_ = false the drained tasks will never be retried, and the RPC channel is healthy (the failure is local disk capacity) — NACK-flushing failed_tasks before returning is both safe and maximally useful.

Lastly, the actual duplicate source — worth a separate follow-up: in GroupOffloadingKeysByBucket, the ungrouped-pool drain neither re-runs is_exist_func nor dedups against the incoming iterator, so a parked key re-issued by the master can land in two buckets of one cycle, single-threaded — the first commits, the second trips PrepareEviction's OBJECT_ALREADY_EXISTS. Placement depends on unordered_map iteration order, which matches the stress-only repro in #2827. Fix is ~10 lines (per-call seen-set + is_exist_func on drained keys); skipped keys reuse the existing all_bucket_keys NACK path. Doesn't block this PR.

[pjdurden]

Thanks for the detailed pass. Pushed a follow-up (a38cbe6) that closes the leak and folds in the two cheap suggestions:

• Fatal-path returns → record-and-break. The KEYS_ULTRA_LIMIT and hard-error branches no longer return inside the loop. On any BatchOffload failure the bucket's host_batch_object keys are pushed to failed_tasks first; a non-soft error then sets abort_error and breaks. Execution falls through to the not-in-any-bucket sweep (which NACKs the unvisited buckets, since they were never added to all_bucket_keys) and the failed_tasks flush, so nothing waits on the put_start_release_timeout_sec_ reaper. The original error is still returned to the caller after the flush.
• Empty-bucket guard. Added if (host_batch_object.empty()) continue; after the D2H staging loop — those keys are already in failed_tasks, and an empty map would otherwise hit the INVALID_KEY hard abort.

Left the GroupOffloadingKeysByBucket ungrouped-drain dedup (the actual duplicate source) out of this PR as you suggested — will open a separate follow-up with the per-call seen-set + is_exist_func on drained keys.