diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 74ab64a..b960acf 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -20,12 +20,20 @@ jobs: run: cargo clippy --workspace --all-targets -- -D warnings - name: Test run: cargo test --workspace --all-targets + - name: Fallback demo + run: cargo run -p gpucache-transport --bin fault-demo --quiet + - name: Paired baseline stub + run: cargo run -p gpucache-bench --bin paired-baseline --quiet - name: Reject accelerator crates in V1 run: | if grep -RInE 'cuda|doca|nixl|rdma|spdk' crates/*/Cargo.toml; then echo "V1 crates must not depend on GPU or DPU stacks" exit 1 fi + if grep -RInE 'reed_solomon|erasure_coding|parity_shard' crates --include '*.rs'; then + echo "V1 must not add erasure coding" + exit 1 + fi - name: Keep the cold tier off the hot path run: | if grep -n 'gpucache-cold' crates/storage/Cargo.toml crates/transport/Cargo.toml crates/protocol/Cargo.toml crates/core-types/Cargo.toml; then @@ -41,7 +49,9 @@ jobs: - uses: actions/checkout@v4 - uses: dtolnay/rust-toolchain@stable - name: Test - run: cargo test --workspace --all-targets + run: | + echo "aarch64 cargo test is a CPU build. It is not a BlueField or DOCA run." + cargo test --workspace --all-targets test-aarch64-qemu: name: test (aarch64 qemu-user) diff --git a/AGENTS.md b/AGENTS.md index a36a7fd..61d1b16 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -4,7 +4,7 @@ GPUCache is the G3.5 shared KV tier. RustFS is the colder G4 backend. Read [READ ## Phase -The active phase is **V1 milestone B**. Do not implement V1 C, V2 RDMA, or V3 DPU work in a V1 B change. Do not publish crates, cut releases, deploy, or edit a website unless asked. +The active phase is **V1 milestone C**. Do not implement V2 RDMA or V3 DPU work in a V1 C change. Do not publish crates, cut releases, deploy, or edit a website unless asked. ## Honesty rules @@ -23,6 +23,7 @@ The active phase is **V1 milestone B**. Do not implement V1 C, V2 RDMA, or V3 DP | `gpucache-storage` | the two crates above | | `gpucache-transport` | the three crates above. TCP only | | `gpucache-cold` | `gpucache-core-types` and `sha2`. Not a dependency of storage or transport | +| `gpucache-bench` | nothing. Prints `NOT_RUN_HARDWARE` baseline rows and must not invent timings | No crate may add CUDA, DOCA, NIXL, RDMA, SPDK, erasure coding, or replica placement. `unsafe` is forbidden. Registered buffers whose completion is unknown stay quarantined until drain or a newer epoch; do not free them only because a lease timer fired. The cold crate stays off the hot path. diff --git a/COMPATIBILITY.md b/COMPATIBILITY.md index 81a46d8..b56f20a 100644 --- a/COMPATIBILITY.md +++ b/COMPATIBILITY.md @@ -1,12 +1,12 @@ # Compatibility manifest -Recorded for V1 milestones A and B. This file is not a performance acceptance report and not a hardware inventory of a BlueField node. +Recorded for V1 milestones A, B, and C. This file is not a performance acceptance report and not a hardware inventory of a BlueField node. | Component | Value | Result | | --- | --- | --- | | OS | Ubuntu 24.04.5 LTS | Recorded on the V1 A build host | | Kernel | Linux 6.12.94+ | Recorded on the V1 A build host | -| Architecture exercised locally | x86_64 native, and aarch64 via qemu-user (`qemu-aarch64-static`) | `PASS` for `cargo test --workspace --all-targets` on both. This is not a BlueField run. | +| Architecture exercised locally for V1 C | x86_64 native | `PASS` for `cargo test --workspace --all-targets`, `fault-demo`, and `paired-baseline`. aarch64 was not executed on this host (`NOT_RUN` here). CI still runs `test-aarch64` and `test-aarch64-qemu`. Neither is a BlueField run. | | Rust | stable 1.99.0 (b940084d7 2026-09-28); MSRV declared as 1.85 | `PASS` | | Python | 3.12.3 present on the build host | Not used by V1 A | | CUDA | Not installed | `NOT_RUN_HARDWARE` | @@ -17,7 +17,7 @@ Recorded for V1 milestones A and B. This file is not a performance acceptance re | DOCA / BlueField BFB / firmware | Not present | `NOT_RUN_HARDWARE` | | RustFS server / client | No endpoint authorized in this environment. Contract mock covers Put, Head, RangeGet, Delete, pack, and manifest | Live probe `NOT_RUN_HARDWARE`. Mock `PASS` | | TCP / file store | Loopback TCP and a temp directory on the build host | `PASS` for `cargo test --workspace --all-targets` on x86_64. Not an NVMe format and not a DPU run | -| Direct Rust deps | `sha2` 0.10.8; dev `proptest` 1.4.0 | Apache-2.0 OR MIT; see `Cargo.lock`. No new direct crate in V1 B | +| Direct Rust deps | `sha2` 0.10.8; dev `proptest` 1.4.0 | Apache-2.0 OR MIT; see `Cargo.lock`. V1 C adds no direct crate. `gpucache-bench` has an empty dependency list | | License of this repo | Apache-2.0 | See `LICENSE` | CI (see `.github/workflows/ci.yml`) builds and runs the same tests on x86_64 and on aarch64 (GitHub ARM runner, plus qemu-user cross execution). An aarch64 result is a CPU result. It does not certify BlueField-3, DOCA, or NVMe-on-DPU. diff --git a/Cargo.lock b/Cargo.lock index 354fec9..e6261d6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -127,6 +127,10 @@ dependencies = [ "r-efi", ] +[[package]] +name = "gpucache-bench" +version = "0.1.0" + [[package]] name = "gpucache-cold" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index dbaa504..d132f7d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -6,6 +6,7 @@ members = [ "crates/storage", "crates/cold", "crates/transport", + "crates/bench", ] [workspace.package] diff --git a/README.md b/README.md index 94d8884..d85c42f 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,7 @@ GPUCache does not do erasure coding and does not keep durability replicas. A los ## Status -**V1 milestones A and B are implemented. DPU acceleration is not.** +**V1 milestones A, B, and C are implemented. DPU acceleration is not.** V1 runs the cache contract on CPU. The long-term deployment target is a storage-side BlueField-3, but this repository does not link DOCA, CUDA, RDMA, or SPDK, and it does not claim that any of those paths exist. An aarch64 build is a CPU build. It is not a BlueField functional test. @@ -18,7 +18,7 @@ V1 runs the cache contract on CPU. The long-term deployment target is a storage- | V1 A: real vLLM KV block export/import | `NOT_RUN_HARDWARE` (no GPU in this environment) | | V1 B: TCP service, file backend, LRU admission, tenant quotas, two-client reuse | Implemented in this tree | | V1 B: RustFS live Put/Head/RangeGet/Delete | `NOT_RUN_HARDWARE` (no authorized endpoint). Contract mock passes | -| V1 C: fuzz, fault injection, engine output comparison, paired baselines | Not started | +| V1 C: fuzz, fault injection, synthetic token checks, paired-baseline stubs | Implemented in this tree. Real engine output is `NOT_RUN_HARDWARE` | | V2: NIC RDMA (NIXL or Mooncake) plus a real RustFS cold path | Not started; blocked on V1 | | V3: BlueField-3 data-path offload and production hardening | Not started; not purchased and not measured. Not an erasure-coding milestone | | 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 | @@ -63,7 +63,7 @@ From a clean checkout, with Rust stable (MSRV 1.85): cargo test --workspace --all-targets ``` -That command is the CPU contract repro for V1 A and V1 B. It includes unit tests, property tests, timeout and cancel tests, a single-process import of a synthetic block, a cross-process import through `kv-block-xfer`, file-backend quota and restart tests, and the TCP reuse tests. The bytes are synthetic. They are not a vLLM KV tensor. +That command is the CPU contract repro for V1 A, V1 B, and V1 C. It includes unit tests, property tests, protocol fuzz, timeout and cancel tests, a single-process import of a synthetic block, a cross-process import through `kv-block-xfer`, file-backend quota and restart tests, the TCP reuse tests, and the V1 C fault matrix. The bytes are synthetic. They are not a vLLM KV tensor. Two localhost clients sharing one file-backed service: @@ -73,6 +73,20 @@ cargo run -p gpucache-transport --bin share-demo A successful run prints `status=REUSE_OK` and `partial_prefix=WHOLE_BLOCK_PREFIX`. It does not print a sliced-block hit. +Hit, miss, and recompute fallback from a clean temp directory: + +```bash +cargo run -p gpucache-transport --bin fault-demo +``` + +Success is `status=FALLBACK_OK` plus `criteria=EXACT_TOKEN_IDS` and `tolerance=NONE`. The process prints that banner only after the checks pass. A wrong tenant or wrong model is not a hit. This is not a vLLM run. + +Paired GPU baselines stay `NOT_RUN_HARDWARE` and print no timings: + +```bash +cargo run -p gpucache-bench --bin paired-baseline +``` + `kv-block-xfer import ` reads one export frame on stdin and imports it into a fresh `MemoryStore`. It does not talk to an engine. ## Layout @@ -81,13 +95,14 @@ A successful run prints `status=REUSE_OK` and `partial_prefix=WHOLE_BLOCK_PREFIX crates/core-types identity, keys, lifecycle, errors crates/protocol batch messages, replies, version negotiation crates/storage MemoryStore, FileStore, kv-block-xfer -crates/transport TCP server, client, share-demo +crates/transport TCP server, client, share-demo, fault-demo crates/cold RustFS contract, pack/manifest, live probe +crates/bench paired-baseline stub (no timings) 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/RISKS.md](docs/RISKS.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/V1C_TEST_MATRIX.md](docs/V1C_TEST_MATRIX.md). ## License diff --git a/THIRD_PARTY_NOTICES.md b/THIRD_PARTY_NOTICES.md index ff9aec8..052e06d 100644 --- a/THIRD_PARTY_NOTICES.md +++ b/THIRD_PARTY_NOTICES.md @@ -1,6 +1,6 @@ # Third-party notices -GPUCache V1 A and V1 B are Apache-2.0. See [LICENSE](LICENSE). +GPUCache V1 A, V1 B, and V1 C are Apache-2.0. See [LICENSE](LICENSE). V1 C adds no direct dependency. Direct dependencies: diff --git a/crates/bench/Cargo.toml b/crates/bench/Cargo.toml new file mode 100644 index 0000000..02fa5e8 --- /dev/null +++ b/crates/bench/Cargo.toml @@ -0,0 +1,16 @@ +[package] +name = "gpucache-bench" +version.workspace = true +edition.workspace = true +license.workspace = true +rust-version.workspace = true +authors.workspace = true +description = "Paired-baseline harness stubs. GPU rows stay NOT_RUN_HARDWARE. No accelerator dependencies." +publish = false + +[[bin]] +name = "paired-baseline" +path = "src/bin/paired_baseline.rs" + +[lints] +workspace = true diff --git a/crates/bench/src/bin/paired_baseline.rs b/crates/bench/src/bin/paired_baseline.rs new file mode 100644 index 0000000..30bf341 --- /dev/null +++ b/crates/bench/src/bin/paired_baseline.rs @@ -0,0 +1,14 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 +//! Print the paired-baseline matrix. Every GPU row is NOT_RUN_HARDWARE. +//! +//! This process does not measure TTFT, TPOT, or token goodput. + +use std::process::ExitCode; + +use gpucache_bench::paired_baseline_report; + +fn main() -> ExitCode { + print!("{}", paired_baseline_report()); + ExitCode::SUCCESS +} diff --git a/crates/bench/src/lib.rs b/crates/bench/src/lib.rs new file mode 100644 index 0000000..4c500a0 --- /dev/null +++ b/crates/bench/src/lib.rs @@ -0,0 +1,60 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 +//! Paired baseline names for the V1 exit criteria. +//! +//! This crate does not run a GPU, an engine, or a timer. It prints +//! `NOT_RUN_HARDWARE` for every row that needs token goodput. Filling in a +//! number here would be a fabricated result. + +/// Text report. It contains no digits, so it cannot smuggle a latency or a token rate. +pub fn paired_baseline_report() -> String { + let lines = [ + "status=NOT_RUN_HARDWARE", + "engine_native_prefix_cache=NOT_RUN_HARDWARE", + "local_offload=NOT_RUN_HARDWARE", + "gpucache_cpu_tcp_goodput=NOT_RUN_HARDWARE", + "scenario_cold=NOT_RUN_HARDWARE", + "scenario_warm=NOT_RUN_HARDWARE", + "scenario_no_reuse=NOT_RUN_HARDWARE", + "scenario_multi_turn=NOT_RUN_HARDWARE", + "scenario_hbm_within=NOT_RUN_HARDWARE", + "scenario_hbm_cliff=NOT_RUN_HARDWARE", + "write_cost=NOT_RUN_HARDWARE", + "samples=NONE", + "numeric_tolerance=NOT_APPLICABLE", + "bypass_when_no_reuse=MISS_RECOMPUTE", + ]; + let mut out = String::new(); + for line in lines { + out.push_str(line); + out.push('\n'); + } + out +} + +#[cfg(test)] +mod tests { + use super::paired_baseline_report; + + #[test] + fn report_has_no_invented_numbers() { + let report = paired_baseline_report(); + assert!(report.chars().all(|c| !c.is_ascii_digit())); + for key in [ + "engine_native_prefix_cache=NOT_RUN_HARDWARE", + "local_offload=NOT_RUN_HARDWARE", + "gpucache_cpu_tcp_goodput=NOT_RUN_HARDWARE", + "scenario_cold=NOT_RUN_HARDWARE", + "scenario_warm=NOT_RUN_HARDWARE", + "scenario_no_reuse=NOT_RUN_HARDWARE", + "scenario_multi_turn=NOT_RUN_HARDWARE", + "scenario_hbm_within=NOT_RUN_HARDWARE", + "scenario_hbm_cliff=NOT_RUN_HARDWARE", + "samples=NONE", + ] { + assert!(report.contains(key), "missing {key}"); + } + assert!(!report.contains("tokens/s")); + assert!(!report.contains("PASS")); + } +} diff --git a/crates/protocol/src/fuzz.rs b/crates/protocol/src/fuzz.rs new file mode 100644 index 0000000..a36259e --- /dev/null +++ b/crates/protocol/src/fuzz.rs @@ -0,0 +1,110 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 +//! Protocol decoder fuzz. Random, truncated, duplicated, and reordered frames +//! must not panic and must not decode as a different successful request. + +use gpucache_core_types::{ + content_checksum, synthetic_identity, CacheError, RequestId, HARD_BYTES_CAP, +}; +use proptest::prelude::*; + +use crate::{ + decode_export, decode_hello, decode_hello_ack, decode_reply, decode_request, encode_export, + encode_hello, encode_request, AuthContext, CommittedExport, ItemHeader, LookupItem, + RequestBody, WIRE_VERSION, +}; + +fn sample_request() -> (AuthContext, RequestBody) { + let identity = synthetic_identity(1, 1); + let auth = AuthContext::new( + identity.tenant.clone(), + identity.trust_domain.clone(), + [9u8; 32], + ) + .unwrap(); + let body = RequestBody::Lookup(vec![LookupItem { + header: ItemHeader { + request_id: RequestId::from_counter(1).unwrap(), + epoch: 1, + deadline_ms: 50, + }, + identity, + }]); + (auth, body) +} + +fn sample_export() -> Vec { + let payload = b"synthetic-kv".to_vec(); + encode_export(&CommittedExport { + identity: synthetic_identity(1, 1), + source_generation: 1, + checksum: content_checksum(&payload), + payload, + }) + .unwrap() +} + +#[test] +fn truncated_duplicate_and_reordered_frames_fail_closed() { + let (auth, body) = sample_request(); + let frame = encode_request(&auth, &body).unwrap(); + let (decoded_auth, decoded_body) = decode_request(&frame).unwrap(); + assert_eq!(decoded_auth, auth); + assert_eq!(decoded_body, body); + for len in 0..frame.len() { + assert!( + decode_request(&frame[..len]).is_err(), + "short frame of {len} bytes must not decode" + ); + } + let mut duplicated = frame.clone(); + duplicated.extend_from_slice(&frame); + assert!(matches!( + decode_request(&duplicated), + Err(CacheError::Malformed { + reason: "trailing bytes" + }) + )); + let mut reordered = frame.clone(); + let mid = reordered.len() / 2; + reordered.rotate_left(mid); + assert!(decode_request(&reordered).is_err()); + let export = sample_export(); + let mut corrupt = export.clone(); + let last = corrupt.len() - 1; + corrupt[last] ^= 0x01; + assert!(matches!( + decode_export(&corrupt), + Err(CacheError::ChecksumMismatch) + )); + assert!(decode_export(&export[..export.len() / 2]).is_err()); + let mut huge = Vec::from(&b"GCK1"[..]); + huge.extend_from_slice(&WIRE_VERSION.to_le_bytes()); + huge.extend_from_slice(&1u16.to_le_bytes()); + huge.extend_from_slice(&(HARD_BYTES_CAP as u32).saturating_add(1).to_le_bytes()); + assert!(decode_request(&huge).is_err()); + assert!(decode_reply(&frame).is_err()); + assert!(decode_hello(&frame).is_err()); + assert!(decode_hello_ack(&frame).is_err()); + let hello = encode_hello(&[WIRE_VERSION]).unwrap(); + assert_eq!(decode_hello(&hello).unwrap(), vec![WIRE_VERSION]); +} + +proptest! { + #![proptest_config(ProptestConfig::with_cases(48))] + #[test] + fn random_frames_do_not_panic_and_roundtrip_when_accepted( + bytes in prop::collection::vec(any::(), 0..384) + ) { + if let Ok((auth, body)) = decode_request(&bytes) { + let again = encode_request(&auth, &body).expect("accepted request encodes"); + let (auth2, body2) = decode_request(&again).expect("re-encoded request decodes"); + assert_eq!(auth, auth2); + assert_eq!(body, body2); + } + let _ = decode_reply(&bytes); + let _ = decode_export(&bytes); + let _ = decode_hello(&bytes); + let _ = decode_hello_ack(&bytes); + } +} diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index 2f42635..5df515e 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -13,6 +13,9 @@ mod codec; mod negotiate; mod reply; +#[cfg(test)] +mod fuzz; + pub use api::{ identity_for_prefix_hit, BatchReply, CommitItem, CommitOutcome, GetHit, GetItem, InvalidateItem, InvalidateOutcome, ItemHeader, ItemResult, LookupItem, LookupOutcome, diff --git a/crates/storage/src/file.rs b/crates/storage/src/file.rs index bb6def1..7a9792c 100644 --- a/crates/storage/src/file.rs +++ b/crates/storage/src/file.rs @@ -29,17 +29,32 @@ const BLOCK_VERSION: u16 = 1; const LRU_MAGIC: &[u8; 4] = b"GLRU"; const LRU_VERSION: u16 = 1; +/// Injected failure for the next durable block write. Production callers leave +/// this unset. It exists so tests can force a short write or a full volume +/// without filling the host disk. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum PersistFault { + DiskFull, + ShortWrite, +} + /// Directory-backed store used by the V1 B TCP service. /// /// Admission evicts the least-recently-used unpinned `COMMITTED` block when a /// tenant or global retained-byte quota would otherwise fail. Pinned, /// quarantined, and in-flight blocks are not evicted. `MemoryStore` itself /// still refuses admission without evicting; only this backend reclaims. +/// +/// A failed durable write is not a hit. The in-memory commit is abandoned and +/// the request id is forgotten, so a replay cannot report success for bytes +/// that never reached the directory. pub struct FileStore { root: PathBuf, inner: MemoryStore, lru: VecDeque, skipped_corrupt: u32, + fault: Option, + fault_remaining: u32, } impl FileStore { @@ -68,6 +83,8 @@ impl FileStore { inner, lru: order, skipped_corrupt, + fault: None, + fault_remaining: 0, }; store.write_manifest()?; store.write_lru()?; @@ -94,6 +111,17 @@ impl FileStore { self.skipped_corrupt } + /// Fail the next `times` committed-block writes. `times == 0` clears the hook. + pub fn inject_persist_fault(&mut self, fault: PersistFault, times: u32) { + if times == 0 { + self.fault = None; + self.fault_remaining = 0; + return; + } + self.fault = Some(fault); + self.fault_remaining = times; + } + pub fn committed_path(&self, generation: u64) -> PathBuf { block_path(&self.root, generation) } @@ -158,7 +186,17 @@ impl FileStore { self.touch(outcome.generation); Ok(outcome) } - Err(err) => Err(err), + Err(err) => { + let _ = self.inner.abandon_durable_commit( + outcome.generation, + request_id, + now_ms, + ); + self.delete_block(outcome.generation); + self.lru + .retain(|generation_in_lru| *generation_in_lru != outcome.generation); + Err(err) + } }, Err(err) => Err(err), }; @@ -292,7 +330,16 @@ impl FileStore { let block = self.inner.export_committed(generation)?; let bytes = encode_block(&block)?; let path = block_path(&self.root, generation); - write_atomic(&path, &bytes) + write_atomic(&path, &bytes, self.take_fault()) + } + + fn take_fault(&mut self) -> Option { + let fault = self.fault?; + self.fault_remaining = self.fault_remaining.saturating_sub(1); + if self.fault_remaining == 0 { + self.fault = None; + } + Some(fault) } fn touch(&mut self, generation: u64) { @@ -313,7 +360,7 @@ impl FileStore { w.raw(MANIFEST_MAGIC); w.u16(MANIFEST_VERSION); w.u64(self.inner.epoch()); - write_atomic(&self.root.join("MANIFEST"), &w.finish()) + write_atomic(&self.root.join("MANIFEST"), &w.finish(), None) } fn write_lru(&self) -> Result<(), CacheError> { @@ -325,7 +372,7 @@ impl FileStore { for generation in &self.lru { w.u64(*generation); } - write_atomic(&self.root.join("LRU"), &w.finish()) + write_atomic(&self.root.join("LRU"), &w.finish(), None) } } @@ -591,17 +638,32 @@ fn read_capped(path: &Path, max: usize) -> Result, CacheError> { Ok(buf) } -fn write_atomic(path: &Path, bytes: &[u8]) -> Result<(), CacheError> { +fn classify_store_io(err: &std::io::Error) -> CacheError { + if err.kind() == std::io::ErrorKind::StorageFull || err.raw_os_error() == Some(28) { + CacheError::Storage { + reason: "disk full", + } + } else { + CacheError::Storage { reason: "store io" } + } +} + +fn write_atomic(path: &Path, bytes: &[u8], fault: Option) -> Result<(), CacheError> { + if matches!(fault, Some(PersistFault::DiskFull)) { + return Err(CacheError::Storage { + reason: "disk full", + }); + } let tmp = path.with_extension("tmp"); - { - let mut file = - File::create(&tmp).map_err(|_| CacheError::Storage { reason: "store io" })?; - file.write_all(bytes) - .map_err(|_| CacheError::Storage { reason: "store io" })?; - file.sync_all() - .map_err(|_| CacheError::Storage { reason: "store io" })?; + let result = write_temp(&tmp, bytes, fault); + if result.is_err() { + let _ = fs::remove_file(&tmp); + return result; + } + if let Err(err) = fs::rename(&tmp, path) { + let _ = fs::remove_file(&tmp); + return Err(classify_store_io(&err)); } - fs::rename(&tmp, path).map_err(|_| CacheError::Storage { reason: "store io" })?; if let Some(parent) = path.parent() { if let Ok(dir) = File::open(parent) { let _ = dir.sync_all(); @@ -609,3 +671,50 @@ fn write_atomic(path: &Path, bytes: &[u8]) -> Result<(), CacheError> { } Ok(()) } + +fn write_temp(tmp: &Path, bytes: &[u8], fault: Option) -> Result<(), CacheError> { + let mut file = File::create(tmp).map_err(|err| classify_store_io(&err))?; + let slice = if matches!(fault, Some(PersistFault::ShortWrite)) { + &bytes[..bytes.len() / 2] + } else { + bytes + }; + file.write_all(slice) + .map_err(|err| classify_store_io(&err))?; + file.sync_all().map_err(|err| classify_store_io(&err))?; + if matches!(fault, Some(PersistFault::ShortWrite)) { + return Err(CacheError::Storage { + reason: "short write", + }); + } + Ok(()) +} + +#[cfg(test)] +mod io_class_tests { + use super::classify_store_io; + use gpucache_core_types::CacheError; + + #[test] + fn storage_full_and_enospc_map_to_disk_full() { + let full = std::io::Error::new(std::io::ErrorKind::StorageFull, "full"); + assert_eq!( + classify_store_io(&full), + CacheError::Storage { + reason: "disk full", + } + ); + let enospc = std::io::Error::from_raw_os_error(28); + assert_eq!( + classify_store_io(&enospc), + CacheError::Storage { + reason: "disk full", + } + ); + let other = std::io::Error::new(std::io::ErrorKind::PermissionDenied, "denied"); + assert_eq!( + classify_store_io(&other), + CacheError::Storage { reason: "store io" } + ); + } +} diff --git a/crates/storage/src/lib.rs b/crates/storage/src/lib.rs index fb459d8..30faba6 100644 --- a/crates/storage/src/lib.rs +++ b/crates/storage/src/lib.rs @@ -10,10 +10,12 @@ mod file; mod memory; +mod synthetic; mod transfer; -pub use file::FileStore; +pub use file::{FileStore, PersistFault}; pub use memory::{MemoryStore, ReclaimRequest, StoreStats}; +pub use synthetic::{decode_synthetic_kv, deterministic_recompute, encode_synthetic_kv}; pub use transfer::{import_committed_block, ImportReport}; #[cfg(test)] diff --git a/crates/storage/src/memory.rs b/crates/storage/src/memory.rs index 945c138..5782b32 100644 --- a/crates/storage/src/memory.rs +++ b/crates/storage/src/memory.rs @@ -508,6 +508,24 @@ impl MemoryStore { Ok(()) } + /// Drop a committed block after the durable write failed. + /// + /// The client was already told the commit failed. A later lookup must miss, + /// and replaying the same request id must not report success. + pub(crate) fn abandon_durable_commit( + &mut self, + generation: u64, + request_id: RequestId, + now_ms: u64, + ) -> Result<(), CacheError> { + let state = self.force_evict_committed(generation, now_ms)?; + if state != BlockState::Absent { + return Err(CacheError::InUse { refs: 1 }); + } + self.idem.remove(&request_id); + Ok(()) + } + pub(crate) fn export_committed(&self, generation: u64) -> Result { let slot = self .slots diff --git a/crates/storage/src/synthetic.rs b/crates/storage/src/synthetic.rs new file mode 100644 index 0000000..5c6b717 --- /dev/null +++ b/crates/storage/src/synthetic.rs @@ -0,0 +1,103 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 +//! Deterministic synthetic KV payload used by CPU tests and the fallback demo. +//! +//! The payload is a versioned list of token ids. Comparison is exact equality +//! of those ids. This is not a vLLM tensor, and it does not define a numeric +//! tolerance for a GPU kernel. A real engine export stays `NOT_RUN_HARDWARE`. + +use gpucache_core_types::{ByteReader, ByteWriter, CacheError}; + +const MAGIC: &[u8; 4] = b"SKV1"; +const VERSION: u16 = 1; +const MAX_TOKENS: usize = 4096; + +/// Encode token ids as a synthetic KV block. The bytes are the token ids. +pub fn encode_synthetic_kv(token_ids: &[u32]) -> Result, CacheError> { + if token_ids.is_empty() || token_ids.len() > MAX_TOKENS { + return Err(CacheError::Malformed { + reason: "synthetic tokens", + }); + } + let mut w = ByteWriter::with_capacity(10 + token_ids.len() * 4); + w.raw(MAGIC); + w.u16(VERSION); + w.u32( + u32::try_from(token_ids.len()).map_err(|_| CacheError::Malformed { + reason: "synthetic tokens", + })?, + ); + for id in token_ids { + w.u32(*id); + } + Ok(w.finish()) +} + +/// Decode a synthetic block. Trailing bytes, a bad magic, and a short buffer fail. +pub fn decode_synthetic_kv(bytes: &[u8]) -> Result, CacheError> { + let mut r = ByteReader::new(bytes); + let magic = r.raw(4)?; + if magic != MAGIC { + return Err(CacheError::Malformed { + reason: "synthetic kv", + }); + } + if r.u16()? != VERSION { + return Err(CacheError::Malformed { + reason: "synthetic kv", + }); + } + let n = usize::try_from(r.u32()?).map_err(|_| CacheError::Malformed { + reason: "synthetic tokens", + })?; + if n == 0 || n > MAX_TOKENS { + return Err(CacheError::Malformed { + reason: "synthetic tokens", + }); + } + let mut out = Vec::with_capacity(n); + for _ in 0..n { + out.push(r.u32()?); + } + r.finish()?; + Ok(out) +} + +/// Local recompute used when the cache misses. It is a pure function of the +/// token ids the caller already has. It does not read another tenant's block. +pub fn deterministic_recompute(token_ids: &[u32]) -> Result, CacheError> { + encode_synthetic_kv(token_ids) +} + +#[cfg(test)] +mod tests { + use super::{decode_synthetic_kv, deterministic_recompute, encode_synthetic_kv}; + use gpucache_core_types::CacheError; + + #[test] + fn exact_token_roundtrip_and_rejected_mutations() { + let tokens = vec![11, 22, 33, 44]; + let bytes = encode_synthetic_kv(&tokens).unwrap(); + assert_eq!(decode_synthetic_kv(&bytes).unwrap(), tokens); + assert_eq!( + decode_synthetic_kv(&deterministic_recompute(&tokens).unwrap()).unwrap(), + tokens + ); + assert!(encode_synthetic_kv(&[]).is_err()); + assert!(matches!( + decode_synthetic_kv(&bytes[..bytes.len() - 1]), + Err(CacheError::Malformed { .. }) + )); + let mut flipped = bytes.clone(); + flipped[0] ^= 0x01; + assert!(decode_synthetic_kv(&flipped).is_err()); + let mut trailed = bytes.clone(); + trailed.push(0); + assert!(matches!( + decode_synthetic_kv(&trailed), + Err(CacheError::Malformed { + reason: "trailing bytes" + }) + )); + } +} diff --git a/crates/storage/tests/fault_matrix.rs b/crates/storage/tests/fault_matrix.rs new file mode 100644 index 0000000..192d7e2 --- /dev/null +++ b/crates/storage/tests/fault_matrix.rs @@ -0,0 +1,825 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 +//! V1 C fault matrix for the file and memory stores. +//! +//! These tests use synthetic token blocks. A hit must decode to the same token +//! ids that were stored. A wrong tenant, model, tokenizer, adapter, or layout +//! must not return those ids. + +use std::fs; +use std::path::PathBuf; +use std::sync::atomic::{AtomicU64, Ordering}; + +use gpucache_core_types::{ + content_checksum, synthetic_identity, AdapterIdentity, Attestation, BlockState, CacheError, + Digest, QuotaKind, RequestId, ResourceLimits, TenantId, +}; +use gpucache_protocol::{ + AuthContext, CommitItem, GetItem, ItemHeader, LookupItem, LookupOutcome, MissReason, + ReleaseCancelItem, ReservePutItem, TenantPolicy, +}; +use gpucache_storage::{ + decode_synthetic_kv, encode_synthetic_kv, FileStore, MemoryStore, PersistFault, +}; + +const FP_A: [u8; 32] = [9u8; 32]; +const FP_B: [u8; 32] = [8u8; 32]; +const TOKENS_A: [u32; 4] = [11, 22, 33, 44]; +const TOKENS_B: [u32; 4] = [90, 91, 92, 93]; + +struct TempDir { + path: PathBuf, +} + +impl TempDir { + fn new() -> Self { + static N: AtomicU64 = AtomicU64::new(0); + let path = std::env::temp_dir().join(format!( + "gpucache-fault-{}-{}", + std::process::id(), + N.fetch_add(1, Ordering::Relaxed) + )); + let _ = fs::remove_dir_all(&path); + fs::create_dir_all(&path).unwrap(); + let path = fs::canonicalize(&path).unwrap(); + Self { path } + } +} + +impl Drop for TempDir { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.path); + } +} + +fn limits(retained: u64, per_tenant: u64) -> ResourceLimits { + let mut limits = ResourceLimits::dev_defaults(); + limits.max_object_bytes = 256; + limits.max_retained_bytes = retained; + limits.max_retained_bytes_per_tenant = per_tenant; + limits.max_in_flight_bytes_per_tenant = retained; + limits.max_registrations = 32; + limits.max_in_flight_requests = 32; + limits +} + +fn policy(nonce: u8, fingerprint: [u8; 32]) -> (TenantPolicy, AuthContext) { + let id = synthetic_identity(nonce, 1); + let policy = + TenantPolicy::new(id.tenant.clone(), id.trust_domain.clone(), fingerprint).unwrap(); + let auth = AuthContext::new(id.tenant, id.trust_domain, fingerprint).unwrap(); + (policy, auth) +} + +fn header(epoch: u64, n: u128) -> ItemHeader { + ItemHeader { + request_id: RequestId::from_counter(n).unwrap(), + epoch, + deadline_ms: 1_000, + } +} + +fn put( + epoch: u64, + n: u128, + bytes: &[u8], + identity: gpucache_core_types::CacheIdentity, +) -> ReservePutItem { + ReservePutItem { + header: header(epoch, n), + identity, + generation: 0, + length: bytes.len() as u64, + checksum: content_checksum(bytes), + payload: Some(bytes.to_vec()), + lease_ttl_ms: 5_000, + } +} + +fn commit_block( + store: &mut FileStore, + auth: &AuthContext, + n: u128, + bytes: &[u8], + identity: gpucache_core_types::CacheIdentity, +) -> u64 { + let epoch = store.epoch(); + let reserved = store + .reserve_put(auth, 0, vec![put(epoch, n, bytes, identity)]) + .unwrap(); + let generation = reserved.items[0].body.as_ref().unwrap().generation; + let committed = store + .commit( + auth, + 0, + vec![CommitItem { + header: header(epoch, n + 1), + generation, + length: bytes.len() as u64, + checksum: content_checksum(bytes), + }], + ) + .unwrap(); + assert!(committed.items[0].body.is_ok()); + generation +} + +fn lookup( + store: &mut FileStore, + auth: &AuthContext, + n: u128, + identity: gpucache_core_types::CacheIdentity, +) -> LookupOutcome { + let looked = store + .lookup( + auth, + 0, + vec![LookupItem { + header: header(store.epoch(), n), + identity, + }], + ) + .unwrap(); + looked.items.into_iter().next().unwrap().body.unwrap() +} + +fn assert_miss(outcome: LookupOutcome) { + assert!( + matches!(outcome, LookupOutcome::Miss { .. }), + "expected miss, got {outcome:?}" + ); +} + +fn payload_of(tokens: &[u32]) -> Vec { + encode_synthetic_kv(tokens).unwrap() +} + +#[test] +fn short_write_is_not_durable_and_replay_is_not_success() { + let dir = TempDir::new(); + let (policy, auth) = policy(1, FP_A); + let bytes = payload_of(&TOKENS_A); + let mut store = FileStore::open(&dir.path, limits(4096, 4096), vec![policy]).unwrap(); + store.inject_persist_fault(PersistFault::ShortWrite, 1); + let identity = synthetic_identity(1, 1); + let epoch = store.epoch(); + let reserved = store + .reserve_put(&auth, 0, vec![put(epoch, 10, &bytes, identity.clone())]) + .unwrap(); + let generation = reserved.items[0].body.as_ref().unwrap().generation; + let commit_item = CommitItem { + header: header(epoch, 11), + generation, + length: bytes.len() as u64, + checksum: content_checksum(&bytes), + }; + let failed = store.commit(&auth, 0, vec![commit_item.clone()]).unwrap(); + assert_eq!( + failed.items[0].body, + Err(CacheError::Storage { + reason: "short write", + }) + ); + assert!(!store.committed_path(generation).is_file()); + assert_miss(lookup(&mut store, &auth, 12, identity.clone())); + let replay = store.commit(&auth, 0, vec![commit_item]).unwrap(); + assert!( + replay.items[0].body.is_err(), + "replay of a failed short write must not report success" + ); + assert_eq!(store.stats().committed, 0); + let generation = commit_block(&mut store, &auth, 20, &bytes, identity.clone()); + let got = store + .get( + &auth, + 0, + vec![GetItem { + header: header(store.epoch(), 30), + generation, + identity, + pin_ttl_ms: 100, + }], + ) + .unwrap(); + let hit = got.items[0].body.as_ref().unwrap(); + assert_eq!(decode_synthetic_kv(&hit.payload).unwrap(), TOKENS_A); +} + +#[test] +fn disk_full_abandon_leaves_a_miss() { + let dir = TempDir::new(); + let (policy, auth) = policy(1, FP_A); + let bytes = payload_of(&TOKENS_A); + let mut store = FileStore::open(&dir.path, limits(4096, 4096), vec![policy]).unwrap(); + store.inject_persist_fault(PersistFault::DiskFull, 1); + let identity = synthetic_identity(1, 1); + let epoch = store.epoch(); + let reserved = store + .reserve_put(&auth, 0, vec![put(epoch, 10, &bytes, identity.clone())]) + .unwrap(); + let generation = reserved.items[0].body.as_ref().unwrap().generation; + let failed = store + .commit( + &auth, + 0, + vec![CommitItem { + header: header(epoch, 11), + generation, + length: bytes.len() as u64, + checksum: content_checksum(&bytes), + }], + ) + .unwrap(); + assert_eq!( + failed.items[0].body, + Err(CacheError::Storage { + reason: "disk full", + }) + ); + assert_miss(lookup(&mut store, &auth, 12, identity)); + assert_eq!(store.stats().committed, 0); + assert_eq!(store.stats().retained_bytes, 0); +} + +#[test] +fn truncated_block_file_is_a_miss_after_restart() { + let dir = TempDir::new(); + let (policy, auth) = policy(1, FP_A); + let bytes = payload_of(&TOKENS_A); + let path; + { + let mut store = + FileStore::open(&dir.path, limits(4096, 4096), vec![policy.clone()]).unwrap(); + let generation = commit_block(&mut store, &auth, 10, &bytes, synthetic_identity(1, 1)); + path = store.committed_path(generation); + } + let original = fs::read(&path).unwrap(); + fs::write(&path, &original[..8]).unwrap(); + let mut store = FileStore::open(&dir.path, limits(4096, 4096), vec![policy]).unwrap(); + assert!(store.skipped_corrupt() >= 1); + assert_miss(lookup(&mut store, &auth, 20, synthetic_identity(1, 1))); +} + +#[cfg(unix)] +#[test] +fn unwritable_blocks_directory_does_not_leave_a_hit() { + use std::os::unix::fs::PermissionsExt; + let dir = TempDir::new(); + let (policy, auth) = policy(1, FP_A); + let bytes = payload_of(&TOKENS_A); + let mut store = FileStore::open(&dir.path, limits(4096, 4096), vec![policy]).unwrap(); + let blocks = dir.path.join("blocks"); + let restore = blocks.clone(); + struct Guard(PathBuf); + impl Drop for Guard { + fn drop(&mut self) { + let _ = fs::set_permissions(&self.0, fs::Permissions::from_mode(0o755)); + } + } + let _guard = Guard(restore); + fs::set_permissions(&blocks, fs::Permissions::from_mode(0o555)).unwrap(); + let identity = synthetic_identity(1, 1); + let epoch = store.epoch(); + let reserved = store + .reserve_put(&auth, 0, vec![put(epoch, 10, &bytes, identity.clone())]) + .unwrap(); + let generation = reserved.items[0].body.as_ref().unwrap().generation; + let failed = store + .commit( + &auth, + 0, + vec![CommitItem { + header: header(epoch, 11), + generation, + length: bytes.len() as u64, + checksum: content_checksum(&bytes), + }], + ) + .unwrap(); + assert!(matches!( + failed.items[0].body, + Err(CacheError::Storage { .. }) + )); + fs::set_permissions(&blocks, fs::Permissions::from_mode(0o755)).unwrap(); + assert_miss(lookup(&mut store, &auth, 12, identity)); + assert_eq!(store.stats().committed, 0); +} + +#[test] +fn duplicate_commit_replays_one_block_and_conflict_hides_bytes() { + let dir = TempDir::new(); + let (policy, auth) = policy(1, FP_A); + let bytes = payload_of(&TOKENS_A); + let mut store = FileStore::open(&dir.path, limits(4096, 4096), vec![policy]).unwrap(); + let identity = synthetic_identity(1, 1); + let epoch = store.epoch(); + let reserved = store + .reserve_put(&auth, 0, vec![put(epoch, 10, &bytes, identity.clone())]) + .unwrap(); + let generation = reserved.items[0].body.as_ref().unwrap().generation; + let item = CommitItem { + header: header(epoch, 11), + generation, + length: bytes.len() as u64, + checksum: content_checksum(&bytes), + }; + let first = store.commit(&auth, 0, vec![item.clone()]).unwrap(); + let second = store.commit(&auth, 0, vec![item]).unwrap(); + assert_eq!(first.items[0].body, second.items[0].body); + assert_eq!(store.stats().committed, 1); + let mut other = bytes.clone(); + other[10] ^= 0xff; + let conflict = store + .reserve_put( + &auth, + 0, + vec![ReservePutItem { + header: header(epoch, 10), + identity: identity.clone(), + generation: 0, + length: other.len() as u64, + checksum: content_checksum(&other), + payload: Some(other), + lease_ttl_ms: 5_000, + }], + ) + .unwrap(); + assert!(matches!( + conflict.items[0].body, + Err(CacheError::IdempotencyConflict) + )); + let got = store + .get( + &auth, + 0, + vec![GetItem { + header: header(epoch, 30), + generation, + identity, + pin_ttl_ms: 100, + }], + ) + .unwrap(); + assert_eq!( + decode_synthetic_kv(&got.items[0].body.as_ref().unwrap().payload).unwrap(), + TOKENS_A + ); +} + +#[test] +fn commit_before_reserve_and_duplicate_batch_ids_fail_closed() { + let dir = TempDir::new(); + let (policy, auth) = policy(1, FP_A); + let mut store = FileStore::open(&dir.path, limits(4096, 4096), vec![policy]).unwrap(); + let epoch = store.epoch(); + let early = store + .commit( + &auth, + 0, + vec![CommitItem { + header: header(epoch, 1), + generation: 99, + length: 4, + checksum: content_checksum(&[1, 2, 3, 4]), + }], + ) + .unwrap(); + assert!(matches!( + early.items[0].body, + Err(CacheError::UnknownGeneration) + )); + let identity = synthetic_identity(1, 1); + let duplicate = store.lookup( + &auth, + 0, + vec![ + LookupItem { + header: header(epoch, 2), + identity: identity.clone(), + }, + LookupItem { + header: header(epoch, 2), + identity, + }, + ], + ); + assert!(matches!( + duplicate, + Err(CacheError::Malformed { + reason: "duplicate request_id" + }) + )); + assert_eq!(store.stats().committed, 0); +} + +#[test] +fn timeout_and_cancel_do_not_return_payload() { + let dir = TempDir::new(); + let (policy, auth) = policy(1, FP_A); + let bytes = payload_of(&TOKENS_A); + let mut store = FileStore::open(&dir.path, limits(4096, 4096), vec![policy]).unwrap(); + let identity = synthetic_identity(1, 1); + let epoch = store.epoch(); + let reserved = store + .reserve_put(&auth, 0, vec![put(epoch, 10, &bytes, identity.clone())]) + .unwrap(); + let generation = reserved.items[0].body.as_ref().unwrap().generation; + let timed = store + .commit( + &auth, + 50, + vec![CommitItem { + header: ItemHeader { + request_id: RequestId::from_counter(11).unwrap(), + epoch, + deadline_ms: 10, + }, + generation, + length: bytes.len() as u64, + checksum: content_checksum(&bytes), + }], + ) + .unwrap(); + assert_eq!(timed.items[0].body, Err(CacheError::TimedOut)); + let cancelled = store + .release_cancel( + &auth, + 0, + vec![ReleaseCancelItem { + header: header(epoch, 12), + generation, + region_id: None, + pin_owner: None, + attestation: Attestation::Unknown, + }], + ) + .unwrap(); + assert_eq!( + cancelled.items[0].body.as_ref().unwrap().state, + BlockState::Quarantined + ); + assert_eq!( + lookup(&mut store, &auth, 13, identity.clone()), + LookupOutcome::Miss { + reason: MissReason::Quarantined + } + ); + let got = store + .get( + &auth, + 0, + vec![GetItem { + header: header(epoch, 14), + generation, + identity, + pin_ttl_ms: 100, + }], + ) + .unwrap(); + assert_eq!(got.items[0].body, Err(CacheError::Cancelled)); +} + +#[test] +fn pinned_read_keeps_exact_tokens_while_eviction_is_refused() { + let dir = TempDir::new(); + let (policy, auth) = policy(1, FP_A); + let bytes_a = payload_of(&TOKENS_A); + let bytes_b = payload_of(&TOKENS_B); + let mut tight = limits(bytes_a.len() as u64, bytes_a.len() as u64); + tight.max_object_bytes = bytes_a.len() as u64; + let mut store = FileStore::open(&dir.path, tight, vec![policy]).unwrap(); + let mut id_a = synthetic_identity(1, 1); + id_a.prefix_chain[0].0[0] = 1; + let mut id_b = synthetic_identity(1, 1); + id_b.prefix_chain[0].0[0] = 2; + let generation = commit_block(&mut store, &auth, 10, &bytes_a, id_a.clone()); + let got = store + .get( + &auth, + 0, + vec![GetItem { + header: header(store.epoch(), 20), + generation, + identity: id_a.clone(), + pin_ttl_ms: 1_000, + }], + ) + .unwrap(); + let pin = got.items[0].body.as_ref().unwrap(); + assert_eq!(decode_synthetic_kv(&pin.payload).unwrap(), TOKENS_A); + let denied = store + .reserve_put(&auth, 0, vec![put(store.epoch(), 30, &bytes_b, id_b)]) + .unwrap(); + assert!(matches!( + denied.items[0].body, + Err(CacheError::QuotaExceeded { .. }) + )); + let again = store + .get( + &auth, + 0, + vec![GetItem { + header: header(store.epoch(), 21), + generation, + identity: id_a.clone(), + pin_ttl_ms: 1_000, + }], + ) + .unwrap(); + assert_eq!( + decode_synthetic_kv(&again.items[0].body.as_ref().unwrap().payload).unwrap(), + TOKENS_A + ); + store + .release_cancel( + &auth, + 0, + vec![ + ReleaseCancelItem { + header: header(store.epoch(), 22), + generation, + region_id: Some(pin.region_id), + pin_owner: Some(RequestId::from_counter(20).unwrap()), + attestation: Attestation::Drained, + }, + ReleaseCancelItem { + header: header(store.epoch(), 23), + generation, + region_id: Some(again.items[0].body.as_ref().unwrap().region_id), + pin_owner: Some(RequestId::from_counter(21).unwrap()), + attestation: Attestation::Drained, + }, + ], + ) + .unwrap(); +} + +#[test] +fn tenant_key_collision_never_returns_the_other_payload() { + let dir = TempDir::new(); + let (policy_a, auth_a) = policy(1, FP_A); + let mut id_b = synthetic_identity(1, 1); + id_b.tenant = TenantId::new(b"tenant-b".to_vec()).unwrap(); + let policy_b = TenantPolicy::new(id_b.tenant.clone(), id_b.trust_domain.clone(), FP_B).unwrap(); + let auth_b = AuthContext::new(id_b.tenant.clone(), id_b.trust_domain.clone(), FP_B).unwrap(); + let mut store = + FileStore::open(&dir.path, limits(4096, 4096), vec![policy_a, policy_b]).unwrap(); + let mut id_a = synthetic_identity(1, 1); + id_a.prefix_chain[0].0[1] = 7; + id_b.prefix_chain[0].0[1] = 7; + let bytes_a = payload_of(&TOKENS_A); + let bytes_b = payload_of(&TOKENS_B); + let gen_a = commit_block(&mut store, &auth_a, 10, &bytes_a, id_a.clone()); + let gen_b = commit_block(&mut store, &auth_b, 20, &bytes_b, id_b.clone()); + assert_ne!(gen_a, gen_b); + let got_a = store + .get( + &auth_a, + 0, + vec![GetItem { + header: header(store.epoch(), 30), + generation: gen_a, + identity: id_a.clone(), + pin_ttl_ms: 100, + }], + ) + .unwrap(); + assert_eq!( + decode_synthetic_kv(&got_a.items[0].body.as_ref().unwrap().payload).unwrap(), + TOKENS_A + ); + let foreign = store + .lookup( + &auth_a, + 0, + vec![LookupItem { + header: header(store.epoch(), 31), + identity: id_b.clone(), + }], + ) + .unwrap(); + assert_eq!(foreign.items[0].body, Err(CacheError::Unauthorized)); + let foreign_get = store + .get( + &auth_a, + 0, + vec![GetItem { + header: header(store.epoch(), 33), + generation: gen_b, + identity: id_b.clone(), + pin_ttl_ms: 100, + }], + ) + .unwrap(); + assert_eq!(foreign_get.items[0].body, Err(CacheError::Unauthorized)); + let got_b = store + .get( + &auth_b, + 0, + vec![GetItem { + header: header(store.epoch(), 32), + generation: gen_b, + identity: id_b, + pin_ttl_ms: 100, + }], + ) + .unwrap(); + assert_eq!( + decode_synthetic_kv(&got_b.items[0].body.as_ref().unwrap().payload).unwrap(), + TOKENS_B + ); +} + +#[test] +fn model_tokenizer_adapter_and_layout_changes_are_not_hits() { + let dir = TempDir::new(); + let (policy, auth) = policy(1, FP_A); + let bytes = payload_of(&TOKENS_A); + let mut store = FileStore::open(&dir.path, limits(4096, 4096), vec![policy]).unwrap(); + let identity = synthetic_identity(1, 1); + let generation = commit_block(&mut store, &auth, 10, &bytes, identity.clone()); + let mut model = identity.clone(); + model.model_revision.0[4] ^= 0x5a; + let mut config = identity.clone(); + config.model_config_hash.0[4] ^= 0x5a; + let mut tokenizer = identity.clone(); + tokenizer.tokenizer_revision.0[4] ^= 0x5a; + let mut adapter = identity.clone(); + adapter.adapter = AdapterIdentity::Lora { + revision: Digest([9u8; 32]), + }; + let mut layout = identity.clone(); + layout.layout.family = 2; + assert_eq!( + lookup(&mut store, &auth, 20, model.clone()), + LookupOutcome::Miss { + reason: MissReason::Absent + } + ); + assert_eq!( + lookup(&mut store, &auth, 21, config), + LookupOutcome::Miss { + reason: MissReason::Absent + } + ); + assert_eq!( + lookup(&mut store, &auth, 22, tokenizer), + LookupOutcome::Miss { + reason: MissReason::Absent + } + ); + assert_eq!( + lookup(&mut store, &auth, 23, adapter), + LookupOutcome::Miss { + reason: MissReason::Absent + } + ); + assert_eq!( + lookup(&mut store, &auth, 24, layout.clone()), + LookupOutcome::Miss { + reason: MissReason::Incompatible + } + ); + let wrong_get = store + .get( + &auth, + 0, + vec![GetItem { + header: header(store.epoch(), 25), + generation, + identity: model, + pin_ttl_ms: 100, + }], + ) + .unwrap(); + assert_eq!(wrong_get.items[0].body, Err(CacheError::Incompatible)); + let layout_get = store + .get( + &auth, + 0, + vec![GetItem { + header: header(store.epoch(), 26), + generation, + identity: layout, + pin_ttl_ms: 100, + }], + ) + .unwrap(); + assert_eq!(layout_get.items[0].body, Err(CacheError::Incompatible)); + let exact = store + .get( + &auth, + 0, + vec![GetItem { + header: header(store.epoch(), 27), + generation, + identity, + pin_ttl_ms: 100, + }], + ) + .unwrap(); + assert_eq!( + decode_synthetic_kv(&exact.items[0].body.as_ref().unwrap().payload).unwrap(), + TOKENS_A + ); +} + +#[test] +fn memory_quota_exhaustion_does_not_evict_or_return_foreign_bytes() { + let bytes = payload_of(&TOKENS_A); + let mut limits = ResourceLimits::dev_defaults(); + limits.max_object_bytes = bytes.len() as u64; + limits.max_retained_bytes = bytes.len() as u64; + limits.max_retained_bytes_per_tenant = bytes.len() as u64; + limits.max_in_flight_bytes_per_tenant = bytes.len() as u64; + limits.max_registrations = 1; + let identity = synthetic_identity(1, 1); + let policy = + TenantPolicy::new(identity.tenant.clone(), identity.trust_domain.clone(), FP_A).unwrap(); + let auth = + AuthContext::new(identity.tenant.clone(), identity.trust_domain.clone(), FP_A).unwrap(); + let mut store = MemoryStore::new(limits, vec![policy]).unwrap(); + let epoch = store.epoch(); + let reserved = store + .reserve_put(&auth, 0, vec![put(epoch, 10, &bytes, identity.clone())]) + .unwrap(); + let generation = reserved.items[0].body.as_ref().unwrap().generation; + store + .commit( + &auth, + 0, + vec![CommitItem { + header: header(epoch, 11), + generation, + length: bytes.len() as u64, + checksum: content_checksum(&bytes), + }], + ) + .unwrap() + .items[0] + .body + .as_ref() + .unwrap(); + let mut other = identity.clone(); + other.prefix_chain[0].0[2] = 9; + let denied = store + .reserve_put(&auth, 0, vec![put(epoch, 20, &bytes, other)]) + .unwrap(); + assert!(matches!( + denied.items[0].body, + Err(CacheError::QuotaExceeded { + kind: QuotaKind::RetainedBytes | QuotaKind::Registrations + }) + )); + assert_eq!(store.stats().committed, 1); + assert_eq!(store.stats().retained_bytes, bytes.len() as u64); + let got = store + .get( + &auth, + 0, + vec![GetItem { + header: header(epoch, 30), + generation, + identity, + pin_ttl_ms: 100, + }], + ) + .unwrap(); + assert_eq!( + decode_synthetic_kv(&got.items[0].body.as_ref().unwrap().payload).unwrap(), + TOKENS_A + ); +} + +proptest::proptest! { + #![proptest_config(proptest::prelude::ProptestConfig::with_cases(12))] + #[test] + fn flipped_model_byte_is_never_a_hit(index in 0usize..32) { + let bytes = payload_of(&TOKENS_A); + let mut limits = ResourceLimits::dev_defaults(); + limits.max_object_bytes = 256; + limits.max_retained_bytes = 4096; + limits.max_retained_bytes_per_tenant = 4096; + limits.max_in_flight_bytes_per_tenant = 4096; + let identity = synthetic_identity(1, 1); + let policy = TenantPolicy::new(identity.tenant.clone(), identity.trust_domain.clone(), FP_A).unwrap(); + let auth = AuthContext::new(identity.tenant.clone(), identity.trust_domain.clone(), FP_A).unwrap(); + let mut store = MemoryStore::new(limits, vec![policy]).unwrap(); + let epoch = store.epoch(); + let reserved = store.reserve_put(&auth, 0, vec![put(epoch, 10, &bytes, identity.clone())]).unwrap(); + let generation = reserved.items[0].body.as_ref().unwrap().generation; + store.commit(&auth, 0, vec![CommitItem { + header: header(epoch, 11), + generation, + length: bytes.len() as u64, + checksum: content_checksum(&bytes), + }]).unwrap().items[0].body.as_ref().unwrap(); + let mut wrong = identity.clone(); + wrong.model_revision.0[index] ^= 0x5a; + let looked = store.lookup(&auth, 0, vec![LookupItem { + header: header(epoch, 12), + identity: wrong, + }]).unwrap(); + assert!(matches!(looked.items[0].body, Ok(LookupOutcome::Miss { .. }))); + } +} diff --git a/crates/transport/Cargo.toml b/crates/transport/Cargo.toml index c6c8950..bd9b455 100644 --- a/crates/transport/Cargo.toml +++ b/crates/transport/Cargo.toml @@ -21,5 +21,9 @@ path = "src/bin/gpucache_tcp.rs" name = "share-demo" path = "src/bin/share_demo.rs" +[[bin]] +name = "fault-demo" +path = "src/bin/fault_demo.rs" + [lints] workspace = true diff --git a/crates/transport/src/bin/fault_demo.rs b/crates/transport/src/bin/fault_demo.rs new file mode 100644 index 0000000..76feaa8 --- /dev/null +++ b/crates/transport/src/bin/fault_demo.rs @@ -0,0 +1,24 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 +//! Hit, miss, and recompute fallback from a clean temp directory. +//! +//! The bytes are synthetic token ids. This is not a vLLM run and not a DPU run. +//! A failed check exits non-zero and does not print `status=FALLBACK_OK`. + +use std::process::ExitCode; + +use gpucache_transport::run_fault_demo; + +fn main() -> ExitCode { + match run_fault_demo() { + Ok(report) => { + println!("{report}"); + ExitCode::SUCCESS + } + Err(err) => { + eprintln!("status=FALLBACK_FAILED"); + eprintln!("error={err}"); + ExitCode::from(2) + } + } +} diff --git a/crates/transport/src/fault_demo.rs b/crates/transport/src/fault_demo.rs new file mode 100644 index 0000000..da39849 --- /dev/null +++ b/crates/transport/src/fault_demo.rs @@ -0,0 +1,612 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 +//! End-to-end hit, miss, and recompute fallback on a clean temp directory. +//! +//! Token comparison is exact. A wrong tenant or wrong model is not a hit. +//! Nothing here talks to a GPU or an inference engine. + +use std::fs; +use std::net::SocketAddr; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::thread; +use std::time::Duration; + +use gpucache_core_types::{ + content_checksum, synthetic_identity, AdapterIdentity, Attestation, CacheError, Digest, + RequestId, ResourceLimits, TenantId, +}; +use gpucache_protocol::{ + AuthContext, CommitItem, GetItem, ItemHeader, LookupItem, LookupOutcome, MissReason, + ReleaseCancelItem, ReservePutItem, TenantPolicy, +}; +use gpucache_storage::{ + decode_synthetic_kv, deterministic_recompute, encode_synthetic_kv, FileStore, +}; + +use crate::client::TcpClient; +use crate::server::{serve, ServerConfig}; + +const FP_A: [u8; 32] = [9u8; 32]; +const FP_B: [u8; 32] = [4u8; 32]; +const TOKENS: [u32; 4] = [11, 22, 33, 44]; + +/// Result of one clean-environment fallback run. Printed only after the checks pass. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct FaultReport { + pub hit_tokens: u32, + pub miss: &'static str, + pub corrupt_fallback: &'static str, + pub timeout_fallback: &'static str, + pub cancel_fallback: &'static str, + pub restart: &'static str, + pub restart_down: &'static str, + pub wrong_tenant: &'static str, + pub wrong_model: &'static str, + pub wrong_tokenizer: &'static str, + pub wrong_adapter: &'static str, + pub layout_mismatch: &'static str, + pub quota_fallback: &'static str, + pub duplicate: &'static str, + pub reorder: &'static str, +} + +impl std::fmt::Display for FaultReport { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + writeln!(f, "status=FALLBACK_OK")?; + writeln!(f, "criteria=EXACT_TOKEN_IDS")?; + writeln!(f, "tolerance=NONE")?; + writeln!(f, "hit=PASS tokens={}", self.hit_tokens)?; + writeln!(f, "miss={}", self.miss)?; + writeln!(f, "corrupt_fallback={}", self.corrupt_fallback)?; + writeln!(f, "timeout_fallback={}", self.timeout_fallback)?; + writeln!(f, "cancel_fallback={}", self.cancel_fallback)?; + writeln!(f, "restart={}", self.restart)?; + writeln!(f, "restart_down={}", self.restart_down)?; + writeln!(f, "wrong_tenant={}", self.wrong_tenant)?; + writeln!(f, "wrong_model={}", self.wrong_model)?; + writeln!(f, "wrong_tokenizer={}", self.wrong_tokenizer)?; + writeln!(f, "wrong_adapter={}", self.wrong_adapter)?; + writeln!(f, "layout_mismatch={}", self.layout_mismatch)?; + writeln!(f, "quota_fallback={}", self.quota_fallback)?; + writeln!(f, "duplicate={}", self.duplicate)?; + write!(f, "reorder={}", self.reorder) + } +} + +pub fn run_fault_demo() -> Result { + let dir = temp_dir()?; + let result = run_in(&dir.path); + let _ = fs::remove_dir_all(&dir.path); + result +} + +struct TempPath { + path: PathBuf, +} + +fn temp_dir() -> Result { + static N: AtomicU64 = AtomicU64::new(0); + let path = std::env::temp_dir().join(format!( + "gpucache-fault-demo-{}-{}", + std::process::id(), + N.fetch_add(1, Ordering::Relaxed) + )); + let _ = fs::remove_dir_all(&path); + fs::create_dir_all(&path).map_err(|_| CacheError::Storage { reason: "store io" })?; + let path = fs::canonicalize(&path).map_err(|_| CacheError::Storage { reason: "store io" })?; + Ok(TempPath { path }) +} + +fn run_in(path: &Path) -> Result { + let writer_id = synthetic_identity(1, 1); + let mut other_id = synthetic_identity(1, 1); + other_id.tenant = TenantId::new(b"tenant-b".to_vec())?; + let policy_a = TenantPolicy::new( + writer_id.tenant.clone(), + writer_id.trust_domain.clone(), + FP_A, + )?; + let policy_b = TenantPolicy::new(other_id.tenant.clone(), other_id.trust_domain.clone(), FP_B)?; + let payload = encode_synthetic_kv(&TOKENS)?; + let store = FileStore::open( + path, + ResourceLimits::dev_defaults(), + vec![policy_a.clone(), policy_b], + )?; + let server = serve(store, ServerConfig::default())?; + let addr = server.addr(); + let auth = AuthContext::new( + writer_id.tenant.clone(), + writer_id.trust_domain.clone(), + FP_A, + )?; + let mut writer = connect(addr, auth.clone())?; + let mut reader = connect(addr, auth.clone())?; + let generation = put_commit(&mut writer, &payload, writer_id.clone(), 10)?; + let hit_tokens = expect_exact_hit(&mut reader, &writer_id, generation, &TOKENS, 300)?; + let miss = expect_miss( + &mut reader, + &mut_prefix(writer_id.clone(), 9)?, + MissReason::Absent, + )?; + fallback_matches(&TOKENS)?; + thread::sleep(Duration::from_millis(20)); + let timed = reader.lookup(vec![LookupItem { + header: ItemHeader { + request_id: RequestId::from_counter(40)?, + epoch: reader.epoch(), + deadline_ms: 1, + }, + identity: writer_id.clone(), + }])?; + if timed.items[0].body != Err(CacheError::TimedOut) { + return Err(CacheError::Malformed { + reason: "fault demo timeout", + }); + } + expect_exact_hit(&mut reader, &writer_id, generation, &TOKENS, 310)?; + let cancel_id = mut_prefix(writer_id.clone(), 3)?; + let cancel_generation = reserve_only(&mut writer, &payload, cancel_id.clone(), 50)?; + let cancelled = writer.release_cancel(vec![ReleaseCancelItem { + header: header(&writer, 52)?, + generation: cancel_generation, + region_id: None, + pin_owner: None, + attestation: Attestation::Unknown, + }])?; + if cancelled.items[0] + .body + .as_ref() + .map(|item| item.quarantined) + != Ok(true) + { + return Err(CacheError::Malformed { + reason: "fault demo cancel", + }); + } + let cancel_fallback = expect_miss(&mut reader, &cancel_id, MissReason::Quarantined)?; + let blocked = reader.get(vec![GetItem { + header: header(&reader, 53)?, + generation: cancel_generation, + identity: cancel_id, + pin_ttl_ms: 100, + }])?; + if blocked.items[0].body != Err(CacheError::Cancelled) { + return Err(CacheError::Malformed { + reason: "fault demo cancel get", + }); + } + let duplicate = replay_same_commit(&mut writer, &payload, writer_id.clone())?; + let reorder = commit_unknown_generation(&mut writer)?; + let wrong_tenant = tenant_denied(addr, &other_id, &writer_id)?; + let wrong_model = { + let mut id = writer_id.clone(); + id.model_revision.0[4] ^= 0x5a; + expect_miss(&mut reader, &id, MissReason::Absent)? + }; + let wrong_tokenizer = { + let mut id = writer_id.clone(); + id.tokenizer_revision.0[4] ^= 0x5a; + expect_miss(&mut reader, &id, MissReason::Absent)? + }; + let wrong_adapter = { + let mut id = writer_id.clone(); + id.adapter = AdapterIdentity::Lora { + revision: Digest([9u8; 32]), + }; + expect_miss(&mut reader, &id, MissReason::Absent)? + }; + let layout_mismatch = { + let mut id = writer_id.clone(); + id.layout.family = 2; + expect_miss(&mut reader, &id, MissReason::Incompatible)? + }; + drop(writer); + drop(reader); + drop(server); + let restart_down = + match TcpClient::connect(addr, auth.clone(), Duration::from_millis(200), 1024 * 1024) { + Err(CacheError::Storage { .. } | CacheError::TimedOut) => "TRANSPORT_FALLBACK", + Ok(_) => { + return Err(CacheError::Malformed { + reason: "fault demo restart", + }) + } + Err(_) => { + return Err(CacheError::Malformed { + reason: "fault demo restart", + }) + } + }; + let block_path = path.join("blocks").join(format!("{generation:016x}.gblk")); + { + let store = FileStore::open( + path, + ResourceLimits::dev_defaults(), + vec![policy_a.clone(), policy_b_from(&other_id)?], + )?; + let server = serve(store, ServerConfig::default())?; + let mut reader = connect(server.addr(), auth.clone())?; + expect_exact_hit(&mut reader, &writer_id, generation, &TOKENS, 320)?; + drop(reader); + drop(server); + } + corrupt_last_byte(&block_path)?; + let store = FileStore::open( + path, + ResourceLimits::dev_defaults(), + vec![policy_a, policy_b_from(&other_id)?], + )?; + if store.skipped_corrupt() == 0 { + return Err(CacheError::Malformed { + reason: "fault demo corrupt", + }); + } + let server = serve(store, ServerConfig::default())?; + let mut reader = connect(server.addr(), auth)?; + let corrupt_fallback = expect_miss(&mut reader, &writer_id, MissReason::Absent)?; + fallback_matches(&TOKENS)?; + drop(reader); + drop(server); + let quota_fallback = quota_fallback_server()?; + Ok(FaultReport { + hit_tokens, + miss, + corrupt_fallback, + timeout_fallback: "RECOMPUTE", + cancel_fallback, + restart: "HIT_EXACT", + restart_down, + wrong_tenant, + wrong_model, + wrong_tokenizer, + wrong_adapter, + layout_mismatch, + quota_fallback, + duplicate, + reorder, + }) +} + +fn policy_b_from( + other_id: &gpucache_core_types::CacheIdentity, +) -> Result { + TenantPolicy::new(other_id.tenant.clone(), other_id.trust_domain.clone(), FP_B) +} + +fn connect(addr: SocketAddr, auth: AuthContext) -> Result { + TcpClient::connect(addr, auth, Duration::from_secs(5), 1024 * 1024) +} + +fn header(client: &TcpClient, n: u128) -> Result { + Ok(ItemHeader { + request_id: RequestId::from_counter(n)?, + epoch: client.epoch(), + deadline_ms: client.deadline(30_000), + }) +} + +fn put_commit( + client: &mut TcpClient, + payload: &[u8], + identity: gpucache_core_types::CacheIdentity, + n: u128, +) -> Result { + let generation = reserve_only(client, payload, identity, n)?; + let committed = client.commit(vec![CommitItem { + header: header(client, n + 1)?, + generation, + length: payload.len() as u64, + checksum: content_checksum(payload), + }])?; + if committed.items[0].body.is_err() { + return Err(CacheError::Malformed { + reason: "fault demo commit", + }); + } + Ok(generation) +} + +fn reserve_only( + client: &mut TcpClient, + payload: &[u8], + identity: gpucache_core_types::CacheIdentity, + n: u128, +) -> Result { + let reserved = client.reserve_put(vec![ReservePutItem { + header: header(client, n)?, + identity, + generation: 0, + length: payload.len() as u64, + checksum: content_checksum(payload), + payload: Some(payload.to_vec()), + lease_ttl_ms: 30_000, + }])?; + reserved.items[0] + .body + .as_ref() + .map(|item| item.generation) + .map_err(|_| CacheError::Malformed { + reason: "fault demo reserve", + }) +} + +fn expect_exact_hit( + client: &mut TcpClient, + identity: &gpucache_core_types::CacheIdentity, + generation: u64, + tokens: &[u32], + n: u128, +) -> Result { + let looked = client.lookup(vec![LookupItem { + header: header(client, n)?, + identity: identity.clone(), + }])?; + let LookupOutcome::Hit { + generation: hit_generation, + .. + } = looked.items[0] + .body + .as_ref() + .map_err(|_| CacheError::Malformed { + reason: "fault demo hit", + })? + else { + return Err(CacheError::Malformed { + reason: "fault demo hit", + }); + }; + if *hit_generation != generation { + return Err(CacheError::Malformed { + reason: "fault demo hit", + }); + } + let got = client.get(vec![GetItem { + header: header(client, n + 1)?, + generation, + identity: identity.clone(), + pin_ttl_ms: 1_000, + }])?; + let hit = got.items[0] + .body + .as_ref() + .map_err(|_| CacheError::Malformed { + reason: "fault demo hit", + })?; + let decoded = decode_synthetic_kv(&hit.payload)?; + if decoded != tokens { + return Err(CacheError::Malformed { + reason: "fault demo token mismatch", + }); + } + let region_id = hit.region_id; + client.release_cancel(vec![ReleaseCancelItem { + header: header(client, n + 2)?, + generation, + region_id: Some(region_id), + pin_owner: Some(RequestId::from_counter(n + 1)?), + attestation: Attestation::Drained, + }])?; + u32::try_from(decoded.len()).map_err(|_| CacheError::Malformed { + reason: "fault demo hit", + }) +} + +fn expect_miss( + client: &mut TcpClient, + identity: &gpucache_core_types::CacheIdentity, + reason: MissReason, +) -> Result<&'static str, CacheError> { + let looked = client.lookup(vec![LookupItem { + header: header(client, 210 + u128::from(reason_tag(reason)))?, + identity: identity.clone(), + }])?; + match &looked.items[0].body { + Ok(LookupOutcome::Miss { reason: got }) if *got == reason => Ok(miss_label(reason)), + _ => Err(CacheError::Malformed { + reason: "fault demo miss", + }), + } +} + +fn reason_tag(reason: MissReason) -> u8 { + match reason { + MissReason::Absent => 1, + MissReason::NotCommitted => 2, + MissReason::Incompatible => 3, + MissReason::Evicting => 4, + MissReason::Quarantined => 5, + MissReason::Expired => 6, + } +} + +fn miss_label(reason: MissReason) -> &'static str { + match reason { + MissReason::Absent => "MISS_ABSENT", + MissReason::NotCommitted => "MISS_NOT_COMMITTED", + MissReason::Incompatible => "MISS_INCOMPATIBLE", + MissReason::Evicting => "MISS_EVICTING", + MissReason::Quarantined => "MISS_QUARANTINED", + MissReason::Expired => "MISS_EXPIRED", + } +} + +fn fallback_matches(tokens: &[u32]) -> Result<(), CacheError> { + let recomputed = deterministic_recompute(tokens)?; + if decode_synthetic_kv(&recomputed)? != tokens { + return Err(CacheError::Malformed { + reason: "fault demo recompute", + }); + } + Ok(()) +} + +fn mut_prefix( + mut identity: gpucache_core_types::CacheIdentity, + mark: u8, +) -> Result { + identity.prefix_chain[0].0[2] = mark; + Ok(identity) +} + +fn replay_same_commit( + client: &mut TcpClient, + payload: &[u8], + identity: gpucache_core_types::CacheIdentity, +) -> Result<&'static str, CacheError> { + let mut identity = identity; + identity.prefix_chain[0].0[2] = 4; + let generation = reserve_only(client, payload, identity.clone(), 60)?; + let item = CommitItem { + header: header(client, 61)?, + generation, + length: payload.len() as u64, + checksum: content_checksum(payload), + }; + let first = client.commit(vec![item.clone()])?; + let second = client.commit(vec![item])?; + if first.items[0].body != second.items[0].body || first.items[0].body.is_err() { + return Err(CacheError::Malformed { + reason: "fault demo duplicate", + }); + } + let again = client.reserve_put(vec![ReservePutItem { + header: header(client, 62)?, + identity, + generation: 0, + length: payload.len() as u64, + checksum: content_checksum(payload), + payload: Some(payload.to_vec()), + lease_ttl_ms: 30_000, + }])?; + if !matches!( + again.items[0].body, + Err(CacheError::InvalidTransition { .. }) + ) { + return Err(CacheError::Malformed { + reason: "fault demo duplicate", + }); + } + Ok("REPLAY_ONE_BLOCK") +} + +fn commit_unknown_generation(client: &mut TcpClient) -> Result<&'static str, CacheError> { + let early = client.commit(vec![CommitItem { + header: header(client, 70)?, + generation: 99, + length: 4, + checksum: content_checksum(&[1, 2, 3, 4]), + }])?; + if early.items[0].body != Err(CacheError::UnknownGeneration) { + return Err(CacheError::Malformed { + reason: "fault demo reorder", + }); + } + Ok("REJECTED") +} + +fn tenant_denied( + addr: SocketAddr, + other_id: &gpucache_core_types::CacheIdentity, + writer_id: &gpucache_core_types::CacheIdentity, +) -> Result<&'static str, CacheError> { + let auth = AuthContext::new(other_id.tenant.clone(), other_id.trust_domain.clone(), FP_B)?; + let mut intruder = connect(addr, auth)?; + let denied = intruder.lookup(vec![LookupItem { + header: header(&intruder, 80)?, + identity: writer_id.clone(), + }])?; + if denied.items[0].body != Err(CacheError::Unauthorized) { + return Err(CacheError::Malformed { + reason: "fault demo tenant", + }); + } + Ok("UNAUTHORIZED") +} + +fn corrupt_last_byte(path: &Path) -> Result<(), CacheError> { + let mut bytes = fs::read(path).map_err(|_| CacheError::Storage { reason: "store io" })?; + if bytes.is_empty() { + return Err(CacheError::Malformed { + reason: "fault demo corrupt", + }); + } + let last = bytes.len() - 1; + bytes[last] ^= 0x01; + fs::write(path, bytes).map_err(|_| CacheError::Storage { reason: "store io" })?; + Ok(()) +} + +fn quota_fallback_server() -> Result<&'static str, CacheError> { + let dir = temp_dir()?; + let result = quota_in(&dir.path); + let _ = fs::remove_dir_all(&dir.path); + result +} + +fn quota_in(path: &Path) -> Result<&'static str, CacheError> { + let payload = encode_synthetic_kv(&TOKENS)?; + let mut limits = ResourceLimits::dev_defaults(); + let len = payload.len() as u64; + limits.max_object_bytes = len; + limits.max_retained_bytes = len; + limits.max_retained_bytes_per_tenant = len; + limits.max_in_flight_bytes_per_tenant = len; + let identity = synthetic_identity(1, 1); + let policy = TenantPolicy::new(identity.tenant.clone(), identity.trust_domain.clone(), FP_A)?; + let auth = AuthContext::new(identity.tenant.clone(), identity.trust_domain.clone(), FP_A)?; + let store = FileStore::open(path, limits, vec![policy])?; + let server = serve(store, ServerConfig::default())?; + let mut client = connect(server.addr(), auth)?; + let generation = put_commit(&mut client, &payload, identity.clone(), 90)?; + let got = client.get(vec![GetItem { + header: header(&client, 93)?, + generation, + identity: identity.clone(), + pin_ttl_ms: 30_000, + }])?; + let pin = got.items[0] + .body + .as_ref() + .map_err(|_| CacheError::Malformed { + reason: "fault demo quota", + })?; + if decode_synthetic_kv(&pin.payload)? != TOKENS { + return Err(CacheError::Malformed { + reason: "fault demo quota", + }); + } + let mut second = identity.clone(); + second.prefix_chain[0].0[2] = 8; + let denied = client.reserve_put(vec![ReservePutItem { + header: header(&client, 94)?, + identity: second.clone(), + generation: 0, + length: len, + checksum: content_checksum(&payload), + payload: Some(payload.clone()), + lease_ttl_ms: 30_000, + }])?; + if !matches!(denied.items[0].body, Err(CacheError::QuotaExceeded { .. })) { + return Err(CacheError::Malformed { + reason: "fault demo quota", + }); + } + let looked = client.lookup(vec![LookupItem { + header: header(&client, 95)?, + identity: second, + }])?; + if !matches!( + looked.items[0].body, + Ok(LookupOutcome::Miss { + reason: MissReason::Absent + }) + ) { + return Err(CacheError::Malformed { + reason: "fault demo quota", + }); + } + fallback_matches(&[90, 91, 92, 93])?; + Ok("RECOMPUTE") +} diff --git a/crates/transport/src/lib.rs b/crates/transport/src/lib.rs index 69474e7..a7c1ce3 100644 --- a/crates/transport/src/lib.rs +++ b/crates/transport/src/lib.rs @@ -11,10 +11,12 @@ mod client; mod clock; mod demo; +mod fault_demo; mod frame; mod pool; mod server; pub use client::TcpClient; pub use demo::{run_share_demo, ShareReport}; +pub use fault_demo::{run_fault_demo, FaultReport}; pub use server::{serve, ServerConfig, ServerHandle}; diff --git a/crates/transport/tests/fault_tcp.rs b/crates/transport/tests/fault_tcp.rs new file mode 100644 index 0000000..27cbc85 --- /dev/null +++ b/crates/transport/tests/fault_tcp.rs @@ -0,0 +1,345 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 +//! TCP fault checks: disk full, short write, and eviction racing a reader. +//! Outcomes are exact bytes or a miss/error. Another block's bytes are a failure. + +use std::fs; +use std::path::PathBuf; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::thread; +use std::time::Duration; + +use gpucache_core_types::{ + content_checksum, synthetic_identity, CacheError, RequestId, ResourceLimits, +}; +use gpucache_protocol::{ + Attestation, AuthContext, CommitItem, GetItem, ItemHeader, LookupItem, LookupOutcome, + MissReason, ReleaseCancelItem, ReservePutItem, TenantPolicy, +}; +use gpucache_storage::{FileStore, PersistFault}; +use gpucache_transport::{run_fault_demo, serve, ServerConfig, TcpClient}; + +const FP: [u8; 32] = [9u8; 32]; + +struct TempDir { + path: PathBuf, +} + +impl TempDir { + fn new() -> Self { + static N: AtomicU64 = AtomicU64::new(0); + let path = std::env::temp_dir().join(format!( + "gpucache-fault-tcp-{}-{}", + std::process::id(), + N.fetch_add(1, Ordering::Relaxed) + )); + let _ = fs::remove_dir_all(&path); + Self { path } + } +} + +impl Drop for TempDir { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.path); + } +} + +fn open(path: &std::path::Path, limits: ResourceLimits) -> (FileStore, AuthContext) { + let id = synthetic_identity(1, 1); + let policy = TenantPolicy::new(id.tenant.clone(), id.trust_domain.clone(), FP).unwrap(); + let auth = AuthContext::new(id.tenant, id.trust_domain, FP).unwrap(); + let store = FileStore::open(path, limits, vec![policy]).unwrap(); + (store, auth) +} + +fn header(client: &TcpClient, n: u128) -> ItemHeader { + ItemHeader { + request_id: RequestId::from_counter(n).unwrap(), + epoch: client.epoch(), + deadline_ms: client.deadline(30_000), + } +} + +fn identity_mark(mark: u8) -> gpucache_core_types::CacheIdentity { + let mut id = synthetic_identity(1, 1); + id.prefix_chain[0].0[0] = mark; + id +} + +fn reserve_commit( + client: &mut TcpClient, + n: u128, + bytes: &[u8], + identity: gpucache_core_types::CacheIdentity, +) -> Result { + let reserved = client.reserve_put(vec![ReservePutItem { + header: header(client, n), + identity, + generation: 0, + length: bytes.len() as u64, + checksum: content_checksum(bytes), + payload: Some(bytes.to_vec()), + lease_ttl_ms: 30_000, + }])?; + let generation = reserved.items[0].body.as_ref().unwrap().generation; + let committed = client.commit(vec![CommitItem { + header: header(client, n + 1), + generation, + length: bytes.len() as u64, + checksum: content_checksum(bytes), + }])?; + committed.items[0] + .body + .as_ref() + .map(|_| generation) + .map_err(|err| err.clone()) +} + +#[test] +fn fault_demo_requires_exact_tokens_before_the_ok_banner() { + let report = run_fault_demo().unwrap(); + assert_eq!(report.hit_tokens, 4); + assert_eq!(report.miss, "MISS_ABSENT"); + assert_eq!(report.corrupt_fallback, "MISS_ABSENT"); + assert_eq!(report.timeout_fallback, "RECOMPUTE"); + assert_eq!(report.cancel_fallback, "MISS_QUARANTINED"); + assert_eq!(report.restart, "HIT_EXACT"); + assert_eq!(report.restart_down, "TRANSPORT_FALLBACK"); + assert_eq!(report.wrong_tenant, "UNAUTHORIZED"); + assert_eq!(report.wrong_model, "MISS_ABSENT"); + assert_eq!(report.wrong_tokenizer, "MISS_ABSENT"); + assert_eq!(report.wrong_adapter, "MISS_ABSENT"); + assert_eq!(report.layout_mismatch, "MISS_INCOMPATIBLE"); + assert_eq!(report.quota_fallback, "RECOMPUTE"); + assert_eq!(report.duplicate, "REPLAY_ONE_BLOCK"); + assert_eq!(report.reorder, "REJECTED"); + let text = report.to_string(); + assert!(text.starts_with("status=FALLBACK_OK")); + assert!(text.contains("criteria=EXACT_TOKEN_IDS")); + assert!(text.contains("tolerance=NONE")); + assert!(!text.contains("TODO")); + assert!(!text.contains("placeholder")); +} + +#[test] +fn disk_full_and_short_write_over_tcp_are_misses() { + let dir = TempDir::new(); + let (mut store, auth) = open(&dir.path, ResourceLimits::dev_defaults()); + store.inject_persist_fault(PersistFault::DiskFull, 1); + let server = serve(store, ServerConfig::default()).unwrap(); + let mut client = TcpClient::connect( + server.addr(), + auth.clone(), + Duration::from_secs(5), + 1024 * 1024, + ) + .unwrap(); + let payload = b"disk-full-block!!".to_vec(); + let identity = identity_mark(1); + let reserved = client + .reserve_put(vec![ReservePutItem { + header: header(&client, 1), + identity: identity.clone(), + generation: 0, + length: payload.len() as u64, + checksum: content_checksum(&payload), + payload: Some(payload.clone()), + lease_ttl_ms: 30_000, + }]) + .unwrap(); + let generation = reserved.items[0].body.as_ref().unwrap().generation; + let failed = client + .commit(vec![CommitItem { + header: header(&client, 2), + generation, + length: payload.len() as u64, + checksum: content_checksum(&payload), + }]) + .unwrap(); + assert!(matches!( + failed.items[0].body, + Err(CacheError::Storage { .. }) + )); + let looked = client + .lookup(vec![LookupItem { + header: header(&client, 3), + identity: identity.clone(), + }]) + .unwrap(); + assert!(matches!( + looked.items[0].body, + Ok(LookupOutcome::Miss { + reason: MissReason::Absent + }) + )); + drop(client); + drop(server); + + let (mut store, auth) = open(&dir.path, ResourceLimits::dev_defaults()); + store.inject_persist_fault(PersistFault::ShortWrite, 1); + let server = serve(store, ServerConfig::default()).unwrap(); + let mut client = + TcpClient::connect(server.addr(), auth, Duration::from_secs(5), 1024 * 1024).unwrap(); + let reserved = client + .reserve_put(vec![ReservePutItem { + header: header(&client, 10), + identity: identity.clone(), + generation: 0, + length: payload.len() as u64, + checksum: content_checksum(&payload), + payload: Some(payload.clone()), + lease_ttl_ms: 30_000, + }]) + .unwrap(); + let generation = reserved.items[0].body.as_ref().unwrap().generation; + let failed = client + .commit(vec![CommitItem { + header: header(&client, 11), + generation, + length: payload.len() as u64, + checksum: content_checksum(&payload), + }]) + .unwrap(); + assert!(matches!( + failed.items[0].body, + Err(CacheError::Storage { .. }) + )); + let looked = client + .lookup(vec![LookupItem { + header: header(&client, 12), + identity, + }]) + .unwrap(); + assert!(matches!( + looked.items[0].body, + Ok(LookupOutcome::Miss { .. }) + )); + let early = client + .commit(vec![CommitItem { + header: header(&client, 13), + generation: 99, + length: 4, + checksum: content_checksum(b"abcd"), + }]) + .unwrap(); + assert!(matches!( + early.items[0].body, + Err(CacheError::UnknownGeneration) + )); +} + +#[test] +fn concurrent_readers_never_observe_another_blocks_bytes() { + let dir = TempDir::new(); + let mut limits = ResourceLimits::dev_defaults(); + limits.max_object_bytes = 16; + limits.max_retained_bytes = 32; + limits.max_retained_bytes_per_tenant = 32; + limits.max_in_flight_bytes_per_tenant = 32; + let (store, auth) = open(&dir.path, limits); + let server = serve(store, ServerConfig::default()).unwrap(); + let addr = server.addr(); + let payload_a = vec![1u8; 16]; + let payload_b = vec![2u8; 16]; + let payload_c = vec![3u8; 16]; + let id_a = identity_mark(1); + let id_b = identity_mark(2); + let id_c = identity_mark(3); + { + let mut client = + TcpClient::connect(addr, auth.clone(), Duration::from_secs(5), 1024 * 1024).unwrap(); + reserve_commit(&mut client, 10, &payload_a, id_a.clone()).unwrap(); + } + let failures = Arc::new(Mutex::new(Vec::new())); + let mut handles = Vec::new(); + for idx in 0..2u128 { + let auth = auth.clone(); + let failures = Arc::clone(&failures); + let id_a = id_a.clone(); + let payload_a = payload_a.clone(); + handles.push(thread::spawn(move || { + let mut client = + TcpClient::connect(addr, auth, Duration::from_secs(5), 1024 * 1024).unwrap(); + for step in 0..8u128 { + let n = 100 + idx * 50 + step * 4; + let looked = client.lookup(vec![LookupItem { + header: header(&client, n), + identity: id_a.clone(), + }]); + let Ok(looked) = looked else { + continue; + }; + match &looked.items[0].body { + Ok(LookupOutcome::Hit { generation, .. }) => { + let got = client.get(vec![GetItem { + header: header(&client, n + 1), + generation: *generation, + identity: id_a.clone(), + pin_ttl_ms: 200, + }]); + let Ok(got) = got else { + continue; + }; + if let Ok(hit) = &got.items[0].body { + if hit.payload != payload_a { + failures.lock().unwrap().push(format!("reader {idx}")); + } + let _ = client.release_cancel(vec![ReleaseCancelItem { + header: header(&client, n + 2), + generation: *generation, + region_id: Some(hit.region_id), + pin_owner: Some(RequestId::from_counter(n + 1).unwrap()), + attestation: Attestation::Drained, + }]); + } + } + Ok(LookupOutcome::Miss { .. }) => {} + Ok(LookupOutcome::PrefixHit { .. }) => { + failures.lock().unwrap().push(format!("prefix {idx}")); + } + Err(_) => {} + } + } + })); + } + handles.push(thread::spawn({ + let auth = auth.clone(); + let id_b = id_b.clone(); + let id_c = id_c.clone(); + let payload_b = payload_b.clone(); + let payload_c = payload_c.clone(); + move || { + let mut client = + TcpClient::connect(addr, auth, Duration::from_secs(5), 1024 * 1024).unwrap(); + let _ = reserve_commit(&mut client, 400, &payload_b, id_b); + let _ = reserve_commit(&mut client, 410, &payload_c, id_c); + } + })); + for handle in handles { + handle.join().unwrap(); + } + assert!( + failures.lock().unwrap().is_empty(), + "cross-block bytes: {:?}", + failures.lock().unwrap() + ); + let mut client = TcpClient::connect(addr, auth, Duration::from_secs(5), 1024 * 1024).unwrap(); + let looked = client + .lookup(vec![LookupItem { + header: header(&client, 500), + identity: id_c, + }]) + .unwrap(); + if let Ok(LookupOutcome::Hit { generation, .. }) = &looked.items[0].body { + let got = client + .get(vec![GetItem { + header: header(&client, 501), + generation: *generation, + identity: identity_mark(3), + pin_ttl_ms: 100, + }]) + .unwrap(); + assert_eq!(got.items[0].body.as_ref().unwrap().payload, payload_c); + } +} diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index e4cf3b6..4721511 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -1,4 +1,4 @@ -# Architecture (V1 B) +# Architecture (V1 C) GPUCache is a shared KV data tier. The hot tier is G3.5. RustFS is an optional colder G4 capacity tier. It is not on the prefill-to-decode path. The inference engine still owns tokenization, attention, and scheduling. This repo does not replace them. @@ -21,11 +21,11 @@ gpucache-core-types (identity, keys, lifecycle, errors) gpucache-cold optional G4 pack/manifest. Not linked by transport. ``` -`MemoryStore` remains the in-process contract. `FileStore` is what the TCP service uses. Planned crates still absent: a standalone metadata service, an engine adapter, a DPU agent, a bench harness, and metrics. Their contracts stay out of `core-types` so GPU and DPU types do not leak in. +`MemoryStore` remains the in-process contract. `FileStore` is what the TCP service uses. `gpucache-bench` only names the paired baselines and prints `NOT_RUN_HARDWARE`. Planned crates still absent: a standalone metadata service, an engine adapter, a DPU agent, and metrics. Their contracts stay out of `core-types` so GPU and DPU types do not leak in. ## Durability -The hot tier does not repair lost blocks. There is no erasure coding and no replica set. Loss, corruption, eviction, and a failed check are clean misses. The engine recomputes and may write the block again. An extra copy of a hot shared prefix is a possible later latency optimization, not a durability feature, and it is not in this tree. Cold durability is RustFS's responsibility. See [adr/0002-no-erasure-coding.md](adr/0002-no-erasure-coding.md). +The hot tier does not repair lost blocks. There is no erasure coding and no replica set. Loss, corruption, a short write, a full disk, eviction, and a failed check are clean misses. The engine recomputes and may write the block again. If the directory write fails after the in-memory commit, that commit is abandoned and the request id is forgotten, so a replay cannot report success for bytes that were not stored. An extra copy of a hot shared prefix is a possible later latency optimization, not a durability feature, and it is not in this tree. Cold durability is RustFS's responsibility. See [adr/0002-no-erasure-coding.md](adr/0002-no-erasure-coding.md). ## Identity @@ -60,7 +60,7 @@ ABSENT -> RESERVED -> WRITING -> COMMITTED -> EVICTING -> ABSENT ## Process model -`MemoryStore` is single-threaded and in-process. The cross-process test exports a committed frame and runs `kv-block-xfer` in a child. The TCP service is a thread per connection with one mutex around `FileStore`. The read-buffer pool is bounded. Restart reloads committed files, bumps the epoch, and drops in-flight reservations. A corrupt file is skipped and is not a hit. +`MemoryStore` is single-threaded and in-process. The cross-process test exports a committed frame and runs `kv-block-xfer` in a child. The TCP service is a thread per connection with one mutex around `FileStore`. The read-buffer pool is bounded. Restart reloads committed files, bumps the epoch, and drops in-flight reservations. A corrupt or truncated file is skipped and is not a hit. A client that keeps talking to a stopped listener gets a transport error and must recompute; it does not keep a stale hit. ## Cold tier diff --git a/docs/BENCHMARKS.md b/docs/BENCHMARKS.md index 8ce2409..9794f86 100644 --- a/docs/BENCHMARKS.md +++ b/docs/BENCHMARKS.md @@ -1,14 +1,22 @@ # Benchmarks -No token-goodput run was executed for V1 A or V1 B. The TCP tests check reuse, quotas, checksums, and restart. They are not a bandwidth result. +No token-goodput run was executed for V1. The TCP tests check reuse, quotas, checksums, restart, and the fault matrix. They are not a bandwidth result. -| Comparison the V1 exit criteria ask for | Result | -| --- | --- | -| Engine-native prefix cache | `NOT_RUN_HARDWARE` | -| Same-box local offload | `NOT_RUN_HARDWARE` | -| GPUCache CPU/TCP | Correctness `PASS` via `cargo test` and `share-demo`. Token goodput `NOT_RUN_HARDWARE` (no engine, no GPU) | -| Cold / warm / no reuse / multi-turn / HBM cliff | `NOT_RUN_HARDWARE` | +Paired baselines use the same prefix-cache idea on both sides when a GPU run exists: the engine's own prefix cache, a same-box local offload, and GPUCache CPU/TCP. Write cost is part of that comparison. This tree does not have a GPU, an engine, or a trace, so every timing row is `NOT_RUN_HARDWARE`. The harness is `cargo run -p gpucache-bench --bin paired-baseline`. It prints no sample count, no TTFT, and no tokens/s. -Memcpy or a synthetic store bandwidth number is not token goodput and is not reported. There is no `FAIL_PERFORMANCE` figure because there is no paired measurement to fail. The CPU contract tests in `cargo test --workspace --all-targets` check correctness only. +| Comparison | Result | Notes | +| --- | --- | --- | +| Engine-native prefix cache | `NOT_RUN_HARDWARE` | Needs a pinned vLLM commit, an approved model, and a GPU. ADR 0001 is still Pending | +| Same-box local offload | `NOT_RUN_HARDWARE` | Same GPU, model, trace, and prefix-cache settings as the engine row. Not run | +| GPUCache CPU/TCP token goodput | `NOT_RUN_HARDWARE` | Correctness is `PASS` via `cargo test`, `share-demo`, and `fault-demo`. Goodput was not measured | +| Cold | `NOT_RUN_HARDWARE` | Empty cache, engine recomputes | +| Warm | `NOT_RUN_HARDWARE` | Same prefix hits the shared block | +| No prefix reuse | `NOT_RUN_HARDWARE` | Miss path. Callers recompute and may skip the shared cache | +| Agent multi-turn | `NOT_RUN_HARDWARE` | Not run | +| HBM within capacity | `NOT_RUN_HARDWARE` | Not run | +| HBM cliff | `NOT_RUN_HARDWARE` | Not run | +| Write cost | `NOT_RUN_HARDWARE` | Not recorded. A memcpy figure would not count | -Do not add a benchmark that prints a success banner without raw samples, a manifest, and the hardware it ran on. +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. + +Correctness criteria for the CPU path are in [CORRECTNESS.md](CORRECTNESS.md). diff --git a/docs/CORRECTNESS.md b/docs/CORRECTNESS.md new file mode 100644 index 0000000..0008671 --- /dev/null +++ b/docs/CORRECTNESS.md @@ -0,0 +1,39 @@ +# Correctness criteria (V1 C) + +GPUCache V1 C checks the CPU contract. It does not check a GPU kernel. + +## What is compared + +Synthetic blocks use `SKV1` version 1: a little-endian count and a list of `u32` token ids. The demo and the fault tests encode those ids, store the bytes, and decode them on every hit. + +The pass rule is exact equality of the token-id list and of the raw payload bytes. `tolerance=NONE`. There is no absolute or relative numeric slack, because this path has no floating-point kernel. + +On a miss, timeout, cancel, checksum failure, corrupt file, short write, full disk, quota refusal, or a stopped server, the caller recomputes with `deterministic_recompute`. That function is a pure encoding of the token ids the caller already has. It does not read another tenant's block. The cache result is not used. + +## What is never a hit + +A lookup or get is a failure of this milestone if it returns bytes for any of these: + +- a different tenant or trust domain +- a different model revision or model-config digest +- a different tokenizer revision +- a different adapter or LoRA revision +- a different layout ABI, dtype, quant, shard, or RoPE +- a block that is not `COMMITTED`, including reserved, writing, evicting, and quarantined +- a checksum mismatch, a truncated file, or a byte flipped on disk +- a commit the service reported as a failed durable write + +Model, tokenizer, and adapter changes are different logical keys. The lookup reason is `MISS_ABSENT`. A layout change keeps the logical key and is `MISS_INCOMPATIBLE`. `Get` of the stored generation with the wrong identity is `Incompatible` and returns no payload. A wrong tenant is `Unauthorized` and returns no payload. + +## Limits + +- These token ids are not vLLM output. A real export, import, and engine-output comparison is `NOT_RUN_HARDWARE` until it runs on a GPU. ADR 0001 stays Pending. +- Non-deterministic GPU kernels are not given a tolerance in this tree. When a GPU run exists, that run must publish its own criterion. It must not loosen the rule that a wrong tenant or a wrong model is a miss. +- `deterministic_recompute` does not approximate attention. It only proves the cache did not substitute a different token list. +- Protocol fuzz is `proptest` over the decoders inside `cargo test`. It is not a multi-hour libFuzzer campaign. Decoders must not panic. A frame that decodes must round-trip. Truncation, duplication, and a rotated frame must fail closed. +- Filling the host disk is `NOT_RUN`. Disk-full behavior is the `PersistFault::DiskFull` path plus the `ENOSPC` / `ErrorKind::StorageFull` classifier. A chmod-unwritable directory covers a real failed write on Unix. +- aarch64 `cargo test` is a CPU build. It is not a BlueField test. + +## Workloads that should bypass the shared cache + +A request with no prefix overlap is a miss. The engine recomputes. Callers can skip the shared cache for that traffic. The CPU tests show the miss. They do not show a token-goodput win. See [BENCHMARKS.md](BENCHMARKS.md). diff --git a/docs/OPERATIONS.md b/docs/OPERATIONS.md index d0b12d5..11883dc 100644 --- a/docs/OPERATIONS.md +++ b/docs/OPERATIONS.md @@ -1,4 +1,4 @@ -# Operations (V1 B) +# Operations (V1 C) V1 B is a library, an import helper, and a localhost TCP service. There is no package and no production config. The service creates files only under the directory you pass. It does not format NVMe, change a NIC, or touch a DPU. @@ -28,6 +28,22 @@ cargo run -p gpucache-transport --bin share-demo This starts a loopback server on an ephemeral port, writes one synthetic block from one client, and reads it from a second client. Success is `status=REUSE_OK`. `partial_prefix=WHOLE_BLOCK_PREFIX` means the longer chain reused the whole shorter block. `intra_block_slice=DEGRADED_MISS` means a shorter `valid_length` was not sliced out of that block. +## Hit, miss, and fallback + +```bash +cargo run -p gpucache-transport --bin fault-demo +``` + +The command uses a fresh temp directory. It commits a synthetic token block, reads it back from a second client, and checks the token ids exactly. An unknown key, a corrupt file after restart, a past deadline, a cancel, and a quota refusal all fail closed. The caller recomputes with the same pure function of those token ids. `status=FALLBACK_OK` is printed only after those checks. `tolerance=NONE` means no numeric slack was applied. A wrong tenant, model, tokenizer, adapter, or layout is not a hit. + +The paired-baseline stub does not measure anything: + +```bash +cargo run -p gpucache-bench --bin paired-baseline +``` + +`status=NOT_RUN_HARDWARE` is the honest result until a GPU run exists. See [BENCHMARKS.md](BENCHMARKS.md) and [CORRECTNESS.md](CORRECTNESS.md). + ## TCP service ```bash diff --git a/docs/PROTOCOL.md b/docs/PROTOCOL.md index 76f8762..4da46a2 100644 --- a/docs/PROTOCOL.md +++ b/docs/PROTOCOL.md @@ -52,6 +52,8 @@ Request deadlines abort the call. Writer leases move `RESERVED` with no buffer b A hit requires the physical key and a `COMMITTED` block. Logical-key overlap with a different layout, dtype, quant, or shard is a miss. +A commit that arrives before its reserve is `UnknownGeneration` and returns no bytes. Two items in one batch that share a `request_id` are rejected before either runs. Replaying a successful commit with the same canonical arguments returns the stored result and does not create a second block. The same id with different arguments is `IdempotencyConflict`. Decoders reject truncated frames, a second frame glued on the end, and a frame whose halves were swapped. A declared length above the hard cap is rejected before the buffer is allocated. + `PrefixHit` is a whole stored block whose hash chain is a strict prefix of the query and whose layout-affecting fields match. `matched_prefix_blocks` is the stored chain length. `Get` uses `identity_for_prefix_hit`. The same chain with a different `valid_length` is an incompatible miss: the service does not return a sliced payload. See [adr/0003-partial-prefix.md](adr/0003-partial-prefix.md). ## Replies diff --git a/docs/REUSE_AUDIT.md b/docs/REUSE_AUDIT.md index e5d864e..a78ae58 100644 --- a/docs/REUSE_AUDIT.md +++ b/docs/REUSE_AUDIT.md @@ -10,4 +10,4 @@ The execution spec asks for a look at LMCache, Mooncake, Dynamo KVBM, NIXL, Open | NIXL | Candidate V2 transport. No Rust binding is vendored. Ordinary NIC RDMA is not a DPU result. | | OpenLake, ContextStore, Fluxon | Not required for identity, protocol, or `MemoryStore`. No version is pinned because no code is imported. | -This is a scope decision, not a line-by-line license audit of those upstream trees. Nothing from them was copied. Direct dependencies are only `sha2` (runtime) and `proptest` (tests). See [THIRD_PARTY_NOTICES.md](../THIRD_PARTY_NOTICES.md). +This is a scope decision, not a line-by-line license audit of those upstream trees. Nothing from them was copied. Direct dependencies are only `sha2` (runtime) and `proptest` (tests). V1 C did not add a crate. See [THIRD_PARTY_NOTICES.md](../THIRD_PARTY_NOTICES.md). diff --git a/docs/RISKS.md b/docs/RISKS.md index 3b20bbd..4173e51 100644 --- a/docs/RISKS.md +++ b/docs/RISKS.md @@ -1,4 +1,4 @@ -# Risks (V1 B) +# Risks (V1 C) | Risk | Status | Note | | --- | --- | --- | @@ -17,5 +17,9 @@ | License of a future DOCA or model stack | `NOT_RUN_LICENSE` | No DOCA binary and no model weights are in the tree. | | GPU rental or DPU purchase | `NOT_RUN_BUDGET` | No paid API, GPU, or DPU was used. | | Hot block loss | Accepted | No erasure coding and no durability replicas. The outcome is a clean miss, then recompute and repopulate. Cold durability is RustFS G4. See ADR 0002. | +| Short write or disk full reported as a durable hit | Mitigated | The in-memory commit is abandoned and the request id is forgotten. See `docs/V1C_TEST_MATRIX.md` | +| Hostile protocol bytes | Mitigated by decoder fuzz | `proptest` inside `cargo test`. Not a multi-hour libFuzzer run | +| Wrong model or tokenizer hit | Mitigated | Those identities miss. Comparison of synthetic tokens is exact. `tolerance=NONE` | +| Invented baseline numbers | Closed in this tree | `paired-baseline` prints `NOT_RUN_HARDWARE` and contains no digits | Minimum reproduction of the CPU contract is `cargo test --workspace --all-targets`. There is no hardware failure log because the hardware was not present. diff --git a/docs/SECURITY.md b/docs/SECURITY.md index 5bcb5f3..f720bb8 100644 --- a/docs/SECURITY.md +++ b/docs/SECURITY.md @@ -1,4 +1,4 @@ -# Security (V1 B) +# Security (V1 C) The library calls still require an `AuthContext`. The TCP service does too. It binds `127.0.0.1` unless the operator sets `--allow-remote`, and it refuses to start with an empty tenant policy. A data request with an unknown tenant, a mismatched trust domain, or a wrong fingerprint is `Unauthorized` and returns no payload. Hello carries versions and the store epoch only. @@ -25,4 +25,4 @@ The fingerprint is a hash of a credential, never the credential. Do not pass a r ## Fallback -Callers must treat every error and every miss as "recompute in the engine". They must not stitch a partial or quarantined block into a decode. +Callers must treat every error and every miss as "recompute in the engine". They must not stitch a partial or quarantined block into a decode. V1 C checks that a wrong tenant, a wrong model, a corrupt file, a short write, and a failed disk write return no foreign payload. The synthetic check is exact token-id equality. See [CORRECTNESS.md](CORRECTNESS.md). diff --git a/docs/V1C_TEST_MATRIX.md b/docs/V1C_TEST_MATRIX.md new file mode 100644 index 0000000..fc96eea --- /dev/null +++ b/docs/V1C_TEST_MATRIX.md @@ -0,0 +1,50 @@ +# V1 C test matrix + +Recorded for the CPU contract. Synthetic token ids are compared exactly (`tolerance=NONE`). This is not a GPU, DPU, or token-goodput report. aarch64 in CI is a CPU build, not a BlueField test. + +| Check | Result | Where | +| --- | --- | --- | +| Unit and property tests, including a flipped model byte | PASS | `cargo test --workspace --all-targets`; `flipped_model_byte_is_never_a_hit` | +| Protocol fuzz: random bytes, truncation, duplicate frame, reordered frame, corrupt export | PASS | `crates/protocol/src/fuzz.rs` | +| Short write abandoned; replay is not success | PASS | `short_write_is_not_durable_and_replay_is_not_success` | +| Truncated `.gblk` is a miss after restart | PASS | `truncated_block_file_is_a_miss_after_restart` | +| Corruption (flipped payload byte) is a miss | PASS | `corrupt_file_is_not_a_hit_after_restart`, `fault-demo` `corrupt_fallback=RECOMPUTE` | +| Duplicate commit replays one block; conflicting id is rejected | PASS | `duplicate_commit_replays_one_block_and_conflict_hides_bytes`, `fault-demo` `duplicate=REPLAY_ONE_BLOCK` | +| Reordered commit-before-reserve and duplicate ids in one batch | PASS | `commit_before_reserve_and_duplicate_batch_ids_fail_closed`, `fault-demo` `reorder=REJECTED` | +| Timeout returns no payload | PASS | `timeout_and_cancel_do_not_return_payload`, `fault-demo` `timeout_fallback=RECOMPUTE` | +| Cancel quarantines and hides bytes | PASS | same tests, `cancel_fallback=MISS_QUARANTINED` | +| Server restart reloads exact tokens; dead listener is a transport fallback | PASS | `fault-demo` `restart=HIT_EXACT`, `restart_down=TRANSPORT_FALLBACK` | +| Disk full (injected) is a miss and releases retained bytes | PASS | `disk_full_abandon_leaves_a_miss`, `disk_full_and_short_write_over_tcp_are_misses` | +| `ENOSPC` / `ErrorKind::StorageFull` classified as disk full | PASS | `storage_full_and_enospc_map_to_disk_full` | +| Host volume actually filled | NOT_RUN | Not done. The injector covers the failure path | +| Unwritable blocks directory (Unix) does not leave a hit | PASS | `unwritable_blocks_directory_does_not_leave_a_hit` | +| Eviction versus pinned read keeps exact tokens | PASS | `pinned_read_keeps_exact_tokens_while_eviction_is_refused` | +| Concurrent TCP readers never observe another block | PASS | `concurrent_readers_never_observe_another_blocks_bytes` | +| Tenant key collision | PASS | `tenant_key_collision_never_returns_the_other_payload`, `fault-demo` `wrong_tenant=UNAUTHORIZED` | +| Model, tokenizer, adapter change | PASS | `model_tokenizer_adapter_and_layout_changes_are_not_hits`, `fault-demo` `MISS_ABSENT` | +| Layout mismatch | PASS | same test, `layout_mismatch=MISS_INCOMPATIBLE` | +| Memory quota exhaustion (no silent eviction in `MemoryStore`) | PASS | `memory_quota_exhaustion_does_not_evict_or_return_foreign_bytes` | +| Pinned file quota fallback | PASS | `fault-demo` `quota_fallback=RECOMPUTE` | +| Exact synthetic token check | PASS | `fault-demo` `criteria=EXACT_TOKEN_IDS`, `tolerance=NONE` | +| Real vLLM output comparison | NOT_RUN_HARDWARE | No GPU. ADR 0001 stays Pending. See [CORRECTNESS.md](CORRECTNESS.md) | +| Engine-native prefix cache, local offload, GPUCache goodput, cold/warm/no-reuse/multi-turn/HBM | NOT_RUN_HARDWARE | `cargo run -p gpucache-bench --bin paired-baseline` | +| RustFS live endpoint | NOT_RUN_HARDWARE | Unchanged from V1 B. Mock still PASS | +| DOCA / model license | NOT_RUN_LICENSE | No DOCA binary and no model weights | +| Paid GPU or DPU | NOT_RUN_BUDGET | No rental and no purchase | +| aarch64 BlueField functional test | NOT_RUN_HARDWARE | An aarch64 build is not a BlueField test | +| x86_64 CPU suite | PASS | `cargo test --workspace --all-targets` on this host (Linux 6.12.94+ x86_64, rustc 1.99.0) | +| aarch64 CPU suite on this x86_64 host | NOT_RUN | `qemu-aarch64-static` and `aarch64-linux-gnu-gcc` are not installed here. CI jobs `test-aarch64` and `test-aarch64-qemu` still run the same `cargo test`. This is not a BlueField test | + +One-command CPU repro: + +```bash +cargo test --workspace --all-targets +``` + +Fallback demo: + +```bash +cargo run -p gpucache-transport --bin fault-demo +``` + +No erasure coding was added. V2 and V3 were not started. diff --git a/docs/adr/0001-integration-route.md b/docs/adr/0001-integration-route.md index 29844d0..71d3aaa 100644 --- a/docs/adr/0001-integration-route.md +++ b/docs/adr/0001-integration-route.md @@ -29,6 +29,10 @@ When a GPU is authorized, the spike should: Prefer the hook that already exists in the pinned revision. Switch only if that spike fails. Do not start the second path in the same change. +## V1 C + +The fault-matrix work did not run the GPU spike. This ADR stays **Pending**. No engine adapter was added. + ## Consequences - Callers must not assume GPUCache can read an arbitrary engine's KV bytes. diff --git a/docs/adr/0002-no-erasure-coding.md b/docs/adr/0002-no-erasure-coding.md index b1c2582..b8fad61 100644 --- a/docs/adr/0002-no-erasure-coding.md +++ b/docs/adr/0002-no-erasure-coding.md @@ -23,6 +23,6 @@ DOCA EC, SPDK RAID targets, and similar offloads are not GPUCache components. Th ## Consequences - The README and roadmap do not list an EC milestone. -- The V1 crates have no EC types, no parity buffers, and no replica placement. The V1 B file backend deletes or skips a bad file; it does not rebuild one. +- The V1 crates have no EC types, no parity buffers, and no replica placement. The file backend deletes or skips a bad file; it does not rebuild one. V1 C did not add parity. A short write or a full disk is a miss. - RustFS remains responsible for any durable cold object it is asked to store. The V1 B cold client is off the hot path and is a contract mock unless an authorized endpoint is configured. - A cache miss after a node loss is correct behavior. The KV block is not the user's only copy.