Harden transfer contracts and isolate cold-cache failures

README.md

@@ -20,9 +20,9 @@ V1 runs the cache contract on CPU. V2 adds a transport trait, a simulated regist
 | V1 B: RustFS live Put/Head/RangeGet/Delete | `NOT_RUN_HARDWARE` (no authorized endpoint). Contract mock passes |
 | V1 C: fuzz, fault injection, synthetic token checks, paired-baseline stubs | Implemented in this tree. Real engine output is `NOT_RUN_HARDWARE` |
 | V2: `KvTransport`, TCP default, feature `rdma-nixl` | In progress. The feature compiles and `NixlTransport::open` refuses. Live RDMA is `NOT_RUN_HARDWARE` |
-| V2: bounded registered buffer pool (simulated pin, reuse, quarantine, backpressure) | Implemented in this tree. Not a NIC registration |
-| V2: shard-local directory, epoch fencing, single-flight, capacity credits | Implemented in-process. One process is one shard. No cross-shard quorum |
-| V2: G4 pack/manifest, async sink, cost-bounded refill | Implemented against the mock. Live RustFS is `NOT_RUN_HARDWARE` |
+| V2: bounded registered buffer pool (simulated pin, reuse, quarantine, backpressure) | Implemented in this tree. Checkout prefers an already-bound slot and does not partially check out a batch. Not a NIC registration |
+| V2: shard-local directory, epoch fencing, single-flight, capacity credits | Implemented in-process. One process is one shard. Finish, failure, and fence leave a notice with no payload. An epoch that cannot advance does not drop the fill. No cross-shard quorum |
+| V2: G4 pack/manifest, async sink, cost-bounded refill | Implemented against the mock. A failed drain keeps the job. A wrong-tenant refill does not return bytes. Live RustFS is `NOT_RUN_HARDWARE` |
 | V2: P/D, G3.5, and G4 hop list | Documented in [docs/V2_DATA_PATHS.md](docs/V2_DATA_PATHS.md). Not a zero-copy claim |
 | V2: paired TTFT/TPOT against native NIXL | `NOT_RUN_HARDWARE` |
 | V3: `dpu-bf3` feature slot, simulated DRAM slab + storage queue, SPDK affinity check | In progress. Devices are not linked. `open_doca` / `open_nvme_device` / `open_spdk` refuse. `dpu-demo` prints `NOT_RUN_HARDWARE` |
@@ -33,6 +33,7 @@ V1 runs the cache contract on CPU. V2 adds a transport trait, a simulated regist
 | V3: G4 retention, manifest version, reclaim, delete verification, tenant cold keys, backfill budget | Implemented against the mock. Live RustFS is `NOT_RUN_HARDWARE` |
 | V3: paired NIC-RDMA baseline vs DPU | `NOT_RUN_HARDWARE`. `tco=TCO_UNKNOWN`. `go_gate=NOT_EVALUATED`. TCP is not the DPU baseline |
 | V3: second engine (SGLang / HiCache) | Not added. Evaluation note only. See [docs/adr/0008-second-engine.md](docs/adr/0008-second-engine.md) |
+| Mooncake-informed CPU contracts | Implemented in this tree. See [docs/MOONCAKE.md](docs/MOONCAKE.md) and [docs/adr/0009-mooncake-lessons.md](docs/adr/0009-mooncake-lessons.md). Mooncake is not linked. No NIC, Soft-RoCE, or DPU number is claimed |
 | Erasure coding and durability replicas | Will not be built. A lost or failed hot block is a clean miss, then recompute and repopulate. Cold durability is RustFS G4 |
 
 The integration route (native vLLM KV connector or LMCache) is **not chosen**. See [docs/adr/0001-integration-route.md](docs/adr/0001-integration-route.md). Hot-tier failure policy is [docs/adr/0002-no-erasure-coding.md](docs/adr/0002-no-erasure-coding.md). The V3 offload candidate is [docs/adr/0006-storage-queue-offload.md](docs/adr/0006-storage-queue-offload.md). It is a hypothesis pending a paired measurement, not a measured bottleneck.
