diff --git a/README.md b/README.md index 2e63185..a5fb7da 100644 --- a/README.md +++ b/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 diff --git a/crates/cold/src/sink.rs b/crates/cold/src/sink.rs index 942a233..ea979f0 100644 --- a/crates/cold/src/sink.rs +++ b/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); + } } diff --git a/crates/dpu/src/queue.rs b/crates/dpu/src/queue.rs index a363869..64e469d 100644 --- a/crates/dpu/src/queue.rs +++ b/crates/dpu/src/queue.rs @@ -77,7 +77,7 @@ impl Slab { }) } - fn insert(&mut self, tenant: &[u8], bytes: &[u8]) -> Result { + fn insert(&mut self, tenant: Vec, bytes: Vec) -> Result { 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, + 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 { 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()); + } + } + } } diff --git a/crates/storage/src/file.rs b/crates/storage/src/file.rs index 7a9792c..911811c 100644 --- a/crates/storage/src/file.rs +++ b/crates/storage/src/file.rs @@ -6,13 +6,13 @@ //! does not open block devices, format disks, or issue NVMe admin commands. //! A lost or corrupt file is a miss. There is no erasure coding and no replica. -use std::collections::VecDeque; +use std::collections::{HashSet, VecDeque}; use std::fs::{self, File}; use std::io::{Read, Write}; use std::path::{Component, Path, PathBuf}; use gpucache_core_types::{ - verify_checksum, BlockState, ByteReader, ByteWriter, CacheError, QuotaKind, ResourceLimits, + verify_checksum, BlockState, ByteReader, ByteWriter, CacheError, ResourceLimits, }; use gpucache_protocol::{ AuthContext, BatchReply, CommitItem, CommitOutcome, GetHit, GetItem, InvalidateItem, @@ -150,22 +150,24 @@ impl FileStore { now_ms: u64, items: Vec, ) -> Result, CacheError> { - self.ensure_batch(items.len())?; + let ids: Vec<_> = items.iter().map(|item| item.header.request_id).collect(); + self.inner.prepare(auth, now_ms, &ids)?; let mut out = Vec::with_capacity(items.len()); - let mut epoch = self.inner.epoch(); for item in items { - self.ensure_room(auth.tenant.as_bytes(), item.length, now_ms)?; - let reply = self.inner.reserve_put(auth, now_ms, vec![item])?; - epoch = reply.store_epoch; - let Some(result) = reply.items.into_iter().next() else { - return Err(CacheError::Malformed { - reason: "reserve batch", + let root = &self.root; + let lru = &mut self.lru; + let body = self + .inner + .reserve_admitting(auth, now_ms, &item, |inner, extra| { + Self::ensure_room(inner, lru, root, auth.tenant.as_bytes(), extra, now_ms) }); - }; - out.push(result); + out.push(ItemResult { + request_id: item.header.request_id, + body, + }); } Ok(BatchReply { - store_epoch: epoch, + store_epoch: self.inner.epoch(), items: out, }) } @@ -188,6 +190,7 @@ impl FileStore { } Err(err) => { let _ = self.inner.abandon_durable_commit( + auth, outcome.generation, request_id, now_ms, @@ -202,6 +205,7 @@ impl FileStore { }; out.push(ItemResult { request_id, body }); } + let _ = self.write_lru(); Ok(BatchReply { store_epoch: reply.store_epoch, items: out, @@ -251,79 +255,61 @@ impl FileStore { Ok(reply) } - fn ensure_batch(&self, len: usize) -> Result<(), CacheError> { - if len == 0 { - return Err(CacheError::Malformed { - reason: "empty batch", - }); - } - if len > self.limits().max_batch_items as usize { - return Err(CacheError::QuotaExceeded { - kind: QuotaKind::BatchItems, - }); - } - Ok(()) - } - - fn ensure_room(&mut self, tenant: &[u8], extra: u64, now_ms: u64) -> Result<(), CacheError> { - if extra > self.limits().max_retained_bytes { - return Ok(()); - } - loop { - let global = self.inner.stats().retained_bytes; - let tenant_used = self.inner.tenant_retained_bytes(tenant); - let global_ok = global.saturating_add(extra) <= self.limits().max_retained_bytes; - let tenant_ok = - tenant_used.saturating_add(extra) <= self.limits().max_retained_bytes_per_tenant; - if global_ok && tenant_ok { - return Ok(()); - } - let only_tenant = if tenant_ok { None } else { Some(tenant) }; - if !self.evict_oldest(now_ms, only_tenant)? { - return Ok(()); - } - } - } - - fn evict_oldest( - &mut self, + fn ensure_room( + inner: &mut MemoryStore, + lru: &mut VecDeque, + root: &Path, + tenant: &[u8], + extra: u64, now_ms: u64, - only_tenant: Option<&[u8]>, ) -> Result { - let candidates = self.inner.eviction_candidates(now_ms); - let n = self.lru.len(); - let mut deferred = Vec::new(); - let mut freed = false; - for _ in 0..n { - let Some(generation) = self.lru.pop_front() else { - break; - }; - let Some(candidate) = candidates.iter().find(|item| item.generation == generation) - else { - continue; - }; - let wrong_tenant = - only_tenant.is_some_and(|tenant| tenant != candidate.tenant.as_slice()); - if candidate.pinned || wrong_tenant { - deferred.push(generation); - continue; - } - let state = self.inner.force_evict_committed(generation, now_ms)?; - self.delete_block(generation); - if state != BlockState::Absent { - self.lru.push_back(generation); - continue; - } - freed = true; - break; - } - for generation in deferred { - self.lru.push_back(generation); + let limits = inner.limits(); + if extra > limits.max_retained_bytes || extra > limits.max_retained_bytes_per_tenant { + return Ok(false); } - if freed { - let _ = self.write_lru(); + // One snapshot and index per admission, rather than rescanning all + // slots and linearly searching the snapshot for every victim. + let candidates: std::collections::HashMap<_, _> = inner + .eviction_candidates(now_ms) + .into_iter() + .map(|candidate| (candidate.generation, candidate)) + .collect(); + let mut global = inner.stats().retained_bytes; + let mut tenant_used = inner.tenant_retained_bytes(tenant); + // If the tenant limit is tight, reclaim from that tenant first. Keep + // skipped entries in their original relative order. + for tenant_only in [true, false] { + let mut kept = VecDeque::with_capacity(lru.len()); + while let Some(generation) = lru.pop_front() { + let global_ok = global.saturating_add(extra) <= limits.max_retained_bytes; + let tenant_ok = + tenant_used.saturating_add(extra) <= limits.max_retained_bytes_per_tenant; + let needs_room = if tenant_only { + !tenant_ok + } else { + !global_ok && tenant_ok + }; + let Some(candidate) = candidates.get(&generation) else { + continue; + }; + let same_tenant = candidate.tenant.as_slice() == tenant; + if !needs_room || candidate.pinned || (tenant_only && !same_tenant) { + kept.push_back(generation); + continue; + } + if inner.force_evict_prepared(generation)? == BlockState::Absent { + let _ = fs::remove_file(block_path(root, generation)); + global = global.saturating_sub(candidate.length); + if same_tenant { + tenant_used = tenant_used.saturating_sub(candidate.length); + } + } else { + kept.push_back(generation); + } + } + *lru = kept; } - Ok(freed) + Ok(true) } fn persist_committed(&mut self, generation: u64) -> Result<(), CacheError> { @@ -345,7 +331,6 @@ impl FileStore { fn touch(&mut self, generation: u64) { self.lru.retain(|existing| *existing != generation); self.lru.push_back(generation); - let _ = self.write_lru(); } fn delete_block(&self, generation: u64) { @@ -514,17 +499,19 @@ fn load_blocks( fn read_lru(root: &Path, loaded: &[u64]) -> VecDeque { let path = root.join("LRU"); - let bytes = fs::read(&path).unwrap_or_default(); + let bytes = read_capped(&path, 10 + 1_000_000 * 8).unwrap_or_default(); + let loaded_set: HashSet<_> = loaded.iter().copied().collect(); + let mut seen = HashSet::new(); let mut order = VecDeque::new(); if let Some(gens) = decode_lru(&bytes) { for generation in gens { - if loaded.contains(&generation) && !order.contains(&generation) { + if loaded_set.contains(&generation) && seen.insert(generation) { order.push_back(generation); } } } for generation in loaded { - if !order.contains(generation) { + if seen.insert(*generation) { order.push_back(*generation); } } diff --git a/crates/storage/src/memory.rs b/crates/storage/src/memory.rs index 5782b32..b9930f3 100644 --- a/crates/storage/src/memory.rs +++ b/crates/storage/src/memory.rs @@ -124,7 +124,7 @@ pub struct MemoryStore { slots: HashMap, by_physical: HashMap, u64>, logical: HashMap, Vec>>, - idem: HashMap, + idem: HashMap<(Vec, Vec, RequestId), IdemRecord>, next_generation: u64, next_region: u64, epoch: u64, @@ -264,7 +264,7 @@ impl MemoryStore { &generation.to_le_bytes(), &observed_epoch.to_le_bytes(), ]); - if let Some(prev) = self.replay(request_id, &fp)? { + if let Some(prev) = self.replay(auth, request_id, &fp)? { return match prev { IdemKind::Release(result) => result, _ => Err(CacheError::IdempotencyConflict), @@ -281,7 +281,13 @@ impl MemoryStore { self.ensure_idem_room(now_ms)?; let result = self.reclaim_fenced_inner(auth, epoch, deadline_ms, generation, observed_epoch, now_ms); - self.remember(request_id, fp, now_ms, IdemKind::Release(result.clone()))?; + self.remember( + auth, + request_id, + fp, + now_ms, + IdemKind::Release(result.clone()), + )?; result } @@ -514,6 +520,7 @@ impl MemoryStore { /// and replaying the same request id must not report success. pub(crate) fn abandon_durable_commit( &mut self, + auth: &AuthContext, generation: u64, request_id: RequestId, now_ms: u64, @@ -522,7 +529,7 @@ impl MemoryStore { if state != BlockState::Absent { return Err(CacheError::InUse { refs: 1 }); } - self.idem.remove(&request_id); + self.idem.remove(&idem_key(auth, request_id)); Ok(()) } @@ -573,6 +580,15 @@ impl MemoryStore { now_ms: u64, ) -> Result { self.poll(now_ms); + self.force_evict_prepared(generation) + } + + /// Evict after a caller has polled this batch's logical time. FileStore + /// uses this under exclusive access after taking its candidate snapshot. + pub(crate) fn force_evict_prepared( + &mut self, + generation: u64, + ) -> Result { let slot = self .slots .get(&generation) @@ -626,9 +642,9 @@ impl MemoryStore { if !layout_compatible(&slot.identity, query) { continue; } - let payload = slot.bytes.clone().unwrap_or_default(); + let payload = slot.bytes.as_deref().unwrap_or_default(); if payload.len() as u64 != slot.length - || verify_checksum(&payload, slot.checksum).is_err() + || verify_checksum(payload, slot.checksum).is_err() { corrupt.push(generation); continue; @@ -665,7 +681,7 @@ impl MemoryStore { } } - fn prepare( + pub(crate) fn prepare( &mut self, auth: &AuthContext, now_ms: u64, @@ -760,9 +776,21 @@ impl MemoryStore { auth: &AuthContext, now_ms: u64, item: &ReservePutItem, + ) -> Result { + self.reserve_admitting(auth, now_ms, item, |_, _| Ok(false)) + } + + /// Reclaim only after validation and an actual retained-byte admission + /// failure. Replays and continuations never request extra retained bytes. + pub(crate) fn reserve_admitting( + &mut self, + auth: &AuthContext, + now_ms: u64, + item: &ReservePutItem, + reclaim: impl FnOnce(&mut Self, u64) -> Result, ) -> Result { let fp = reserve_fingerprint(item)?; - if let Some(prev) = self.replay(item.header.request_id, &fp)? { + if let Some(prev) = self.replay(auth, item.header.request_id, &fp)? { return match prev { IdemKind::Reserve(result) => result, _ => Err(CacheError::IdempotencyConflict), @@ -770,8 +798,20 @@ impl MemoryStore { } self.preflight(&item.header, now_ms)?; self.ensure_idem_room(now_ms)?; - let result = self.reserve_new(auth, now_ms, item); + let mut result = self.reserve_new(auth, now_ms, item); + if item.generation == 0 + && matches!( + result, + Err(CacheError::QuotaExceeded { + kind: QuotaKind::RetainedBytes | QuotaKind::TenantRetainedBytes + }) + ) + && reclaim(self, item.length)? + { + result = self.reserve_new(auth, now_ms, item); + } self.remember( + auth, item.header.request_id, fp, now_ms, @@ -937,7 +977,7 @@ impl MemoryStore { &item.length.to_le_bytes(), &item.checksum.0, ]); - if let Some(prev) = self.replay(item.header.request_id, &fp)? { + if let Some(prev) = self.replay(auth, item.header.request_id, &fp)? { return match prev { IdemKind::Commit(result) => result, _ => Err(CacheError::IdempotencyConflict), @@ -947,6 +987,7 @@ impl MemoryStore { self.ensure_idem_room(now_ms)?; let result = self.commit_new(auth, now_ms, item); self.remember( + auth, item.header.request_id, fp, now_ms, @@ -975,7 +1016,7 @@ impl MemoryStore { slot.fail, slot.length, slot.checksum, - slot.bytes.clone().unwrap_or_default(), + slot.bytes.as_deref().unwrap_or_default(), ) }; if reserve_epoch != self.epoch { @@ -996,7 +1037,7 @@ impl MemoryStore { actual: item.length, }); } - if item.checksum != declared_sum || verify_checksum(&payload, declared_sum).is_err() { + if item.checksum != declared_sum || verify_checksum(payload, declared_sum).is_err() { self.reject_write(generation)?; return Err(CacheError::ChecksumMismatch); } @@ -1087,7 +1128,7 @@ impl MemoryStore { item: &ReleaseCancelItem, ) -> Result { let fp = release_fingerprint(item); - if let Some(prev) = self.replay(item.header.request_id, &fp)? { + if let Some(prev) = self.replay(auth, item.header.request_id, &fp)? { return match prev { IdemKind::Release(result) => result, _ => Err(CacheError::IdempotencyConflict), @@ -1097,6 +1138,7 @@ impl MemoryStore { self.ensure_idem_room(now_ms)?; let result = self.release_new(auth, now_ms, item); self.remember( + auth, item.header.request_id, fp, now_ms, @@ -1180,7 +1222,7 @@ impl MemoryStore { &item.header.epoch.to_le_bytes(), &item.generation.to_le_bytes(), ]); - if let Some(prev) = self.replay(item.header.request_id, &fp)? { + if let Some(prev) = self.replay(auth, item.header.request_id, &fp)? { return match prev { IdemKind::Invalidate(result) => result, _ => Err(CacheError::IdempotencyConflict), @@ -1190,6 +1232,7 @@ impl MemoryStore { self.ensure_idem_room(now_ms)?; let result = self.invalidate_new(auth, now_ms, item); self.remember( + auth, item.header.request_id, fp, now_ms, @@ -1483,27 +1526,27 @@ impl MemoryStore { self.tenant_inflight.get(&tenant).copied().unwrap_or(0), inflight_delta, )?; - if target.retained && !current.retained { - if new_retained > self.limits.max_retained_bytes { + if target.inflight && !current.inflight { + if new_requests > u64::from(self.limits.max_in_flight_requests) { return Err(CacheError::QuotaExceeded { - kind: QuotaKind::RetainedBytes, + kind: QuotaKind::InFlightRequests, }); } - if new_tenant_retained > self.limits.max_retained_bytes_per_tenant { + if new_tenant_inflight > self.limits.max_in_flight_bytes_per_tenant { return Err(CacheError::QuotaExceeded { - kind: QuotaKind::TenantRetainedBytes, + kind: QuotaKind::InFlightBytes, }); } } - if target.inflight && !current.inflight { - if new_requests > u64::from(self.limits.max_in_flight_requests) { + if target.retained && !current.retained { + if new_retained > self.limits.max_retained_bytes { return Err(CacheError::QuotaExceeded { - kind: QuotaKind::InFlightRequests, + kind: QuotaKind::RetainedBytes, }); } - if new_tenant_inflight > self.limits.max_in_flight_bytes_per_tenant { + if new_tenant_retained > self.limits.max_retained_bytes_per_tenant { return Err(CacheError::QuotaExceeded { - kind: QuotaKind::InFlightBytes, + kind: QuotaKind::TenantRetainedBytes, }); } } @@ -1536,10 +1579,11 @@ impl MemoryStore { fn replay( &self, + auth: &AuthContext, id: RequestId, fingerprint: &Checksum, ) -> Result, CacheError> { - match self.idem.get(&id) { + match self.idem.get(&idem_key(auth, id)) { None => Ok(None), Some(record) if &record.fingerprint == fingerprint => Ok(Some(record.kind.clone())), Some(_) => Err(CacheError::IdempotencyConflict), @@ -1558,13 +1602,14 @@ impl MemoryStore { fn remember( &mut self, + auth: &AuthContext, id: RequestId, fingerprint: Checksum, now_ms: u64, kind: IdemKind, ) -> Result<(), CacheError> { self.idem.insert( - id, + idem_key(auth, id), IdemRecord { fingerprint, inserted_ms: now_ms, @@ -1581,6 +1626,14 @@ impl MemoryStore { } } +fn idem_key(auth: &AuthContext, id: RequestId) -> (Vec, Vec, RequestId) { + ( + auth.tenant.as_bytes().to_vec(), + auth.trust_domain.as_bytes().to_vec(), + id, + ) +} + fn layout_compatible(stored: &CacheIdentity, query: &CacheIdentity) -> bool { stored.version == query.version && stored.model_revision == query.model_revision diff --git a/crates/storage/src/tests.rs b/crates/storage/src/tests.rs index 7ea09b5..fefac4f 100644 --- a/crates/storage/src/tests.rs +++ b/crates/storage/src/tests.rs @@ -839,3 +839,31 @@ fn hex(bytes: &[u8]) -> String { } out } + +#[test] +fn replay_state_is_scoped_to_the_authenticated_tenant() { + let mut other = synthetic_identity(1, 1); + other.tenant = TenantId::new(b"tenant-b".to_vec()).unwrap(); + let policy = TenantPolicy::new(other.tenant.clone(), other.trust_domain.clone(), FP).unwrap(); + let other_auth = + AuthContext::new(other.tenant.clone(), other.trust_domain.clone(), FP).unwrap(); + let (mut store, auth) = open(Some(policy)); + let generation = commit_bytes(&mut store, &auth, 10, b"payload"); + let denied = store + .commit( + &other_auth, + 0, + vec![CommitItem { + header: hdr(&store, 11, 0), + generation, + length: 7, + checksum: content_checksum(b"payload"), + }], + ) + .unwrap(); + assert_eq!(denied.items[0].body, Err(CacheError::Unauthorized)); + let mut request = put_item(&store, 10, 0, b"other", 1); + request.identity = other; + let independent = store.reserve_put(&other_auth, 0, vec![request]).unwrap(); + assert!(independent.items[0].body.is_ok()); +} diff --git a/crates/storage/tests/cross_process.rs b/crates/storage/tests/cross_process.rs index 6b10217..64de139 100644 --- a/crates/storage/tests/cross_process.rs +++ b/crates/storage/tests/cross_process.rs @@ -64,20 +64,23 @@ fn cross_process_corrupt_frame_is_rejected() { assert!(stderr.contains("checksum"), "stderr={stderr}"); } -fn xfer_bin() -> String { - std::env::var("CARGO_BIN_EXE_kv_block_xfer").unwrap_or_else(|_| { - let mut path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")); - path.pop(); - path.pop(); - path.push("target"); - path.push("debug"); - path.push("kv-block-xfer"); - path.to_string_lossy().into_owned() - }) -} - fn run_import(frame: &[u8]) -> std::process::Output { - let mut child = Command::new(xfer_bin()) + // Cargo supplies the binary for the selected target/profile at compile + // time. Subprocesses do not inherit Cargo's target runner automatically. + let binary = env!("CARGO_BIN_EXE_kv-block-xfer"); + // CI's runner is an executable plus whitespace-separated arguments; + // reuse it without invoking a shell or relying on binfmt registration. + let runner = std::env::var("CARGO_TARGET_AARCH64_UNKNOWN_LINUX_GNU_RUNNER").ok(); + let mut command = match runner.as_deref() { + Some(runner) => { + let mut parts = runner.split_whitespace(); + let mut command = Command::new(parts.next().expect("nonempty target runner")); + command.args(parts).arg(binary); + command + } + None => Command::new(binary), + }; + let mut child = command .arg("import") .arg(FP_HEX) .stdin(Stdio::piped()) @@ -85,13 +88,15 @@ fn run_import(frame: &[u8]) -> std::process::Output { .stderr(Stdio::piped()) .spawn() .expect("spawn kv-block-xfer"); - child - .stdin - .as_mut() - .expect("stdin") - .write_all(frame) - .expect("write frame"); - child.wait_with_output().expect("wait") + let wrote = child.stdin.as_mut().expect("stdin").write_all(frame); + let output = child.wait_with_output().expect("wait"); + assert!( + wrote.is_ok(), + "write frame: {wrote:?}; status={}; stderr={}", + output.status, + String::from_utf8_lossy(&output.stderr) + ); + output } fn hex(bytes: &[u8]) -> String { diff --git a/crates/storage/tests/file_backend.rs b/crates/storage/tests/file_backend.rs index 35d778a..7d90fbf 100644 --- a/crates/storage/tests/file_backend.rs +++ b/crates/storage/tests/file_backend.rs @@ -424,3 +424,109 @@ fn tenant_quota_does_not_evict_another_tenant() { Ok(LookupOutcome::Hit { .. }) )); } + +#[test] +fn rejected_reservations_do_not_evict_committed_data() { + for case in 0..7 { + let dir = TempDir::new(); + let (policy, auth) = policy(1, FP_A); + let mut tight = limits(32, 16); + tight.max_object_bytes = 16; + let mut store = FileStore::open(&dir.path, tight, vec![policy]).unwrap(); + let identity = synthetic_identity(1, 1); + let generation = commit_block(&mut store, &auth, 10, &block(1), identity.clone()); + let mut other = identity.clone(); + other.prefix_chain[0].0[0] ^= 0xff; + let mut request = put(store.epoch(), 20, &block(2), other); + let mut caller = auth.clone(); + match case { + 0 => caller.credential_fingerprint = FP_B, + 1 => request.header.epoch += 1, + 2 => request.header.deadline_ms = 1, + 3 => request.length = 17, + 4 => request.payload = Some(vec![2; 15]), + 5 => request.identity.tenant = TenantId::new(b"other".to_vec()).unwrap(), + 6 => request.identity = identity.clone(), + _ => unreachable!(), + } + let _ = store.reserve_put(&caller, 2, vec![request]); + assert!(store.committed_path(generation).exists(), "case {case}"); + let reply = store + .lookup( + &auth, + 2, + vec![LookupItem { + header: header(store.epoch(), 30), + identity, + }], + ) + .unwrap(); + assert!( + matches!(reply.items[0].body, Ok(LookupOutcome::Hit { .. })), + "case {case}" + ); + } +} + +#[test] +fn replayed_reservation_does_not_evict_its_own_commit() { + let dir = TempDir::new(); + let (policy, auth) = policy(1, FP_A); + let mut store = FileStore::open(&dir.path, limits(32, 16), vec![policy]).unwrap(); + let identity = synthetic_identity(1, 1); + let generation = commit_block(&mut store, &auth, 10, &block(1), identity.clone()); + let reply = store + .reserve_put(&auth, 0, vec![put(store.epoch(), 10, &block(1), identity)]) + .unwrap(); + assert_eq!(reply.items[0].body.as_ref().unwrap().generation, generation); + assert!(store.committed_path(generation).exists()); +} + +#[test] +fn duplicate_reservation_batch_is_rejected_before_mutation() { + let dir = TempDir::new(); + let (policy, auth) = policy(1, FP_A); + let mut store = FileStore::open(&dir.path, limits(64, 64), vec![policy]).unwrap(); + let request = put(store.epoch(), 10, &block(1), synthetic_identity(1, 1)); + assert!(matches!( + store.reserve_put(&auth, 0, vec![request.clone(), request]), + Err(CacheError::Malformed { + reason: "duplicate request_id" + }) + )); + assert_eq!(store.stats().retained_bytes, 0); +} + +#[test] +fn lookup_and_get_keep_recency_in_memory_without_writing_lru() { + let dir = TempDir::new(); + let (policy, auth) = policy(1, FP_A); + let mut store = FileStore::open(&dir.path, limits(64, 64), vec![policy]).unwrap(); + let identity = synthetic_identity(1, 1); + let generation = commit_block(&mut store, &auth, 10, &block(1), identity.clone()); + let sentinel = b"do not rewrite advisory recency on reads"; + fs::write(dir.path.join("LRU"), sentinel).unwrap(); + store + .lookup( + &auth, + 0, + vec![LookupItem { + header: header(store.epoch(), 20), + identity: identity.clone(), + }], + ) + .unwrap(); + store + .get( + &auth, + 0, + vec![GetItem { + header: header(store.epoch(), 21), + generation, + identity, + pin_ttl_ms: 100, + }], + ) + .unwrap(); + assert_eq!(fs::read(dir.path.join("LRU")).unwrap(), sentinel); +} diff --git a/crates/transport/examples/cpu_audit.rs b/crates/transport/examples/cpu_audit.rs new file mode 100644 index 0000000..206e33b --- /dev/null +++ b/crates/transport/examples/cpu_audit.rs @@ -0,0 +1,137 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 +//! Synthetic CPU microbenchmarks, not inference or NIC performance. +use std::hint::black_box; +use std::time::Instant; + +use gpucache_core_types::{content_checksum, synthetic_identity, RequestId, ResourceLimits}; +use gpucache_protocol::{ + AuthContext, CommitItem, ItemHeader, LookupItem, ReservePutItem, TenantPolicy, +}; +use gpucache_storage::FileStore; +use gpucache_transport::{Admit, FillRequest, ShardDirectory, ShardParams}; + +fn header(id: u128) -> ItemHeader { + ItemHeader { + request_id: RequestId::from_counter(id).unwrap(), + epoch: 1, + deadline_ms: 1_000_000, + } +} + +fn main() { + println!("scope=CPU_SYNTHETIC hardware=NOT_RUN_HARDWARE tco=TCO_UNKNOWN"); + let root = std::env::temp_dir().join(format!("gpucache-audit-bench-{}", std::process::id())); + std::fs::create_dir(&root).expect("fresh benchmark directory"); + let identity = synthetic_identity(1, 1); + let auth = AuthContext::new( + identity.tenant.clone(), + identity.trust_domain.clone(), + [9; 32], + ) + .unwrap(); + let policy = TenantPolicy::new( + identity.tenant.clone(), + identity.trust_domain.clone(), + [9; 32], + ) + .unwrap(); + let mut store = FileStore::open(&root, ResourceLimits::dev_defaults(), vec![policy]).unwrap(); + let checksum = content_checksum(b"synthetic"); + let mut identities = Vec::new(); + for n in 0..128u128 { + let mut id = identity.clone(); + id.prefix_chain[0].0[..16].copy_from_slice(&n.to_le_bytes()); + let reserved = store + .reserve_put( + &auth, + 0, + vec![ReservePutItem { + header: header(2 * n + 1), + identity: id.clone(), + generation: 0, + length: 9, + checksum, + payload: Some(b"synthetic".to_vec()), + lease_ttl_ms: 1000, + }], + ) + .unwrap(); + let generation = reserved.items[0].body.as_ref().unwrap().generation; + store + .commit( + &auth, + 0, + vec![CommitItem { + header: header(2 * n + 2), + generation, + length: 9, + checksum, + }], + ) + .unwrap() + .items[0] + .body + .as_ref() + .unwrap(); + identities.push(id); + } + let start = Instant::now(); + for n in 0..512 { + black_box( + store + .lookup( + &auth, + 0, + vec![LookupItem { + header: header(10_000 + n as u128), + identity: identities[n % 128].clone(), + }], + ) + .unwrap(), + ); + } + println!( + "workload=file_lookup blocks=128 operations=512 elapsed_ns={}", + start.elapsed().as_nanos() + ); + drop(store); + std::fs::remove_dir_all(root).unwrap(); + + let mut dir = + ShardDirectory::single_node(ShardParams::new(1_000_000, 1_000_000, 8192, 4, 4).unwrap()); + let requests: Vec<_> = (1..=8192u128) + .map(|n| FillRequest { + physical_key: n.to_le_bytes().to_vec(), + tenant: b"tenant-a".to_vec(), + bytes: 1, + request: n.to_le_bytes(), + epoch: 1, + }) + .collect(); + let mut ids = Vec::new(); + for req in &requests { + let Admit::Leader { flight_id, .. } = dir.begin_or_join(req).unwrap() else { + panic!("leader") + }; + ids.push(flight_id); + } + let start = Instant::now(); + for n in 0..8192 { + black_box(dir.begin_or_join(&requests[n * 7919 % 8192]).unwrap()); + } + println!( + "workload=directory_replay flights=8192 operations=8192 elapsed_ns={}", + start.elapsed().as_nanos() + ); + let start = Instant::now(); + for n in 0..8192 { + dir.complete_flight(ids[n], 1, requests[n].request).unwrap(); + } + println!( + "workload=directory_complete flights=8192 operations=8192 elapsed_ns={}", + start.elapsed().as_nanos() + ); + assert_eq!(dir.inflight(), 0); + assert_eq!(dir.credits_used(), 0); +} diff --git a/crates/transport/src/client.rs b/crates/transport/src/client.rs index 64285c9..08a08b0 100644 --- a/crates/transport/src/client.rs +++ b/crates/transport/src/client.rs @@ -24,6 +24,7 @@ pub struct TcpClient { max_frame: usize, epoch: u64, clock: Clock, + usable: bool, } impl TcpClient { @@ -38,9 +39,10 @@ impl TcpClient { reason: "server config", }); } - let mut stream = TcpStream::connect(addr).map_err(|_| CacheError::Storage { - reason: "transport io", - })?; + let mut stream = + TcpStream::connect_timeout(&addr, timeout).map_err(|_| CacheError::Storage { + reason: "transport io", + })?; stream .set_read_timeout(Some(timeout)) .map_err(|_| CacheError::Storage { @@ -56,7 +58,7 @@ impl TcpClient { let hello = encode_hello(&[WIRE_VERSION])?; write_envelope(&mut stream, 0, &hello, max_frame)?; let envelope = read_envelope(&mut stream, pool.checkout()?, max_frame)?; - let (version, epoch) = decode_hello_ack(&envelope.body)?; + let (version, epoch) = decode_hello_ack(envelope.body())?; if version != WIRE_VERSION { return Err(CacheError::WireVersionUnsupported { local: WIRE_VERSION, @@ -70,6 +72,7 @@ impl TcpClient { max_frame, epoch, clock: Clock::new(envelope.server_now), + usable: true, }) } @@ -143,13 +146,96 @@ impl TcpClient { request: RequestBody, map: impl FnOnce(ReplyBody) -> Result, ) -> Result { + if !self.usable { + return Err(CacheError::Storage { + reason: "transport io", + }); + } let bytes = encode_request(&self.auth, &request)?; + // No request correlation exists at the envelope layer. After an + // incomplete exchange, a late reply must never answer a later call. + self.usable = false; write_envelope(&mut self.stream, 0, &bytes, self.max_frame)?; let envelope = read_envelope(&mut self.stream, self.pool.checkout()?, self.max_frame)?; self.clock.observe(envelope.server_now); - match decode_reply(&envelope.body)? { - ReplyBody::Failed { error, .. } => Err(error), - other => map(other), + match decode_reply(envelope.body())? { + ReplyBody::Failed { error, .. } => { + self.usable = true; + Err(error) + } + other => { + let result = map(other); + if result.is_ok() { + self.usable = true; + } + result + } } } } + +#[cfg(test)] +mod tests { + use super::*; + use gpucache_core_types::{synthetic_identity, RequestId}; + use gpucache_protocol::{encode_hello_ack, encode_reply, ItemHeader, ItemResult}; + use std::net::TcpListener; + use std::sync::mpsc; + + #[test] + fn timed_out_connection_cannot_consume_a_late_reply_for_another_call() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let (release, wait) = mpsc::channel(); + let server = std::thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + let pool = Arc::new(Pool::new(1)); + drop(read_envelope(&mut stream, pool.checkout().unwrap(), 4096).unwrap()); + write_envelope(&mut stream, 1, &encode_hello_ack(WIRE_VERSION, 1), 4096).unwrap(); + drop(read_envelope(&mut stream, pool.checkout().unwrap(), 4096).unwrap()); + wait.recv().unwrap(); + let body = encode_reply(&ReplyBody::Lookup(BatchReply { + store_epoch: 1, + items: vec![ItemResult { + request_id: RequestId::from_counter(1).unwrap(), + body: Ok(LookupOutcome::Miss { + reason: gpucache_protocol::MissReason::Absent, + }), + }], + })) + .unwrap(); + write_envelope(&mut stream, 1, &body, 4096).unwrap(); + }); + let identity = synthetic_identity(1, 1); + let auth = AuthContext::new( + identity.tenant.clone(), + identity.trust_domain.clone(), + [9; 32], + ) + .unwrap(); + let mut client = TcpClient::connect(addr, auth, Duration::from_secs(5), 4096).unwrap(); + client + .stream + .set_read_timeout(Some(Duration::from_millis(20))) + .unwrap(); + let item = LookupItem { + header: ItemHeader { + request_id: RequestId::from_counter(1).unwrap(), + epoch: 1, + deadline_ms: 1000, + }, + identity, + }; + assert_eq!(client.lookup(vec![item.clone()]), Err(CacheError::TimedOut)); + release.send(()).unwrap(); + server.join().unwrap(); + let mut next = item; + next.header.request_id = RequestId::from_counter(2).unwrap(); + assert!(matches!( + client.lookup(vec![next]), + Err(CacheError::Storage { + reason: "transport io" + }) + )); + } +} diff --git a/crates/transport/src/directory.rs b/crates/transport/src/directory.rs index dd28b64..9baf4b9 100644 --- a/crates/transport/src/directory.rs +++ b/crates/transport/src/directory.rs @@ -88,6 +88,8 @@ pub struct ShardDirectory { epoch: u64, members: Vec, flights: BTreeMap<[u8; 32], Flight>, + by_request: BTreeMap<(Vec, [u8; 16]), [u8; 32]>, + by_id: BTreeMap, credits_used: u64, credit_bytes: u64, tenant_used: BTreeMap, u64>, @@ -114,6 +116,8 @@ impl ShardDirectory { incarnation: 1, }], flights: BTreeMap::new(), + by_request: BTreeMap::new(), + by_id: BTreeMap::new(), credits_used: 0, credit_bytes: params.credit_bytes, tenant_used: BTreeMap::new(), @@ -177,6 +181,9 @@ impl ShardDirectory { if self.members[pos].incarnation == incarnation { return Ok(self.epoch); } + if incarnation < self.members[pos].incarnation { + return Err(CacheError::FencingRejected { reason: "member" }); + } self.members[pos].incarnation = incarnation; return Ok(self.fence_inflight()); } @@ -213,11 +220,15 @@ impl ShardDirectory { self.check_epoch(req.epoch)?; validate_fill(req)?; let key = flight_key(&req.physical_key); - if let Some(existing) = self.key_for_request(req.request) { + let request_key = (req.tenant.clone(), req.request); + if let Some(existing) = self.by_request.get(&request_key).copied() { if existing != key { return Err(CacheError::IdempotencyConflict); } let flight = &self.flights[&existing]; + if flight.bytes != req.bytes { + return Err(CacheError::IdempotencyConflict); + } if flight.leader == req.request { return Ok(Admit::Leader { flight_id: flight.id, @@ -233,12 +244,16 @@ impl ShardDirectory { if flight.tenant.as_slice() != req.tenant.as_slice() { return Err(CacheError::Unauthorized); } + if flight.bytes != req.bytes { + return Err(CacheError::IdempotencyConflict); + } if flight.waiters.len() >= self.max_waiters { return Err(CacheError::QuotaExceeded { kind: QuotaKind::InFlightRequests, }); } flight.waiters.insert(req.request); + self.by_request.insert(request_key, key); return Ok(Admit::Joined { flight_id: flight.id, epoch: flight.epoch, @@ -252,6 +267,8 @@ impl ShardDirectory { self.reserve(&req.tenant, req.bytes)?; let id = self.next_flight; self.next_flight += 1; + self.by_request.insert(request_key, key); + self.by_id.insert(id, key); self.flights.insert( key, Flight { @@ -294,12 +311,7 @@ impl ShardDirectory { request: [u8; 16], ) -> Result<(), CacheError> { self.check_epoch(request_epoch)?; - let Some(key) = self - .flights - .iter() - .find(|(_, flight)| flight.id == flight_id) - .map(|(key, _)| *key) - else { + let Some(key) = self.by_id.get(&flight_id).copied() else { return Err(CacheError::UnknownGeneration); }; let flight = &self.flights[&key]; @@ -308,12 +320,22 @@ impl ShardDirectory { } let tenant = flight.tenant.clone(); let bytes = flight.bytes; - self.flights.remove(&key); + let flight = self + .flights + .remove(&key) + .ok_or(CacheError::UnknownGeneration)?; + self.by_id.remove(&flight_id); + self.by_request.remove(&(tenant.clone(), flight.leader)); + for waiter in flight.waiters { + self.by_request.remove(&(tenant.clone(), waiter)); + } self.release(&tenant, bytes) } fn fence_inflight(&mut self) -> u64 { self.epoch = self.epoch.saturating_add(1); + self.by_request.clear(); + self.by_id.clear(); let flights = std::mem::take(&mut self.flights); for (_, flight) in flights { let _ = self.release(&flight.tenant, flight.bytes); @@ -363,16 +385,6 @@ impl ShardDirectory { .ok_or(CacheError::Storage { reason: "credits" })?; Ok(()) } - - fn key_for_request(&self, request: [u8; 16]) -> Option<[u8; 32]> { - self.flights.iter().find_map(|(key, flight)| { - if flight.leader == request || flight.waiters.contains(&request) { - Some(*key) - } else { - None - } - }) - } } fn flight_key(physical_key: &[u8]) -> [u8; 32] { @@ -564,4 +576,69 @@ mod tests { assert_eq!(dir.inflight(), 1); assert_eq!(dir.credits_used(), 10); } + #[test] + fn tenant_and_size_are_checked_before_replay_or_join() { + let mut dir = ShardDirectory::single_node(params()); + dir.begin_or_join(&fill(b"a", b"tenant-a", 10, 1, 1)) + .unwrap(); + assert_eq!( + dir.begin_or_join(&fill(b"a", b"tenant-b", 10, 1, 1)), + Err(CacheError::Unauthorized) + ); + assert_eq!( + dir.begin_or_join(&fill(b"a", b"tenant-a", 20, 1, 1)), + Err(CacheError::IdempotencyConflict) + ); + assert_eq!( + dir.begin_or_join(&fill(b"a", b"tenant-a", 20, 2, 1)), + Err(CacheError::IdempotencyConflict) + ); + dir.begin_or_join(&fill(b"b", b"tenant-b", 10, 1, 1)) + .unwrap(); + assert_eq!(dir.credits_used(), 20); + } + + #[test] + fn indexes_are_reclaimed_on_completion_failure_and_fencing() { + let mut dir = ShardDirectory::single_node(params()); + for finish in 0..3 { + let epoch = dir.epoch(); + let Admit::Leader { flight_id, .. } = dir + .begin_or_join(&fill(b"a", b"tenant-a", 10, 1, epoch)) + .unwrap() + else { + panic!("leader") + }; + dir.begin_or_join(&fill(b"a", b"tenant-a", 10, 2, epoch)) + .unwrap(); + match finish { + 0 => dir + .complete_flight(flight_id, epoch, 1u128.to_le_bytes()) + .unwrap(), + 1 => dir + .fail_flight(flight_id, epoch, 1u128.to_le_bytes()) + .unwrap(), + _ => { + dir.fence_epoch(); + } + } + assert!(dir.by_request.is_empty()); + assert!(dir.by_id.is_empty()); + assert_eq!(dir.credits_used(), 0); + } + } + + #[test] + fn stale_membership_does_not_fence_current_work() { + let mut dir = ShardDirectory::single_node(params()); + let epoch = dir.note_member(1, 2).unwrap(); + dir.begin_or_join(&fill(b"a", b"tenant-a", 10, 1, epoch)) + .unwrap(); + assert_eq!( + dir.note_member(1, 1), + Err(CacheError::FencingRejected { reason: "member" }) + ); + assert_eq!(dir.epoch(), epoch); + assert_eq!(dir.inflight(), 1); + } } diff --git a/crates/transport/src/frame.rs b/crates/transport/src/frame.rs index 04894fe..58ce8ad 100644 --- a/crates/transport/src/frame.rs +++ b/crates/transport/src/frame.rs @@ -1,7 +1,7 @@ // Copyright 2024-2026 RustFS, Inc. // SPDX-License-Identifier: Apache-2.0 -use std::io::{Read, Write}; +use std::io::{IoSlice, Read, Write}; use std::net::TcpStream; use gpucache_core_types::CacheError; @@ -10,7 +10,13 @@ use crate::pool::Pooled; pub struct Envelope { pub server_now: u64, - pub body: Vec, + body: Pooled, +} + +impl Envelope { + pub fn body(&self) -> &[u8] { + self.body.as_slice() + } } pub fn write_envelope( @@ -25,15 +31,32 @@ pub fn write_envelope( actual: body.len() as u64, }); } - let payload_len = u32::try_from(8 + body.len()).map_err(|_| CacheError::ObjectTooLarge { - limit: max_frame as u64, - actual: body.len() as u64, - })?; - let mut buf = Vec::with_capacity(4 + payload_len as usize); - buf.extend_from_slice(&payload_len.to_le_bytes()); - buf.extend_from_slice(&server_now.to_le_bytes()); - buf.extend_from_slice(body); - stream.write_all(&buf).map_err(map_io)?; + let payload_len = body + .len() + .checked_add(8) + .and_then(|len| u32::try_from(len).ok()) + .ok_or(CacheError::ObjectTooLarge { + limit: max_frame as u64, + actual: body.len() as u64, + })?; + let mut header = [0u8; 12]; + header[..4].copy_from_slice(&payload_len.to_le_bytes()); + header[4..].copy_from_slice(&server_now.to_le_bytes()); + write_parts(stream, &header, body).map_err(map_io) +} + +// Keep the payload borrowed. A short vectored write can stop in either slice. +fn write_parts(writer: &mut impl Write, header: &[u8], body: &[u8]) -> std::io::Result<()> { + let mut parts = [IoSlice::new(header), IoSlice::new(body)]; + let mut remaining = &mut parts[..]; + while !remaining.is_empty() { + match writer.write_vectored(remaining) { + Ok(0) => return Err(std::io::ErrorKind::WriteZero.into()), + Ok(n) => IoSlice::advance_slices(&mut remaining, n), + Err(err) if err.kind() == std::io::ErrorKind::Interrupted => continue, + Err(err) => return Err(err), + } + } Ok(()) } @@ -52,11 +75,14 @@ pub fn read_envelope( actual: payload_len.saturating_sub(8) as u64, }); } - buffer.buf_mut().resize(payload_len, 0); + let mut now = [0u8; 8]; + stream.read_exact(&mut now).map_err(map_io)?; + buffer.buf_mut().resize(payload_len - 8, 0); stream.read_exact(buffer.buf_mut()).map_err(map_io)?; - let server_now = u64::from_le_bytes(buffer.buf_mut()[..8].try_into().unwrap_or([0; 8])); - let body = buffer.buf_mut()[8..].to_vec(); - Ok(Envelope { server_now, body }) + Ok(Envelope { + server_now: u64::from_le_bytes(now), + body: buffer, + }) } pub fn map_io(err: std::io::Error) -> CacheError { @@ -68,3 +94,80 @@ pub fn map_io(err: std::io::Error) -> CacheError { } } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::pool::Pool; + use std::net::TcpListener; + use std::sync::Arc; + + struct ShortWriter { + bytes: Vec, + limit: usize, + interrupted: bool, + } + impl Write for ShortWriter { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + if !self.interrupted { + self.interrupted = true; + return Err(std::io::ErrorKind::Interrupted.into()); + } + let n = buf.len().min(self.limit); + self.bytes.extend_from_slice(&buf[..n]); + Ok(n) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + #[test] + fn partial_writes_and_interrupts_preserve_the_wire_bytes() { + for limit in 1..32 { + let mut writer = ShortWriter { + bytes: Vec::new(), + limit, + interrupted: false, + }; + write_parts(&mut writer, b"header", b"payload").unwrap(); + assert_eq!(writer.bytes, b"headerpayload"); + } + let mut writer = ShortWriter { + bytes: Vec::new(), + limit: 0, + interrupted: true, + }; + assert_eq!( + write_parts(&mut writer, b"header", b"payload") + .unwrap_err() + .kind(), + std::io::ErrorKind::WriteZero + ); + } + + #[test] + fn envelope_owns_the_pooled_buffer_and_preserves_wire_format() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let mut sender = TcpStream::connect(listener.local_addr().unwrap()).unwrap(); + let (mut receiver, _) = listener.accept().unwrap(); + let pool = Arc::new(Pool::new(1)); + let mut buffer = pool.checkout().unwrap(); + buffer.buf_mut().resize(32, 0); + let allocation = buffer.as_slice().as_ptr(); + write_envelope(&mut sender, 42, b"payload", 32).unwrap(); + let envelope = read_envelope(&mut receiver, buffer, 32).unwrap(); + assert_eq!(envelope.server_now, 42); + assert_eq!(envelope.body(), b"payload"); + assert_eq!(envelope.body().as_ptr(), allocation); + assert!(pool.checkout().is_err()); + drop(envelope); + assert_eq!(pool.checkout().unwrap().as_slice().as_ptr(), allocation); + write_envelope(&mut sender, 7, b"abc", 32).unwrap(); + let mut raw = [0u8; 15]; + receiver.read_exact(&mut raw).unwrap(); + assert_eq!(&raw[..4], &11u32.to_le_bytes()); + assert_eq!(&raw[4..12], &7u64.to_le_bytes()); + assert_eq!(&raw[12..], b"abc"); + } +} diff --git a/crates/transport/src/pool.rs b/crates/transport/src/pool.rs index 91520d0..0f0775a 100644 --- a/crates/transport/src/pool.rs +++ b/crates/transport/src/pool.rs @@ -56,6 +56,10 @@ impl Pool { } impl Pooled { + pub fn as_slice(&self) -> &[u8] { + &self.buf + } + pub fn buf_mut(&mut self) -> &mut Vec { &mut self.buf } diff --git a/crates/transport/src/server.rs b/crates/transport/src/server.rs index 4f89e95..1fec976 100644 --- a/crates/transport/src/server.rs +++ b/crates/transport/src/server.rs @@ -43,7 +43,11 @@ impl Default for ServerConfig { impl ServerConfig { pub fn validate(&self) -> Result<(), CacheError> { - if self.max_connections == 0 || self.pool_buffers == 0 || self.max_frame_bytes < 64 { + if self.max_connections == 0 + || self.pool_buffers == 0 + || self.max_frame_bytes < 64 + || self.io_timeout.is_zero() + { return Err(CacheError::Malformed { reason: "server config", }); @@ -172,33 +176,45 @@ fn connection(mut stream: TcpStream, shared: Arc) { Ok(envelope) => envelope, Err(_) => break, }; - let reply = { - let mut store = shared.store.lock().unwrap_or_else(|err| err.into_inner()); - handle_frame(&mut store, &envelope.body) - }; + let reply = handle_frame(&shared.store, envelope.body()); + // The decoder owns the request; return the read buffer before writing. + drop(envelope); if write_envelope(&mut stream, mono_ms(), &reply, shared.cfg.max_frame_bytes).is_err() { break; } } } -fn handle_frame(store: &mut FileStore, body: &[u8]) -> Vec { +fn handle_frame(store: &Mutex, body: &[u8]) -> Vec { if body.len() >= 8 && &body[..4] == b"GCK1" { let op = u16::from_le_bytes([body[6], body[7]]); if op == OP_HELLO { - return match decode_hello(body) - .and_then(|versions| negotiate(&[WIRE_VERSION], &versions)) - { - Ok(version) => encode_hello_ack(version, store.epoch()), - Err(err) => fail(OP_HELLO, store.epoch(), &err), + let negotiated = + decode_hello(body).and_then(|versions| negotiate(&[WIRE_VERSION], &versions)); + let epoch = store.lock().unwrap_or_else(|err| err.into_inner()).epoch(); + return match negotiated { + Ok(version) => encode_hello_ack(version, epoch), + Err(err) => fail(OP_HELLO, epoch, &err), }; } } - let now = mono_ms(); - match decode_request(body) { - Ok((auth, request)) => dispatch(store, &auth, request, now), - Err(err) => fail(0, store.epoch(), &err), - } + let decoded = decode_request(body); + let reply = { + let mut store = store.lock().unwrap_or_else(|err| err.into_inner()); + // Time is checked after waiting for the store, not before contention. + match decoded { + Ok((auth, request)) => dispatch(&mut store, &auth, request, mono_ms()), + Err(error) => ReplyBody::Failed { + op: 0, + store_epoch: store.epoch(), + error, + }, + } + }; + encode_reply(&reply).unwrap_or_else(|error| { + let epoch = store.lock().unwrap_or_else(|err| err.into_inner()).epoch(); + fail(0, epoch, &error) + }) } fn dispatch( @@ -206,38 +222,27 @@ fn dispatch( auth: &gpucache_protocol::AuthContext, request: RequestBody, now: u64, -) -> Vec { +) -> ReplyBody { let op = request_op(&request); - let encoded = match request { - RequestBody::Lookup(items) => store - .lookup(auth, now, items) - .map(ReplyBody::Lookup) - .and_then(|body| encode_reply(&body)), - RequestBody::ReservePut(items) => store - .reserve_put(auth, now, items) - .map(ReplyBody::Reserve) - .and_then(|body| encode_reply(&body)), - RequestBody::Commit(items) => store - .commit(auth, now, items) - .map(ReplyBody::Commit) - .and_then(|body| encode_reply(&body)), - RequestBody::Get(items) => store - .get(auth, now, items) - .map(ReplyBody::Get) - .and_then(|body| encode_reply(&body)), + let result = match request { + RequestBody::Lookup(items) => store.lookup(auth, now, items).map(ReplyBody::Lookup), + RequestBody::ReservePut(items) => { + store.reserve_put(auth, now, items).map(ReplyBody::Reserve) + } + RequestBody::Commit(items) => store.commit(auth, now, items).map(ReplyBody::Commit), + RequestBody::Get(items) => store.get(auth, now, items).map(ReplyBody::Get), RequestBody::ReleaseCancel(items) => store .release_cancel(auth, now, items) - .map(ReplyBody::Release) - .and_then(|body| encode_reply(&body)), + .map(ReplyBody::Release), RequestBody::Invalidate(items) => store .invalidate(auth, now, items) - .map(ReplyBody::Invalidate) - .and_then(|body| encode_reply(&body)), + .map(ReplyBody::Invalidate), }; - match encoded { - Ok(bytes) => bytes, - Err(err) => fail(op, store.epoch(), &err), - } + result.unwrap_or_else(|error| ReplyBody::Failed { + op, + store_epoch: store.epoch(), + error, + }) } fn fail(op: u16, epoch: u64, err: &CacheError) -> Vec { diff --git a/docs/OPERATIONS.md b/docs/OPERATIONS.md index fa8ca9d..1d5d3d1 100644 --- a/docs/OPERATIONS.md +++ b/docs/OPERATIONS.md @@ -134,3 +134,25 @@ See [../COMPATIBILITY.md](../COMPATIBILITY.md) and [V3_DPU_TARGETS.md](V3_DPU_TA ## Not operated here No NVMe format, no DPU mode change, no firmware, no IOMMU or BIOS change, no production network change, no hardware purchase. Those need a separate authorization. There is no website deploy and no crates.io publish in this milestone. G4 live calls stay env-gated. + +## CPU audit changes (2026-10-15) + +LRU recency is advisory. Lookup and Get update it in memory without filesystem writes. +Commit batches and invalidation checkpoint it; restart may lose recent read ordering. +Block persistence still uses the existing atomic-write and sync policy. A stale LRU +checkpoint can change eviction order, never tenant identity or checksum validation. + +A TCP client becomes unusable after an incomplete or malformed exchange. Recompute +and reconnect before issuing another cache request. The connect call uses its +configured timeout. Socket I/O timeouts are not total RPC deadlines and do not +bound a store lock or a filesystem syscall. Zero server I/O timeouts are rejected. + +The buffer pool owns receive payloads through decoding; framing writes borrow the +encoded payload through vectored I/O. This removes framing copies, not all copies. +Serialization, storage ownership, and Get replies still copy bytes. Parsing and +reply serialization run outside the store mutex; storage I/O remains inside it. + +Run the optional local microbenchmarks with +`cargo run --release -p gpucache-transport --example cpu_audit`. +These measure synthetic CPU/file operations only. They do not measure RDMA, GPU +inference, TTFT, TPOT, DPU savings, or TCO. diff --git a/docs/PROTOCOL.md b/docs/PROTOCOL.md index 4da46a2..e15edbb 100644 --- a/docs/PROTOCOL.md +++ b/docs/PROTOCOL.md @@ -28,7 +28,7 @@ Every item has: - `epoch` — must equal the store epoch - `deadline_ms` — absolute monotonic deadline; `0` is malformed; `now >= deadline` is `TimedOut` -Mutating ops are idempotent inside a bounded window. The same `request_id` and the same canonical arguments replay the stored result. The same id with different arguments is `IdempotencyConflict`. Lookup and Get are not logged; they are safe to repeat. A deadline miss before any mutation is not logged, so the caller can retry that id with a later deadline. +Mutating ops are idempotent inside a bounded window. Within one authenticated tenant and trust domain, the same `request_id` and the same canonical arguments replay the stored result. The same id with different arguments is `IdempotencyConflict`. Lookup and Get are not logged. Each successful Get creates a reader pin, including repeated calls with the same request id; callers must release every pin. A global reader-pin admission cap is not implemented. A deadline miss before any mutation is not logged, so the caller can retry that id with a later deadline. ## Operations diff --git a/docs/RISKS.md b/docs/RISKS.md index d4c42fc..54b3b1f 100644 --- a/docs/RISKS.md +++ b/docs/RISKS.md @@ -4,7 +4,7 @@ | --- | --- | --- | | Wrong-tenant or wrong-layout hit | Mitigated in `MemoryStore` tests | Physical key includes tenant and layout. Incompatible lookup returns a miss and no bytes. | | Reclaim of a buffer that might still be copied | Mitigated by quarantine | Timeout and `KnownIdle` do not free a registered region. | -| Unbounded memory | Mitigated by quotas | Admission fails instead of evicting. Quarantine still holds bytes until fence or drain, which is intentional and counted against the quota. | +| Unbounded memory | Partially mitigated | Retained payloads and writers have quotas; FileStore can evict unpinned committed blocks. Reader pins and some auxiliary metadata remain uncapped. Protocol reply copies also need byte admission. Quarantine stays charged until fence or drain. | | Idempotency window expiry | Accepted | After TTL, a request id may run again. Callers must not reuse ids they still care about. | | `Drained` is only tenant-authenticated | Accepted for V1 A | Not a device proof. Documented in SECURITY.md. | | No real engine bytes | `NOT_RUN_HARDWARE` | Synthetic frames can pass while a real vLLM layout still mismatches. | diff --git a/docs/SECURITY.md b/docs/SECURITY.md index dc07248..e337a91 100644 --- a/docs/SECURITY.md +++ b/docs/SECURITY.md @@ -2,7 +2,7 @@ The library calls still require an `AuthContext`. The TCP service does too. It binds `127.0.0.1` unless the operator sets `--allow-remote`, and it refuses to start with an empty tenant policy. A data request with an unknown tenant, a mismatched trust domain, or a wrong fingerprint is `Unauthorized` and returns no payload. Hello carries versions and the store epoch only. -The fingerprint is a hash of a credential, never the credential. Do not pass a raw secret on the command line in a place that records process arguments if you can avoid it; V1 still takes the hex fingerprint as an argument because there is no secret store yet. An all-zero fingerprint is rejected before a request is built. +The fingerprint is derived from a credential, but the current protocol accepts that fingerprint directly: it is a bearer-equivalent secret. Capturing it on plaintext TCP permits impersonation. Do not pass a raw secret on the command line in a place that records process arguments if you can avoid it; V1 still takes the hex fingerprint as an argument because there is no secret store yet. An all-zero fingerprint is rejected before a request is built. ## What is enforced @@ -27,7 +27,7 @@ These sit beside the V1 gate. They do not replace it and they do not add mTLS. S ## What is not enforced -- mTLS is not linked. The TCP service does not call `Acl` or `ReplayGuard` yet; those are library hooks a caller can use beside the V1 fingerprint. `--allow-remote` does not add them and does not terminate TLS. Treat a non-loopback bind as a trusted fabric, or do not bind it. +- mTLS is not linked. The TCP service does not call `Acl`, `ReplayGuard`, `RateGate`, or `SecretRing` yet; those are library hooks a caller can use beside the V1 fingerprint. `--allow-remote` does not add them and does not terminate TLS. Treat a non-loopback bind as a trusted fabric, or do not bind it. - Replay protection beyond the idempotency window and the bounded nonce window is not a permanent cache. - `Attestation::Drained` is accepted from any caller who already passed the tenant gate. It is not a device measurement. A buggy client inside the tenant can reclaim its own in-flight buffer early. - The idempotency table forgets entries after `idempotency_ttl_ms` or when it is full (new ids are then rejected until expiry). Reusing an id after expiry can run the operation again. diff --git a/docs/V2_DATA_PATHS.md b/docs/V2_DATA_PATHS.md index 87efa47..66639c4 100644 --- a/docs/V2_DATA_PATHS.md +++ b/docs/V2_DATA_PATHS.md @@ -41,7 +41,7 @@ The hot TCP service does not depend on `gpucache-cold`. RustFS is not on the pre | Async sink | `AsyncSink` queues that pair up to a job and byte cap. `drain` puts the object, then the manifest. `cancel` drops the queue and writes nothing | Host buffers until drain. Not a runtime thread pool | | Mock store | In-process map. Contract test covers Put, Head, RangeGet, Delete | Not RustFS. Not a measurement | | Live HTTP | `live_probe` runs only when `GPUCACHE_RUSTFS_*` is set. Otherwise `NOT_RUN_HARDWARE` | Full HTTP body copy. Not GDS | -| Refill | Head the manifest. If its length exceeds the budget, return `OverBudget` and do not download it. Otherwise download the manifest, and RangeGet one generation only when that range fits. Checksum mismatch is an error, not a hit | Host copy of the manifest and, when allowed, one payload range | +| Refill | Head the manifest. If its length exceeds the budget, return `OverBudget` and do not download it. Otherwise download the manifest, and RangeGet one generation only when manifest bytes plus that range fit the total byte budget. Checksum mismatch is an error, not a hit | Host copy of the manifest and, when allowed, one payload range | | Promote into G3.5 | A later hot `Put` through the TCP API | A separate host copy. Not chained to the GET | GDS and BlueField storage DMA are in development and were not run. diff --git a/docs/V3_FAILURES.md b/docs/V3_FAILURES.md index b5bede1..4afaf52 100644 --- a/docs/V3_FAILURES.md +++ b/docs/V3_FAILURES.md @@ -1,15 +1,15 @@ # V3 failure matrix -Each row is a clean miss. The engine recomputes and may fill again. The hook returns no payload, does not report a wrong hit, and does not stay blocked. `StorageQueue::apply_fault` applies the DPU-local rows to the simulated slab. RustFS rows do not wipe the hot slab; `refill_degraded` returns `Missing` without calling the store. +Each row is a clean miss. The engine recomputes and may fill again. The hook returns no payload, does not report a wrong hit. The decision flag is not proof of sustained write availability: hidden slots retain their byte charges. `StorageQueue::apply_fault` applies the DPU-local rows to the simulated slab. RustFS rows do not wipe the hot slab; `refill_degraded` returns `Missing` without calling the store. | Fault | What the hook does | Result | | --- | --- | --- | | DPU restart | Drop in-flight tasks and drop slab slots | Clean miss. A new write can complete. Old slot ids are not readable | -| GPU reset | Hide every slot and drop in-flight tasks. Client completion is unknown, so bytes stay unreadable | Clean miss. The queue still accepts a new write | -| NIC flap | Same hide. An in-flight transfer is not a hit | Clean miss. Not a permanent block | +| GPU reset | Hide every slot and drop in-flight tasks. Client completion is unknown, so bytes stay unreadable | Clean miss. New writes require remaining slab capacity | +| NIC flap | Same hide. An in-flight transfer is not a hit | Clean miss. Hidden bytes can exhaust the slab | | Storage loss | Hide slots. Hidden bytes are not a durability replica and are not returned | Clean miss, then recompute | | NVMe timeout | Drop one in-flight task. Do not expose it | Clean miss. Later submits still run | -| Control-plane loss | Hide slots and refuse new submits with `Cancelled` | No read while membership is unknown. `restore_control` allows new writes. Old slots stay hidden | +| Control-plane loss | Hide slots and refuse new submits with `Cancelled` | No read while membership is unknown. `restore_control` permits admission if capacity remains. Old slots stay hidden and charged | | RustFS slow | Do not wait and do not call the object store | `RefillOutcome::Missing` | | RustFS unavailable | Same as slow | `RefillOutcome::Missing` | @@ -18,3 +18,5 @@ Checksum mismatch hides the simulated slot after the failing read. A wrong tenan Lease expiry is still not proof that a copy stopped. The simulated slab does not free a slot only because a timer fired. Restart and the other faults above are explicit fencing events, not a lease callback. This matrix is a CPU contract. It is not a fault campaign on BlueField, GPU, NIC, or RustFS hardware. + +A full hidden slab has no drain/reclaim API yet. In this CPU simulation, an explicit DPU restart clears it. Do not translate that into permission to free real DMA buffers: hardware integration needs verified drain and device fencing. Pending write bytes and committed slab bytes share the configured byte cap. diff --git a/docs/adr/0005-shard-fencing.md b/docs/adr/0005-shard-fencing.md index 8a4335c..185fe10 100644 --- a/docs/adr/0005-shard-fencing.md +++ b/docs/adr/0005-shard-fencing.md @@ -12,7 +12,7 @@ A distributed KV tier needs to know which process may fill a block, and it needs `ShardDirectory` is local to one shard. One process is one shard. A single machine is shard 0 using the same methods as any other shard id. -- Membership is a list of `(node_id, incarnation)` on that shard. A new member, a removed member, or a changed incarnation bumps the shard epoch and fails in-flight fills. An unchanged incarnation does not. The last member cannot be removed. +- Membership is a list of `(node_id, incarnation)` on that shard. A new member, a removed member, or a newer incarnation bumps the shard epoch and fails in-flight fills. An unchanged incarnation does not. An older incarnation is rejected without changing the epoch. The last member cannot be removed. - There is no cross-shard quorum and no per-token index. The map key is the SHA-256 of the physical byte key. One entry is one in-flight fill. - Single-flight: the first same-epoch caller is the leader and reserves capacity credits once. A later same-tenant caller joins and does not reserve again. A different tenant is `Unauthorized`. The directory stores no payload. `Admit` is `Leader` or `Joined` only. - Completing or failing a fill requires the leader request id and the current epoch. A stale epoch is `EpochMismatch`. After the epoch moves, the fill is gone, credits are released, and a new fill may start. That new fill is not a hit. diff --git a/docs/adr/0007-trusted-fabric.md b/docs/adr/0007-trusted-fabric.md index ed398f7..b021422 100644 --- a/docs/adr/0007-trusted-fabric.md +++ b/docs/adr/0007-trusted-fabric.md @@ -17,10 +17,10 @@ The V3 boundary is a **trusted fabric**, documented here, plus CPU hooks. mTLS i Operators: - Leave the listener on `127.0.0.1` unless the host network is already restricted to cache peers. -- `--allow-remote` does not add authentication and does not start TLS. The fingerprint and the V3 ACL still apply. +- `--allow-remote` does not add authentication and does not start TLS. The V1 fingerprint gate still applies. The TCP service does not call the V3 ACL, replay guard, rate gate, or secret ring. - Do not place this TCP service on an open network and call that mTLS. -In-process hooks, all caller-clocked and bounded: +In-process hooks (the ACL, nonce window, rate windows, and audit log have caps; secret-rotation history currently grows without a cap): - `Acl` maps a fingerprint to read or write. An empty ACL fails closed. A read grant is not a write grant. The fingerprint compare scans every entry. - `ReplayGuard` accepts a nonce once inside a caller-supplied skew. The window is bounded. Evicting the oldest nonce means that nonce can appear again later. This is not an infinite replay cache.