Validate CPU engine contracts and coordinated 1-to-10 shard scaling

README.md

@@ -27,16 +27,16 @@ 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 |
 | V3: second engine (SGLang / HiCache) | Not added. Evaluation note only. See [docs/adr/0008-second-engine.md](docs/adr/0008-second-engine.md) |
 | Mooncake-informed CPU contracts | Implemented in this tree. See [docs/MOONCAKE.md](docs/MOONCAKE.md) and [docs/adr/0009-mooncake-lessons.md](docs/adr/0009-mooncake-lessons.md). Mooncake is not linked. No NIC, Soft-RoCE, or DPU number is claimed |
 | Erasure coding and durability replicas | Will not be built. A lost or failed hot block is a clean miss, then recompute and repopulate. Cold durability is RustFS G4 |
 
-The integration route (native vLLM KV connector or LMCache) is **not chosen**. See [docs/adr/0001-integration-route.md](docs/adr/0001-integration-route.md). Hot-tier failure policy is [docs/adr/0002-no-erasure-coding.md](docs/adr/0002-no-erasure-coding.md). The V3 offload candidate is [docs/adr/0006-storage-queue-offload.md](docs/adr/0006-storage-queue-offload.md). It is a hypothesis pending a paired measurement, not a measured bottleneck.
+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
 

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<DpuReport, CacheError> {
     } 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)]

crates/dpu/src/lib.rs

@@ -12,13 +12,15 @@
 mod ablation;
 mod degrade;
 mod demo;
+mod message;
 mod queue;
 mod reactor;
 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,

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<Self, CacheError> {
+        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<SimulationReply, CacheError> {
+        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());
+    }
+}

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

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<dyn std::error::Error>> {
+    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(())
+}