Fix cache isolation and admission; optimize CPU hot paths

README.md

@@ -8,7 +8,7 @@ GPUCache does not do erasure coding and does not keep durability replicas. A los
 
 ## Status
 
-**V1 milestones A, B, and C are implemented. The V2 first slice is in this tree. V3 is in progress: a CPU contract for a storage-side BlueField path, plus production hooks that do not need a DPU. DPU acceleration is not measured.**
+**V1 milestones A, B, and C are implemented. The V2 first slice is in this tree. V3 is in progress: a CPU contract for a storage-side BlueField path, plus experimental control and security hooks that do not need a DPU. DPU acceleration is not measured.**
 
 V1 runs the cache contract on CPU. V2 adds a transport trait, a simulated registered-buffer pool, a shard-local directory, and a stricter G4 cold contract. V3 adds an optional `dpu-bf3` feature that does not link DOCA or SPDK, a simulated DRAM-slab storage queue, and CPU-side authz, replay, quota, and cold-ops hooks. The long-term deployment target is a storage-side BlueField-3. The GPU client keeps an ordinary ConnectX. This repository does not link DOCA, CUDA, RDMA, or SPDK, and it does not claim that any of those paths ran. Feature `rdma-nixl` does not link libnixl. Feature `dpu-bf3` does not link a device. An aarch64 build is a CPU build. It is not a BlueField functional test. GDS, DOCA, and BlueField offload are in development / `NOT_RUN_HARDWARE` / optional feature. They are not "already implemented" and not "deployed".
 
@@ -29,7 +29,7 @@ V1 runs the cache contract on CPU. V2 adds a transport trait, a simulated regist
 | V3: one candidate offload (storage queue) and ablation stubs | Interface only. P2P, DPA, compression, and crypto offload stay `NOT_RUN_HARDWARE` when enabled. Hot-tier EC cannot be enabled |
 | V3: authz, replay window, integrity seal, rate limit, secret rotation, audit | CPU hooks in this tree. Not mTLS. Trusted-fabric boundary is [docs/adr/0007-trusted-fabric.md](docs/adr/0007-trusted-fabric.md) |
 | V3: rebalance, rolling upgrade, fault-domain isolation | In-process stubs on the V2 shard directory. They fence fills. They do not copy a replica |
-| V3: degrade matrix (DPU restart, GPU reset, NIC flap, storage loss, NVMe timeout, control-plane loss, RustFS slow/down) | Decisions are clean misses. No payload, no wrong hit, no permanent block. Not a hardware fault injection |
+| V3: degrade matrix (DPU restart, GPU reset, NIC flap, storage loss, NVMe timeout, control-plane loss, RustFS slow/down) | Simulated fault decisions request recompute. Hidden slab bytes remain charged and can exhaust capacity until explicit restart; recovery is not proven on hardware |
 | 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) |
@@ -55,7 +55,7 @@ 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 manifest or the payload range exceeds its 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. V2 adds a bounded async sink and a refill that returns no bytes when the combined manifest and payload downloads exceed its byte budget.
 
 Lifecycle:
 
@@ -65,7 +65,7 @@ ABSENT -> RESERVED -> WRITING -> COMMITTED -> EVICTING -> ABSENT
 
 A failed or timed-out registered region goes to `QUARANTINED`. Lease expiry does not prove that a copy has stopped, so that buffer stays allocated until the owner attests drain or a newer store epoch fences it. Same tokens do not imply the same bytes: the logical prefix key and the physical byte key are different encodings. V1 reuses only within one tenant, one model, and one layout. A reshard, quant, or layout mismatch is a miss. There is no cross-tenant dedup. A longer hash chain can reuse a whole shorter committed block (`PrefixHit`); the service does not slice inside a block. See [docs/adr/0003-partial-prefix.md](docs/adr/0003-partial-prefix.md).
 
-Resource limits cover object size, in-flight requests, memory registrations, per-tenant in-flight bytes, and retained bytes. The idempotency table is bounded.
+Resource limits cover object size, writer admissions, memory registrations, per-tenant in-flight bytes, and retained payload bytes. They are not a complete process-RSS bound: reader pins, protocol reply allocations, and some auxiliary metadata still need admission limits. The idempotency table is bounded and scoped by authenticated tenant and trust domain.
 
 ## What you can run
 

