From 1c9279cf7d12b0a4463d69c7c498c546185f06df Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 15 Oct 2026 06:03:03 +0000 Subject: [PATCH] V1 C: fault matrix, exact synthetic tokens, and baseline stubs Add protocol fuzz, fault injection, and concurrency tests for short writes, corruption, duplicates, reordering, timeout, cancel, restart, disk full, eviction races, tenant collisions, identity changes, and quota exhaustion. A failed durable write is not a hit. fault-demo checks hit, miss, and recompute with exact token ids. GPU baselines stay NOT_RUN_HARDWARE. Co-authored-by: RustFS --- .github/workflows/ci.yml | 12 +- AGENTS.md | 3 +- COMPATIBILITY.md | 6 +- Cargo.lock | 4 + Cargo.toml | 1 + README.md | 25 +- THIRD_PARTY_NOTICES.md | 2 +- crates/bench/Cargo.toml | 16 + crates/bench/src/bin/paired_baseline.rs | 14 + crates/bench/src/lib.rs | 60 ++ crates/protocol/src/fuzz.rs | 110 ++++ crates/protocol/src/lib.rs | 3 + crates/storage/src/file.rs | 135 +++- crates/storage/src/lib.rs | 4 +- crates/storage/src/memory.rs | 18 + crates/storage/src/synthetic.rs | 103 +++ crates/storage/tests/fault_matrix.rs | 825 ++++++++++++++++++++++++ crates/transport/Cargo.toml | 4 + crates/transport/src/bin/fault_demo.rs | 24 + crates/transport/src/fault_demo.rs | 612 ++++++++++++++++++ crates/transport/src/lib.rs | 2 + crates/transport/tests/fault_tcp.rs | 345 ++++++++++ docs/ARCHITECTURE.md | 8 +- docs/BENCHMARKS.md | 26 +- docs/CORRECTNESS.md | 39 ++ docs/OPERATIONS.md | 18 +- docs/PROTOCOL.md | 2 + docs/REUSE_AUDIT.md | 2 +- docs/RISKS.md | 6 +- docs/SECURITY.md | 4 +- docs/V1C_TEST_MATRIX.md | 50 ++ docs/adr/0001-integration-route.md | 4 + docs/adr/0002-no-erasure-coding.md | 2 +- 33 files changed, 2445 insertions(+), 44 deletions(-) create mode 100644 crates/bench/Cargo.toml create mode 100644 crates/bench/src/bin/paired_baseline.rs create mode 100644 crates/bench/src/lib.rs create mode 100644 crates/protocol/src/fuzz.rs create mode 100644 crates/storage/src/synthetic.rs create mode 100644 crates/storage/tests/fault_matrix.rs create mode 100644 crates/transport/src/bin/fault_demo.rs create mode 100644 crates/transport/src/fault_demo.rs create mode 100644 crates/transport/tests/fault_tcp.rs create mode 100644 docs/CORRECTNESS.md create mode 100644 docs/V1C_TEST_MATRIX.md 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.