diff --git a/README.md b/README.md index 56ca2c9..9480f83 100644 --- a/README.md +++ b/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 diff --git a/crates/bench/src/lib.rs b/crates/bench/src/lib.rs index 730a5ee..d75c960 100644 --- a/crates/bench/src/lib.rs +++ b/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}"); } diff --git a/crates/cold/src/demo.rs b/crates/cold/src/demo.rs index a385e3a..28cded7 100644 --- a/crates/cold/src/demo.rs +++ b/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 { }); } 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")); diff --git a/crates/cold/src/lib.rs b/crates/cold/src/lib.rs index 3fcf8d1..900a785 100644 --- a/crates/cold/src/lib.rs +++ b/crates/cold/src/lib.rs @@ -14,6 +14,7 @@ mod http; mod mock; mod ops; mod pack; +mod refill; mod sigv4; mod sink; @@ -21,11 +22,12 @@ 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}; diff --git a/crates/cold/src/ops.rs b/crates/cold/src/ops.rs index d5d6548..824bc43 100644 --- a/crates/cold/src/ops.rs +++ b/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 { + 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 { 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 { + Err(CacheError::Storage { + reason: "object store", + }) + } + fn range_get(&self, _key: &str, _offset: u64, _length: u64) -> Result, 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 + ); + } + } } diff --git a/crates/cold/src/pack.rs b/crates/cold/src/pack.rs index ad3eb3b..1c74c40 100644 --- a/crates/cold/src/pack.rs +++ b/crates/cold/src/pack.rs @@ -10,7 +10,9 @@ const OBJECT_MAGIC: &[u8; 4] = b"GPKG"; const MANIFEST_MAGIC: &[u8; 4] = b"GPMF"; const OBJECT_VERSION: u16 = 1; /// Manifest layout this crate will read. Any other version is a miss. -pub const MANIFEST_VERSION: u16 = 1; +/// Version 3 binds every range to its complete canonical cache identity. +/// Versions 1 and 2 are clean refill misses and must be recomputed. +pub const MANIFEST_VERSION: u16 = 3; #[derive(Clone, Debug, PartialEq, Eq)] pub struct ObjectMeta { @@ -32,6 +34,11 @@ pub struct PayloadRange { pub offset: u64, pub length: u64, pub checksum: Checksum, + /// Tenant that owns this generation. A refill for a different tenant must + /// not return the payload. + pub tenant: Vec, + /// SHA-256 of the validated complete canonical CacheIdentity. + pub identity_digest: Checksum, } #[derive(Clone, Debug, PartialEq, Eq)] @@ -66,8 +73,27 @@ pub fn pack_blocks(namespace: &str, blocks: &[PackedBlock]) -> Result 128 { + return Err(CacheError::Malformed { reason: "cold key" }); + } + if block.payload.is_empty() { + return Err(CacheError::Malformed { + reason: "cold pack", + }); + } verify_checksum(&block.payload, block.checksum)?; if block.payload.len() > HARD_BYTES_CAP { return Err(CacheError::ObjectTooLarge { @@ -75,14 +101,15 @@ pub fn pack_blocks(namespace: &str, blocks: &[PackedBlock]) -> Result Result Result Result { } let mut ranges = Vec::with_capacity(n); for _ in 0..n { - ranges.push(PayloadRange { + let range = PayloadRange { generation: r.u64()?, offset: r.u64()?, length: r.u64()?, checksum: Checksum(r.array32()?), - }); + tenant: r.bytes(128)?.to_vec(), + identity_digest: Checksum(r.array32()?), + }; + if range.tenant.is_empty() + || range.length == 0 + || range.length > HARD_BYTES_CAP as u64 + || range.offset.checked_add(range.length).is_none() + || ranges + .iter() + .any(|earlier: &PayloadRange| earlier.generation == range.generation) + { + return Err(CacheError::Malformed { + reason: "cold manifest", + }); + } + ranges.push(range); } r.finish()?; Ok(Manifest { @@ -231,6 +278,112 @@ pub fn read_manifest(bytes: &[u8]) -> Result { }) } +pub(crate) fn identity_digest(identity: &CacheIdentity) -> Result { + identity.validate_wire()?; + Ok(content_checksum(&identity.to_bytes()?)) +} + +fn digest_hex(checksum: Checksum) -> String { + const HEX: &[u8; 16] = b"0123456789abcdef"; + let mut value = String::with_capacity(64); + for byte in checksum.0 { + value.push(char::from(HEX[usize::from(byte >> 4)])); + value.push(char::from(HEX[usize::from(byte & 15)])); + } + value +} + +fn content_key(namespace: &str, kind: &str, checksum: Checksum) -> Result { + let key = format!("{namespace}/{}/{kind}", digest_hex(checksum)); + if !key_ok(namespace) || !key_ok(&key) { + return Err(CacheError::Malformed { reason: "cold key" }); + } + Ok(key) +} + +fn content_namespace<'a>( + key: &'a str, + kind: &str, + checksum: Checksum, +) -> Result<&'a str, CacheError> { + let suffix = format!("/{kind}"); + let Some(prefix) = key.strip_suffix(&suffix) else { + return Err(CacheError::Malformed { reason: "cold key" }); + }; + let Some((namespace, digest)) = prefix.rsplit_once('/') else { + return Err(CacheError::Malformed { reason: "cold key" }); + }; + if !key_ok(key) || !key_ok(namespace) { + return Err(CacheError::Malformed { reason: "cold key" }); + } + if digest != digest_hex(checksum) { + return Err(CacheError::ChecksumMismatch); + } + Ok(namespace) +} + +/// Validate the content-addressed manifest and its reference to the object. +pub(crate) fn validate_manifest_binding( + manifest_key: &str, + manifest_bytes: &[u8], + manifest: &Manifest, +) -> Result<(), CacheError> { + let manifest_namespace = + content_namespace(manifest_key, "manifest", content_checksum(manifest_bytes))?; + let object_namespace = + content_namespace(&manifest.object_key, "object", manifest.object_checksum)?; + if manifest_namespace != object_namespace { + return Err(CacheError::Malformed { reason: "cold key" }); + } + Ok(()) +} + +/// Reject a manually assembled job that could overwrite another published pack +/// or claim a range/identity absent from its object. No storage calls are made. +pub(crate) fn validate_pack_job( + object_key: &str, + object_bytes: &[u8], + manifest_key: &str, + manifest_bytes: &[u8], +) -> Result<(), CacheError> { + let manifest = read_manifest(manifest_bytes)?; + validate_manifest_binding(manifest_key, manifest_bytes, &manifest)?; + if manifest.object_key != object_key + || content_checksum(object_bytes) != manifest.object_checksum + { + return Err(CacheError::ChecksumMismatch); + } + let mut object = ByteReader::new(object_bytes); + if object.raw(4)? != OBJECT_MAGIC + || object.u16()? != OBJECT_VERSION + || object.u32()? as usize != manifest.ranges.len() + { + return Err(CacheError::Malformed { + reason: "cold pack", + }); + } + for range in &manifest.ranges { + let generation = object.u64()?; + let checksum = Checksum(object.array32()?); + let identity = CacheIdentity::from_bytes(object.bytes(HARD_BYTES_CAP)?)?; + let offset = object_bytes.len() - object.remaining() + 4; + let payload = object.bytes(HARD_BYTES_CAP)?; + if generation != range.generation + || checksum != range.checksum + || range.offset != offset as u64 + || range.length != payload.len() as u64 + || range.tenant != identity.tenant.as_bytes() + || range.identity_digest != identity_digest(&identity)? + { + return Err(CacheError::Malformed { + reason: "cold pack", + }); + } + verify_checksum(payload, checksum)?; + } + object.finish() +} + pub(crate) fn key_ok(key: &str) -> bool { !key.is_empty() && key.len() <= 256 @@ -262,9 +415,9 @@ mod tests { MANIFEST_VERSION ); let mut bad = pack.manifest_bytes.clone(); - bad[4] = 2; + bad[4] = 9; bad[5] = 0; - assert_eq!(manifest_version(&bad).unwrap(), 2); + assert_eq!(manifest_version(&bad).unwrap(), 9); assert!(matches!( read_manifest(&bad), Err(CacheError::Storage { @@ -272,4 +425,89 @@ mod tests { }) )); } + + #[test] + fn duplicate_generation_is_rejected() { + let mut blocks = sample_blocks().unwrap(); + blocks[1].generation = blocks[0].generation; + assert_eq!( + pack_blocks("dup", &blocks).unwrap_err(), + CacheError::Malformed { + reason: "cold pack" + } + ); + } + + #[test] + fn content_derived_keys_validate_the_final_length() { + let blocks = sample_blocks().unwrap(); + let pack = pack_blocks(&"n".repeat(182), &blocks).unwrap(); + assert_eq!(pack.manifest_key.len(), 256); + assert!(pack.object_key.len() <= 256); + validate_pack_job( + &pack.object_key, + &pack.object_bytes, + &pack.manifest_key, + &pack.manifest_bytes, + ) + .unwrap(); + assert_eq!( + pack_blocks(&"n".repeat(183), &blocks).unwrap_err(), + CacheError::Malformed { reason: "cold key" } + ); + } + + #[test] + fn publication_rejects_forged_identity_even_with_content_derived_keys() { + let blocks = sample_blocks().unwrap(); + let pack = pack_blocks("forged", &blocks).unwrap(); + let mut manifest_bytes = pack.manifest_bytes.clone(); + // The final field is the last range's complete identity digest. + *manifest_bytes.last_mut().unwrap() ^= 1; + let manifest_key = + content_key("forged", "manifest", content_checksum(&manifest_bytes)).unwrap(); + assert_eq!( + validate_pack_job( + &pack.object_key, + &pack.object_bytes, + &manifest_key, + &manifest_bytes, + ) + .unwrap_err(), + CacheError::Malformed { + reason: "cold pack" + } + ); + } + + #[test] + fn publication_rejects_cross_namespace_and_body_substitution() { + let blocks = sample_blocks().unwrap(); + let pack = pack_blocks("bound", &blocks).unwrap(); + let foreign_manifest = content_key( + "foreign", + "manifest", + content_checksum(&pack.manifest_bytes), + ) + .unwrap(); + assert!(validate_pack_job( + &pack.object_key, + &pack.object_bytes, + &foreign_manifest, + &pack.manifest_bytes, + ) + .is_err()); + let mut wrong_body = pack.object_bytes.clone(); + *wrong_body.last_mut().unwrap() ^= 1; + assert_eq!( + validate_pack_job( + &pack.object_key, + &wrong_body, + &pack.manifest_key, + &pack.manifest_bytes, + ) + .unwrap_err(), + CacheError::ChecksumMismatch + ); + } } diff --git a/crates/cold/src/refill.rs b/crates/cold/src/refill.rs new file mode 100644 index 0000000..34b1ac8 --- /dev/null +++ b/crates/cold/src/refill.rs @@ -0,0 +1,129 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 + +//! Budgeted refill with mandatory complete identity matching. +//! +//! The caller supplies an expected identity from its authenticated tenant and +//! trust-domain context. This library does not authenticate that caller. Raw +//! ObjectStore access is a trusted internal capability, not a tenant API. + +use gpucache_core_types::{content_checksum, CacheError, CacheIdentity}; + +use crate::pack::{ + identity_digest, key_ok, manifest_version, read_manifest, validate_manifest_binding, + ObjectStore, MANIFEST_VERSION, +}; + +#[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, +} + +impl RefillBudget { + pub fn new(max_bytes: u64, max_ranges: u32) -> Result { + if max_bytes == 0 || max_ranges == 0 { + return Err(CacheError::Malformed { + reason: "refill budget", + }); + } + Ok(Self { + max_bytes, + max_ranges, + }) + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum RefillOutcome { + Loaded { + payload: Vec, + }, + /// Manifest or payload exceeds the combined budget. No payload bytes. + OverBudget, + /// Missing or incompatible cache entry. The caller recomputes. + Missing, +} + +/// Fetch a generation only for its complete expected cache identity. +/// +/// The identity must come from an authenticated upstream request. A generation +/// alone is not an identity, and this function has no unguarded overload. +pub fn refill_generation( + store: &dyn ObjectStore, + manifest_key: &str, + generation: u64, + budget: &RefillBudget, + expected: &CacheIdentity, +) -> Result { + refill_matching(store, manifest_key, generation, budget, expected) +} + +/// Match tenant and complete canonical identity before any payload download. +/// Tenant mismatch is Unauthorized; other identity mismatch is a clean miss. +/// Legacy manifests lack the required binding and are clean misses too. +pub fn refill_matching( + store: &dyn ObjectStore, + manifest_key: &str, + generation: u64, + budget: &RefillBudget, + expected: &CacheIdentity, +) -> Result { + let expected_digest = identity_digest(expected)?; + if !key_ok(manifest_key) { + return Err(CacheError::Malformed { reason: "cold key" }); + } + let meta = match store.head(manifest_key) { + Ok(meta) => meta, + Err(CacheError::Storage { + reason: "object missing", + }) => return Ok(RefillOutcome::Missing), + Err(err) => return Err(err), + }; + if meta.len > budget.max_bytes { + return Ok(RefillOutcome::OverBudget); + } + let manifest_bytes = store.range_get(manifest_key, 0, meta.len)?; + if manifest_bytes.len() as u64 != meta.len || content_checksum(&manifest_bytes) != meta.checksum + { + return Err(CacheError::ChecksumMismatch); + } + // Check the complete header first. An invalid/truncated magic is corrupt, + // not a legacy miss. Versions 1/2 cannot prove a complete identity match. + if manifest_version(&manifest_bytes)? != MANIFEST_VERSION { + return Ok(RefillOutcome::Missing); + } + let manifest = read_manifest(&manifest_bytes)?; + validate_manifest_binding(manifest_key, &manifest_bytes, &manifest)?; + if manifest.ranges.len() as u32 > budget.max_ranges { + return Ok(RefillOutcome::OverBudget); + } + let Some(range) = manifest + .ranges + .iter() + .find(|range| range.generation == generation) + else { + return Ok(RefillOutcome::Missing); + }; + if range.tenant.as_slice() != expected.tenant.as_bytes() { + return Err(CacheError::Unauthorized); + } + if range.identity_digest != expected_digest { + return Ok(RefillOutcome::Missing); + } + if range.length > budget.max_bytes.saturating_sub(meta.len) { + return Ok(RefillOutcome::OverBudget); + } + let payload = match store.range_get(&manifest.object_key, range.offset, range.length) { + Ok(payload) => payload, + Err(CacheError::Storage { + reason: "object missing", + }) => return Ok(RefillOutcome::Missing), + Err(err) => return Err(err), + }; + if payload.len() as u64 != range.length || content_checksum(&payload) != range.checksum { + return Err(CacheError::ChecksumMismatch); + } + Ok(RefillOutcome::Loaded { payload }) +} diff --git a/crates/cold/src/sink.rs b/crates/cold/src/sink.rs index ea979f0..5904962 100644 --- a/crates/cold/src/sink.rs +++ b/crates/cold/src/sink.rs @@ -1,18 +1,18 @@ // Copyright 2024-2026 RustFS, Inc. // SPDX-License-Identifier: Apache-2.0 -//! Decoupled cold writes and a budget-capped refill. -//! -//! `AsyncSink` is a bounded queue drained by the caller. It is not a Tokio -//! runtime and it does not run on the hot TCP path. `refill_generation` -//! returns payload bytes only after a checksum match that fits the budget. -//! Over budget is an empty outcome: the engine recomputes. +//! Bounded, caller-driven cold writes, separate from the hot TCP path. use std::collections::VecDeque; -use gpucache_core_types::{content_checksum, CacheError, QuotaKind}; +use gpucache_core_types::{CacheError, QuotaKind}; -use crate::pack::{key_ok, read_manifest, Manifest, ObjectStore, Pack}; +use crate::pack::{validate_pack_job, ObjectStore, Pack}; + +/// Hard metadata bound shared by pending and failed jobs. +pub const MAX_SINK_JOBS: usize = 4096; +/// Total write attempts per job, including its first attempt. +pub const MAX_WRITE_ATTEMPTS: u8 = 3; #[derive(Clone, Debug, PartialEq, Eq)] pub struct ColdJob { @@ -35,64 +35,55 @@ impl ColdJob { pub fn byte_weight(&self) -> u64 { self.object_bytes.len() as u64 + self.manifest_bytes.len() as u64 } - - fn weight(&self) -> u64 { - self.byte_weight() - } } -#[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, +struct PendingJob { + job: ColdJob, + attempts: u8, } -impl RefillBudget { - pub fn new(max_bytes: u64, max_ranges: u32) -> Result { - if max_bytes == 0 || max_ranges == 0 { - return Err(CacheError::Malformed { - reason: "refill budget", - }); - } - Ok(Self { - max_bytes, - max_ranges, - }) - } +/// A terminal failure that can be removed independently of healthy work. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct FailedColdJob { + pub job: ColdJob, + pub error: CacheError, + pub attempts: u8, } -#[derive(Clone, Debug, PartialEq, Eq)] -pub enum RefillOutcome { - Loaded { - payload: Vec, - }, - /// Manifest or payload exceeds the budget. No payload bytes. - OverBudget, - /// Clean miss. The caller recomputes. - Missing, +/// Per-call progress, including partial success. No unbounded error log is kept. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct DrainReport { + pub attempted: u32, + pub completed: u32, + pub requeued: u32, + pub failed: u32, + pub first_error: Option, } pub struct AsyncSink { - queue: VecDeque, + queue: VecDeque, + failures: VecDeque, max_jobs: usize, max_bytes: u64, queued_bytes: u64, + failed_bytes: u64, cancelled: bool, } impl AsyncSink { pub fn new(max_jobs: usize, max_bytes: u64) -> Result { - if max_jobs == 0 || max_bytes == 0 { + if max_jobs == 0 || max_jobs > MAX_SINK_JOBS || max_bytes == 0 { return Err(CacheError::Malformed { reason: "cold sink", }); } Ok(Self { queue: VecDeque::new(), + failures: VecDeque::new(), max_jobs, max_bytes, queued_bytes: 0, + failed_bytes: 0, cancelled: false, }) } @@ -100,288 +91,147 @@ impl AsyncSink { pub fn queued(&self) -> usize { self.queue.len() } + pub fn failed_count(&self) -> usize { + self.failures.len() + } + pub fn queued_bytes(&self) -> u64 { + self.queued_bytes + } + pub fn failed_bytes(&self) -> u64 { + self.failed_bytes + } + pub fn retained_bytes(&self) -> u64 { + self.queued_bytes + self.failed_bytes + } pub fn enqueue(&mut self, job: ColdJob) -> Result<(), CacheError> { if self.cancelled { return Err(CacheError::Cancelled); } - if !key_ok(&job.object_key) || !key_ok(&job.manifest_key) { - return Err(CacheError::Malformed { reason: "cold key" }); - } - let add = job.weight(); + let add = job.byte_weight(); if add > self.max_bytes { return Err(CacheError::ObjectTooLarge { limit: self.max_bytes, actual: add, }); } - let next = self.queued_bytes.saturating_add(add); - if self.queue.len() >= self.max_jobs || next > self.max_bytes { + let next = self + .retained_bytes() + .checked_add(add) + .ok_or(CacheError::QuotaExceeded { + kind: QuotaKind::InFlightBytes, + })?; + if self.queue.len() + self.failures.len() >= self.max_jobs || next > self.max_bytes { return Err(CacheError::QuotaExceeded { kind: QuotaKind::InFlightBytes, }); } - self.queued_bytes = next; - self.queue.push_back(job); + // Public fields cannot bypass immutable publication keys or checksums. + validate_pack_job( + &job.object_key, + &job.object_bytes, + &job.manifest_key, + &job.manifest_bytes, + )?; + self.queued_bytes += add; + self.queue.push_back(PendingJob { job, attempts: 0 }); Ok(()) } - /// Put at most `max_jobs` queued packs. The object is written first, then - /// the manifest, so a manifest is not visible without its object. A - /// failed manifest write deletes the object just written. - pub fn drain( + /// Attempt at most `max_jobs` jobs present at call entry, each at most once. + /// Transient failures go to the tail; permanent or exhausted failures are + /// isolated. Both queues share the original job and byte admission limits. + /// + /// A failed PUT may have committed before its acknowledgement was lost. + /// Never delete its object here: its manifest may already reference it. + /// Content-derived keys make retries idempotent and prevent older jobs + /// from overwriting another pack in the same namespace. + pub fn drain_report( &mut self, store: &mut dyn ObjectStore, max_jobs: usize, - ) -> Result { + ) -> Result { if self.cancelled { return Err(CacheError::Cancelled); } - let mut done = 0u32; - while (done as usize) < max_jobs { - let Some(job) = self.queue.pop_front() else { - break; - }; - self.queued_bytes = self.queued_bytes.saturating_sub(job.weight()); - store.put(&job.object_key, &job.object_bytes)?; - if let Err(err) = store.put(&job.manifest_key, &job.manifest_bytes) { - let _ = store.delete(&job.object_key); - return Err(err); + let limit = max_jobs.min(self.queue.len()); + let mut report = DrainReport::default(); + for _ in 0..limit { + let mut pending = self.queue.pop_front().expect("bounded queue snapshot"); + pending.attempts += 1; + report.attempted += 1; + let job = &pending.job; + let result = store + .put(&job.object_key, &job.object_bytes) + .and_then(|()| store.put(&job.manifest_key, &job.manifest_bytes)); + match result { + Ok(()) => { + self.queued_bytes -= pending.job.byte_weight(); + report.completed += 1; + } + Err(error) => { + if report.first_error.is_none() { + report.first_error = Some(error.clone()); + } + if retryable(&error) && pending.attempts < MAX_WRITE_ATTEMPTS { + self.queue.push_back(pending); + report.requeued += 1; + } else { + let weight = pending.job.byte_weight(); + self.queued_bytes -= weight; + self.failed_bytes += weight; + self.failures.push_back(FailedColdJob { + job: pending.job, + error, + attempts: pending.attempts, + }); + report.failed += 1; + } + } } - done = done.saturating_add(1); } - Ok(done) + Ok(report) + } + + /// Compatibility wrapper. The whole bounded round runs before returning + /// its first error. Use `drain_report` to observe partial success directly. + pub fn drain( + &mut self, + store: &mut dyn ObjectStore, + max_jobs: usize, + ) -> Result { + let report = self.drain_report(store, max_jobs)?; + match report.first_error { + Some(error) => Err(error), + None => Ok(report.completed), + } } - /// Drop queued work. Later enqueue and drain fail with `Cancelled`. + /// Transfer ownership of one failed job and release exactly its credits. + /// Taking a job does not retry it or delete any possibly published object. + pub fn take_failed(&mut self) -> Option { + let failed = self.failures.pop_front()?; + self.failed_bytes -= failed.job.byte_weight(); + Some(failed) + } + + /// Drop all pending and failed work. Later enqueue and drain are cancelled. pub fn cancel(&mut self) { self.queue.clear(); + self.failures.clear(); self.queued_bytes = 0; + self.failed_bytes = 0; self.cancelled = true; } } -/// Fetch one generation from a packed object if it fits `budget`. -/// -/// The manifest is not downloaded when its length exceeds the budget. The -/// payload range is not downloaded when its length exceeds the budget. -pub fn refill_generation( - store: &dyn ObjectStore, - manifest_key: &str, - generation: u64, - budget: &RefillBudget, -) -> Result { - if !key_ok(manifest_key) { - return Err(CacheError::Malformed { reason: "cold key" }); - } - let meta = match store.head(manifest_key) { - Ok(meta) => meta, - Err(CacheError::Storage { - reason: "object missing", - }) => return Ok(RefillOutcome::Missing), - Err(err) => return Err(err), - }; - if meta.len > budget.max_bytes { - return Ok(RefillOutcome::OverBudget); - } - let manifest_bytes = store.range_get(manifest_key, 0, meta.len)?; - if manifest_bytes.len() as u64 != meta.len || content_checksum(&manifest_bytes) != meta.checksum - { - return Err(CacheError::ChecksumMismatch); - } - let manifest = read_manifest(&manifest_bytes)?; - if manifest.ranges.len() as u32 > budget.max_ranges { - return Ok(RefillOutcome::OverBudget); - } - let Some(range) = manifest - .ranges - .iter() - .find(|range| range.generation == generation) - else { - return Ok(RefillOutcome::Missing); - }; - 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) -} - -fn load_range( - store: &dyn ObjectStore, - manifest: &Manifest, - offset: u64, - length: u64, - checksum: gpucache_core_types::Checksum, -) -> Result { - let payload = match store.range_get(&manifest.object_key, offset, length) { - Ok(payload) => payload, - Err(CacheError::Storage { - reason: "object missing", - }) => return Ok(RefillOutcome::Missing), - Err(err) => return Err(err), - }; - if payload.len() as u64 != length || content_checksum(&payload) != checksum { - return Err(CacheError::ChecksumMismatch); - } - Ok(RefillOutcome::Loaded { payload }) +fn retryable(error: &CacheError) -> bool { + matches!( + error, + CacheError::TimedOut | CacheError::Storage { .. } | CacheError::QuotaExceeded { .. } + ) } #[cfg(test)] -mod tests { - use super::*; - use crate::pack::{pack_blocks, PackedBlock}; - use crate::MockObjectStore; - use gpucache_core_types::{content_checksum, synthetic_identity}; - use std::cell::Cell; - - fn block(generation: u64, byte: u8, len: usize) -> PackedBlock { - let payload = vec![byte; len]; - let mut identity = synthetic_identity(byte, 1); - identity.prefix_chain[0].0[0] = byte; - PackedBlock { - generation, - identity, - checksum: content_checksum(&payload), - payload, - } - } - - struct CountStore { - inner: MockObjectStore, - manifest_ranges: Cell, - object_ranges: Cell, - } - - impl ObjectStore for CountStore { - fn put(&mut self, key: &str, body: &[u8]) -> Result<(), CacheError> { - self.inner.put(key, body) - } - - fn head(&self, key: &str) -> Result { - self.inner.head(key) - } - - fn range_get(&self, key: &str, offset: u64, length: u64) -> Result, CacheError> { - if key.ends_with("/manifest") { - self.manifest_ranges.set(self.manifest_ranges.get() + 1); - } - if key.ends_with("/object") { - self.object_ranges.set(self.object_ranges.get() + 1); - } - self.inner.range_get(key, offset, length) - } - - fn delete(&mut self, key: &str) -> Result<(), CacheError> { - self.inner.delete(key) - } - } - - #[test] - fn drain_then_refill_checks_the_budget_before_the_payload() { - let blocks = vec![block(1, 1, 32), block(2, 2, 4096)]; - let pack = pack_blocks("refill", &blocks).unwrap(); - let mut store = CountStore { - inner: MockObjectStore::default(), - manifest_ranges: Cell::new(0), - object_ranges: Cell::new(0), - }; - let mut sink = AsyncSink::new(2, 8 * 1024 * 1024).unwrap(); - sink.enqueue(ColdJob::from_pack(&pack)).unwrap(); - assert_eq!(sink.drain(&mut store, 1).unwrap(), 1); - let manifest_len = pack.manifest_bytes.len() as u64; - assert!(manifest_len < 4096); - let tight = RefillBudget::new(manifest_len, 16).unwrap(); - let skipped = refill_generation(&store, &pack.manifest_key, 2, &tight).unwrap(); - assert_eq!(skipped, RefillOutcome::OverBudget); - assert_eq!(store.manifest_ranges.get(), 1); - assert_eq!(store.object_ranges.get(), 0); - let wide = RefillBudget::new(8 * 1024 * 1024, 16).unwrap(); - let loaded = refill_generation(&store, &pack.manifest_key, 2, &wide).unwrap(); - assert_eq!( - loaded, - RefillOutcome::Loaded { - payload: blocks[1].payload.clone(), - } - ); - assert_eq!(store.object_ranges.get(), 1); - } - - #[test] - fn cancel_drops_the_queue_and_writes_nothing() { - let blocks = vec![block(1, 3, 8), block(2, 4, 8)]; - let pack = pack_blocks("cancel", &blocks).unwrap(); - let mut sink = AsyncSink::new(1, 1024 * 1024).unwrap(); - sink.enqueue(ColdJob::from_pack(&pack)).unwrap(); - let again = sink.enqueue(ColdJob::from_pack(&pack)).unwrap_err(); - assert!(matches!( - again, - CacheError::QuotaExceeded { - kind: QuotaKind::InFlightBytes - } - )); - sink.cancel(); - assert_eq!(sink.queued(), 0); - let mut store = MockObjectStore::default(); - assert_eq!( - sink.drain(&mut store, 4).unwrap_err(), - CacheError::Cancelled - ); - assert!(matches!( - store.head(&pack.object_key), - Err(CacheError::Storage { - reason: "object missing" - }) - )); - } - - #[test] - fn huge_manifest_is_not_downloaded() { - let blocks = vec![block(1, 5, 8), block(2, 6, 8)]; - let pack = pack_blocks("tiny", &blocks).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(); - let budget = RefillBudget::new(1, 16).unwrap(); - let outcome = refill_generation(&store, &pack.manifest_key, 1, &budget).unwrap(); - assert_eq!(outcome, RefillOutcome::OverBudget); - 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); - } -} +#[path = "sink_tests.rs"] +mod tests; diff --git a/crates/cold/src/sink_tests.rs b/crates/cold/src/sink_tests.rs new file mode 100644 index 0000000..a49b74c --- /dev/null +++ b/crates/cold/src/sink_tests.rs @@ -0,0 +1,383 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 + +use super::*; +use crate::{pack_blocks, MockObjectStore, ObjectMeta, PackedBlock}; +use gpucache_core_types::{content_checksum, synthetic_identity}; +use std::collections::BTreeMap; + +#[derive(Clone)] +enum Fault { + Before(CacheError), + After(CacheError), +} + +#[derive(Default)] +struct FaultStore { + inner: MockObjectStore, + faults: BTreeMap>, + puts: BTreeMap, + deletes: usize, +} + +impl FaultStore { + fn fail(&mut self, key: &str, fault: Fault, count: usize) { + self.faults + .entry(key.to_string()) + .or_default() + .extend(vec![fault; count]); + } + + fn puts(&self, key: &str) -> usize { + self.puts.get(key).copied().unwrap_or(0) + } +} + +impl ObjectStore for FaultStore { + fn put(&mut self, key: &str, body: &[u8]) -> Result<(), CacheError> { + *self.puts.entry(key.to_string()).or_default() += 1; + let fault = self.faults.get_mut(key).and_then(VecDeque::pop_front); + if let Some(Fault::Before(error)) = &fault { + return Err(error.clone()); + } + self.inner.put(key, body)?; + if let Some(Fault::After(error)) = fault { + return Err(error); + } + Ok(()) + } + + fn head(&self, key: &str) -> Result { + self.inner.head(key) + } + fn range_get(&self, key: &str, offset: u64, length: u64) -> Result, CacheError> { + self.inner.range_get(key, offset, length) + } + fn delete(&mut self, key: &str) -> Result<(), CacheError> { + self.deletes += 1; + self.inner.delete(key) + } +} + +fn pack(namespace: &str, byte: u8) -> Pack { + let payload = vec![byte; 32]; + let mut identity = synthetic_identity(byte, 1); + identity.prefix_chain[0].0[0] = byte; + pack_blocks( + namespace, + &[PackedBlock { + generation: u64::from(byte), + identity, + checksum: content_checksum(&payload), + payload, + }], + ) + .unwrap() +} + +#[test] +fn permanent_failure_is_isolated_without_blocking_healthy_work() { + let bad = ColdJob::from_pack(&pack("denied", 1)); + let good = ColdJob::from_pack(&pack("healthy", 2)); + let bad_weight = bad.byte_weight(); + let mut sink = AsyncSink::new(2, bad_weight + good.byte_weight()).unwrap(); + let mut store = FaultStore::default(); + store.fail(&bad.object_key, Fault::Before(CacheError::Unauthorized), 8); + sink.enqueue(bad.clone()).unwrap(); + sink.enqueue(good.clone()).unwrap(); + let report = sink.drain_report(&mut store, usize::MAX).unwrap(); + assert_eq!( + report, + DrainReport { + attempted: 2, + completed: 1, + failed: 1, + first_error: Some(CacheError::Unauthorized), + ..DrainReport::default() + } + ); + assert_eq!((sink.queued(), sink.failed_count()), (0, 1)); + assert!(store.head(&good.manifest_key).is_ok()); + assert_eq!((sink.queued_bytes(), sink.failed_bytes()), (0, bad_weight)); + assert_eq!( + sink.take_failed().unwrap(), + FailedColdJob { + job: bad, + error: CacheError::Unauthorized, + attempts: 1, + } + ); + assert_eq!(sink.retained_bytes(), 0); + assert!(sink.take_failed().is_none()); + sink.enqueue(good).unwrap(); +} + +#[test] +fn transient_failure_rotates_and_each_round_attempts_a_job_once() { + let bad = ColdJob::from_pack(&pack("retry", 1)); + let good = ColdJob::from_pack(&pack("healthy", 2)); + let mut sink = AsyncSink::new(2, bad.byte_weight() + good.byte_weight()).unwrap(); + let mut store = FaultStore::default(); + store.fail(&bad.object_key, Fault::Before(CacheError::TimedOut), 1); + sink.enqueue(bad.clone()).unwrap(); + sink.enqueue(good.clone()).unwrap(); + let first = sink.drain_report(&mut store, 1).unwrap(); + assert_eq!((first.attempted, first.requeued), (1, 1)); + assert_eq!( + sink.retained_bytes(), + bad.byte_weight() + good.byte_weight() + ); + assert_eq!(sink.drain_report(&mut store, 1).unwrap().completed, 1); + assert!(store.head(&good.manifest_key).is_ok()); + assert_eq!(store.puts(&bad.object_key), 1); + assert_eq!(sink.queued_bytes(), bad.byte_weight()); + assert_eq!( + sink.drain_report(&mut store, usize::MAX).unwrap().completed, + 1 + ); + assert_eq!(store.puts(&bad.object_key), 2); + assert_eq!(sink.retained_bytes(), 0); +} + +#[test] +fn retry_exhaustion_is_bounded_and_failed_jobs_keep_both_credits() { + let job = ColdJob::from_pack(&pack("unavailable", 1)); + let mut sink = AsyncSink::new(1, job.byte_weight()).unwrap(); + let mut store = FaultStore::default(); + store.fail( + &job.object_key, + Fault::Before(CacheError::Storage { + reason: "object store", + }), + 9, + ); + sink.enqueue(job.clone()).unwrap(); + for attempt in 1..=MAX_WRITE_ATTEMPTS { + assert_eq!( + sink.drain_report(&mut store, usize::MAX).unwrap().attempted, + 1 + ); + assert_eq!(store.puts(&job.object_key), usize::from(attempt)); + assert_eq!(sink.retained_bytes(), job.byte_weight()); + } + assert_eq!((sink.queued(), sink.failed_count()), (0, 1)); + assert_eq!( + sink.drain_report(&mut store, usize::MAX).unwrap(), + DrainReport::default() + ); + assert!(matches!( + sink.enqueue(job.clone()), + Err(CacheError::QuotaExceeded { .. }) + )); + assert_eq!(sink.take_failed().unwrap().attempts, MAX_WRITE_ATTEMPTS); + assert_eq!(sink.retained_bytes(), 0); + sink.enqueue(job).unwrap(); +} + +#[test] +fn failed_jobs_share_the_byte_budget_even_with_free_job_slots() { + let job = ColdJob::from_pack(&pack("budget", 1)); + let weight = job.byte_weight(); + let mut sink = AsyncSink::new(3, weight).unwrap(); + let mut store = FaultStore::default(); + store.fail(&job.object_key, Fault::Before(CacheError::Unauthorized), 1); + sink.enqueue(job.clone()).unwrap(); + sink.drain_report(&mut store, 1).unwrap(); + assert_eq!(sink.failed_bytes(), weight); + assert!(matches!( + sink.enqueue(job.clone()), + Err(CacheError::QuotaExceeded { .. }) + )); + sink.take_failed().unwrap(); + sink.enqueue(job).unwrap(); +} + +#[test] +fn failed_jobs_share_the_job_limit_even_with_free_byte_credit() { + let job = ColdJob::from_pack(&pack("job-budget", 1)); + let mut sink = AsyncSink::new(1, job.byte_weight() * 2).unwrap(); + let mut store = FaultStore::default(); + store.fail(&job.object_key, Fault::Before(CacheError::Unauthorized), 1); + sink.enqueue(job.clone()).unwrap(); + sink.drain_report(&mut store, 1).unwrap(); + assert!(sink.retained_bytes() < sink.max_bytes); + assert!(matches!( + sink.enqueue(job.clone()), + Err(CacheError::QuotaExceeded { .. }) + )); + sink.take_failed().unwrap(); + sink.enqueue(job).unwrap(); +} + +#[test] +fn compatibility_drain_reports_error_after_other_jobs_make_progress() { + let bad = ColdJob::from_pack(&pack("denied", 1)); + let good = ColdJob::from_pack(&pack("healthy", 2)); + let mut sink = AsyncSink::new(2, bad.byte_weight() + good.byte_weight()).unwrap(); + let mut store = FaultStore::default(); + store.fail(&bad.object_key, Fault::Before(CacheError::Unauthorized), 1); + sink.enqueue(bad).unwrap(); + sink.enqueue(good.clone()).unwrap(); + assert_eq!(sink.drain(&mut store, 2), Err(CacheError::Unauthorized)); + assert!(store.head(&good.manifest_key).is_ok()); + assert_eq!(sink.failed_count(), 1); +} + +#[test] +fn ambiguous_manifest_put_preserves_a_published_object_and_retries_idempotently() { + let pack = pack("ack-lost", 1); + let job = ColdJob::from_pack(&pack); + let mut sink = AsyncSink::new(1, job.byte_weight()).unwrap(); + let mut store = FaultStore::default(); + store.fail(&pack.manifest_key, Fault::After(CacheError::TimedOut), 1); + sink.enqueue(job).unwrap(); + assert_eq!(sink.drain(&mut store, 1), Err(CacheError::TimedOut)); + assert!(store.head(&pack.manifest_key).is_ok()); + assert_eq!( + store + .range_get(&pack.object_key, 0, pack.object_bytes.len() as u64) + .unwrap(), + pack.object_bytes + ); + assert_eq!(store.deletes, 0); + assert_eq!(sink.drain(&mut store, 1), Ok(1)); + assert_eq!((sink.retained_bytes(), store.deletes), (0, 0)); +} + +#[test] +fn failed_manifest_before_commit_retains_the_unreferenced_object() { + let pack = pack("unpublished", 1); + let job = ColdJob::from_pack(&pack); + let mut sink = AsyncSink::new(1, job.byte_weight()).unwrap(); + let mut store = FaultStore::default(); + store.fail(&pack.manifest_key, Fault::Before(CacheError::TimedOut), 1); + sink.enqueue(job).unwrap(); + assert_eq!(sink.drain(&mut store, 1), Err(CacheError::TimedOut)); + assert!(store.head(&pack.object_key).is_ok()); + assert!(store.head(&pack.manifest_key).is_err()); + assert_eq!(store.deletes, 0); + assert_eq!(sink.drain(&mut store, 1), Ok(1)); +} + +#[test] +fn object_put_ack_loss_does_not_publish_or_delete_prematurely() { + let pack = pack("object-ack-lost", 1); + let job = ColdJob::from_pack(&pack); + let mut sink = AsyncSink::new(1, job.byte_weight()).unwrap(); + let mut store = FaultStore::default(); + store.fail(&pack.object_key, Fault::After(CacheError::TimedOut), 1); + sink.enqueue(job).unwrap(); + assert_eq!(sink.drain(&mut store, 1), Err(CacheError::TimedOut)); + assert!(store.head(&pack.object_key).is_ok()); + assert!(store.head(&pack.manifest_key).is_err()); + assert_eq!(sink.drain(&mut store, 1), Ok(1)); + assert_eq!(store.deletes, 0); +} + +#[test] +fn interleaved_retries_in_one_namespace_cannot_overwrite_another_pack() { + let old = pack("shared", 1); + let new = pack("shared", 2); + assert_ne!(old.object_key, new.object_key); + assert_ne!(old.manifest_key, new.manifest_key); + let old_job = ColdJob::from_pack(&old); + let new_job = ColdJob::from_pack(&new); + let mut sink = AsyncSink::new(2, old_job.byte_weight() + new_job.byte_weight()).unwrap(); + let mut store = FaultStore::default(); + store.fail(&old.manifest_key, Fault::Before(CacheError::TimedOut), 1); + sink.enqueue(old_job).unwrap(); + sink.enqueue(new_job).unwrap(); + let first = sink.drain_report(&mut store, usize::MAX).unwrap(); + assert_eq!( + (first.attempted, first.completed, first.requeued), + (2, 1, 1) + ); + assert_eq!(sink.drain(&mut store, 1), Ok(1)); + for pack in [old, new] { + assert_eq!( + store + .range_get(&pack.manifest_key, 0, pack.manifest_bytes.len() as u64) + .unwrap(), + pack.manifest_bytes + ); + assert_eq!( + store + .range_get(&pack.object_key, 0, pack.object_bytes.len() as u64) + .unwrap(), + pack.object_bytes + ); + } +} + +#[test] +fn malformed_public_jobs_do_not_consume_admission_credits() { + let job = ColdJob::from_pack(&pack("validated", 1)); + let mut sink = AsyncSink::new(1, job.byte_weight() * 2).unwrap(); + for mutated in 0..4 { + let mut bad = job.clone(); + match mutated { + 0 => bad.object_key = "mutable/object".to_string(), + 1 => bad.manifest_key = "mutable/manifest".to_string(), + 2 => bad.object_bytes[0] ^= 1, + _ => bad.manifest_bytes[0] ^= 1, + } + assert!(sink.enqueue(bad).is_err()); + assert_eq!((sink.queued(), sink.retained_bytes()), (0, 0)); + } + sink.enqueue(job).unwrap(); +} + +#[test] +fn cancellation_releases_pending_and_failed_credits_and_stays_closed() { + let bad = ColdJob::from_pack(&pack("denied", 1)); + let good = ColdJob::from_pack(&pack("healthy", 2)); + let mut sink = AsyncSink::new(2, bad.byte_weight() + good.byte_weight()).unwrap(); + let mut store = FaultStore::default(); + store.fail(&bad.object_key, Fault::Before(CacheError::Unauthorized), 1); + sink.enqueue(bad).unwrap(); + sink.enqueue(good.clone()).unwrap(); + sink.drain_report(&mut store, 1).unwrap(); + assert_eq!((sink.queued(), sink.failed_count()), (1, 1)); + sink.cancel(); + assert_eq!( + (sink.queued(), sink.failed_count(), sink.retained_bytes()), + (0, 0, 0) + ); + assert!(sink.take_failed().is_none()); + assert_eq!(sink.drain(&mut store, 4), Err(CacheError::Cancelled)); + assert_eq!(sink.drain_report(&mut store, 4), Err(CacheError::Cancelled)); + assert_eq!(sink.enqueue(good.clone()), Err(CacheError::Cancelled)); + assert_eq!(store.puts(&good.object_key), 0); +} + +#[test] +fn zero_work_and_empty_queue_do_not_touch_the_store() { + let job = ColdJob::from_pack(&pack("idle", 1)); + let mut sink = AsyncSink::new(1, job.byte_weight()).unwrap(); + let mut store = FaultStore::default(); + assert_eq!( + sink.drain_report(&mut store, usize::MAX).unwrap(), + DrainReport::default() + ); + sink.enqueue(job.clone()).unwrap(); + assert_eq!( + sink.drain_report(&mut store, 0).unwrap(), + DrainReport::default() + ); + assert_eq!(sink.retained_bytes(), job.byte_weight()); + assert!(store.puts.is_empty()); +} + +#[test] +fn metadata_and_payload_admissions_are_bounded() { + assert!(AsyncSink::new(0, 1).is_err()); + assert!(AsyncSink::new(MAX_SINK_JOBS + 1, 1).is_err()); + assert!(AsyncSink::new(1, 0).is_err()); + let job = ColdJob::from_pack(&pack("bounds", 1)); + let mut sink = AsyncSink::new(1, job.byte_weight() - 1).unwrap(); + assert!(matches!( + sink.enqueue(job), + Err(CacheError::ObjectTooLarge { .. }) + )); + assert_eq!(sink.retained_bytes(), 0); +} diff --git a/crates/cold/tests/refill_safety.rs b/crates/cold/tests/refill_safety.rs new file mode 100644 index 0000000..20c1948 --- /dev/null +++ b/crates/cold/tests/refill_safety.rs @@ -0,0 +1,443 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 + +use std::cell::Cell; + +use gpucache_cold::{ + pack_blocks, read_manifest, refill_degraded, refill_for_tenant, refill_generation, + refill_matching, AsyncSink, ColdAvailability, ColdJob, Manifest, MockObjectStore, ObjectMeta, + ObjectStore, Pack, PackedBlock, RefillBudget, RefillOutcome, MANIFEST_VERSION, +}; +use gpucache_core_types::{ + content_checksum, synthetic_identity, AdapterIdentity, ByteWriter, CacheError, CacheIdentity, + Digest, KvDtype, MultimodalIdentity, PositionEncoding, QuantFormat, TenantId, TrustDomain, +}; + +#[derive(Default)] +struct CountStore { + inner: MockObjectStore, + manifest_ranges: Cell, + object_ranges: Cell, + heads: Cell, +} + +impl CountStore { + fn publish(&mut self, pack: &Pack) { + self.put(&pack.object_key, &pack.object_bytes).unwrap(); + self.put(&pack.manifest_key, &pack.manifest_bytes).unwrap(); + } +} + +impl ObjectStore for CountStore { + fn put(&mut self, key: &str, body: &[u8]) -> Result<(), CacheError> { + self.inner.put(key, body) + } + + fn head(&self, key: &str) -> Result { + self.heads.set(self.heads.get() + 1); + self.inner.head(key) + } + + fn range_get(&self, key: &str, offset: u64, length: u64) -> Result, CacheError> { + if key.ends_with("/manifest") { + self.manifest_ranges.set(self.manifest_ranges.get() + 1); + } else if key.ends_with("/object") { + self.object_ranges.set(self.object_ranges.get() + 1); + } + self.inner.range_get(key, offset, length) + } + + fn delete(&mut self, key: &str) -> Result<(), CacheError> { + self.inner.delete(key) + } +} + +fn block(generation: u64, byte: u8, len: usize) -> PackedBlock { + let payload = vec![byte; len]; + PackedBlock { + generation, + identity: synthetic_identity(byte, 1), + checksum: content_checksum(&payload), + payload, + } +} + +fn wide_budget() -> RefillBudget { + RefillBudget::new(8 * 1024 * 1024, 16).unwrap() +} + +#[test] +fn drain_then_refill_checks_the_budget_before_the_payload() { + let blocks = [block(1, 1, 32), block(2, 2, 4096)]; + let pack = pack_blocks("refill", &blocks).unwrap(); + let mut store = CountStore::default(); + let mut sink = AsyncSink::new(2, 8 * 1024 * 1024).unwrap(); + sink.enqueue(ColdJob::from_pack(&pack)).unwrap(); + assert_eq!(sink.drain(&mut store, 1).unwrap(), 1); + let tight = RefillBudget::new(pack.manifest_bytes.len() as u64, 16).unwrap(); + assert_eq!( + refill_generation(&store, &pack.manifest_key, 2, &tight, &blocks[1].identity).unwrap(), + RefillOutcome::OverBudget + ); + assert_eq!(store.manifest_ranges.get(), 1); + assert_eq!(store.object_ranges.get(), 0); + assert_eq!( + refill_generation( + &store, + &pack.manifest_key, + 2, + &wide_budget(), + &blocks[1].identity, + ) + .unwrap(), + RefillOutcome::Loaded { + payload: blocks[1].payload.clone(), + } + ); + assert_eq!(store.object_ranges.get(), 1); +} + +#[test] +fn huge_manifest_is_not_downloaded() { + let blocks = [block(1, 5, 8), block(2, 6, 8)]; + let pack = pack_blocks("tiny", &blocks).unwrap(); + let mut store = CountStore::default(); + store.publish(&pack); + assert_eq!( + refill_generation( + &store, + &pack.manifest_key, + 1, + &RefillBudget::new(1, 16).unwrap(), + &blocks[0].identity, + ) + .unwrap(), + RefillOutcome::OverBudget + ); + 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 block = block(1, 1, 32); + let pack = pack_blocks("total-budget", std::slice::from_ref(&block)).unwrap(); + let mut store = CountStore::default(); + store.publish(&pack); + let total = pack.manifest_bytes.len() as u64 + 32; + assert_eq!( + refill_generation( + &store, + &pack.manifest_key, + 1, + &RefillBudget::new(total - 1, 1).unwrap(), + &block.identity, + ) + .unwrap(), + RefillOutcome::OverBudget + ); + assert_eq!(store.object_ranges.get(), 0); + assert!(matches!( + refill_generation( + &store, + &pack.manifest_key, + 1, + &RefillBudget::new(total, 1).unwrap(), + &block.identity, + ) + .unwrap(), + RefillOutcome::Loaded { .. } + )); + assert_eq!(store.object_ranges.get(), 1); +} + +#[test] +fn wrong_tenant_refill_does_not_download_the_payload() { + let block = block(1, 1, 32); + let pack = pack_blocks("tenant", std::slice::from_ref(&block)).unwrap(); + let mut store = CountStore::default(); + store.publish(&pack); + let mut foreign = block.identity.clone(); + foreign.tenant = TenantId::new(b"other-tenant".to_vec()).unwrap(); + assert_eq!( + refill_matching(&store, &pack.manifest_key, 1, &wide_budget(), &foreign), + Err(CacheError::Unauthorized) + ); + assert_eq!(store.object_ranges.get(), 0); + assert_eq!( + refill_matching( + &store, + &pack.manifest_key, + 1, + &wide_budget(), + &block.identity + ) + .unwrap(), + RefillOutcome::Loaded { + payload: block.payload, + } + ); + assert_eq!(store.object_ranges.get(), 1); +} + +#[test] +fn every_public_refill_entry_requires_and_checks_the_expected_identity() { + let block = block(1, 1, 32); + let pack = pack_blocks("entrypoints", std::slice::from_ref(&block)).unwrap(); + let mut store = CountStore::default(); + store.publish(&pack); + let mut wrong = block.identity.clone(); + wrong.model_revision.0[0] ^= 1; + let budget = wide_budget(); + for result in [ + refill_generation(&store, &pack.manifest_key, 1, &budget, &wrong), + refill_matching(&store, &pack.manifest_key, 1, &budget, &wrong), + refill_degraded( + &store, + &pack.manifest_key, + 1, + &budget, + ColdAvailability::Ready, + &wrong, + ), + refill_for_tenant( + &store, + &pack.manifest_key, + 1, + &budget, + ColdAvailability::Ready, + &wrong, + ), + ] { + assert_eq!(result.unwrap(), RefillOutcome::Missing); + } + assert_eq!(store.object_ranges.get(), 0); +} + +#[test] +fn complete_identity_mismatches_are_misses_before_payload_io() { + let block = block(1, 1, 32); + let pack = pack_blocks("identity", std::slice::from_ref(&block)).unwrap(); + let mut store = CountStore::default(); + store.publish(&pack); + let base = block.identity; + let mut variants: Vec = Vec::new(); + let mut changed = base.clone(); + changed.trust_domain = TrustDomain::new(b"another-domain".to_vec()).unwrap(); + variants.push(changed); + for field in 0..4 { + let mut changed = base.clone(); + match field { + 0 => changed.model_revision.0[0] ^= 1, + 1 => changed.model_config_hash.0[0] ^= 1, + 2 => changed.tokenizer_revision.0[0] ^= 1, + _ => changed.prefix_chain[0].0[0] ^= 1, + } + variants.push(changed); + } + let mut changed = base.clone(); + changed.adapter = AdapterIdentity::Lora { + revision: Digest([9; 32]), + }; + variants.push(changed); + let mut changed = base.clone(); + changed.kv_dtype = KvDtype::F16; + variants.push(changed); + let mut changed = base.clone(); + changed.quant_format = QuantFormat::Fp8; + variants.push(changed); + let mut changed = base.clone(); + changed.layout.minor += 1; + variants.push(changed); + let mut changed = base.clone(); + changed.shard.tp_size = 2; + changed.shard.tp_rank = 1; + variants.push(changed); + let mut changed = base.clone(); + changed.position = PositionEncoding::Absolute; + variants.push(changed); + let mut changed = base.clone(); + changed.rope.theta_bits = 500_000.0f64.to_bits(); + variants.push(changed); + let mut changed = base.clone(); + changed.valid_length -= 1; + variants.push(changed); + let mut changed = base.clone(); + changed.block_token_count += 1; + variants.push(changed); + let mut changed = base.clone(); + changed.multimodal = Some(MultimodalIdentity { + processor_revision: Digest([2; 32]), + input_digest: Digest([3; 32]), + }); + variants.push(changed); + for expected in variants { + assert_eq!( + refill_matching(&store, &pack.manifest_key, 1, &wide_budget(), &expected).unwrap(), + RefillOutcome::Missing + ); + } + assert_eq!(store.object_ranges.get(), 0); +} + +fn encode_manifest(manifest: &Manifest, version: u16) -> Vec { + let mut bytes = ByteWriter::new(); + bytes.raw(b"GPMF"); + bytes.u16(version); + bytes.raw(&manifest.object_checksum.0); + bytes.bytes(manifest.object_key.as_bytes()).unwrap(); + bytes.u32(manifest.ranges.len() as u32); + for range in &manifest.ranges { + bytes.u64(range.generation); + bytes.u64(range.offset); + bytes.u64(range.length); + bytes.raw(&range.checksum.0); + if version >= 2 { + bytes.bytes(&range.tenant).unwrap(); + } + if version >= 3 { + bytes.raw(&range.identity_digest.0); + } + } + bytes.finish() +} + +#[test] +fn legacy_v1_and_v2_fixtures_are_clean_misses_without_payload_reads() { + let block = block(1, 1, 32); + let pack = pack_blocks("legacy", std::slice::from_ref(&block)).unwrap(); + let manifest = read_manifest(&pack.manifest_bytes).unwrap(); + for version in [1, 2] { + let bytes = encode_manifest(&manifest, version); + let mut store = CountStore::default(); + store.put("legacy/manifest", &bytes).unwrap(); + store.put(&pack.object_key, &pack.object_bytes).unwrap(); + assert_eq!( + refill_generation( + &store, + "legacy/manifest", + 1, + &wide_budget(), + &block.identity + ) + .unwrap(), + RefillOutcome::Missing + ); + assert_eq!(store.manifest_ranges.get(), 1); + assert_eq!(store.object_ranges.get(), 0); + } +} + +fn manifest_key(namespace: &str, bytes: &[u8]) -> String { + let digest: String = content_checksum(bytes) + .0 + .iter() + .map(|byte| format!("{byte:02x}")) + .collect(); + format!("{namespace}/{digest}/manifest") +} + +#[test] +fn malformed_v3_manifest_fails_closed_before_payload_io() { + let block = block(1, 1, 32); + let pack = pack_blocks("malformed", std::slice::from_ref(&block)).unwrap(); + let mut empty_tenant = read_manifest(&pack.manifest_bytes).unwrap(); + empty_tenant.ranges[0].tenant.clear(); + let mut duplicate = read_manifest(&pack.manifest_bytes).unwrap(); + duplicate.ranges.push(duplicate.ranges[0].clone()); + let mut overflow = read_manifest(&pack.manifest_bytes).unwrap(); + overflow.ranges[0].offset = u64::MAX; + let mut trailing = pack.manifest_bytes.clone(); + trailing.push(0); + let mut magic = pack.manifest_bytes.clone(); + magic[0] ^= 1; + for bytes in [ + b"GPMF".to_vec(), + magic, + trailing, + encode_manifest(&empty_tenant, MANIFEST_VERSION), + encode_manifest(&duplicate, MANIFEST_VERSION), + encode_manifest(&overflow, MANIFEST_VERSION), + ] { + let key = manifest_key("malformed", &bytes); + let mut store = CountStore::default(); + store.put(&key, &bytes).unwrap(); + assert!(refill_matching(&store, &key, 1, &wide_budget(), &block.identity).is_err()); + assert_eq!(store.object_ranges.get(), 0); + } +} + +#[test] +fn corrupt_payload_and_content_key_substitution_return_no_loaded_bytes() { + let block = block(1, 1, 32); + let pack = pack_blocks("corruption", std::slice::from_ref(&block)).unwrap(); + let mut store = CountStore::default(); + store.publish(&pack); + let mut corrupt = pack.object_bytes.clone(); + *corrupt.last_mut().unwrap() ^= 1; + store.put(&pack.object_key, &corrupt).unwrap(); + assert_eq!( + refill_matching( + &store, + &pack.manifest_key, + 1, + &wide_budget(), + &block.identity + ), + Err(CacheError::ChecksumMismatch) + ); + store + .put("corruption/incorrect/manifest", &pack.manifest_bytes) + .unwrap(); + assert!(refill_matching( + &store, + "corruption/incorrect/manifest", + 1, + &wide_budget(), + &block.identity, + ) + .is_err()); + assert_eq!(store.object_ranges.get(), 1); +} + +#[test] +fn invalid_expected_identity_fails_before_any_store_call() { + let mut expected = synthetic_identity(1, 1); + expected.valid_length = 0; + let store = CountStore::default(); + assert!(refill_matching(&store, "ns/manifest", 1, &wide_budget(), &expected).is_err()); + assert_eq!(store.heads.get(), 0); + assert_eq!(store.manifest_ranges.get(), 0); + assert_eq!(store.object_ranges.get(), 0); +} + +#[test] +fn missing_generation_and_range_limit_do_not_download_payloads() { + let blocks = [block(1, 1, 32), block(2, 2, 32)]; + let pack = pack_blocks("missing", &blocks).unwrap(); + let mut store = CountStore::default(); + store.publish(&pack); + assert_eq!( + refill_matching( + &store, + &pack.manifest_key, + 9, + &wide_budget(), + &blocks[0].identity + ) + .unwrap(), + RefillOutcome::Missing + ); + assert_eq!( + refill_matching( + &store, + &pack.manifest_key, + 1, + &RefillBudget::new(1_000_000, 1).unwrap(), + &blocks[0].identity + ) + .unwrap(), + RefillOutcome::OverBudget + ); + assert_eq!(store.object_ranges.get(), 0); +} diff --git a/crates/transport/src/directory.rs b/crates/transport/src/directory.rs index 9baf4b9..a3384d6 100644 --- a/crates/transport/src/directory.rs +++ b/crates/transport/src/directory.rs @@ -7,9 +7,11 @@ //! id 0. The map is keyed by the SHA-256 of a physical byte key, one entry //! per in-flight fill, not per token. Membership changes bump the epoch and //! drop in-flight fills. There is no cross-shard quorum and no payload store. -//! Credits are an admission limit, not a replica set. +//! Credits are an admission limit, not a replica set. A finished, failed, or +//! fenced fill leaves a bounded notice. That notice is not a hit and carries +//! no bytes. An epoch that cannot advance does not drop in-flight work. -use std::collections::{BTreeMap, BTreeSet}; +use std::collections::{BTreeMap, BTreeSet, VecDeque}; use gpucache_core_types::{content_checksum, CacheError, QuotaKind, HARD_BYTES_CAP}; @@ -74,6 +76,18 @@ pub enum Admit { Joined { flight_id: u64, epoch: u64 }, } +/// Observable end of one fill. None of these variants carries payload bytes. +/// `LeaderFinished` is not a cache hit: the caller still has to `Get` a +/// committed block or recompute. A missing notice is [`CacheError::UnknownGeneration`], +/// which is also a recompute. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum FillNotice { + Pending, + LeaderFinished, + LeaderFailed, + Fenced, +} + struct Flight { id: u64, epoch: u64, @@ -98,6 +112,9 @@ pub struct ShardDirectory { max_waiters: usize, max_members: usize, next_flight: u64, + notices: BTreeMap<(Vec, [u8; 16]), FillNotice>, + notice_order: VecDeque<(Vec, [u8; 16])>, + max_notices: usize, } impl ShardDirectory { @@ -126,6 +143,12 @@ impl ShardDirectory { max_waiters: params.max_waiters, max_members: params.max_members, next_flight: 1, + notices: BTreeMap::new(), + notice_order: VecDeque::new(), + max_notices: params + .max_flights + .saturating_mul(params.max_waiters.saturating_add(1)) + .max(1), } } @@ -151,7 +174,9 @@ impl ShardDirectory { /// Drop in-flight fills and bump the epoch. Credits return to the pool. /// Callers must treat the dropped fills as misses. This is not a copy. - pub fn fence_epoch(&mut self) -> u64 { + /// When the epoch cannot advance, in-flight fills stay and the error is + /// returned instead. + pub fn fence_epoch(&mut self) -> Result { self.fence_inflight() } @@ -184,19 +209,21 @@ impl ShardDirectory { if incarnation < self.members[pos].incarnation { return Err(CacheError::FencingRejected { reason: "member" }); } + self.ensure_epoch_can_advance()?; self.members[pos].incarnation = incarnation; - return Ok(self.fence_inflight()); + return self.fence_inflight(); } if self.members.len() >= self.max_members { return Err(CacheError::Malformed { reason: "membership", }); } + self.ensure_epoch_can_advance()?; self.members.push(MemberView { node_id, incarnation, }); - Ok(self.fence_inflight()) + self.fence_inflight() } pub fn remove_member(&mut self, node_id: u64) -> Result { @@ -212,8 +239,9 @@ impl ShardDirectory { if self.members.len() == 1 { return Err(CacheError::FencingRejected { reason: "member" }); } + self.ensure_epoch_can_advance()?; self.members.remove(pos); - Ok(self.fence_inflight()) + self.fence_inflight() } pub fn begin_or_join(&mut self, req: &FillRequest) -> Result { @@ -292,7 +320,12 @@ impl ShardDirectory { request_epoch: u64, request: [u8; 16], ) -> Result<(), CacheError> { - self.finish(flight_id, request_epoch, request) + self.finish( + flight_id, + request_epoch, + request, + FillNotice::LeaderFinished, + ) } pub fn fail_flight( @@ -301,7 +334,20 @@ impl ShardDirectory { request_epoch: u64, request: [u8; 16], ) -> Result<(), CacheError> { - self.finish(flight_id, request_epoch, request) + self.finish(flight_id, request_epoch, request, FillNotice::LeaderFailed) + } + + /// Status of one leader or joiner. In-flight work is `Pending`. A terminal + /// notice is not a payload. An evicted or unknown request is not a hit. + pub fn fill_notice(&self, tenant: &[u8], request: [u8; 16]) -> Result { + let key = (tenant.to_vec(), request); + if self.by_request.contains_key(&key) { + return Ok(FillNotice::Pending); + } + self.notices + .get(&key) + .copied() + .ok_or(CacheError::UnknownGeneration) } fn finish( @@ -309,6 +355,7 @@ impl ShardDirectory { flight_id: u64, request_epoch: u64, request: [u8; 16], + notice: FillNotice, ) -> Result<(), CacheError> { self.check_epoch(request_epoch)?; let Some(key) = self.by_id.get(&flight_id).copied() else { @@ -326,21 +373,59 @@ impl ShardDirectory { .ok_or(CacheError::UnknownGeneration)?; self.by_id.remove(&flight_id); self.by_request.remove(&(tenant.clone(), flight.leader)); + self.record_notice(tenant.clone(), flight.leader, notice); for waiter in flight.waiters { self.by_request.remove(&(tenant.clone(), waiter)); + self.record_notice(tenant.clone(), waiter, notice); } self.release(&tenant, bytes) } - fn fence_inflight(&mut self) -> u64 { - self.epoch = self.epoch.saturating_add(1); + fn fence_inflight(&mut self) -> Result { + let next = self.epoch.checked_add(1).ok_or(CacheError::QuotaExceeded { + kind: QuotaKind::InFlightRequests, + })?; + self.epoch = next; self.by_request.clear(); self.by_id.clear(); let flights = std::mem::take(&mut self.flights); for (_, flight) in flights { + self.record_notice(flight.tenant.clone(), flight.leader, FillNotice::Fenced); + for waiter in &flight.waiters { + self.record_notice(flight.tenant.clone(), *waiter, FillNotice::Fenced); + } let _ = self.release(&flight.tenant, flight.bytes); } - self.epoch + Ok(self.epoch) + } + + fn ensure_epoch_can_advance(&self) -> Result<(), CacheError> { + if self.epoch == u64::MAX { + return Err(CacheError::QuotaExceeded { + kind: QuotaKind::InFlightRequests, + }); + } + Ok(()) + } + + fn record_notice(&mut self, tenant: Vec, request: [u8; 16], notice: FillNotice) { + let key = (tenant, request); + if self.notices.contains_key(&key) { + self.notice_order.retain(|existing| existing != &key); + } + self.notices.insert(key.clone(), notice); + self.notice_order.push_back(key); + while self.notices.len() > self.max_notices { + let Some(old) = self.notice_order.pop_front() else { + break; + }; + self.notices.remove(&old); + } + } + + #[cfg(test)] + pub(crate) fn set_epoch_for_test(&mut self, epoch: u64) { + self.epoch = epoch; } fn reserve(&mut self, tenant: &[u8], bytes: u64) -> Result<(), CacheError> { @@ -619,7 +704,7 @@ mod tests { .fail_flight(flight_id, epoch, 1u128.to_le_bytes()) .unwrap(), _ => { - dir.fence_epoch(); + dir.fence_epoch().unwrap(); } } assert!(dir.by_request.is_empty()); @@ -641,4 +726,151 @@ mod tests { assert_eq!(dir.epoch(), epoch); assert_eq!(dir.inflight(), 1); } + + #[test] + fn finish_and_failure_are_notices_without_bytes_and_a_fence_is_not_success() { + let mut dir = ShardDirectory::single_node(params()); + let leader = dir + .begin_or_join(&fill(b"phys-a", b"tenant-a", 10, 1, 1)) + .unwrap(); + dir.begin_or_join(&fill(b"phys-a", b"tenant-a", 10, 2, 1)) + .unwrap(); + assert_eq!( + dir.fill_notice(b"tenant-a", 2u128.to_le_bytes()).unwrap(), + FillNotice::Pending + ); + let Admit::Leader { flight_id, .. } = leader else { + panic!("leader"); + }; + dir.fail_flight(flight_id, 1, 1u128.to_le_bytes()).unwrap(); + assert_eq!(dir.inflight(), 0); + assert_eq!(dir.credits_used(), 0); + assert_eq!( + dir.fill_notice(b"tenant-a", 1u128.to_le_bytes()).unwrap(), + FillNotice::LeaderFailed + ); + assert_eq!( + dir.fill_notice(b"tenant-a", 2u128.to_le_bytes()).unwrap(), + FillNotice::LeaderFailed + ); + let again = dir + .begin_or_join(&fill(b"phys-b", b"tenant-a", 10, 3, 1)) + .unwrap(); + let Admit::Leader { flight_id, .. } = again else { + panic!("leader"); + }; + dir.begin_or_join(&fill(b"phys-b", b"tenant-a", 10, 4, 1)) + .unwrap(); + dir.complete_flight(flight_id, 1, 3u128.to_le_bytes()) + .unwrap(); + assert_eq!( + dir.fill_notice(b"tenant-a", 4u128.to_le_bytes()).unwrap(), + FillNotice::LeaderFinished + ); + assert_eq!(dir.credits_used(), 0); + let fenced = dir + .begin_or_join(&fill(b"phys-c", b"tenant-a", 10, 5, 1)) + .unwrap(); + dir.begin_or_join(&fill(b"phys-c", b"tenant-a", 10, 6, 1)) + .unwrap(); + let Admit::Leader { flight_id, epoch } = fenced else { + panic!("leader"); + }; + dir.fence_epoch().unwrap(); + assert_eq!( + dir.complete_flight(flight_id, epoch, 5u128.to_le_bytes()) + .unwrap_err(), + CacheError::EpochMismatch { + store: 2, + request: 1, + } + ); + assert_eq!( + dir.fill_notice(b"tenant-a", 5u128.to_le_bytes()).unwrap(), + FillNotice::Fenced + ); + assert_eq!( + dir.fill_notice(b"tenant-a", 6u128.to_le_bytes()).unwrap(), + FillNotice::Fenced + ); + assert_ne!( + dir.fill_notice(b"tenant-a", 5u128.to_le_bytes()).unwrap(), + FillNotice::LeaderFinished + ); + assert_eq!( + dir.fill_notice(b"tenant-a", 9u128.to_le_bytes()) + .unwrap_err(), + CacheError::UnknownGeneration + ); + } + + #[test] + fn notice_table_drops_the_oldest_instead_of_growing() { + let mut dir = ShardDirectory::single_node(ShardParams::new(100, 100, 1, 1, 2).unwrap()); + assert_eq!(dir.max_notices, 2); + for request in 1u128..=3 { + let epoch = dir.epoch(); + let Admit::Leader { flight_id, .. } = dir + .begin_or_join(&fill(b"phys", b"tenant-a", 10, request, epoch)) + .unwrap() + else { + panic!("leader"); + }; + dir.complete_flight(flight_id, epoch, request.to_le_bytes()) + .unwrap(); + } + assert_eq!( + dir.fill_notice(b"tenant-a", 1u128.to_le_bytes()) + .unwrap_err(), + CacheError::UnknownGeneration + ); + assert_eq!( + dir.fill_notice(b"tenant-a", 3u128.to_le_bytes()).unwrap(), + FillNotice::LeaderFinished + ); + assert_eq!(dir.notices.len(), 2); + } + + #[test] + fn epoch_overflow_does_not_drop_the_fill_or_publish_a_new_member() { + let mut dir = ShardDirectory::single_node(params()); + dir.set_epoch_for_test(u64::MAX); + let admit = dir + .begin_or_join(&fill(b"phys-a", b"tenant-a", 10, 1, u64::MAX)) + .unwrap(); + let Admit::Leader { flight_id, .. } = admit else { + panic!("leader"); + }; + let err = dir.note_member(1, 2).unwrap_err(); + assert_eq!( + err, + CacheError::QuotaExceeded { + kind: QuotaKind::InFlightRequests, + } + ); + assert_eq!(dir.epoch(), u64::MAX); + assert_eq!(dir.inflight(), 1); + assert_eq!(dir.credits_used(), 10); + assert_eq!(dir.members()[0].incarnation, 1); + assert_eq!( + dir.fill_notice(b"tenant-a", 1u128.to_le_bytes()).unwrap(), + FillNotice::Pending + ); + let err = dir.fence_epoch().unwrap_err(); + assert_eq!( + err, + CacheError::QuotaExceeded { + kind: QuotaKind::InFlightRequests, + } + ); + assert_eq!(dir.inflight(), 1); + dir.complete_flight(flight_id, u64::MAX, 1u128.to_le_bytes()) + .unwrap(); + assert_eq!(dir.inflight(), 0); + assert_eq!(dir.credits_used(), 0); + assert_eq!( + dir.fill_notice(b"tenant-a", 1u128.to_le_bytes()).unwrap(), + FillNotice::LeaderFinished + ); + } } diff --git a/crates/transport/src/iface.rs b/crates/transport/src/iface.rs index 2cff1f0..b5ff494 100644 --- a/crates/transport/src/iface.rs +++ b/crates/transport/src/iface.rs @@ -26,11 +26,26 @@ pub struct TransferDesc<'a> { pub bytes: &'a [u8], } +/// Why `submit_preferring_rdma` did not stay on the preferred path. +/// +/// `RdmaNotLinked` is the CPU contract for this tree. It is not a NIC retry +/// and it is not a measured fallback latency. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum FallbackReason { + /// The receipt's `kind` is the path that was asked for. + None, + /// The RDMA slot refused to open. Bytes were staged on TCP instead. + RdmaNotLinked, +} + /// What `submit` actually did. `staged_bytes` is the length copied into the /// host staging buffer. It is not a wire-byte count and not a latency. +/// `attempted` is the path the caller preferred. `kind` is the path that ran. #[derive(Clone, Debug, PartialEq, Eq)] pub struct TransferReceipt { pub kind: TransportKind, + pub attempted: TransportKind, + pub fallback: FallbackReason, pub host_copies: u32, pub registration_reused: bool, pub dma_posted: bool, @@ -100,6 +115,8 @@ impl KvTransport for TcpTransport { drop(staged); Ok(TransferReceipt { kind: TransportKind::Tcp, + attempted: TransportKind::Tcp, + fallback: FallbackReason::None, host_copies: 1, registration_reused: false, dma_posted: false, @@ -109,6 +126,54 @@ impl KvTransport for TcpTransport { } } +/// Prefer the RDMA slot, then stage on TCP when that slot is not linked. +/// +/// This build's `NixlTransport::open` always fails closed. The receipt then +/// names TCP as the path that ran and `RdmaNotLinked` as the reason. `dma_posted` +/// stays false. The function does not invent a NIC retry or a bandwidth number. +pub fn submit_preferring_rdma( + tcp: &mut TcpTransport, + desc: &TransferDesc<'_>, +) -> Result { + match NixlTransport::open() { + Ok(mut nixl) => match nixl.submit(desc) { + Ok(receipt) => { + if receipt.dma_posted || receipt.hardware != HardwarePath::NotRunHardware { + return Err(CacheError::Storage { + reason: "rdma not linked", + }); + } + Ok(TransferReceipt { + attempted: TransportKind::RdmaNixl, + fallback: FallbackReason::None, + ..receipt + }) + } + Err(CacheError::Storage { + reason: "rdma not linked", + }) => tcp_after_rdma_refusal(tcp, desc), + Err(err) => Err(err), + }, + Err(CacheError::Storage { + reason: "rdma not linked", + }) => tcp_after_rdma_refusal(tcp, desc), + Err(err) => Err(err), + } +} + +fn tcp_after_rdma_refusal( + tcp: &mut TcpTransport, + desc: &TransferDesc<'_>, +) -> Result { + let mut receipt = tcp.submit(desc)?; + receipt.attempted = TransportKind::RdmaNixl; + receipt.kind = TransportKind::Tcp; + receipt.fallback = FallbackReason::RdmaNotLinked; + receipt.dma_posted = false; + receipt.hardware = HardwarePath::CpuContract; + Ok(receipt) +} + /// NIXL slot. Constructing this value does not open a device. /// /// `open` fails in every build, including `--features rdma-nixl`. libnixl is @@ -164,6 +229,8 @@ mod tests { let mut tcp = TcpTransport::new(64).unwrap(); let receipt = tcp.submit(&TransferDesc { bytes: b"kv-bytes" }).unwrap(); assert_eq!(receipt.kind, TransportKind::Tcp); + assert_eq!(receipt.attempted, TransportKind::Tcp); + assert_eq!(receipt.fallback, FallbackReason::None); assert_eq!(receipt.host_copies, 1); assert!(!receipt.dma_posted); assert!(!receipt.registration_reused); @@ -205,4 +272,34 @@ mod tests { }) )); } + + #[test] + fn rdma_refusal_falls_back_to_tcp_without_posting() { + let mut tcp = TcpTransport::new(32).unwrap(); + let receipt = + submit_preferring_rdma(&mut tcp, &TransferDesc { bytes: b"kv-block" }).unwrap(); + assert_eq!(receipt.attempted, TransportKind::RdmaNixl); + assert_eq!(receipt.kind, TransportKind::Tcp); + assert_eq!(receipt.fallback, FallbackReason::RdmaNotLinked); + assert!(!receipt.dma_posted); + assert_eq!(receipt.hardware, HardwarePath::CpuContract); + assert_eq!(receipt.host_copies, 1); + assert_eq!(receipt.staged_bytes, 8); + assert_eq!(tcp.submits(), 1); + assert_eq!(rdma_probe(), "NOT_RUN_HARDWARE"); + } + + #[test] + fn rdma_fallback_keeps_tcp_backpressure() { + let mut tcp = TcpTransport::new(2).unwrap(); + let err = submit_preferring_rdma(&mut tcp, &TransferDesc { bytes: b"abcd" }).unwrap_err(); + assert_eq!( + err, + CacheError::QuotaExceeded { + kind: QuotaKind::InFlightBytes, + } + ); + assert_eq!(tcp.submits(), 0); + assert_eq!(tcp.staged_bytes(), 0); + } } diff --git a/crates/transport/src/lib.rs b/crates/transport/src/lib.rs index eca1b5d..8178411 100644 --- a/crates/transport/src/lib.rs +++ b/crates/transport/src/lib.rs @@ -26,11 +26,11 @@ mod stage; pub use client::TcpClient; pub use demo::{run_share_demo, ShareReport}; -pub use directory::{Admit, FillRequest, MemberView, ShardDirectory, ShardParams}; +pub use directory::{Admit, FillNotice, FillRequest, MemberView, ShardDirectory, ShardParams}; pub use fault_demo::{run_fault_demo, FaultReport}; pub use iface::{ - rdma_feature_selected, rdma_probe, HardwarePath, KvTransport, NixlTransport, TcpTransport, - TransferDesc, TransferReceipt, TransportKind, + rdma_feature_selected, rdma_probe, submit_preferring_rdma, FallbackReason, HardwarePath, + KvTransport, NixlTransport, TcpTransport, TransferDesc, TransferReceipt, TransportKind, }; pub use plane::{ControlPlane, RebalancePlan, UpgradeStamp}; pub use rate::RateGate; diff --git a/crates/transport/src/plane.rs b/crates/transport/src/plane.rs index 2488677..15e561f 100644 --- a/crates/transport/src/plane.rs +++ b/crates/transport/src/plane.rs @@ -8,9 +8,9 @@ use std::collections::BTreeMap; -use gpucache_core_types::CacheError; +use gpucache_core_types::{CacheError, QuotaKind}; -use crate::directory::{Admit, FillRequest, ShardDirectory, ShardParams}; +use crate::directory::{Admit, FillNotice, FillRequest, ShardDirectory, ShardParams}; #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum Phase { @@ -88,6 +88,10 @@ impl ControlPlane { self.dir.complete_flight(flight_id, request_epoch, request) } + pub fn fill_notice(&self, tenant: &[u8], request: [u8; 16]) -> Result { + self.dir.fill_notice(tenant, request) + } + pub fn assign_domain(&mut self, node_id: u64, domain: u16) -> Result<(), CacheError> { if !self .dir @@ -105,6 +109,7 @@ impl ControlPlane { /// Remove every member in `domain`. The last remaining member stays. /// In-flight fills are fenced and become misses. No payload is copied. + /// If the epoch cannot advance for the entire batch, nothing is changed. pub fn isolate_domain(&mut self, domain: u16) -> Result { let victims: Vec = self .dir @@ -121,6 +126,14 @@ impl ControlPlane { if victims.len() >= self.dir.members().len() { return Err(CacheError::FencingRejected { reason: "member" }); } + // Each removal fences once. Reserve the entire epoch range before + // changing membership, domains, credits, or any in-flight notice. + self.dir + .epoch() + .checked_add(victims.len() as u64) + .ok_or(CacheError::QuotaExceeded { + kind: QuotaKind::InFlightRequests, + })?; let mut epoch = self.dir.epoch(); for node_id in victims { epoch = self.dir.remove_member(node_id)?; @@ -140,7 +153,7 @@ impl ControlPlane { reason: "rebalance", }); } - let source_epoch = self.dir.fence_epoch(); + let source_epoch = self.dir.fence_epoch()?; let destination = ShardDirectory::open(to_shard, self.params.clone()); Ok(RebalancePlan { from_shard: self.dir.shard_id(), @@ -162,7 +175,7 @@ impl ControlPlane { if self.phase != Phase::Draining { return Err(CacheError::Malformed { reason: "upgrade" }); } - let epoch = self.dir.fence_epoch(); + let epoch = self.dir.fence_epoch()?; self.software_generation = self.software_generation.saturating_add(1); self.upgraded = true; self.phase = Phase::Serving; @@ -188,7 +201,7 @@ impl ControlPlane { if !self.upgraded { return Err(CacheError::Malformed { reason: "upgrade" }); } - let epoch = self.dir.fence_epoch(); + let epoch = self.dir.fence_epoch()?; Ok(UpgradeStamp { generation: self.software_generation, epoch, @@ -215,6 +228,34 @@ mod tests { } } + fn three_member_plane(epoch: u64) -> ControlPlane { + let mut plane = ControlPlane::single_node(params()); + plane.dir.note_member(2, 1).unwrap(); + plane.dir.note_member(3, 1).unwrap(); + plane.assign_domain(1, 7).unwrap(); + plane.assign_domain(2, 7).unwrap(); + plane.assign_domain(3, 9).unwrap(); + plane.dir.set_epoch_for_test(epoch); + plane + } + + fn begin_leader_and_waiter(plane: &mut ControlPlane) -> (u64, FillRequest, FillRequest) { + let leader = fill(plane.epoch()); + let Admit::Leader { flight_id, .. } = plane.begin_or_join(&leader).unwrap() else { + panic!("leader"); + }; + let mut waiter = leader.clone(); + waiter.request = 2u128.to_le_bytes(); + assert_eq!( + plane.begin_or_join(&waiter).unwrap(), + Admit::Joined { + flight_id, + epoch: plane.epoch() + } + ); + (flight_id, leader, waiter) + } + #[test] fn rebalance_fences_the_source_and_starts_an_empty_destination() { let mut plane = ControlPlane::single_node(params()); @@ -234,6 +275,10 @@ mod tests { .unwrap_err(); assert!(matches!(err, CacheError::EpochMismatch { .. })); assert_eq!(plan.source_epoch, plane.epoch()); + assert_eq!( + plane.fill_notice(b"tenant-a", 1u128.to_le_bytes()).unwrap(), + FillNotice::Fenced + ); } #[test] @@ -280,6 +325,12 @@ mod tests { let node = plane.dir.members()[0].node_id; plane.assign_domain(node, 7).unwrap(); plane.begin_or_join(&fill(1)).unwrap(); + assert_eq!( + plane.isolate_domain(99).unwrap_err(), + CacheError::Malformed { + reason: "fault domain" + } + ); let err = plane.isolate_domain(7).unwrap_err(); assert_eq!(err, CacheError::FencingRejected { reason: "member" }); assert_eq!(plane.inflight(), 1); @@ -292,4 +343,113 @@ mod tests { assert_eq!(plane.inflight(), 0); assert!(plane.dir.members().iter().all(|member| member.node_id != 9)); } + + #[test] + fn isolate_domain_epoch_shortage_keeps_the_entire_batch_and_fills_unchanged() { + let mut plane = three_member_plane(u64::MAX - 1); + let (flight_id, leader, waiter) = begin_leader_and_waiter(&mut plane); + let members = plane.dir.members().to_vec(); + let domains = plane.domains.clone(); + + assert_eq!( + plane.isolate_domain(7).unwrap_err(), + CacheError::QuotaExceeded { + kind: QuotaKind::InFlightRequests + } + ); + assert_eq!(plane.epoch(), u64::MAX - 1); + assert_eq!(plane.dir.members(), members); + assert_eq!(plane.domains, domains); + assert_eq!(plane.inflight(), 1); + assert_eq!(plane.credits_used(), leader.bytes); + assert!(plane.serving()); + for request in [&leader, &waiter] { + assert_eq!( + plane.fill_notice(&request.tenant, request.request).unwrap(), + FillNotice::Pending + ); + } + assert_eq!( + plane.begin_or_join(&leader).unwrap(), + Admit::Leader { + flight_id, + epoch: leader.epoch + } + ); + assert_eq!( + plane.begin_or_join(&waiter).unwrap(), + Admit::Joined { + flight_id, + epoch: waiter.epoch + } + ); + assert_eq!(plane.credits_used(), leader.bytes); + plane + .complete_flight(flight_id, leader.epoch, leader.request) + .unwrap(); + assert_eq!(plane.credits_used(), 0); + assert_eq!(plane.inflight(), 0); + } + + #[test] + fn isolate_domain_with_exact_epoch_budget_removes_every_victim_and_fences_fills() { + let mut plane = three_member_plane(u64::MAX - 2); + let (flight_id, leader, waiter) = begin_leader_and_waiter(&mut plane); + + assert_eq!(plane.isolate_domain(7).unwrap(), u64::MAX); + assert_eq!(plane.epoch(), u64::MAX); + assert_eq!(plane.dir.members().len(), 1); + assert_eq!(plane.dir.members()[0].node_id, 3); + assert_eq!(plane.domains, BTreeMap::from([(3, 9)])); + assert_eq!(plane.inflight(), 0); + assert_eq!(plane.credits_used(), 0); + assert!(plane.serving()); + for request in [&leader, &waiter] { + assert_eq!( + plane.fill_notice(&request.tenant, request.request).unwrap(), + FillNotice::Fenced + ); + } + assert_eq!( + plane + .complete_flight(flight_id, leader.epoch, leader.request) + .unwrap_err(), + CacheError::EpochMismatch { + store: u64::MAX, + request: leader.epoch + } + ); + } + + #[test] + fn isolate_domain_at_epoch_max_keeps_a_single_victim_and_current_fills() { + let mut plane = three_member_plane(u64::MAX); + plane.assign_domain(2, 9).unwrap(); + let (flight_id, leader, waiter) = begin_leader_and_waiter(&mut plane); + let members = plane.dir.members().to_vec(); + let domains = plane.domains.clone(); + + assert_eq!( + plane.isolate_domain(7).unwrap_err(), + CacheError::QuotaExceeded { + kind: QuotaKind::InFlightRequests + } + ); + assert_eq!(plane.epoch(), u64::MAX); + assert_eq!(plane.dir.members(), members); + assert_eq!(plane.domains, domains); + assert_eq!(plane.inflight(), 1); + assert_eq!(plane.credits_used(), leader.bytes); + for request in [&leader, &waiter] { + assert_eq!( + plane.fill_notice(&request.tenant, request.request).unwrap(), + FillNotice::Pending + ); + } + plane + .complete_flight(flight_id, leader.epoch, leader.request) + .unwrap(); + assert_eq!(plane.inflight(), 0); + assert_eq!(plane.credits_used(), 0); + } } diff --git a/crates/transport/src/registered.rs b/crates/transport/src/registered.rs index a16646f..0267c1f 100644 --- a/crates/transport/src/registered.rs +++ b/crates/transport/src/registered.rs @@ -4,9 +4,10 @@ //! Bounded buffer pool that *simulates* pin/registration. //! //! Slots are ordinary `Vec` allocations. Nothing here calls `ibv_reg_mr` -//! or a NIC. Unknown completion quarantines a slot. Lease expiry does not -//! free it. A new region id is assigned only when an allocation is bound -//! again; retired ids are not recycled. +//! or a NIC. Checkout prefers a slot that already has a binding, and a batch +//! that cannot be satisfied is left untouched. Unknown completion quarantines +//! a slot. Lease expiry does not free it. A new region id is assigned only +//! when an allocation is bound again; retired ids are not recycled. use gpucache_core_types::{CacheError, QuotaKind}; @@ -167,63 +168,99 @@ impl RegisteredPool { Ok(()) } - /// Take `n` idle slots. Already-bound slots do not count as a new - /// registration. The call changes nothing when it returns an error. + /// Take `n` idle slots. Already-bound slots are preferred and do not count + /// as a new registration. The call changes nothing when it returns an error, + /// including when the region-id space cannot cover every new binding. pub fn checkout_batch(&mut self, n: usize) -> Result { if n == 0 || n > self.regions.len() { return Err(CacheError::Malformed { reason: "buffer pool", }); } - let available = self - .regions - .iter() - .filter(|region| matches!(region.state, RegionState::Unbound | RegionState::Idle)) - .count(); - if available < n { + if self.next_batch == u64::MAX { return Err(CacheError::QuotaExceeded { kind: QuotaKind::Registrations, }); } - if self.next_batch == u64::MAX { + let mut idle = Vec::new(); + let mut unbound = Vec::new(); + for (index, region) in self.regions.iter().enumerate() { + match region.state { + RegionState::Idle => { + if region.id.is_none() { + return Err(CacheError::Storage { + reason: "buffer pool", + }); + } + idle.push(index); + } + RegionState::Unbound => unbound.push(index), + RegionState::CheckedOut { .. } | RegionState::Quarantined { .. } => {} + } + } + if idle.len() + unbound.len() < n { return Err(CacheError::QuotaExceeded { kind: QuotaKind::Registrations, }); } + let mut chosen = Vec::with_capacity(n); + for index in idle { + if chosen.len() == n { + break; + } + chosen.push(index); + } + let fresh = n - chosen.len(); + let available_ids = u64::MAX - self.next_id; + if fresh as u64 > available_ids { + return Err(CacheError::QuotaExceeded { + kind: QuotaKind::Registrations, + }); + } + for index in unbound.into_iter().take(fresh) { + chosen.push(index); + } + let batch = self.next_batch; - self.next_batch += 1; - let mut slots = Vec::with_capacity(n); - let mut index = 0; - while slots.len() < n && index < self.regions.len() { - let state = self.regions[index].state; - match state { + let mut next_id = self.next_id; + let mut planned = Vec::with_capacity(n); + for index in chosen { + match self.regions[index].state { RegionState::Idle => { - self.reused_registrations = self.reused_registrations.saturating_add(1); let id = self.regions[index].id.ok_or(CacheError::Storage { reason: "buffer pool", })?; - self.regions[index].state = RegionState::CheckedOut { batch }; - slots.push(id); + planned.push((index, id, false)); } RegionState::Unbound => { - if self.regions[index].had_registration { - self.backing_reused = self.backing_reused.saturating_add(1); - } - let id = self.alloc_id()?; - self.regions[index].id = Some(id); - self.regions[index].had_registration = true; - self.regions[index].state = RegionState::CheckedOut { batch }; - self.registration_ops = self.registration_ops.saturating_add(1); - slots.push(id); + let id = SlotId(next_id); + next_id = next_id.saturating_add(1); + planned.push((index, id, true)); + } + RegionState::CheckedOut { .. } | RegionState::Quarantined { .. } => { + return Err(CacheError::Storage { + reason: "buffer pool", + }); } - RegionState::CheckedOut { .. } | RegionState::Quarantined { .. } => {} } - index += 1; } - if slots.len() != n { - return Err(CacheError::Storage { - reason: "buffer pool", - }); + + self.next_batch = self.next_batch.saturating_add(1); + self.next_id = next_id; + let mut slots = Vec::with_capacity(n); + for (index, id, is_new) in planned { + if is_new { + if self.regions[index].had_registration { + self.backing_reused = self.backing_reused.saturating_add(1); + } + self.regions[index].id = Some(id); + self.regions[index].had_registration = true; + self.registration_ops = self.registration_ops.saturating_add(1); + } else { + self.reused_registrations = self.reused_registrations.saturating_add(1); + } + self.regions[index].state = RegionState::CheckedOut { batch }; + slots.push(id); } Ok(CheckedBatch { id: batch, slots }) } @@ -272,9 +309,12 @@ impl RegisteredPool { } /// Move in-flight and already-quarantined slots to fenced quarantine. - /// Does not free them. - pub fn fence_epoch(&mut self) -> u64 { - self.epoch = self.epoch.saturating_add(1); + /// Does not free them. If the epoch cannot advance, the pool is unchanged. + pub fn fence_epoch(&mut self) -> Result { + let next = self.epoch.checked_add(1).ok_or(CacheError::QuotaExceeded { + kind: QuotaKind::Registrations, + })?; + self.epoch = next; for region in &mut self.regions { if matches!( region.state, @@ -283,7 +323,7 @@ impl RegisteredPool { region.state = RegionState::Quarantined { fenced: true }; } } - self.epoch + Ok(self.epoch) } /// Free slots that fencing already marked. Unfenced quarantine stays. @@ -299,15 +339,14 @@ impl RegisteredPool { Ok(freed) } - fn alloc_id(&mut self) -> Result { - if self.next_id == u64::MAX { - return Err(CacheError::QuotaExceeded { - kind: QuotaKind::Registrations, - }); - } - let id = SlotId(self.next_id); - self.next_id += 1; - Ok(id) + #[cfg(test)] + fn set_next_id_for_test(&mut self, next_id: u64) { + self.next_id = next_id; + } + + #[cfg(test)] + fn set_epoch_for_test(&mut self, epoch: u64) { + self.epoch = epoch; } fn ensure_batch(&self, batch: &CheckedBatch) -> Result<(), CacheError> { @@ -403,7 +442,7 @@ mod tests { let id = batch.slots()[0]; pool.on_lease_expiry(id).unwrap(); assert_eq!(pool.reclaim_fenced().unwrap(), 0); - pool.fence_epoch(); + pool.fence_epoch().unwrap(); assert_eq!(pool.stats().quarantined, 1); assert_eq!(pool.reclaim_fenced().unwrap(), 1); assert_eq!(pool.stats().idle_unbound, 2); @@ -418,4 +457,62 @@ mod tests { assert_eq!(stats.idle_unbound, 2); assert_eq!(stats.registration_ops, 0); } + + #[test] + fn checkout_prefers_an_idle_registration_over_a_lower_unbound_slot() { + let mut pool = RegisteredPool::new(2, 8).unwrap(); + let first = pool.checkout_batch(1).unwrap(); + let second = pool.checkout_batch(1).unwrap(); + let idle_id = second.slots()[0]; + pool.complete_batch(&first, SimCompletion::Unknown).unwrap(); + pool.attest_drain(first.slots()[0]).unwrap(); + pool.complete_batch(&second, SimCompletion::Completed) + .unwrap(); + assert_eq!(pool.stats().idle_unbound, 1); + assert_eq!(pool.stats().idle_registered, 1); + let reused = pool.checkout_batch(1).unwrap(); + assert_eq!(reused.slots()[0], idle_id); + let stats = pool.stats(); + assert_eq!(stats.registration_ops, 2); + assert_eq!(stats.reused_registrations, 1); + assert_eq!(stats.idle_unbound, 1); + } + + #[test] + fn id_exhaustion_does_not_check_out_a_prefix_of_the_batch() { + let mut pool = RegisteredPool::new(2, 4).unwrap(); + pool.set_next_id_for_test(u64::MAX - 1); + let err = pool.checkout_batch(2).unwrap_err(); + assert_eq!( + err, + CacheError::QuotaExceeded { + kind: QuotaKind::Registrations, + } + ); + let stats = pool.stats(); + assert_eq!(stats.idle_unbound, 2); + assert_eq!(stats.checked_out, 0); + assert_eq!(stats.registration_ops, 0); + assert_eq!(stats.epoch, 1); + } + + #[test] + fn fence_at_epoch_max_leaves_checked_out_slots_in_place() { + let mut pool = RegisteredPool::new(1, 4).unwrap(); + let batch = pool.checkout_batch(1).unwrap(); + pool.set_epoch_for_test(u64::MAX); + let err = pool.fence_epoch().unwrap_err(); + assert_eq!( + err, + CacheError::QuotaExceeded { + kind: QuotaKind::Registrations, + } + ); + assert_eq!(pool.stats().epoch, u64::MAX); + assert_eq!(pool.stats().checked_out, 1); + assert_eq!(pool.stats().quarantined, 0); + pool.complete_batch(&batch, SimCompletion::Completed) + .unwrap(); + assert_eq!(pool.stats().idle_registered, 1); + } } diff --git a/crates/transport/src/shard_demo.rs b/crates/transport/src/shard_demo.rs index bcacd40..f7cd203 100644 --- a/crates/transport/src/shard_demo.rs +++ b/crates/transport/src/shard_demo.rs @@ -9,8 +9,11 @@ use core::fmt; use gpucache_core_types::CacheError; -use crate::directory::{Admit, FillRequest, ShardDirectory, ShardParams}; -use crate::iface::{rdma_probe, NixlTransport, TcpTransport}; +use crate::directory::{Admit, FillNotice, FillRequest, ShardDirectory, ShardParams}; +use crate::iface::{ + rdma_probe, submit_preferring_rdma, FallbackReason, HardwarePath, NixlTransport, TcpTransport, + TransferDesc, TransportKind, +}; use crate::registered::{RegisteredPool, SimCompletion}; use crate::stage::stage_through_tcp; @@ -24,9 +27,11 @@ impl fmt::Display for ShardReport { writeln!(f, "flight=JOINED")?; writeln!(f, "credits=SINGLE_CHARGE")?; writeln!(f, "stale_epoch=FENCED")?; + writeln!(f, "fill_notice=FENCED_NOT_A_HIT")?; writeln!(f, "unknown_completion=QUARANTINED")?; writeln!(f, "lease_expiry=HELD")?; writeln!(f, "tcp_path=HOST_COPY")?; + writeln!(f, "rdma_fallback=TCP_CPU_CONTRACT")?; writeln!(f, "dma_posted=NO")?; writeln!(f, "rdma={rdma}", rdma = rdma_probe())?; writeln!(f, "registration=SIMULATED")?; @@ -78,6 +83,11 @@ pub fn run_shard_demo() -> Result { { return Err(CacheError::FencingRejected { reason: "epoch" }); } + if dir.fill_notice(&leader_req.tenant, leader_req.request)? != FillNotice::Fenced + || dir.fill_notice(&joined_req.tenant, joined_req.request)? != FillNotice::Fenced + { + return Err(CacheError::FencingRejected { reason: "epoch" }); + } let mut pool = RegisteredPool::new(1, 16)?; let batch = pool.checkout_batch(1)?; @@ -101,6 +111,17 @@ pub fn run_shard_demo() -> Result { reason: "rdma not linked", }); } + let fallback = submit_preferring_rdma(&mut tcp, &TransferDesc { bytes: b"fallback" })?; + if fallback.attempted != TransportKind::RdmaNixl + || fallback.kind != TransportKind::Tcp + || fallback.fallback != FallbackReason::RdmaNotLinked + || fallback.dma_posted + || fallback.hardware != HardwarePath::CpuContract + { + return Err(CacheError::Storage { + reason: "rdma not linked", + }); + } Ok(ShardReport) } @@ -115,6 +136,8 @@ mod tests { assert!(text.contains("topology=SINGLE_SHARD")); assert!(text.contains("rdma=NOT_RUN_HARDWARE")); assert!(text.contains("registration=SIMULATED")); + assert!(text.contains("fill_notice=FENCED_NOT_A_HIT")); + assert!(text.contains("rdma_fallback=TCP_CPU_CONTRACT")); assert!(text.contains("dma_posted=NO")); assert!(!text.contains("GB/s")); assert!(!text.contains("TTFT")); diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index eda9483..f3b60fa 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -81,11 +81,13 @@ ABSENT -> RESERVED -> WRITING -> COMMITTED -> EVICTING -> ABSENT `KvTransport` is the byte-path trait. `TcpTransport::submit` copies into a bounded host buffer and reports `dma_posted=false`. `NixlTransport::open` fails closed. Feature `rdma-nixl` does not link libnixl. See [adr/0004-nixl-transport.md](adr/0004-nixl-transport.md) and [V2_DATA_PATHS.md](V2_DATA_PATHS.md). -`RegisteredPool` is a fixed set of `Vec` slots. Checkout reuses a binding instead of allocating a new region id. Unknown completion quarantines the slot. Lease expiry does not free it. `fence_epoch` then `reclaim_fenced` does, and so does `attest_drain`. Retired ids are not recycled. +`RegisteredPool` is a fixed set of `Vec` slots. Checkout prefers a slot that is already bound, then an unbound slot. A batch that cannot be fully satisfied does not change the pool. Unknown completion quarantines the slot. Lease expiry does not free it. `fence_epoch` then `reclaim_fenced` does, and so does `attest_drain`. If the epoch cannot advance, checked-out slots stay checked out. Retired ids are not recycled. -`ShardDirectory` is local to one shard. Shard 0 is the single-node case and uses the same methods. A membership change bumps the epoch, drops in-flight fills, and releases credits. Single-flight joins do not charge twice and do not return payload bytes. There is no per-token central directory and no per-block quorum. See [adr/0005-shard-fencing.md](adr/0005-shard-fencing.md). +`ShardDirectory` is local to one shard. Shard 0 is the single-node case and uses the same methods. A membership change bumps the epoch, drops in-flight fills, and releases credits. Single-flight joins do not charge twice and do not return payload bytes. `FillNotice` tells a leader and each joiner whether the fill is pending, finished, failed, or fenced. Finished is not a hit. Fenced is not success. There is no per-token central directory and no per-block quorum. See [adr/0005-shard-fencing.md](adr/0005-shard-fencing.md) and [MOONCAKE.md](MOONCAKE.md). -The cold sink queues a pack and drains it with Put. Refill Heads the manifest first. Over budget returns `OverBudget` and no payload. That outcome is a recompute, not a hit. +The cold sink validates a content-addressed object/manifest pair before admission. Each drain round attempts a bounded snapshot once per job: temporary errors retry at the tail up to three total attempts, while permanent or exhausted errors enter an individually retrievable failed queue. Pending and failed jobs share the job and byte limits. `drain_report` exposes partial progress and the first job error; compatible `drain` returns that error after the round. Ambiguous PUT failures preserve objects that an already visible manifest may reference. The sink does not implement reference-safe orphan reclamation; later explicit cleanup must establish ownership and absence of live manifest references. + +Manifest v3 binds each range to its tenant and the digest of its validated canonical `CacheIdentity`. All public refill APIs require the full expected identity from an upstream authenticated and authorized request. Wrong tenants are rejected before payload download; other identity mismatches and legacy v1/v2 manifests are `Missing`, with no payload download. Over-budget requests and corrupt bytes cannot become hits. Old entries require recompute and repack; the cold crate does not authenticate callers or prove a producer's model computation. See [MOONCAKE.md](MOONCAKE.md) for the API migration and trust boundary. ## V3 offload and hardening diff --git a/docs/BENCHMARKS.md b/docs/BENCHMARKS.md index 154add9..8cd129c 100644 --- a/docs/BENCHMARKS.md +++ b/docs/BENCHMARKS.md @@ -24,6 +24,8 @@ Paired baselines use the same prefix-cache idea on both sides when a GPU run exi There is no `FAIL_PERFORMANCE` figure because there is no paired measurement to fail. Do not add a benchmark that prints a success banner without raw samples, a manifest, and the hardware it ran on. The V2 CPU tests do not fill this gap. +Mooncake's published transfer and serving figures are not rows in this table. `paired-baseline` prints `mooncake_transfer_engine=NOT_LINKED` and `mooncake_published_numbers=NOT_REPRODUCED`. Those lines are labels. They are not a reproduction and they contain no sample. See [MOONCAKE.md](MOONCAKE.md). + ## V3 paired gate The harness is `cargo run -p gpucache-bench --bin v3-paired`. @@ -38,6 +40,7 @@ The comparison it names is **NIC RDMA baseline vs DPU storage queue**, under the | Host core-seconds, DPU CPU, DPU DRAM, GPU interference | `NOT_RUN_HARDWARE` | | Wire bytes, SSD write amplification, wall power | `NOT_RUN_HARDWARE` | | TCO | `TCO_UNKNOWN` | +| Mooncake RDMA | `NOT_A_PERFORMANCE_RESULT`. Not linked, and not a stand-in for the NIC RDMA baseline | | Go gate | `NOT_EVALUATED` | | PB-scale | `NOT_CLAIMED` | | Production ready | `NOT_CLAIMED` | diff --git a/docs/MOONCAKE.md b/docs/MOONCAKE.md new file mode 100644 index 0000000..e5d0e6a --- /dev/null +++ b/docs/MOONCAKE.md @@ -0,0 +1,54 @@ +# Mooncake reverse notes + +This note compares public Mooncake behavior with GPUCache and records what this tree changed. It is not a port of Mooncake, and it is not a performance report. Sources used on 2026-10-16: + +- Mooncake Transfer Engine design: (Segment, BatchTransfer, topology-aware path selection, endpoint pooling, TCP when `--protocol=tcp`). +- FAST 2025 paper, arXiv:2407.00079, "Mooncake: A KVCache-centric Disaggregated Architecture for LLM Serving" (paged KV blocks, prefix hashes, LRU, hot-key replicas, Transfer Engine, `getTransferStatus`). + +No number from those sources is reproduced here. GPUCache did not run Mooncake, Soft-RoCE, RDMA, or a GPU for this change. The decision record is [adr/0009-mooncake-lessons.md](adr/0009-mooncake-lessons.md). + +## What Mooncake does well + +Mooncake separates KV reuse from the compute engine. Prefill and decode can sit on different machines. KV blocks live in a pool outside GPU HBM, and a miss still falls back to compute. + +The Transfer Engine treats registration as a long-lived property of a buffer, not as something to repeat on every slice. A batch names many non-contiguous ranges. Completion is a status the caller polls, so a failure is visible instead of looking like a successful transfer. When a NIC or a queue is unhealthy, the engine tries another path, and TCP is an explicit protocol rather than a silent relabel of RDMA. Endpoints that have been evicted stay alive until their outstanding work drains. + +The store key mixes a block hash with its prefix, so layout-sensitive bytes are not treated as interchangeable. Eviction skips a block that a request is still using. + +## Where that diverges from GPUCache + +| Mooncake | GPUCache constraint | +| --- | --- | +| Extra copies of hot blocks to balance NIC load | Hot-tier replicas and parity are rejected. A lost block is a clean miss, then recompute. Cold durability is RustFS. | +| Central segment metadata (etcd, Redis, or HTTP) | The directory is shard-local. There is no per-token index and no quorum on the miss path. | +| In-tree C++ Transfer Engine, with a bindgen Rust wrapper | `unsafe` is forbidden. NIXL is the named RDMA slot and is not linked. Mooncake is not linked either. | +| Transfer slices can use several NICs | A prefix hit is the whole stored block. The service does not slice inside a block. | +| Published transfer and serving measurements | This tree has no GPU and no NIC. Those measurements are not GPUCache results. TCP is not the DPU baseline. | + +## What this tree changed + +These are CPU contracts. They do not post a work request. + +- **Pool.** Checkout uses an already-bound slot before it binds a free one. If the batch cannot be completed, including when region ids are exhausted, no slot changes state. +- **Fallback.** `submit_preferring_rdma` records an RDMA attempt and a TCP result with reason `RdmaNotLinked`. `dma_posted` is false. `hardware` on that path is `CpuContract`. +- **Fencing.** Leaders and joiners can read `Pending`, `LeaderFinished`, `LeaderFailed`, or `Fenced`. The notice has no bytes. `LeaderFinished` still requires a store `Get` or a recompute. A fenced fill is not success. A single fence refuses epoch overflow. `ControlPlane::isolate_domain` also checks the epoch budget for every victim before removing any member; an insufficient budget preserves the entire membership change and its current fills. +- **Cold path.** Publication uses content-addressed object and manifest keys. The sink validates their pairing before admission, bounds retries, and isolates terminal failures without blocking unrelated work. An ambiguous PUT error never triggers deletion of an object that a visible manifest could reference. Manifest version 3 binds each range to the complete expected `CacheIdentity`; every public refill path requires that identity before it can return payload bytes. A slow or down cold tier is still `Missing` and does not touch the store. +- **Bench labels.** `paired-baseline` prints `mooncake_transfer_engine=NOT_LINKED` and `mooncake_published_numbers=NOT_REPRODUCED`. `v3-paired` prints `mooncake_rdma=NOT_A_PERFORMANCE_RESULT`. Neither report contains a digit. + +## Cold failure contract and migration + +`AsyncSink::drain_report` attempts at most `min(max_jobs, queued_at_call_start)` jobs. A job is attempted once per call. Temporary storage, timeout, and quota errors move it to the queue tail until `MAX_WRITE_ATTEMPTS` (3) is reached. Permanent errors and exhausted retries enter the failed queue. `DrainReport` separates attempted, completed, requeued, and failed counts and retains the first job error. The compatible `drain` method processes that same round and then returns the first error, if any; an error does not undo jobs that completed in that round. + +Pending and failed jobs share the configured job and byte limits. `MAX_SINK_JOBS` (4096) is the metadata ceiling. `failed_count` and `take_failed` let an operator retrieve an individual failed job and release its retained credit. A full failed queue can still cause admission backpressure until its jobs are taken. `cancel` explicitly clears both queues and closes the sink. It is not a per-job recovery operation. + +Object keys are `namespace//object`; manifest keys are `namespace//manifest`. Each digest covers its own complete bytes. Retries retain those keys, so an older pack cannot overwrite a newer pack under the same namespace. A failed publication can leave an unreferenced object. `AsyncSink` does not implement reference-safe orphan reclamation. A later explicit cleanup must establish ownership and absence of live manifest references; a timeout, an unsuccessful PUT response, or one HEAD result is not permission to delete an object. + +Manifest v3 stores the tenant and the digest of the validated canonical `CacheIdentity` for each range. `refill_generation` and `refill_matching` now require `expected: &CacheIdentity`; the availability wrappers `refill_degraded` and `refill_for_tenant` require it as well. There is no public tenant-free refill path. Versions 1 and 2 yield `Missing` without downloading payload bytes. Existing cold entries must be recomputed and repacked with v3; merely changing the version field cannot migrate them. + +The expected identity must come from the caller's authenticated, authorized inference context. This crate does not authenticate a tenant, authorize arbitrary object-store keys, or prove that a producer computed the correct model state. Matching identity digests and payload checksums detects mismatched metadata and bytes within that upstream trust boundary. The same tenant name in another trust domain, model revision, adapter, prefix, or layout is not interchangeable. + +The required regression coverage is listed in [V2_TEST_MATRIX.md](V2_TEST_MATRIX.md). A Grok bot or VM operator should use the existing [VM runner](VM_REGRESSION.md#pr-8-cold-and-fencing-regressions), retain its exit code and artifacts, and preserve all `NOT_RUN_HARDWARE` labels. + +## What stays out + +No Mooncake crate, no etcd client, no replica placement, no intra-block slice, and no Soft-RoCE timing. Live RDMA remains `NOT_RUN_HARDWARE`. diff --git a/docs/V2_DATA_PATHS.md b/docs/V2_DATA_PATHS.md index 66639c4..4a8f0fd 100644 --- a/docs/V2_DATA_PATHS.md +++ b/docs/V2_DATA_PATHS.md @@ -10,10 +10,10 @@ These are the copies this tree actually performs, and the copies it does not. No | --- | --- | --- | | Prefill GPU HBM | Not executed | `NOT_RUN_HARDWARE` | | GPU → host | Not linked. No CUDA copy and no GPUDirect | `NOT_RUN_HARDWARE` | -| Simulated register | `RegisteredPool` binds a `Vec` slot. A second checkout of the same slot does not allocate a new region id. Unknown completion quarantines the slot. Lease expiry does not free it | Host allocation. Not `ibv_reg_mr` | +| Simulated register | `RegisteredPool` binds a `Vec` slot and prefers one that is already bound. A batch is all or nothing. Unknown completion quarantines the slot. Lease expiry does not free it | Host allocation. Not `ibv_reg_mr` | | Host staging | `TcpTransport::submit` and `stage_through_tcp` copy the payload into a bounded buffer, then drop it. Over the cap is backpressure (`QuotaExceeded`) and the slots are quarantined | One host copy. `dma_posted=false` | | TCP RPC | Existing `TcpClient` / `serve` path. Request bytes are still GCK1. The read side uses the V1 buffer pool | Host copy into the socket and the read buffer | -| NIC | Kernel TCP. Feature `rdma-nixl` does not open NIXL | No RDMA work request. `NOT_RUN_HARDWARE` | +| NIC | Kernel TCP. `submit_preferring_rdma` records an RDMA attempt that falls back to this TCP copy with reason `RdmaNotLinked`. Feature `rdma-nixl` does not open NIXL | No RDMA work request. `dma_posted=false`. `NOT_RUN_HARDWARE` | | Decode host → GPU | Not executed | `NOT_RUN_HARDWARE` | A future NIXL path, once `libnixl` is linked on a machine that has it, would add registration, a posted transfer, and a completion event between the two host or GPU buffers. That path is not in this binary. See [adr/0004-nixl-transport.md](adr/0004-nixl-transport.md). @@ -23,7 +23,7 @@ A future NIXL path, once `libnixl` is linked on a machine that has it, would add | Hop | What this tree does | Copy or registration | | --- | --- | --- | | Lookup | Metadata only. Hit, whole-block prefix hit, or miss. No payload | No data copy | -| Directory | `ShardDirectory` records one in-flight fill per physical-key checksum. A joiner does not reserve a second credit. The directory does not return bytes | No data copy. No per-token central lookup. No quorum | +| Directory | `ShardDirectory` records one in-flight fill per physical-key checksum. A joiner does not reserve a second credit. `FillNotice` is pending, finished, failed, or fenced, and it does not return bytes. `LeaderFinished` is not a hit | No data copy. No per-token central lookup. No quorum | | Hit `Get` | TCP returns the committed block | Host copies, same as path 1 | | Miss, stale epoch, wrong tenant, checksum failure | No bytes. The engine recomputes | No data copy | | File backend | `FileStore` writes an ordinary file under the directory the operator passed | Host copy plus the kernel page cache. Not an NVMe format | @@ -37,13 +37,21 @@ The hot TCP service does not depend on `gpucache-cold`. RustFS is not on the pre | Hop | What this tree does | Copy or registration | | --- | --- | --- | -| Pack | `pack_blocks` writes one object and one manifest in host memory | Host copy into the pack buffers | -| 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 | +| Pack | `pack_blocks` writes an object and a v3 manifest in host memory. Each range carries its tenant and a digest of its validated canonical `CacheIdentity`. Object and manifest keys contain their respective content digests | Host copy into the pack buffers | +| Async sink | Admission validates the complete pack/key pairing. `drain_report` puts the object, then the manifest; temporary failures retry at the tail up to three attempts, and permanent/exhausted failures enter the failed queue. Pending and failed jobs share the job and byte caps. Ambiguous PUT errors preserve the object. `take_failed` releases an individual failed job; `cancel` clears both queues and closes the sink | Host buffers remain charged until completion, failed-job retrieval, or explicit cancellation. 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 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 | +| Refill | Every public refill API requires the full expected `CacheIdentity`. Head the manifest and enforce the download budget. Versions 1 and 2 return `Missing` without fetching payload bytes. For v3, validate the manifest, tenant and canonical identity digest before requesting a payload range. Download only when manifest plus payload fit the total byte budget, then verify payload length and checksum | 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 | +`drain_report` visits at most the queue length captured at call start, bounded further by `max_jobs`; a requeued task cannot be attempted repeatedly in that call. The report preserves counts and the first job error. The compatible `drain` call returns that first error after the round, even if other jobs completed. Pending and failed jobs together cannot exceed the configured limits or the 4096-job hard ceiling. A full failed queue causes backpressure until the operator retrieves its jobs. + +Retries use the original content-addressed keys. A failed publication may leave an object without a manifest. Retain its keys for explicit later cleanup after ownership and reference checks. The sink implements neither a reference-safe orphan collector nor a crash-persistent orphan inventory, and does not infer safe deletion from a PUT failure or a HEAD result. It cannot provide a cross-object storage transaction. + +Callers migrate to `refill_generation(..., expected)` or `refill_matching(..., expected)` and pass that same identity through `refill_degraded` or `refill_for_tenant`. Old v1/v2 entries require recompute and repack; they are not converted by changing a header. The expected identity comes from the upstream authenticated and authorized request. The cold crate does not establish that identity, authorize arbitrary store keys, or verify the producer's model computation. + +`RefillBudget` accounts for manifest and payload bodies requested through the refill interface. It is not a backend wire-traffic or RSS bound: the current live HTTP probe also fetches an object body during HEAD to derive its checksum. Mock budget tests do not qualify that backend behavior. + GDS and BlueField storage DMA are in development and were not run. ## What a caller must do on failure diff --git a/docs/V2_TEST_MATRIX.md b/docs/V2_TEST_MATRIX.md index e651425..1cce935 100644 --- a/docs/V2_TEST_MATRIX.md +++ b/docs/V2_TEST_MATRIX.md @@ -6,20 +6,44 @@ CPU contract for the first V2 slice. Live RDMA, live RustFS, GPU export, and tok | --- | --- | --- | | Default workspace tests, including V1 A/B/C | PASS | `cargo test --workspace --all-targets` | | TCP transport trait: one host copy, backpressure, `dma_posted=false` | PASS | `iface::tests` | -| `NixlTransport::open` refuses without a device | PASS | `iface::tests::rdma_open_does_not_post`, `tests/v2_control.rs` | +| `NixlTransport::open` refuses without a device. `submit_preferring_rdma` labels the TCP fallback `RdmaNotLinked` and does not post | PASS | `iface::tests`, `tests/v2_control.rs` | | Feature `rdma-nixl` still refuses `open` | PASS | `cargo test -p gpucache-transport --all-targets --features rdma-nixl` | -| Registration reuse, batch backpressure, quarantine, lease hold, fence reclaim | PASS | `registered::tests` | +| Registration reuse, prefer an idle binding, atomic checkout, quarantine, lease hold, fence reclaim, epoch overflow | PASS | `registered::tests` | | Failed TCP stage quarantines the simulated slots | PASS | `stage::tests` | | Single-node directory is one shard; join charges credits once | PASS | `directory::tests` | | Two shard values do not share flights | PASS | `directory::tests::two_shard_values_do_not_share_flights` | -| Stale epoch fences the fill and releases credits | PASS | `directory::tests::stale_epoch_releases_credits_and_hides_the_fill` | +| Stale epoch fences the fill and releases credits. Finish, failure, and fence are distinct notices with no payload. Epoch overflow keeps the fill | PASS | `directory::tests` | | Other tenant cannot join; quota does not drop the leader | PASS | `directory::tests` | | `shard-demo` prints `SHARD_OK` and `rdma=NOT_RUN_HARDWARE` | PASS | `cargo run -p gpucache-transport --bin shard-demo` | | Cold mock Put/Head/RangeGet/Delete, pack, manifest | PASS | `contract_smoke` | -| Async sink drain, cancel writes nothing, queue backpressure | PASS | `sink::tests` | -| Refill over budget does not download the payload (or the manifest, when the manifest itself is over budget) | PASS | `sink::tests` | +| Async sink publication, bounded retry, failed-job retrieval, cancellation and shared backpressure | REQUIRED_CPU | `sink::tests`; checklist below | +| Identity-bound v3 refill, legacy v1/v2 misses, and download budgets | REQUIRED_CPU | Cold refill/pack tests; checklist below | | `cold-demo` prints `COLD_CONTRACT_OK` and `gds=NOT_RUN_HARDWARE` | PASS | `cargo run -p gpucache-cold --bin cold-demo` | | Live RustFS Put/Head/RangeGet/Delete | NOT_RUN_HARDWARE | No `GPUCACHE_RUSTFS_*` endpoint | | Live NIXL / Soft-RoCE / hardware RDMA | NOT_RUN_HARDWARE | `libnixl` is not linked. See [adr/0004-nixl-transport.md](adr/0004-nixl-transport.md) | | GPU KV export, NIXL P/D, TTFT, TPOT, goodput | NOT_RUN_HARDWARE | `cargo run -p gpucache-bench --bin paired-baseline` | | BlueField, DOCA, GDS, SPDK | NOT_RUN_HARDWARE | Not started. Not an erasure-coding milestone | + +## Cold publication and batch-fencing regressions + +This checklist defines required coverage for the updated CPU contract. It is not a retained test report: `REQUIRED_CPU` does not mean that a new checkout has passed. Record the exact revision and command outcomes with the existing [VM runner](VM_REGRESSION.md#pr-8-cold-and-fencing-regressions). The workspace test gate includes these Rust tests; no separate PR-specific runner is needed. + +| Contract | Required outcome | Test area | +| --- | --- | --- | +| Permanent sink error followed by a healthy job | The bad job is isolated, the healthy job publishes, and the failed job can be retrieved independently | `sink::tests` | +| Temporary sink error | Each selected job is attempted once per drain call; unrelated jobs advance; retry attempts never exceed `MAX_WRITE_ATTEMPTS` (3) | `sink::tests` | +| Retry exhaustion | Move the job to the failed queue; later drain calls do not automatically retry it | `sink::tests` | +| Partial progress and reporting | `drain_report` counts attempts, completions, retries and terminal failures; `drain` returns the first error after processing the round without undoing completed jobs | `sink::tests` | +| Shared accounting | Pending and failed jobs jointly obey `max_jobs` and `max_bytes`; failed-job retrieval and cancel release exact credit; metadata cannot exceed `MAX_SINK_JOBS` (4096) | `sink::tests` | +| Ambiguous object or manifest PUT | Simulate commit followed by an error, not only failure before writing; preserve any object a visible manifest could reference; no speculative DELETE | `sink::tests` | +| Retry ordering under one namespace | Distinct packs use distinct content-addressed keys; a delayed older retry cannot overwrite the later pack | Cold pack/sink tests | +| Forged or mismatched queued pack | Reject an inconsistent key, manifest/object pairing, identity digest, range or payload checksum before issuing a store write | Cold pack/sink tests | +| Full expected identity | A wrong tenant, trust domain, model/config/tokenizer revision, adapter, prefix, position/RoPE, dtype/quantization, layout/shard or valid length cannot return payload bytes | `tests/refill_safety.rs::complete_identity_mismatches_are_misses_before_payload_io` | +| Every public refill entry point | `refill_generation`, `refill_matching`, `refill_degraded` and `refill_for_tenant` all require and enforce the expected `CacheIdentity`; no tenant-free helper bypass | `tests/refill_safety.rs::every_public_refill_entry_requires_and_checks_the_expected_identity` | +| Legacy manifests | Valid legacy v1/v2 headers yield `Missing` with zero payload-range reads; old entries require recompute and repack | `tests/refill_safety.rs::legacy_v1_and_v2_fixtures_are_clean_misses_without_payload_reads` | +| Budget and malformed input | Over-budget manifests are not downloaded through refill; disallowed payload ranges are not downloaded; malformed v3 metadata and corrupt payloads fail closed | `tests/refill_safety.rs`, including `malformed_v3_manifest_fails_closed_before_payload_io` and `corrupt_payload_and_content_key_substitution_return_no_loaded_bytes` | +| Slow or unavailable G4 | Return `Missing` without calling the object store | `ops::tests` | +| Domain-isolation epoch shortage | Insufficient epoch budget for the complete victim set leaves members, domains, epoch, credits and leader/joiner notices unchanged | `plane::tests::isolate_domain_epoch_shortage_keeps_the_entire_batch_and_fills_unchanged` | +| Domain-isolation exact budget | Every victim is removed and every affected old fill is fenced; a one-victim overflow also makes no change | `plane::tests` | + +These tests use trusted CPU callers and synthetic bytes. They do not authenticate an external tenant, prove KV model semantics, qualify orphan garbage collection, or exercise real RDMA/DPU drain. Hardware rows above remain `NOT_RUN_HARDWARE`. diff --git a/docs/VM_REGRESSION.md b/docs/VM_REGRESSION.md index 4036f17..b5d5c8e 100644 --- a/docs/VM_REGRESSION.md +++ b/docs/VM_REGRESSION.md @@ -48,6 +48,34 @@ cargo test -p gpucache-transport --all-targets --features rdma-nixl cargo test -p gpucache-dpu --all-targets --features dpu-bf3 ``` +## PR 8 cold and fencing regressions + +A Grok bot or VM operator should run the existing CPU profile against the exact updated commit. Do not use an earlier green PR check as evidence for the updated implementation. From the repository root, choose a new or empty output directory: + +```sh +python3 scripts/ci/vm_regression.py --profile cpu --output artifacts/pr8-cpu-01 +``` + +Use `--offline` only when the locked dependencies are already cached. Keep the exit code and the entire output directory. Do not contact a bot or publish its report as part of this command. The runner's workspace test gate already discovers the new cold and transport tests, so there is no separate PR-specific profile or runner. + +For focused diagnosis before rerunning the complete profile: + +```sh +cargo test --locked -p gpucache-cold --all-targets +cargo test --locked -p gpucache-transport --all-targets +``` + +The acceptance checklist is [V2_TEST_MATRIX.md](V2_TEST_MATRIX.md#cold-publication-and-batch-fencing-regressions). Confirm all of the following in the retained test logs: + +- A permanent or exhausted sink job is isolated and individually retrievable. Temporary retries are bounded, do not prevent unrelated jobs from completing, and do not amplify one task repeatedly within a drain call. +- Pending and failed jobs share the job/byte limits. Retrieving a failed job or cancelling releases the expected credit. Partial success and the first error are reported accurately. +- A PUT that commits before returning an error does not trigger speculative deletion. Content-addressed object/manifest keys keep an old retry from replacing a newer pack. +- Every refill API requires the full expected `CacheIdentity`. Cross-tenant, cross-trust-domain, cross-model, prefix and layout mismatches return no payload. The cold crate still relies on upstream authentication and authorization. +- Legacy manifest versions 1 and 2 yield `Missing` without payload reads; malformed v3 data and checksum failures cannot become hits. Recompute and repack old entries instead of relabeling their version. +- Multi-member domain isolation preflights its complete epoch budget. Insufficient budget leaves membership, credits and active fill notices unchanged. + +Retained cold objects after a failed publication are not evidence that garbage collection completed. Explicit orphan reclamation needs separate ownership/reference checks and is not qualified by this CPU gate. A green result does not add native NIXL, GDS, BlueField, RustFS hardware, or 10,000-GPU evidence. + ## Artifact contract | Path | Contents | Interpretation | diff --git a/docs/adr/0009-mooncake-lessons.md b/docs/adr/0009-mooncake-lessons.md new file mode 100644 index 0000000..c0561e7 --- /dev/null +++ b/docs/adr/0009-mooncake-lessons.md @@ -0,0 +1,44 @@ +# ADR 0009: What to take from Mooncake, and what to leave + +- Status: **Accepted** +- Date: 2026-10-16 +- Updated: 2026-10-17 +- Milestone: V3 + +## Context + +Mooncake (kvcache-ai/Mooncake; the FAST 2025 paper "Mooncake: A KVCache-centric Disaggregated Architecture for LLM Serving", arXiv:2407.00079) is a public KV-cache transfer stack used beside LLM serving. Its Transfer Engine registers DRAM or VRAM buffers, opens a remote segment, submits a batch, and polls `getTransferStatus`. RDMA paths prefer a NIC that is local to the buffer's NUMA or GPU, and they can fall back to another NIC or to TCP. Failed endpoints stay unreclaimed until outstanding slices drain. The store keeps paged KV blocks keyed by a hash of the block and its prefix, evicts with LRU, and places extra copies of hot blocks to spread load. + +GPUCache is a different product shape. The hot tier is recomputable G3.5 state. RustFS is the cold tier. [0002-no-erasure-coding.md](0002-no-erasure-coding.md) forbids hot-tier replicas and parity. [0004-nixl-transport.md](0004-nixl-transport.md) already refused to link Mooncake: the in-tree Rust binding uses bindgen and `unsafe`, and this workspace forbids both. [0005-shard-fencing.md](0005-shard-fencing.md) keeps the directory shard-local. [0003-partial-prefix.md](0003-partial-prefix.md) refuses to slice a stored block. + +This environment has no GPU, no NIC, and no `libnixl`. A Mooncake throughput figure from the paper or from Transfer Engine's own bench is not a GPUCache result. + +## Decision + +Adopt four CPU-side contracts that match Mooncake's failure and registration behavior, without linking Mooncake and without claiming a device path: + +1. **Registration reuse.** `RegisteredPool::checkout_batch` prefers a slot that already has a binding. A batch that cannot be fully satisfied, including region-id exhaustion, leaves the pool unchanged. +2. **Explicit transport fallback.** `submit_preferring_rdma` tries the NIXL slot, which still refuses to open. The receipt then says the attempt was RDMA, the path that ran was TCP, and the reason is `RdmaNotLinked`. `dma_posted` stays false. TCP is not relabeled as RDMA and is not a DPU baseline. +3. **Fill notices.** A leader finish, a leader failure, and a membership fence are different notices. None of them carries payload bytes. `LeaderFinished` is not a hit. A fence is not success. The notice table is bounded; a dropped notice is `UnknownGeneration`, which means recompute. An epoch that cannot advance does not drop in-flight fills and does not publish the membership change. Domain isolation preflights the full victim count, so insufficient epoch budget cannot partially remove a domain. +4. **Bounded cold retry and identity-bound refill.** A drain round attempts each selected job once. Temporary failures are retried at the tail for at most three attempts; permanent failures and exhausted retries enter an individually retrievable failed queue. Both queues share byte and metadata limits. Content-addressed pack keys prevent an older retry from overwriting a later publication. Ambiguous PUT failures preserve the object instead of deleting bytes that a visible manifest may reference. Manifest v3 binds each range to the full expected `CacheIdentity`, which every public refill API requires before downloading its payload. + +Do not adopt: + +- Hot-block replicas or `change_replica`. An extra copy would be a latency hint, and it is still not in this tree. It is not a durability promise. +- A central metadata service on the miss path. No etcd, Redis, or HTTP segment directory was added. +- Linking `mooncake-transfer-engine`, NIXL, DOCA, or a NIC. +- Slicing one KV block across transfers. A prefix hit remains a whole stored block. +- Any Mooncake, Soft-RoCE, NIC, or DPU timing. The bench reports name those rows `NOT_LINKED`, `NOT_REPRODUCED`, or `NOT_RUN_HARDWARE`. + +## Consequences + +- `cargo test --workspace --all-targets` still needs no NIC and no Mooncake library. +- Feature `rdma-nixl` still does not link libnixl. `NixlTransport::open` still returns `rdma not linked`. +- Callers that only watch a fill disappear must read `FillNotice`. Disappearance after `Fenced` or `LeaderFailed` is a miss. +- Manifest version is 3 because each range records its tenant and a digest of the validated canonical `CacheIdentity`. Versions 1 and 2 are `Missing`, with no payload download; recompute and repack instead of relabeling an old manifest. +- `refill_generation` and `refill_matching` require `expected: &CacheIdentity`; `refill_degraded` and `refill_for_tenant` forward that same required identity. Migrate every caller, including administrative helpers. No tenant-free public shortcut remains. +- Identity comparison assumes an authenticated and authorized upstream caller and trusted producer. This crate adds neither an authenticator nor proof that KV bytes equal model execution. +- `drain_report` exposes attempted/completed/requeued/failed counts and the first job error. The original `drain` signature remains, but a returned error can follow successful writes in the same round. Cancelled sinks reject drain at the outer `Result` level. +- Failed jobs retain admission credit until `take_failed` transfers them out; operators must handle that queue. `cancel` discards all pending and failed jobs and permanently closes the sink. +- Failed publications may leave orphaned objects. Explicit later reclamation must establish ownership and absence of live references; the sink does not delete them speculatively after an ambiguous PUT error. +- See [../MOONCAKE.md](../MOONCAKE.md) for the comparison this decision is based on.