From c04256839a9e6e74cc24c0a4dd6c106b670edab4 Mon Sep 17 00:00:00 2001 From: RustFS <155562731+loverustfs@users.noreply.github.com> Date: Tue, 20 Oct 2026 23:17:12 +0800 Subject: [PATCH 1/4] Add honest CPU engine and DPU simulation evidence gates --- README.md | 8 +- crates/dpu/src/demo.rs | 62 ++++++- crates/dpu/src/lib.rs | 2 + crates/dpu/src/message.rs | 182 +++++++++++++++++++ crates/storage/Cargo.toml | 4 + crates/storage/src/bin/engine_spike.rs | 29 +++ crates/storage/src/engine_spike.rs | 131 +++++++++++++ crates/storage/src/lib.rs | 2 + crates/transport/src/demo.rs | 8 + docs/CPU_COMPLETION.md | 56 ++++++ docs/OPERATIONS.md | 36 ++++ docs/VM_REGRESSION.md | 6 +- docs/adr/0001-integration-route.md | 4 + docs/adr/0013-cpu-spikes-and-evidence.md | 42 +++++ scripts/ci/cpu_simulation_pair.py | 68 +++++++ scripts/ci/tests/test_cpu_simulation_pair.py | 44 +++++ scripts/ci/vm_regression.py | 7 + 17 files changed, 684 insertions(+), 7 deletions(-) create mode 100644 crates/dpu/src/message.rs create mode 100644 crates/storage/src/bin/engine_spike.rs create mode 100644 crates/storage/src/engine_spike.rs create mode 100644 docs/CPU_COMPLETION.md create mode 100644 docs/adr/0013-cpu-spikes-and-evidence.md create mode 100644 scripts/ci/cpu_simulation_pair.py create mode 100644 scripts/ci/tests/test_cpu_simulation_pair.py diff --git a/README.md b/README.md index 9480f83..ed61ffa 100644 --- a/README.md +++ b/README.md @@ -27,8 +27,8 @@ V1 runs the cache contract on CPU. V2 adds a transport trait, a simulated regist | V2: paired TTFT/TPOT against native NIXL | `NOT_RUN_HARDWARE` | | V3: `dpu-bf3` feature slot, simulated DRAM slab + storage queue, SPDK affinity check | In progress. Devices are not linked. `open_doca` / `open_nvme_device` / `open_spdk` refuse. `dpu-demo` prints `NOT_RUN_HARDWARE` | | V3: one candidate offload (storage queue) and ablation stubs | Interface only. P2P, DPA, compression, and crypto offload stay `NOT_RUN_HARDWARE` when enabled. Hot-tier EC cannot be enabled | -| V3: authz, replay window, integrity seal, rate limit, secret rotation, audit | CPU hooks in this tree. Not mTLS. Trusted-fabric boundary is [docs/adr/0007-trusted-fabric.md](docs/adr/0007-trusted-fabric.md) | -| V3: rebalance, rolling upgrade, fault-domain isolation | In-process stubs on the V2 shard directory. They fence fills. They do not copy a replica | +| V3: authz, replay window, integrity seal, rate limit, secret rotation, audit | ACL and rate admission enforced on every TCP data operation; bounded CPU replay/rotation hooks remain separate. Not mTLS. Trusted-fabric boundary is [docs/adr/0007-trusted-fabric.md](docs/adr/0007-trusted-fabric.md) | +| V3: rebalance, rolling upgrade, fault-domain isolation | Atomic bounded membership updates and a bounded TCP shard client, with epoch fencing and restart tests. No membership consensus or durability replicas | | V3: degrade matrix (DPU restart, GPU reset, NIC flap, storage loss, NVMe timeout, control-plane loss, RustFS slow/down) | Simulated fault decisions request recompute. Unknown completions remain quarantined and charged until explicit drain. Tenant-scoped invalidation/reclaim cannot free referenced slots. Recovery is not proven on hardware | | V3: G4 retention, manifest version, reclaim, delete verification, tenant cold keys, backfill budget | Implemented against the mock. Live RustFS is `NOT_RUN_HARDWARE` | | V3: paired NIC-RDMA baseline vs DPU | `NOT_RUN_HARDWARE`. `tco=TCO_UNKNOWN`. `go_gate=NOT_EVALUATED`. TCP is not the DPU baseline | @@ -36,7 +36,7 @@ V1 runs the cache contract on CPU. V2 adds a transport trait, a simulated regist | Mooncake-informed CPU contracts | Implemented in this tree. See [docs/MOONCAKE.md](docs/MOONCAKE.md) and [docs/adr/0009-mooncake-lessons.md](docs/adr/0009-mooncake-lessons.md). Mooncake is not linked. No NIC, Soft-RoCE, or DPU number is claimed | | Erasure coding and durability replicas | Will not be built. A lost or failed hot block is a clean miss, then recompute and repopulate. Cold durability is RustFS G4 | -The integration route (native vLLM KV connector or LMCache) is **not chosen**. See [docs/adr/0001-integration-route.md](docs/adr/0001-integration-route.md). Hot-tier failure policy is [docs/adr/0002-no-erasure-coding.md](docs/adr/0002-no-erasure-coding.md). The V3 offload candidate is [docs/adr/0006-storage-queue-offload.md](docs/adr/0006-storage-queue-offload.md). It is a hypothesis pending a paired measurement, not a measured bottleneck. +The integration route (native vLLM KV connector or LMCache) is **not chosen**. `engine-spike` exercises an independently specified ABI against explicitly fake blocks; it is not engine integration. The CPU completion gap map and remaining boundaries are in [docs/CPU_COMPLETION.md](docs/CPU_COMPLETION.md). See [docs/adr/0001-integration-route.md](docs/adr/0001-integration-route.md). Hot-tier failure policy is [docs/adr/0002-no-erasure-coding.md](docs/adr/0002-no-erasure-coding.md). The V3 offload candidate is [docs/adr/0006-storage-queue-offload.md](docs/adr/0006-storage-queue-offload.md). It is a hypothesis pending a paired measurement, not a measured bottleneck. ## What V1 A does @@ -68,7 +68,7 @@ ABSENT -> RESERVED -> WRITING -> COMMITTED -> EVICTING -> ABSENT A failed or timed-out registered region goes to `QUARANTINED`. Lease expiry does not prove that a copy has stopped, so that buffer stays allocated until the owner attests drain or a newer store epoch fences it. Same tokens do not imply the same bytes: the logical prefix key and the physical byte key are different encodings. V1 reuses only within one tenant, one model, and one layout. A reshard, quant, or layout mismatch is a miss. There is no cross-tenant dedup. A longer hash chain can reuse a whole shorter committed block (`PrefixHit`); the service does not slice inside a block. See [docs/adr/0003-partial-prefix.md](docs/adr/0003-partial-prefix.md). -Resource limits cover object size, writer admissions, memory registrations, per-tenant in-flight bytes, and retained payload bytes. They are not a complete process-RSS bound: reader pins, protocol reply allocations, and some auxiliary metadata still need admission limits. The idempotency table is bounded and scoped by authenticated tenant and trust domain. +Resource limits cover object size, writer admissions, reader pins, aggregate Get reply payloads, memory registrations, per-tenant in-flight bytes, and retained payload bytes. Reader pins retain credit after expiry until explicit drain/fence. Routed clients bound connections and retry records; directory and cold metadata have hard ceilings. These are admission bounds, not a measured process-RSS guarantee. The idempotency table is bounded and scoped by authenticated tenant and trust domain. ## What you can run diff --git a/crates/dpu/src/demo.rs b/crates/dpu/src/demo.rs index b03c0fd..f0b861c 100644 --- a/crates/dpu/src/demo.rs +++ b/crates/dpu/src/demo.rs @@ -8,22 +8,28 @@ use core::fmt; -use gpucache_core_types::CacheError; +use gpucache_core_types::{content_checksum, CacheError}; use crate::ablation::{Ablation, Knob}; use crate::queue::{ dpu_feature_selected, open_doca, open_nvme_device, open_spdk, ArmAgent, StorageQueue, }; use crate::target::planned_targets; +use crate::{Fault, HostSimulation, SimulationMessage, SimulationReply}; #[derive(Clone, Debug, PartialEq, Eq)] pub struct DpuReport { pub feature: &'static str, + pub payload_sha256: String, } impl fmt::Display for DpuReport { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { writeln!(f, "status=DPU_CONTRACT_OK")?; + writeln!(f, "payload_sha256={}", self.payload_sha256)?; + writeln!(f, "message_path=HOST_SIMULATION")?; + writeln!(f, "fault_path=QUARANTINE_DRAIN_PASS")?; + writeln!(f, "performance=NOT_A_PERFORMANCE_RESULT")?; writeln!(f, "arm_process=STARTED")?; writeln!(f, "offload_active=false")?; writeln!(f, "offload=STORAGE_QUEUE")?; @@ -93,7 +99,59 @@ pub fn run_dpu_demo() -> Result { } else { "UNSELECTED" }; - Ok(DpuReport { feature }) + // Exercise typed host messages and a lost-completion failure path too. + let mut host = HostSimulation::new(64, 2)?; + let payload = b"shared-kv-block"; + let SimulationReply::Submitted(write) = host.handle(SimulationMessage::Write { + tenant: b"tenant-a", + payload, + })? + else { + return Err(CacheError::Incompatible); + }; + let SimulationReply::Completed(done) = host.handle(SimulationMessage::Complete(write))? else { + return Err(CacheError::Incompatible); + }; + let slot = done.slot.ok_or(CacheError::UnknownGeneration)?; + let SimulationReply::Submitted(read) = host.handle(SimulationMessage::Read { + tenant: b"tenant-a", + slot, + })? + else { + return Err(CacheError::Incompatible); + }; + let SimulationReply::Completed(got) = host.handle(SimulationMessage::Complete(read))? else { + return Err(CacheError::Incompatible); + }; + if got.payload.as_deref() != Some(payload.as_slice()) { + return Err(CacheError::ChecksumMismatch); + } + let SimulationReply::Submitted(pending) = host.handle(SimulationMessage::Write { + tenant: b"tenant-a", + payload, + })? + else { + return Err(CacheError::Incompatible); + }; + host.handle(SimulationMessage::Fault(Fault::NicFlap))?; + if host.handle(SimulationMessage::Complete(pending)).is_ok() + || host.stats().quarantined_tasks != 1 + { + return Err(CacheError::Incompatible); + } + host.handle(SimulationMessage::Drain(pending))?; + if host.stats().drain_attestations != 1 || host.stats().quarantined_tasks != 0 { + return Err(CacheError::Incompatible); + } + let payload_sha256 = content_checksum(payload) + .0 + .iter() + .map(|b| format!("{b:02x}")) + .collect(); + Ok(DpuReport { + feature, + payload_sha256, + }) } #[cfg(test)] diff --git a/crates/dpu/src/lib.rs b/crates/dpu/src/lib.rs index 6acc48a..0614503 100644 --- a/crates/dpu/src/lib.rs +++ b/crates/dpu/src/lib.rs @@ -12,6 +12,7 @@ mod ablation; mod degrade; mod demo; +mod message; mod queue; mod reactor; mod target; @@ -19,6 +20,7 @@ mod target; pub use ablation::{Ablation, Knob}; pub use degrade::{all_faults, decide, DegradeDecision, Fault}; pub use demo::{run_dpu_demo, DpuReport}; +pub use message::{HostSimulation, SimulationMessage, SimulationReply, SimulationStats}; pub use queue::{ dpu_feature_selected, dpu_probe, open_doca, open_nvme_device, open_spdk, ArmAgent, Completion, StorageQueue, TaskId, diff --git a/crates/dpu/src/message.rs b/crates/dpu/src/message.rs new file mode 100644 index 0000000..3dc95e0 --- /dev/null +++ b/crates/dpu/src/message.rs @@ -0,0 +1,182 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 +//! Synchronous host-side messages. No IPC, DMA, DOCA, or device ABI is claimed. +//! The caller is the trusted queue owner. There is no extra unbounded inbox. + +use gpucache_core_types::CacheError; + +use crate::{Completion, DegradeDecision, Fault, StorageQueue, TaskId}; + +pub enum SimulationMessage<'a> { + Write { tenant: &'a [u8], payload: &'a [u8] }, + Read { tenant: &'a [u8], slot: u64 }, + Complete(TaskId), + CancelUnknown(TaskId), + Drain(TaskId), + Fault(Fault), + RestoreControl, +} + +#[derive(Debug)] +pub enum SimulationReply { + Submitted(TaskId), + Completed(Completion), + Acknowledged, + Degraded(DegradeDecision), +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct SimulationStats { + pub messages: u64, + pub quarantined_tasks: usize, + pub retained_bytes: u64, + pub drain_attestations: u64, + pub fault_events: u64, +} + +pub struct HostSimulation { + queue: StorageQueue, + messages: u64, + drains: u64, + faults: u64, +} + +impl HostSimulation { + pub fn new(cap_bytes: u64, max_tasks: usize) -> Result { + Ok(Self { + queue: StorageQueue::open_simulated(cap_bytes, max_tasks)?, + messages: 0, + drains: 0, + faults: 0, + }) + } + + pub fn stats(&self) -> SimulationStats { + SimulationStats { + messages: self.messages, + quarantined_tasks: self.queue.quarantined(), + retained_bytes: self.queue.retained_bytes(), + drain_attestations: self.drains, + fault_events: self.faults, + } + } + + pub fn handle( + &mut self, + message: SimulationMessage<'_>, + ) -> Result { + self.messages = self.messages.saturating_add(1); + Ok(match message { + SimulationMessage::Write { tenant, payload } => { + SimulationReply::Submitted(self.queue.submit_write(tenant, payload)?) + } + SimulationMessage::Read { tenant, slot } => { + SimulationReply::Submitted(self.queue.submit_read(tenant, slot)?) + } + SimulationMessage::Complete(task) => { + SimulationReply::Completed(self.queue.complete(task)?) + } + SimulationMessage::CancelUnknown(task) => { + self.queue.quarantine(task)?; + SimulationReply::Acknowledged + } + SimulationMessage::Drain(task) => { + self.queue.attest_drained(task)?; + self.drains = self.drains.saturating_add(1); + SimulationReply::Acknowledged + } + SimulationMessage::Fault(fault) => { + let decision = self.queue.apply_fault(fault); + self.faults = self.faults.saturating_add(1); + SimulationReply::Degraded(decision) + } + SimulationMessage::RestoreControl => { + self.queue.restore_control(); + SimulationReply::Acknowledged + } + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn cancel_storm_keeps_unknown_completion_credit_until_drain() { + let mut host = HostSimulation::new(8, 1).unwrap(); + let SimulationReply::Submitted(task) = host + .handle(SimulationMessage::Write { + tenant: b"a", + payload: b"12345678", + }) + .unwrap() + else { + panic!("submit reply") + }; + for _ in 0..1024 { + host.handle(SimulationMessage::CancelUnknown(task)).unwrap(); + assert!(host.handle(SimulationMessage::Complete(task)).is_err()); + assert!(host + .handle(SimulationMessage::Write { + tenant: b"a", + payload: b"x" + }) + .is_err()); + assert_eq!(host.stats().quarantined_tasks, 1); + assert_eq!(host.stats().retained_bytes, 8); + } + host.handle(SimulationMessage::Drain(task)).unwrap(); + assert_eq!(host.stats().retained_bytes, 0); + assert_eq!(host.stats().drain_attestations, 1); + assert!(host.handle(SimulationMessage::Drain(task)).is_err()); + } + + #[test] + fn every_injected_fault_is_a_payload_free_recompute_decision() { + for fault in crate::all_faults() { + let mut host = HostSimulation::new(8, 2).unwrap(); + let SimulationReply::Degraded(decision) = + host.handle(SimulationMessage::Fault(*fault)).unwrap() + else { + panic!("fault reply") + }; + assert!(decision.recompute); + assert!(!decision.wrong_hit); + assert_eq!(host.stats().fault_events, 1); + } + } + + #[test] + fn control_recovery_does_not_drain_quarantined_work() { + let mut host = HostSimulation::new(8, 1).unwrap(); + let SimulationReply::Submitted(task) = host + .handle(SimulationMessage::Write { + tenant: b"a", + payload: b"12345678", + }) + .unwrap() + else { + panic!("submit reply") + }; + host.handle(SimulationMessage::Fault(Fault::ControlPlaneLoss)) + .unwrap(); + assert!(host + .handle(SimulationMessage::Write { + tenant: b"a", + payload: b"x" + }) + .is_err()); + host.handle(SimulationMessage::RestoreControl).unwrap(); + assert_eq!(host.stats().retained_bytes, 8); + assert_eq!(host.stats().quarantined_tasks, 1); + assert!(host.handle(SimulationMessage::Complete(task)).is_err()); + host.handle(SimulationMessage::Drain(task)).unwrap(); + assert!(host + .handle(SimulationMessage::Write { + tenant: b"a", + payload: b"x" + }) + .is_ok()); + } +} diff --git a/crates/storage/Cargo.toml b/crates/storage/Cargo.toml index 99e9a2a..850e8c7 100644 --- a/crates/storage/Cargo.toml +++ b/crates/storage/Cargo.toml @@ -19,5 +19,9 @@ proptest = "=1.4.0" name = "kv-block-xfer" path = "src/bin/kv_block_xfer.rs" +[[bin]] +name = "engine-spike" +path = "src/bin/engine_spike.rs" + [lints] workspace = true diff --git a/crates/storage/src/bin/engine_spike.rs b/crates/storage/src/bin/engine_spike.rs new file mode 100644 index 0000000..2d0d2df --- /dev/null +++ b/crates/storage/src/bin/engine_spike.rs @@ -0,0 +1,29 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 +//! A fake-block ABI exercise. Never reports real engine integration as PASS. +use gpucache_core_types::{content_checksum, synthetic_identity}; +use gpucache_protocol::CommittedExport; +use gpucache_storage::{import_committed_block, EngineBlockContract, ENGINE_SPIKE_ABI_VERSION}; + +fn main() -> Result<(), Box> { + let payload = b"FAKE_BLOCK_NOT_ENGINE_OUTPUT".to_vec(); + let block = CommittedExport { + identity: synthetic_identity(1, 1), + source_generation: 1, + checksum: content_checksum(&payload), + payload, + }; + let contract = EngineBlockContract::new( + ENGINE_SPIKE_ABI_VERSION, + block.identity.clone(), + block.payload.len() as u64, + )?; + let imported = contract.roundtrip(&block)?; + import_committed_block(&imported, [9; 32], 0)?; + println!("status=FAKE_BLOCK_CONTRACT_OK"); + println!("engine=vLLM_NOT_LINKED"); + println!("real_export_import=NOT_RUN_HARDWARE"); + println!("gpu_model_run=NOT_RUN_BUDGET"); + println!("performance=NOT_A_PERFORMANCE_RESULT"); + Ok(()) +} diff --git a/crates/storage/src/engine_spike.rs b/crates/storage/src/engine_spike.rs new file mode 100644 index 0000000..2bfdfe6 --- /dev/null +++ b/crates/storage/src/engine_spike.rs @@ -0,0 +1,131 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 +//! CPU ABI spike for opaque whole blocks. This is not a vLLM adapter or an FFI. +//! The expected identity comes from the caller, never from an untrusted export. + +use gpucache_core_types::{ + verify_checksum, CacheError, CacheIdentity, IdentityLimits, ResourceLimits, +}; +use gpucache_protocol::{decode_export, encode_export, CommittedExport}; + +/// Version of this Rust-side spike contract, independent of GCK1/GCKR/GEXP. +pub const ENGINE_SPIKE_ABI_VERSION: u16 = 1; + +#[derive(Clone, Debug)] +pub struct EngineBlockContract { + identity: CacheIdentity, + length: u64, +} + +impl EngineBlockContract { + pub fn new(abi_version: u16, identity: CacheIdentity, length: u64) -> Result { + if abi_version != ENGINE_SPIKE_ABI_VERSION { + return Err(CacheError::Incompatible); + } + let limits = ResourceLimits::dev_defaults(); + identity.validate(&IdentityLimits::from_resources(&limits))?; + if length == 0 || length > limits.max_object_bytes { + return Err(CacheError::ObjectTooLarge { + limit: limits.max_object_bytes, + actual: length, + }); + } + Ok(Self { identity, length }) + } + + /// Validate before allocating an import buffer or mutating engine state. + /// This authenticates neither the producer nor its model computation. + pub fn validate(&self, block: &CommittedExport) -> Result<(), CacheError> { + if block.identity != self.identity { + return Err(CacheError::Incompatible); + } + if block.source_generation == 0 { + return Err(CacheError::UnknownGeneration); + } + if block.payload.len() as u64 != self.length { + return Err(CacheError::LengthMismatch { + expected: self.length, + actual: block.payload.len() as u64, + }); + } + verify_checksum(&block.payload, block.checksum) + } + + /// Exercise the existing export codec with a caller-supplied ABI contract. + /// No tensor conversion, prefix slicing, or model execution occurs. + pub fn roundtrip(&self, block: &CommittedExport) -> Result { + self.validate(block)?; + let imported = decode_export(&encode_export(block)?)?; + self.validate(&imported)?; + Ok(imported) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use gpucache_core_types::{content_checksum, synthetic_identity, Digest, TenantId}; + + fn fake() -> CommittedExport { + let payload = b"FAKE_BLOCK_NOT_ENGINE_OUTPUT".to_vec(); + CommittedExport { + identity: synthetic_identity(1, 1), + source_generation: 7, + checksum: content_checksum(&payload), + payload, + } + } + + #[test] + fn fake_whole_block_roundtrip_preserves_bytes_and_source_generation() { + let block = fake(); + let contract = + EngineBlockContract::new(1, block.identity.clone(), block.payload.len() as u64) + .unwrap(); + assert_eq!(contract.roundtrip(&block).unwrap(), block); + assert!(EngineBlockContract::new(2, block.identity.clone(), 1).is_err()); + assert!(EngineBlockContract::new(1, block.identity.clone(), 0).is_err()); + } + + #[test] + fn wrong_tenant_model_layout_and_partial_block_cannot_be_imported() { + let block = fake(); + let mut identities = vec![block.identity.clone(); 5]; + identities[0].tenant = TenantId::new(b"other".to_vec()).unwrap(); + identities[1].layout.family += 1; + identities[2].prefix_chain.push(Digest([5; 32])); + identities[3].valid_length -= 1; + identities[4].model_revision = Digest([6; 32]); + for expected in identities { + let contract = + EngineBlockContract::new(1, expected, block.payload.len() as u64).unwrap(); + assert_eq!(contract.roundtrip(&block), Err(CacheError::Incompatible)); + } + } + + #[test] + fn corrupt_truncated_and_uncommitted_exports_fail_before_import() { + let block = fake(); + let contract = + EngineBlockContract::new(1, block.identity.clone(), block.payload.len() as u64) + .unwrap(); + let mut changed = block.clone(); + changed.payload[0] ^= 1; + assert_eq!( + contract.roundtrip(&changed), + Err(CacheError::ChecksumMismatch) + ); + changed = block.clone(); + changed.payload.pop(); + assert!(matches!( + contract.roundtrip(&changed), + Err(CacheError::LengthMismatch { .. }) + )); + changed = block; + changed.source_generation = 0; + assert_eq!( + contract.roundtrip(&changed), + Err(CacheError::UnknownGeneration) + ); + } +} diff --git a/crates/storage/src/lib.rs b/crates/storage/src/lib.rs index 30faba6..d1c9583 100644 --- a/crates/storage/src/lib.rs +++ b/crates/storage/src/lib.rs @@ -8,11 +8,13 @@ #![forbid(unsafe_code)] +mod engine_spike; mod file; mod memory; mod synthetic; mod transfer; +pub use engine_spike::{EngineBlockContract, ENGINE_SPIKE_ABI_VERSION}; pub use file::{FileStore, PersistFault}; pub use memory::{MemoryStore, ReclaimRequest, StoreStats}; pub use synthetic::{decode_synthetic_kv, deterministic_recompute, encode_synthetic_kv}; diff --git a/crates/transport/src/demo.rs b/crates/transport/src/demo.rs index 34a5d32..24546dc 100644 --- a/crates/transport/src/demo.rs +++ b/crates/transport/src/demo.rs @@ -24,6 +24,7 @@ use crate::server::{serve, ServerConfig}; /// same chain is an incompatible miss: this demo does not slice inside a block. #[derive(Clone, Debug, PartialEq, Eq)] pub struct ShareReport { + pub payload_sha256: String, pub exact_hit: &'static str, pub unknown_miss: &'static str, pub partial_prefix: &'static str, @@ -37,6 +38,8 @@ pub struct ShareReport { impl std::fmt::Display for ShareReport { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { writeln!(f, "status=REUSE_OK")?; + writeln!(f, "payload_sha256={}", self.payload_sha256)?; + writeln!(f, "performance=NOT_A_PERFORMANCE_RESULT")?; writeln!(f, "exact_hit={}", self.exact_hit)?; writeln!(f, "unknown_miss={}", self.unknown_miss)?; writeln!( @@ -196,6 +199,11 @@ fn run_in(path: &std::path::Path) -> Result { }); } Ok(ShareReport { + payload_sha256: content_checksum(&got) + .0 + .iter() + .map(|b| format!("{b:02x}")) + .collect(), exact_hit: "PASS", unknown_miss: "PASS", partial_prefix: "WHOLE_BLOCK_PREFIX", diff --git a/docs/CPU_COMPLETION.md b/docs/CPU_COMPLETION.md new file mode 100644 index 0000000..bce3301 --- /dev/null +++ b/docs/CPU_COMPLETION.md @@ -0,0 +1,56 @@ +# CPU/TCP completion audit + +Baseline: `8af53146e47a` (verify the full revision in the regression environment +report). This audit describes software contracts, not production qualification. + +| Area | Baseline gap / preserved invariant | Implementation and regression evidence | +| --- | --- | --- | +| A: hot authorization | ACL and rate hooks were not called by TCP | `server::HotSecurity::admit` gates every data operation before storage dispatch; `tests/hot_security.rs` exercises socket-level denial, tenant isolation and rate exhaustion | +| A: lifecycle | Only committed, compatible blocks may hit; unknown completion retains credit | Existing lifecycle/fault tests retained. `MemoryStore::get_one` bounds reader pins and aggregate reply payloads before copying; reader cancel storm tests retain expired pin credit until explicit drain/fence | +| A: prefix | Whole stored blocks only | ADR 0003 remains unchanged; prefix and incompatible valid-length tests run in workspace regression | +| B: client | No routed connection pool or retry ledger | `routing::ShardedClient` bounds sockets, routes and retries; shard sessions preserve six RPC methods and shard-local generations; `tests/sharded_tcp.rs` and routing unit tests exercise lost replies, concurrent restart and stale epochs | +| B: membership | No atomic batch membership update | `ShardDirectory::apply_membership` validates before fencing/publication; `ControlPlane::apply_membership` uses the same atomic path; failed fences preserve fills, credits and notices | +| C: availability | Actual cold read errors could escape as errors | `refill_matching` returns `Missing` for availability failures without store mutation; `tests/failure_contract.rs` covers HEAD, manifest and payload stages | +| C: publication and reclaim | Retention errors could discard queued work or starve healthy jobs | Cold reclaim retains and rotates failed work; renewal cancels stale expiry; identity/digest/v3-only publication and ambiguous-PUT non-deletion tests remain enforced | +| D: simulation | Queue existed without a host message facade or functional TCP pair | `HostSimulation`, message cancellation/fault tests and `cpu_simulation_pair.py` compare synthetic byte checksums; no timing/performance result | +| E: engine spike | No independent expected-ABI import harness | `EngineBlockContract` and `engine-spike` reject incompatible imports; real engine route remains Pending | +| F: observability | Descriptor/staging counts could be mistaken for wire measurements | `ClientIoStats` separates complete RPC body/envelope byte counts; pool/directory stats expose fence, drain and quarantine state; these exclude network headers, retransmits and partial writes | + +## Remaining explicit boundaries + +- Deployments must authenticate and protect the network boundary. GCK1 carries a + credential fingerprint, not a TLS session or a nonce. mTLS, replay-proof + transport authentication and automatic secret rotation are not implemented. + Empty/restrictive ACL and unlinked-feature tests enforce fail-closed behavior. +- Automatic mutation retries are disabled by default. Opt-in requires an + operator guarantee that server idempotency retention exceeds request TTL. + `Get` is never automatically retried because it creates a pin. Tests enforce + that an ambiguous Get or default mutation sends only one request. +- Routing uses explicit operator-fenced epoch snapshots. This is not distributed + consensus, automatic discovery or a live control-plane daemon. Prefixes on a + different shard are a clean miss; the client does not fan out globally. +- Synchronous filesystem/object-store operations cannot be preempted by the + core trait. Adapters must enforce I/O deadlines; a declared slow cold tier + short-circuits without I/O. The mock tests do not prove a wall-clock bound on + arbitrary blocking adapters. +- Automatic orphan discovery/GC is not implemented. Tests ensure a failed or + ambiguously acknowledged PUT does not cause deletion. Explicit retention + reclaim only touches enrolled objects and verifies deletion. +- No fleet-size, PB-scale, 10,000-GPU, day-long soak, physical NIC flap or + hardware drain has been qualified. More complete CPU contracts do not make + this repository production-ready. + +## Reproduction + +```sh +cargo fmt --all -- --check +cargo clippy --workspace --all-targets -- -D warnings +cargo test --workspace --all-targets +cargo test -p gpucache-transport --all-targets --features rdma-nixl +python3 scripts/ci/vm_regression.py --profile cpu --offline --output artifacts/astra-full-cpu +``` + +Fetch locked dependencies once before an offline run. A new output directory is +required for every retry so failed evidence is retained. `results.json` records +the actual commands and logs; `environment.json` records revision and dirty +state. Command runtimes are never reported as product latency or throughput. diff --git a/docs/OPERATIONS.md b/docs/OPERATIONS.md index 1d5d3d1..ff67b55 100644 --- a/docs/OPERATIONS.md +++ b/docs/OPERATIONS.md @@ -135,6 +135,42 @@ See [../COMPATIBILITY.md](../COMPATIBILITY.md) and [V3_DPU_TARGETS.md](V3_DPU_TA No NVMe format, no DPU mode change, no firmware, no IOMMU or BIOS change, no production network change, no hardware purchase. Those need a separate authorization. There is no website deploy and no crates.io publish in this milestone. G4 live calls stay env-gated. +## CPU fault and rolling-upgrade runbook + +All device faults below are CPU simulations. Stop admission before changing an +owner; publish routing only after every endpoint in the new snapshot has fenced successfully. +Never convert lease expiry into a drain attestation. + +| Event | Operator sequence | CPU regression | +| --- | --- | --- | +| Process restart | Drop the server handle and wait for connection workers to exit; reopen the file store, which advances its persisted epoch; publish a newer route snapshot; discard old generation handles and recompute misses | `cargo test -p gpucache-transport --test sharded_tcp`; `cargo test -p gpucache-storage --test file_backend` | +| NIC flap (simulated) | Quarantine unknown completions and preserve credits; resume admission only with healthy control; explicit CPU drain/fence is required before reuse | `cargo test -p gpucache-dpu --lib`; inspect `HostSimulation::stats` and `RegisteredPool::stats` | +| Disk full | Refuse the failed fill, recompute, free space under the ordinary-file data directory, and retry with a fresh request after recovery; never expose a partially committed block | `cargo test -p gpucache-storage --test fault_matrix` | +| Cold sink storm | Inspect `drain_report`; healthy pending jobs advance while failed jobs retain shared credit. Diagnose permanent failures, then explicitly `take_failed` or re-enqueue under policy; do not delete on ambiguous PUT | `cargo test -p gpucache-cold`; inspect pending/failed counts and retained bytes | +| Epoch stuck/exhausted | Leave the old published membership and notices intact; stop admission and reconcile operator state. Do not reset a live epoch or force-release quarantined regions | `cargo test -p gpucache-transport --lib`; fence overflow/batch atomicity tests | + +For rebalance, externally stop new traffic, record old epochs/in-flight Put IDs, +apply the complete bounded membership snapshot while ControlPlane is Serving, +and fence the actual stores. The control-plane object does not fence a remote +TCP process. Publish routes only after **every** endpoint in the replacement +snapshot has an epoch above the previous snapshot's maximum. Old sessions and +late replies must fail; test a new Lookup/Get/Put cycle before resuming traffic. + +For software upgrade, use `begin_upgrade` to stop fills, drain or fail old work, +then `finish_upgrade` to advance the software generation and fence. Do not call +`apply_membership` while Draining: it deliberately refuses. A failed preflight +leaves membership, generation, notices and credits unchanged. Before finish, +rollback may resume without a fence; after finish it requires another fence. +There are no hot durability replicas to rebuild or migrate. Hardware/firmware +rolling upgrades remain outside this checklist. + +Byte counters describe completed application envelopes, not NIC traffic: +`ClientIoStats` excludes TCP/IP headers, retransmissions and incomplete frames. +`TransferReceipt::staged_bytes` is a host copy count. Reader pins include +expired/stuck pins; `max_in_flight_requests` independently caps readers and +writers. The aggregate payload in a Get batch is capped by `max_object_bytes`. +Reduce batch size after admission refusal; never bypass identity checks. + ## CPU audit changes (2026-10-15) LRU recency is advisory. Lookup and Get update it in memory without filesystem writes. diff --git a/docs/VM_REGRESSION.md b/docs/VM_REGRESSION.md index b5d5c8e..d39c325 100644 --- a/docs/VM_REGRESSION.md +++ b/docs/VM_REGRESSION.md @@ -118,7 +118,7 @@ A missing tool, timeout, crashed child, failed assertion or invalid invocation r 3. Verify the identity, tenant, checksum, commit, prefix, timeout/cancel, quarantine and epoch tests passed. The current tests use synthetic bytes. 4. Check that demos retain their scope labels: `registration=SIMULATED`, `dma_posted=false` where reported, `offload_active=false`, `doca=NOT_LINKED` and `NOT_RUN_HARDWARE`. A changed label requires implementation evidence, not a report rewrite. 5. Inspect failures in the raw logs and reproduce the individual command in a fresh temporary environment. Preserve the failing seed and case if a property/fuzz test prints them. A recorded runner seed alone does not guarantee that a particular property-test library consumed it or can replay every failure. -6. Report limitations with the artifact. Current V3 security hooks are not wired into the TCP service; current limits do not bound every reader pin, reply allocation or metadata structure. Hidden simulated slab bytes stay charged until explicit CPU drain/invalidation/reclaim rules permit release and can cause admission denial. Those APIs are not hardware drain evidence. A green CPU report does not remove these limitations. +6. Report limitations with the artifact. TCP now enforces ACL and rate admission. Reader pins, aggregate Get payloads, routed sockets/retry records and cold metadata have ceilings. Replay nonces, mTLS and automatic rotation are still outside GCK1. Hidden simulated slab bytes stay charged until explicit CPU drain/invalidation/reclaim rules permit release and can cause admission denial. Those APIs are not hardware drain evidence. A green CPU report does not remove these limitations. ## Reruns and failure handling @@ -136,3 +136,7 @@ The following is **PROPOSED / NOT IMPLEMENTED**. A later native runner may suppo - BF3: storage-side topology, BFB/OS/DOCA/firmware, actual queue submit/complete and the paired NIC-RDMA baseline. Arm process start alone is insufficient. These profiles remain unavailable until implemented and exercised. Physical fleet qualification, a day-long device soak, PB-scale and 10,000-GPU reliability cannot be derived from this VM runner. + +## CPU completion additions + +`engine-spike` checks fake-block ABI validation and keeps real engine work `NOT_RUN_HARDWARE` / `NOT_RUN_BUDGET`. `cpu-simulation-pair` compares checked synthetic payload digests from localhost TCP and host simulation, includes fault checks, and requires `NOT_A_PERFORMANCE_RESULT`. Neither gate contains product performance samples. See [CPU_COMPLETION.md](CPU_COMPLETION.md) for the gap map and remaining boundaries. diff --git a/docs/adr/0001-integration-route.md b/docs/adr/0001-integration-route.md index 092df9d..c7d61ec 100644 --- a/docs/adr/0001-integration-route.md +++ b/docs/adr/0001-integration-route.md @@ -42,3 +42,7 @@ The V3 CPU slice did not run the spike and did not add SGLang or HiCache. See [0 - Callers must not assume GPUCache can read an arbitrary engine's KV bytes. - Resharding, a different quant format, or a different layout ABI is a miss, not a conversion. - A later ADR will replace this status with Accepted and name the one route. Until then, engine adapters stay out of the tree. + +## CPU ABI spike (2026-10-20) + +ADR 0013 adds `EngineBlockContract` and a fake-block harness with independent expected identity, length, generation and checksum checks. This is a CPU spike, not an engine adapter or exported KV tensor. The integration route remains Pending; real export/import is `NOT_RUN_HARDWARE`, and the GPU/model campaign is `NOT_RUN_BUDGET`. diff --git a/docs/adr/0013-cpu-spikes-and-evidence.md b/docs/adr/0013-cpu-spikes-and-evidence.md new file mode 100644 index 0000000..b5c1e31 --- /dev/null +++ b/docs/adr/0013-cpu-spikes-and-evidence.md @@ -0,0 +1,42 @@ +# ADR 0013: CPU spike contracts and evidence boundaries + +- Status: Accepted for CPU contracts; engine and hardware integration Pending +- Date: 2026-10-20 +- Milestone: V3 + +## Decision + +`EngineBlockContract` is a Rust-side ABI spike for opaque whole blocks. It +validates an independently supplied full identity, exact payload length, +nonzero source generation and checksum before an import. Its ABI version is +independent of the unchanged GCK1/GCKR/GEXP formats. It performs no layout +conversion, tensor slicing, authentication or model computation. + +`engine-spike` uses visibly synthetic bytes. Its success label is +`FAKE_BLOCK_CONTRACT_OK`; real export/import is `NOT_RUN_HARDWARE` and the +GPU/model campaign is `NOT_RUN_BUDGET`. ADR 0001 stays Pending. This spike does +not select native vLLM versus LMCache and is not an engine adapter or FFI. + +`HostSimulation` dispatches synchronous typed storage-queue messages on the +host without another queue or inbox. Unknown cancellation keeps credits until +an explicit simulated drain. Fault replies contain recompute decisions, not +payloads. It is not a network/device message format. + +The CPU runner compares independently checked DPU-simulation and localhost +TCP payload digests and executes both failure paths. This is a functional +pair, labeled `NOT_A_PERFORMANCE_RESULT`. It is not the NIC-RDMA versus +BlueField comparison required for TCO. Device opens continue to fail closed +with empty feature dependencies; aarch64 remains CPU evidence only. + +## Verification and limitations + +Workspace tests exercise cross-identity imports, corrupt/truncated exports, +cancel storms, explicit drain and simulated fault decisions. The bounded +export property suite generates valid payloads and checks truncation, trailing +bytes and payload bit flips. The runner includes `engine-spike` and +`cpu-simulation-pair` as required CPU gates. + +No CUDA/DOCA/SPDK/NIXL/Mooncake dependency is added. Real vLLM, RDMA, GDS, +BF3 and live RustFS remain `NOT_RUN_HARDWARE`. No model was downloaded and no +license evaluation was run. `TCO_UNKNOWN` and `go_gate=NOT_EVALUATED` stay in +force. No hot erasure coding, durability replica or performance sample exists. diff --git a/scripts/ci/cpu_simulation_pair.py b/scripts/ci/cpu_simulation_pair.py new file mode 100644 index 0000000..83ac306 --- /dev/null +++ b/scripts/ci/cpu_simulation_pair.py @@ -0,0 +1,68 @@ +#!/usr/bin/env python3 +# Copyright 2024-2026 RustFS, Inc. +# SPDX-License-Identifier: Apache-2.0 +"""Compare deterministic CPU simulation and TCP functional evidence, never timing.""" +import hashlib +import pathlib +import subprocess + +from vm_regression import NO_LIVE_ENDPOINT, sanitized_environment + + +def fields(text): + result = {} + for line in text.splitlines(): + if "=" in line: + key, value = line.split("=", 1) + if key in result: + raise ValueError("duplicate report field: " + key) + result[key] = value + return result + + +def validate(dpu, tcp, fault): + expected = hashlib.sha256(b"shared-kv-block").hexdigest() + required = ( + (dpu, "status", "DPU_CONTRACT_OK"), + (dpu, "hardware", "NOT_RUN_HARDWARE"), + (dpu, "offload_active", "false"), + (dpu, "message_path", "HOST_SIMULATION"), + (dpu, "fault_path", "QUARANTINE_DRAIN_PASS"), + (dpu, "performance", "NOT_A_PERFORMANCE_RESULT"), + (tcp, "status", "REUSE_OK"), + (tcp, "bytes_match", "PASS"), + (tcp, "performance", "NOT_A_PERFORMANCE_RESULT"), + (fault, "status", "FALLBACK_OK"), + (dpu, "payload_sha256", expected), + (tcp, "payload_sha256", expected), + ) + for report, key, value in required: + if report.get(key) != value: + raise ValueError("functional pair mismatch: " + key) + + +def main(): + root = pathlib.Path(__file__).resolve().parents[2] + reports = [] + for package, binary in ( + ("gpucache-dpu", "dpu-demo"), + ("gpucache-transport", "share-demo"), + ("gpucache-transport", "fault-demo"), + ): + command = ["cargo", *NO_LIVE_ENDPOINT, "run", "--locked", "--offline", + "-p", package, "--bin", binary, "--quiet"] + completed = subprocess.run(command, cwd=root, env=sanitized_environment(0), + capture_output=True, text=True, timeout=60, check=True) + for line in completed.stdout.splitlines(): + print(binary + "." + line) + reports.append(fields(completed.stdout)) + validate(*reports) + print("status=CPU_SIMULATION_PAIR_OK") + print("comparison=FUNCTIONAL_TCP_AND_HOST_SIMULATION") + print("hardware=NOT_RUN_HARDWARE") + print("performance=NOT_A_PERFORMANCE_RESULT") + print("tco=TCO_UNKNOWN") + + +if __name__ == "__main__": + main() diff --git a/scripts/ci/tests/test_cpu_simulation_pair.py b/scripts/ci/tests/test_cpu_simulation_pair.py new file mode 100644 index 0000000..cdd75d2 --- /dev/null +++ b/scripts/ci/tests/test_cpu_simulation_pair.py @@ -0,0 +1,44 @@ +# Copyright 2024-2026 RustFS, Inc. +# SPDX-License-Identifier: Apache-2.0 +import hashlib +import pathlib +import sys +import unittest + +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1])) +import cpu_simulation_pair as pair + + +class PairTests(unittest.TestCase): + def reports(self): + digest = hashlib.sha256(b"shared-kv-block").hexdigest() + return ( + dict(status="DPU_CONTRACT_OK", hardware="NOT_RUN_HARDWARE", + offload_active="false", message_path="HOST_SIMULATION", + fault_path="QUARANTINE_DRAIN_PASS", + performance="NOT_A_PERFORMANCE_RESULT", payload_sha256=digest), + dict(status="REUSE_OK", bytes_match="PASS", + performance="NOT_A_PERFORMANCE_RESULT", payload_sha256=digest), + dict(status="FALLBACK_OK"), + ) + + def test_missing_wrong_and_contradictory_evidence_fails(self): + pair.validate(*self.reports()) + for report_index, key, value in ( + (0, "offload_active", "true"), (0, "payload_sha256", "incorrect"), + (1, "bytes_match", "FAIL"), (2, "status", "FAIL"), + ): + reports = self.reports() + reports[report_index][key] = value + with self.assertRaises(ValueError): + pair.validate(*reports) + with self.assertRaises(ValueError): + pair.fields("status=PASS\nstatus=FAIL\n") + reports = self.reports() + del reports[0]["hardware"] + with self.assertRaises(ValueError): + pair.validate(*reports) + + +if __name__ == "__main__": + unittest.main() diff --git a/scripts/ci/vm_regression.py b/scripts/ci/vm_regression.py index ab452ac..0c0a6d2 100644 --- a/scripts/ci/vm_regression.py +++ b/scripts/ci/vm_regression.py @@ -69,6 +69,9 @@ def cpu_gates(offline=False): "--all-features") + locked), ] demos = [ + ("engine-spike", "gpucache-storage", "engine-spike", + ("status=FAKE_BLOCK_CONTRACT_OK", "real_export_import=NOT_RUN_HARDWARE", + "gpu_model_run=NOT_RUN_BUDGET", "performance=NOT_A_PERFORMANCE_RESULT")), ("share-demo", "gpucache-transport", "share-demo", ("status=REUSE_OK",)), ("fault-demo", "gpucache-transport", "fault-demo", ("status=FALLBACK_OK",)), ("shard-demo", "gpucache-transport", "shard-demo", @@ -84,6 +87,10 @@ def cpu_gates(offline=False): for name, package, binary, expected in demos: gates.append(Gate(name, cargo + ("run", "-p", package, "--bin", binary, "--quiet") + locked, expected)) + gates.append(Gate("cpu-simulation-pair", + (sys.executable, "scripts/ci/cpu_simulation_pair.py"), + ("status=CPU_SIMULATION_PAIR_OK", "hardware=NOT_RUN_HARDWARE", + "performance=NOT_A_PERFORMANCE_RESULT", "tco=TCO_UNKNOWN"))) return gates From a6de65fc9ad58642856d4922faef40952ca293d4 Mon Sep 17 00:00:00 2001 From: RustFS <155562731+loverustfs@users.noreply.github.com> Date: Wed, 21 Oct 2026 08:45:39 +0800 Subject: [PATCH 2/4] Demonstrate coordinated CPU TCP scaling from one to ten shards --- crates/transport/examples/scaling_smoke.rs | 522 +++++++++++++++++++++ docs/SCALING.md | 245 ++++++++++ 2 files changed, 767 insertions(+) create mode 100644 crates/transport/examples/scaling_smoke.rs create mode 100644 docs/SCALING.md diff --git a/crates/transport/examples/scaling_smoke.rs b/crates/transport/examples/scaling_smoke.rs new file mode 100644 index 0000000..7152645 --- /dev/null +++ b/crates/transport/examples/scaling_smoke.rs @@ -0,0 +1,522 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 +//! One process, independent localhost TCP services, synthetic whole blocks. +//! This exercises orchestration; it is not discovery, migration, or a benchmark. + +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; + +use gpucache_core_types::{ + content_checksum, encode_physical_key, synthetic_identity, Attestation, BlockState, CacheError, + CacheIdentity, IdentityLimits, RequestId, ResourceLimits, +}; +use gpucache_protocol::{ + AuthContext, BatchReply, CommitItem, GetItem, ItemHeader, LookupItem, LookupOutcome, + ReleaseCancelItem, ReservePutItem, TenantPolicy, +}; +use gpucache_storage::FileStore; +use gpucache_transport::{ + serve, ClientLimits, ServerConfig, ServerHandle, ShardEndpoint, ShardSession, ShardedClient, + TcpClient, +}; + +const COUNTS: [usize; 3] = [1, 3, 10]; +const MAX_PREPARE_OPENS: usize = 3; +const FRAME_BYTES: usize = 1024 * 1024; + +struct Scratch(PathBuf); +impl Scratch { + fn new() -> std::io::Result { + static NEXT: AtomicU64 = AtomicU64::new(0); + for _ in 0..100 { + let path = std::env::temp_dir().join(format!( + "gpucache-scaling-smoke-{}-{}", + std::process::id(), + NEXT.fetch_add(1, Ordering::Relaxed) + )); + match std::fs::create_dir(&path) { + Ok(()) => return Ok(Self(path)), + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {} + Err(error) => return Err(error), + } + } + Err(std::io::Error::other("no unused scaling smoke directory")) + } +} +impl Drop for Scratch { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } +} + +#[derive(Debug, PartialEq, Eq)] +struct Report { + roundtrips: usize, + transitions: usize, + stages: Vec, +} + +#[derive(Debug, PartialEq, Eq)] +struct Stage { + count: usize, + epoch: u64, + revision: u64, + verified_shards: Vec, +} + +struct Requests(u128); +impl Requests { + fn header(&mut self, session: &ShardSession) -> Result { + self.at(session.epoch(), session.deadline(10_000)) + } + + fn at(&mut self, epoch: u64, deadline_ms: u64) -> Result { + let request_id = RequestId::from_counter(self.0)?; + self.0 = self + .0 + .checked_add(1) + .ok_or(CacheError::IdempotencyConflict)?; + Ok(ItemHeader { + request_id, + epoch, + deadline_ms, + }) + } +} + +fn auth() -> Result { + let identity = synthetic_identity(1, 1); + // Public, synthetic test credential. Never use this fingerprint remotely. + AuthContext::new(identity.tenant, identity.trust_domain, [9; 32]) +} + +fn policy() -> Result { + let auth = auth()?; + TenantPolicy::new(auth.tenant, auth.trust_domain, auth.credential_fingerprint) +} + +fn check(ok: bool, reason: &'static str) -> Result<(), CacheError> { + if ok { + Ok(()) + } else { + Err(CacheError::Storage { reason }) + } +} + +fn one(batch: BatchReply) -> Result { + check(batch.items.len() == 1, "scaling reply length")?; + batch + .items + .into_iter() + .next() + .ok_or(CacheError::UnknownGeneration)? + .body +} + +/// Only called while this demo owns the directory exclusively and every old +/// service has stopped and joined. It never edits a MANIFEST or resets an epoch. +/// Fresh roots need two opens for epoch 2 and three opens for epoch 3. The demo +/// has a fixed three-open bound, not an unbounded production epoch reconciler. +fn prepare_store(root: &Path, target: u64) -> Result { + check((1..=3).contains(&target), "scaling target epoch")?; + for _ in 0..MAX_PREPARE_OPENS { + let store = FileStore::open(root, ResourceLimits::dev_defaults(), vec![policy()?])?; + if store.epoch() == target { + return Ok(store); + } + if store.epoch() > target { + return Err(CacheError::EpochMismatch { + store: store.epoch(), + request: target, + }); + } + drop(store); + } + Err(CacheError::FencingRejected { + reason: "scaling epoch preparation", + }) +} + +fn start_servers( + root: &Path, + count: usize, + epoch: u64, +) -> Result<(Vec, Vec), CacheError> { + let mut servers = Vec::with_capacity(count); + let mut routes = Vec::with_capacity(count); + for shard in 0..count { + let store = prepare_store(&root.join(format!("shard-{shard}")), epoch)?; + check(store.epoch() == epoch, "scaling store epoch")?; + let server = serve(store, ServerConfig::default())?; + routes.push(ShardEndpoint { + shard_id: shard as u64, + address: server.addr(), + epoch, + }); + servers.push(server); + } + Ok((servers, routes)) +} + +fn identity(nonce: u64) -> CacheIdentity { + let mut id = synthetic_identity(1, 1); + id.prefix_chain[0].0[..8].copy_from_slice(&nonce.to_le_bytes()); + id +} + +fn identity_on( + client: &ShardedClient, + shard: u64, + first: u64, +) -> Result { + for nonce in first..first + 10_000 { + let id = identity(nonce); + if client.shard_for(&id)? == shard { + return Ok(id); + } + } + Err(CacheError::Storage { + reason: "scaling identity search exhausted", + }) +} + +fn payload(id: &CacheIdentity) -> Result, CacheError> { + let key = encode_physical_key( + id, + &IdentityLimits::from_resources(&ResourceLimits::dev_defaults()), + )?; + // Distinct synthetic bytes per identity; never an engine KV tensor. + Ok(content_checksum(&key).0.to_vec()) +} + +fn lookup( + session: &mut ShardSession, + id: &CacheIdentity, + requests: &mut Requests, +) -> Result { + one(session.lookup(vec![LookupItem { + header: requests.header(session)?, + identity: id.clone(), + }])?) +} + +fn reserve( + session: &mut ShardSession, + id: &CacheIdentity, + requests: &mut Requests, +) -> Result { + let bytes = payload(id)?; + Ok(one(session.reserve_put(vec![ReservePutItem { + header: requests.header(session)?, + identity: id.clone(), + generation: 0, + length: bytes.len() as u64, + checksum: content_checksum(&bytes), + payload: Some(bytes), + lease_ttl_ms: 10_000, + }])?)? + .generation) +} + +fn write_and_read( + session: &mut ShardSession, + id: &CacheIdentity, + requests: &mut Requests, +) -> Result { + let bytes = payload(id)?; + let generation = reserve(session, id, requests)?; + one(session.commit(vec![CommitItem { + header: requests.header(session)?, + generation, + length: bytes.len() as u64, + checksum: content_checksum(&bytes), + }])?)?; + match lookup(session, id, requests)? { + LookupOutcome::Hit { + generation: got, + length, + checksum, + .. + } if got == generation + && length == bytes.len() as u64 + && checksum == content_checksum(&bytes) => {} + _ => return Err(CacheError::Incompatible), + } + let header = requests.header(session)?; + let pin_owner = header.request_id; + let hit = one(session.get(vec![GetItem { + header, + generation, + identity: id.clone(), + pin_ttl_ms: 10_000, + }])?)?; + check(hit.payload == bytes, "scaling payload mismatch")?; + let released = one(session.release_cancel(vec![ReleaseCancelItem { + header: requests.header(session)?, + generation, + region_id: Some(hit.region_id), + pin_owner: Some(pin_owner), + attestation: Attestation::Drained, + }])?)?; + check( + released.state == BlockState::Committed && !released.quarantined, + "scaling pin release", + )?; + Ok(generation) +} + +fn predictor(count: usize, epoch: u64) -> Result { + // Pure placement calculation: new() and shard_for() open no sockets. Real + // listening ports are selected later and do not affect rendezvous placement. + let routes = (0..count) + .map(|shard| ShardEndpoint { + shard_id: shard as u64, + address: ([127, 0, 0, 1], 9735 + shard as u16).into(), + epoch, + }) + .collect(); + ShardedClient::new(auth()?, routes, ClientLimits::default()) +} + +fn run_scaling_smoke() -> Result> { + let scratch = Scratch::new()?; + let (mut servers, routes) = start_servers(&scratch.0, COUNTS[0], 1)?; + let client = ShardedClient::new(auth()?, routes, ClientLimits::default())?; + let mut requests = Requests(1); + let mut report = Report { + roundtrips: 0, + transitions: 0, + stages: Vec::new(), + }; + + for (phase, &count) in COUNTS.iter().enumerate() { + if phase > 0 { + let target_epoch = phase as u64 + 1; + let next = predictor(count, target_epoch)?; + let mut moved = None; + for nonce in 500_000 + phase as u64 * 10_000..500_000 + phase as u64 * 10_000 + 10_000 { + let id = identity(nonce); + let old_shard = client.shard_for(&id)?; + if next.shard_for(&id)? != old_shard { + moved = Some((id, old_shard)); + break; + } + } + let (moved_id, old_shard) = moved.ok_or(CacheError::Storage { + reason: "scaling remap search exhausted", + })?; + let mut stale = client.route(&moved_id)?; + write_and_read(&mut stale, &moved_id, &mut requests)?; + let pending_id = identity_on(&client, old_shard, 900_000 + phase as u64 * 10_000)?; + let pending_generation = reserve(&mut stale, &pending_id, &mut requests)?; + + // All calls are synchronous and every Get above was drained. There + // is no other client: admission is paused here, including reads. + // Dropping every handle stops and joins its connection workers. + // This is explicit orchestration, not a remotely triggered fence. + drop(servers); + let (started, replacement) = start_servers(&scratch.0, count, target_epoch)?; + servers = started; + let old_address = replacement + .iter() + .find(|route| route.shard_id == old_shard) + .ok_or(CacheError::UnknownGeneration)? + .address; + let old_revision = client.stats().revision; + check( + client.replace_routes(old_revision, replacement)? == old_revision + 1, + "scaling route revision", + )?; + check( + matches!( + lookup(&mut stale, &moved_id, &mut requests), + Err(CacheError::FencingRejected { + reason: "route revision" + }) + ), + "scaling stale lookup", + )?; + let pending_bytes = payload(&pending_id)?; + let commit = stale.commit(vec![CommitItem { + header: requests.header(&stale)?, + generation: pending_generation, + length: pending_bytes.len() as u64, + checksum: content_checksum(&pending_bytes), + }]); + check( + matches!( + commit, + Err(CacheError::FencingRejected { + reason: "route revision" + }) + ), + "scaling stale commit", + )?; + drop(stale); + + // The old committed copy remains on its original store. Inspect it + // directly; shard sessions intentionally refuse wrong-route reads. + let mut original = + TcpClient::connect(old_address, auth()?, Duration::from_secs(5), FRAME_BYTES)?; + let old_copy = one(original.lookup(vec![LookupItem { + header: requests.at(original.epoch(), original.deadline(10_000))?, + identity: moved_id.clone(), + }])?)?; + check( + matches!(old_copy, LookupOutcome::Hit { .. }), + "scaling original committed copy", + )?; + let pending_copy = one(original.lookup(vec![LookupItem { + header: requests.at(original.epoch(), original.deadline(10_000))?, + identity: pending_id.clone(), + }])?)?; + check( + matches!(pending_copy, LookupOutcome::Miss { .. }), + "scaling original uncommitted copy", + )?; + let old_epoch = one(original.lookup(vec![LookupItem { + header: requests.at(target_epoch - 1, original.deadline(10_000))?, + identity: moved_id.clone(), + }])?); + check( + matches!(old_epoch, Err(CacheError::EpochMismatch { store, request }) + if store == target_epoch && request == target_epoch - 1), + "scaling remote old epoch", + )?; + drop(original); + let mut new_owner = client.route(&moved_id)?; + check(new_owner.shard_id() != old_shard, "scaling expected remap")?; + check( + matches!( + lookup(&mut new_owner, &moved_id, &mut requests)?, + LookupOutcome::Miss { .. } + ), + "scaling must not migrate", + )?; + // Simulated recomputation repopulates the new owner explicitly. + write_and_read(&mut new_owner, &moved_id, &mut requests)?; + drop(new_owner); + let mut pending_owner = client.route(&pending_id)?; + check( + matches!( + lookup(&mut pending_owner, &pending_id, &mut requests)?, + LookupOutcome::Miss { .. } + ), + "scaling uncommitted block survived", + )?; + report.transitions += 1; + } + let mut stage = Stage { + count, + epoch: phase as u64 + 1, + revision: client.stats().revision, + verified_shards: Vec::new(), + }; + for shard in 0..count { + let id = identity_on( + &client, + shard as u64, + phase as u64 * 100_000 + shard as u64 * 10_000, + )?; + let mut session = client.route(&id)?; + check( + session.shard_id() == shard as u64 && session.epoch() == phase as u64 + 1, + "scaling session placement", + )?; + check( + matches!( + lookup(&mut session, &id, &mut requests)?, + LookupOutcome::Miss { .. } + ), + "scaling initial miss", + )?; + write_and_read(&mut session, &id, &mut requests)?; + report.roundtrips += 1; + stage.verified_shards.push(shard as u64); + } + check(client.stats().shards == count, "scaling shard count")?; + check( + client.stats().connections <= count, + "scaling connection count", + )?; + report.stages.push(stage); + } + drop(client); + drop(servers); + Ok(report) +} + +fn main() -> Result<(), Box> { + let report = run_scaling_smoke()?; + println!("status=SCALING_SMOKE_OK"); + println!("topology=SINGLE_PROCESS_LOCALHOST"); + for stage in &report.stages { + println!( + "stage_shards={} store_epoch={} local_revision={}", + stage.count, stage.epoch, stage.revision + ); + if stage.epoch > 1 { + println!("transition_to_shards={} stale_session=REJECTED stale_commit=REJECTED old_epoch=REJECTED uncommitted=MISS remapped=MISS_THEN_RECOMPUTE original_copy=RETAINED", stage.count); + } + for shard in &stage.verified_shards { + println!( + "shard={} epoch={} lookup_put_commit_get_drained=OK", + shard, stage.epoch + ); + } + } + println!("shard_counts=1,3,10"); + println!("verified_store_epochs=1,2,3"); + println!("per_shard_roundtrips={}", report.roundtrips); + println!("transition_fences={}", report.transitions); + println!("stale_sessions=REJECTED"); + println!("inflight_puts=FENCED_MISS"); + println!("remapped_blocks=MISS_THEN_RECOMPUTE"); + println!("automatic_payload_migration=NONE"); + println!("durability_replication=NONE"); + println!("engine_blocks=SYNTHETIC"); + println!("hardware=NOT_RUN_HARDWARE"); + println!("performance=NOT_A_PERFORMANCE_RESULT"); + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn one_three_ten_localhost_services_fence_and_recompute_without_migration() { + let report = run_scaling_smoke().unwrap(); + assert_eq!(report.roundtrips, 14); + assert_eq!(report.transitions, 2); + assert_eq!(report.stages.len(), COUNTS.len()); + for (phase, stage) in report.stages.iter().enumerate() { + assert_eq!(stage.count, COUNTS[phase]); + assert_eq!(stage.epoch, phase as u64 + 1); + assert_eq!(stage.revision, phase as u64 + 1); + assert_eq!( + stage.verified_shards, + (0..COUNTS[phase] as u64).collect::>() + ); + } + } + + #[test] + fn epoch_preparation_refuses_to_reset_or_run_without_a_bound() { + let scratch = Scratch::new().unwrap(); + let root = scratch.0.join("epoch"); + assert!(prepare_store(&root, 0).is_err()); + assert!(prepare_store(&root, 4).is_err()); + let store = prepare_store(&root, 3).unwrap(); + assert_eq!(store.epoch(), 3); + drop(store); + assert!(matches!( + prepare_store(&root, 2), + Err(CacheError::EpochMismatch { + store: 4, + request: 2 + }) + )); + } +} diff --git a/docs/SCALING.md b/docs/SCALING.md new file mode 100644 index 0000000..6547716 --- /dev/null +++ b/docs/SCALING.md @@ -0,0 +1,245 @@ +# Scaling the CPU/TCP contract from 1 to 3 to 10 services + +GPUCache currently provides independent file-backed TCP services and a Rust +`ShardedClient`. A caller supplies a complete routing snapshot and coordinates +ownership changes. Starting more services does not automatically form a cluster. +There is no `--join`, seed discovery, membership daemon, replicated metadata +service, remote fence RPC, payload migration, or durability replica placement. + +The executable example below verifies 1 → 3 → 10 services on one host. It uses +one process, separate loopback listeners and separate store directories. It is +CPU correctness evidence, not a multi-host deployment qualification, an inference +engine integration, a throughput result, or a production-readiness claim. + +## Run the executable example + +From a checkout containing +[`scaling_smoke.rs`](../crates/transport/examples/scaling_smoke.rs): + +```bash +cargo run -p gpucache-transport --example scaling_smoke --offline +cargo test -p gpucache-transport --example scaling_smoke --offline +``` + +`--offline` requires the locked dependencies to be cached already. Omit that flag +for the initial dependency download if necessary. The example's positive test is +also executed by `cargo test --workspace --all-targets`. + +The example creates an exclusive temporary root, starts independent `FileStore` +services on OS-selected loopback ports, and removes its temporary data afterward. +Its public test credential and synthetic 32-byte blocks are for this local example +only. It does not connect to a GPU, DPU, inference engine, or RustFS endpoint. + +Each phase checks every shard using a full physical `CacheIdentity`: + +1. Lookup returns a miss for a new identity. +2. Reserve/Put accepts the payload; Commit publishes its returned generation. +3. Lookup returns that generation, length, and checksum. +4. Get returns the exact payload for that identity. +5. Release/Cancel supplies the Get request's pin owner and `Attestation::Drained`; + the block remains committed and is not quarantined. + +At each transition the example retains one old session and an uncommitted +reservation, completes all Get lifecycles, and stops and joins every old server. +It opens the surviving and new stores at the next common epoch, replaces the +complete client snapshot, and checks these outcomes: + +- The old session's Lookup and pending Commit are rejected by its route revision. +- A direct request carrying the previous store epoch is rejected by the restarted + server, independently of the client's local route fence. +- The pending block is absent from its original restarted store and its current + routed owner. +- A committed identity whose owner changes remains on its original store but is + initially absent on its new owner. Explicit simulated recomputation populates + the new owner. No payload migration took place. + +The output includes a line for every verified shard, for example: + +```text +status=SCALING_SMOKE_OK +topology=SINGLE_PROCESS_LOCALHOST +stage_shards=1 store_epoch=1 local_revision=1 +shard=0 epoch=1 lookup_put_commit_get_drained=OK +stage_shards=3 store_epoch=2 local_revision=2 +transition_to_shards=3 stale_session=REJECTED stale_commit=REJECTED old_epoch=REJECTED uncommitted=MISS remapped=MISS_THEN_RECOMPUTE original_copy=RETAINED +shard=0 epoch=2 lookup_put_commit_get_drained=OK +shard=1 epoch=2 lookup_put_commit_get_drained=OK +shard=2 epoch=2 lookup_put_commit_get_drained=OK +stage_shards=10 store_epoch=3 local_revision=3 +transition_to_shards=10 stale_session=REJECTED stale_commit=REJECTED old_epoch=REJECTED uncommitted=MISS remapped=MISS_THEN_RECOMPUTE original_copy=RETAINED +``` + +Ten further `shard=0` through `shard=9` success lines follow at epoch 3. The final +summary is: + +```text +shard_counts=1,3,10 +verified_store_epochs=1,2,3 +per_shard_roundtrips=14 +transition_fences=2 +stale_sessions=REJECTED +inflight_puts=FENCED_MISS +remapped_blocks=MISS_THEN_RECOMPUTE +automatic_payload_migration=NONE +durability_replication=NONE +engine_blocks=SYNTHETIC +hardware=NOT_RUN_HARDWARE +performance=NOT_A_PERFORMANCE_RESULT +``` + +The 14 roundtrips count the one-per-shard checks across the three phases. The +transition checks perform additional operations. The output does not report a +performance measurement. + +## Four different identifiers + +| Value | Meaning and owner | +| --- | --- | +| `ShardEndpoint::shard_id` | A stable placement ID assigned by the caller. A server does not advertise this ID in the current wire handshake. Keep its mapping to a store under operator control. | +| `ShardEndpoint::epoch` | The expected actual epoch of that endpoint's `FileStore`. The TCP handshake checks it before using a routed connection. | +| `ClientStats::revision` | A local client's route snapshot revision. It starts at 1 and advances on replacement. It is not a global consensus term or a store epoch. | +| Generation | A block identifier local to one shard and epoch. Two stores can both allocate generation 1. Preserve shard and epoch with the generation. | + +`ShardDirectory` and `ControlPlane` are additional in-process APIs. Their member +incarnations and directory epochs do not configure a remote `FileStore`, and +`gpucache-tcp serve` does not instantiate a membership coordinator. + +The full physical identity includes tenant, trust domain, model and layout +compatibility, plus block/prefix identity. Rendezvous placement uses this identity +and stable shard IDs; endpoint order and address changes do not change the hash. +An added shard moves some identities to new owners. There is no promise that +exactly one third or one tenth of a small workload lands on each shard. + +Keep `(shard_id, epoch, generation)` with outstanding work. Reopen the original +owner with `client.shard(shard_id)` only while that epoch remains valid. Never +route a Commit, Release, or Invalidate by hashing a generation number. Partition +identity-bearing batches by shard before submission. Request IDs must remain +unique for new logical requests; do not rewrite an old request's epoch to make it +pass a fence. + +## Plan the service addresses and ownership + +These are example address assignments, not discovery configuration: + +| Phase | Stable IDs | Separate hosts on a private network | Same-host fixed-port alternative | +| --- | --- | --- | --- | +| 1 service | `0` | `10.0.0.11:9735` | `127.0.0.1:9735` | +| 3 services | `0..2` | `10.0.0.11:9735` through `10.0.0.13:9735` | ports `9735..9737` | +| 10 services | `0..9` | `10.0.0.11:9735` through `10.0.0.20:9735` | ports `9735..9744` | + +Each service owns a distinct data directory. One host may run several services, +but those services do not create independent host failure domains. The runnable +example uses ephemeral ports rather than these fixed ones. + +The existing standalone CLI can start a file-backed TCP service: + +```bash +cargo run -p gpucache-transport --bin gpucache-tcp --offline -- serve \ + --data-dir ./work/shard-0 \ + --tenant example-tenant --trust example-domain \ + --fingerprint "$GPUCACHE_FINGERPRINT_HEX" \ + --bind 127.0.0.1:9735 +``` + +`GPUCACHE_FINGERPRINT_HEX` must contain the caller-provisioned 64-hex credential; +the CLI does not generate or distribute it. Use a protected trusted network and +`--allow-remote` for a non-loopback bind. The fingerprint is bearer-equivalent +and is not a TLS client certificate. This transport does not provide TLS or +mTLS; see [ADR 0007](adr/0007-trusted-fabric.md). + +The CLI has no shard ID, expected epoch, resource-limit, or membership flags. +Launching three CLI processes alone is not the 1 → 3 transition. The caller must +supply real addresses and verified store epochs to `ShardedClient::new`, retain +the client for pool reuse, then use `route(&identity)` to obtain a session. The +example's `start_servers`, `prepare_store`, and `run_scaling_smoke` functions are +the tested library bootstrap and orchestration reference. + +## Perform a coordinated transition + +The currently supported route replacement requires a coordinated pause. It does +not implement a zero-downtime rolling expansion. Use the same sequence for 1 → 3 +and 3 → 10; removing or re-adding a shard also follows the snapshot rule. + +1. **Pause admission in every old client**, including Lookup/Get and writes. + Stopping writes alone leaves read pins and active operations. Retain original + shard/epoch/generation ownership for outstanding requests. +2. Complete or cancel known operations and finish every known Get drain/release + lifecycle. Unknown completion remains quarantined and charged; lease expiry + is not evidence of drain. The synthetic example drains every Get explicitly. +3. Stop all old service owners and wait for them and their connection workers to + exit. The library example drops every `ServerHandle`, whose destructor stops + and joins its workers. With independently managed hosts, the operator must + provide the equivalent stop-and-wait barrier. A disconnected host is not + proven stopped. Do not reopen a directory or transfer ownership while an old + owner may still serve it. +4. Exclusively open each surviving and new store and verify its actual epoch. + Choose a target greater than the highest epoch in the old snapshot. **Every + endpoint in the new snapshot must exceed that old maximum**, including newly + added endpoints. Do not assume a newly created directory has the right epoch. +5. Start the new service set using those verified store instances. If using local + directory/control-plane membership, validate and apply its complete bounded + snapshot while external admission is still paused. `ControlPlane` must be + `Serving`: its `apply_membership` rejects the software-upgrade `Draining` + state. This API is not a remote fence. +6. Distribute the same complete endpoint snapshot to all callers, with authenticated + distribution supplied by the deployment. For each existing client call + `replace_routes(client.stats().revision, endpoints)` and check success. Each + client has its own expected revision; distribution is not atomic across + processes. Resume only after the required callers acknowledge the new view. +7. Drop retired sessions and recompute any absent, failed, or ambiguously completed + block as new work. A new owner can miss even when the previous store still + contains committed bytes. Repopulate explicitly and resume admission. + +`replace_routes` closes idle connections and retires old checked-out sessions; +it also suppresses replies that arrive after publication. Those checked-out +connections remain charged until dropped. Publishing routes by itself cannot +stop a remote old server or fence another client's stale view. + +### How the example reaches epochs 1, 2, and 3 + +`FileStore::open` creates a fresh directory at epoch 1. Every later exclusive open +of that store increments its persisted manifest epoch. There is no public +`FileStore::set_epoch` API or CLI `--epoch` flag. + +| Target snapshot | Surviving stores | Fresh stores | +| --- | --- | --- | +| 1 service, epoch 1 | none | Open shard 0 once. | +| 3 services, epoch 2 | After stop/join, reopen shard 0 once. | Open, close, and reopen shards 1 and 2. | +| 10 services, epoch 3 | After stop/join, reopen shards 0 through 2 once. | Open and close twice, then open shards 3 through 9 a third time. | + +The example's preparation helper is bounded to three opens and targets 1 through +3. It returns the final open store directly to `serve`, so starting the service +does not open it again. A CLI invocation opens its own store and therefore would +increment an existing epoch again. Never edit `MANIFEST`, delete an old root to +reset its epoch, or run overlapping owners to bypass the rule. The caller must +guarantee exclusive directory ownership during preparation. + +This helper is deliberately limited to a fresh disposable demonstration. An +existing deployment at epochs 7, 7, and 9 needs every replacement endpoint above +9, not simply epoch 3 because it has ten services. The repository does not yet +provide a general fleet epoch-administration tool. Any failure or unexpected +epoch requires stopping the transition and reconciling the actual store states +before publishing a snapshot; do not silently lower the target. + +## Limits and expected misses + +Client defaults allow 64 shards, 64 total connections, and 4 connections per +shard. A server defaults to 16 accepted connections across all clients. Multiple +client processes each have their own pool, so increasing the shard count does +not establish a capacity plan. Configure the library's `ClientLimits`, +`ServerConfig`, and store `ResourceLimits` for the actual workload and handle +quota rejection with recomputation. A 10-service smoke test is not evidence for +10 physical hosts or PB scale. + +Mutation retries remain disabled by default. Enabling them requires an explicit +server idempotency-retention contract longer than the maximum client request +lifetime. Get is never replayed automatically because it creates a pin. Errors, +timeouts, and epoch failures are not hits. See [ADR 0011](adr/0011-bounded-shard-client.md) +for the bounded pool, replay, and membership contracts. + +Whole-block prefix reuse is shard-local. A longer identity can route away from a +stored shorter prefix; that lookup is a clean miss. There is no cross-shard prefix +search or intra-block slicing. Membership changes do not copy blocks. Old local +copies may remain until eviction, but no replica protocol maintains them and +no durability promise follows. A lost hot block is recomputed; cold durability +belongs to RustFS G4, outside this hot-path example. From 4c5ebb023eb8f89d7630df804dfbd718771efd31 Mon Sep 17 00:00:00 2001 From: RustFS <155562731+loverustfs@users.noreply.github.com> Date: Wed, 21 Oct 2026 08:46:13 +0800 Subject: [PATCH 3/4] Gate coordinated 1-to-3-to-10 shard scaling in CPU regression --- docs/VM_REGRESSION.md | 4 +++- scripts/ci/vm_regression.py | 9 +++++++++ 2 files changed, 12 insertions(+), 1 deletion(-) diff --git a/docs/VM_REGRESSION.md b/docs/VM_REGRESSION.md index d39c325..47076eb 100644 --- a/docs/VM_REGRESSION.md +++ b/docs/VM_REGRESSION.md @@ -39,7 +39,7 @@ cargo clippy --workspace --all-targets -- -D warnings cargo test --workspace --all-targets ``` -The runner uses `--locked` for Cargo resolution and fetches the locked dependencies before its offline metadata check, so a clean VM does not need an already populated cache unless `--offline` is selected. It checks Clippy with default and all features, checks the dependency contract and runner unit tests, runs doc tests and combined-feature tests, and exercises the CPU-compatible `rdma-nixl` and `dpu-bf3` feature slots. It also runs the reuse, fallback, shard, cold, DPU and paired-report demos. The current list contains 18 gates; the command list in `results.json` is authoritative for a retained run. Optional feature compilation does not link a device, and a demo's successful exit does not convert an unavailable hardware row into a pass. +The runner uses `--locked` for Cargo resolution and fetches the locked dependencies before its offline metadata check, so a clean VM does not need an already populated cache unless `--offline` is selected. It checks Clippy with default and all features, checks the dependency contract and runner unit tests, runs doc tests and combined-feature tests, and exercises the CPU-compatible `rdma-nixl` and `dpu-bf3` feature slots. It also runs the reuse, fallback, shard, cold, DPU and paired-report demos. The current list contains 21 gates; the command list in `results.json` is authoritative for a retained run. Optional feature compilation does not link a device, and a demo's successful exit does not convert an unavailable hardware row into a pass. Manual reproduction of the optional feature contracts is: @@ -140,3 +140,5 @@ These profiles remain unavailable until implemented and exercised. Physical flee ## CPU completion additions `engine-spike` checks fake-block ABI validation and keeps real engine work `NOT_RUN_HARDWARE` / `NOT_RUN_BUDGET`. `cpu-simulation-pair` compares checked synthetic payload digests from localhost TCP and host simulation, includes fault checks, and requires `NOT_A_PERFORMANCE_RESULT`. Neither gate contains product performance samples. See [CPU_COMPLETION.md](CPU_COMPLETION.md) for the gap map and remaining boundaries. + +`scaling-smoke` checks 1 → 3 → 10 independent localhost TCP servers with coordinated stop/join, increasing epochs, stale-session and in-flight-write rejection, per-shard roundtrips and remapped-key recomputation. It is `SINGLE_PROCESS_LOCALHOST`, uses synthetic blocks, and is not physical multi-host or performance evidence. See [SCALING.md](SCALING.md). diff --git a/scripts/ci/vm_regression.py b/scripts/ci/vm_regression.py index 0c0a6d2..c7f9afe 100644 --- a/scripts/ci/vm_regression.py +++ b/scripts/ci/vm_regression.py @@ -91,6 +91,15 @@ def cpu_gates(offline=False): (sys.executable, "scripts/ci/cpu_simulation_pair.py"), ("status=CPU_SIMULATION_PAIR_OK", "hardware=NOT_RUN_HARDWARE", "performance=NOT_A_PERFORMANCE_RESULT", "tco=TCO_UNKNOWN"))) + gates.append(Gate("scaling-smoke", cargo + ("run", "-p", "gpucache-transport", + "--example", "scaling_smoke", "--quiet") + + locked, + ("status=SCALING_SMOKE_OK", "topology=SINGLE_PROCESS_LOCALHOST", + "shard_counts=1,3,10", "verified_store_epochs=1,2,3", + "stale_sessions=REJECTED", "inflight_puts=FENCED_MISS", + "remapped_blocks=MISS_THEN_RECOMPUTE", + "hardware=NOT_RUN_HARDWARE", + "performance=NOT_A_PERFORMANCE_RESULT"))) return gates From 18f544b660235f82728df691658201def7ef4144 Mon Sep 17 00:00:00 2001 From: RustFS <155562731+loverustfs@users.noreply.github.com> Date: Wed, 21 Oct 2026 08:48:14 +0800 Subject: [PATCH 4/4] Clarify scaling client limits and TCP frame configuration --- docs/SCALING.md | 31 ++++++++++++++++++++++++++++--- 1 file changed, 28 insertions(+), 3 deletions(-) diff --git a/docs/SCALING.md b/docs/SCALING.md index 6547716..8ab9ecf 100644 --- a/docs/SCALING.md +++ b/docs/SCALING.md @@ -226,9 +226,34 @@ before publishing a snapshot; do not silently lower the target. Client defaults allow 64 shards, 64 total connections, and 4 connections per shard. A server defaults to 16 accepted connections across all clients. Multiple client processes each have their own pool, so increasing the shard count does -not establish a capacity plan. Configure the library's `ClientLimits`, -`ServerConfig`, and store `ResourceLimits` for the actual workload and handle -quota rejection with recomputation. A 10-service smoke test is not evidence for +not establish a capacity plan. Choose limits when constructing the client; +`replace_routes` does not modify `ClientLimits`. For example, application code +that already supplies `auth` and verified initial `endpoints` can provision this +one-connection-per-shard configuration before the first stage: + +```rust +use gpucache_transport::{ClientLimits, ShardedClient}; + +let limits = ClientLimits { + max_shards: 10, + max_connections: 10, + max_connections_per_shard: 1, + max_frame_bytes: 2 * 1024 * 1024, + ..ClientLimits::default() +}; +let client = ShardedClient::new(auth, endpoints, limits)?; +``` + +Reducing only `max_connections` to 3 while retaining the default per-shard cap of +4 is invalid. The configuration above supports the final ten-shard topology from +the outset. Size the library's `ServerConfig` and store `ResourceLimits` as well, +and handle quota rejection with recomputation. + +The standalone CLI server uses a 2 MiB frame-body cap, matched above by +`max_frame_bytes`. Frame limits are not negotiated; configure matching client +and server limits. A client with a smaller receive cap can reject a reply that +its server admitted. The executable `scaling_smoke` instead uses matching 1 MiB +library defaults on both sides. A 10-service smoke test is not evidence for 10 physical hosts or PB scale. Mutation retries remain disabled by default. Enabling them requires an explicit