crates/cold/src/sink.rs

@@ -43,6 +43,7 @@ impl ColdJob {
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub struct RefillBudget {
+    /// Total downloaded manifest plus payload bytes per refill.
     pub max_bytes: u64,
     pub max_ranges: u32,
 }
@@ -199,7 +200,7 @@ pub fn refill_generation(
     else {
         return Ok(RefillOutcome::Missing);
     };
-    if range.length == 0 || range.length > budget.max_bytes {
+    if range.length == 0 || range.length > budget.max_bytes.saturating_sub(meta.len) {
         return Ok(RefillOutcome::OverBudget);
     }
     load_range(store, &manifest, range.offset, range.length, range.checksum)
@@ -349,4 +350,38 @@ mod tests {
         assert_eq!(store.manifest_ranges.get(), 0);
         assert_eq!(store.object_ranges.get(), 0);
     }
+    #[test]
+    fn refill_charges_manifest_and_payload_to_one_budget() {
+        let pack = pack_blocks("total-budget", &[block(1, 1, 32)]).unwrap();
+        let mut store = CountStore {
+            inner: MockObjectStore::default(),
+            manifest_ranges: Cell::new(0),
+            object_ranges: Cell::new(0),
+        };
+        store.put(&pack.manifest_key, &pack.manifest_bytes).unwrap();
+        store.put(&pack.object_key, &pack.object_bytes).unwrap();
+        let total = pack.manifest_bytes.len() as u64 + 32;
+        assert_eq!(
+            refill_generation(
+                &store,
+                &pack.manifest_key,
+                1,
+                &RefillBudget::new(total - 1, 1).unwrap()
+            )
+            .unwrap(),
+            RefillOutcome::OverBudget
+        );
+        assert_eq!(store.object_ranges.get(), 0);
+        assert!(matches!(
+            refill_generation(
+                &store,
+                &pack.manifest_key,
+                1,
+                &RefillBudget::new(total, 1).unwrap()
+            )
+            .unwrap(),
+            RefillOutcome::Loaded { .. }
+        ));
+        assert_eq!(store.object_ranges.get(), 1);
+    }
 }

crates/dpu/src/queue.rs

@@ -77,7 +77,7 @@ impl Slab {
         })
     }
 