@@ -55,7 +56,9 @@ The batch API is Lookup, Reserve/Put, Commit, Get, Release/Cancel, and Invalidat
 
 Lookup can return `PrefixHit` when a committed block's hash chain is a strict prefix of the query and the layout, dtype, shard, and model identity match. The hit is the whole stored block. `Get` must use `identity_for_prefix_hit`. A shorter `valid_length` on the same chain is an incompatible miss. The service does not slice inside a block.
 
-`gpucache-cold` packs blocks into an object plus a manifest and can Put, Head, RangeGet, and Delete through a trait. The TCP hot path does not call it. Without `GPUCACHE_RUSTFS_*` variables the live probe is `NOT_RUN_HARDWARE`. The mock still runs. V2 adds a bounded async sink and a refill that returns no bytes when the combined manifest and payload downloads exceed its byte budget.
+`gpucache-cold` packs blocks into an object plus a manifest and can Put, Head, RangeGet, and Delete through a trait. The TCP hot path does not call it. Without `GPUCACHE_RUSTFS_*` variables the live probe is `NOT_RUN_HARDWARE`. The mock still runs. Its bounded sink isolates permanent failures, caps transient retries, and allows healthy jobs to advance. Pending and failed jobs share admission limits; `take_failed` releases one failed job without cancelling other work. Content-derived object and manifest keys prevent interleaved retries from overwriting another pack. A PUT error does not trigger object deletion because publication may have succeeded before an acknowledgement was lost.
+
+Every public cold refill requires the full expected `CacheIdentity` from the upstream authenticated request. Manifest v3 binds each range to that identity, and mismatches return no payload. Legacy v1/v2 entries are clean misses and must be recomputed and repacked. The combined manifest and payload downloads must fit the refill byte budget. This identity check does not authenticate a caller or validate the producer's model computation. See [docs/V2_DATA_PATHS.md](docs/V2_DATA_PATHS.md) for the API migration and publication contract.
 
 Lifecycle:
 
