diff --git a/crates/cold/src/failure.rs b/crates/cold/src/failure.rs new file mode 100644 index 0000000..8670956 --- /dev/null +++ b/crates/cold/src/failure.rs @@ -0,0 +1,28 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 + +//! Object-store failures that indicate temporary unavailability. Corruption, +//! invalid requests, authorization failures and unknown labels fail closed. + +use gpucache_core_types::CacheError; + +pub(crate) fn transient(error: &CacheError) -> bool { + matches!( + error, + CacheError::TimedOut + | CacheError::Storage { + reason: "object store" | "object unavailable" + } + | CacheError::QuotaExceeded { .. } + ) +} + +pub(crate) fn refill_miss(error: &CacheError) -> bool { + transient(error) + || matches!( + error, + CacheError::Storage { + reason: "object missing" + } + ) +} diff --git a/crates/cold/src/http.rs b/crates/cold/src/http.rs index 63e03d7..2014a9a 100644 --- a/crates/cold/src/http.rs +++ b/crates/cold/src/http.rs @@ -5,7 +5,7 @@ use std::io::{Read, Write}; use std::net::TcpStream; use std::time::{Duration, SystemTime, UNIX_EPOCH}; -use gpucache_core_types::{content_checksum, CacheError}; +use gpucache_core_types::{content_checksum, CacheError, QuotaKind}; use crate::pack::{contract_smoke, ObjectMeta, ObjectStore}; use crate::sigv4::{authorization, format_amz_time, hex, sha256, SignInput}; @@ -154,11 +154,7 @@ impl ObjectStore for HttpStore { fn put(&mut self, key: &str, body: &[u8]) -> Result<(), CacheError> { let key = self.full_key(key)?; let status = request(self, "PUT", &key, body, None, false)?; - if !(200..300).contains(&status.code) { - return Err(CacheError::Storage { - reason: "object store", - }); - } + check_status(status.code)?; Ok(()) } @@ -170,17 +166,9 @@ impl ObjectStore for HttpStore { reason: "object missing", }); } - if !(200..300).contains(&status.code) { - return Err(CacheError::Storage { - reason: "object store", - }); - } + check_status(status.code)?; let fetched = request(self, "GET", &key, b"", None, true)?; - if !(200..300).contains(&fetched.code) { - return Err(CacheError::Storage { - reason: "object store", - }); - } + check_status(fetched.code)?; if let Some(len) = status.content_length { if len != fetched.body.len() as u64 { return Err(CacheError::LengthMismatch { @@ -200,11 +188,7 @@ impl ObjectStore for HttpStore { let end = offset.saturating_add(length).saturating_sub(1); let header = format!("bytes={offset}-{end}"); let status = request(self, "GET", &key, b"", Some(&header), true)?; - if !(200..300).contains(&status.code) { - return Err(CacheError::Storage { - reason: "object store", - }); - } + check_status(status.code)?; if status.body.len() as u64 != length { return Err(CacheError::LengthMismatch { expected: length, @@ -220,9 +204,28 @@ impl ObjectStore for HttpStore { if status.code == 404 || (200..300).contains(&status.code) { return Ok(()); } - Err(CacheError::Storage { - reason: "object store", - }) + check_status(status.code) + } +} + +/// Preserve permanent authentication/request failures for sink classification. +fn check_status(code: u16) -> Result<(), CacheError> { + match code { + 200..=299 => Ok(()), + 401 | 403 => Err(CacheError::Unauthorized), + 404 => Err(CacheError::Storage { + reason: "object missing", + }), + 408 => Err(CacheError::TimedOut), + 429 => Err(CacheError::QuotaExceeded { + kind: QuotaKind::RateLimit, + }), + 500..=599 => Err(CacheError::Storage { + reason: "object unavailable", + }), + _ => Err(CacheError::Malformed { + reason: "object response", + }), } } @@ -324,7 +327,7 @@ fn request( if err.kind() == std::io::ErrorKind::WouldBlock || err.kind() == std::io::ErrorKind::TimedOut => { - break; + return Err(CacheError::TimedOut); } Err(_) => { return Err(CacheError::Storage { @@ -337,32 +340,52 @@ fn request( } fn parse_response(raw: &[u8], read_body: bool) -> Result { - let text = std::str::from_utf8(raw).map_err(|_| CacheError::Storage { - reason: "object store", - })?; - let (head, body) = text.split_once("\r\n\r\n").unwrap_or((text, "")); - let mut lines = head.lines(); - let status = lines.next().ok_or(CacheError::Storage { - reason: "object store", - })?; - let code = status - .split_whitespace() - .nth(1) + let malformed = || CacheError::Malformed { + reason: "object response", + }; + // KV packs contain arbitrary bytes; only the HTTP header is UTF-8 text. + let boundary = raw + .windows(4) + .position(|bytes| bytes == b"\r\n\r\n") + .ok_or_else(malformed)?; + let text = std::str::from_utf8(&raw[..boundary]).map_err(|_| malformed())?; + let mut lines = text.split("\r\n"); + let status = lines.next().ok_or_else(malformed)?; + let mut fields = status.split_whitespace(); + if !matches!(fields.next(), Some("HTTP/1.0" | "HTTP/1.1")) { + return Err(malformed()); + } + let code = fields + .next() .and_then(|code| code.parse::().ok()) - .ok_or(CacheError::Storage { - reason: "object store", - })?; + .filter(|code| (100..=599).contains(code)) + .ok_or_else(malformed)?; let mut content_length = None; for line in lines { - let Some((name, value)) = line.split_once(':') else { - continue; - }; + let (name, value) = line.split_once(':').ok_or_else(malformed)?; + if name.eq_ignore_ascii_case("transfer-encoding") { + // The dependency-free smoke client does not decode chunked bodies. + return Err(malformed()); + } if name.eq_ignore_ascii_case("content-length") { - content_length = value.trim().parse::().ok(); + if content_length.is_some() { + return Err(malformed()); + } + content_length = Some(value.trim().parse::().map_err(|_| malformed())?); } } - let body = if read_body { - body.as_bytes().to_vec() + let received = &raw[boundary + 4..]; + let body = if read_body && (200..300).contains(&code) { + // An incomplete response, including a read timeout after headers, must + // never yield a truncated success or consume bytes past the declared body. + let expected = content_length.ok_or_else(malformed)?; + if received.len() as u64 != expected { + return Err(CacheError::LengthMismatch { + expected, + actual: received.len() as u64, + }); + } + received.to_vec() } else { Vec::new() }; @@ -421,3 +444,56 @@ mod tests { ); } } + +#[cfg(test)] +mod response_tests { + use super::*; + + #[test] + fn http_status_preserves_permanent_and_transient_backend_failures() { + for code in [401, 403] { + assert_eq!(check_status(code), Err(CacheError::Unauthorized)); + } + for code in [400, 301, 409, 413, 416] { + let error = check_status(code).unwrap_err(); + assert!(!crate::failure::transient(&error)); + } + for code in [408, 429, 500, 502, 503, 504] { + let error = check_status(code).unwrap_err(); + assert!(crate::failure::transient(&error)); + assert!(crate::failure::refill_miss(&error)); + } + assert!(crate::failure::refill_miss(&check_status(404).unwrap_err())); + assert_eq!(check_status(204), Ok(())); + // Error status classification does not require a success payload body. + let status = parse_response(b"HTTP/1.1 503 Unavailable\r\n\r\n", true).unwrap(); + assert_eq!( + check_status(status.code), + Err(CacheError::Storage { + reason: "object unavailable" + }) + ); + } + + #[test] + fn response_parser_preserves_binary_bodies_and_requires_complete_framing() { + let mut response = b"HTTP/1.1 206 Partial Content\r\nContent-Length: 4\r\n\r\n".to_vec(); + response.extend_from_slice(&[0, 255, 128, 42]); + let parsed = parse_response(&response, true).unwrap(); + assert_eq!(parsed.code, 206); + assert_eq!(parsed.body, [0, 255, 128, 42]); + for malformed in [ + b"HTTP/1.1 200 OK\r\nContent-Length: 1".as_slice(), + b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nx", + b"HTTP/1.1 200 OK\r\nContent-Length: 1\r\n\r\nxx", + b"HTTP/1.1 200 OK\r\nContent-Length: 1\r\nContent-Length: 1\r\n\r\nx", + b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n0\r\n\r\n", + b"HTTP/1.1 200 OK\r\n\r\nx", + b"HTTP/1.1 200 OK\r\nContent-Length: garbage\r\n\r\nx", + ] { + assert!(parse_response(malformed, true).is_err()); + } + // HEAD describes object length but contains no body. + assert!(parse_response(b"HTTP/1.1 200 OK\r\nContent-Length: 900\r\n\r\n", false).is_ok()); + } +} diff --git a/crates/cold/src/lib.rs b/crates/cold/src/lib.rs index 900a785..d2df336 100644 --- a/crates/cold/src/lib.rs +++ b/crates/cold/src/lib.rs @@ -10,6 +10,7 @@ #![forbid(unsafe_code)] mod demo; +mod failure; mod http; mod mock; mod ops; @@ -24,6 +25,7 @@ pub use mock::MockObjectStore; pub use ops::{ enqueue_backfill, refill_degraded, refill_for_tenant, verify_deleted, BackfillLimit, ColdAvailability, ColdKeys, ColdOps, ObjectRecord, ReclaimStep, Retention, + MAX_COLD_KEY_HISTORY, MAX_COLD_RECORDS, MAX_COLD_TENANTS, }; pub use pack::{ contract_smoke, manifest_version, pack_blocks, read_manifest, Manifest, ObjectMeta, diff --git a/crates/cold/src/ops.rs b/crates/cold/src/ops.rs index 824bc43..5281d89 100644 --- a/crates/cold/src/ops.rs +++ b/crates/cold/src/ops.rs @@ -6,8 +6,9 @@ //! //! Live RustFS stays outside this module. A slow or unavailable cold tier //! returns [`RefillOutcome::Missing`] without reading the store, so the -//! caller recomputes and is not blocked. Expired objects are deleted only -//! after the following head shows them gone. +//! caller recomputes and is not blocked. Explicit retention reclaim is complete +//! only after a following HEAD shows the object gone. Unknown objects are never +//! discovered or collected automatically. use std::collections::{BTreeMap, VecDeque}; @@ -17,6 +18,11 @@ use crate::pack::{key_ok, ObjectStore, MANIFEST_VERSION}; use crate::refill::{refill_generation, refill_matching, RefillBudget, RefillOutcome}; use crate::sink::{AsyncSink, ColdJob}; +/// Hard metadata ceilings, independent of packed payload byte budgets. +pub const MAX_COLD_TENANTS: usize = 1024; +pub const MAX_COLD_KEY_HISTORY: usize = 4096; +pub const MAX_COLD_RECORDS: usize = 4096; + #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub struct Retention { max_age_ms: u64, @@ -126,6 +132,8 @@ struct ColdKey { pub struct ColdKeys { items: Vec, next: BTreeMap, u64>, + max_tenants: usize, + max_history: usize, } impl ColdKeys { @@ -133,7 +141,28 @@ impl ColdKeys { Self { items: Vec::new(), next: BTreeMap::new(), + max_tenants: MAX_COLD_TENANTS, + max_history: MAX_COLD_KEY_HISTORY, + } + } + + /// Lower local limits for metadata. Retired generations remain charged: + /// dropping their history must never make a retired credential valid again. + pub fn with_limits(max_tenants: usize, max_history: usize) -> Result { + if max_tenants == 0 + || max_tenants > MAX_COLD_TENANTS + || max_history == 0 + || max_history > MAX_COLD_KEY_HISTORY + { + return Err(CacheError::Malformed { + reason: "cold key limits", + }); } + Ok(Self { + max_tenants, + max_history, + ..Self::new() + }) } pub fn issue(&mut self, tenant: &[u8]) -> Result { @@ -141,7 +170,9 @@ impl ColdKeys { if self.next.get(tenant).copied().unwrap_or(0) != 0 { return Err(CacheError::IdempotencyConflict); } - Ok(self.insert(tenant, KeyState::Current)) + let generation = self.prepare_insert(tenant)?; + self.insert(tenant, generation); + Ok(generation) } pub fn rotate(&mut self, tenant: &[u8]) -> Result { @@ -153,12 +184,15 @@ impl ColdKeys { { return Err(CacheError::UnknownGeneration); } + // Capacity and generation exhaustion must not demote the current key. + let generation = self.prepare_insert(tenant)?; for item in &mut self.items { if item.tenant.as_slice() == tenant && item.state == KeyState::Current { item.state = KeyState::Overlap; } } - Ok(self.insert(tenant, KeyState::Current)) + self.insert(tenant, generation); + Ok(generation) } pub fn retire(&mut self, tenant: &[u8], generation: u64) -> Result<(), CacheError> { @@ -211,20 +245,31 @@ impl ColdKeys { Ok(()) } - fn insert(&mut self, tenant: &[u8], state: KeyState) -> u64 { - let generation = self - .next + fn prepare_insert(&self, tenant: &[u8]) -> Result { + if self.items.len() >= self.max_history + || (!self.next.contains_key(tenant) && self.next.len() >= self.max_tenants) + { + return Err(CacheError::QuotaExceeded { + kind: QuotaKind::InFlightRequests, + }); + } + self.next .get(tenant) .copied() .unwrap_or(0) - .saturating_add(1); + .checked_add(1) + .ok_or(CacheError::FencingRejected { + reason: "cold key generation exhausted", + }) + } + + fn insert(&mut self, tenant: &[u8], generation: u64) { self.next.insert(tenant.to_vec(), generation); self.items.push(ColdKey { tenant: tenant.to_vec(), generation, - state, + state: KeyState::Current, }); - generation } } @@ -239,6 +284,7 @@ pub struct ColdOps { records: BTreeMap, reclaim: VecDeque, keys: ColdKeys, + max_records: usize, } impl ColdOps { @@ -248,9 +294,24 @@ impl ColdOps { records: BTreeMap::new(), reclaim: VecDeque::new(), keys: ColdKeys::new(), + max_records: MAX_COLD_RECORDS, } } + /// Bound tracked objects and, consequently, their deduplicated reclaim queue. + /// Failed reclaim keeps its record and its credit until verified completion. + pub fn with_capacity(retention: Retention, max_records: usize) -> Result { + if max_records == 0 || max_records > MAX_COLD_RECORDS { + return Err(CacheError::Malformed { + reason: "cold record limits", + }); + } + Ok(Self { + max_records, + ..Self::new(retention) + }) + } + pub fn keys(&self) -> &ColdKeys { &self.keys } @@ -270,6 +331,13 @@ impl ColdOps { } self.keys .allow_write(&record.tenant, record.key_generation)?; + if !self.records.contains_key(&record.key) && self.records.len() >= self.max_records { + return Err(CacheError::QuotaExceeded { + kind: QuotaKind::InFlightRequests, + }); + } + // A renewed record invalidates an earlier expiration decision. + self.reclaim.retain(|key| key != &record.key); self.records.insert(record.key.clone(), record); Ok(()) } @@ -294,36 +362,36 @@ impl ColdOps { added } + /// Attempt one explicitly scheduled retention record. Every failure retains + /// its record and moves it to the tail, so a bad backend key cannot starve + /// healthy keys. A checksum mismatch never authorizes deleting other bytes. pub fn reclaim_one(&mut self, store: &mut dyn ObjectStore) -> Result { let Some(key) = self.reclaim.pop_front() else { return Ok(ReclaimStep::Idle); }; - let Some(record) = self.records.get(&key).cloned() else { + let Some(record) = self.records.get(&key) else { return Ok(ReclaimStep::Idle); }; - match store.head(&key) { + let result = match store.head(&key) { Err(CacheError::Storage { reason: "object missing", - }) => { + }) => Ok(()), + Err(err) => Err(err), + Ok(meta) if meta.checksum != record.checksum => Err(CacheError::ChecksumMismatch), + Ok(_) => store + .delete(&key) + .and_then(|()| verify_deleted(store, &key)), + }; + match result { + Ok(()) => { self.records.remove(&key); - return Ok(ReclaimStep::Verified); + Ok(ReclaimStep::Verified) } - Err(err) => return Err(err), - Ok(meta) if meta.checksum != record.checksum => { - store.delete(&key)?; - verify_deleted(store, &key)?; - self.records.remove(&key); - return Err(CacheError::ChecksumMismatch); + Err(err) => { + self.reclaim.push_back(key); + Err(err) } - Ok(_) => {} } - store.delete(&key)?; - if let Err(err) = verify_deleted(store, &key) { - self.reclaim.push_front(key); - return Err(err); - } - self.records.remove(&key); - Ok(ReclaimStep::Verified) } } @@ -553,3 +621,7 @@ mod tests { } } } + +#[cfg(test)] +#[path = "reclaim_tests.rs"] +mod reclaim_tests; diff --git a/crates/cold/src/pack.rs b/crates/cold/src/pack.rs index 1c74c40..cfb8ca2 100644 --- a/crates/cold/src/pack.rs +++ b/crates/cold/src/pack.rs @@ -57,6 +57,13 @@ pub struct Pack { pub ranges: Vec, } +/// Trusted backend capability, separate from the authenticated hot path. +/// +/// Adapters report transient I/O failures as `TimedOut`, `Storage` with reason +/// "object store" or "object unavailable", or `QuotaExceeded`. Missing reads +/// use `Storage { reason: "object missing" }`. Authorization, malformed input, +/// corruption, and other storage labels are permanent write failures. A failed +/// PUT may already be visible: consumers must not infer permission to DELETE. pub trait ObjectStore { fn put(&mut self, key: &str, body: &[u8]) -> Result<(), CacheError>; fn head(&self, key: &str) -> Result; diff --git a/crates/cold/src/reclaim_tests.rs b/crates/cold/src/reclaim_tests.rs new file mode 100644 index 0000000..6309454 --- /dev/null +++ b/crates/cold/src/reclaim_tests.rs @@ -0,0 +1,226 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 + +use super::*; +use crate::{MockObjectStore, ObjectMeta}; +use gpucache_core_types::content_checksum; +use std::cell::Cell; + +struct FaultStore { + inner: MockObjectStore, + head_fault: usize, + heads: Cell, + delete_fault: bool, + deletes: usize, +} + +impl ObjectStore for FaultStore { + fn put(&mut self, key: &str, body: &[u8]) -> Result<(), CacheError> { + self.inner.put(key, body) + } + fn head(&self, key: &str) -> Result { + if key == "a" { + self.heads.set(self.heads.get() + 1); + if self.heads.get() == self.head_fault { + return Err(CacheError::TimedOut); + } + } + self.inner.head(key) + } + fn range_get(&self, key: &str, offset: u64, length: u64) -> Result, CacheError> { + self.inner.range_get(key, offset, length) + } + fn delete(&mut self, key: &str) -> Result<(), CacheError> { + self.deletes += 1; + if key == "a" && self.delete_fault { + self.delete_fault = false; + return Err(CacheError::TimedOut); + } + self.inner.delete(key) + } +} + +fn record(key: &str, generation: u64, written_ms: u64) -> ObjectRecord { + ObjectRecord { + key: key.to_string(), + written_ms, + checksum: content_checksum(b"body"), + tenant: b"tenant".to_vec(), + key_generation: generation, + manifest_version: MANIFEST_VERSION, + } +} + +#[test] +fn reclaim_head_delete_and_verification_failures_preserve_work_and_allow_fair_progress() { + for (head_fault, delete_fault) in [(1, false), (0, true), (2, false)] { + let mut store = FaultStore { + inner: MockObjectStore::default(), + head_fault, + heads: Cell::new(0), + delete_fault, + deletes: 0, + }; + let mut ops = ColdOps::new(Retention::new(10).unwrap()); + let generation = ops.keys_mut().issue(b"tenant").unwrap(); + for key in ["a", "b"] { + store.put(key, b"body").unwrap(); + ops.record_put(record(key, generation, 0)).unwrap(); + } + assert_eq!(ops.schedule_expired(10), 2); + assert_eq!(ops.reclaim_one(&mut store), Err(CacheError::TimedOut)); + assert_eq!(ops.reclaim.len(), 2); + assert_eq!(ops.records.len(), 2); + assert_eq!(ops.schedule_expired(10), 0); + assert_eq!(ops.reclaim_one(&mut store).unwrap(), ReclaimStep::Verified); + assert!(store.inner.head("b").is_err()); + assert!(ops.records.contains_key("a")); + assert_eq!(ops.reclaim_one(&mut store).unwrap(), ReclaimStep::Verified); + assert!(store.inner.head("a").is_err()); + assert!(ops.records.is_empty()); + assert_eq!(ops.reclaim_one(&mut store).unwrap(), ReclaimStep::Idle); + } +} + +#[test] +fn mismatched_reclaim_record_never_deletes_unexpected_content() { + let mut store = FaultStore { + inner: MockObjectStore::default(), + head_fault: 0, + heads: Cell::new(0), + delete_fault: false, + deletes: 0, + }; + let mut ops = ColdOps::new(Retention::new(10).unwrap()); + let generation = ops.keys_mut().issue(b"tenant").unwrap(); + store.put("a", b"different-owner-or-corrupt-body").unwrap(); + store.put("b", b"body").unwrap(); + ops.record_put(record("a", generation, 0)).unwrap(); + ops.record_put(record("b", generation, 0)).unwrap(); + ops.schedule_expired(10); + assert_eq!( + ops.reclaim_one(&mut store), + Err(CacheError::ChecksumMismatch) + ); + assert_eq!(store.deletes, 0); + assert_eq!(ops.reclaim_one(&mut store).unwrap(), ReclaimStep::Verified); + assert_eq!( + ops.reclaim_one(&mut store), + Err(CacheError::ChecksumMismatch) + ); + assert_eq!(store.deletes, 1); + assert_eq!(ops.records.len(), 1); + assert_eq!(ops.reclaim.len(), 1); + assert_eq!( + store.inner.head("a").unwrap().checksum, + content_checksum(b"different-owner-or-corrupt-body") + ); +} + +#[test] +fn renewing_a_record_invalidates_its_previously_scheduled_expiration() { + let mut store = MockObjectStore::default(); + let mut ops = ColdOps::new(Retention::new(10).unwrap()); + let generation = ops.keys_mut().issue(b"tenant").unwrap(); + store.put("a", b"body").unwrap(); + ops.record_put(record("a", generation, 0)).unwrap(); + assert_eq!(ops.schedule_expired(10), 1); + ops.record_put(record("a", generation, 10)).unwrap(); + assert_eq!(ops.reclaim_one(&mut store).unwrap(), ReclaimStep::Idle); + assert!(store.head("a").is_ok()); + assert_eq!(ops.schedule_expired(19), 0); + assert_eq!(ops.schedule_expired(20), 1); + assert_eq!(ops.reclaim_one(&mut store).unwrap(), ReclaimStep::Verified); +} + +#[test] +fn cold_key_tenant_and_history_admission_is_bounded_and_atomic() { + let mut keys = ColdKeys::with_limits(1, 2).unwrap(); + let first = keys.issue(b"tenant").unwrap(); + assert!(matches!( + keys.issue(b"other"), + Err(CacheError::QuotaExceeded { .. }) + )); + assert_eq!((keys.next.len(), keys.items.len()), (1, 1)); + assert!(!keys.next.contains_key(b"other".as_slice())); + let second = keys.rotate(b"tenant").unwrap(); + keys.retire(b"tenant", first).unwrap(); + for _ in 0..128 { + assert!(matches!( + keys.rotate(b"tenant"), + Err(CacheError::QuotaExceeded { .. }) + )); + assert_eq!((keys.next.len(), keys.items.len()), (1, 2)); + assert_eq!(keys.next[b"tenant".as_slice()], second); + assert_eq!(keys.allow_write(b"tenant", second), Ok(())); + assert_eq!( + keys.allow_read(b"tenant", first), + Err(CacheError::Unauthorized) + ); + } + assert_eq!(keys.issue(b"tenant"), Err(CacheError::IdempotencyConflict)); + assert!(ColdKeys::with_limits(0, 1).is_err()); + assert!(ColdKeys::with_limits(1, 0).is_err()); + assert!(ColdKeys::with_limits(MAX_COLD_TENANTS + 1, 1).is_err()); + assert!(ColdKeys::with_limits(1, MAX_COLD_KEY_HISTORY + 1).is_err()); +} + +#[test] +fn exhausted_cold_key_generation_does_not_demote_current_authority() { + let mut keys = ColdKeys::with_limits(1, 2).unwrap(); + let current = keys.issue(b"tenant").unwrap(); + keys.next.insert(b"tenant".to_vec(), u64::MAX); + assert_eq!( + keys.rotate(b"tenant"), + Err(CacheError::FencingRejected { + reason: "cold key generation exhausted" + }) + ); + assert_eq!(keys.allow_write(b"tenant", current), Ok(())); + assert_eq!(keys.items.len(), 1); + assert_eq!(keys.next[b"tenant".as_slice()], u64::MAX); +} + +#[test] +fn full_record_table_keeps_failed_reclaim_credit_and_allows_replacement() { + let mut store = FaultStore { + inner: MockObjectStore::default(), + head_fault: 1, + heads: Cell::new(0), + delete_fault: false, + deletes: 0, + }; + let retention = Retention::new(10).unwrap(); + let mut ops = ColdOps::with_capacity(retention, 2).unwrap(); + let generation = ops.keys_mut().issue(b"tenant").unwrap(); + for key in ["a", "b"] { + store.put(key, b"body").unwrap(); + ops.record_put(record(key, generation, 0)).unwrap(); + } + assert_eq!(ops.schedule_expired(10), 2); + assert_eq!(ops.reclaim_one(&mut store), Err(CacheError::TimedOut)); + let scheduled = ops.reclaim.clone(); + for _ in 0..128 { + assert!(matches!( + ops.record_put(record("c", generation, 0)), + Err(CacheError::QuotaExceeded { .. }) + )); + assert_eq!(ops.records.len(), 2); + assert_eq!(ops.reclaim, scheduled); + assert_eq!(ops.schedule_expired(10), 0); + } + // An authenticated replacement does not grow metadata at capacity and must + // cancel its stale expiry. Failed admissions above did not alter this queue. + ops.record_put(record("a", generation, 100)).unwrap(); + assert_eq!(ops.records.len(), 2); + assert_eq!(ops.reclaim.len(), 1); + assert_eq!(ops.reclaim_one(&mut store).unwrap(), ReclaimStep::Verified); + assert!(store.head("a").is_ok()); + ops.record_put(record("c", generation, 100)).unwrap(); + assert_eq!(ops.records.len(), 2); + assert_eq!(ops.schedule_expired(110), 2); + assert_eq!(ops.schedule_expired(110), 0); + assert_eq!(ops.reclaim.len(), 2); + assert!(ColdOps::with_capacity(retention, 0).is_err()); + assert!(ColdOps::with_capacity(retention, MAX_COLD_RECORDS + 1).is_err()); +} diff --git a/crates/cold/src/refill.rs b/crates/cold/src/refill.rs index 34b1ac8..7c51e9e 100644 --- a/crates/cold/src/refill.rs +++ b/crates/cold/src/refill.rs @@ -63,6 +63,9 @@ pub fn refill_generation( /// Match tenant and complete canonical identity before any payload download. /// Tenant mismatch is Unauthorized; other identity mismatch is a clean miss. /// Legacy manifests lack the required binding and are clean misses too. +/// Missing objects, backend timeouts, unavailability and backpressure are clean +/// misses at every read stage; validation and authorization errors stay errors. +/// This read-only operation never retries, publishes, or removes an object. pub fn refill_matching( store: &dyn ObjectStore, manifest_key: &str, @@ -76,15 +79,17 @@ pub fn refill_matching( } let meta = match store.head(manifest_key) { Ok(meta) => meta, - Err(CacheError::Storage { - reason: "object missing", - }) => return Ok(RefillOutcome::Missing), + Err(err) if crate::failure::refill_miss(&err) => return Ok(RefillOutcome::Missing), Err(err) => return Err(err), }; if meta.len > budget.max_bytes { return Ok(RefillOutcome::OverBudget); } - let manifest_bytes = store.range_get(manifest_key, 0, meta.len)?; + let manifest_bytes = match store.range_get(manifest_key, 0, meta.len) { + Ok(bytes) => bytes, + Err(err) if crate::failure::refill_miss(&err) => return Ok(RefillOutcome::Missing), + Err(err) => return Err(err), + }; if manifest_bytes.len() as u64 != meta.len || content_checksum(&manifest_bytes) != meta.checksum { return Err(CacheError::ChecksumMismatch); @@ -117,9 +122,7 @@ pub fn refill_matching( } let payload = match store.range_get(&manifest.object_key, range.offset, range.length) { Ok(payload) => payload, - Err(CacheError::Storage { - reason: "object missing", - }) => return Ok(RefillOutcome::Missing), + Err(err) if crate::failure::refill_miss(&err) => return Ok(RefillOutcome::Missing), Err(err) => return Err(err), }; if payload.len() as u64 != range.length || content_checksum(&payload) != range.checksum { diff --git a/crates/cold/src/sink.rs b/crates/cold/src/sink.rs index 5904962..0db165a 100644 --- a/crates/cold/src/sink.rs +++ b/crates/cold/src/sink.rs @@ -173,7 +173,7 @@ impl AsyncSink { if report.first_error.is_none() { report.first_error = Some(error.clone()); } - if retryable(&error) && pending.attempts < MAX_WRITE_ATTEMPTS { + if crate::failure::transient(&error) && pending.attempts < MAX_WRITE_ATTEMPTS { self.queue.push_back(pending); report.requeued += 1; } else { @@ -225,13 +225,6 @@ impl AsyncSink { } } -fn retryable(error: &CacheError) -> bool { - matches!( - error, - CacheError::TimedOut | CacheError::Storage { .. } | CacheError::QuotaExceeded { .. } - ) -} - #[cfg(test)] #[path = "sink_tests.rs"] mod tests; diff --git a/crates/cold/src/sink_tests.rs b/crates/cold/src/sink_tests.rs index a49b74c..fb71c1a 100644 --- a/crates/cold/src/sink_tests.rs +++ b/crates/cold/src/sink_tests.rs @@ -381,3 +381,93 @@ fn metadata_and_payload_admissions_are_bounded() { )); assert_eq!(sink.retained_bytes(), 0); } + +#[test] +fn permanent_storage_corruption_is_not_retried_but_temporary_errors_are() { + let permanent = [ + CacheError::Unauthorized, + CacheError::Malformed { reason: "cold key" }, + CacheError::ChecksumMismatch, + CacheError::Storage { + reason: "store corrupt", + }, + CacheError::Storage { + reason: "object missing", + }, + CacheError::Storage { + reason: "unknown backend error", + }, + ]; + let transient = [ + CacheError::TimedOut, + CacheError::Storage { + reason: "object store", + }, + CacheError::Storage { + reason: "object unavailable", + }, + CacheError::QuotaExceeded { + kind: QuotaKind::InFlightBytes, + }, + ]; + for (errors, should_retry) in [(&permanent[..], false), (&transient[..], true)] { + for error in errors { + let job = ColdJob::from_pack(&pack("classification", 1)); + let mut sink = AsyncSink::new(1, job.byte_weight()).unwrap(); + let mut store = FaultStore::default(); + store.fail(&job.manifest_key, Fault::Before(error.clone()), 1); + sink.enqueue(job.clone()).unwrap(); + let report = sink.drain_report(&mut store, usize::MAX).unwrap(); + assert_eq!(report.attempted, 1); + assert_eq!(report.requeued, u32::from(should_retry)); + assert_eq!(report.failed, u32::from(!should_retry)); + assert_eq!(sink.retained_bytes(), job.byte_weight()); + assert_eq!(store.deletes, 0); + assert!(store.head(&job.object_key).is_ok()); + } + } +} + +#[test] +fn failed_job_storm_never_releases_credit_or_collects_orphans_automatically() { + use crate::{ColdOps, ReclaimStep, Retention}; + + let pack = pack("orphan", 1); + let job = ColdJob::from_pack(&pack); + let mut sink = AsyncSink::new(2, job.byte_weight() * 2).unwrap(); + let mut store = FaultStore::default(); + store.fail( + &job.manifest_key, + Fault::Before(CacheError::Unauthorized), + 2, + ); + sink.enqueue(job.clone()).unwrap(); + sink.enqueue(job.clone()).unwrap(); + assert_eq!(sink.drain_report(&mut store, usize::MAX).unwrap().failed, 2); + for _ in 0..128 { + assert!(matches!( + sink.enqueue(job.clone()), + Err(CacheError::QuotaExceeded { .. }) + )); + assert_eq!( + sink.drain_report(&mut store, usize::MAX).unwrap(), + DrainReport::default() + ); + assert_eq!( + (sink.failed_count(), sink.retained_bytes()), + (2, job.byte_weight() * 2) + ); + } + let mut ops = ColdOps::new(Retention::new(1).unwrap()); + assert_eq!(ops.schedule_expired(u64::MAX), 0); + assert_eq!(ops.reclaim_one(&mut store).unwrap(), ReclaimStep::Idle); + sink.take_failed().unwrap(); + sink.cancel(); + drop(sink); + assert_eq!(store.deletes, 0); + assert_eq!( + store.head(&pack.object_key).unwrap().len, + pack.object_bytes.len() as u64 + ); + assert!(store.head(&pack.manifest_key).is_err()); +} diff --git a/crates/cold/tests/failure_contract.rs b/crates/cold/tests/failure_contract.rs new file mode 100644 index 0000000..9338623 --- /dev/null +++ b/crates/cold/tests/failure_contract.rs @@ -0,0 +1,185 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 + +use std::cell::RefCell; + +use gpucache_cold::{ + pack_blocks, refill_degraded, refill_for_tenant, refill_generation, refill_matching, + ColdAvailability, MockObjectStore, ObjectMeta, ObjectStore, PackedBlock, RefillBudget, + RefillOutcome, +}; +use gpucache_core_types::{content_checksum, synthetic_identity, CacheError, QuotaKind}; + +struct ReadFaultStore { + inner: MockObjectStore, + fail_at: usize, + error: CacheError, + calls: RefCell>, +} + +impl ReadFaultStore { + fn read(&self, label: &str) -> Result<(), CacheError> { + self.calls.borrow_mut().push(label.to_string()); + if self.calls.borrow().len() == self.fail_at { + return Err(self.error.clone()); + } + Ok(()) + } +} + +impl ObjectStore for ReadFaultStore { + fn put(&mut self, _key: &str, _body: &[u8]) -> Result<(), CacheError> { + panic!("refill must not publish") + } + fn head(&self, key: &str) -> Result { + self.read("head")?; + self.inner.head(key) + } + fn range_get(&self, key: &str, offset: u64, length: u64) -> Result, CacheError> { + self.read(if key.ends_with("/manifest") { + "manifest" + } else { + "object" + })?; + self.inner.range_get(key, offset, length) + } + fn delete(&mut self, _key: &str) -> Result<(), CacheError> { + panic!("refill must not delete") + } +} + +#[test] +fn actual_read_outages_at_every_stage_are_clean_misses_without_retries_or_mutation() { + let identity = synthetic_identity(1, 1); + let payload = vec![7; 32]; + let pack = pack_blocks( + "read-outage", + &[PackedBlock { + generation: 1, + identity: identity.clone(), + checksum: content_checksum(&payload), + payload, + }], + ) + .unwrap(); + let budget = RefillBudget::new(1024 * 1024, 1).unwrap(); + let mut inner = MockObjectStore::default(); + inner.put(&pack.object_key, &pack.object_bytes).unwrap(); + inner.put(&pack.manifest_key, &pack.manifest_bytes).unwrap(); + let errors = [ + CacheError::TimedOut, + CacheError::Storage { + reason: "object missing", + }, + CacheError::Storage { + reason: "object unavailable", + }, + CacheError::Storage { + reason: "object store", + }, + CacheError::QuotaExceeded { + kind: QuotaKind::InFlightBytes, + }, + ]; + for fail_at in 1..=3 { + for error in &errors { + for entry in 0..4 { + let store = ReadFaultStore { + inner: inner.clone(), + fail_at, + error: error.clone(), + calls: RefCell::new(Vec::new()), + }; + let result = match entry { + 0 => refill_generation(&store, &pack.manifest_key, 1, &budget, &identity), + 1 => refill_matching(&store, &pack.manifest_key, 1, &budget, &identity), + 2 => refill_degraded( + &store, + &pack.manifest_key, + 1, + &budget, + ColdAvailability::Ready, + &identity, + ), + _ => refill_for_tenant( + &store, + &pack.manifest_key, + 1, + &budget, + ColdAvailability::Ready, + &identity, + ), + }; + assert_eq!(result.unwrap(), RefillOutcome::Missing); + assert_eq!(store.calls.borrow().len(), fail_at); + assert_eq!( + store + .inner + .range_get(&pack.object_key, 0, pack.object_bytes.len() as u64) + .unwrap(), + pack.object_bytes + ); + assert_eq!( + store + .inner + .range_get(&pack.manifest_key, 0, pack.manifest_bytes.len() as u64) + .unwrap(), + pack.manifest_bytes + ); + } + } + } +} + +#[test] +fn backend_validation_errors_remain_distinguishable_from_outages() { + for error in [ + CacheError::Unauthorized, + CacheError::ChecksumMismatch, + CacheError::Storage { + reason: "store corrupt", + }, + CacheError::Malformed { reason: "cold key" }, + ] { + let store = ReadFaultStore { + inner: MockObjectStore::default(), + fail_at: 1, + error: error.clone(), + calls: RefCell::new(Vec::new()), + }; + assert_eq!( + refill_generation( + &store, + "ns/manifest", + 1, + &RefillBudget::new(1024, 1).unwrap(), + &synthetic_identity(1, 1) + ), + Err(error) + ); + assert_eq!(store.calls.borrow().len(), 1); + } +} + +#[test] +fn declared_slow_and_unavailable_skip_all_reads_for_both_wrappers() { + for availability in [ColdAvailability::Slow, ColdAvailability::Unavailable] { + let store = ReadFaultStore { + inner: MockObjectStore::default(), + fail_at: 1, + error: CacheError::Unauthorized, + calls: RefCell::new(Vec::new()), + }; + let budget = RefillBudget::new(1024, 1).unwrap(); + let identity = synthetic_identity(1, 1); + assert_eq!( + refill_degraded(&store, "ns/manifest", 1, &budget, availability, &identity).unwrap(), + RefillOutcome::Missing + ); + assert_eq!( + refill_for_tenant(&store, "ns/manifest", 1, &budget, availability, &identity).unwrap(), + RefillOutcome::Missing + ); + assert!(store.calls.borrow().is_empty()); + } +} diff --git a/docs/V2_DATA_PATHS.md b/docs/V2_DATA_PATHS.md index 4a8f0fd..459862a 100644 --- a/docs/V2_DATA_PATHS.md +++ b/docs/V2_DATA_PATHS.md @@ -46,6 +46,16 @@ The hot TCP service does not depend on `gpucache-cold`. RustFS is not on the pre `drain_report` visits at most the queue length captured at call start, bounded further by `max_jobs`; a requeued task cannot be attempted repeatedly in that call. The report preserves counts and the first job error. The compatible `drain` call returns that first error after the round, even if other jobs completed. Pending and failed jobs together cannot exceed the configured limits or the 4096-job hard ceiling. A full failed queue causes backpressure until the operator retrieves its jobs. +Actual backend timeout, unavailability, missing-object and backpressure errors return +`Missing` at HEAD, manifest GET and payload GET. Corruption and authorization stay +explicit errors, and no failure causes inline retries or store mutation. Backend +implementations must enforce deadlines; the synchronous refill API cannot stop a +backend call that never returns. See [ADR 0012](adr/0012-cold-failure-contract.md) +for the error labels, retention behavior, and explicit cleanup policy. ColdKeys +caps tenant metadata at 1024 and generation history at 4096; ColdOps caps records +and the deduplicated reclaim queue at 4096. Local constructors can lower these +ceilings. Failed reclaim and retired key history retain their metadata credits. + Retries use the original content-addressed keys. A failed publication may leave an object without a manifest. Retain its keys for explicit later cleanup after ownership and reference checks. The sink implements neither a reference-safe orphan collector nor a crash-persistent orphan inventory, and does not infer safe deletion from a PUT failure or a HEAD result. It cannot provide a cross-object storage transaction. Callers migrate to `refill_generation(..., expected)` or `refill_matching(..., expected)` and pass that same identity through `refill_degraded` or `refill_for_tenant`. Old v1/v2 entries require recompute and repack; they are not converted by changing a header. The expected identity comes from the upstream authenticated and authorized request. The cold crate does not establish that identity, authorize arbitrary store keys, or verify the producer's model computation. diff --git a/docs/adr/0012-cold-failure-contract.md b/docs/adr/0012-cold-failure-contract.md new file mode 100644 index 0000000..149dce0 --- /dev/null +++ b/docs/adr/0012-cold-failure-contract.md @@ -0,0 +1,101 @@ +# ADR 0012: Bounded cold failure handling and explicit retention + +- Status: **Accepted** +- Date: 2026-10-20 +- Milestone: V3 CPU contract completion + +## Context + +The G4 mock already binds v3 manifests to complete canonical identities, derives +publication keys from content, and caps pending plus failed sink jobs. A PUT +failure can mean that publication succeeded but its acknowledgement was lost. +Deleting its object would corrupt an already published manifest. + +The remaining gaps were failure handling: actual backend read outages escaped as +errors even though predeclared Slow/Unavailable returned Missing; all storage +write errors were retried, including corruption; retention HEAD/DELETE errors +could discard scheduled work, and repeated delete-verification failure could +block the next key. Replacing a retention record also left an old expiration +queued. The optional HTTP probe treated binary response bodies as UTF-8 and +mapped authorization failures to transient storage errors. + +## Decision + +1. Keep GPMF manifest version 3 and existing digest keys unchanged. Versions 1, + 2, and unknown future versions are clean misses without payload reads. No + CacheIdentity or GCK1/GCKR wire semantics change. +2. ObjectStore adapters classify timeouts as `TimedOut`, temporary backend I/O + as `Storage` with reason `object store` or `object unavailable`, and temporary + backpressure as `QuotaExceeded`. The missing-object label is `object missing`. + Authorization, malformed input, corruption, and unknown storage labels are + permanent write errors. Unknown errors therefore cannot silently retry. +3. At each refill read stage (HEAD, manifest GET, payload GET), missing objects + and temporary failures return `Missing`. There is no inline retry or store + mutation. Predeclared Slow/Unavailable performs no store calls. Validation + and authorization failures remain distinguishable errors with no payload; + upstream still recomputes. This synchronous API cannot interrupt a backend + that never returns: adapters must enforce their own deadlines. +4. The sink attempts each job at most once per bounded drain round, rotating + temporary failures to the tail and exhausting them after three total + attempts. Permanent errors go directly to the failed queue. Pending and + failed jobs share the configured byte/job limits and the 4096-job ceiling. + Retrieval or explicit cancellation releases host credits only. +5. **Orphan garbage collection is not automatic.** Neither sink drain, + `take_failed`, cancellation, drop, nor ColdOps scans for orphan objects or + deletes one because PUT failed. ColdOps only acts on records explicitly + supplied by a trusted caller and explicitly scheduled for expiration. It is + a retention primitive, not a proof that an object is unreferenced. +6. Retention errors preserve the record and rotate its queued key to the tail. + A checksum mismatch refuses DELETE. Renewing a record cancels any queued + expiration for that key; the renewed timestamp must expire and be scheduled + again. Only a confirmed missing object completes reclaim. +7. Cold key metadata has hard ceilings of 1024 tenants and 4096 generation + records, including retired generations. Rotation checks history capacity and + generation overflow before demoting the current key. Retention holds at most + 4096 object records and at most one queued reclaim per record. Constructors + permit lower limits. Replacing a record at capacity remains allowed; a failed + reclaim retains its admission credit. Saturation returns `QuotaExceeded` + using the existing `InFlightRequests` metadata label, without a wire change. +8. The optional HTTP smoke adapter preserves permanent status errors, parses + binary bodies without UTF-8 conversion, and rejects truncated, overlong, + ambiguous-length, and unsupported chunked success responses. These are CPU + parser checks, not evidence of live RustFS compatibility. + +## Operator cleanup policy + +A failed publication may retain an orphan indefinitely. Preserve the failed job +or an external inventory before releasing its host credits. After producer +quiescence and an operator-chosen grace period, an external authority must check +ownership and every manifest reference, including in-flight publications, before +a delete is authorized. This crate has no reference index, bucket scanner, +crash-persistent inventory, or automatic orphan collector. A negative HEAD alone +is not a proof of safe deletion. Do not register unknown orphan keys in ColdOps +as a shortcut around those checks. + +ColdOps records and cold key-rotation history remain in-memory metadata despite +these hard ceilings. Retired key history is never silently evicted to create +space: capacity exhaustion requires operator action and must not revive old +credentials. The bounds are not a claim of bounded overall process RSS or a +crash-safe inventory. A future integrated collector needs its own ADR and +reference-safe protocol. + +## Evidence and limitations + +`cargo test -p gpucache-cold --all-targets` covers: + +- v1/v2 misses, complete identity mismatch, wrong-tenant rejection, content-key + substitution, and combined manifest/payload budgets (`refill_safety.rs`); +- every refill entry point and read stage under actual timeout, unavailable, + missing, and backpressure errors, plus no-call degraded wrappers + (`failure_contract.rs`); +- permanent/transient classification, fair drain, retry exhaustion, retained + failed-job credits, and a saturated-queue storm that performs no automatic + orphan collection (`sink_tests.rs`); +- retention HEAD, DELETE, and verification faults, fair progress, unexpected + content preservation, renewed-record expiration, metadata capacity failure + atomicity, and key-generation overflow (`reclaim_tests.rs`); +- HTTP status classification and binary/framing parser failures (`http.rs`). + +Mock ObjectStore remains the default. Live RustFS, GDS, RDMA and BF3 remain +`NOT_RUN_HARDWARE`. No credentials, hardware, performance results, or new +external dependencies were introduced. The cold crate stays off the hot path.