-    fn insert(&mut self, tenant: &[u8], bytes: &[u8]) -> Result<u64, CacheError> {
+    fn insert(&mut self, tenant: Vec<u8>, bytes: Vec<u8>) -> Result<u64, CacheError> {
         let len = bytes.len() as u64;
         let next = self
             .used
@@ -93,12 +93,13 @@ impl Slab {
         let id = self.next;
         self.next += 1;
         self.used = next;
+        let checksum = content_checksum(&bytes);
         self.slots.insert(
             id,
             Slot {
-                tenant: tenant.to_vec(),
-                bytes: bytes.to_vec(),
-                checksum: content_checksum(bytes),
+                tenant,
+                bytes,
+                checksum,
                 exposed: true,
             },
         );
@@ -146,6 +147,7 @@ enum Pending {
 pub struct StorageQueue {
     slab: Slab,
     pending: BTreeMap<u64, Pending>,
+    pending_bytes: u64,
     next_task: u64,
     max_inflight: usize,
     control_up: bool,
@@ -159,6 +161,7 @@ impl StorageQueue {
         Ok(Self {
             slab: Slab::new(cap_bytes)?,
             pending: BTreeMap::new(),
+            pending_bytes: 0,
             next_task: 1,
             max_inflight,
             control_up: true,
@@ -175,6 +178,13 @@ impl StorageQueue {
 
     pub fn submit_write(&mut self, tenant: &[u8], payload: &[u8]) -> Result<TaskId, CacheError> {
         self.admit(tenant, payload.len())?;
+        let next = self.pending_bytes.saturating_add(payload.len() as u64);
+        if next.saturating_add(self.slab.used) > self.slab.cap {
+            return Err(CacheError::QuotaExceeded {
+                kind: gpucache_core_types::QuotaKind::RetainedBytes,
+            });
+        }
+        self.pending_bytes = next;
         self.enqueue(Pending::Write {
             tenant: tenant.to_vec(),
             payload: payload.to_vec(),
@@ -195,7 +205,8 @@ impl StorageQueue {
         };
         let (slot, payload) = match pending {
             Pending::Write { tenant, payload } => {
-                let id = self.slab.insert(&tenant, &payload)?;
+                self.pending_bytes -= payload.len() as u64;
+                let id = self.slab.insert(tenant, payload)?;
                 (Some(id), None)
             }
             Pending::Read { tenant, slot } => {
@@ -207,8 +218,12 @@ impl StorageQueue {
     }
 
     pub fn fail(&mut self, task: TaskId) -> Result<(), CacheError> {
-        if self.pending.remove(&task.0).is_none() {
-            return Err(CacheError::UnknownGeneration);
+        let pending = self
+            .pending
+            .remove(&task.0)
+            .ok_or(CacheError::UnknownGeneration)?;
+        if let Pending::Write { payload, .. } = pending {
+            self.pending_bytes -= payload.len() as u64;
         }
         Ok(())
     }
@@ -217,21 +232,24 @@ impl StorageQueue {
         match fault {
             Fault::DpuRestart => {
                 self.pending.clear();
+                self.pending_bytes = 0;
                 self.slab.restart();
                 self.control_up = true;
             }
             Fault::GpuReset | Fault::NicFlap | Fault::StorageLoss => {
                 self.pending.clear();
+                self.pending_bytes = 0;
                 self.slab.hide_all();
                 self.control_up = true;
             }
             Fault::NvmeTimeout => {
                 if let Some(id) = self.pending.keys().next().copied() {
-                    self.pending.remove(&id);
+                    let _ = self.fail(TaskId(id));
                 }
             }
             Fault::ControlPlaneLoss => {
                 self.pending.clear();
+                self.pending_bytes = 0;
                 self.slab.hide_all();
                 self.control_up = false;
             }
@@ -438,4 +456,51 @@ mod tests {
             }
         );
     }
+    #[test]
+    fn pending_writes_share_the_slab_budget_and_release_credit_on_failure() {
+        let mut queue = StorageQueue::open_simulated(8, 4).unwrap();
+        let first = queue.submit_write(b"tenant-a", b"12345678").unwrap();
+        assert!(matches!(
+            queue.submit_write(b"tenant-a", b"x"),
+            Err(CacheError::QuotaExceeded { .. })
+        ));
+        assert_eq!(queue.pending_bytes, 8);
+        queue.fail(first).unwrap();
+        assert_eq!(queue.pending_bytes, 0);
+        let first = queue.submit_write(b"tenant-a", b"12345678").unwrap();
+        queue.complete(first).unwrap();
+        assert_eq!(queue.pending_bytes, 0);
+        assert!(matches!(
+            queue.submit_write(b"tenant-a", b"x"),
+            Err(CacheError::QuotaExceeded { .. })
+        ));
+        queue.apply_fault(Fault::DpuRestart);
+        queue.submit_write(b"tenant-a", b"12345678").unwrap();
+    }
+
+    #[test]
+    fn faults_release_pending_credit_but_do_not_free_hidden_slots() {
+        for fault in [
+            Fault::DpuRestart,
+            Fault::GpuReset,
+            Fault::NicFlap,
+            Fault::StorageLoss,
+            Fault::NvmeTimeout,
+            Fault::ControlPlaneLoss,
+        ] {
+            let mut queue = StorageQueue::open_simulated(8, 4).unwrap();
+            let first = queue.submit_write(b"tenant-a", b"1234").unwrap();
+            queue.complete(first).unwrap();
+            queue.submit_write(b"tenant-a", b"5678").unwrap();
+            queue.apply_fault(fault);
+            assert_eq!(queue.pending_bytes, 0);
+            if fault == Fault::ControlPlaneLoss {
+                queue.restore_control();
+            }
+            queue.submit_write(b"tenant-a", b"5678").unwrap();
+            if fault != Fault::DpuRestart {
+                assert!(queue.submit_write(b"tenant-a", b"x").is_err());
+            }
+        }
+    }
 }