@@ -161,7 +164,7 @@ docs/adr             decision records
 COMPATIBILITY.md     environment manifest
 ```
 
-Design notes: [docs/ARCHITECTURE.md](docs/ARCHITECTURE.md), [docs/PROTOCOL.md](docs/PROTOCOL.md), [docs/SECURITY.md](docs/SECURITY.md), [docs/OPERATIONS.md](docs/OPERATIONS.md), [docs/BENCHMARKS.md](docs/BENCHMARKS.md), [docs/CORRECTNESS.md](docs/CORRECTNESS.md), [docs/RISKS.md](docs/RISKS.md), [docs/V1C_TEST_MATRIX.md](docs/V1C_TEST_MATRIX.md), [docs/V2_DATA_PATHS.md](docs/V2_DATA_PATHS.md), [docs/V2_TEST_MATRIX.md](docs/V2_TEST_MATRIX.md), [docs/V3_DPU_TARGETS.md](docs/V3_DPU_TARGETS.md), [docs/V3_TEST_MATRIX.md](docs/V3_TEST_MATRIX.md), [docs/V3_FAILURES.md](docs/V3_FAILURES.md).
+Design notes: [docs/ARCHITECTURE.md](docs/ARCHITECTURE.md), [docs/PROTOCOL.md](docs/PROTOCOL.md), [docs/SECURITY.md](docs/SECURITY.md), [docs/OPERATIONS.md](docs/OPERATIONS.md), [docs/BENCHMARKS.md](docs/BENCHMARKS.md), [docs/CORRECTNESS.md](docs/CORRECTNESS.md), [docs/RISKS.md](docs/RISKS.md), [docs/MOONCAKE.md](docs/MOONCAKE.md), [docs/V1C_TEST_MATRIX.md](docs/V1C_TEST_MATRIX.md), [docs/V2_DATA_PATHS.md](docs/V2_DATA_PATHS.md), [docs/V2_TEST_MATRIX.md](docs/V2_TEST_MATRIX.md), [docs/V3_DPU_TARGETS.md](docs/V3_DPU_TARGETS.md), [docs/V3_TEST_MATRIX.md](docs/V3_TEST_MATRIX.md), [docs/V3_FAILURES.md](docs/V3_FAILURES.md).
 
 ## License
 

crates/bench/src/lib.rs

@@ -30,6 +30,9 @@ pub fn paired_baseline_report() -> String {
         "samples=NONE",
         "numeric_tolerance=NOT_APPLICABLE",
         "bypass_when_no_reuse=MISS_RECOMPUTE",
+        "mooncake_transfer_engine=NOT_LINKED",
+        "mooncake_published_numbers=NOT_REPRODUCED",
+        "rdma_fallback=TCP_CPU_CONTRACT",
     ];
     let mut out = String::new();
     for line in lines {
@@ -73,6 +76,7 @@ pub fn v3_paired_report() -> String {
         "pb_scale=NOT_CLAIMED",
         "production_ready=NOT_CLAIMED",
         "soft_roce=NOT_A_PERFORMANCE_RESULT",
+        "mooncake_rdma=NOT_A_PERFORMANCE_RESULT",
     ];
     let mut out = String::new();
     for line in lines {
@@ -108,6 +112,9 @@ mod tests {
             "tpot=NOT_RUN_HARDWARE",
             "goodput=NOT_RUN_HARDWARE",
             "samples=NONE",
+            "mooncake_transfer_engine=NOT_LINKED",
+            "mooncake_published_numbers=NOT_REPRODUCED",
+            "rdma_fallback=TCP_CPU_CONTRACT",
         ] {
             assert!(report.contains(key), "missing {key}");
         }
@@ -129,6 +136,7 @@ mod tests {
             "pb_scale=NOT_CLAIMED",
             "production_ready=NOT_CLAIMED",
             "soft_roce=NOT_A_PERFORMANCE_RESULT",
+            "mooncake_rdma=NOT_A_PERFORMANCE_RESULT",
         ] {
             assert!(report.contains(key), "missing {key}");
         }

crates/cold/src/demo.rs

@@ -8,11 +8,12 @@
 
 use core::fmt;
 
-use gpucache_core_types::{content_checksum, synthetic_identity, CacheError};
+use gpucache_core_types::{content_checksum, synthetic_identity, CacheError, TenantId};
 
 use crate::http::{live_probe, ProbeStatus};
 use crate::pack::{contract_smoke, pack_blocks, PackedBlock};
-use crate::sink::{refill_generation, AsyncSink, ColdJob, RefillBudget, RefillOutcome};
+use crate::refill::{refill_generation, refill_matching, RefillBudget, RefillOutcome};
+use crate::sink::{AsyncSink, ColdJob};
 use crate::MockObjectStore;
 
 #[derive(Clone, Debug, PartialEq, Eq)]
@@ -27,6 +28,7 @@ impl fmt::Display for ColdReport {
         writeln!(f, "async_sink=DRAINED")?;
         writeln!(f, "refill=CHECKSUM_OK")?;
         writeln!(f, "over_budget=NO_BYTES")?;
+        writeln!(f, "wrong_tenant=NO_BYTES")?;
         writeln!(f, "rustfs_live={}", self.live)?;
         writeln!(f, "gds=NOT_RUN_HARDWARE")?;
         Ok(())
@@ -46,18 +48,30 @@ pub fn run_cold_demo() -> Result<ColdReport, CacheError> {
         });
     }
     let budget = RefillBudget::new(1024 * 1024, 16)?;
-    match refill_generation(&store, &pack.manifest_key, 2, &budget)? {
+    let expected = &blocks[1].identity;
+    match refill_generation(&store, &pack.manifest_key, 2, &budget, expected)? {
         RefillOutcome::Loaded { payload } if payload == blocks[1].payload => {}
         _ => {
             return Err(CacheError::ChecksumMismatch);
         }
     }
     let tight = RefillBudget::new(1, 16)?;
-    if refill_generation(&store, &pack.manifest_key, 2, &tight)? != RefillOutcome::OverBudget {
+    if refill_generation(&store, &pack.manifest_key, 2, &tight, expected)?
+        != RefillOutcome::OverBudget
+    {
         return Err(CacheError::Malformed {
             reason: "refill budget",
         });
     }
+    match refill_matching(&store, &pack.manifest_key, 2, &budget, expected)? {
+        RefillOutcome::Loaded { payload } if payload == blocks[1].payload => {}
+        _ => return Err(CacheError::ChecksumMismatch),
+    }
+    let mut foreign = expected.clone();
+    foreign.tenant = TenantId::new(b"other-tenant".to_vec())?;
+    if refill_matching(&store, &pack.manifest_key, 2, &budget, &foreign).is_ok() {
+        return Err(CacheError::Unauthorized);
+    }
     let live = match live_probe().status {
         ProbeStatus::NotRunHardware { .. } => "NOT_RUN_HARDWARE",
         ProbeStatus::Passed => "PASSED",
@@ -96,6 +110,7 @@ mod tests {
         let text = report.to_string();
         assert!(text.contains("status=COLD_CONTRACT_OK"));
         assert!(text.contains("over_budget=NO_BYTES"));
+        assert!(text.contains("wrong_tenant=NO_BYTES"));
         assert!(text.contains("gds=NOT_RUN_HARDWARE"));
         if std::env::var_os("GPUCACHE_RUSTFS_ENDPOINT").is_none() {
             assert!(text.contains("rustfs_live=NOT_RUN_HARDWARE"));

crates/cold/src/lib.rs

@@ -14,18 +14,20 @@ mod http;
 mod mock;
 mod ops;
 mod pack;
+mod refill;
 mod sigv4;
 mod sink;
 
 pub use demo::{run_cold_demo, ColdReport};
 pub use http::{live_probe, ProbeReport, ProbeStatus};
 pub use mock::MockObjectStore;
 pub use ops::{
-    enqueue_backfill, refill_degraded, verify_deleted, BackfillLimit, ColdAvailability, ColdKeys,
-    ColdOps, ObjectRecord, ReclaimStep, Retention,
+    enqueue_backfill, refill_degraded, refill_for_tenant, verify_deleted, BackfillLimit,
+    ColdAvailability, ColdKeys, ColdOps, ObjectRecord, ReclaimStep, Retention,
 };
 pub use pack::{
     contract_smoke, manifest_version, pack_blocks, read_manifest, Manifest, ObjectMeta,
-    ObjectStore, PackedBlock, PayloadRange, MANIFEST_VERSION,
+    ObjectStore, Pack, PackedBlock, PayloadRange, MANIFEST_VERSION,
 };
-pub use sink::{refill_generation, AsyncSink, ColdJob, RefillBudget, RefillOutcome};
+pub use refill::{refill_generation, refill_matching, RefillBudget, RefillOutcome};
+pub use sink::{AsyncSink, ColdJob, DrainReport, FailedColdJob, MAX_SINK_JOBS, MAX_WRITE_ATTEMPTS};

crates/cold/src/ops.rs

@@ -11,10 +11,11 @@
 
 use std::collections::{BTreeMap, VecDeque};
 
-use gpucache_core_types::{CacheError, Checksum, QuotaKind};
+use gpucache_core_types::{CacheError, CacheIdentity, Checksum, QuotaKind};
 
 use crate::pack::{key_ok, ObjectStore, MANIFEST_VERSION};
-use crate::sink::{refill_generation, AsyncSink, ColdJob, RefillBudget, RefillOutcome};
+use crate::refill::{refill_generation, refill_matching, RefillBudget, RefillOutcome};
+use crate::sink::{AsyncSink, ColdJob};
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub struct Retention {
@@ -43,16 +44,38 @@ pub enum ColdAvailability {
     Unavailable,
 }
 
+/// Refill only for the complete expected identity from an authenticated caller.
+/// The historical name does not provide a tenant-only or unauthenticated path.
+pub fn refill_for_tenant(
+    store: &dyn ObjectStore,
+    manifest_key: &str,
+    generation: u64,
+    budget: &RefillBudget,
+    availability: ColdAvailability,
+    expected: &CacheIdentity,
+) -> Result<RefillOutcome, CacheError> {
+    match availability {
+        ColdAvailability::Slow | ColdAvailability::Unavailable => Ok(RefillOutcome::Missing),
+        ColdAvailability::Ready => {
+            refill_matching(store, manifest_key, generation, budget, expected)
+        }
+    }
+}
+
+/// Availability wrapper around the same mandatory identity-matching refill.
 pub fn refill_degraded(
     store: &dyn ObjectStore,
     manifest_key: &str,
     generation: u64,
     budget: &RefillBudget,
     availability: ColdAvailability,
+    expected: &CacheIdentity,
 ) -> Result<RefillOutcome, CacheError> {
     match availability {
         ColdAvailability::Slow | ColdAvailability::Unavailable => Ok(RefillOutcome::Missing),
-        ColdAvailability::Ready => refill_generation(store, manifest_key, generation, budget),
+        ColdAvailability::Ready => {
+            refill_generation(store, manifest_key, generation, budget, expected)
+        }
     }
 }
 
@@ -434,7 +457,7 @@ mod tests {
         let mut ops = ColdOps::new(Retention::new(10).unwrap());
         let generation = ops.keys_mut().issue(b"tenant-b").unwrap();
         let mut bad = record("ns/object", generation, 0);
-        bad.manifest_version = 2;
+        bad.manifest_version = 9;
         assert!(ops.record_put(bad).is_err());
     }
 
@@ -471,9 +494,62 @@ mod tests {
         store.put("ns/manifest", b"not-a-manifest").unwrap();
         let budget = RefillBudget::new(1024, 1).unwrap();
         for availability in [ColdAvailability::Slow, ColdAvailability::Unavailable] {
-            let outcome = refill_degraded(&store, "ns/manifest", 1, &budget, availability).unwrap();
+            let outcome = refill_degraded(
+                &store,
+                "ns/manifest",
+                1,
+                &budget,
+                availability,
+                &synthetic_identity(1, 1),
+            )
+            .unwrap();
             assert_eq!(outcome, RefillOutcome::Missing);
         }
         assert!(store.head("ns/manifest").is_ok());
     }
+
+    struct ClosedStore;
+
+    impl ObjectStore for ClosedStore {
+        fn put(&mut self, _key: &str, _body: &[u8]) -> Result<(), CacheError> {
+            Err(CacheError::Storage {
+                reason: "object store",
+            })
+        }
+        fn head(&self, _key: &str) -> Result<crate::pack::ObjectMeta, CacheError> {
+            Err(CacheError::Storage {
+                reason: "object store",
+            })
+        }
+        fn range_get(&self, _key: &str, _offset: u64, _length: u64) -> Result<Vec<u8>, CacheError> {
+            Err(CacheError::Storage {
+                reason: "object store",
+            })
+        }
+        fn delete(&mut self, _key: &str) -> Result<(), CacheError> {
+            Err(CacheError::Storage {
+                reason: "object store",
+            })
+        }
+    }
+
+    #[test]
+    fn degraded_tenant_refill_does_not_touch_the_store() {
+        let budget = RefillBudget::new(64, 1).unwrap();
+        let store = ClosedStore;
+        for availability in [ColdAvailability::Slow, ColdAvailability::Unavailable] {
+            assert_eq!(
+                refill_for_tenant(
+                    &store,
+                    "ns/manifest",
+                    1,
+                    &budget,
+                    availability,
+                    &synthetic_identity(1, 1),
+                )
+                .unwrap(),
+                RefillOutcome::Missing
+            );
+        }
+    }
 }