From 74e2ef4478492cbd60f11b80ac4280c8a479d2a4 Mon Sep 17 00:00:00 2001 From: loverustfs Date: Sat, 17 Oct 2026 15:05:08 +0800 Subject: [PATCH 1/4] fix: enforce V3 replay, reply and quarantine invariants --- crates/core-types/src/identity.rs | 128 +++++++ crates/dpu/src/degrade.rs | 6 +- crates/dpu/src/queue.rs | 518 ++++++++++++++++++++++++--- crates/protocol/src/auth.rs | 78 +++- crates/protocol/src/codec.rs | 79 +++- crates/protocol/src/harden.rs | 247 ++++++++++++- crates/storage/src/file.rs | 6 +- crates/storage/src/memory.rs | 2 + crates/storage/src/tests.rs | 30 ++ crates/storage/tests/file_backend.rs | 18 + crates/transport/src/client.rs | 287 ++++++++++++++- 11 files changed, 1322 insertions(+), 77 deletions(-) diff --git a/crates/core-types/src/identity.rs b/crates/core-types/src/identity.rs index 72b6623..7c61ba9 100644 --- a/crates/core-types/src/identity.rs +++ b/crates/core-types/src/identity.rs @@ -209,6 +209,15 @@ impl CacheIdentity { Self::decode_body(bytes, false) } + /// Validate protocol hard limits independently of tighter store quotas. + pub fn validate_wire(&self) -> Result<(), CacheError> { + self.validate(&IdentityLimits { + max_field_bytes: crate::limits::DEFAULT_MAX_FIELD_BYTES, + max_prefix_links: 4096, + max_block_tokens: u32::MAX, + }) + } + pub fn validate(&self, limits: &IdentityLimits) -> Result<(), CacheError> { if self.version != IDENTITY_VERSION { return Err(CacheError::IdentityInvalid { @@ -483,6 +492,9 @@ impl CacheIdentity { reason: "physical field on logical key", }); } + // Syntax alone is insufficient: callers also decode exports and keys + // without passing through a store's admission checks. + id.validate_wire()?; Ok(id) } } @@ -667,3 +679,119 @@ pub fn logical_fields_diverge(a: &CacheIdentity, b: &CacheIdentity) -> bool { || a.prefix_chain != b.prefix_chain || a.multimodal != b.multimodal } + +#[cfg(test)] +mod tests { + use super::*; + use crate::{decode_physical_key, encode_physical_key, synthetic_identity, ResourceLimits}; + + #[test] + fn identity_decoder_rejects_structurally_encoded_invalid_identity() { + let base = synthetic_identity(1, 1); + let mut cases = Vec::new(); + let mut id = base.clone(); + id.layout.family = 0; + cases.push(id); + let mut id = base.clone(); + id.shard.tp_rank = id.shard.tp_size; + cases.push(id); + let mut id = base.clone(); + id.block_token_count = 0; + cases.push(id); + let mut id = base.clone(); + id.valid_length = id.block_token_count + 1; + cases.push(id); + let mut id = base; + id.rope.scaling_factor_bits = 1.0f64.to_bits(); + cases.push(id); + for id in cases { + let bytes = id.encode_body(false).unwrap(); + assert!(matches!( + CacheIdentity::from_bytes(&bytes), + Err(CacheError::IdentityInvalid { .. }) + )); + } + } + + #[test] + fn wire_validation_preserves_hard_bounds_without_imposing_store_defaults() { + let mut id = synthetic_identity(1, 1); + id.block_token_count = u32::MAX; + id.valid_length = u32::MAX; + id.prefix_chain = vec![Digest([1; 32]); 4096]; + assert_eq!( + CacheIdentity::from_bytes(&id.to_bytes().unwrap()).unwrap(), + id + ); + id.prefix_chain.push(Digest([2; 32])); + assert!(id.validate_wire().is_err()); + assert!(CacheIdentity::from_bytes(&id.to_bytes().unwrap()).is_err()); + } + + #[test] + fn physical_keys_separate_all_security_and_layout_fields() { + let base = synthetic_identity(1, 1); + let mut variants = Vec::new(); + let mut changed = base.clone(); + changed.tenant = TenantId::new(b"other-tenant".to_vec()).unwrap(); + variants.push(changed); + let mut changed = base.clone(); + changed.trust_domain = TrustDomain::new(b"other-trust".to_vec()).unwrap(); + variants.push(changed); + let mut changed = base.clone(); + changed.model_revision = Digest([9; 32]); + variants.push(changed); + let mut changed = base.clone(); + changed.model_config_hash = Digest([9; 32]); + variants.push(changed); + let mut changed = base.clone(); + changed.tokenizer_revision = Digest([9; 32]); + variants.push(changed); + let mut changed = base.clone(); + changed.adapter = AdapterIdentity::Lora { + revision: Digest([9; 32]), + }; + variants.push(changed); + let mut changed = base.clone(); + changed.kv_dtype = KvDtype::F16; + variants.push(changed); + let mut changed = base.clone(); + changed.quant_format = QuantFormat::Fp8; + variants.push(changed); + let mut changed = base.clone(); + changed.layout.minor += 1; + variants.push(changed); + let mut changed = base.clone(); + changed.shard.tp_size = 2; + changed.shard.tp_rank = 1; + variants.push(changed); + let mut changed = base.clone(); + changed.position = PositionEncoding::Absolute; + variants.push(changed); + let mut changed = base.clone(); + changed.rope.theta_bits = 500_000.0f64.to_bits(); + variants.push(changed); + let mut changed = base.clone(); + changed.prefix_chain[0] = Digest([9; 32]); + variants.push(changed); + let mut changed = base.clone(); + changed.block_token_count += 1; + variants.push(changed); + let mut changed = base.clone(); + changed.valid_length -= 1; + variants.push(changed); + let mut changed = base.clone(); + changed.multimodal = Some(MultimodalIdentity { + processor_revision: Digest([8; 32]), + input_digest: Digest([9; 32]), + }); + variants.push(changed); + let limits = IdentityLimits::from_resources(&ResourceLimits::dev_defaults()); + let key = encode_physical_key(&base, &limits).unwrap(); + for changed in variants { + let other_key = encode_physical_key(&changed, &limits).unwrap(); + assert_ne!(key, other_key); + assert_eq!(decode_physical_key(&other_key).unwrap(), changed); + } + } +} diff --git a/crates/dpu/src/degrade.rs b/crates/dpu/src/degrade.rs index 859ed89..34fa75e 100644 --- a/crates/dpu/src/degrade.rs +++ b/crates/dpu/src/degrade.rs @@ -4,8 +4,10 @@ //! Failure outcomes for the V3 degrade matrix. //! //! Every listed fault is a clean miss. The engine recomputes. The decision -//! carries no payload, does not report a wrong hit, and does not stick in a -//! permanent block. Device-side effects live on `StorageQueue::apply_fault`. +//! carries no payload and does not report a wrong hit. `blocked=false` means +//! the caller may fall back to recompute; it does not promise queue admission. +//! Unknown completions retain quarantine credit until explicit drain or a +//! simulated restart fence. Effects live on `StorageQueue::apply_fault`. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum Fault { diff --git a/crates/dpu/src/queue.rs b/crates/dpu/src/queue.rs index 64e469d..4827ccb 100644 --- a/crates/dpu/src/queue.rs +++ b/crates/dpu/src/queue.rs @@ -11,11 +11,16 @@ use std::collections::BTreeMap; use gpucache_core_types::{ - content_checksum, verify_checksum, CacheError, Checksum, HARD_BYTES_CAP, + content_checksum, verify_checksum, BlockState, CacheError, Checksum, QuotaKind, HARD_BYTES_CAP, }; use crate::degrade::{decide, DegradeDecision, Fault}; +// Bound queue-owned metadata independently of payload bytes. These are CPU +// contract limits, not measured NIC queue or registration limits. +const MAX_QUEUE_TASKS: usize = 4096; +const MAX_SLAB_SLOTS: usize = 4096; + #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub struct TaskId(pub u64); @@ -85,7 +90,7 @@ impl Slab { .ok_or(CacheError::QuotaExceeded { kind: gpucache_core_types::QuotaKind::RetainedBytes, })?; - if next > self.cap || self.next == u64::MAX { + if next > self.cap || self.next == u64::MAX || self.slots.len() >= MAX_SLAB_SLOTS { return Err(CacheError::QuotaExceeded { kind: gpucache_core_types::QuotaKind::RetainedBytes, }); @@ -106,7 +111,7 @@ impl Slab { Ok(id) } - fn read(&mut self, tenant: &[u8], id: u64) -> Result, CacheError> { + fn readable_slot(&mut self, tenant: &[u8], id: u64) -> Result<&Slot, CacheError> { let Some(slot) = self.slots.get_mut(&id) else { return Err(CacheError::Storage { reason: "clean miss", @@ -124,7 +129,11 @@ impl Slab { slot.exposed = false; return Err(CacheError::ChecksumMismatch); } - Ok(slot.bytes.clone()) + Ok(slot) + } + + fn read(&mut self, tenant: &[u8], id: u64) -> Result, CacheError> { + Ok(self.readable_slot(tenant, id)?.bytes.clone()) } fn hide_all(&mut self) { @@ -144,10 +153,28 @@ enum Pending { Read { tenant: Vec, slot: u64 }, } +impl Pending { + fn write_bytes(&self) -> u64 { + match self { + Self::Write { payload, .. } => payload.len() as u64, + Self::Read { .. } => 0, + } + } + + fn reads_slot(&self, tenant: &[u8], id: u64) -> bool { + matches!(self, Self::Read { tenant: owner, slot } if owner.as_slice() == tenant && *slot == id) + } +} + +/// CPU-only storage task owner. Completion, failure, fault injection and drain +/// attestations are trusted owner operations; a task id is not a tenant token. +/// Returned payloads belong to the caller and are outside the queue's byte cap. pub struct StorageQueue { slab: Slab, pending: BTreeMap, pending_bytes: u64, + quarantined: BTreeMap, + quarantined_bytes: u64, next_task: u64, max_inflight: usize, control_up: bool, @@ -155,13 +182,15 @@ pub struct StorageQueue { impl StorageQueue { pub fn open_simulated(cap_bytes: u64, max_inflight: usize) -> Result { - if max_inflight == 0 { + if max_inflight == 0 || max_inflight > MAX_QUEUE_TASKS { return Err(CacheError::Malformed { reason: "dpu slab" }); } Ok(Self { slab: Slab::new(cap_bytes)?, pending: BTreeMap::new(), pending_bytes: 0, + quarantined: BTreeMap::new(), + quarantined_bytes: 0, next_task: 1, max_inflight, control_up: true, @@ -176,15 +205,37 @@ impl StorageQueue { self.pending.len() } + /// Tasks with unknown completion still hold their payloads and admission + /// credit until an explicit drain attestation or simulated restart fence. + pub fn quarantined(&self) -> usize { + self.quarantined.len() + } + + /// All queue-owned payload bytes, including hidden slots and quarantine. + pub fn retained_bytes(&self) -> u64 { + self.slab.used + self.pending_bytes + self.quarantined_bytes + } + pub fn submit_write(&mut self, tenant: &[u8], payload: &[u8]) -> Result { self.admit(tenant, payload.len())?; - let next = self.pending_bytes.saturating_add(payload.len() as u64); - if next.saturating_add(self.slab.used) > self.slab.cap { + if self.retained_bytes() + payload.len() as u64 > self.slab.cap { return Err(CacheError::QuotaExceeded { - kind: gpucache_core_types::QuotaKind::RetainedBytes, + kind: QuotaKind::RetainedBytes, }); } - self.pending_bytes = next; + let reserved_writes = self + .pending + .values() + .chain(self.quarantined.values()) + .filter(|task| matches!(task, Pending::Write { .. })) + .count(); + let reserved_slots = self.slab.slots.len() + reserved_writes; + if reserved_slots >= MAX_SLAB_SLOTS || reserved_writes as u64 >= u64::MAX - self.slab.next { + return Err(CacheError::QuotaExceeded { + kind: QuotaKind::RetainedBytes, + }); + } + self.pending_bytes += payload.len() as u64; self.enqueue(Pending::Write { tenant: tenant.to_vec(), payload: payload.to_vec(), @@ -193,6 +244,9 @@ impl StorageQueue { pub fn submit_read(&mut self, tenant: &[u8], slot: u64) -> Result { self.admit(tenant, 1)?; + // Bind only to a slot that exists and is readable now. A guessed id + // must never become a read of a slot allocated by a later write. + self.slab.readable_slot(tenant, slot)?; self.enqueue(Pending::Read { tenant: tenant.to_vec(), slot, @@ -200,6 +254,11 @@ impl StorageQueue { } pub fn complete(&mut self, task: TaskId) -> Result { + if self.quarantined.contains_key(&task.0) { + return Err(CacheError::FencingRejected { + reason: "task quarantined", + }); + } let Some(pending) = self.pending.remove(&task.0) else { return Err(CacheError::UnknownGeneration); }; @@ -217,7 +276,14 @@ impl StorageQueue { Ok(self.receipt(task, slot, payload)) } + /// Report a confirmed terminal failure. Unknown completion (including a + /// timeout or cancellation request) must use `quarantine` instead. pub fn fail(&mut self, task: TaskId) -> Result<(), CacheError> { + if self.quarantined.contains_key(&task.0) { + return Err(CacheError::FencingRejected { + reason: "task quarantined", + }); + } let pending = self .pending .remove(&task.0) @@ -228,28 +294,112 @@ impl StorageQueue { Ok(()) } + pub fn quarantine(&mut self, task: TaskId) -> Result<(), CacheError> { + if self.quarantined.contains_key(&task.0) { + return Ok(()); + } + let pending = self + .pending + .remove(&task.0) + .ok_or(CacheError::UnknownGeneration)?; + let bytes = pending.write_bytes(); + self.pending_bytes -= bytes; + self.quarantined_bytes += bytes; + self.quarantined.insert(task.0, pending); + Ok(()) + } + + /// Trusted CPU simulation attestation that this task can no longer touch + /// its buffers. A timer or control-plane restoration is not an attestation. + /// A hardware backend would need a verified device drain before this call. + pub fn attest_drained(&mut self, task: TaskId) -> Result<(), CacheError> { + let pending = self + .quarantined + .remove(&task.0) + .ok_or(CacheError::UnknownGeneration)?; + self.quarantined_bytes -= pending.write_bytes(); + Ok(()) + } + + /// Hide a tenant's slot. Already queued reads must observe the invalidation + /// at completion. The slot remains charged until it is safely reclaimed. + pub fn invalidate(&mut self, tenant: &[u8], slot: u64) -> Result<(), CacheError> { + self.check_control_and_tenant(tenant)?; + let stored = self + .slab + .slots + .get_mut(&slot) + .ok_or(CacheError::UnknownGeneration)?; + if stored.tenant.as_slice() != tenant { + return Err(CacheError::Unauthorized); + } + stored.exposed = false; + Ok(()) + } + + /// Reclaim an invalidated CPU slot only when no active or quarantined read + /// references it. This does not attest a hardware registration drain. + pub fn reclaim_invalidated(&mut self, tenant: &[u8], slot: u64) -> Result<(), CacheError> { + self.check_control_and_tenant(tenant)?; + let stored = self + .slab + .slots + .get(&slot) + .ok_or(CacheError::UnknownGeneration)?; + if stored.tenant.as_slice() != tenant { + return Err(CacheError::Unauthorized); + } + if stored.exposed { + return Err(CacheError::NotReclaimable { + from: BlockState::Committed, + }); + } + let refs = self + .pending + .values() + .chain(self.quarantined.values()) + .filter(|task| task.reads_slot(tenant, slot)) + .count(); + if refs > 0 { + return Err(CacheError::InUse { refs: refs as u32 }); + } + let removed = self + .slab + .slots + .remove(&slot) + .ok_or(CacheError::UnknownGeneration)?; + self.slab.used -= removed.bytes.len() as u64; + Ok(()) + } + + fn quarantine_all(&mut self) { + self.quarantined.append(&mut self.pending); + self.quarantined_bytes += self.pending_bytes; + self.pending_bytes = 0; + } + pub fn apply_fault(&mut self, fault: Fault) -> DegradeDecision { match fault { Fault::DpuRestart => { + // Explicit CPU epoch fence. Keep monotonic ids and do not + // restore control authority lost before this restart. self.pending.clear(); self.pending_bytes = 0; + self.quarantined.clear(); + self.quarantined_bytes = 0; self.slab.restart(); - self.control_up = true; } Fault::GpuReset | Fault::NicFlap | Fault::StorageLoss => { - self.pending.clear(); - self.pending_bytes = 0; + self.quarantine_all(); self.slab.hide_all(); - self.control_up = true; } Fault::NvmeTimeout => { - if let Some(id) = self.pending.keys().next().copied() { - let _ = self.fail(TaskId(id)); - } + // This fault has no task id, so no outstanding task can be + // assumed complete. Targeted failures use quarantine(task). + self.quarantine_all(); } Fault::ControlPlaneLoss => { - self.pending.clear(); - self.pending_bytes = 0; + self.quarantine_all(); self.slab.hide_all(); self.control_up = false; } @@ -263,10 +413,8 @@ impl StorageQueue { } fn admit(&self, tenant: &[u8], len: usize) -> Result<(), CacheError> { - if !self.control_up { - return Err(CacheError::Cancelled); - } - if tenant.is_empty() || tenant.len() > 128 || len == 0 { + self.check_control_and_tenant(tenant)?; + if len == 0 { return Err(CacheError::Malformed { reason: "dpu task" }); } if len > HARD_BYTES_CAP || len as u64 > self.slab.cap { @@ -275,7 +423,9 @@ impl StorageQueue { actual: len as u64, }); } - if self.pending.len() >= self.max_inflight || self.next_task == u64::MAX { + if self.pending.len() + self.quarantined.len() >= self.max_inflight + || self.next_task == u64::MAX + { return Err(CacheError::QuotaExceeded { kind: gpucache_core_types::QuotaKind::InFlightRequests, }); @@ -283,6 +433,16 @@ impl StorageQueue { Ok(()) } + fn check_control_and_tenant(&self, tenant: &[u8]) -> Result<(), CacheError> { + if !self.control_up { + return Err(CacheError::Cancelled); + } + if tenant.is_empty() || tenant.len() > 128 { + return Err(CacheError::Malformed { reason: "dpu task" }); + } + Ok(()) + } + fn enqueue(&mut self, pending: Pending) -> Result { let id = self.next_task; self.next_task += 1; @@ -372,9 +532,8 @@ mod tests { let read = queue.submit_read(b"tenant-a", slot).unwrap(); let got = queue.complete(read).unwrap(); assert_eq!(got.payload.as_deref(), Some(b"kv-block".as_slice())); - let foreign = queue.submit_read(b"tenant-b", slot).unwrap(); assert_eq!( - queue.complete(foreign).unwrap_err(), + queue.submit_read(b"tenant-b", slot).unwrap_err(), CacheError::Unauthorized ); } @@ -385,9 +544,8 @@ mod tests { let write = queue.submit_write(b"tenant-a", b"secret-bytes").unwrap(); queue.fail(write).unwrap(); assert_eq!(queue.inflight(), 0); - let read = queue.submit_read(b"tenant-a", 1).unwrap(); assert_eq!( - queue.complete(read).unwrap_err(), + queue.submit_read(b"tenant-a", 1).unwrap_err(), CacheError::Storage { reason: "clean miss" } @@ -396,11 +554,10 @@ mod tests { #[test] fn faults_miss_and_then_accept_new_work() { - let mut queue = StorageQueue::open_simulated(128, 4).unwrap(); - let write = queue.submit_write(b"tenant-a", b"block").unwrap(); - let wrote = queue.complete(write).unwrap(); - let slot = wrote.slot.unwrap(); for fault in all_faults() { + let mut queue = StorageQueue::open_simulated(128, 4).unwrap(); + let write = queue.submit_write(b"tenant-a", b"block").unwrap(); + let slot = queue.complete(write).unwrap().slot.unwrap(); if matches!(fault, Fault::RustFsSlow | Fault::RustFsUnavailable) { let before = queue.inflight(); let decision = queue.apply_fault(*fault); @@ -410,13 +567,14 @@ mod tests { assert_eq!(queue.inflight(), before); continue; } - let _ = queue.submit_write(b"tenant-a", b"inflight"); + let old_write = queue.submit_write(b"tenant-a", b"inflight").unwrap(); let decision = queue.apply_fault(*fault); assert_eq!(decision.outcome, "CLEAN_MISS"); assert!(decision.recompute); assert!(!decision.wrong_hit); assert!(!decision.blocked); assert_eq!(queue.inflight(), 0); + assert!(queue.complete(old_write).is_err()); if *fault == Fault::ControlPlaneLoss { assert!(!queue.control_up()); assert_eq!( @@ -425,8 +583,15 @@ mod tests { ); queue.restore_control(); } - let read = queue.submit_read(b"tenant-a", slot).unwrap(); - assert!(queue.complete(read).is_err()); + if *fault == Fault::NvmeTimeout { + let read = queue.submit_read(b"tenant-a", slot).unwrap(); + assert_eq!( + queue.complete(read).unwrap().payload.as_deref(), + Some(b"block".as_slice()) + ); + } else { + assert!(queue.submit_read(b"tenant-a", slot).is_err()); + } let again = queue.submit_write(b"tenant-a", b"fresh").unwrap(); let done = queue.complete(again).unwrap(); assert!(done.payload.is_none()); @@ -443,14 +608,12 @@ mod tests { let stored = queue.slab.slots.get_mut(&slot).unwrap(); stored.bytes[0] ^= 0xff; } - let read = queue.submit_read(b"tenant-a", slot).unwrap(); assert_eq!( - queue.complete(read).unwrap_err(), + queue.submit_read(b"tenant-a", slot).unwrap_err(), CacheError::ChecksumMismatch ); - let again = queue.submit_read(b"tenant-a", slot).unwrap(); assert_eq!( - queue.complete(again).unwrap_err(), + queue.submit_read(b"tenant-a", slot).unwrap_err(), CacheError::Storage { reason: "clean miss" } @@ -479,7 +642,7 @@ mod tests { } #[test] - fn faults_release_pending_credit_but_do_not_free_hidden_slots() { + fn faults_keep_unknown_writes_charged_until_drain_or_restart() { for fault in [ Fault::DpuRestart, Fault::GpuReset, @@ -491,16 +654,287 @@ mod tests { let mut queue = StorageQueue::open_simulated(8, 4).unwrap(); let first = queue.submit_write(b"tenant-a", b"1234").unwrap(); queue.complete(first).unwrap(); - queue.submit_write(b"tenant-a", b"5678").unwrap(); + let pending = queue.submit_write(b"tenant-a", b"5678").unwrap(); queue.apply_fault(fault); assert_eq!(queue.pending_bytes, 0); if fault == Fault::ControlPlaneLoss { queue.restore_control(); } - queue.submit_write(b"tenant-a", b"5678").unwrap(); - if fault != Fault::DpuRestart { + if fault == Fault::DpuRestart { + assert_eq!(queue.quarantined(), 0); + assert_eq!(queue.retained_bytes(), 0); + } else { + assert_eq!(queue.quarantined(), 1); + assert_eq!(queue.retained_bytes(), 8); assert!(queue.submit_write(b"tenant-a", b"x").is_err()); + assert!(queue.complete(pending).is_err()); + assert!(queue.fail(pending).is_err()); + assert_eq!(queue.retained_bytes(), 8); + queue.attest_drained(pending).unwrap(); + assert_eq!(queue.retained_bytes(), 4); } + queue.submit_write(b"tenant-a", b"5678").unwrap(); } } + + #[test] + fn timeout_quarantines_every_affected_task_and_late_completion_never_publishes() { + let mut queue = StorageQueue::open_simulated(8, 2).unwrap(); + let first = queue.submit_write(b"tenant-a", b"1234").unwrap(); + let second = queue.submit_write(b"tenant-b", b"5678").unwrap(); + queue.apply_fault(Fault::NvmeTimeout); + assert_eq!(queue.inflight(), 0); + assert_eq!(queue.quarantined(), 2); + assert_eq!(queue.retained_bytes(), 8); + for task in [first, second] { + queue.quarantine(task).unwrap(); + assert_eq!( + queue.complete(task).unwrap_err(), + CacheError::FencingRejected { + reason: "task quarantined" + } + ); + } + assert!(queue.slab.slots.is_empty()); + assert!(matches!( + queue.submit_read(b"tenant-a", 1), + Err(CacheError::QuotaExceeded { + kind: QuotaKind::InFlightRequests + }) + )); + queue.attest_drained(first).unwrap(); + assert_eq!(queue.retained_bytes(), 4); + let next = queue.submit_write(b"tenant-a", b"abcd").unwrap(); + let slot = queue.complete(next).unwrap().slot.unwrap(); + assert_eq!(queue.slab.slots[&slot].tenant, b"tenant-a"); + assert_eq!(queue.retained_bytes(), 8); + assert_eq!( + queue.attest_drained(first).unwrap_err(), + CacheError::UnknownGeneration + ); + } + + #[test] + fn repeated_faults_preserve_exact_quarantine_credit_and_require_a_drain() { + let mut queue = StorageQueue::open_simulated(8, 2).unwrap(); + let write = queue.submit_write(b"tenant-a", b"12345678").unwrap(); + assert_eq!( + queue.attest_drained(write).unwrap_err(), + CacheError::UnknownGeneration + ); + assert_eq!(queue.inflight(), 1); + for fault in [ + Fault::NvmeTimeout, + Fault::NvmeTimeout, + Fault::NicFlap, + Fault::StorageLoss, + Fault::GpuReset, + ] { + queue.apply_fault(fault); + assert_eq!(queue.quarantined(), 1); + assert_eq!(queue.retained_bytes(), 8); + assert!(queue.submit_write(b"tenant-a", b"x").is_err()); + } + queue.attest_drained(write).unwrap(); + assert_eq!(queue.quarantined(), 0); + assert_eq!(queue.retained_bytes(), 0); + assert!(queue.complete(write).is_err()); + queue.submit_write(b"tenant-a", b"12345678").unwrap(); + } + + #[test] + fn unrelated_faults_never_restore_lost_control_authority() { + let mut queue = StorageQueue::open_simulated(16, 4).unwrap(); + let write = queue.submit_write(b"tenant-a", b"1234").unwrap(); + queue.apply_fault(Fault::ControlPlaneLoss); + for fault in all_faults() { + queue.apply_fault(*fault); + assert!(!queue.control_up()); + assert_eq!( + queue.submit_write(b"tenant-a", b"x").unwrap_err(), + CacheError::Cancelled + ); + } + queue.restore_control(); + assert!(queue.complete(write).is_err()); + let new = queue.submit_write(b"tenant-a", b"fresh").unwrap(); + queue.complete(new).unwrap(); + } + + #[test] + fn control_restore_keeps_quarantine_and_hidden_slots() { + let mut queue = StorageQueue::open_simulated(8, 2).unwrap(); + let write = queue.submit_write(b"tenant-a", b"1234").unwrap(); + let slot = queue.complete(write).unwrap().slot.unwrap(); + let pending = queue.submit_write(b"tenant-a", b"5678").unwrap(); + queue.apply_fault(Fault::ControlPlaneLoss); + queue.restore_control(); + assert_eq!(queue.quarantined(), 1); + assert_eq!(queue.retained_bytes(), 8); + assert!(queue.submit_write(b"tenant-a", b"x").is_err()); + assert!(queue.submit_read(b"tenant-a", slot).is_err()); + assert!(queue.complete(pending).is_err()); + } + + #[test] + fn invalidation_is_tenant_scoped_and_queued_reads_cannot_escape_it() { + let mut queue = StorageQueue::open_simulated(8, 4).unwrap(); + let write = queue.submit_write(b"tenant-a", b"12345678").unwrap(); + let slot = queue.complete(write).unwrap().slot.unwrap(); + assert_eq!( + queue.invalidate(b"tenant-b", slot).unwrap_err(), + CacheError::Unauthorized + ); + assert!(queue.slab.slots[&slot].exposed); + assert_eq!( + queue.reclaim_invalidated(b"tenant-b", slot).unwrap_err(), + CacheError::Unauthorized + ); + assert!(matches!( + queue.reclaim_invalidated(b"tenant-a", slot), + Err(CacheError::NotReclaimable { .. }) + )); + let read = queue.submit_read(b"tenant-a", slot).unwrap(); + queue.invalidate(b"tenant-a", slot).unwrap(); + assert_eq!( + queue.reclaim_invalidated(b"tenant-a", slot).unwrap_err(), + CacheError::InUse { refs: 1 } + ); + assert!(queue.complete(read).is_err()); + queue.reclaim_invalidated(b"tenant-a", slot).unwrap(); + assert_eq!(queue.retained_bytes(), 0); + let next = queue.submit_write(b"tenant-b", b"new-data").unwrap(); + let new_slot = queue.complete(next).unwrap().slot.unwrap(); + assert_ne!(new_slot, slot); + assert!(queue.submit_read(b"tenant-a", slot).is_err()); + } + + #[test] + fn quarantined_reads_prevent_reclamation_until_attested_drained() { + let mut queue = StorageQueue::open_simulated(8, 2).unwrap(); + let write = queue.submit_write(b"tenant-a", b"12345678").unwrap(); + let slot = queue.complete(write).unwrap().slot.unwrap(); + let read = queue.submit_read(b"tenant-a", slot).unwrap(); + queue.apply_fault(Fault::NicFlap); + assert_eq!(queue.quarantined(), 1); + assert_eq!( + queue.reclaim_invalidated(b"tenant-a", slot).unwrap_err(), + CacheError::InUse { refs: 1 } + ); + queue.attest_drained(read).unwrap(); + queue.reclaim_invalidated(b"tenant-a", slot).unwrap(); + assert_eq!(queue.retained_bytes(), 0); + } + + #[test] + fn restart_fences_old_task_and_slot_ids_without_reusing_them() { + let mut queue = StorageQueue::open_simulated(16, 4).unwrap(); + let write = queue.submit_write(b"tenant-a", b"1234").unwrap(); + let old_slot = queue.complete(write).unwrap().slot.unwrap(); + let old_write = queue.submit_write(b"tenant-a", b"5678").unwrap(); + let old_read = queue.submit_read(b"tenant-a", old_slot).unwrap(); + queue.apply_fault(Fault::NicFlap); + queue.apply_fault(Fault::DpuRestart); + assert_eq!(queue.retained_bytes(), 0); + assert_eq!(queue.quarantined(), 0); + let new_write = queue.submit_write(b"tenant-b", b"new-data").unwrap(); + let new_slot = queue.complete(new_write).unwrap().slot.unwrap(); + assert!(new_write.0 > old_read.0); + assert!(new_slot > old_slot); + for task in [write, old_write, old_read] { + assert_eq!( + queue.complete(task).unwrap_err(), + CacheError::UnknownGeneration + ); + } + assert!(queue.submit_read(b"tenant-a", old_slot).is_err()); + assert_eq!(queue.retained_bytes(), 8); + } + + #[test] + fn queue_and_slot_metadata_have_independent_hard_bounds() { + assert!(StorageQueue::open_simulated(16, MAX_QUEUE_TASKS + 1).is_err()); + let mut queue = StorageQueue::open_simulated(8192, 2).unwrap(); + for _ in 0..MAX_SLAB_SLOTS - 1 { + let task = queue.submit_write(b"tenant-a", b"x").unwrap(); + queue.complete(task).unwrap(); + } + let last = queue.submit_write(b"tenant-a", b"x").unwrap(); + // Reserve the metadata slot when admitting the write, not at complete. + assert!(queue.submit_write(b"tenant-a", b"x").is_err()); + queue.quarantine(last).unwrap(); + assert!(queue.submit_write(b"tenant-a", b"x").is_err()); + queue.attest_drained(last).unwrap(); + let last = queue.submit_write(b"tenant-a", b"x").unwrap(); + queue.complete(last).unwrap(); + assert_eq!(queue.slab.slots.len(), MAX_SLAB_SLOTS); + assert!(queue.retained_bytes() < queue.slab.cap); + assert!(queue.submit_write(b"tenant-a", b"x").is_err()); + } + + #[test] + fn nonexistent_read_cannot_bind_to_a_future_slot() { + let mut queue = StorageQueue::open_simulated(8, 2).unwrap(); + assert_eq!( + queue.submit_read(b"tenant-a", 1).unwrap_err(), + CacheError::Storage { + reason: "clean miss" + } + ); + assert_eq!(queue.inflight(), 0); + assert_eq!(queue.next_task, 1); + let write = queue.submit_write(b"tenant-a", b"new-data").unwrap(); + let slot = queue.complete(write).unwrap().slot.unwrap(); + assert_eq!(slot, 1); + assert_eq!(queue.inflight(), 0); + let read = queue.submit_read(b"tenant-a", slot).unwrap(); + assert_eq!( + queue.complete(read).unwrap().payload.as_deref(), + Some(b"new-data".as_slice()) + ); + queue.invalidate(b"tenant-a", slot).unwrap(); + assert!(queue.submit_read(b"tenant-a", slot).is_err()); + assert_eq!(queue.inflight(), 0); + } + + #[test] + fn checksum_is_rechecked_after_read_admission() { + let mut queue = StorageQueue::open_simulated(8, 2).unwrap(); + let write = queue.submit_write(b"tenant-a", b"new-data").unwrap(); + let slot = queue.complete(write).unwrap().slot.unwrap(); + let read = queue.submit_read(b"tenant-a", slot).unwrap(); + queue.slab.slots.get_mut(&slot).unwrap().bytes[0] ^= 0xff; + assert_eq!( + queue.complete(read).unwrap_err(), + CacheError::ChecksumMismatch + ); + assert_eq!(queue.inflight(), 0); + assert!(!queue.slab.slots[&slot].exposed); + assert!(queue.submit_read(b"tenant-a", slot).is_err()); + } + + #[test] + fn malformed_inputs_and_exhausted_ids_do_not_consume_credit() { + let mut queue = StorageQueue::open_simulated(8, 2).unwrap(); + for tenant in [&b""[..], &[b'a'; 129][..]] { + assert!(queue.submit_write(tenant, b"x").is_err()); + assert!(queue.submit_read(tenant, 1).is_err()); + } + assert!(queue.submit_write(b"tenant-a", b"").is_err()); + assert!(queue.submit_write(b"tenant-a", b"123456789").is_err()); + queue.next_task = u64::MAX; + assert!(queue.submit_write(b"tenant-a", b"x").is_err()); + assert_eq!(queue.inflight(), 0); + assert_eq!(queue.retained_bytes(), 0); + queue.next_task = 1; + queue.slab.next = u64::MAX; + assert!(queue.submit_write(b"tenant-a", b"x").is_err()); + assert_eq!(queue.retained_bytes(), 0); + queue.slab.next = u64::MAX - 1; + let last = queue.submit_write(b"tenant-a", b"x").unwrap(); + assert!(queue.submit_write(b"tenant-a", b"y").is_err()); + assert_eq!(queue.retained_bytes(), 1); + assert_eq!(queue.complete(last).unwrap().slot, Some(u64::MAX - 1)); + assert!(queue.submit_write(b"tenant-a", b"z").is_err()); + } } diff --git a/crates/protocol/src/auth.rs b/crates/protocol/src/auth.rs index 2c41feb..4a3ae74 100644 --- a/crates/protocol/src/auth.rs +++ b/crates/protocol/src/auth.rs @@ -2,10 +2,11 @@ // SPDX-License-Identifier: Apache-2.0 use gpucache_core_types::{CacheError, TenantId, TrustDomain}; +use std::fmt; /// Minimum tenant gate. This is not mTLS and not a production authenticator. /// The fingerprint is a hash of a credential, never the credential itself. -#[derive(Clone, Debug, PartialEq, Eq)] +#[derive(Clone, PartialEq, Eq)] pub struct AuthContext { pub tenant: TenantId, pub trust_domain: TrustDomain, @@ -27,9 +28,14 @@ impl AuthContext { credential_fingerprint, }) } + + /// Revalidate at trust boundaries because fields remain public for V1 callers. + pub fn validate(&self) -> Result<(), CacheError> { + validate_fingerprint(&self.credential_fingerprint) + } } -#[derive(Clone, Debug, PartialEq, Eq)] +#[derive(Clone, PartialEq, Eq)] pub struct TenantPolicy { pub tenant: TenantId, pub trust_domain: TrustDomain, @@ -51,6 +57,39 @@ impl TenantPolicy { credential_fingerprint, }) } + + /// Revalidate policy values assembled without the checked constructor. + pub fn validate(&self) -> Result<(), CacheError> { + validate_fingerprint(&self.credential_fingerprint) + } +} + +fn validate_fingerprint(fingerprint: &[u8; 32]) -> Result<(), CacheError> { + if fingerprints_equal(fingerprint, &[0; 32]) { + Err(CacheError::Unauthorized) + } else { + Ok(()) + } +} + +impl fmt::Debug for AuthContext { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("AuthContext") + .field("tenant", &self.tenant) + .field("trust_domain", &self.trust_domain) + .field("credential_fingerprint", &"[REDACTED]") + .finish() + } +} + +impl fmt::Debug for TenantPolicy { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("TenantPolicy") + .field("tenant", &self.tenant) + .field("trust_domain", &self.trust_domain) + .field("credential_fingerprint", &"[REDACTED]") + .finish() + } } pub fn fingerprints_equal(a: &[u8; 32], b: &[u8; 32]) -> bool { @@ -60,3 +99,38 @@ pub fn fingerprints_equal(a: &[u8; 32], b: &[u8; 32]) -> bool { } diff == 0 } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn credential_fingerprints_are_redacted_from_debug_output() { + let tenant = TenantId::new(b"tenant-a".to_vec()).unwrap(); + let trust = TrustDomain::new(b"trust-a".to_vec()).unwrap(); + let ctx = AuthContext::new(tenant.clone(), trust.clone(), [173; 32]).unwrap(); + let policy = TenantPolicy::new(tenant, trust, [173; 32]).unwrap(); + for output in [format!("{ctx:?}"), format!("{policy:?}")] { + assert!(output.contains("[REDACTED]")); + assert!(!output.contains("173")); + } + } + + #[test] + fn revalidation_rejects_public_field_constructor_bypass() { + let tenant = TenantId::new(b"tenant-a".to_vec()).unwrap(); + let trust = TrustDomain::new(b"trust-a".to_vec()).unwrap(); + let ctx = AuthContext { + tenant: tenant.clone(), + trust_domain: trust.clone(), + credential_fingerprint: [0; 32], + }; + let policy = TenantPolicy { + tenant, + trust_domain: trust, + credential_fingerprint: [0; 32], + }; + assert_eq!(ctx.validate(), Err(CacheError::Unauthorized)); + assert_eq!(policy.validate(), Err(CacheError::Unauthorized)); + } +} diff --git a/crates/protocol/src/codec.rs b/crates/protocol/src/codec.rs index f485338..b120883 100644 --- a/crates/protocol/src/codec.rs +++ b/crates/protocol/src/codec.rs @@ -50,6 +50,11 @@ pub struct CommittedExport { } pub fn encode_request(auth: &AuthContext, body: &RequestBody) -> Result, CacheError> { + if body.len() > MAX_ITEMS { + return Err(CacheError::QuotaExceeded { + kind: gpucache_core_types::QuotaKind::BatchItems, + }); + } let mut w = ByteWriter::with_capacity(128); w.raw(WIRE_MAGIC); w.u16(WIRE_VERSION); @@ -215,7 +220,7 @@ pub fn decode_request(bytes: &[u8]) -> Result<(AuthContext, RequestBody), CacheE } pub fn encode_hello(versions: &[u16]) -> Result, CacheError> { - if versions.len() > 32 { + if versions.is_empty() || versions.len() > 32 { return Err(CacheError::Malformed { reason: "hello versions", }); @@ -343,6 +348,7 @@ pub fn request_op(body: &RequestBody) -> u16 { } fn write_auth(w: &mut ByteWriter, auth: &AuthContext) -> Result<(), CacheError> { + auth.validate()?; w.bytes(auth.tenant.as_bytes())?; w.bytes(auth.trust_domain.as_bytes())?; w.raw(&auth.credential_fingerprint); @@ -357,6 +363,7 @@ fn read_auth(r: &mut ByteReader<'_>) -> Result { } fn write_header(w: &mut ByteWriter, header: &ItemHeader) -> Result<(), CacheError> { + RequestId::new(header.request_id.0)?; w.raw(&header.request_id.0); w.u64(header.epoch); w.u64(header.deadline_ms); @@ -374,6 +381,7 @@ fn read_header(r: &mut ByteReader<'_>) -> Result { } fn write_identity(w: &mut ByteWriter, id: &CacheIdentity) -> Result<(), CacheError> { + id.validate_wire()?; w.bytes(&id.to_bytes()?) } @@ -429,6 +437,7 @@ fn write_opt_request(w: &mut ByteWriter, value: Option) -> Result<(), match value { None => w.u8(0), Some(id) => { + RequestId::new(id.0)?; w.u8(1); w.raw(&id.0); } @@ -590,4 +599,72 @@ mod tests { assert_eq!(back, body); } } + + #[test] + fn encoder_rejects_public_auth_and_request_id_constructor_bypasses() { + let mut invalid_auth = auth(); + invalid_auth.credential_fingerprint = [0; 32]; + let body = RequestBody::Lookup(vec![LookupItem { + header: header(1), + identity: synthetic_identity(1, 1), + }]); + assert_eq!( + encode_request(&invalid_auth, &body), + Err(CacheError::Unauthorized) + ); + let mut bad_header = header(1); + bad_header.request_id = RequestId([0; 16]); + let body = RequestBody::Lookup(vec![LookupItem { + header: bad_header, + identity: synthetic_identity(1, 1), + }]); + assert!(matches!( + encode_request(&auth(), &body), + Err(CacheError::Malformed { .. }) + )); + } + + #[test] + fn encoder_does_not_emit_oversized_batches_or_empty_negotiation() { + let item = LookupItem { + header: header(1), + identity: synthetic_identity(1, 1), + }; + assert_eq!( + encode_request(&auth(), &RequestBody::Lookup(vec![item; MAX_ITEMS + 1])), + Err(CacheError::QuotaExceeded { + kind: gpucache_core_types::QuotaKind::BatchItems, + }) + ); + assert!(encode_hello(&[]).is_err()); + } + + #[test] + fn export_does_not_accept_invalid_identity_with_a_valid_payload_checksum() { + let payload = b"valid-payload".to_vec(); + let mut identity = synthetic_identity(1, 1); + identity.valid_length = 0; + let frame = CommittedExport { + identity: identity.clone(), + source_generation: 1, + checksum: content_checksum(&payload), + payload: payload.clone(), + }; + assert!(encode_export(&frame).is_err()); + + // Build a hostile frame without the checked encoder. + let mut w = ByteWriter::new(); + w.raw(EXPORT_MAGIC); + w.u16(EXPORT_VERSION); + w.bytes(&identity.to_bytes().unwrap()).unwrap(); + w.u64(1); + w.raw(&content_checksum(&payload).0); + w.bytes(&payload).unwrap(); + assert_eq!( + decode_export(&w.finish()), + Err(CacheError::IdentityInvalid { + reason: "valid length" + }) + ); + } } diff --git a/crates/protocol/src/harden.rs b/crates/protocol/src/harden.rs index 9786486..fc3c0f2 100644 --- a/crates/protocol/src/harden.rs +++ b/crates/protocol/src/harden.rs @@ -12,7 +12,8 @@ use std::collections::VecDeque; use gpucache_core_types::{ - content_checksum, CacheError, Checksum, TenantId, TrustDomain, HARD_BYTES_CAP, + content_checksum, ByteWriter, CacheError, Checksum, QuotaKind, TenantId, TrustDomain, + HARD_BYTES_CAP, }; use crate::auth::{fingerprints_equal, AuthContext}; @@ -86,6 +87,7 @@ impl Acl { } pub fn grant(&mut self, ctx: AuthContext, write: bool) -> Result<(), CacheError> { + ctx.validate()?; if let Some(entry) = self.entries.iter_mut().find(|entry| { entry.tenant == ctx.tenant && entry.trust == ctx.trust_domain @@ -109,6 +111,7 @@ impl Acl { } pub fn check(&self, ctx: &AuthContext, write: bool) -> Result<(), CacheError> { + ctx.validate()?; let mut allowed = false; for entry in &self.entries { let same = entry.tenant == ctx.tenant @@ -129,7 +132,14 @@ impl Acl { pub struct ReplayGuard { skew_ms: u64, capacity: usize, - seen: VecDeque<([u8; 32], [u8; 16])>, + seen: VecDeque, + last_now_ms: Option, +} + +struct ReplayEntry { + scope: [u8; 32], + nonce: [u8; 16], + observed_ms: u64, } impl ReplayGuard { @@ -141,13 +151,15 @@ impl ReplayGuard { skew_ms, capacity, seen: VecDeque::new(), + last_now_ms: None, }) } /// Accept `nonce` once inside the skew window. /// - /// The window is bounded. After `capacity` newer nonces, the oldest nonce - /// can be presented again. That is a window, not a permanent replay cache. + /// Entries are retained for the whole window, even when the guard is full. + /// Capacity exhaustion rejects new admissions instead of enabling replay. + /// `now_ms` must come from the server clock, never an untrusted request. pub fn observe( &mut self, tenant: &[u8], @@ -162,17 +174,47 @@ impl ReplayGuard { return Err(CacheError::Unauthorized); } let tenant_key = content_checksum(tenant).0; + self.observe_scope(tenant_key, nonce, now_ms, issued_ms) + } + + fn observe_scope( + &mut self, + scope: [u8; 32], + nonce: &[u8; 16], + now_ms: u64, + issued_ms: u64, + ) -> Result<(), CacheError> { + if *nonce == [0; 16] { + return Err(CacheError::Malformed { reason: "replay" }); + } + if self.last_now_ms.is_some_and(|last| now_ms < last) + || issued_ms > now_ms + || now_ms - issued_ms > self.skew_ms + { + return Err(CacheError::Unauthorized); + } + self.last_now_ms = Some(now_ms); + // Subtraction cannot overflow: the clock is nondecreasing and every + // retained observation came from an earlier accepted server time. + self.seen + .retain(|entry| now_ms - entry.observed_ms <= self.skew_ms); if self .seen .iter() - .any(|(seen_tenant, seen_nonce)| *seen_tenant == tenant_key && seen_nonce == nonce) + .any(|entry| entry.scope == scope && entry.nonce == *nonce) { return Err(CacheError::IdempotencyConflict); } if self.seen.len() == self.capacity { - self.seen.pop_front(); + return Err(CacheError::QuotaExceeded { + kind: QuotaKind::IdempotencyTable, + }); } - self.seen.push_back((tenant_key, *nonce)); + self.seen.push_back(ReplayEntry { + scope, + nonce: *nonce, + observed_ms: now_ms, + }); Ok(()) } } @@ -198,6 +240,11 @@ pub fn seal(key_id: u64, generation: u64, payload: &[u8]) -> Result Result<(), CacheError> { + if payload.is_empty() || payload.len() > HARD_BYTES_CAP { + return Err(CacheError::Malformed { + reason: "integrity", + }); + } let actual = tag(seal.key_id, seal.generation, payload); if actual != seal.checksum { return Err(CacheError::ChecksumMismatch); @@ -234,6 +281,8 @@ pub struct SecretRing { log: SecurityLog, } +const MAX_SECRET_GENERATIONS: usize = 1024; + impl SecretRing { pub fn open(fingerprint: [u8; 32], audit_cap: usize) -> Result { if fingerprint == [0u8; 32] { @@ -244,7 +293,7 @@ impl SecretRing { next_id: 1, log: SecurityLog::new(audit_cap)?, }; - let id = ring.push(fingerprint, GenState::Current); + let id = ring.push(fingerprint, GenState::Current)?; ring.log.push(AuditEvent::Rotate { generation: id }); Ok(ring) } @@ -256,16 +305,19 @@ impl SecretRing { if self .items .iter() - .any(|item| item.fingerprint == fingerprint && item.state != GenState::Retired) + .any(|item| fingerprints_equal(&item.fingerprint, &fingerprint)) { return Err(CacheError::IdempotencyConflict); } + // Preflight before demoting the current credential: a rejected rotation + // must leave the previous write authority and audit history unchanged. + self.check_capacity()?; for item in &mut self.items { if item.state == GenState::Current { item.state = GenState::Overlap; } } - let id = self.push(fingerprint, GenState::Current); + let id = self.push(fingerprint, GenState::Current)?; self.log.push(AuditEvent::Rotate { generation: id }); Ok(id) } @@ -311,15 +363,30 @@ impl SecretRing { Err(CacheError::Unauthorized) } - fn push(&mut self, fingerprint: [u8; 32], state: GenState) -> u64 { + fn check_capacity(&self) -> Result<(), CacheError> { + if self.items.len() >= MAX_SECRET_GENERATIONS { + return Err(CacheError::QuotaExceeded { + kind: QuotaKind::IdempotencyTable, + }); + } + self.next_id + .checked_add(1) + .ok_or(CacheError::FencingRejected { + reason: "secret generation exhausted", + })?; + Ok(()) + } + + fn push(&mut self, fingerprint: [u8; 32], state: GenState) -> Result { + self.check_capacity()?; let id = self.next_id; - self.next_id = self.next_id.saturating_add(1); + self.next_id += 1; self.items.push(Gen { id, fingerprint, state, }); - id + Ok(id) } } @@ -342,8 +409,15 @@ pub fn authorize_request( log.push(AuditEvent::Deny); return Err(err); } - if let Err(err) = replay.observe( - ctx.tenant.as_bytes(), + // Scope replay by the complete authenticated principal, with unambiguous + // lengths. The same tenant label in another trust domain is independent. + let mut scope = ByteWriter::new(); + scope.raw(b"gpucache-replay-v1"); + scope.bytes(ctx.tenant.as_bytes())?; + scope.bytes(ctx.trust_domain.as_bytes())?; + scope.raw(&ctx.credential_fingerprint); + if let Err(err) = replay.observe_scope( + content_checksum(&scope.finish()).0, &stamp.nonce, stamp.now_ms, stamp.issued_ms, @@ -475,4 +549,147 @@ mod tests { let events: Vec<_> = log.events().copied().collect(); assert_eq!(events, vec![AuditEvent::Replay, AuditEvent::Deny]); } + + #[test] + fn full_replay_window_never_evicts_a_live_nonce() { + let mut guard = ReplayGuard::new(100, 1).unwrap(); + guard.observe(b"tenant-a", &[1; 16], 10, 10).unwrap(); + assert_eq!( + guard.observe(b"tenant-a", &[2; 16], 11, 11), + Err(CacheError::QuotaExceeded { + kind: QuotaKind::IdempotencyTable, + }) + ); + assert_eq!( + guard.observe(b"tenant-a", &[1; 16], 110, 10), + Err(CacheError::IdempotencyConflict) + ); + guard.observe(b"tenant-a", &[2; 16], 111, 111).unwrap(); + assert_eq!(guard.seen.len(), 1); + assert_eq!( + guard.observe(b"tenant-a", &[1; 16], 111, 10), + Err(CacheError::Unauthorized) + ); + } + + #[test] + fn replay_window_rejects_clock_rollback_without_losing_entries() { + let mut guard = ReplayGuard::new(10, 2).unwrap(); + guard.observe(b"tenant-a", &[1; 16], 100, 100).unwrap(); + assert_eq!( + guard.observe(b"tenant-a", &[2; 16], 99, 99), + Err(CacheError::Unauthorized) + ); + assert_eq!( + guard.observe(b"tenant-a", &[1; 16], 100, 100), + Err(CacheError::IdempotencyConflict) + ); + } + + #[test] + fn replay_window_does_not_overflow_near_the_clock_limit() { + let mut guard = ReplayGuard::new(10, 1).unwrap(); + guard + .observe(b"tenant-a", &[1; 16], u64::MAX - 1, u64::MAX - 1) + .unwrap(); + assert_eq!( + guard.observe(b"tenant-a", &[1; 16], u64::MAX, u64::MAX - 1), + Err(CacheError::IdempotencyConflict) + ); + } + + #[test] + fn authenticated_replay_scopes_include_trust_and_credential() { + let a = ctx(b"tenant-a", 1); + let mut b = a.clone(); + b.trust_domain = TrustDomain::new(b"other-trust".to_vec()).unwrap(); + let c = ctx(b"tenant-a", 2); + let mut acl = Acl::new(3).unwrap(); + for principal in [&a, &b, &c] { + acl.grant(principal.clone(), true).unwrap(); + } + let mut replay = ReplayGuard::new(20, 3).unwrap(); + let mut log = SecurityLog::new(8).unwrap(); + let stamp = RequestStamp { + write: true, + nonce: [1; 16], + now_ms: 10, + issued_ms: 10, + }; + for principal in [&a, &b, &c] { + authorize_request(&acl, &mut replay, &mut log, principal, stamp).unwrap(); + } + assert_eq!( + authorize_request(&acl, &mut replay, &mut log, &a, stamp), + Err(CacheError::IdempotencyConflict) + ); + } + + #[test] + fn zero_credentials_cannot_be_granted_via_public_fields() { + let mut zero = ctx(b"tenant-a", 1); + zero.credential_fingerprint = [0; 32]; + let mut acl = Acl::new(1).unwrap(); + assert_eq!(acl.grant(zero.clone(), true), Err(CacheError::Unauthorized)); + assert_eq!(acl.check(&zero, false), Err(CacheError::Unauthorized)); + } + + #[test] + fn failed_rotation_preserves_current_write_authority() { + let mut ring = SecretRing::open([1; 32], 8).unwrap(); + ring.next_id = u64::MAX; + let before_log = ring.log.len(); + assert_eq!( + ring.rotate([2; 32]), + Err(CacheError::FencingRejected { + reason: "secret generation exhausted", + }) + ); + assert_eq!(ring.allow_write(&[1; 32]), Ok(1)); + assert_eq!(ring.allow_read(&[2; 32]), Err(CacheError::Unauthorized)); + assert_eq!(ring.log.len(), before_log); + } + + #[test] + fn retired_credentials_cannot_be_resurrected_by_rotation() { + let mut ring = SecretRing::open([1; 32], 8).unwrap(); + ring.rotate([2; 32]).unwrap(); + ring.retire(1).unwrap(); + assert_eq!(ring.rotate([1; 32]), Err(CacheError::IdempotencyConflict)); + assert_eq!(ring.allow_read(&[1; 32]), Err(CacheError::Unauthorized)); + assert_eq!(ring.allow_write(&[2; 32]), Ok(2)); + } + + #[test] + fn secret_history_is_bounded_and_full_rotation_is_atomic() { + let mut ring = SecretRing::open([1; 32], 8).unwrap(); + ring.items.resize_with(MAX_SECRET_GENERATIONS, || Gen { + id: 0, + fingerprint: [3; 32], + state: GenState::Retired, + }); + assert_eq!( + ring.rotate([2; 32]), + Err(CacheError::QuotaExceeded { + kind: QuotaKind::IdempotencyTable, + }) + ); + assert_eq!(ring.allow_write(&[1; 32]), Ok(1)); + assert_eq!(ring.items.len(), MAX_SECRET_GENERATIONS); + } + + #[test] + fn seal_verification_rejects_empty_payloads_even_with_matching_tags() { + let forged = IntegritySeal { + key_id: 1, + generation: 1, + checksum: tag(1, 1, &[]), + }; + assert_eq!( + open_seal(&forged, &[]), + Err(CacheError::Malformed { + reason: "integrity", + }) + ); + } } diff --git a/crates/storage/src/file.rs b/crates/storage/src/file.rs index 911811c..dc38253 100644 --- a/crates/storage/src/file.rs +++ b/crates/storage/src/file.rs @@ -72,7 +72,11 @@ impl FileStore { let manifest_existed = root.join("MANIFEST").is_file(); let previous_epoch = read_manifest(&root)?.unwrap_or(1); let epoch = if manifest_existed { - previous_epoch.saturating_add(1).max(1) + previous_epoch + .checked_add(1) + .ok_or(CacheError::FencingRejected { + reason: "epoch exhausted", + })? } else { 1 }; diff --git a/crates/storage/src/memory.rs b/crates/storage/src/memory.rs index b9930f3..8c98d67 100644 --- a/crates/storage/src/memory.rs +++ b/crates/storage/src/memory.rs @@ -150,6 +150,7 @@ impl MemoryStore { limits.validate()?; let mut map = HashMap::with_capacity(policies.len()); for policy in policies { + policy.validate()?; if map .insert(policy.tenant.as_bytes().to_vec(), policy) .is_some() @@ -1405,6 +1406,7 @@ impl MemoryStore { auth: &AuthContext, identity: Option<&CacheIdentity>, ) -> Result<(), CacheError> { + auth.validate()?; let policy = self .policies .get(auth.tenant.as_bytes()) diff --git a/crates/storage/src/tests.rs b/crates/storage/src/tests.rs index fefac4f..2e01c60 100644 --- a/crates/storage/src/tests.rs +++ b/crates/storage/src/tests.rs @@ -15,6 +15,36 @@ use crate::{MemoryStore, ReclaimRequest}; const FP: [u8; 32] = [9u8; 32]; +#[test] +fn direct_policy_construction_cannot_enable_a_zero_credential() { + let identity = synthetic_identity(1, 1); + let policy = TenantPolicy { + tenant: identity.tenant, + trust_domain: identity.trust_domain, + credential_fingerprint: [0; 32], + }; + assert!(matches!( + MemoryStore::new(limits(), vec![policy]), + Err(CacheError::Unauthorized) + )); +} + +#[test] +fn mutated_auth_context_is_rejected_at_the_store_boundary() { + let (mut store, mut auth) = open(None); + auth.credential_fingerprint = [0; 32]; + let reply = store.lookup( + &auth, + 0, + vec![LookupItem { + header: hdr(&store, 1, 0), + identity: synthetic_identity(1, 1), + }], + ); + assert_eq!(reply, Err(CacheError::Unauthorized)); + assert_eq!(store.stats().retained_bytes, 0); +} + fn limits() -> ResourceLimits { ResourceLimits::dev_defaults() } diff --git a/crates/storage/tests/file_backend.rs b/crates/storage/tests/file_backend.rs index 7d90fbf..3c8dc39 100644 --- a/crates/storage/tests/file_backend.rs +++ b/crates/storage/tests/file_backend.rs @@ -154,6 +154,24 @@ fn empty_policy_is_refused() { )); } +#[test] +fn exhausted_restart_epoch_is_rejected_without_reusing_the_old_epoch() { + let dir = TempDir::new(); + let (policy, _) = policy(1, FP_A); + drop(FileStore::open(&dir.path, limits(64, 64), vec![policy.clone()]).unwrap()); + let path = dir.path.join("MANIFEST"); + let mut manifest = fs::read(&path).unwrap(); + manifest[6..14].copy_from_slice(&u64::MAX.to_le_bytes()); + fs::write(&path, &manifest).unwrap(); + assert!(matches!( + must_err(FileStore::open(&dir.path, limits(64, 64), vec![policy])), + CacheError::FencingRejected { + reason: "epoch exhausted" + } + )); + assert_eq!(fs::read(path).unwrap(), manifest); +} + #[test] fn restart_reloads_committed_bytes_and_bumps_epoch() { let dir = TempDir::new(); diff --git a/crates/transport/src/client.rs b/crates/transport/src/client.rs index 08a08b0..1b3f2be 100644 --- a/crates/transport/src/client.rs +++ b/crates/transport/src/client.rs @@ -5,12 +5,12 @@ use std::net::{SocketAddr, TcpStream}; use std::sync::Arc; use std::time::Duration; -use gpucache_core_types::CacheError; +use gpucache_core_types::{CacheError, RequestId}; use gpucache_protocol::{ - decode_hello_ack, decode_reply, encode_hello, encode_request, AuthContext, BatchReply, - CommitItem, CommitOutcome, GetHit, GetItem, InvalidateItem, InvalidateOutcome, LookupItem, - LookupOutcome, ReleaseCancelItem, ReleaseOutcome, ReplyBody, RequestBody, ReserveOutcome, - ReservePutItem, WIRE_VERSION, + decode_hello_ack, decode_reply, encode_hello, encode_request, request_op, AuthContext, + BatchReply, CommitItem, CommitOutcome, GetHit, GetItem, InvalidateItem, InvalidateOutcome, + LookupItem, LookupOutcome, ReleaseCancelItem, ReleaseOutcome, ReplyBody, RequestBody, + ReserveOutcome, ReservePutItem, WIRE_VERSION, }; use crate::clock::Clock; @@ -144,34 +144,86 @@ impl TcpClient { fn call( &mut self, request: RequestBody, - map: impl FnOnce(ReplyBody) -> Result, - ) -> Result { + map: impl FnOnce(ReplyBody) -> Result, CacheError>, + ) -> Result, CacheError> { if !self.usable { return Err(CacheError::Storage { reason: "transport io", }); } let bytes = encode_request(&self.auth, &request)?; + let expected_op = request_op(&request); + let expected_ids = request_ids(&request); // No request correlation exists at the envelope layer. After an // incomplete exchange, a late reply must never answer a later call. self.usable = false; write_envelope(&mut self.stream, 0, &bytes, self.max_frame)?; let envelope = read_envelope(&mut self.stream, self.pool.checkout()?, self.max_frame)?; - self.clock.observe(envelope.server_now); match decode_reply(envelope.body())? { - ReplyBody::Failed { error, .. } => { - self.usable = true; + ReplyBody::Failed { + op, + store_epoch, + error, + } => { + self.check_epoch(store_epoch)?; + if op == expected_op { + self.clock.observe(envelope.server_now); + self.usable = true; + } else if op != 0 { + return Err(CacheError::Malformed { reason: "reply op" }); + } + // Op zero is a server-side decode failure with no reliable + // request association. Preserve its error but keep the stream + // poisoned so it cannot answer a subsequent request. Err(error) } other => { - let result = map(other); - if result.is_ok() { - self.usable = true; + let batch = map(other)?; + self.check_epoch(batch.store_epoch)?; + if batch.items.len() != expected_ids.len() { + return Err(CacheError::Malformed { + reason: "reply batch length", + }); + } + if batch + .items + .iter() + .zip(&expected_ids) + .any(|(item, expected)| item.request_id != *expected) + { + return Err(CacheError::Malformed { + reason: "reply request id", + }); } - result + self.clock.observe(envelope.server_now); + self.usable = true; + Ok(batch) } } } + + fn check_epoch(&self, received: u64) -> Result<(), CacheError> { + if received != self.epoch { + return Err(CacheError::EpochMismatch { + store: received, + request: self.epoch, + }); + } + Ok(()) + } +} + +fn request_ids(request: &RequestBody) -> Vec { + match request { + RequestBody::Lookup(items) => items.iter().map(|item| item.header.request_id).collect(), + RequestBody::ReservePut(items) => items.iter().map(|item| item.header.request_id).collect(), + RequestBody::Commit(items) => items.iter().map(|item| item.header.request_id).collect(), + RequestBody::Get(items) => items.iter().map(|item| item.header.request_id).collect(), + RequestBody::ReleaseCancel(items) => { + items.iter().map(|item| item.header.request_id).collect() + } + RequestBody::Invalidate(items) => items.iter().map(|item| item.header.request_id).collect(), + } } #[cfg(test)] @@ -182,6 +234,213 @@ mod tests { use std::net::TcpListener; use std::sync::mpsc; + fn lookup_item(id: u128) -> LookupItem { + LookupItem { + header: ItemHeader { + request_id: RequestId::from_counter(id).unwrap(), + epoch: 1, + deadline_ms: 1000, + }, + identity: synthetic_identity(1, 1), + } + } + + fn lookup_reply(epoch: u64, ids: &[u128]) -> ReplyBody { + ReplyBody::Lookup(BatchReply { + store_epoch: epoch, + items: ids + .iter() + .map(|id| ItemResult { + request_id: RequestId::from_counter(*id).unwrap(), + body: Ok(LookupOutcome::Miss { + reason: gpucache_protocol::MissReason::Absent, + }), + }) + .collect(), + }) + } + + fn fake_server(replies: Vec>) -> (TcpClient, std::thread::JoinHandle<()>) { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let server = std::thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + stream + .set_write_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let pool = Arc::new(Pool::new(1)); + drop(read_envelope(&mut stream, pool.checkout().unwrap(), 4096).unwrap()); + write_envelope(&mut stream, 1, &encode_hello_ack(WIRE_VERSION, 1), 4096).unwrap(); + for reply in replies { + drop(read_envelope(&mut stream, pool.checkout().unwrap(), 4096).unwrap()); + write_envelope(&mut stream, 2, &reply, 4096).unwrap(); + } + }); + let identity = synthetic_identity(1, 1); + let auth = AuthContext::new(identity.tenant, identity.trust_domain, [9; 32]).unwrap(); + let client = TcpClient::connect(addr, auth, Duration::from_secs(5), 4096).unwrap(); + (client, server) + } + + fn assert_poisoned_reply(reply: Vec, ids: &[u128], expected: CacheError) { + let (mut client, server) = fake_server(vec![reply]); + let items = ids.iter().map(|id| lookup_item(*id)).collect(); + assert_eq!(client.lookup(items), Err(expected)); + assert!(!client.usable); + assert_eq!( + client.lookup(vec![lookup_item(99)]), + Err(CacheError::Storage { + reason: "transport io" + }) + ); + server.join().unwrap(); + } + + #[test] + fn wrong_request_id_poisons_the_connection() { + assert_poisoned_reply( + encode_reply(&lookup_reply(1, &[9])).unwrap(), + &[1], + CacheError::Malformed { + reason: "reply request id", + }, + ); + } + + #[test] + fn reordered_batch_items_poison_the_connection() { + assert_poisoned_reply( + encode_reply(&lookup_reply(1, &[2, 1])).unwrap(), + &[1, 2], + CacheError::Malformed { + reason: "reply request id", + }, + ); + } + + #[test] + fn missing_and_extra_batch_items_poison_the_connection() { + for (reply_ids, request_ids) in [(vec![1], vec![1, 2]), (vec![1, 2], vec![1])] { + assert_poisoned_reply( + encode_reply(&lookup_reply(1, &reply_ids)).unwrap(), + &request_ids, + CacheError::Malformed { + reason: "reply batch length", + }, + ); + } + } + + #[test] + fn truncated_reply_body_poisons_the_connection() { + let mut reply = encode_reply(&lookup_reply(1, &[1])).unwrap(); + reply.pop(); + assert_poisoned_reply( + reply, + &[1], + CacheError::Malformed { + reason: "truncated buffer", + }, + ); + } + + #[test] + fn wrong_epoch_poisons_success_and_failed_replies() { + let replies = [ + lookup_reply(2, &[1]), + ReplyBody::Failed { + op: gpucache_protocol::OP_LOOKUP, + store_epoch: 2, + error: CacheError::Unauthorized, + }, + ]; + for reply in replies { + assert_poisoned_reply( + encode_reply(&reply).unwrap(), + &[1], + CacheError::EpochMismatch { + store: 2, + request: 1, + }, + ); + } + } + + #[test] + fn wrong_operation_poisons_success_and_failed_replies() { + let replies = [ + ReplyBody::Reserve(BatchReply { + store_epoch: 1, + items: Vec::new(), + }), + ReplyBody::Failed { + op: gpucache_protocol::OP_GET, + store_epoch: 1, + error: CacheError::Unauthorized, + }, + ]; + for reply in replies { + assert_poisoned_reply( + encode_reply(&reply).unwrap(), + &[1], + CacheError::Malformed { reason: "reply op" }, + ); + } + } + + #[test] + fn unassociated_server_decode_error_is_returned_without_reusing_the_stream() { + assert_poisoned_reply( + encode_reply(&ReplyBody::Failed { + op: 0, + store_epoch: 1, + error: CacheError::Unauthorized, + }) + .unwrap(), + &[1], + CacheError::Unauthorized, + ); + } + + #[test] + fn correlated_batch_and_item_errors_keep_the_stream_synchronized() { + let failed = ReplyBody::Failed { + op: gpucache_protocol::OP_LOOKUP, + store_epoch: 1, + error: CacheError::Unauthorized, + }; + let item_error = ReplyBody::Lookup(BatchReply { + store_epoch: 1, + items: vec![ItemResult { + request_id: RequestId::from_counter(2).unwrap(), + body: Err(CacheError::TimedOut), + }], + }); + let replies = [failed, item_error, lookup_reply(1, &[3])] + .iter() + .map(|reply| encode_reply(reply).unwrap()) + .collect(); + let (mut client, server) = fake_server(replies); + assert_eq!( + client.lookup(vec![lookup_item(1)]), + Err(CacheError::Unauthorized) + ); + assert!(client.usable); + assert_eq!( + client.lookup(vec![lookup_item(2)]).unwrap().items[0].body, + Err(CacheError::TimedOut) + ); + assert!(client.usable); + assert!(matches!( + client.lookup(vec![lookup_item(3)]).unwrap().items[0].body, + Ok(LookupOutcome::Miss { .. }) + )); + server.join().unwrap(); + } + #[test] fn timed_out_connection_cannot_consume_a_late_reply_for_another_call() { let listener = TcpListener::bind("127.0.0.1:0").unwrap(); From dc4b539738f4c0cbaa59dad2b0f1c651e9a4e71b Mon Sep 17 00:00:00 2001 From: loverustfs Date: Sat, 17 Oct 2026 15:05:23 +0800 Subject: [PATCH 2/4] test: add fail-closed VM regression reports and CI gates --- .github/workflows/ci.yml | 36 +-- .gitignore | 2 + scripts/ci/check_contract.py | 93 ++++++++ scripts/ci/tests/test_vm_regression.py | 289 ++++++++++++++++++++++++ scripts/ci/vm_regression.py | 295 +++++++++++++++++++++++++ 5 files changed, 691 insertions(+), 24 deletions(-) create mode 100644 scripts/ci/check_contract.py create mode 100644 scripts/ci/tests/test_vm_regression.py create mode 100644 scripts/ci/vm_regression.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 60303af..8143af3 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -14,28 +14,16 @@ jobs: - uses: dtolnay/rust-toolchain@stable with: components: clippy, rustfmt - - name: Format - run: cargo fmt --all -- --check - - name: Clippy - run: cargo clippy --workspace --all-targets -- -D warnings - - name: Test - run: cargo test --workspace --all-targets - - name: Fallback demo - run: cargo run -p gpucache-transport --bin fault-demo --quiet - - name: Paired baseline stub - run: cargo run -p gpucache-bench --bin paired-baseline --quiet - - name: Shard and cold demos - run: | - cargo run -p gpucache-transport --bin shard-demo --quiet - cargo run -p gpucache-cold --bin cold-demo --quiet - - name: V3 DPU contract and paired gate - run: | - cargo run -p gpucache-dpu --bin dpu-demo --quiet - cargo run -p gpucache-bench --bin v3-paired --quiet - - name: Optional RDMA feature still skips hardware - run: cargo test -p gpucache-transport --all-targets --features rdma-nixl - - name: Optional DPU feature still skips hardware - run: cargo test -p gpucache-dpu --all-targets --features dpu-bf3 + - name: Reproducible CPU VM regression + run: python3 scripts/ci/vm_regression.py --profile cpu --output artifacts/vm-regression + - name: Retain regression evidence + if: always() + uses: actions/upload-artifact@v4 + with: + name: cpu-vm-regression-x86_64 + path: artifacts/vm-regression/ + if-no-files-found: error + retention-days: 14 - name: Reject accelerator linkage and erasure coding run: | python3 - <<'PY' @@ -83,7 +71,7 @@ jobs: - name: Test run: | echo "aarch64 cargo test is a CPU build. It is not a BlueField or DOCA run." - cargo test --workspace --all-targets + cargo test --workspace --all-targets --locked test-aarch64-qemu: name: test (aarch64 qemu-user) @@ -102,4 +90,4 @@ jobs: env: CARGO_TARGET_AARCH64_UNKNOWN_LINUX_GNU_LINKER: aarch64-linux-gnu-gcc CARGO_TARGET_AARCH64_UNKNOWN_LINUX_GNU_RUNNER: qemu-aarch64-static -L /usr/aarch64-linux-gnu - run: cargo test --workspace --all-targets --target aarch64-unknown-linux-gnu + run: cargo test --workspace --all-targets --locked --target aarch64-unknown-linux-gnu diff --git a/.gitignore b/.gitignore index 3fe0ed2..eb61a14 100644 --- a/.gitignore +++ b/.gitignore @@ -5,3 +5,5 @@ /tmp .cargo/registry .cargo/git +/artifacts/ +**/__pycache__/ diff --git a/scripts/ci/check_contract.py b/scripts/ci/check_contract.py new file mode 100644 index 0000000..e3012d3 --- /dev/null +++ b/scripts/ci/check_contract.py @@ -0,0 +1,93 @@ +#!/usr/bin/env python3 +# Copyright 2024-2026 RustFS, Inc. +# SPDX-License-Identifier: Apache-2.0 +"""Check the V3 dependency boundaries using Cargo's resolved metadata graph.""" + +import argparse +import json +import pathlib +import subprocess +import sys + + +ALLOWED = { + "gpucache-core-types": {"sha2", "proptest"}, + "gpucache-protocol": {"gpucache-core-types", "proptest"}, + "gpucache-storage": {"gpucache-core-types", "gpucache-protocol", "proptest"}, + "gpucache-transport": { + "gpucache-core-types", "gpucache-protocol", "gpucache-storage" + }, + "gpucache-cold": {"gpucache-core-types", "sha2"}, + "gpucache-dpu": {"gpucache-core-types"}, + "gpucache-bench": set(), +} + + +def check_metadata(metadata): + """Return violations; dependency aliases cannot bypass package checks.""" + errors = [] + members = set(metadata["workspace_members"]) + packages = {p["name"]: p for p in metadata["packages"] if p["id"] in members} + root = pathlib.Path(metadata["workspace_root"]).resolve() + for name in sorted(set(packages) - set(ALLOWED)): + errors.append("unreviewed workspace crate: " + name) + for name in sorted(set(ALLOWED) - set(packages)): + errors.append("missing workspace crate: " + name) + for name, package in sorted(packages.items()): + expected = root / "crates" / name.removeprefix("gpucache-") / "Cargo.toml" + if pathlib.Path(package["manifest_path"]).resolve() != expected: + errors.append(name + ": workspace manifest is outside its reviewed crate directory") + for dependency in package["dependencies"]: + # Cargo reports the actual package name even when `package = ...` + # or target-specific dependency tables rename the local binding. + actual = dependency["name"] + if actual not in ALLOWED.get(name, set()): + errors.append("{}: dependency {} is not allowed".format(name, actual)) + if actual in ALLOWED: + expected_dependency = packages.get(actual) + actual_path = dependency.get("path") + if not expected_dependency or not actual_path or pathlib.Path(actual_path).resolve() != pathlib.Path( + expected_dependency["manifest_path"] + ).resolve().parent: + errors.append("{}: {} must resolve to its reviewed workspace crate".format(name, actual)) + elif actual in {"sha2", "proptest"} and ( + dependency.get("path") is not None + or not (dependency.get("source") or "").startswith("registry+") + ): + errors.append("{}: {} must be a locked registry dependency".format(name, actual)) + if actual == "proptest" and dependency.get("kind") != "dev": + errors.append(name + ": proptest must remain a dev dependency") + for feature in ("rdma-nixl", "dpu-bf3"): + values = package.get("features", {}).get(feature) + if values is not None and values != []: + errors.append("{}: {} must remain an empty feature slot".format(name, feature)) + return errors + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--repo", type=pathlib.Path, + default=pathlib.Path(__file__).resolve().parents[2]) + args = parser.parse_args(argv) + try: + completed = subprocess.run( + ["cargo", "metadata", "--locked", "--offline", "--no-deps", "--format-version", "1"], + cwd=args.repo, capture_output=True, text=True, timeout=60, check=False, + ) + if completed.returncode: + print("CONTRACT_FAILED: cargo metadata failed", file=sys.stderr) + return 1 + errors = check_metadata(json.loads(completed.stdout)) + except (OSError, ValueError, KeyError, subprocess.TimeoutExpired) as error: + print("CONTRACT_FAILED: " + type(error).__name__, file=sys.stderr) + return 1 + for error in errors: + print(error, file=sys.stderr) + if errors: + return 1 + print("status=CPU_DEPENDENCY_CONTRACT_OK") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/ci/tests/test_vm_regression.py b/scripts/ci/tests/test_vm_regression.py new file mode 100644 index 0000000..3504c1d --- /dev/null +++ b/scripts/ci/tests/test_vm_regression.py @@ -0,0 +1,289 @@ +# Copyright 2024-2026 RustFS, Inc. +# SPDX-License-Identifier: Apache-2.0 +"""Runner regressions; an optional tiny Cargo probe checks endpoint isolation.""" + +import contextlib +import io +import json +import os +import pathlib +import shutil +import subprocess +import sys +import tempfile +import unittest +from unittest import mock +import xml.etree.ElementTree as ET + +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1])) +import check_contract as contract # noqa: E402 +import vm_regression as vm # noqa: E402 + + +class ExecutionTests(unittest.TestCase): + def setUp(self): + self.directory = tempfile.TemporaryDirectory(prefix="gpucache-runner-test-") + self.root = pathlib.Path(self.directory.name) + (self.root / "logs").mkdir() + + def tearDown(self): + self.directory.cleanup() + + def execute(self, code, expected=(), timeout=5): + gate = vm.Gate("probe", (sys.executable, "-c", code), expected) + return vm.execute(gate, self.root, self.root / "logs" / "probe.log", + timeout, vm.sanitized_environment(7)) + + def test_success_requires_exit_and_expected_markers(self): + result = self.execute("print('status=READY')", ("status=READY",)) + self.assertEqual(result["status"], vm.PASS) + self.assertEqual(result["returncode"], 0) + self.assertEqual(result["log"], "logs/probe.log") + + def test_zero_exit_with_missing_contract_is_failure(self): + result = self.execute("print('not the contract')", ("status=READY",)) + self.assertEqual(result["status"], vm.FAIL) + self.assertEqual(result["missing_markers"], ["status=READY"]) + + def test_success_marker_must_be_a_complete_line(self): + result = self.execute("print('error=status=READY_NOT_READY')", ("status=READY",)) + self.assertEqual(result["status"], vm.FAIL) + + def test_contradictory_success_markers_fail_closed(self): + result = self.execute("print('status=READY'); print('status=FAILED')", ("status=READY",)) + self.assertEqual(result["status"], vm.FAIL) + self.assertEqual(result["contradictory_markers"], ["status=FAILED"]) + + def test_failure_cannot_be_hidden_by_success_banner(self): + result = self.execute("print('status=READY'); raise SystemExit(7)", ("status=READY",)) + self.assertEqual(result["status"], vm.FAIL) + self.assertEqual(result["returncode"], 7) + + def test_missing_executable_is_reported(self): + gate = vm.Gate("missing", (str(self.root / "nonexistent-executable"),)) + result = vm.execute(gate, self.root, self.root / "logs" / "missing.log", 1, {}) + self.assertEqual(result["status"], vm.FAIL) + self.assertIsNone(result["returncode"]) + + def test_timeout_terminates_process(self): + result = self.execute("import time; time.sleep(60)", timeout=0.2) + self.assertEqual(result["status"], vm.FAIL) + self.assertEqual(result["reason"], "timed out") + self.assertLess(result["duration_seconds"], 5) + + @unittest.skipUnless(os.name == "posix", "POSIX process groups") + def test_timeout_kills_the_entire_child_process_group(self): + child = mock.Mock(pid=1234) + child.wait.side_effect = [subprocess.TimeoutExpired(["fake"], 1), -9] + with mock.patch.object(vm.subprocess, "Popen", return_value=child), \ + mock.patch.object(vm.os, "killpg") as kill: + result = vm.execute(vm.Gate("timeout", ("fake",)), self.root, + self.root / "logs" / "timeout.log", 1, {}) + kill.assert_called_once_with(1234, vm.signal.SIGKILL) + self.assertEqual(result["reason"], "timed out") + + @unittest.skipUnless(os.name == "posix", "POSIX process groups") + def test_failed_parent_exit_still_cleans_up_descendant_processes(self): + child = mock.Mock(pid=1234) + child.wait.return_value = 3 + with mock.patch.object(vm.subprocess, "Popen", return_value=child), \ + mock.patch.object(vm.os, "killpg") as kill: + result = vm.execute(vm.Gate("failed", ("fake",)), self.root, + self.root / "logs" / "failed.log", 1, {}) + kill.assert_called_once_with(1234, vm.signal.SIGKILL) + self.assertEqual(result["status"], vm.FAIL) + + def test_individual_rust_test_results_are_preserved(self): + result = self.execute("print('test module::case ... ok')") + self.assertEqual(result["rust_tests"], [{"name": "module::case", "status": "ok"}]) + + def test_cpu_environment_does_not_enable_a_live_rustfs_probe(self): + with mock.patch.dict(os.environ, {"GPUCACHE_RUSTFS_ENDPOINT": "https://private", + "GPUCACHE_RUSTFS_SECRET_KEY": "secret"}): + env = vm.sanitized_environment(7) + self.assertEqual(env["GPUCACHE_RUSTFS_ENDPOINT"], "") + self.assertNotIn("GPUCACHE_RUSTFS_SECRET_KEY", env) + self.assertNotIn("PROPTEST_RNG_SEED", env) + + +class PlanTests(unittest.TestCase): + def test_every_cargo_build_uses_the_lockfile(self): + for gate in vm.cpu_gates(): + if gate.argv[0] == "cargo" and gate.name != "format": + self.assertIn("--locked", gate.argv, gate.name) + + def test_offline_applies_to_every_cargo_build(self): + for gate in vm.cpu_gates(offline=True): + if gate.argv[0] == "cargo" and gate.name != "format": + self.assertIn("--offline", gate.argv, gate.name) + + def test_cargo_cannot_reenable_live_rustfs_through_global_config(self): + for gate in vm.cpu_gates(): + if gate.argv[0] == "cargo": + self.assertEqual(gate.argv[1:1 + len(vm.NO_LIVE_ENDPOINT)], vm.NO_LIVE_ENDPOINT) + + def test_required_gates_are_unique_and_cover_features_and_demos(self): + names = [gate.name for gate in vm.cpu_gates()] + self.assertEqual(len(names), len(set(names))) + for name in ("workspace-tests", "doc-tests", "rdma-feature-contract", + "dpu-feature-contract", "combined-feature-tests", "share-demo", + "fault-demo", "cold-demo", "dpu-demo", "v3-paired"): + self.assertIn(name, names) + + def test_invalid_timeouts_are_rejected(self): + for value in ("0", "-1", "nan", "inf"): + with self.assertRaises(vm.argparse.ArgumentTypeError): + vm.positive_number(value) + + +class ReportTests(unittest.TestCase): + def setUp(self): + self.directory = tempfile.TemporaryDirectory(prefix="gpucache-report-test-") + self.repo = pathlib.Path(self.directory.name) + (self.repo / "Cargo.toml").write_text("[workspace]\n", encoding="utf-8") + self.output = self.repo / "report" + + def tearDown(self): + self.directory.cleanup() + + def run_main(self, *args): + with mock.patch.object(vm, "environment_report", return_value={"schema_version": 1}), \ + contextlib.redirect_stdout(io.StringIO()): + return vm.main(["--repo", str(self.repo), "--output", str(self.output)] + list(args)) + + def test_all_hardware_profiles_fail_closed_and_write_evidence(self): + for profile in vm.HARDWARE: + self.output = self.repo / profile + with mock.patch.object(vm, "execute") as execute: + self.assertEqual(self.run_main("--profile", profile), 2) + execute.assert_not_called() + result = json.loads((self.output / "results.json").read_text()) + self.assertEqual(result["status"], vm.NOT_RUN) + capabilities = json.loads((self.output / "capabilities.json").read_text()) + self.assertEqual(capabilities[profile], vm.NOT_RUN) + self.assertEqual(capabilities["cpu_contract"], "NOT_RUN") + self.assertEqual(ET.parse(self.output / "junit.xml").getroot().attrib["failures"], "1") + + def test_cpu_success_does_not_qualify_hardware(self): + gate = vm.Gate("real-python-check", (sys.executable, "-c", "print('checked')")) + with mock.patch.object(vm, "cpu_gates", return_value=[gate]): + self.assertEqual(self.run_main(), 0) + result = json.loads((self.output / "results.json").read_text()) + capabilities = json.loads((self.output / "capabilities.json").read_text()) + self.assertEqual(result["status"], vm.PASS) + self.assertEqual(capabilities["hardware_qualification"], "INCOMPLETE") + self.assertEqual(capabilities["rdma"], vm.NOT_RUN) + self.assertEqual(capabilities["cpu_contract"], vm.PASS) + + def test_failure_is_nonzero_and_remaining_gates_are_collected(self): + gates = [vm.Gate("fails", (sys.executable, "-c", "raise SystemExit(3)")), + vm.Gate("runs-after-failure", (sys.executable, "-c", "print('checked')"))] + with mock.patch.object(vm, "cpu_gates", return_value=gates): + self.assertEqual(self.run_main(), 1) + result = json.loads((self.output / "results.json").read_text()) + self.assertEqual([row["status"] for row in result["required_gates"]], [vm.FAIL, vm.PASS]) + self.assertEqual(result["status"], vm.FAIL) + + def test_nonempty_output_is_not_overwritten(self): + self.output.mkdir() + sentinel = self.output / "keep.txt" + sentinel.write_text("previous evidence", encoding="utf-8") + with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit) as raised: + self.run_main() + self.assertEqual(raised.exception.code, 2) + self.assertEqual(sentinel.read_text(), "previous evidence") + + def test_file_output_is_rejected_without_overwriting_it(self): + self.output.write_text("previous evidence", encoding="utf-8") + with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit) as raised: + self.run_main() + self.assertEqual(raised.exception.code, 2) + + +def metadata(dependencies=(), features=None): + packages = [] + for name in contract.ALLOWED: + package = {"id": name, "name": name, "dependencies": [], "features": {}, + "manifest_path": "/workspace/crates/" + name.removeprefix("gpucache-") + "/Cargo.toml"} + if name == "gpucache-transport": + package["dependencies"] = list(dependencies) + package["features"] = features or {} + packages.append(package) + return {"workspace_root": "/workspace", "workspace_members": list(contract.ALLOWED), "packages": packages} + + +class DependencyContractTests(unittest.TestCase): + def test_expected_graph_passes(self): + self.assertEqual(contract.check_metadata(metadata()), []) + + def test_renamed_accelerator_dependency_is_not_allowed(self): + value = metadata([{"name": "cuda-driver", "rename": "innocent", "kind": None}]) + self.assertTrue(contract.check_metadata(value)) + + def test_target_specific_and_build_dependencies_cannot_bypass_check(self): + for kind in (None, "build", "dev"): + value = metadata([{"name": "libnixl", "kind": kind, "target": "cfg(target_arch = \"aarch64\")"}]) + self.assertTrue(contract.check_metadata(value)) + + def test_same_name_foreign_crate_cannot_replace_a_workspace_dependency(self): + value = metadata([{"name": "gpucache-core-types", "kind": None, "path": "/foreign"}]) + self.assertTrue(contract.check_metadata(value)) + + def test_empty_feature_cannot_enable_another_feature(self): + value = metadata(features={"rdma-nixl": ["dep:cuda-driver"]}) + self.assertTrue(contract.check_metadata(value)) + + def test_unreviewed_workspace_members_fail(self): + value = metadata() + value["workspace_members"].append("new") + value["packages"].append({"id": "new", "name": "new", "dependencies": [], "features": {}, + "manifest_path": "/workspace/crates/new/Cargo.toml"}) + self.assertTrue(contract.check_metadata(value)) + + def test_missing_workspace_members_fail(self): + value = metadata() + value["packages"] = value["packages"][:-1] + self.assertTrue(contract.check_metadata(value)) + + +@unittest.skipUnless(shutil.which("cargo"), "Cargo required for configuration-injection probe") +class CargoEndpointIsolationTests(unittest.TestCase): + def test_global_and_checkout_force_relative_config_cannot_enable_a_live_probe(self): + with tempfile.TemporaryDirectory(prefix="gpucache-cargo-env-") as directory: + root = pathlib.Path(directory) + home, project = root / "cargo-home", root / "probe" + home.mkdir() + (project / "src").mkdir(parents=True) + (project / ".cargo").mkdir() + (project / "Cargo.toml").write_text( + '[package]\nname="gpucache-env-probe"\nversion="0.0.0"\nedition="2021"\n', + encoding="utf-8", + ) + (project / "src/main.rs").write_text( + 'fn main() { if !std::env::var("GPUCACHE_RUSTFS_ENDPOINT").unwrap_or_default().is_empty() { std::process::exit(86); } }', + encoding="utf-8", + ) + hostile = ('[env]\nGPUCACHE_RUSTFS_ENDPOINT = ' + '{ value="https://example.invalid", force=true, relative=true }\n') + env = vm.sanitized_environment(7) + env["CARGO_HOME"] = str(home) + env["CARGO_TARGET_DIR"] = str(root / "target") + for global_config, checkout_config in ((True, False), (False, True), (True, True)): + with self.subTest(global_config=global_config, checkout_config=checkout_config): + (home / "config.toml").write_text(hostile if global_config else "", encoding="utf-8") + (project / ".cargo/config.toml").write_text(hostile if checkout_config else "", encoding="utf-8") + baseline = subprocess.run( + ["cargo", "run", "--quiet", "--offline"], cwd=project, env=env, + capture_output=True, text=True, timeout=60, check=False, + ) + self.assertEqual(baseline.returncode, 86, baseline.stderr) + result = subprocess.run( + ["cargo"] + list(vm.NO_LIVE_ENDPOINT) + ["run", "--quiet", "--offline"], + cwd=project, env=env, capture_output=True, text=True, timeout=60, + check=False, + ) + self.assertEqual(result.returncode, 0, result.stderr) + + +if __name__ == "__main__": + unittest.main() diff --git a/scripts/ci/vm_regression.py b/scripts/ci/vm_regression.py new file mode 100644 index 0000000..a6eb75e --- /dev/null +++ b/scripts/ci/vm_regression.py @@ -0,0 +1,295 @@ +#!/usr/bin/env python3 +# Copyright 2024-2026 RustFS, Inc. +# SPDX-License-Identifier: Apache-2.0 +"""Run the portable CPU gates and write reproducible VM regression evidence. + +No third-party Python packages, root access, device writes, model downloads, +cloud services, or live RustFS credentials are required. Hardware profiles +fail closed until real qualified implementations exist in this repository. +""" + +import argparse +import datetime +import hashlib +import json +import os +import pathlib +import platform +import re +import signal +import subprocess +import sys +import time +import xml.etree.ElementTree as ET +from dataclasses import dataclass + + +SCHEMA_VERSION = 1 +HARDWARE = ("rdma", "gds", "bf3", "gpu-engine") +PASS = "PASS" +FAIL = "FAIL" +NOT_RUN = "NOT_RUN_HARDWARE" +NO_LIVE_ENDPOINT = ( + "--config", 'env.GPUCACHE_RUSTFS_ENDPOINT.value=""', + "--config", "env.GPUCACHE_RUSTFS_ENDPOINT.force=true", + "--config", "env.GPUCACHE_RUSTFS_ENDPOINT.relative=false", +) + + +@dataclass(frozen=True) +class Gate: + name: str + argv: tuple + expected: tuple = () + + +def cpu_gates(offline=False): + # Highest-precedence Cargo config also blocks [env] injection from a + # checkout or an operator's global Cargo config, including force=true. + cargo = ("cargo",) + NO_LIVE_ENDPOINT + locked = ("--locked",) + (("--offline",) if offline else ()) + gates = [ + Gate("format", cargo + ("fmt", "--all", "--", "--check")), + Gate("dependency-contract", (sys.executable, "scripts/ci/check_contract.py"), + ("status=CPU_DEPENDENCY_CONTRACT_OK",)), + Gate("runner-tests", (sys.executable, "-m", "unittest", "discover", + "-s", "scripts/ci/tests", "-v")), + Gate("clippy", cargo + ("clippy", "--workspace", "--all-targets", "--all-features") + + locked + ("--", "-D", "warnings")), + Gate("workspace-tests", cargo + ("test", "--workspace", "--all-targets") + locked), + Gate("doc-tests", cargo + ("test", "--workspace", "--doc") + locked), + Gate("rdma-feature-contract", cargo + ("test", "-p", "gpucache-transport", + "--all-targets", "--features", "rdma-nixl") + locked), + Gate("dpu-feature-contract", cargo + ("test", "-p", "gpucache-dpu", + "--all-targets", "--features", "dpu-bf3") + locked), + Gate("combined-feature-tests", cargo + ("test", "--workspace", "--all-targets", + "--all-features") + locked), + ] + demos = [ + ("share-demo", "gpucache-transport", "share-demo", ("status=REUSE_OK",)), + ("fault-demo", "gpucache-transport", "fault-demo", ("status=FALLBACK_OK",)), + ("shard-demo", "gpucache-transport", "shard-demo", + ("status=SHARD_OK", "registration=SIMULATED", "rdma=NOT_RUN_HARDWARE")), + ("cold-demo", "gpucache-cold", "cold-demo", + ("status=COLD_CONTRACT_OK", "rustfs_live=NOT_RUN_HARDWARE")), + ("dpu-demo", "gpucache-dpu", "dpu-demo", + ("status=DPU_CONTRACT_OK", "offload_active=false", "hardware=NOT_RUN_HARDWARE")), + ("paired-baseline", "gpucache-bench", "paired-baseline", ("status=NOT_RUN_HARDWARE",)), + ("v3-paired", "gpucache-bench", "v3-paired", + ("status=NOT_RUN_HARDWARE", "tco=TCO_UNKNOWN", "go_gate=NOT_EVALUATED")), + ] + for name, package, binary, expected in demos: + gates.append(Gate(name, cargo + ("run", "-p", package, "--bin", binary, "--quiet") + + locked, expected)) + return gates + + +def sanitized_environment(seed): + env = os.environ.copy() + # CPU regression must never probe an operator's configured live endpoint. + for key in list(env): + if key.startswith("GPUCACHE_RUSTFS_"): + del env[key] + env["GPUCACHE_RUSTFS_ENDPOINT"] = "" + env.pop("PROPTEST_RNG_SEED", None) + env["CARGO_TERM_COLOR"] = "never" + env["RUST_BACKTRACE"] = "1" + # proptest 1.4 does not consume PROPTEST_RNG_SEED. The campaign seed is + # report metadata only; failing property cases use proptest persistence. + env["PYTHONDONTWRITEBYTECODE"] = "1" + return env + + +def execute(gate, repo, log, timeout, env): + started = time.monotonic() + returncode = None + reason = None + with log.open("wb") as output: + try: + child = subprocess.Popen( + list(gate.argv), cwd=repo, env=env, stdout=output, + stderr=subprocess.STDOUT, start_new_session=(os.name == "posix"), + ) + except OSError as error: + reason = "launch failed: " + type(error).__name__ + else: + try: + returncode = child.wait(timeout=timeout) + except (subprocess.TimeoutExpired, KeyboardInterrupt) as error: + reason = "timed out" if isinstance(error, subprocess.TimeoutExpired) else "interrupted" + finally: + # Include child Cargo/test processes; leaving a cache server or + # a cargo lock behind makes the next VM run nondeterministic. + # A failed command can leave descendants even after it exits. + try: + if os.name == "posix": + os.killpg(child.pid, signal.SIGKILL) + elif child.poll() is None: + child.kill() + except ProcessLookupError: + pass + returncode = child.wait() + text = log.read_text(encoding="utf-8", errors="replace") + lines = set(text.splitlines()) + missing = [marker for marker in gate.expected if marker not in lines] + contradictory = [line for marker in gate.expected for line in lines + if "=" in marker and line.startswith(marker.split("=", 1)[0] + "=") + and line != marker] + if reason is None and returncode != 0: + reason = "nonzero exit" + if reason is None and missing: + reason = "missing required output markers" + if reason is None and contradictory: + reason = "contradictory output markers" + # This is elapsed command runtime, never a GPU/engine latency benchmark. + return { + "name": gate.name, "status": FAIL if reason else PASS, + "command": list(gate.argv), "returncode": returncode, + "duration_seconds": round(time.monotonic() - started, 6), + "reason": reason, "missing_markers": missing, + "contradictory_markers": sorted(contradictory), + "log": str(log.relative_to(log.parent.parent)), + "rust_tests": [ + {"name": match.group(1), "status": match.group(2)} + for match in re.finditer(r"^test (.+?) \.\.\. (ok|FAILED|ignored)$", text, re.MULTILINE) + ], + } + + +def probe(argv, repo): + try: + result = subprocess.run(argv, cwd=repo, capture_output=True, text=True, + timeout=10, check=False) + return result.stdout.strip() if result.returncode == 0 else "UNAVAILABLE" + except (OSError, subprocess.TimeoutExpired): + return "UNAVAILABLE" + + +def environment_report(repo, seed): + lock = repo / "Cargo.lock" + return { + "schema_version": SCHEMA_VERSION, + "recorded_at_utc": datetime.datetime.now(datetime.timezone.utc).isoformat(), + "repository_commit": probe(["git", "rev-parse", "HEAD"], repo), + "working_tree_dirty": probe(["git", "status", "--porcelain", "--untracked-files=normal"], repo) != "", + "cargo_lock_sha256": hashlib.sha256(lock.read_bytes()).hexdigest() if lock.is_file() else None, + "os": platform.system(), "os_release": platform.release(), + "architecture": platform.machine(), "python": platform.python_version(), + "rustc": probe(["rustc", "--version", "--verbose"], repo), + "cargo": probe(["cargo", "--version"], repo), + "seed": seed, + "seed_scope": "campaign metadata; proptest uses its own persisted failure seeds", + "live_rustfs_probe": "DISABLED_FOR_CPU_REGRESSION", + "performance_measurements": "NONE", + } + + +def capability_report(cpu_status): + return { + "schema_version": SCHEMA_VERSION, + "cpu_contract": cpu_status, + "simulated_registration": "IMPLEMENTED_SIMULATED", + "simulated_dpu_queue": "IMPLEMENTED_SIMULATED", + "rdma": NOT_RUN, "gds": NOT_RUN, "bf3": NOT_RUN, + "gpu-engine": NOT_RUN, "live_rustfs": NOT_RUN, + "hardware_qualification": "INCOMPLETE", + "tco": "TCO_UNKNOWN", "fleet_qualification": "NOT_RUN_HARDWARE", + } + + +def write_json(path, data): + # Atomic replacement avoids a half-written PASS document after interruption. + temporary = path.with_name(path.name + ".tmp") + temporary.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n", encoding="utf-8") + temporary.replace(path) + + +def write_reports(output, environment, results, status, profile): + capabilities = capability_report(status if profile == "cpu" else "NOT_RUN") + write_json(output / "environment.json", environment) + write_json(output / "capabilities.json", capabilities) + write_json(output / "results.json", { + "schema_version": SCHEMA_VERSION, "profile": profile, "status": status, + "required_gates": results, "hardware_qualification": "INCOMPLETE", + }) + suite = ET.Element("testsuite", name="gpucache-vm-" + profile, + tests=str(len(results)), + failures=str(sum(row["status"] != PASS for row in results))) + for row in results: + case = ET.SubElement(suite, "testcase", name=row["name"], + time=str(row["duration_seconds"])) + if row["status"] != PASS: + ET.SubElement(case, "failure", message=row["reason"] or row["status"]) + ET.SubElement(case, "system-out").text = row["log"] + ET.ElementTree(suite).write(output / "junit.xml", encoding="utf-8", xml_declaration=True) + lines = ["# GPUCache VM regression", "", "Profile: `{}`".format(profile), + "CPU/selected-profile result: **{}**".format(status), "", + "Hardware qualification: **INCOMPLETE**. RDMA, GDS, GPU engine and BF3 are `NOT_RUN_HARDWARE`.", + "Command durations are execution times, not TTFT, TPOT or goodput.", "", + "| Gate | Result | Log |", "| --- | --- | --- |"] + lines.extend("| {} | {} | {} |".format(row["name"], row["status"], row["log"]) + for row in results) + (output / "summary.md").write_text("\n".join(lines) + "\n", encoding="utf-8") + + +def positive_number(value): + parsed = float(value) + if not (0 < parsed < float("inf")): + raise argparse.ArgumentTypeError("must be finite and positive") + return parsed + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--profile", choices=("cpu",) + HARDWARE, default="cpu") + parser.add_argument("--output", type=pathlib.Path, default=pathlib.Path("artifacts/vm-regression")) + parser.add_argument("--repo", type=pathlib.Path, + default=pathlib.Path(__file__).resolve().parents[2]) + parser.add_argument("--timeout", type=positive_number, default=900, + help="timeout in seconds for each gate") + parser.add_argument("--seed", type=int, default=20261017) + parser.add_argument("--offline", action="store_true") + args = parser.parse_args(argv) + repo = args.repo.resolve() + output = args.output.resolve() + if not (repo / "Cargo.toml").is_file(): + parser.error("--repo must contain the GPUCache Cargo workspace") + # Use a new report directory. Never overwrite another run's evidence or + # clean an operator-supplied arbitrary directory. + if output.exists() and (not output.is_dir() or any(output.iterdir())): + parser.error("--output must be empty or new; preserve previous regression evidence") + output.mkdir(parents=True, exist_ok=True) + (output / "logs").mkdir() + environment = environment_report(repo, args.seed) + results = [] + write_reports(output, environment, results, "RUNNING", args.profile) + if args.profile != "cpu": + results.append({ + "name": args.profile + "-qualification", "status": NOT_RUN, + "command": [], "returncode": None, "duration_seconds": 0, + "reason": "real hardware adapter and qualification runner are not implemented", + "missing_markers": [], "log": "", "rust_tests": [], + }) + write_reports(output, environment, results, NOT_RUN, args.profile) + print("status=NOT_RUN_HARDWARE profile=" + args.profile) + return 2 + env = sanitized_environment(args.seed) + for index, gate in enumerate(cpu_gates(args.offline)): + log = output / "logs" / "{:02d}-{}.log".format(index + 1, gate.name) + print("running=" + gate.name, flush=True) + row = execute(gate, repo, log, args.timeout, env) + results.append(row) + write_reports(output, environment, results, "RUNNING", args.profile) + print("gate={} status={}".format(gate.name, row["status"]), flush=True) + if row["reason"] == "interrupted": + break + status = PASS if len(results) == len(cpu_gates(args.offline)) and all( + row["status"] == PASS for row in results + ) else FAIL + write_reports(output, environment, results, status, args.profile) + print("status={} hardware=NOT_RUN_HARDWARE report={}".format(status, output / "summary.md")) + return 0 if status == PASS else 1 + + +if __name__ == "__main__": + sys.exit(main()) From ff5af1ee29d8cdaebad0097813a49f5fe09305a6 Mon Sep 17 00:00:00 2001 From: loverustfs Date: Sat, 17 Oct 2026 15:05:28 +0800 Subject: [PATCH 3/4] docs: define phased KV cache replacement and VM acceptance --- .github/pull_request_template.md | 19 ++++++ Cargo.toml | 2 +- README.md | 12 +++- docs/PR_PLAN.md | 59 +++++++++++++++++ docs/REPLACEMENT_STRATEGY.md | 62 +++++++++++++++++ docs/ROADMAP.md | 74 +++++++++++++++++++++ docs/V3_FAILURES.md | 12 ++-- docs/V3_TEST_MATRIX.md | 6 ++ docs/VM_REGRESSION.md | 110 +++++++++++++++++++++++++++++++ 9 files changed, 350 insertions(+), 6 deletions(-) create mode 100644 .github/pull_request_template.md create mode 100644 docs/PR_PLAN.md create mode 100644 docs/REPLACEMENT_STRATEGY.md create mode 100644 docs/ROADMAP.md create mode 100644 docs/VM_REGRESSION.md diff --git a/.github/pull_request_template.md b/.github/pull_request_template.md new file mode 100644 index 0000000..4bacc7f --- /dev/null +++ b/.github/pull_request_template.md @@ -0,0 +1,19 @@ +## Change + +Describe the concrete problem and resulting behavior. Link the relevant ADR or roadmap stage. + +## Validation + +List the commands actually run and their results. Link regression artifacts when available. A stub returning zero is not a hardware pass. + +## Compatibility and failure behavior + +Describe affected identity/layout versions, tenant boundaries, admission limits, buffer/pin lifecycle, and miss/recompute behavior where relevant. + +## Evidence limits + +List unrun device/engine checks explicitly (`NOT_RUN_HARDWARE` where applicable). Keep `TCO_UNKNOWN` and `NOT_EVALUATED` until paired evidence exists. Distinguish CPU simulation, VM execution, real hardware and proposed work. + +## Scope and rollback + +State remaining limitations and the rollback/fallback for changed behavior. GPUCache does not add hot-tier erasure coding or durability replicas. Do not include credentials, prompts or KV payloads in this PR. diff --git a/Cargo.toml b/Cargo.toml index f97bfed..2eb9237 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,7 +16,7 @@ edition = "2021" license = "Apache-2.0" rust-version = "1.85" authors = ["RustFS, Inc."] -repository = "https://github.com/rustfs/gpucache" +repository = "https://github.com/rustfs/GPUCache" [workspace.dependencies] sha2 = "=0.10.8" diff --git a/README.md b/README.md index a5fb7da..56ca2c9 100644 --- a/README.md +++ b/README.md @@ -29,7 +29,7 @@ V1 runs the cache contract on CPU. V2 adds a transport trait, a simulated regist | 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: degrade matrix (DPU restart, GPU reset, NIC flap, storage loss, NVMe timeout, control-plane loss, RustFS slow/down) | Simulated fault decisions request recompute. Hidden slab bytes remain charged and can exhaust capacity until explicit restart; recovery is not proven on hardware | +| 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) | @@ -69,6 +69,16 @@ Resource limits cover object size, writer admissions, memory registrations, per- ## What you can run +For an unattended VM or Grok bot regression, run: + +```bash +python3 scripts/ci/vm_regression.py --profile cpu --output artifacts/vm-regression +``` + +The runner checks the locked dependency contract, formatting, Clippy, workspace and doc tests, optional and combined features, and CPU demos. It writes JSON, JUnit, a summary and per-gate logs. Hardware profiles return `NOT_RUN_HARDWARE` and exit 2; a CPU pass does not qualify RDMA, GDS, a GPU engine or BF3. Live RustFS access is disabled for this profile, including Cargo-configured endpoints. See the [VM runbook](docs/VM_REGRESSION.md), [staged roadmap](docs/ROADMAP.md), [replacement strategy](docs/REPLACEMENT_STRATEGY.md), and [English PR plan](docs/PR_PLAN.md). + +V3 regression hardening rejects active replay-window eviction, cross-domain replay, retired credential revival and malformed identities. TCP replies must match the operation, epoch and ordered request IDs before a connection can be reused. The DPU queue simulation bounds task/slot counts and retains unknown-completion credits; these checks do not constitute device isolation or hardware drain. + From a clean checkout, with Rust stable (MSRV 1.85): ```bash diff --git a/docs/PR_PLAN.md b/docs/PR_PLAN.md new file mode 100644 index 0000000..3750692 --- /dev/null +++ b/docs/PR_PLAN.md @@ -0,0 +1,59 @@ +# Staged PR plan + +This is an English review plan. Current implementation scope remains V3 CPU work. Every follow-up below is **PROPOSED / NOT IMPLEMENTED**, not a completed milestone. Crate publication, releases, hardware purchases, deployment and website changes remain outside this implementation plan. + +The accepted [no-EC decision](adr/0002-no-erasure-coding.md), [integration decision](adr/0001-integration-route.md), [storage-queue candidate](adr/0006-storage-queue-offload.md) and crate dependency rules in [AGENTS.md](../AGENTS.md) take precedence over optional future designs. + +## Current V3 review slices + +| Proposed PR title | Scope | Required evidence | +| --- | --- | --- | +| `docs: define delivery gates and scoped replacement strategy` | Roadmap, compatibility boundaries, replacement gates and VM runbook | Links resolve; future work is labeled proposed; no simulated/hardware claims are conflated | +| `ci: add dependency-free VM regression reports` | Standard-library Python CPU runner, environment/capability/results reports, command logs and strict unavailable hardware profiles | Clean-checkout CPU execution; invalid arguments, missing toolchain, failed command and timeout fail predictably; unavailable hardware cannot silently pass | +| `fix: close reproduced V3 CPU contract defects` | Only defects established by regression evidence; retain V1/V2 behavior | Focused regression for each defect, full required Rust checks, updated limitations where behavior remains incomplete | + +These slices may be reviewed independently when their prerequisites are present. A documentation PR must not describe a planned runner, SDK or integration as implemented before its code is reviewed and executable. + +## Proposed follow-up PR sequence + +| Order and proposed title | Prerequisite | Reviewable result and acceptance | +| --- | --- | --- | +| 1. `feat(sdk): add a typed Rust client and backend conformance` | Existing protocol read and preserved | Narrow client facade, examples, bounded retries and pin/cancel ownership; shared conformance for memory/file/TCP; no GPU dependency in the core | +| 2. `docs(adr): select the primary vLLM integration route` | Authorized GPU, approved/pinned model and a real spike | Commit/model/layout manifest, actual block export/import, output comparison and native hook evidence; accept one route in ADR 0001 | +| 3. `feat(adapter): implement the selected engine cache route` | ADR 0001 Accepted | One native vLLM connector or LMCache bridge; real restore and failure fallback; narrow binding and compatibility matrix | +| 4. `feat(cluster): add bounded multi-node cache routing` | Service/metadata design recorded | Actual discovery/routing, stale ownership and drain behavior; multi-process fault tests; cache misses after loss remain valid; no hot replicas | +| 5. `feat(transport): add qualified NIXL RDMA transfers` | Dependency/safety ADRs revisited; licenses reviewed; hardware available | Actual registration/completion and fencing, device/path evidence, fault tests and native transport baseline; TCP stays usable | +| 6. `feat(cold): validate live RustFS refill and optional GDS` | Authorized endpoint; independent GDS prerequisites | Live object/manifest integrity, deletion and budget evidence; optional GDS direct/fallback evidence; G4 remains off mandatory PD path | +| 7. `feat(dpu): evaluate storage-side BF3 queue offload` | Real NIC-RDMA baseline and compatible BFB/DOCA/firmware | One storage-queue path, raw paired samples and predefined go/no-go decision; no Arm-start, TCP-baseline or unmeasured TCO shortcut | +| 8. `feat(adapter): qualify SGLang HiCache integration` | Primary path stable; ADR 0008 revisited | Separate engine/layout qualification and TP consistency; no claim of interchangeable engine KV bytes | +| 9. `test(scale): publish staged fleet and fault qualification` | Functioning service/integrations and an authorized fleet | Separate simulated/VM/physical scale, failure envelope, SLO and reserve-capacity evidence; no extrapolated 10,000-GPU certification | + +Hardware PRs can remain unstarted indefinitely without weakening the portable CPU implementation. A no-go for the DPU candidate is a valid measured result. Do not add an acceleration dependency merely to make an optional feature name look complete. + +## PR description contract + +Lead with the concrete problem and resulting behavior. For example: “Repeated client reads can retain pins without an admission cap. This change adds a per-tenant read-pin limit and returns a bounded miss/error when the limit is reached.” State only what the code and evidence demonstrate. + +Each PR should include: + +- Scope and prerequisite decisions, with the exact changed behavior. +- Validation commands and results, including a link/path to raw regression artifacts when produced. +- Compatibility, lifecycle, quota and security implications relevant to the change. +- Hardware not exercised, using existing labels such as `NOT_RUN_HARDWARE`. +- Remaining limitations and a concrete rollback or fallback where behavior changes. + +Do not paste secrets, bearer fingerprints, raw prompts or KV bytes into a PR, issue or report. Do not use checkboxes to turn an unavailable run into a pass. Document-only changes can use link/content checks; implementation changes run the appropriate meaningful checks as well as repository-required validation. + +## Review gates + +Required Rust checks for code changes remain: + +```sh +cargo fmt --all -- --check +cargo clippy --workspace --all-targets -- -D warnings +cargo test --workspace --all-targets +``` + +Optional feature checks must preserve their existing CPU-only refusal behavior until a separately accepted integration changes it. `dpu-demo`, `paired-baseline` and `v3-paired` returning exit code zero is not a hardware pass. The VM report must preserve the printed unavailable labels. + +Use small commits with English messages and rewrite the PR title/body around the final result if scope changes. An unrelated defect should be recorded separately unless it blocks the stated behavior. Each slice should leave a concrete, reviewable result with its evidence limits intact. diff --git a/docs/REPLACEMENT_STRATEGY.md b/docs/REPLACEMENT_STRATEGY.md new file mode 100644 index 0000000..f1f8477 --- /dev/null +++ b/docs/REPLACEMENT_STRATEGY.md @@ -0,0 +1,62 @@ +# Adoption and replacement strategy + +Status: **PROPOSED / NOT IMPLEMENTED**. This document describes future integration choices. It does not add an engine adapter, a Rust SDK crate, a native transport or a hardware dependency. Current implementation follows the V3 CPU boundary. [ADR 0001](adr/0001-integration-route.md) and [ADR 0008](adr/0008-second-engine.md) remain Pending. + +GPUCache should replace a measured cache-tier capability in a supported configuration. It should preserve the inference engine, its fastest local cache and a working transport where appropriate. A claim that GPUCache replaces a whole inference stack would exceed its data-tier scope. + +## Adoption sequence + +1. **Conformance first.** Publish the Rust client contract and prove existing identity, commit, checksum, tenant, pin and miss behavior through portable backends. This SDK is proposed, not currently a workspace crate. +2. **Choose one primary integration.** On authorized GPU hardware, pin vLLM and model revisions and run the real block round trip specified in ADR 0001. Prefer the existing hook in that revision. Select the native connector or LMCache bridge; do not ship two half-built routes. +3. **Start as an optional shared tier.** Preserve GPU-local prefix caching. Cache unavailability should trigger bounded fallback/recompute, with admission and deadlines that prevent refill and prefill storms from consuming the fleet. +4. **Measure against the deployment being replaced.** Keep policy, capacity, auth semantics, trace, model, GPU layout and SLO constant. Include write/population costs and no-reuse traffic. Publish regressions as well as wins. +5. **Expand only after evidence.** Additional engines, distributed production routing, RDMA, GDS and BF3 require the distinct gates in the [roadmap](ROADMAP.md). They are not implied by a storage backend adapter. + +## Capability-level migration + +| Existing deployment | Proposed first integration | Replacement claim allowed after acceptance | +| --- | --- | --- | +| vLLM APC | Retain GPU-local APC; select a supported external connector route after ADR 0001 | A tested external shared-cache path for the pinned engine/model/layout; not replacement of vLLM scheduling or attention | +| LMCache | If selected, implement GPUCache behind the supported storage interface and preserve the existing engine bridge | The tested backend capability and modes; not automatic equivalence to CacheBlend, compression or every LMCache backend | +| SGLang HiCache | Evaluate an L3 backend after the primary route works; retain RadixAttention and private local tiers | The qualified L3 behavior for that SGLang revision; not a universal KV format or an already implemented adapter | +| Mooncake Store | Evaluate GPUCache as an alternate shared store while preserving a supported TE where useful | Equivalent tested object/commit/failure behavior through the selected engine integration; no undocumented wire-compatibility promise | +| NIXL | Use a qualified NIXL transport integration where appropriate | A compatible cache tier using a transport library; GPUCache need not replace NIXL | +| NVIDIA Dynamo | Expose reuse through a supported engine integration and, later, versioned KV lifecycle events | Verified cache-tier integration and router visibility; not replacement of orchestration, autoscaling or inference workers | +| File, Redis or object-store offload | Add a compatibility probe and explicit migration tool only when their layouts are understood | Migration of compatible bytes, or safe regeneration for incompatible entries; not arbitrary cross-model KV reuse | + +The table identifies proposed integration boundaries rather than promising feature parity with any project. Validate upstream interfaces against pinned official revisions. Product names and plugin interfaces alone are not a compatibility test. + +## Compatibility contract + +A reusable block must match its authenticated tenant and trust domain, model revision/configuration, tokenizer, adapter, multimodal identity where relevant, position/RoPE configuration, dtype/quantization, layout ABI, TP/PP shard and valid lengths. Preserve the distinction between logical prefix identity and physical byte identity. + +Only a checksum-valid committed block is readable. A whole-block `PrefixHit` requires the documented prefix identity and matched length; it is not an instruction to slice arbitrary KV tensors or decode the shorter block as the entire request. See [protocol](PROTOCOL.md) and [partial-prefix ADR](adr/0003-partial-prefix.md). + +Non-prefix KV composition, resharding and format conversion are separate inference features. Storing or transporting bytes does not implement them. A model/layout change is a miss until an independently validated conversion is accepted. + +## Side-by-side evaluation and rollback + +The first proposed deployment should support a reversible baseline/candidate selection. Use an isolated namespace and bounded cache budget. Do not silently merge tenants or deduplicate private prefixes across tenants. Shadow writes may test population and integrity, but cannot establish real restore correctness by themselves. + +Before moving traffic, demonstrate real engine restore, declared output equivalence, error/fallback behavior and paired SLO goodput. After moving a bounded cohort, monitor lookup latency, saved prefill work, admission denials, pin lifetime, quarantine, fallback rate and useful output tokens. Cache hit ratio alone is not the decision metric. + +Rollback should stop new candidate traffic, cancel or drain outstanding work under the lifecycle contract, and preserve the baseline engine path. Unknown completion remains quarantined until verified drain/fencing. Leaving a candidate cache namespace behind is preferable to freeing a region that might still receive DMA. Cold-data retention/deletion follows its explicit policy; rollback is not permission to delete user objects. + +## Reliability and isolation boundaries + +GPUCache does not maintain hot durability replicas, parity or replica repair. Rebalance and fault-domain hooks currently fence fills rather than copy payloads. A lost hot block is a miss and may be recomputed/repopulated. RustFS supplies the colder durability boundary. See [ADR 0002](adr/0002-no-erasure-coding.md). + +Multi-node availability must be tested within a defined failure envelope. More cache misses after node loss can increase prefill pressure; reserve capacity, per-tenant admission, refill budgets and bounded retries are necessary before promising an SLO. A failed decode worker still needs engine-level request recovery. Never claim arbitrary simultaneous node loss has no effect. + +Current TCP uses a bearer-equivalent credential fingerprint on a trusted fabric. mTLS is not linked; V3 ACL/replay/rate/rotation hooks are not wired into the listener. These are migration constraints, not completed network security. See [security](SECURITY.md) and [trusted-fabric ADR](adr/0007-trusted-fabric.md). + +## Evidence required in a replacement report + +- Exact baseline/candidate commits and configurations, model revision, hardware/topology and active data path. +- Supported engine operations and modes, plus explicit unsupported combinations. +- Correctness criterion, raw samples, command logs, warm/cold/no-reuse tests and failure results. +- Population cost, TTFT/TPOT/ITL distributions, SLO goodput and capacity/resource costs. +- Rollback procedure and observed behavior under cache outage, noisy tenants and partial transfer failure. +- A scoped conclusion such as “replaces the tested remote KV backend for configuration X.” + +Until those results exist, use “proposed integration” or `NOT_RUN_HARDWARE`. Hardware acceleration and TCO remain unmeasured. See [benchmarks](BENCHMARKS.md). diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md new file mode 100644 index 0000000..8654ecd --- /dev/null +++ b/docs/ROADMAP.md @@ -0,0 +1,74 @@ +# GPUCache roadmap + +This roadmap separates the current V3 CPU scope from proposed follow-up work. A proposed stage is a design direction, not an implemented feature or a release commitment. [AGENTS.md](../AGENTS.md), the accepted ADRs, and the [compatibility manifest](../COMPATIBILITY.md) define the current implementation boundary. + +GPUCache is the G3.5 shared KV tier. RustFS is the colder G4 backend, off the prefill-to-decode hot path. A lost hot block is a miss followed by engine recomputation. GPUCache will not add hot-tier erasure coding or durability replicas. See [ADR 0002](adr/0002-no-erasure-coding.md). + +## Current evidence + +| Area | Current scope | Evidence boundary | +| --- | --- | --- | +| V1 A/B/C | Identity, lifecycle, protocol, memory/file stores, loopback TCP reuse, synthetic correctness and fault tests | CPU contracts. Synthetic bytes are not real engine KV tensors | +| V2 first slice | Transport trait, simulated registered buffers, shard-local directory, fencing, single-flight, credits, mock G4 packs and refill | One process is one shard. No distributed metadata service or quorum | +| V3 | Simulated storage-side DRAM slab and storage queue, quarantine/drain and tenant invalidation/reclaim contracts; CPU security/control hooks; mock cold operations | DOCA, SPDK, RDMA, CUDA, GDS and a BlueField device are not linked | +| Engine integration | Native vLLM connector versus LMCache bridge remains undecided | [ADR 0001](adr/0001-integration-route.md) is Pending; real round trip is `NOT_RUN_HARDWARE` | +| DPU decision | Storage queue is the sole first offload candidate | Hypothesis only. Paired NIC-RDMA versus DPU measurements are `NOT_RUN_HARDWARE`; `TCO_UNKNOWN` | + +The V3 hooks are not complete service enforcement. The TCP service does not call the V3 ACL, replay guard, rate gate or secret ring, and it does not terminate mTLS. Existing payload and writer limits are not a complete process-RSS bound. See [security](SECURITY.md) and [protocol](PROTOCOL.md). + +## V3 completion work + +Current V3 work remains CPU-only. Preserve the crate dependency restrictions and `unsafe` prohibition in AGENTS.md. + +1. Make the existing CPU contracts reproducible from a clean Linux VM, with structured results, raw command logs and explicit unavailable hardware rows. See [VM regression](VM_REGRESSION.md). +2. Keep every V1/V2 test passing while correcting documented V3 contract defects. A successful simulation must continue to report `offload_active=false` and `NOT_RUN_HARDWARE`. +3. Preserve the failure boundary: hiding bytes requests recompute but retains slab capacity until the explicit CPU drain/invalidation/reclaim rules permit release. Quarantined tasks retain credits and cannot publish a late completion; the simulation caps task and slot counts at 4,096. A CPU attestation is not evidence of a hardware drain. Never free a future DMA region merely because its lease expired. +4. Make follow-up work reviewable as small English PRs with their own acceptance evidence. See [PR plan](PR_PLAN.md). + +V3 completion does not close the pending engine route, certify a physical fault campaign, or certify service operation on an untrusted network. + +## Proposed follow-up stages + +All stages below are **PROPOSED / NOT IMPLEMENTED**. Progress through their technical prerequisites in order. Hardware linkage requires revisiting the current dependency and safety restrictions; this roadmap does not override them. + +| Proposed stage | Deliverable | Acceptance gate | +| --- | --- | --- | +| R1: Rust client SDK | A typed client facade over the existing identity, lifecycle and batch protocol; documented deadlines, cancellation, pins, bounded retries and compatible-version negotiation | Memory/file/TCP conformance; wrong tenant/model/layout returns no payload; every successful read pin is released; request and auxiliary-state admission is explicitly bounded and tested | +| R2: One real engine route | Run the vLLM spike, accept ADR 0001, then implement only the selected native connector or LMCache route | Pin engine/model revisions, layout, dtype and TP/PP; export/import a real GPU KV block; compare output under a declared determinism criterion; exercise miss and failure fallback | +| R3: Multi-node service | Node discovery and shard routing, bounded stale metadata, epoch fencing, admission credits, drain and controlled refill | Multi-process/VM campaign for simultaneous node loss, partition, restart, rolling upgrade and hot-prefix storms; correct surviving hits or bounded misses; no claim of seamless active-decode recovery | +| R4: Native NIXL/RDMA path | Optional transport integration with actual registration, completion, cancellation/drain and quarantine | Real RDMA hosts; path/device evidence; tenant access bounds; NIC failure and unknown completion cannot expose or prematurely recycle buffers; a simulated pool is insufficient | +| R5: Live cold backend and optional GDS | Live RustFS pack/manifest operations, cost-bounded refill, then a separately qualified GDS path | Authorized endpoint and integrity/delete/budget tests; GDS topology, cuFile/filesystem and direct-versus-fallback evidence; cold writes/refill do not become mandatory PD hops | +| R6: Storage-side BF3 experiment | The single storage-queue offload in ADR 0006, only after a real NIC-RDMA baseline | Same behavior, capacity, policy, auth, GPU/model/trace and SLO on both sides; raw repeated samples; mechanism, power and full cost evidence; a no-go retains portable backends | +| R7: Second engine and fleet qualification | SGLang/HiCache evaluation after the primary interface is stable; staged fleet and fault qualification | Accept ADR 0008 only after its prerequisites; qualify each engine/layout separately; publish simulated, VM and physical GPU scale as separate results | + +R4, R5 and R6 are distinct evidence gates. A GDS pass does not certify RDMA or DPU offload. Starting a daemon on Arm does not satisfy R6. An installed SDK, device name, compile or successful stub command does not satisfy a hardware gate. + +## Stable boundaries for the proposed SDK + +| Component | Owns | Must not assume | +| --- | --- | --- | +| Engine adapter | Engine block tables, GPU execution ordering, tensor export/import, layout identity and recompute decisions | Arbitrary KV formats are interchangeable; a token hash alone identifies physical bytes | +| Rust SDK | Protocol versions, typed hit/miss/error outcomes, deadlines, request IDs, cancellation and retry classification | Every operation is freely retryable; repeated Get is pin-free; an auth failure is a valid hit | +| Transport | Transfer of approved byte ranges and trustworthy completion reporting | Completion publishes a block; lease expiry proves DMA stopped; a CPU pointer is a GPU region | +| Cache service | Commit visibility, checksums, admission, eviction, isolation and epoch fencing | A transport library supplies membership, tenant policy or persistence automatically | +| Cold backend | RustFS object/manifest operations, retention and bounded refill | Cold storage is an obligatory P-to-D hop or repairs a lost hot block automatically | + +Keep GPU/DPU types and dependencies out of the portable core. A proposed foreign-language binding should be narrow and versioned; it is not a second implementation of identity or lifecycle rules. Compatibility mismatch remains a miss unless a future, independently tested conversion is explicitly accepted. + +## Performance and reliability gates + +Predeclare workload, sample count, cache state, concurrency, prompt/output lengths, reuse ratio, SLO and error policy before a paired run. Include population/write cost, cold misses, warm hits, no-reuse traffic, HBM pressure, noisy neighbors and failures. Report TTFT and TPOT/ITL distributions, SLO goodput, useful output tokens, host/DPU resources, network/PCIe traffic and total cost where measurable. + +PD separation is a scheduling option, not a universal throughput improvement. Native prefix caching, local offload and native PD transport remain comparison configurations. A DPU decision requires the NIC-RDMA baseline in [benchmarks](BENCHMARKS.md); TCP is not that baseline. Existing cost targets are hypotheses until the paired go gate is confirmed. + +The 10,000-GPU objective requires physical fleet evidence and a stated fault envelope, spare inference capacity and SLO. A 10,000-client simulator only tests its simulated workload. Cache node loss may correctly cause misses, extra prefill work and higher latency. Interrupted generation needs engine-level recovery/replay; preserving a cache API is not proof of uninterrupted decoding. + +## Evidence labels + +- `PASS` means the named executed check passed in its recorded environment. +- `SIMULATED` describes an executed CPU stand-in; it does not upgrade a hardware row. +- `NOT_RUN_HARDWARE` means required devices or a real path were not exercised. +- `NOT_RUN_LICENSE` and `NOT_RUN_BUDGET` preserve their existing meanings. +- `TCO_UNKNOWN`, `NOT_EVALUATED` and `NOT_CLAIMED` remain until supporting evidence exists. + +Do not rewrite unavailable results as zero timings, zero cost, an implicit pass, PB-scale verification or production readiness. [Replacement strategy](REPLACEMENT_STRATEGY.md) describes adoption without those claims. diff --git a/docs/V3_FAILURES.md b/docs/V3_FAILURES.md index 4afaf52..f0ae647 100644 --- a/docs/V3_FAILURES.md +++ b/docs/V3_FAILURES.md @@ -4,11 +4,11 @@ Each row is a clean miss. The engine recomputes and may fill again. The hook ret | Fault | What the hook does | Result | | --- | --- | --- | -| DPU restart | Drop in-flight tasks and drop slab slots | Clean miss. A new write can complete. Old slot ids are not readable | -| GPU reset | Hide every slot and drop in-flight tasks. Client completion is unknown, so bytes stay unreadable | Clean miss. New writes require remaining slab capacity | +| DPU restart | Explicitly fence simulated tasks and drop slab slots; keep task/slot ids monotonic | Clean miss. Old ids are not reusable. A lost control-plane authority still requires `restore_control` | +| GPU reset | Hide every slot and quarantine in-flight tasks. Completion is unknown, so credits and bytes stay charged | Clean miss. New writes require remaining slab and queue capacity | | NIC flap | Same hide. An in-flight transfer is not a hit | Clean miss. Hidden bytes can exhaust the slab | | Storage loss | Hide slots. Hidden bytes are not a durability replica and are not returned | Clean miss, then recompute | -| NVMe timeout | Drop one in-flight task. Do not expose it | Clean miss. Later submits still run | +| NVMe timeout | Quarantine all pending tasks because the fault carries no task id; preserve healthy committed slots | No incomplete payload is exposed. Queue/byte credits remain charged until explicit drain | | Control-plane loss | Hide slots and refuse new submits with `Cancelled` | No read while membership is unknown. `restore_control` permits admission if capacity remains. Old slots stay hidden and charged | | RustFS slow | Do not wait and do not call the object store | `RefillOutcome::Missing` | | RustFS unavailable | Same as slow | `RefillOutcome::Missing` | @@ -19,4 +19,8 @@ Lease expiry is still not proof that a copy stopped. The simulated slab does not This matrix is a CPU contract. It is not a fault campaign on BlueField, GPU, NIC, or RustFS hardware. -A full hidden slab has no drain/reclaim API yet. In this CPU simulation, an explicit DPU restart clears it. Do not translate that into permission to free real DMA buffers: hardware integration needs verified drain and device fencing. Pending write bytes and committed slab bytes share the configured byte cap. +`quarantine` retains a task whose completion is unknown. `attest_drained` releases only a quarantined task after the trusted queue owner confirms drain; a late `complete` or `fail` cannot publish or free it. `invalidate` and `reclaim_invalidated` are tenant-scoped. Active or quarantined readers prevent slot reclaim. `fail` describes a confirmed failed completion, not an unknown timeout. + +Pending writes reserve slab bytes and slot ids at admission. Task and slot metadata each have a hard ceiling of 4096. Pending/quarantined writes and committed/hidden slots share the configured byte cap. Returned payload vectors are caller-owned and are not a complete process-RSS bound. + +These explicit CPU drain/reclaim operations and restart fences are not permission to free real DMA buffers. Hardware integration still needs verified transport quiescence and device fencing. Other faults and DPU restart do not silently restore lost control authority. diff --git a/docs/V3_TEST_MATRIX.md b/docs/V3_TEST_MATRIX.md index b9f7fcc..650ab13 100644 --- a/docs/V3_TEST_MATRIX.md +++ b/docs/V3_TEST_MATRIX.md @@ -17,11 +17,15 @@ Default features do not need DOCA, SPDK, libnixl, or a DPU. `--features dpu-bf3` | --- | --- | --- | | `dpu-demo` simulated round trip, devices closed, EC off | Expected `DPU_CONTRACT_OK` with `offload_active=false` | CPU contract. Hardware `NOT_RUN_HARDWARE` | | Storage-queue submit, complete, fail, wrong tenant, checksum flip | Unit tests in `gpucache-dpu` | CPU contract | +| Unknown completion, late completion, repeated faults, explicit drain, tenant invalidate/reclaim and referenced slots | Deterministic storage-queue regression tests; unknown tasks retain credit | CPU contract, not hardware DMA quiescence | +| Task/slot ceilings, pending slot-id reservation, restart id fencing, control-loss preservation | Storage-queue regression tests | CPU contract | | Arm `start` does not set offload | Unit test | CPU contract | | Reactor id mismatch | `AffineQueue` unit test | Contract only. SPDK `NOT_RUN_HARDWARE` | | Ablation knobs and hot-tier EC refusal | Unit test | EC stays off | | Fault matrix | `StorageQueue::apply_fault` tests | CPU contract. Hardware faults `NOT_RUN_HARDWARE` | | ACL, replay, seal, secret rotation, audit contents | `gpucache-protocol` tests | CPU contract. mTLS `NOT_RUN` (not linked) | +| Replay capacity/clock rewind/domain isolation, retired secret revival, redacted credentials and malformed identities | Protocol and core-type regression tests | CPU hooks; not a TCP security rollout | +| TCP response operation, epoch, item count/order and request id association | Fake-peer regression tests; malformed exchanges poison the connection | Real localhost TCP with synthetic bytes | | Rate limit and noisy neighbor | `RateGate` tests | CPU contract. Caller clock, no sleep | | Rebalance, rolling upgrade, rollback, fault domain | `ControlPlane` tests | CPU contract. No payload copy | | G4 retention, verified delete, cold key retire, backfill cap, slow RustFS | `gpucache-cold` tests | Mock `PASS`. Live RustFS `NOT_RUN_HARDWARE` | @@ -32,4 +36,6 @@ Default features do not need DOCA, SPDK, libnixl, or a DPU. `--features dpu-bf3` | 24 hour soak, wall power, PB-scale | Not run | `NOT_RUN_HARDWARE` / `NOT_RUN_BUDGET` | | vLLM or SGLang KV round trip | Not run | `NOT_RUN_HARDWARE` | +The [VM runner](VM_REGRESSION.md) executes the required CPU gates and writes versioned JSON, JUnit and logs. Its own tests cover command failure, timeouts, descendant cleanup, exact output contracts, dependency-boundary bypasses and hardware-profile refusal. CPU regression forces the live RustFS endpoint empty, including overrides from Cargo configuration. + `status=DPU_CONTRACT_OK` means the simulation matched. It is not a device pass and not `production ready`. diff --git a/docs/VM_REGRESSION.md b/docs/VM_REGRESSION.md new file mode 100644 index 0000000..6822452 --- /dev/null +++ b/docs/VM_REGRESSION.md @@ -0,0 +1,110 @@ +# VM regression runbook + +This runbook is for a Grok bot or another operator running the repository inside a disposable VM. The runner collects reproducible evidence; it does not contact a bot, allocate a VM, buy hardware, deploy a service or publish results externally. + +The current executable profile is the CPU contract. Native RDMA, real engine KV export/import, GDS and BF3 offload are not implemented hardware paths in this tree. Their strict profiles must produce an explicit unavailable result and return nonzero. See [V3 test matrix](V3_TEST_MATRIX.md), [compatibility](../COMPATIBILITY.md) and [roadmap](ROADMAP.md). + +## Prerequisites + +- A clean checkout at a recorded commit, including `Cargo.lock` and `rust-toolchain.toml`. +- Linux x86_64 or aarch64, Python 3.9 or newer, Git, and a Rust toolchain with Cargo, rustfmt and Clippy. Follow the repository toolchain; the declared MSRV is 1.85. +- Sufficient RAM, CPU, disk space and a writable ordinary temporary directory for the build and file-store tests. Do not point test storage at a production disk or a device node. +- Cargo dependency access, or a prepopulated dependency cache for an offline VM. The Python runner requires no third-party Python package and no `pip install`. +- Loopback networking and child-process execution for TCP and import tests. No GPU, RDMA adapter, DOCA, NIXL or SPDK SDK is required for the CPU profile. + +Record resource limits when comparing runs. A VM process-duration comparison is not a GPU inference benchmark. An aarch64 or QEMU run remains CPU evidence, not BlueField evidence. + +## Run from the repository root + +```sh +python3 scripts/ci/vm_regression.py --profile cpu --output artifacts/vm-regression +``` + +The output path is relative to the calling directory unless an absolute path is given. It must be new or empty; the runner refuses to overwrite an earlier run. Keep the process exit code and the full artifact directory, including logs. Run `python3 scripts/ci/vm_regression.py --help` for the implemented command-line interface. + +| Option | Default | Use | +| --- | --- | --- | +| `--profile` | `cpu` | Select the required qualification profile | +| `--output` | `artifacts/vm-regression` | New or empty artifact directory | +| `--repo` | Repository containing the runner | Explicit checkout root, which must contain `Cargo.toml` | +| `--timeout` | `900` seconds | Positive, finite timeout for each gate; timed-out child process groups are killed | +| `--seed` | `20261017` | Recorded run seed; preserve any failing property-test case separately | +| `--offline` | Off | Pass Cargo `--offline`; dependencies must already be cached | + +The CPU profile should cover the repository-required checks: + +```sh +cargo fmt --all -- --check +cargo clippy --workspace --all-targets -- -D warnings +cargo test --workspace --all-targets +``` + +The runner uses `--locked` for Cargo resolution, checks Clippy with 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 16 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: + +```sh +cargo test -p gpucache-transport --all-targets --features rdma-nixl +cargo test -p gpucache-dpu --all-targets --features dpu-bf3 +``` + +## Artifact contract + +| Path | Contents | Interpretation | +| --- | --- | --- | +| `environment.json` | Repository revision/state, OS/architecture, available toolchain versions and run metadata | An observed environment manifest, not a certification or a dump of all environment variables | +| `capabilities.json` | CPU and optional path availability, including simulation and unrun hardware limitations | An installed tool or visible device is not proof that the data path executed | +| `results.json` | Selected profile, each command/check, exit/status, duration and log references | Required selected-profile failures remain failures; duration is command wall time, not TTFT/TPOT | +| `summary.md` | Human-readable outcome, failed/unavailable checks and evidence limits | The VM operator can review this first, then verify raw results/logs | +| `junit.xml` | One test case per executed required gate, with failure status and log reference | Additional CI-compatible view; unavailable selected hardware qualification is a failure, not a skipped pass | +| `logs/` | Per-command stdout/stderr and diagnostic output | Retain the evidence behind a result; no credentials, prompts or KV payloads should be added | + +JSON reports carry `schema_version: 1`; consumers should reject incompatible versions rather than guess fields. `results.json` uses `required_gates` for the ordered rows. CPU child processes have live RustFS credentials removed and the endpoint forced empty. Every Cargo gate additionally overrides the endpoint with highest-precedence `--config`, so checkout or global Cargo `[env]` settings cannot silently enable a live probe. Environment reporting does not dump those variables. + +Do not infer a result from a missing file or empty log. A runner failure before complete reporting is an incomplete run. Do not manufacture results for commands that never started. Intermediate reports use `RUNNING`; that is not a successful completed run. + +## Profile and exit policy + +| Profile | Current behavior | Required outcome | +| --- | --- | --- | +| `cpu` | Executes portable checks and demos | Zero only when every required CPU check passes; preserve independent `NOT_RUN_HARDWARE` rows | +| `gpu-engine` | Real engine adapter and GPU KV round trip unavailable | `NOT_RUN_HARDWARE`, exit 2 | +| `rdma` | Native NIXL/RDMA backend is not linked | `NOT_RUN_HARDWARE`, exit 2 | +| `gds` | cuFile/GDS path is not linked | `NOT_RUN_HARDWARE`, exit 2 | +| `bf3` | DOCA/SPDK/NVMe device path is not linked | `NOT_RUN_HARDWARE`, exit 2 | + +For example, this command must not silently pass on the current tree: + +```sh +python3 scripts/ci/vm_regression.py --profile bf3 --output artifacts/vm-regression-bf3 +``` + +Selecting a strict hardware profile must not fall back to TCP, CPU simulation, Soft-RoCE or a stub and report success. A detected GPU/DPU or a successful SDK import cannot override an unimplemented path. Unsupported profiles can still write useful reports before returning nonzero. + +A missing tool, timeout, crashed child, failed assertion or invalid invocation returns nonzero. A failed CPU run exits 1; a passed CPU run exits 0. Invalid arguments exit nonzero before running gates and need not produce an artifact directory. A CPU pass means the required CPU checks passed in that environment; it does not mean that unavailable hardware checks passed. Keep `TCO_UNKNOWN`, `go_gate=NOT_EVALUATED` and the absence of performance samples intact. + +## What to inspect + +1. Confirm the revision and environment match the requested test. Note any dirty checkout; do not label a modified tree as the clean commit. +2. Inspect the selected profile and process exit code. Confirm failures, timeouts and unsupported selected profiles are visible in the summary and results. +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. + +## Reruns and failure handling + +Fix a demonstrated defect, then rerun the affected check and required suite. Keep the failed run separately from the corrected run; do not overwrite failure evidence with a later success. Infrastructure failures and implementation failures may need different fixes, but neither should be rewritten as a pass. + +Do not change firmware, BFB, IOMMU/ACS, DPU mode, NIC setup, production mounts or NVMe formatting to make the CPU profile pass. A future hardware campaign needs its own environment manifest and acceptance evidence. Existing [DPU targets](V3_DPU_TARGETS.md) list unset targets, not VM setup commands. + +## Proposed hardware qualification + +The following is **PROPOSED / NOT IMPLEMENTED**. A later native runner may support device-backed execution after the corresponding backend exists. Its report must identify the actual selected path and retain raw measurements, topology and fault results. + +- GPU engine: pinned engine/model, real KV export/import, declared deterministic output criterion, baseline/candidate configurations and fallback cases. +- RDMA: actual HCA/driver/firmware, registered regions, access bounds, completion/drain and NIC-failure evidence. Soft-RoCE can be a labeled functional test, not the native performance baseline. +- GDS: CUDA/cuFile, filesystem/device support, direct versus compatibility/fallback path, alignment and integrity evidence. +- 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. From f02dcc3eb356a7679486c42e64f09a4677620e23 Mon Sep 17 00:00:00 2001 From: loverustfs Date: Sat, 17 Oct 2026 15:06:03 +0800 Subject: [PATCH 4/4] test: bootstrap clean VM dependencies and check default features --- docs/VM_REGRESSION.md | 2 +- scripts/ci/tests/test_vm_regression.py | 5 +++++ scripts/ci/vm_regression.py | 3 +++ 3 files changed, 9 insertions(+), 1 deletion(-) diff --git a/docs/VM_REGRESSION.md b/docs/VM_REGRESSION.md index 6822452..4036f17 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, checks Clippy with 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 16 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 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. Manual reproduction of the optional feature contracts is: diff --git a/scripts/ci/tests/test_vm_regression.py b/scripts/ci/tests/test_vm_regression.py index 3504c1d..566dda9 100644 --- a/scripts/ci/tests/test_vm_regression.py +++ b/scripts/ci/tests/test_vm_regression.py @@ -130,6 +130,11 @@ def test_required_gates_are_unique_and_cover_features_and_demos(self): "fault-demo", "cold-demo", "dpu-demo", "v3-paired"): self.assertIn(name, names) + def test_clean_vm_fetches_locked_dependencies_before_offline_metadata(self): + names = [gate.name for gate in vm.cpu_gates()] + self.assertLess(names.index("locked-dependencies"), names.index("dependency-contract")) + self.assertIn("clippy-default", names) + def test_invalid_timeouts_are_rejected(self): for value in ("0", "-1", "nan", "inf"): with self.assertRaises(vm.argparse.ArgumentTypeError): diff --git a/scripts/ci/vm_regression.py b/scripts/ci/vm_regression.py index a6eb75e..ab452ac 100644 --- a/scripts/ci/vm_regression.py +++ b/scripts/ci/vm_regression.py @@ -50,10 +50,13 @@ def cpu_gates(offline=False): locked = ("--locked",) + (("--offline",) if offline else ()) gates = [ Gate("format", cargo + ("fmt", "--all", "--", "--check")), + Gate("locked-dependencies", cargo + ("fetch",) + locked), Gate("dependency-contract", (sys.executable, "scripts/ci/check_contract.py"), ("status=CPU_DEPENDENCY_CONTRACT_OK",)), Gate("runner-tests", (sys.executable, "-m", "unittest", "discover", "-s", "scripts/ci/tests", "-v")), + Gate("clippy-default", cargo + ("clippy", "--workspace", "--all-targets") + + locked + ("--", "-D", "warnings")), Gate("clippy", cargo + ("clippy", "--workspace", "--all-targets", "--all-features") + locked + ("--", "-D", "warnings")), Gate("workspace-tests", cargo + ("test", "--workspace", "--all-targets") + locked),