From b649ed450ed21f16d190f8513638119f44581546 Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Thu, 23 Jul 2026 23:41:23 +1000 Subject: [PATCH 1/4] feat(reliability): retry, circuit breaker, graceful degradation, cold-miss single-flight (LAB-518) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Brings cachekit-rs to reliability-tier parity with the Python and TypeScript SDKs: - New `reliability` feature (default, native-only): retry with truncated exponential backoff + jitter driven by the previously-dead BackendErrorKind::is_retryable() classification, and a closed/open/half-open circuit breaker with a rolling failure window. Composed as breaker(retry(op)) around every backend data operation, mirroring cachekit-ts's ReliabilityExecutor. Only retryable-kind errors count toward opening the circuit — permanent/auth errors are request-specific and must not cut off healthy traffic. - Presets: on by default in production/encrypted/io, off in minimal (ts preset posture). Builder: .reliability() / .no_reliability(). - New BackendErrorKind::CircuitOpen for fail-fast errors. - Graceful degradation in #[cachekit]: backend failures fail OPEN on the plain path (function runs uncached, matching cachekit-py); the secure path stays fail-CLOSED — encrypted workloads never silently degrade. - Cold-miss single-flight (CacheKit::single_flight, always available): per-key in-process async mutex dedups concurrent fills; with `reliability`, lock-capable backends (CachekitIO, Redis via the existing LockableBackend) additionally suppress fills cross-process through a distributed fill lock with poll-for-fill on contention. Leaders never re-check the cache — a re-check would be a second billable miss under metered-misses pricing. - Backend::as_lockable() default hook (None) overridden by CachekitIO and RedisBackend so dyn Backend can expose its lock capability. Co-authored-by: multica-agent --- README.md | 36 ++ crates/cachekit-macros/src/lib.rs | 51 +- crates/cachekit/Cargo.toml | 10 +- crates/cachekit/src/backend/cachekitio.rs | 6 +- crates/cachekit/src/backend/mod.rs | 20 + crates/cachekit/src/backend/redis.rs | 4 + crates/cachekit/src/client.rs | 50 ++ crates/cachekit/src/error.rs | 14 + crates/cachekit/src/flight.rs | 227 +++++++++ crates/cachekit/src/intents.rs | 38 +- crates/cachekit/src/lib.rs | 14 + crates/cachekit/src/reliability.rs | 541 +++++++++++++++++++++ crates/cachekit/tests/macro_tests.rs | 159 ++++++ crates/cachekit/tests/reliability_tests.rs | 509 +++++++++++++++++++ 14 files changed, 1664 insertions(+), 15 deletions(-) create mode 100644 crates/cachekit/src/flight.rs create mode 100644 crates/cachekit/src/reliability.rs create mode 100644 crates/cachekit/tests/reliability_tests.rs diff --git a/README.md b/README.md index 974e508..a582c1b 100644 --- a/README.md +++ b/README.md @@ -39,6 +39,7 @@ | `cachekitio` | ✅ | HTTP backend for [api.cachekit.io](https://api.cachekit.io) via [reqwest](https://crates.io/crates/reqwest) + rustls | | `encryption` | ✅ | Zero-knowledge AES-256-GCM via [cachekit-core](https://crates.io/crates/cachekit-core) | | `l1` | ✅ | In-process L1 cache via [moka](https://crates.io/crates/moka) | +| `reliability` | ✅ | Retry with backoff + jitter, circuit breaker, distributed fill locks (native only) | | `redis` | ❌ | Redis backend via [fred](https://crates.io/crates/fred) (native only) | | `workers` | ❌ | Cloudflare Workers backend via [worker](https://crates.io/crates/worker) | | `macros` | ❌ | `#[cachekit]` proc-macro decorator (mints [interop/v1](#cross-sdk-interop-mode) keys) | @@ -61,6 +62,7 @@ cachekit-rs = { version = "0.2", default-features = false, features = ["workers" > **Mutually exclusive features:** > - `workers` + `redis` — Workers runtime cannot use fred > - `workers` + `l1` — moka requires std threads unavailable in wasm32 +> - `workers` + `reliability` — retry/breaker timers need tokio `time`, unavailable in wasm32 --- @@ -283,6 +285,40 @@ When the `l1` feature is enabled (default), CacheKit maintains an in-process [mo --- +## Reliability + +With the `reliability` feature (default, native only), the `production`, `encrypted`, and `io` presets wrap every backend operation in a reliability stack; `minimal` stays bare for maximum throughput: + +| Layer | What it does | Defaults | +|:------|:-------------|:---------| +| **Retry** | Truncated exponential backoff + jitter on transient/timeout errors (`BackendErrorKind::is_retryable`); permanent and auth errors propagate immediately | 3 attempts, 100 ms base, 5 s cap, jitter ×[0.5, 1.5) | +| **Circuit breaker** | closed → open after N retryable failures in a rolling window; fails fast (`BackendErrorKind::CircuitOpen`) while open; half-open probes recovery | threshold 5, window 60 s, open 5 s, 3 probes, close after 3 successes | +| **Graceful degradation** | On backend failure, `#[cachekit]`-wrapped functions run uncached (fail-open). `secure` paths fail **closed** — encrypted workloads never silently degrade | built into the macro | +| **Single-flight** | Concurrent misses of one key collapse to a single execution: per-key in-process lock, plus a distributed fill lock across processes on lock-capable backends (cachekit.io, Redis) | in-process always on; cross-process 5 s lock, 100 ms polls | + +Retry sits *inside* the breaker (one exhausted retry sequence = one breaker failure), degradation and single-flight sit in the `#[cachekit]` macro around the miss path — the same composition as the TypeScript SDK's `ReliabilityExecutor` and the Python decorator. + +```rust,ignore +use std::time::Duration; +use cachekit::{CacheKit, ReliabilityConfig, RetryConfig}; + +// Presets enable it — override or disable per client: +let cache = CacheKit::production("redis://localhost:6379").await? + .reliability(ReliabilityConfig { + retry: Some(RetryConfig { max_attempts: 5, ..RetryConfig::default() }), + ..ReliabilityConfig::default() + }) + .build()?; + +let bare = CacheKit::production("redis://localhost:6379").await? + .no_reliability() + .build()?; +``` + +Requires a tokio runtime for backoff timers (the `redis` and `cachekitio` backends already do). + +--- + ## Environment Variables | Variable | Required | Description | diff --git a/crates/cachekit-macros/src/lib.rs b/crates/cachekit-macros/src/lib.rs index 60f72cf..087cfc7 100644 --- a/crates/cachekit-macros/src/lib.rs +++ b/crates/cachekit-macros/src/lib.rs @@ -203,6 +203,19 @@ fn extract_ok_type(ret: &ReturnType) -> syn::Result { /// - A stored entry that cannot be decoded as the return type is treated as /// a miss and overwritten (self-healing), never an error loop. /// +/// # Reliability behaviour +/// +/// - **Graceful degradation**: on a backend failure the plain path fails +/// *open* — the function executes uncached and its result is returned +/// (matching cachekit-py). With `secure`, backend and decryption errors +/// fail *closed* and propagate to the caller: an encrypted workload never +/// silently degrades. +/// - **Cold-miss single-flight**: concurrent calls that miss on the same +/// key are collapsed to one execution per process (and per fleet, when +/// the backend supports distributed fill locks — CachekitIO and Redis do, +/// with the `reliability` feature). Waiters re-read the filled entry +/// instead of recomputing. See `cachekit::flight`. +/// /// # Example /// /// ```ignore @@ -297,6 +310,17 @@ fn expand(args: &MacroArgs, mut func: ItemFn) -> syn::Result { ) }; + // Graceful degradation (LAB-518): on a backend failure the plain path + // fails OPEN — the wrapped function runs uncached, mirroring cachekit-py's + // BackendError → execute-without-caching posture. The `secure` path stays + // fail-CLOSED: backend and decryption errors reach the caller, so an + // encrypted workload never silently degrades. + let fail_open_arm = if args.secure { + quote! {} + } else { + quote! { Err(cachekit::error::CachekitError::Backend(_)) => {} } + }; + // Capture the original function body. let original_body = &func.block; @@ -329,15 +353,37 @@ fn expand(args: &MacroArgs, mut func: ItemFn) -> syn::Result { // cannot be decoded as #ok_type (poisoned, foreign-shaped, or a // Python-internal CK frame) — treat it as a miss so the fresh // result OVERWRITES it, instead of hard-failing every call until - // TTL expiry. All other errors (namespaced-client Config, backend - // failures) propagate. + // TTL expiry. Backend errors fail open on the plain path (run + // the function uncached) and fail closed on the secure path. + // Other errors (namespaced-client Config, decryption) propagate. match #get_expr { Ok(Some(__ck_cached)) => return Ok(__ck_cached), Ok(None) => {} Err(cachekit::error::CachekitError::Serialization(_)) => {} + #fail_open_arm Err(__ck_err) => return Err(__ck_err.into()), } + // Cold-miss single-flight: collapse concurrent fills of this key + // to one execution (misses are billable). While another worker is + // filling, re-check the cache instead of recomputing. + let mut __ck_flight = #client_ident.single_flight(&__ck_key).await; + while __ck_flight.awaiting_fill().await { + match #get_expr { + Ok(Some(__ck_cached)) => { + __ck_flight.release().await; + return Ok(__ck_cached); + } + Ok(None) => {} + Err(cachekit::error::CachekitError::Serialization(_)) => {} + #fail_open_arm + Err(__ck_err) => { + __ck_flight.release().await; + return Err(__ck_err.into()); + } + } + } + // Execute original function body let __ck_result: #ret_ty = (async #original_body).await; @@ -346,6 +392,7 @@ fn expand(args: &MacroArgs, mut func: ItemFn) -> syn::Result { #set_expr } + __ck_flight.release().await; __ck_result } }; diff --git a/crates/cachekit/Cargo.toml b/crates/cachekit/Cargo.toml index 3dab328..e6a2c3c 100644 --- a/crates/cachekit/Cargo.toml +++ b/crates/cachekit/Cargo.toml @@ -15,7 +15,7 @@ categories = ["caching", "web-programming", "cryptography"] name = "cachekit" [features] -default = ["cachekitio", "encryption", "l1"] +default = ["cachekitio", "encryption", "l1", "reliability"] cachekitio = ["dep:reqwest", "dep:serde_json"] redis = ["dep:fred"] workers = ["dep:worker", "dep:js-sys", "dep:getrandom", "dep:serde_json"] @@ -24,6 +24,9 @@ workers = ["dep:worker", "dep:js-sys", "dep:getrandom", "dep:serde_json"] encryption = ["cachekit-core/encryption"] l1 = ["dep:moka"] macros = ["dep:cachekit-macros"] +# Retry with backoff + circuit breaker around backend ops, and cross-process +# cold-miss suppression via LockableBackend. Native only (needs tokio timers). +reliability = ["tokio/time"] # Opt into ?Send / Rc on native targets (for tokio::task::LocalSet, single-threaded # runtimes, etc.). Same code paths that wasm32 already uses. unsync = [] @@ -41,6 +44,9 @@ urlencoding = "2" hex = "0.4" url = "2" uuid = { version = "1", features = ["v4", "js"] } +# `sync` only: runtime-independent primitives for cold-miss single-flight. +# Compiles on wasm32; the `reliability` feature adds `time` (native only). +tokio = { version = "1", default-features = false, features = ["sync"] } # Optional: HTTP backend (native) reqwest = { version = "0.12", optional = true, default-features = false, features = ["rustls-tls", "json"] } @@ -62,7 +68,7 @@ getrandom = { version = "0.2", optional = true, features = ["js"] } cachekit-macros = { version = "0.4.0", path = "../cachekit-macros", optional = true } [dev-dependencies] -tokio = { version = "1", features = ["macros", "rt-multi-thread"] } +tokio = { version = "1", features = ["macros", "rt-multi-thread", "sync", "time"] } serde_json = "1" serial_test = "3" diff --git a/crates/cachekit/src/backend/cachekitio.rs b/crates/cachekit/src/backend/cachekitio.rs index f28b416..cdc5f8e 100644 --- a/crates/cachekit/src/backend/cachekitio.rs +++ b/crates/cachekit/src/backend/cachekitio.rs @@ -4,7 +4,7 @@ use std::time::Duration; use async_trait::async_trait; use zeroize::Zeroizing; -use crate::backend::{Backend, HealthStatus}; +use crate::backend::{Backend, HealthStatus, LockableBackend}; use crate::error::{BackendError, BackendErrorKind}; use crate::metrics::{metrics_headers, MetricsProvider}; use crate::session::session_headers; @@ -247,6 +247,10 @@ impl Backend for CachekitIO { Err(self.error_from_response(resp).await) } } + + fn as_lockable(&self) -> Option<&dyn LockableBackend> { + Some(self) + } } // ── Builder ─────────────────────────────────────────────────────────────────── diff --git a/crates/cachekit/src/backend/mod.rs b/crates/cachekit/src/backend/mod.rs index 76feae5..49bed44 100644 --- a/crates/cachekit/src/backend/mod.rs +++ b/crates/cachekit/src/backend/mod.rs @@ -49,6 +49,16 @@ pub trait Backend: Send + Sync { /// Return health/status information for this backend. async fn health(&self) -> Result; + + /// Expose this backend's [`LockableBackend`] capability, if it has one. + /// + /// Trait objects (`dyn Backend`) cannot be cross-cast to a sibling trait, + /// so backends that support distributed locking opt in by overriding this + /// to return `Some(self)`. Used by the client's cold-miss single-flight + /// for cross-process fill suppression. Default: `None`. + fn as_lockable(&self) -> Option<&dyn LockableBackend> { + None + } } /// Async cache backend abstraction (`?Send` variant). @@ -77,6 +87,16 @@ pub trait Backend { /// Return health/status information for this backend. async fn health(&self) -> Result; + + /// Expose this backend's [`LockableBackend`] capability, if it has one. + /// + /// Trait objects (`dyn Backend`) cannot be cross-cast to a sibling trait, + /// so backends that support distributed locking opt in by overriding this + /// to return `Some(self)`. Used by the client's cold-miss single-flight + /// for cross-process fill suppression. Default: `None`. + fn as_lockable(&self) -> Option<&dyn LockableBackend> { + None + } } // ── TtlInspectable ─────────────────────────────────────────────────────────── diff --git a/crates/cachekit/src/backend/redis.rs b/crates/cachekit/src/backend/redis.rs index 78f6595..03fbb30 100644 --- a/crates/cachekit/src/backend/redis.rs +++ b/crates/cachekit/src/backend/redis.rs @@ -124,6 +124,10 @@ impl Backend for RedisBackend { details, }) } + + fn as_lockable(&self) -> Option<&dyn LockableBackend> { + Some(self) + } } // ── TtlInspectable impl ─────────────────────────────────────────────────────── diff --git a/crates/cachekit/src/client.rs b/crates/cachekit/src/client.rs index f10b422..938c9c6 100644 --- a/crates/cachekit/src/client.rs +++ b/crates/cachekit/src/client.rs @@ -77,6 +77,7 @@ pub struct CacheKit { default_ttl: Duration, namespace: Option, max_payload_bytes: usize, + flight: crate::flight::FlightMap, #[cfg(feature = "l1")] l1: Option, @@ -344,6 +345,24 @@ impl CacheKit { Ok(self.backend.exists(&full_key).await?) } + // ── Single-flight ───────────────────────────────────────────────────────── + + /// Begin a cold-miss single-flight for `key` (see [`crate::flight`]). + /// + /// Call after a cache miss, before computing the value. Concurrent + /// in-process fills of the same key are collapsed to one; with the + /// `reliability` feature and a lock-capable backend (CachekitIO, Redis), + /// fills are also suppressed across processes via a distributed fill + /// lock. The `#[cachekit]` macro does this automatically. + /// + /// The key is namespaced like every cache operation but not validated — + /// this call is infallible; an invalid key simply fails later at the + /// actual cache operation. + pub async fn single_flight(&self, key: &str) -> crate::flight::SingleFlight { + let full_key = self.namespaced_key(key); + crate::flight::SingleFlight::acquire(&self.flight, &self.backend, &full_key).await + } + // ── Secure cache ───────────────────────────────────────────────────────── /// Return a [`SecureCache`] handle that encrypts all values before storage. @@ -526,6 +545,9 @@ pub struct CacheKitBuilder { #[cfg(feature = "encryption")] encryption: Option, + + #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))] + reliability: Option, } impl CacheKitBuilder { @@ -578,6 +600,26 @@ impl CacheKitBuilder { self } + /// Wrap the backend in the reliability stack (retry with exponential + /// backoff + jitter, circuit breaker) — see [`crate::reliability`]. + /// + /// Enabled by default with production settings by the `production`, + /// `encrypted`, and `io` intent presets; off for `minimal` and for + /// manually-built clients. + #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))] + pub fn reliability(mut self, config: crate::reliability::ReliabilityConfig) -> Self { + self.reliability = Some(config); + self + } + + /// Remove any reliability configuration (backend errors propagate on + /// first failure, no circuit breaker). Overrides a preset's default. + #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))] + pub fn no_reliability(mut self) -> Self { + self.reliability = None; + self + } + /// Configure encryption from raw master key bytes and tenant ID. /// /// The master key must be at least 16 bytes (32 recommended). @@ -650,11 +692,19 @@ impl CacheKitBuilder { Some(crate::l1::L1Cache::new(capacity)) }; + // Apply the reliability stack last so it decorates the final backend. + #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))] + let backend = match self.reliability { + Some(config) => crate::reliability::wrap_reliable(backend, config), + None => backend, + }; + Ok(CacheKit { backend, default_ttl: self.default_ttl.unwrap_or(Duration::from_secs(300)), namespace: self.namespace, max_payload_bytes: self.max_payload_bytes.unwrap_or(5 * 1024 * 1024), + flight: crate::flight::FlightMap::default(), #[cfg(feature = "l1")] l1, diff --git a/crates/cachekit/src/error.rs b/crates/cachekit/src/error.rs index 1ca9462..8a4bd93 100644 --- a/crates/cachekit/src/error.rs +++ b/crates/cachekit/src/error.rs @@ -48,6 +48,10 @@ pub enum BackendErrorKind { Timeout, /// Credentials are invalid or missing — retrying will not help. Authentication, + /// The circuit breaker is open — the call failed fast without reaching + /// the backend. Not retryable *now*; the breaker re-probes on its own + /// schedule (see `reliability::CircuitBreakerConfig::open_timeout`). + CircuitOpen, } impl BackendErrorKind { @@ -65,6 +69,7 @@ impl std::fmt::Display for BackendErrorKind { Self::Permanent => write!(f, "permanent"), Self::Timeout => write!(f, "timeout"), Self::Authentication => write!(f, "authentication"), + Self::CircuitOpen => write!(f, "circuit-open"), } } } @@ -126,6 +131,15 @@ impl BackendError { } } + /// Create a circuit-open backend error (call failed fast, backend not reached). + pub fn circuit_open(message: impl Into) -> Self { + Self { + kind: BackendErrorKind::CircuitOpen, + message: message.into(), + source: None, + } + } + /// Sanitize error messages to strip API keys (CWE-532). pub fn sanitize_message(msg: &str, api_key: &str) -> String { if api_key.is_empty() { diff --git a/crates/cachekit/src/flight.rs b/crates/cachekit/src/flight.rs new file mode 100644 index 0000000..c919fa5 --- /dev/null +++ b/crates/cachekit/src/flight.rs @@ -0,0 +1,227 @@ +//! Cold-miss single-flight: dedup concurrent fills of the same key. +//! +//! Under metered-misses pricing a stampede is literally billable — N tasks +//! missing the same key at once means N backend misses and N executions of +//! the wrapped function. [`CacheKit::single_flight`](crate::CacheKit::single_flight) +//! collapses that to one: +//! +//! - **In-process** (always available): a per-key async mutex. The first +//! task through becomes the *leader* and computes; concurrent tasks queue +//! behind it and re-check the cache once the leader finishes. +//! - **Cross-process** (`reliability` feature, native, backend implements +//! `LockableBackend` — CachekitIO and Redis do): the leader additionally +//! takes a distributed fill lock. If another process already holds it, +//! this process polls the cache for the other side's fill instead of +//! recomputing, and computes anyway once the poll budget is exhausted +//! (fail-open — a stampede beats unavailability). +//! +//! The `#[cachekit]` macro wires this in automatically around its miss path. +//! Manual usage follows the same shape: +//! +//! ```no_run +//! # async fn example(cache: &cachekit::CacheKit) -> Result<(), cachekit::CachekitError> { +//! if let Some(_v) = cache.get::("expensive").await? { +//! return Ok(()); +//! } +//! let mut flight = cache.single_flight("expensive").await; +//! while flight.awaiting_fill().await { +//! if let Some(_v) = cache.get::("expensive").await? { +//! flight.release().await; // another worker filled it +//! return Ok(()); +//! } +//! } +//! let value = "computed".to_owned(); // expensive work — runs once +//! cache.set("expensive", &value).await?; +//! flight.release().await; +//! # Ok(()) +//! # } +//! ``` + +use std::collections::HashMap; +use std::sync::{Arc, Mutex, PoisonError, Weak}; + +use crate::client::SharedBackend; + +/// Above this many live entries, dead map slots are swept opportunistically. +const SWEEP_THRESHOLD: usize = 128; + +/// How long a distributed fill lock is held server-side before auto-expiry. +#[cfg(all(feature = "reliability", not(target_arch = "wasm32")))] +const FILL_LOCK_TIMEOUT_MS: u64 = 5_000; + +/// Poll cadence while waiting for another process's fill. +#[cfg(all(feature = "reliability", not(target_arch = "wasm32")))] +const FILL_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_millis(100); + +/// Poll budget: 50 × 100 ms ≈ the fill lock timeout. +#[cfg(all(feature = "reliability", not(target_arch = "wasm32")))] +const FILL_POLL_BUDGET: u32 = 50; + +// ── FlightMap ──────────────────────────────────────────────────────────────── + +/// Per-key async mutexes for in-process fill dedup. Weak entries let finished +/// flights drop their state without an explicit removal protocol. +#[derive(Default)] +pub(crate) struct FlightMap { + entries: Mutex>>>, +} + +impl FlightMap { + fn handle(&self, key: &str) -> Arc> { + let mut map = self.entries.lock().unwrap_or_else(PoisonError::into_inner); + // ponytail: O(n) sweep once the map grows; a doubly-indexed structure + // is not worth it until someone caches millions of distinct cold keys. + if map.len() > SWEEP_THRESHOLD { + map.retain(|_, w| w.strong_count() > 0); + } + if let Some(existing) = map.get(key).and_then(Weak::upgrade) { + return existing; + } + let fresh = Arc::new(tokio::sync::Mutex::new(())); + map.insert(key.to_owned(), Arc::downgrade(&fresh)); + fresh + } +} + +// ── SingleFlight guard ─────────────────────────────────────────────────────── + +enum Role { + /// First worker in: compute without re-checking (a re-check would be a + /// second billable miss under metered-misses pricing). + Leader, + /// Queued behind a local leader that has since finished: re-check the + /// cache once — the leader's fill is in L1 — then compute if it missed. + LocalFollower { rechecked: bool }, + /// Another *process* holds the distributed fill lock: poll the cache for + /// its fill, then compute anyway when the budget runs out (fail-open). + #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))] + RemoteContested { polls_left: u32 }, +} + +/// Guard for a single-flight fill, returned by +/// [`CacheKit::single_flight`](crate::CacheKit::single_flight). +/// +/// Holds the per-key in-process lock for its whole lifetime, and the +/// distributed fill lock (if one was acquired) until [`Self::release`]. +/// Dropping without `release` is safe: the in-process lock frees immediately +/// and a distributed lock expires server-side after its timeout. +pub struct SingleFlight { + _local: tokio::sync::OwnedMutexGuard<()>, + role: Role, + #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))] + dist: Option, +} + +#[cfg(all(feature = "reliability", not(target_arch = "wasm32")))] +struct DistLock { + backend: SharedBackend, + full_key: String, + lock_id: String, +} + +impl SingleFlight { + /// `true` while another worker may still be filling this key — re-check + /// the cache after every `true` before computing yourself: + /// + /// - Leader: immediately `false` (compute, don't re-read your own miss). + /// - Queued behind a local leader: `true` exactly once. + /// - Contested cross-process: sleeps one poll interval per call, `true` + /// until the poll budget is spent. + pub async fn awaiting_fill(&mut self) -> bool { + match &mut self.role { + Role::Leader => false, + Role::LocalFollower { rechecked } => { + let first = !*rechecked; + *rechecked = true; + first + } + #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))] + Role::RemoteContested { polls_left } => { + if *polls_left == 0 { + return false; + } + *polls_left -= 1; + tokio::time::sleep(FILL_POLL_INTERVAL).await; + true + } + } + } + + /// Release the flight. Best-effort: frees the distributed fill lock (if + /// held) so other processes stop waiting early; errors are ignored — the + /// lock expires server-side regardless. + pub async fn release(self) { + #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))] + if let Some(dist) = self.dist { + if let Some(lockable) = dist.backend.as_lockable() { + let _ = lockable.release_lock(&dist.full_key, &dist.lock_id).await; + } + } + } + + pub(crate) async fn acquire(map: &FlightMap, backend: &SharedBackend, full_key: &str) -> Self { + let handle = map.handle(full_key); + match Arc::clone(&handle).try_lock_owned() { + Ok(local) => Self::lead(local, backend, full_key).await, + Err(_) => { + // Contended: a local leader is filling. Queue behind it. + let local = handle.lock_owned().await; + Self { + _local: local, + role: Role::LocalFollower { rechecked: false }, + #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))] + dist: None, + } + } + } + } + + /// Local leader: attempt cross-process suppression via the backend's + /// distributed lock, when available. Lock-infrastructure errors fail + /// open to a plain leader — suppression is an optimisation, never an + /// availability dependency. + #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))] + async fn lead( + local: tokio::sync::OwnedMutexGuard<()>, + backend: &SharedBackend, + full_key: &str, + ) -> Self { + let (role, dist) = match backend.as_lockable() { + Some(lockable) => match lockable.acquire_lock(full_key, FILL_LOCK_TIMEOUT_MS).await { + Ok(Some(lock_id)) => ( + Role::Leader, + Some(DistLock { + backend: backend.clone(), + full_key: full_key.to_owned(), + lock_id, + }), + ), + Ok(None) => ( + Role::RemoteContested { + polls_left: FILL_POLL_BUDGET, + }, + None, + ), + Err(_) => (Role::Leader, None), + }, + None => (Role::Leader, None), + }; + Self { + _local: local, + role, + dist, + } + } + + #[cfg(not(all(feature = "reliability", not(target_arch = "wasm32"))))] + async fn lead( + local: tokio::sync::OwnedMutexGuard<()>, + _backend: &SharedBackend, + _full_key: &str, + ) -> Self { + Self { + _local: local, + role: Role::Leader, + } + } +} diff --git a/crates/cachekit/src/intents.rs b/crates/cachekit/src/intents.rs index 019a818..5655a4e 100644 --- a/crates/cachekit/src/intents.rs +++ b/crates/cachekit/src/intents.rs @@ -5,12 +5,17 @@ //! use case and returns a [`CacheKitBuilder`] so callers can override any //! setting before building. //! -//! | Intent | Backend | L1 | Encryption | Auto-reconnect | Default TTL | -//! |------------|-----------|------|------------|----------------|-------------| -//! | `minimal` | Redis | Off | No | No | 300 s | -//! | `production` | Redis | On | No | Yes | 600 s | -//! | `encrypted` | Redis | On | AES-256-GCM | Yes | 600 s | -//! | `io` | cachekit.io | On | No | n/a (HTTP) | 3 600 s | +//! | Intent | Backend | L1 | Encryption | Auto-reconnect | Reliability¹ | Default TTL | +//! |------------|-----------|------|------------|----------------|--------------|-------------| +//! | `minimal` | Redis | Off | No | No | Off | 300 s | +//! | `production` | Redis | On | No | Yes | On | 600 s | +//! | `encrypted` | Redis | On | AES-256-GCM | Yes | On | 600 s | +//! | `io` | cachekit.io | On | No | n/a (HTTP) | On | 3 600 s | +//! +//! ¹ Retry with backoff + jitter and a circuit breaker around backend ops +//! (requires the `reliability` feature, on by default — see +//! [`crate::reliability`]). Override with +//! [`CacheKitBuilder::reliability`] / [`CacheKitBuilder::no_reliability`]. use std::time::Duration; @@ -38,6 +43,8 @@ impl CacheKit { /// connection is not re-established) /// * L1 cache: **off** /// * Encryption: **no** + /// * Reliability: **off** — no retry, no circuit breaker; every backend + /// error propagates on first failure /// * Default TTL: **300 s** /// /// Good for: product catalogs, public data, development. @@ -75,6 +82,7 @@ impl CacheKit { /// exponential backoff after a dropped connection) /// * L1 cache: **on** (1 000 entries) /// * Encryption: **no** + /// * Reliability: **on** — retry with backoff + jitter, circuit breaker /// * Default TTL: **600 s** /// /// Good for: user sessions, API responses, production services. @@ -101,10 +109,13 @@ impl CacheKit { .build()?; drop(backend.connect().await?); - Ok(CacheKitBuilder::default() + let builder = CacheKitBuilder::default() .backend(wrap(backend)) .default_ttl(Duration::from_secs(600)) - .l1_capacity(1000)) + .l1_capacity(1000); + #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))] + let builder = builder.reliability(crate::reliability::ReliabilityConfig::default()); + Ok(builder) } /// **Encrypted** — zero-knowledge encrypted Redis cache. @@ -113,6 +124,7 @@ impl CacheKit { /// exponential backoff after a dropped connection) /// * L1 cache: **on** (1 000 entries, stores ciphertext) /// * Encryption: **AES-256-GCM** with HKDF-SHA256 + /// * Reliability: **on** — retry with backoff + jitter, circuit breaker /// * Default TTL: **600 s** /// * Tenant ID: `"default"` (override via /// [`.encryption_from_bytes()`](CacheKitBuilder::encryption_from_bytes)) @@ -148,6 +160,8 @@ impl CacheKit { .default_ttl(Duration::from_secs(600)) .l1_capacity(1000) .encryption_from_bytes(master_key, "default")?; + #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))] + let builder = builder.reliability(crate::reliability::ReliabilityConfig::default()); let backend = crate::backend::redis::RedisBackend::builder() .url(redis_url) @@ -164,6 +178,7 @@ impl CacheKit { /// * L1 cache: **on** (1 000 entries) /// * Encryption: **no** (add via /// [`.encryption()`](CacheKitBuilder::encryption)) + /// * Reliability: **on** — retry with backoff + jitter, circuit breaker /// * Default TTL: **3 600 s** /// /// Good for: serverless, edge compute, managed caching without Redis. @@ -188,9 +203,12 @@ impl CacheKit { .api_key(api_key) .build()?; - Ok(CacheKitBuilder::default() + let builder = CacheKitBuilder::default() .backend(wrap(backend)) .default_ttl(Duration::from_secs(3600)) - .l1_capacity(1000)) + .l1_capacity(1000); + #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))] + let builder = builder.reliability(crate::reliability::ReliabilityConfig::default()); + Ok(builder) } } diff --git a/crates/cachekit/src/lib.rs b/crates/cachekit/src/lib.rs index bcf4c2d..59b3443 100644 --- a/crates/cachekit/src/lib.rs +++ b/crates/cachekit/src/lib.rs @@ -16,6 +16,9 @@ compile_error!( #[cfg(all(feature = "workers", feature = "l1"))] compile_error!("features `workers` and `l1` are mutually exclusive — moka requires std threads unavailable in wasm32"); +#[cfg(all(feature = "workers", feature = "reliability"))] +compile_error!("features `workers` and `reliability` are mutually exclusive — retry/breaker timers need tokio `time`, unavailable in wasm32"); + /// Pluggable cache backend trait and implementations (CachekitIO, Redis, Workers). pub mod backend; /// High-level cache client with dual-layer (L1/L2) support. @@ -24,6 +27,8 @@ pub mod client; pub mod config; /// Error types for cache operations and backend communication. pub mod error; +/// Cold-miss single-flight: dedup concurrent fills of the same key. +pub mod flight; /// Interop mode (interop/v1): cross-SDK cache keys and plain-MessagePack values. pub mod interop; /// L1 cache hit-rate metrics for CachekitIO request headers. @@ -46,6 +51,10 @@ pub mod encryption; #[cfg(feature = "l1")] pub mod l1; +/// Reliability tier: retry with backoff + jitter, circuit breaker. +#[cfg(all(feature = "reliability", not(target_arch = "wasm32")))] +pub mod reliability; + // Re-exports pub use client::{CacheKit, CacheKitBuilder, SharedBackend}; pub use config::CachekitConfig; @@ -59,6 +68,11 @@ pub use encryption::EncryptionLayer; #[cfg(feature = "macros")] pub use cachekit_macros::cachekit; +pub use flight::SingleFlight; + +#[cfg(all(feature = "reliability", not(target_arch = "wasm32")))] +pub use reliability::{CircuitBreakerConfig, ReliabilityConfig, RetryConfig}; + /// Convenient glob import for the most common types. pub mod prelude { pub use crate::{ diff --git a/crates/cachekit/src/reliability.rs b/crates/cachekit/src/reliability.rs new file mode 100644 index 0000000..cda3011 --- /dev/null +++ b/crates/cachekit/src/reliability.rs @@ -0,0 +1,541 @@ +//! Reliability tier: retry with exponential backoff + jitter, and a +//! closed/open/half-open circuit breaker around backend operations. +//! +//! Both are composed by a private `ReliableBackend` decorator around any +//! [`crate::backend::Backend`], applied by the builder when a [`ReliabilityConfig`] is set +//! (see [`crate::CacheKitBuilder::reliability`]). The intent presets +//! `production`, `encrypted`, and `io` enable it by default; `minimal` does +//! not — mirroring the TypeScript SDK's preset posture. +//! +//! Composition order matches the TypeScript SDK's `ReliabilityExecutor`: +//! the retry loop is *inside* the breaker, so one exhausted retry sequence +//! counts as a single breaker failure, and a fast-failing open breaker never +//! spends time retrying. +//! +//! Unlike the TypeScript breaker (which counts every error), only errors +//! classified retryable by [`crate::error::BackendErrorKind::is_retryable`] (`Transient`, +//! `Timeout`) count toward opening the circuit: they are the backend-health +//! signals. `Permanent` / `Authentication` errors are request-specific — five +//! malformed requests must not cut off healthy traffic. +//! +//! Requires a tokio runtime for backoff timers (`redis` and `cachekitio` +//! backends already do). Not available on wasm32 targets. + +use std::future::Future; +use std::sync::{Mutex, PoisonError}; +use std::time::{Duration, Instant}; + +use async_trait::async_trait; + +use crate::backend::{Backend, HealthStatus, LockableBackend}; +use crate::client::SharedBackend; +use crate::error::BackendError; + +// ── Configuration ──────────────────────────────────────────────────────────── + +/// Retry policy configuration (truncated exponential backoff with jitter). +#[derive(Debug, Clone, PartialEq)] +pub struct RetryConfig { + /// Total attempts, including the first (default: 3). `0` behaves as `1`. + pub max_attempts: u32, + /// Backoff base delay; attempt *n* waits `base_delay * 2^n` (default: 100 ms). + pub base_delay: Duration, + /// Backoff ceiling (default: 5 s). + pub max_delay: Duration, + /// Multiply each delay by a random factor in `[0.5, 1.5)` (default: true). + pub jitter: bool, +} + +impl Default for RetryConfig { + fn default() -> Self { + Self { + max_attempts: 3, + base_delay: Duration::from_millis(100), + max_delay: Duration::from_secs(5), + jitter: true, + } + } +} + +/// Circuit breaker configuration. +/// +/// Defaults mirror the TypeScript SDK's production preset +/// (`PRODUCTION_RELIABILITY` in `cachekit-ts/src/intents.ts`). +#[derive(Debug, Clone, PartialEq)] +pub struct CircuitBreakerConfig { + /// Retryable failures within [`Self::rolling_window`] before the circuit + /// opens (default: 5). + pub failure_threshold: u32, + /// Successes in half-open state required to close the circuit (default: 3). + pub success_threshold: u32, + /// How long the circuit stays open before allowing half-open probes + /// (default: 5 s). + pub open_timeout: Duration, + /// Maximum concurrent probe calls in half-open state (default: 3). + pub half_open_max_calls: u32, + /// Rolling window for failure counting (default: 60 s). + pub rolling_window: Duration, +} + +impl Default for CircuitBreakerConfig { + fn default() -> Self { + Self { + failure_threshold: 5, + success_threshold: 3, + open_timeout: Duration::from_secs(5), + half_open_max_calls: 3, + rolling_window: Duration::from_secs(60), + } + } +} + +/// Reliability stack configuration: which layers to apply around backend ops. +/// +/// The `Default` enables both layers with production defaults. Disable a +/// layer by setting its field to `None`: +/// +/// ``` +/// use cachekit::reliability::ReliabilityConfig; +/// +/// let retry_only = ReliabilityConfig { +/// circuit_breaker: None, +/// ..ReliabilityConfig::default() +/// }; +/// assert!(retry_only.retry.is_some()); +/// ``` +#[derive(Debug, Clone, PartialEq)] +pub struct ReliabilityConfig { + /// Retry policy, or `None` to propagate every error on first failure. + pub retry: Option, + /// Circuit breaker, or `None` to never fail fast. + pub circuit_breaker: Option, +} + +impl Default for ReliabilityConfig { + fn default() -> Self { + Self { + retry: Some(RetryConfig::default()), + circuit_breaker: Some(CircuitBreakerConfig::default()), + } + } +} + +// ── Jitter ─────────────────────────────────────────────────────────────────── + +/// Uniform random in `[0, 1)`. uuid v4 is the crate's existing entropy source +/// (getrandom-backed); jitter needs decorrelation across clients, not crypto +/// quality — 53 bits is plenty. +fn random_unit() -> f64 { + let bits = uuid::Uuid::new_v4().as_u128() & ((1u128 << 53) - 1); + (bits as f64) / ((1u64 << 53) as f64) +} + +// ── RetryPolicy ────────────────────────────────────────────────────────────── + +/// Retries an operation on errors where [`crate::error::BackendErrorKind::is_retryable`] is +/// true, sleeping a truncated exponential backoff (with jitter) between +/// attempts. `Permanent` and `Authentication` errors propagate immediately. +#[derive(Debug)] +pub(crate) struct RetryPolicy { + config: RetryConfig, +} + +impl RetryPolicy { + pub(crate) fn new(config: RetryConfig) -> Self { + Self { config } + } + + fn delay(&self, attempt: u32) -> Duration { + let exp = self + .config + .base_delay + .saturating_mul(2u32.saturating_pow(attempt)); + let capped = exp.min(self.config.max_delay); + if self.config.jitter { + capped.mul_f64(0.5 + random_unit()) + } else { + capped + } + } + + pub(crate) async fn execute(&self, f: F) -> Result + where + F: Fn() -> Fut, + Fut: Future>, + { + let mut attempt: u32 = 0; + loop { + match f().await { + Ok(v) => return Ok(v), + Err(e) if e.kind.is_retryable() && attempt + 1 < self.config.max_attempts => { + tokio::time::sleep(self.delay(attempt)).await; + attempt += 1; + } + Err(e) => return Err(e), + } + } + } +} + +// ── CircuitBreaker ─────────────────────────────────────────────────────────── + +/// Circuit breaker states, exposed for observability and tests. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CircuitState { + /// Normal operation; calls pass through. + Closed, + /// Failing fast; calls return a [`crate::error::BackendErrorKind::CircuitOpen`] error + /// without reaching the backend. + Open, + /// Probing recovery with a bounded number of trial calls. + HalfOpen, +} + +#[derive(Debug)] +enum State { + Closed, + Open { since: Instant }, + HalfOpen, +} + +#[derive(Debug)] +struct BreakerInner { + state: State, + /// Timestamps of counted failures inside the rolling window. + failures: Vec, + half_open_successes: u32, + half_open_calls: u32, +} + +/// How a completed call is reported back to the breaker. +enum Outcome { + Success, + /// A retryable-kind failure — a backend-health signal. + Failure, + /// A non-retryable failure (permanent/auth) — request-specific, does not + /// count toward opening the circuit but must release its half-open slot, + /// or a burst of permanent errors would wedge the breaker half-open. + Neutral, +} + +/// State machine: closed → (failures ≥ threshold in window) → open → +/// (open_timeout elapsed) → half-open → (successes ≥ threshold) → closed, +/// or (any counted failure) → open. +#[derive(Debug)] +pub(crate) struct CircuitBreaker { + config: CircuitBreakerConfig, + inner: Mutex, +} + +impl CircuitBreaker { + pub(crate) fn new(config: CircuitBreakerConfig) -> Self { + Self { + config, + inner: Mutex::new(BreakerInner { + state: State::Closed, + failures: Vec::new(), + half_open_successes: 0, + half_open_calls: 0, + }), + } + } + + fn lock(&self) -> std::sync::MutexGuard<'_, BreakerInner> { + // A poisoned lock means a panic mid-update; breaker state is advisory, + // so recovering the guard is strictly better than propagating panics. + self.inner.lock().unwrap_or_else(PoisonError::into_inner) + } + + /// Current state (transitions open → half-open lazily on inspection). + /// Test-only until the observability tier (LAB-101) needs it at runtime. + #[cfg(test)] + pub(crate) fn state(&self) -> CircuitState { + let mut inner = self.lock(); + self.maybe_half_open(&mut inner); + match inner.state { + State::Closed => CircuitState::Closed, + State::Open { .. } => CircuitState::Open, + State::HalfOpen => CircuitState::HalfOpen, + } + } + + fn maybe_half_open(&self, inner: &mut BreakerInner) { + if let State::Open { since } = inner.state { + if since.elapsed() >= self.config.open_timeout { + inner.state = State::HalfOpen; + inner.half_open_successes = 0; + inner.half_open_calls = 0; + } + } + } + + /// Admit a call, or fail fast with a circuit-open error. + fn try_acquire(&self) -> Result<(), BackendError> { + let mut inner = self.lock(); + self.maybe_half_open(&mut inner); + match inner.state { + State::Closed => Ok(()), + State::Open { .. } => Err(BackendError::circuit_open( + "circuit breaker is open: backend calls are failing fast", + )), + State::HalfOpen => { + if inner.half_open_calls >= self.config.half_open_max_calls { + Err(BackendError::circuit_open( + "circuit breaker is half-open and the probe limit is reached", + )) + } else { + inner.half_open_calls += 1; + Ok(()) + } + } + } + } + + fn record(&self, outcome: &Outcome) { + let mut inner = self.lock(); + match outcome { + Outcome::Success => { + if matches!(inner.state, State::HalfOpen) { + inner.half_open_successes += 1; + if inner.half_open_successes >= self.config.success_threshold { + inner.state = State::Closed; + inner.failures.clear(); + inner.half_open_successes = 0; + inner.half_open_calls = 0; + } + } + } + Outcome::Failure => match inner.state { + State::HalfOpen => { + inner.state = State::Open { + since: Instant::now(), + }; + inner.half_open_successes = 0; + inner.half_open_calls = 0; + } + State::Closed => { + let now = Instant::now(); + inner.failures.push(now); + let window = self.config.rolling_window; + inner.failures.retain(|t| now.duration_since(*t) <= window); + if inner.failures.len() >= self.config.failure_threshold as usize { + inner.state = State::Open { since: now }; + inner.failures.clear(); + } + } + // Open without an admitted call cannot report a failure; + // ignore rather than extend the open window. + State::Open { .. } => {} + }, + Outcome::Neutral => { + if matches!(inner.state, State::HalfOpen) { + inner.half_open_calls = inner.half_open_calls.saturating_sub(1); + } + } + } + } +} + +// ── ReliableBackend ────────────────────────────────────────────────────────── + +/// Decorator that applies the reliability stack to every cache operation of +/// an inner [`Backend`]: `breaker(retry(op))`. +/// +/// - `get`/`set`/`delete`/`exists` are retried on retryable errors and gated +/// by the circuit breaker. +/// - `health` passes through unguarded — it is a diagnostic and must keep +/// reporting truthfully while the breaker fails data calls fast. +/// - [`Backend::as_lockable`] forwards to the inner backend so distributed +/// fill locks bypass retry/breaker (locks are best-effort advisory). +pub(crate) struct ReliableBackend { + inner: SharedBackend, + retry: Option, + breaker: Option, +} + +impl ReliableBackend { + pub(crate) fn new(inner: SharedBackend, config: ReliabilityConfig) -> Self { + Self { + inner, + retry: config.retry.map(RetryPolicy::new), + breaker: config.circuit_breaker.map(CircuitBreaker::new), + } + } + + async fn guarded(&self, f: F) -> Result + where + F: Fn() -> Fut, + Fut: Future>, + { + if let Some(cb) = &self.breaker { + cb.try_acquire()?; + } + let result = match &self.retry { + Some(retry) => retry.execute(f).await, + None => f().await, + }; + if let Some(cb) = &self.breaker { + let outcome = match &result { + Ok(_) => Outcome::Success, + Err(e) if e.kind.is_retryable() => Outcome::Failure, + Err(_) => Outcome::Neutral, + }; + cb.record(&outcome); + } + result + } +} + +#[cfg_attr(not(feature = "unsync"), async_trait)] +#[cfg_attr(feature = "unsync", async_trait(?Send))] +impl Backend for ReliableBackend { + async fn get(&self, key: &str) -> Result>, BackendError> { + self.guarded(|| self.inner.get(key)).await + } + + async fn set( + &self, + key: &str, + value: Vec, + ttl: Option, + ) -> Result<(), BackendError> { + // Clone per attempt: the inner call consumes the buffer. + self.guarded(|| self.inner.set(key, value.clone(), ttl)) + .await + } + + async fn delete(&self, key: &str) -> Result { + self.guarded(|| self.inner.delete(key)).await + } + + async fn exists(&self, key: &str) -> Result { + self.guarded(|| self.inner.exists(key)).await + } + + async fn health(&self) -> Result { + self.inner.health().await + } + + fn as_lockable(&self) -> Option<&dyn LockableBackend> { + self.inner.as_lockable() + } +} + +/// Wrap `inner` in a [`ReliableBackend`] and re-share it. +#[cfg(not(feature = "unsync"))] +pub(crate) fn wrap_reliable(inner: SharedBackend, config: ReliabilityConfig) -> SharedBackend { + std::sync::Arc::new(ReliableBackend::new(inner, config)) +} + +/// Wrap `inner` in a [`ReliableBackend`] and re-share it (`?Send` variant). +#[cfg(feature = "unsync")] +pub(crate) fn wrap_reliable(inner: SharedBackend, config: ReliabilityConfig) -> SharedBackend { + std::rc::Rc::new(ReliableBackend::new(inner, config)) +} + +// ── Unit tests ─────────────────────────────────────────────────────────────── + +#[cfg(test)] +#[allow(clippy::expect_used)] // test-only: failed acquire/probe should panic loudly +mod tests { + use super::*; + use crate::error::BackendErrorKind; + + fn breaker(failure_threshold: u32, open_timeout: Duration) -> CircuitBreaker { + CircuitBreaker::new(CircuitBreakerConfig { + failure_threshold, + success_threshold: 2, + open_timeout, + half_open_max_calls: 2, + rolling_window: Duration::from_secs(60), + }) + } + + #[test] + fn breaker_opens_after_threshold_and_fails_fast() { + let cb = breaker(3, Duration::from_secs(60)); + for _ in 0..3 { + cb.try_acquire().expect("closed breaker admits calls"); + cb.record(&Outcome::Failure); + } + assert_eq!(cb.state(), CircuitState::Open); + let err = cb.try_acquire().expect_err("open breaker fails fast"); + assert_eq!(err.kind, BackendErrorKind::CircuitOpen); + assert!(!err.kind.is_retryable()); + } + + #[test] + fn breaker_ignores_permanent_errors() { + let cb = breaker(2, Duration::from_secs(60)); + for _ in 0..10 { + cb.try_acquire().expect("closed breaker admits calls"); + cb.record(&Outcome::Neutral); + } + assert_eq!(cb.state(), CircuitState::Closed); + } + + #[test] + fn breaker_half_open_recovers_on_successes() { + let cb = breaker(1, Duration::from_millis(0)); + cb.try_acquire().expect("closed breaker admits calls"); + cb.record(&Outcome::Failure); + // open_timeout of zero → immediately half-open on next inspection + assert_eq!(cb.state(), CircuitState::HalfOpen); + for _ in 0..2 { + cb.try_acquire().expect("half-open admits probes"); + cb.record(&Outcome::Success); + } + assert_eq!(cb.state(), CircuitState::Closed); + } + + #[test] + fn breaker_half_open_reopens_on_failure() { + let cb = breaker(1, Duration::from_millis(0)); + cb.try_acquire().expect("closed breaker admits calls"); + cb.record(&Outcome::Failure); + assert_eq!(cb.state(), CircuitState::HalfOpen); + cb.try_acquire().expect("half-open admits a probe"); + cb.record(&Outcome::Failure); + // Freshly re-opened with a zero timeout flips half-open again on + // inspection, so assert via the internal state before inspecting. + assert!(matches!(cb.lock().state, State::Open { .. })); + } + + #[test] + fn breaker_half_open_slot_released_by_neutral_outcome() { + let cb = breaker(1, Duration::from_millis(0)); + cb.try_acquire().expect("closed breaker admits calls"); + cb.record(&Outcome::Failure); + assert_eq!(cb.state(), CircuitState::HalfOpen); + // Exhaust both probe slots with permanent errors... + cb.try_acquire().expect("probe slot 1"); + cb.record(&Outcome::Neutral); + cb.try_acquire().expect("probe slot 2"); + cb.record(&Outcome::Neutral); + // ...and the breaker still admits probes instead of wedging. + cb.try_acquire() + .expect("neutral outcomes release their probe slots"); + } + + #[test] + fn retry_delay_is_capped_and_jittered() { + let policy = RetryPolicy::new(RetryConfig { + max_attempts: 5, + base_delay: Duration::from_millis(100), + max_delay: Duration::from_millis(300), + jitter: true, + }); + for attempt in 0..10 { + let d = policy.delay(attempt); + // cap 300ms × jitter [0.5, 1.5) → strictly under 450ms + assert!(d < Duration::from_millis(450), "attempt {attempt}: {d:?}"); + } + let no_jitter = RetryPolicy::new(RetryConfig { + jitter: false, + ..RetryConfig::default() + }); + assert_eq!(no_jitter.delay(0), Duration::from_millis(100)); + assert_eq!(no_jitter.delay(1), Duration::from_millis(200)); + assert_eq!(no_jitter.delay(20), Duration::from_secs(5)); + } +} diff --git a/crates/cachekit/tests/macro_tests.rs b/crates/cachekit/tests/macro_tests.rs index aca40b6..30311bf 100644 --- a/crates/cachekit/tests/macro_tests.rs +++ b/crates/cachekit/tests/macro_tests.rs @@ -297,3 +297,162 @@ async fn macro_key_delegates_to_interop_key() { let keys: Vec = backend.inner.store.lock().await.keys().cloned().collect(); assert_eq!(keys, vec![expected]); } + +// ── Reliability behaviour (LAB-518) ────────────────────────────────────────── + +/// Backend where every data operation fails with a transient error. +#[derive(Debug, Default, Clone)] +struct DownBackend; + +impl DownBackend { + fn shared() -> SharedBackend { + #[cfg(not(any(target_arch = "wasm32", feature = "unsync")))] + { + std::sync::Arc::new(Self) + } + #[cfg(any(target_arch = "wasm32", feature = "unsync"))] + { + std::rc::Rc::new(Self) + } + } +} + +#[cfg_attr(not(any(target_arch = "wasm32", feature = "unsync")), async_trait)] +#[cfg_attr(any(target_arch = "wasm32", feature = "unsync"), async_trait(?Send))] +impl Backend for DownBackend { + async fn get(&self, _key: &str) -> Result>, BackendError> { + Err(BackendError::transient("backend down")) + } + + async fn set( + &self, + _key: &str, + _value: Vec, + _ttl: Option, + ) -> Result<(), BackendError> { + Err(BackendError::transient("backend down")) + } + + async fn delete(&self, _key: &str) -> Result { + Err(BackendError::transient("backend down")) + } + + async fn exists(&self, _key: &str) -> Result { + Err(BackendError::transient("backend down")) + } + + async fn health(&self) -> Result { + Err(BackendError::transient("backend down")) + } +} + +static FAIL_OPEN_RUNS: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0); + +#[cachekit(client = cache, ttl = 60, interop = "fail_open_op", namespace = "reliab")] +async fn fail_open_op(cache: &CacheKit, id: u64) -> Result { + FAIL_OPEN_RUNS.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Ok(User { + name: format!("degraded {id}"), + }) +} + +#[tokio::test] +async fn macro_fails_open_when_backend_down() { + let cache = CacheKit::builder() + .backend(DownBackend::shared()) + .no_l1() + .build() + .expect("client builds"); + + // Every call runs the function uncached — graceful degradation, matching + // cachekit-py's BackendError → execute-without-caching posture. + let user = fail_open_op(&cache, 1) + .await + .expect("fail-open: function runs uncached"); + assert_eq!(user.name, "degraded 1"); + let _ = fail_open_op(&cache, 1).await.expect("still degrading"); + assert_eq!( + FAIL_OPEN_RUNS.load(std::sync::atomic::Ordering::SeqCst), + 2, + "nothing was cached while the backend was down" + ); +} + +#[cfg(feature = "encryption")] +static SECURE_RUNS: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0); + +#[cfg(feature = "encryption")] +#[cachekit(client = cache, ttl = 60, interop = "secure_op", namespace = "reliab", secure)] +async fn secure_op(cache: &CacheKit, id: u64) -> Result { + SECURE_RUNS.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Ok(User { + name: format!("secret {id}"), + }) +} + +#[cfg(feature = "encryption")] +#[tokio::test] +async fn macro_secure_fails_closed_when_backend_down() { + let cache = CacheKit::builder() + .backend(DownBackend::shared()) + .encryption_from_bytes(&[7u8; 32], "default") + .expect("encryption configures") + .no_l1() + .build() + .expect("client builds"); + + // Encrypted paths never silently degrade: the backend error reaches the + // caller and the wrapped function does NOT run. + let err = secure_op(&cache, 1) + .await + .expect_err("fail-closed: error propagates"); + assert!(matches!(err, CachekitError::Backend(_)), "got: {err:?}"); + assert_eq!( + SECURE_RUNS.load(std::sync::atomic::Ordering::SeqCst), + 0, + "secure path must not fail open into uncached execution" + ); +} + +static SLOW_OP_RUNS: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0); + +#[cachekit(client = cache, ttl = 60, interop = "slow_op", namespace = "flight")] +async fn slow_op(cache: &CacheKit, id: u64) -> Result { + SLOW_OP_RUNS.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + tokio::time::sleep(Duration::from_millis(50)).await; + Ok(User { + name: format!("slow {id}"), + }) +} + +#[tokio::test] +async fn macro_single_flight_collapses_concurrent_misses() { + let (cache, backend) = mock_client_counting(); + let cache = std::sync::Arc::new(cache); + + // Barrier-align the tasks so their initial cache checks all miss before + // the leader finishes computing. + let barrier = std::sync::Arc::new(tokio::sync::Barrier::new(5)); + let tasks: Vec<_> = (0..5) + .map(|_| { + let cache = std::sync::Arc::clone(&cache); + let barrier = std::sync::Arc::clone(&barrier); + tokio::spawn(async move { + barrier.wait().await; + slow_op(&cache, 9).await + }) + }) + .collect(); + + for task in tasks { + let user = task.await.expect("task completes").expect("call succeeds"); + assert_eq!(user.name, "slow 9"); + } + + assert_eq!( + SLOW_OP_RUNS.load(std::sync::atomic::Ordering::SeqCst), + 1, + "five concurrent misses must execute the function exactly once" + ); + assert_eq!(backend.sets(), 1, "and write the cache exactly once"); +} diff --git a/crates/cachekit/tests/reliability_tests.rs b/crates/cachekit/tests/reliability_tests.rs new file mode 100644 index 0000000..75e9060 --- /dev/null +++ b/crates/cachekit/tests/reliability_tests.rs @@ -0,0 +1,509 @@ +//! Integration tests for the reliability tier (LAB-518): retry, circuit +//! breaker, and single-flight distributed-lock wiring. +//! +//! Run with: +//! cargo test --test reliability_tests --features reliability + +#![cfg(all(feature = "reliability", not(target_arch = "wasm32")))] + +mod common; + +use std::collections::HashMap; +use std::sync::atomic::{AtomicU32, Ordering}; +use std::sync::Arc; +use std::time::Duration; + +use async_trait::async_trait; + +use crate::common::MockBackend; +use cachekit::backend::{Backend, HealthStatus, LockableBackend}; +use cachekit::client::SharedBackend; +use cachekit::error::{BackendError, BackendErrorKind}; +use cachekit::reliability::{CircuitBreakerConfig, ReliabilityConfig, RetryConfig}; +use cachekit::{CacheKit, CachekitError}; + +// ── ScriptedBackend ────────────────────────────────────────────────────────── + +/// Backend whose `get` fails `failures_before_success` times with a chosen +/// error kind, then returns a hit. Counts every backend call. +#[derive(Debug)] +struct ScriptedInner { + calls: AtomicU32, + failures_before_success: u32, + kind: BackendErrorKind, +} + +#[derive(Debug, Clone)] +struct ScriptedBackend { + inner: Arc, +} + +impl ScriptedBackend { + fn new_with_handle( + failures_before_success: u32, + kind: BackendErrorKind, + ) -> (SharedBackend, Self) { + let backend = Self { + inner: Arc::new(ScriptedInner { + calls: AtomicU32::new(0), + failures_before_success, + kind, + }), + }; + let handle = backend.clone(); + #[cfg(not(feature = "unsync"))] + let shared: SharedBackend = Arc::new(backend); + #[cfg(feature = "unsync")] + let shared: SharedBackend = std::rc::Rc::new(backend); + (shared, handle) + } + + fn calls(&self) -> u32 { + self.inner.calls.load(Ordering::SeqCst) + } + + fn fail(&self) -> Result>, BackendError> { + match self.inner.kind { + BackendErrorKind::Transient => Err(BackendError::transient("scripted failure")), + BackendErrorKind::Timeout => Err(BackendError::timeout("scripted timeout")), + BackendErrorKind::Authentication => Err(BackendError::auth("scripted auth")), + _ => Err(BackendError::permanent("scripted failure")), + } + } +} + +#[cfg_attr(not(feature = "unsync"), async_trait)] +#[cfg_attr(feature = "unsync", async_trait(?Send))] +impl Backend for ScriptedBackend { + async fn get(&self, _key: &str) -> Result>, BackendError> { + let n = self.inner.calls.fetch_add(1, Ordering::SeqCst); + if n < self.inner.failures_before_success { + self.fail() + } else { + // Any valid MessagePack payload: the msgpack encoding of 7u32. + Ok(Some(rmp_encoded_seven())) + } + } + + async fn set( + &self, + _key: &str, + _value: Vec, + _ttl: Option, + ) -> Result<(), BackendError> { + Ok(()) + } + + async fn delete(&self, _key: &str) -> Result { + Ok(false) + } + + async fn exists(&self, _key: &str) -> Result { + Ok(false) + } + + async fn health(&self) -> Result { + Ok(HealthStatus { + is_healthy: true, + latency_ms: 0.0, + backend_type: "scripted".to_owned(), + details: HashMap::new(), + }) + } +} + +/// MessagePack for the integer 7 — a decodable `u32` payload. +fn rmp_encoded_seven() -> Vec { + vec![0x07] +} + +// ── Config helpers ─────────────────────────────────────────────────────────── + +fn fast_retry(max_attempts: u32) -> RetryConfig { + RetryConfig { + max_attempts, + base_delay: Duration::from_millis(1), + max_delay: Duration::from_millis(5), + jitter: false, + } +} + +fn client_with(backend: SharedBackend, config: ReliabilityConfig) -> CacheKit { + CacheKit::builder() + .backend(backend) + .reliability(config) + .no_l1() + .build() + .expect("client builds") +} + +fn backend_kind(err: &CachekitError) -> Option<&BackendErrorKind> { + match err { + CachekitError::Backend(b) => Some(&b.kind), + _ => None, + } +} + +// ── Retry ──────────────────────────────────────────────────────────────────── + +#[tokio::test] +async fn retry_recovers_from_transient_failures() { + let (shared, handle) = ScriptedBackend::new_with_handle(2, BackendErrorKind::Transient); + let client = client_with( + shared, + ReliabilityConfig { + retry: Some(fast_retry(3)), + circuit_breaker: None, + }, + ); + + let value: u32 = client + .get("k") + .await + .expect("third attempt succeeds") + .expect("value present"); + assert_eq!(value, 7); + assert_eq!(handle.calls(), 3, "two transient failures then a success"); +} + +#[tokio::test] +async fn retry_recovers_from_timeouts() { + let (shared, handle) = ScriptedBackend::new_with_handle(1, BackendErrorKind::Timeout); + let client = client_with( + shared, + ReliabilityConfig { + retry: Some(fast_retry(3)), + circuit_breaker: None, + }, + ); + + let value: Option = client.get("k").await.expect("retry covers the timeout"); + assert_eq!(value, Some(7)); + assert_eq!(handle.calls(), 2); +} + +#[tokio::test] +async fn retry_does_not_touch_permanent_errors() { + let (shared, handle) = ScriptedBackend::new_with_handle(u32::MAX, BackendErrorKind::Permanent); + let client = client_with( + shared, + ReliabilityConfig { + retry: Some(fast_retry(3)), + circuit_breaker: None, + }, + ); + + let err = client + .get::("k") + .await + .expect_err("permanent error propagates"); + assert_eq!(backend_kind(&err), Some(&BackendErrorKind::Permanent)); + assert_eq!(handle.calls(), 1, "permanent errors are never retried"); +} + +#[tokio::test] +async fn retry_exhausts_attempts_then_propagates() { + let (shared, handle) = ScriptedBackend::new_with_handle(u32::MAX, BackendErrorKind::Transient); + let client = client_with( + shared, + ReliabilityConfig { + retry: Some(fast_retry(3)), + circuit_breaker: None, + }, + ); + + let err = client.get::("k").await.expect_err("all attempts fail"); + assert_eq!(backend_kind(&err), Some(&BackendErrorKind::Transient)); + assert_eq!(handle.calls(), 3, "exactly max_attempts calls"); +} + +// ── Circuit breaker ────────────────────────────────────────────────────────── + +#[tokio::test] +async fn breaker_opens_and_fails_fast_without_touching_backend() { + let (shared, handle) = ScriptedBackend::new_with_handle(u32::MAX, BackendErrorKind::Transient); + let client = client_with( + shared, + ReliabilityConfig { + retry: None, + circuit_breaker: Some(CircuitBreakerConfig { + failure_threshold: 2, + open_timeout: Duration::from_secs(60), + ..CircuitBreakerConfig::default() + }), + }, + ); + + for _ in 0..2 { + let err = client.get::("k").await.expect_err("backend failing"); + assert_eq!(backend_kind(&err), Some(&BackendErrorKind::Transient)); + } + assert_eq!(handle.calls(), 2); + + let err = client.get::("k").await.expect_err("circuit is open"); + assert_eq!(backend_kind(&err), Some(&BackendErrorKind::CircuitOpen)); + assert_eq!(handle.calls(), 2, "open circuit never reaches the backend"); +} + +#[tokio::test] +async fn breaker_half_open_probe_recovers() { + // One failure opens the circuit; after open_timeout a probe succeeds and + // (success_threshold: 1) closes it again. + let (shared, handle) = ScriptedBackend::new_with_handle(1, BackendErrorKind::Transient); + let client = client_with( + shared, + ReliabilityConfig { + retry: None, + circuit_breaker: Some(CircuitBreakerConfig { + failure_threshold: 1, + success_threshold: 1, + open_timeout: Duration::from_millis(50), + half_open_max_calls: 3, + rolling_window: Duration::from_secs(60), + }), + }, + ); + + client + .get::("k") + .await + .expect_err("first call fails and opens"); + let err = client.get::("k").await.expect_err("open: fails fast"); + assert_eq!(backend_kind(&err), Some(&BackendErrorKind::CircuitOpen)); + assert_eq!(handle.calls(), 1); + + tokio::time::sleep(Duration::from_millis(60)).await; + + let value: Option = client.get("k").await.expect("half-open probe passes"); + assert_eq!(value, Some(7)); + let value: Option = client.get("k").await.expect("circuit closed again"); + assert_eq!(value, Some(7)); + assert_eq!(handle.calls(), 3); +} + +#[tokio::test] +async fn breaker_counts_one_failure_per_exhausted_retry_sequence() { + // retry(2) inside breaker(threshold 2): two client calls = 4 backend + // attempts but only 2 breaker failures — the third call fails fast. + let (shared, handle) = ScriptedBackend::new_with_handle(u32::MAX, BackendErrorKind::Transient); + let client = client_with( + shared, + ReliabilityConfig { + retry: Some(fast_retry(2)), + circuit_breaker: Some(CircuitBreakerConfig { + failure_threshold: 2, + open_timeout: Duration::from_secs(60), + ..CircuitBreakerConfig::default() + }), + }, + ); + + for _ in 0..2 { + client.get::("k").await.expect_err("backend failing"); + } + assert_eq!(handle.calls(), 4, "2 calls x 2 attempts"); + + let err = client.get::("k").await.expect_err("circuit open"); + assert_eq!(backend_kind(&err), Some(&BackendErrorKind::CircuitOpen)); + assert_eq!(handle.calls(), 4); +} + +// ── Single-flight: in-process ──────────────────────────────────────────────── + +#[tokio::test] +async fn single_flight_leader_never_waits() { + let client = CacheKit::builder() + .backend(MockBackend::shared()) + .no_l1() + .build() + .expect("client builds"); + + let mut flight = client.single_flight("cold").await; + assert!( + !flight.awaiting_fill().await, + "an uncontested leader computes immediately — a re-check would be a second billable miss" + ); + flight.release().await; +} + +#[tokio::test] +async fn single_flight_follower_rechecks_once() { + let client = Arc::new( + CacheKit::builder() + .backend(MockBackend::shared()) + .no_l1() + .build() + .expect("client builds"), + ); + + let leader = client.single_flight("cold").await; + + let follower_client = Arc::clone(&client); + let follower = tokio::spawn(async move { + let mut flight = follower_client.single_flight("cold").await; + let mut rechecks = 0; + while flight.awaiting_fill().await { + rechecks += 1; + } + flight.release().await; + rechecks + }); + + // The follower must be parked behind the leader's in-process lock. + tokio::time::sleep(Duration::from_millis(50)).await; + assert!(!follower.is_finished(), "follower waits for the leader"); + + leader.release().await; + let rechecks = follower.await.expect("follower task completes"); + assert_eq!( + rechecks, 1, + "a queued follower re-checks the cache exactly once" + ); +} + +// ── Single-flight: distributed lock wiring ─────────────────────────────────── + +/// Lock-capable mock: records acquire/release calls; `grant` controls whether +/// the distributed lock is granted or contested. +#[derive(Debug)] +struct LockingInner { + store: tokio::sync::Mutex>>, + grant: bool, + acquires: AtomicU32, + releases: AtomicU32, +} + +#[derive(Debug, Clone)] +struct LockingBackend { + inner: Arc, +} + +impl LockingBackend { + fn new_with_handle(grant: bool) -> (SharedBackend, Self) { + let backend = Self { + inner: Arc::new(LockingInner { + store: tokio::sync::Mutex::new(HashMap::new()), + grant, + acquires: AtomicU32::new(0), + releases: AtomicU32::new(0), + }), + }; + let handle = backend.clone(); + #[cfg(not(feature = "unsync"))] + let shared: SharedBackend = Arc::new(backend); + #[cfg(feature = "unsync")] + let shared: SharedBackend = std::rc::Rc::new(backend); + (shared, handle) + } +} + +#[cfg_attr(not(feature = "unsync"), async_trait)] +#[cfg_attr(feature = "unsync", async_trait(?Send))] +impl Backend for LockingBackend { + async fn get(&self, key: &str) -> Result>, BackendError> { + Ok(self.inner.store.lock().await.get(key).cloned()) + } + + async fn set( + &self, + key: &str, + value: Vec, + _ttl: Option, + ) -> Result<(), BackendError> { + self.inner.store.lock().await.insert(key.to_owned(), value); + Ok(()) + } + + async fn delete(&self, key: &str) -> Result { + Ok(self.inner.store.lock().await.remove(key).is_some()) + } + + async fn exists(&self, key: &str) -> Result { + Ok(self.inner.store.lock().await.contains_key(key)) + } + + async fn health(&self) -> Result { + Ok(HealthStatus { + is_healthy: true, + latency_ms: 0.0, + backend_type: "locking-mock".to_owned(), + details: HashMap::new(), + }) + } + + fn as_lockable(&self) -> Option<&dyn LockableBackend> { + Some(self) + } +} + +#[cfg_attr(not(feature = "unsync"), async_trait)] +#[cfg_attr(feature = "unsync", async_trait(?Send))] +impl LockableBackend for LockingBackend { + async fn acquire_lock( + &self, + _key: &str, + _timeout_ms: u64, + ) -> Result, BackendError> { + self.inner.acquires.fetch_add(1, Ordering::SeqCst); + Ok(self.inner.grant.then(|| "lock-1".to_owned())) + } + + async fn release_lock(&self, _key: &str, lock_id: &str) -> Result { + assert_eq!(lock_id, "lock-1"); + self.inner.releases.fetch_add(1, Ordering::SeqCst); + Ok(true) + } +} + +#[tokio::test] +async fn single_flight_leader_takes_and_releases_distributed_lock() { + let (shared, handle) = LockingBackend::new_with_handle(true); + let client = CacheKit::builder() + .backend(shared) + .no_l1() + .build() + .expect("client builds"); + + let mut flight = client.single_flight("cold").await; + assert!(!flight.awaiting_fill().await, "lock granted → leader"); + assert_eq!(handle.inner.acquires.load(Ordering::SeqCst), 1); + + flight.release().await; + assert_eq!( + handle.inner.releases.load(Ordering::SeqCst), + 1, + "release() frees the distributed fill lock" + ); +} + +#[tokio::test] +async fn single_flight_contested_lock_polls_and_finds_remote_fill() { + let (shared, handle) = LockingBackend::new_with_handle(false); + let client = CacheKit::builder() + .backend(shared) + .no_l1() + .build() + .expect("client builds"); + + // Simulate another process completing its fill while we wait. + handle + .inner + .store + .lock() + .await + .insert("cold".to_owned(), rmp_encoded_seven()); + + let mut flight = client.single_flight("cold").await; + assert!( + flight.awaiting_fill().await, + "contested lock → poll for the other process's fill" + ); + let value: Option = client.get("cold").await.expect("get succeeds"); + assert_eq!(value, Some(7), "remote fill is visible on re-check"); + flight.release().await; + assert_eq!( + handle.inner.releases.load(Ordering::SeqCst), + 0, + "no lock held, nothing to release" + ); +} From 57d96fffd07c7a069c97acf56c160291d2258897 Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Fri, 24 Jul 2026 01:04:29 +1000 Subject: [PATCH 2/4] fix(reliability): release half-open probe slot on non-closing success (LAB-518) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The circuit breaker's half_open_calls counter caps in-flight recovery probes, but a Success that had not yet crossed success_threshold never released its slot — only Neutral did. Any config with success_threshold > half_open_max_calls (all fields are pub, unvalidated) therefore wedged the breaker half-open permanently: slots fill, successes stall below threshold, and every call fails fast with CircuitOpen even against a recovered backend. The default 3/3 preset escaped only because threshold == cap. Mirror the Neutral arm and release the slot on a non-closing success; add a regression test with threshold(3) > cap(1). Surfaced by the crypto/protocol expert-panel review. Co-authored-by: multica-agent --- crates/cachekit/src/reliability.rs | 37 ++++++++++++++++++++++++++++++ 1 file changed, 37 insertions(+) diff --git a/crates/cachekit/src/reliability.rs b/crates/cachekit/src/reliability.rs index cda3011..621fb42 100644 --- a/crates/cachekit/src/reliability.rs +++ b/crates/cachekit/src/reliability.rs @@ -302,6 +302,17 @@ impl CircuitBreaker { inner.failures.clear(); inner.half_open_successes = 0; inner.half_open_calls = 0; + } else { + // Release this probe's slot. `half_open_calls` caps the + // number of *in-flight* probes, so a success that does + // not yet close the breaker must free its slot (exactly + // as `Neutral` does). Without this, a config with + // success_threshold > half_open_max_calls wedges the + // breaker half-open forever: the slots fill, successes + // stall below the threshold, and every subsequent call + // fails fast with CircuitOpen even against a healthy + // backend. + inner.half_open_calls = inner.half_open_calls.saturating_sub(1); } } } @@ -517,6 +528,32 @@ mod tests { .expect("neutral outcomes release their probe slots"); } + #[test] + fn breaker_half_open_closes_when_success_threshold_exceeds_probe_cap() { + // success_threshold (3) deliberately exceeds half_open_max_calls (1): + // with a single in-flight probe slot, the breaker can only ever reach + // three successes if each non-closing success RELEASES its slot. Before + // the fix this wedged half-open forever — the slot filled after the + // first success (which stalled at 1 < 3), so no further probe was + // admitted and the breaker never re-closed. + let cb = CircuitBreaker::new(CircuitBreakerConfig { + failure_threshold: 1, + success_threshold: 3, + open_timeout: Duration::from_millis(0), + half_open_max_calls: 1, + rolling_window: Duration::from_secs(60), + }); + cb.try_acquire().expect("closed breaker admits calls"); + cb.record(&Outcome::Failure); + assert_eq!(cb.state(), CircuitState::HalfOpen); + for _ in 0..3 { + cb.try_acquire() + .expect("a non-closing success must release its probe slot"); + cb.record(&Outcome::Success); + } + assert_eq!(cb.state(), CircuitState::Closed); + } + #[test] fn retry_delay_is_capped_and_jittered() { let policy = RetryPolicy::new(RetryConfig { From 387442505a0929dc000f5b7b37a5e917534dd7e5 Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Fri, 24 Jul 2026 15:12:37 +1000 Subject: [PATCH 3/4] =?UTF-8?q?fix(reliability):=20panel=20FIX-FIRST=20?= =?UTF-8?q?=E2=80=94=20RAII=20probe=20permit,=20outage-scoped=20fail-open,?= =?UTF-8?q?=20API=20cuts=20(LAB-518)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Expert-panel review findings on PR #43: - CRIT: half-open probe slots leaked when a guarded future was cancelled (caller timeout/select!) or panicked before recording an outcome — leak half_open_max_calls of them and the breaker wedges half-open forever, CircuitOpen on every call even against a recovered backend. try_acquire now returns an RAII ProbePermit: complete() hands accounting to the outcome arms; Drop releases the slot (like Neutral, no transition) on every other exit path. The manual counter pairs that leaked twice are gone. Unit tests cover dropped-permit release and closed-state permits not touching half-open accounting; an integration test cancels a hanging probe via tokio::time::timeout and proves the breaker still recovers. - MAJ: the plain-path fail-open arm swallowed permanent/auth backend errors — a wrong API key silently ran uncached forever while looking healthy. Fail-open is now scoped to outage-class errors (retryable + CircuitOpen); permanent/auth propagate. secure unchanged (fail-closed on everything). - MIN: SingleFlight::awaiting_fill → wait_for_fill — it sleeps and mutates; the old name read as a pure predicate. - Cuts: no_reliability() deleted (opt out by passing a config with both layers None — build() skips the no-op wrap); CircuitState demoted to cfg(test) until LAB-101 consumes it at runtime. Co-authored-by: multica-agent --- README.md | 5 +- crates/cachekit-macros/src/lib.rs | 38 +++-- crates/cachekit/src/client.rs | 19 +-- crates/cachekit/src/flight.rs | 4 +- crates/cachekit/src/intents.rs | 4 +- crates/cachekit/src/reliability.rs | 164 ++++++++++++++++----- crates/cachekit/tests/macro_tests.rs | 79 ++++++++++ crates/cachekit/tests/reliability_tests.rs | 111 +++++++++++++- 8 files changed, 358 insertions(+), 66 deletions(-) diff --git a/README.md b/README.md index a582c1b..86f3370 100644 --- a/README.md +++ b/README.md @@ -293,7 +293,7 @@ With the `reliability` feature (default, native only), the `production`, `encryp |:------|:-------------|:---------| | **Retry** | Truncated exponential backoff + jitter on transient/timeout errors (`BackendErrorKind::is_retryable`); permanent and auth errors propagate immediately | 3 attempts, 100 ms base, 5 s cap, jitter ×[0.5, 1.5) | | **Circuit breaker** | closed → open after N retryable failures in a rolling window; fails fast (`BackendErrorKind::CircuitOpen`) while open; half-open probes recovery | threshold 5, window 60 s, open 5 s, 3 probes, close after 3 successes | -| **Graceful degradation** | On backend failure, `#[cachekit]`-wrapped functions run uncached (fail-open). `secure` paths fail **closed** — encrypted workloads never silently degrade | built into the macro | +| **Graceful degradation** | On outage-class backend failure (transient, timeout, open breaker), `#[cachekit]`-wrapped functions run uncached (fail-open); permanent/auth errors propagate — a wrong API key fails loudly. `secure` paths fail **closed** on everything — encrypted workloads never silently degrade | built into the macro | | **Single-flight** | Concurrent misses of one key collapse to a single execution: per-key in-process lock, plus a distributed fill lock across processes on lock-capable backends (cachekit.io, Redis) | in-process always on; cross-process 5 s lock, 100 ms polls | Retry sits *inside* the breaker (one exhausted retry sequence = one breaker failure), degradation and single-flight sit in the `#[cachekit]` macro around the miss path — the same composition as the TypeScript SDK's `ReliabilityExecutor` and the Python decorator. @@ -310,8 +310,9 @@ let cache = CacheKit::production("redis://localhost:6379").await? }) .build()?; +// Opt a preset out: a config with both layers `None` applies no wrapping. let bare = CacheKit::production("redis://localhost:6379").await? - .no_reliability() + .reliability(ReliabilityConfig { retry: None, circuit_breaker: None }) .build()?; ``` diff --git a/crates/cachekit-macros/src/lib.rs b/crates/cachekit-macros/src/lib.rs index 087cfc7..83a6378 100644 --- a/crates/cachekit-macros/src/lib.rs +++ b/crates/cachekit-macros/src/lib.rs @@ -205,11 +205,14 @@ fn extract_ok_type(ret: &ReturnType) -> syn::Result { /// /// # Reliability behaviour /// -/// - **Graceful degradation**: on a backend failure the plain path fails -/// *open* — the function executes uncached and its result is returned -/// (matching cachekit-py). With `secure`, backend and decryption errors -/// fail *closed* and propagate to the caller: an encrypted workload never -/// silently degrades. +/// - **Graceful degradation**: on an outage-class backend failure — +/// transient, timeout, or an open circuit breaker — the plain path fails +/// *open*: the function executes uncached and its result is returned. +/// Permanent and authentication errors propagate even on the plain path +/// (a wrong API key must fail loudly, not silently disable caching +/// forever). With `secure`, *every* backend and decryption error fails +/// *closed* and propagates: an encrypted workload never silently +/// degrades. /// - **Cold-miss single-flight**: concurrent calls that miss on the same /// key are collapsed to one execution per process (and per fleet, when /// the backend supports distributed fill locks — CachekitIO and Redis do, @@ -310,15 +313,26 @@ fn expand(args: &MacroArgs, mut func: ItemFn) -> syn::Result { ) }; - // Graceful degradation (LAB-518): on a backend failure the plain path - // fails OPEN — the wrapped function runs uncached, mirroring cachekit-py's - // BackendError → execute-without-caching posture. The `secure` path stays - // fail-CLOSED: backend and decryption errors reach the caller, so an - // encrypted workload never silently degrades. + // Graceful degradation (LAB-518): on an OUTAGE-class backend failure — + // retryable (transient/timeout) or a fast-failing open circuit breaker — + // the plain path fails OPEN: the wrapped function runs uncached, so a + // cache outage costs performance, not availability. Permanent and + // authentication errors PROPAGATE even on the plain path: a wrong API + // key that silently fell open would run uncached forever with zero + // signal while looking healthy (expert-panel finding). The `secure` + // path stays fail-CLOSED on everything: backend and decryption errors + // reach the caller, so an encrypted workload never silently degrades. let fail_open_arm = if args.secure { quote! {} } else { - quote! { Err(cachekit::error::CachekitError::Backend(_)) => {} } + quote! { + Err(cachekit::error::CachekitError::Backend(__ck_be)) + if __ck_be.kind.is_retryable() + || matches!( + __ck_be.kind, + cachekit::error::BackendErrorKind::CircuitOpen + ) => {} + } }; // Capture the original function body. @@ -368,7 +382,7 @@ fn expand(args: &MacroArgs, mut func: ItemFn) -> syn::Result { // to one execution (misses are billable). While another worker is // filling, re-check the cache instead of recomputing. let mut __ck_flight = #client_ident.single_flight(&__ck_key).await; - while __ck_flight.awaiting_fill().await { + while __ck_flight.wait_for_fill().await { match #get_expr { Ok(Some(__ck_cached)) => { __ck_flight.release().await; diff --git a/crates/cachekit/src/client.rs b/crates/cachekit/src/client.rs index 938c9c6..e856c9a 100644 --- a/crates/cachekit/src/client.rs +++ b/crates/cachekit/src/client.rs @@ -605,21 +605,14 @@ impl CacheKitBuilder { /// /// Enabled by default with production settings by the `production`, /// `encrypted`, and `io` intent presets; off for `minimal` and for - /// manually-built clients. + /// manually-built clients. To opt a preset out, pass a config with both + /// layers `None` — an empty config applies no wrapping at all. #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))] pub fn reliability(mut self, config: crate::reliability::ReliabilityConfig) -> Self { self.reliability = Some(config); self } - /// Remove any reliability configuration (backend errors propagate on - /// first failure, no circuit breaker). Overrides a preset's default. - #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))] - pub fn no_reliability(mut self) -> Self { - self.reliability = None; - self - } - /// Configure encryption from raw master key bytes and tenant ID. /// /// The master key must be at least 16 bytes (32 recommended). @@ -693,10 +686,14 @@ impl CacheKitBuilder { }; // Apply the reliability stack last so it decorates the final backend. + // A config with neither layer set is the documented opt-out: skip the + // (no-op) decorator entirely. #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))] let backend = match self.reliability { - Some(config) => crate::reliability::wrap_reliable(backend, config), - None => backend, + Some(config) if config.retry.is_some() || config.circuit_breaker.is_some() => { + crate::reliability::wrap_reliable(backend, config) + } + _ => backend, }; Ok(CacheKit { diff --git a/crates/cachekit/src/flight.rs b/crates/cachekit/src/flight.rs index c919fa5..6385cbb 100644 --- a/crates/cachekit/src/flight.rs +++ b/crates/cachekit/src/flight.rs @@ -24,7 +24,7 @@ //! return Ok(()); //! } //! let mut flight = cache.single_flight("expensive").await; -//! while flight.awaiting_fill().await { +//! while flight.wait_for_fill().await { //! if let Some(_v) = cache.get::("expensive").await? { //! flight.release().await; // another worker filled it //! return Ok(()); @@ -127,7 +127,7 @@ impl SingleFlight { /// - Queued behind a local leader: `true` exactly once. /// - Contested cross-process: sleeps one poll interval per call, `true` /// until the poll budget is spent. - pub async fn awaiting_fill(&mut self) -> bool { + pub async fn wait_for_fill(&mut self) -> bool { match &mut self.role { Role::Leader => false, Role::LocalFollower { rechecked } => { diff --git a/crates/cachekit/src/intents.rs b/crates/cachekit/src/intents.rs index 5655a4e..94d916b 100644 --- a/crates/cachekit/src/intents.rs +++ b/crates/cachekit/src/intents.rs @@ -14,8 +14,8 @@ //! //! ¹ Retry with backoff + jitter and a circuit breaker around backend ops //! (requires the `reliability` feature, on by default — see -//! [`crate::reliability`]). Override with -//! [`CacheKitBuilder::reliability`] / [`CacheKitBuilder::no_reliability`]. +//! [`crate::reliability`]). Override via [`CacheKitBuilder::reliability`]; +//! a config with both layers `None` disables the stack entirely. use std::time::Duration; diff --git a/crates/cachekit/src/reliability.rs b/crates/cachekit/src/reliability.rs index 621fb42..e4dd3d9 100644 --- a/crates/cachekit/src/reliability.rs +++ b/crates/cachekit/src/reliability.rs @@ -179,9 +179,12 @@ impl RetryPolicy { // ── CircuitBreaker ─────────────────────────────────────────────────────────── -/// Circuit breaker states, exposed for observability and tests. +/// Circuit breaker states. Test-only until the observability tier (LAB-101) +/// exposes breaker state at runtime — a public type with no producer is API +/// noise (expert-panel cut). +#[cfg(test)] #[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum CircuitState { +pub(crate) enum CircuitState { /// Normal operation; calls pass through. Closed, /// Failing fast; calls return a [`crate::error::BackendErrorKind::CircuitOpen`] error @@ -270,11 +273,21 @@ impl CircuitBreaker { } /// Admit a call, or fail fast with a circuit-open error. - fn try_acquire(&self) -> Result<(), BackendError> { + /// + /// Returns an RAII [`ProbePermit`]: if the guarded future is cancelled + /// (caller timeout/`select!`) or panics before an outcome is recorded, + /// the permit's `Drop` releases any half-open probe slot it took — + /// otherwise `half_open_max_calls` cancelled probes would wedge the + /// breaker half-open forever, fast-failing every call even against a + /// recovered backend. + fn try_acquire(&self) -> Result, BackendError> { let mut inner = self.lock(); self.maybe_half_open(&mut inner); match inner.state { - State::Closed => Ok(()), + State::Closed => Ok(ProbePermit { + breaker: self, + took_slot: false, + }), State::Open { .. } => Err(BackendError::circuit_open( "circuit breaker is open: backend calls are failing fast", )), @@ -285,7 +298,10 @@ impl CircuitBreaker { )) } else { inner.half_open_calls += 1; - Ok(()) + Ok(ProbePermit { + breaker: self, + took_slot: true, + }) } } } @@ -347,6 +363,46 @@ impl CircuitBreaker { } } +// ── ProbePermit ────────────────────────────────────────────────────────────── + +/// RAII token for a breaker-admitted call. +/// +/// Slot accounting lives in exactly one of two places: [`Self::complete`] +/// (normal return — the outcome arms of `record` own the bookkeeping from +/// there) or `Drop` (cancel/panic — release the slot like `Neutral`, no +/// transition). Manual increment/decrement pairs leaked twice before this +/// guard existed; do not reintroduce them. +#[derive(Debug)] +struct ProbePermit<'a> { + breaker: &'a CircuitBreaker, + /// Whether this admission consumed a half-open probe slot. + took_slot: bool, +} + +impl ProbePermit<'_> { + /// Report the call's outcome and disarm the drop-release. + fn complete(mut self, outcome: &Outcome) { + self.took_slot = false; + self.breaker.record(outcome); + } +} + +impl Drop for ProbePermit<'_> { + fn drop(&mut self) { + if !self.took_slot { + return; + } + // No outcome was recorded: the guarded future was cancelled mid-await + // or panicked. Free the probe slot so the half-open window can keep + // probing; if the breaker transitioned meanwhile (counters reset), + // the saturating decrement is a no-op. + let mut inner = self.breaker.lock(); + if matches!(inner.state, State::HalfOpen) { + inner.half_open_calls = inner.half_open_calls.saturating_sub(1); + } + } +} + // ── ReliableBackend ────────────────────────────────────────────────────────── /// Decorator that applies the reliability stack to every cache operation of @@ -378,20 +434,21 @@ impl ReliableBackend { F: Fn() -> Fut, Fut: Future>, { - if let Some(cb) = &self.breaker { - cb.try_acquire()?; - } + let permit = match &self.breaker { + Some(cb) => Some(cb.try_acquire()?), + None => None, + }; let result = match &self.retry { Some(retry) => retry.execute(f).await, None => f().await, }; - if let Some(cb) = &self.breaker { + if let Some(permit) = permit { let outcome = match &result { Ok(_) => Outcome::Success, Err(e) if e.kind.is_retryable() => Outcome::Failure, Err(_) => Outcome::Neutral, }; - cb.record(&outcome); + permit.complete(&outcome); } result } @@ -462,12 +519,17 @@ mod tests { }) } + /// Admit a call and immediately report its outcome. + fn admit_and(cb: &CircuitBreaker, outcome: &Outcome) { + let permit = cb.try_acquire().expect("breaker admits the call"); + permit.complete(outcome); + } + #[test] fn breaker_opens_after_threshold_and_fails_fast() { let cb = breaker(3, Duration::from_secs(60)); for _ in 0..3 { - cb.try_acquire().expect("closed breaker admits calls"); - cb.record(&Outcome::Failure); + admit_and(&cb, &Outcome::Failure); } assert_eq!(cb.state(), CircuitState::Open); let err = cb.try_acquire().expect_err("open breaker fails fast"); @@ -479,8 +541,7 @@ mod tests { fn breaker_ignores_permanent_errors() { let cb = breaker(2, Duration::from_secs(60)); for _ in 0..10 { - cb.try_acquire().expect("closed breaker admits calls"); - cb.record(&Outcome::Neutral); + admit_and(&cb, &Outcome::Neutral); } assert_eq!(cb.state(), CircuitState::Closed); } @@ -488,13 +549,11 @@ mod tests { #[test] fn breaker_half_open_recovers_on_successes() { let cb = breaker(1, Duration::from_millis(0)); - cb.try_acquire().expect("closed breaker admits calls"); - cb.record(&Outcome::Failure); + admit_and(&cb, &Outcome::Failure); // open_timeout of zero → immediately half-open on next inspection assert_eq!(cb.state(), CircuitState::HalfOpen); for _ in 0..2 { - cb.try_acquire().expect("half-open admits probes"); - cb.record(&Outcome::Success); + admit_and(&cb, &Outcome::Success); } assert_eq!(cb.state(), CircuitState::Closed); } @@ -502,11 +561,9 @@ mod tests { #[test] fn breaker_half_open_reopens_on_failure() { let cb = breaker(1, Duration::from_millis(0)); - cb.try_acquire().expect("closed breaker admits calls"); - cb.record(&Outcome::Failure); + admit_and(&cb, &Outcome::Failure); assert_eq!(cb.state(), CircuitState::HalfOpen); - cb.try_acquire().expect("half-open admits a probe"); - cb.record(&Outcome::Failure); + admit_and(&cb, &Outcome::Failure); // Freshly re-opened with a zero timeout flips half-open again on // inspection, so assert via the internal state before inspecting. assert!(matches!(cb.lock().state, State::Open { .. })); @@ -515,17 +572,16 @@ mod tests { #[test] fn breaker_half_open_slot_released_by_neutral_outcome() { let cb = breaker(1, Duration::from_millis(0)); - cb.try_acquire().expect("closed breaker admits calls"); - cb.record(&Outcome::Failure); + admit_and(&cb, &Outcome::Failure); assert_eq!(cb.state(), CircuitState::HalfOpen); // Exhaust both probe slots with permanent errors... - cb.try_acquire().expect("probe slot 1"); - cb.record(&Outcome::Neutral); - cb.try_acquire().expect("probe slot 2"); - cb.record(&Outcome::Neutral); + admit_and(&cb, &Outcome::Neutral); + admit_and(&cb, &Outcome::Neutral); // ...and the breaker still admits probes instead of wedging. - cb.try_acquire() + let permit = cb + .try_acquire() .expect("neutral outcomes release their probe slots"); + permit.complete(&Outcome::Neutral); } #[test] @@ -543,14 +599,56 @@ mod tests { half_open_max_calls: 1, rolling_window: Duration::from_secs(60), }); - cb.try_acquire().expect("closed breaker admits calls"); - cb.record(&Outcome::Failure); + admit_and(&cb, &Outcome::Failure); assert_eq!(cb.state(), CircuitState::HalfOpen); for _ in 0..3 { - cb.try_acquire() + let permit = cb + .try_acquire() .expect("a non-closing success must release its probe slot"); - cb.record(&Outcome::Success); + permit.complete(&Outcome::Success); + } + assert_eq!(cb.state(), CircuitState::Closed); + } + + #[test] + fn breaker_dropped_permit_releases_probe_slot() { + // A probe future cancelled (caller timeout / select!) or panicked + // before recording an outcome must not consume its slot forever: + // exhaust every half-open slot with plain drops and the breaker must + // still admit probes instead of wedging half-open until restart. + let cb = breaker(1, Duration::from_millis(0)); + admit_and(&cb, &Outcome::Failure); + assert_eq!(cb.state(), CircuitState::HalfOpen); + for _ in 0..2 { + let permit = cb.try_acquire().expect("half-open admits a probe"); + drop(permit); // cancelled before any outcome } + let permit = cb + .try_acquire() + .expect("dropped permits release their probe slots"); + permit.complete(&Outcome::Success); + } + + #[test] + fn breaker_closed_permit_drop_does_not_touch_half_open_accounting() { + // A call admitted while CLOSED holds no probe slot; cancelling it + // must not free (or corrupt) slots in a half-open window that opened + // after its admission. + let cb = breaker(1, Duration::from_millis(0)); + let closed_permit = cb.try_acquire().expect("closed breaker admits calls"); + // Another call's failure opens the breaker, then zero timeout flips + // it half-open with a fresh probe window. + admit_and(&cb, &Outcome::Failure); + assert_eq!(cb.state(), CircuitState::HalfOpen); + let p1 = cb.try_acquire().expect("probe slot 1"); + let p2 = cb.try_acquire().expect("probe slot 2"); + drop(closed_permit); // must be a no-op: it never took a slot + assert!( + cb.try_acquire().is_err(), + "probe cap must still be enforced after a closed-state permit drops" + ); + p1.complete(&Outcome::Success); + p2.complete(&Outcome::Success); assert_eq!(cb.state(), CircuitState::Closed); } diff --git a/crates/cachekit/tests/macro_tests.rs b/crates/cachekit/tests/macro_tests.rs index 30311bf..d7a91ab 100644 --- a/crates/cachekit/tests/macro_tests.rs +++ b/crates/cachekit/tests/macro_tests.rs @@ -456,3 +456,82 @@ async fn macro_single_flight_collapses_concurrent_misses() { ); assert_eq!(backend.sets(), 1, "and write the cache exactly once"); } + +/// Backend where every operation fails with an authentication error — +/// non-retryable, NOT outage-class. +#[derive(Debug, Default, Clone)] +struct AuthFailBackend; + +impl AuthFailBackend { + fn shared() -> SharedBackend { + #[cfg(not(any(target_arch = "wasm32", feature = "unsync")))] + { + std::sync::Arc::new(Self) + } + #[cfg(any(target_arch = "wasm32", feature = "unsync"))] + { + std::rc::Rc::new(Self) + } + } +} + +#[cfg_attr(not(any(target_arch = "wasm32", feature = "unsync")), async_trait)] +#[cfg_attr(any(target_arch = "wasm32", feature = "unsync"), async_trait(?Send))] +impl Backend for AuthFailBackend { + async fn get(&self, _key: &str) -> Result>, BackendError> { + Err(BackendError::auth("invalid API key")) + } + + async fn set( + &self, + _key: &str, + _value: Vec, + _ttl: Option, + ) -> Result<(), BackendError> { + Err(BackendError::auth("invalid API key")) + } + + async fn delete(&self, _key: &str) -> Result { + Err(BackendError::auth("invalid API key")) + } + + async fn exists(&self, _key: &str) -> Result { + Err(BackendError::auth("invalid API key")) + } + + async fn health(&self) -> Result { + Err(BackendError::auth("invalid API key")) + } +} + +static AUTH_FAIL_RUNS: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0); + +#[cachekit(client = cache, ttl = 60, interop = "auth_fail_op", namespace = "reliab")] +async fn auth_fail_op(cache: &CacheKit, id: u64) -> Result { + AUTH_FAIL_RUNS.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Ok(User { + name: format!("never {id}"), + }) +} + +#[tokio::test] +async fn macro_propagates_permanent_backend_errors_on_plain_path() { + // Fail-open covers OUTAGES (transient/timeout/circuit-open). A wrong API + // key is not an outage: silently falling open would run uncached forever + // with zero signal while looking healthy (expert-panel finding). + let cache = CacheKit::builder() + .backend(AuthFailBackend::shared()) + .no_l1() + .build() + .expect("client builds"); + + let err = auth_fail_op(&cache, 1) + .await + .expect_err("authentication errors must propagate, not fail open"); + assert!(matches!(err, CachekitError::Backend(_)), "got: {err:?}"); + assert_eq!( + AUTH_FAIL_RUNS.load(std::sync::atomic::Ordering::SeqCst), + 0, + "the function must not run when the error is not outage-class" + ); +} diff --git a/crates/cachekit/tests/reliability_tests.rs b/crates/cachekit/tests/reliability_tests.rs index 75e9060..a630b10 100644 --- a/crates/cachekit/tests/reliability_tests.rs +++ b/crates/cachekit/tests/reliability_tests.rs @@ -320,7 +320,7 @@ async fn single_flight_leader_never_waits() { let mut flight = client.single_flight("cold").await; assert!( - !flight.awaiting_fill().await, + !flight.wait_for_fill().await, "an uncontested leader computes immediately — a re-check would be a second billable miss" ); flight.release().await; @@ -342,7 +342,7 @@ async fn single_flight_follower_rechecks_once() { let follower = tokio::spawn(async move { let mut flight = follower_client.single_flight("cold").await; let mut rechecks = 0; - while flight.awaiting_fill().await { + while flight.wait_for_fill().await { rechecks += 1; } flight.release().await; @@ -465,7 +465,7 @@ async fn single_flight_leader_takes_and_releases_distributed_lock() { .expect("client builds"); let mut flight = client.single_flight("cold").await; - assert!(!flight.awaiting_fill().await, "lock granted → leader"); + assert!(!flight.wait_for_fill().await, "lock granted → leader"); assert_eq!(handle.inner.acquires.load(Ordering::SeqCst), 1); flight.release().await; @@ -495,7 +495,7 @@ async fn single_flight_contested_lock_polls_and_finds_remote_fill() { let mut flight = client.single_flight("cold").await; assert!( - flight.awaiting_fill().await, + flight.wait_for_fill().await, "contested lock → poll for the other process's fill" ); let value: Option = client.get("cold").await.expect("get succeeds"); @@ -507,3 +507,106 @@ async fn single_flight_contested_lock_polls_and_finds_remote_fill() { "no lock held, nothing to release" ); } + +// ── Cancel-safety (panel CRIT: probe slot must survive cancellation) ───────── + +/// Scripted per call index: fail (opens the breaker), hang (the probe that +/// gets cancelled), then succeed. +#[derive(Debug, Clone, Default)] +struct HangOnceBackend { + calls: Arc, +} + +impl HangOnceBackend { + fn shared() -> SharedBackend { + #[cfg(not(feature = "unsync"))] + { + Arc::new(Self::default()) + } + #[cfg(feature = "unsync")] + { + std::rc::Rc::new(Self::default()) + } + } +} + +#[cfg_attr(not(feature = "unsync"), async_trait)] +#[cfg_attr(feature = "unsync", async_trait(?Send))] +impl Backend for HangOnceBackend { + async fn get(&self, _key: &str) -> Result>, BackendError> { + match self.calls.fetch_add(1, Ordering::SeqCst) { + 0 => Err(BackendError::transient("opening failure")), + 1 => std::future::pending().await, + _ => Ok(Some(rmp_encoded_seven())), + } + } + + async fn set( + &self, + _key: &str, + _value: Vec, + _ttl: Option, + ) -> Result<(), BackendError> { + Ok(()) + } + + async fn delete(&self, _key: &str) -> Result { + Ok(false) + } + + async fn exists(&self, _key: &str) -> Result { + Ok(false) + } + + async fn health(&self) -> Result { + Ok(HealthStatus { + is_healthy: true, + latency_ms: 0.0, + backend_type: "hang-once".to_owned(), + details: HashMap::new(), + }) + } +} + +#[tokio::test] +async fn cancelled_probe_does_not_wedge_the_breaker() { + // One probe slot, one success to close. The half-open probe hangs and the + // caller times out — cancelling the guarded future mid-await. Without the + // RAII permit releasing the slot on drop, that single slot leaks and the + // breaker fast-fails CircuitOpen forever, even after the backend recovers. + let client = client_with( + HangOnceBackend::shared(), + ReliabilityConfig { + retry: None, + circuit_breaker: Some(CircuitBreakerConfig { + failure_threshold: 1, + success_threshold: 1, + open_timeout: Duration::from_millis(5), + half_open_max_calls: 1, + rolling_window: Duration::from_secs(60), + }), + }, + ); + + client + .get::("k") + .await + .expect_err("first call opens the breaker"); + tokio::time::sleep(Duration::from_millis(10)).await; // → half-open + + let cancelled = tokio::time::timeout(Duration::from_millis(20), client.get::("k")).await; + assert!( + cancelled.is_err(), + "the hanging probe must be cancelled by the timeout" + ); + + let value: Option = client + .get("k") + .await + .expect("a fresh probe is admitted after the cancelled one released its slot"); + assert_eq!( + value, + Some(7), + "breaker recovered instead of wedging half-open" + ); +} From d710fbd2a0b0c7a090dc0a38614c01eeb89ac370 Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Fri, 24 Jul 2026 15:25:29 +1000 Subject: [PATCH 4/4] test(reliability): cover the CircuitOpen fail-open branch; document the 5s fill-lock ceiling (LAB-518) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeRabbit round 2: the macro's fail-open guard special-cases BackendErrorKind::CircuitOpen but no test tripped a real breaker on the plain path — added one (threshold 1: transient fall-open opens the circuit, second call falls open on CircuitOpen). Also documented that FILL_LOCK_TIMEOUT_MS bounds cross-process suppression: fills slower than 5 s lose the lock mid-compute and may recompute concurrently (fail-open by design; owner-checked release prevents wrongful deletion). Co-authored-by: multica-agent --- crates/cachekit/src/flight.rs | 8 ++++ crates/cachekit/tests/macro_tests.rs | 64 ++++++++++++++++++++++++++++ 2 files changed, 72 insertions(+) diff --git a/crates/cachekit/src/flight.rs b/crates/cachekit/src/flight.rs index 6385cbb..5256829 100644 --- a/crates/cachekit/src/flight.rs +++ b/crates/cachekit/src/flight.rs @@ -46,6 +46,14 @@ use crate::client::SharedBackend; const SWEEP_THRESHOLD: usize = 128; /// How long a distributed fill lock is held server-side before auto-expiry. +/// +/// Also the cross-process suppression ceiling: a fill that runs longer than +/// this loses its lock mid-compute and another process may recompute +/// concurrently (fail-open by design — release is owner-checked, so an +/// expired lock is never wrongfully deleted). Workloads whose fills +/// routinely approach 5 s keep in-process dedup but should not rely on +/// cross-process suppression. +/// ponytail: fixed TTL, no heartbeat — add lock renewal if slow fills matter. #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))] const FILL_LOCK_TIMEOUT_MS: u64 = 5_000; diff --git a/crates/cachekit/tests/macro_tests.rs b/crates/cachekit/tests/macro_tests.rs index d7a91ab..16f2706 100644 --- a/crates/cachekit/tests/macro_tests.rs +++ b/crates/cachekit/tests/macro_tests.rs @@ -535,3 +535,67 @@ async fn macro_propagates_permanent_backend_errors_on_plain_path() { "the function must not run when the error is not outage-class" ); } + +#[cfg(all(feature = "reliability", not(target_arch = "wasm32")))] +static CIRCUIT_OPEN_RUNS: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0); + +#[cfg(all(feature = "reliability", not(target_arch = "wasm32")))] +#[cachekit(client = cache, ttl = 60, interop = "circuit_open_op", namespace = "reliab")] +async fn circuit_open_op(cache: &CacheKit, id: u64) -> Result { + CIRCUIT_OPEN_RUNS.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Ok(User { + name: format!("breaker-open {id}"), + }) +} + +/// The fail-open arm's other branch: a fast-failing OPEN circuit breaker +/// (`BackendErrorKind::CircuitOpen`) must also fall open on the plain path. +#[cfg(all(feature = "reliability", not(target_arch = "wasm32")))] +#[tokio::test] +async fn macro_fails_open_when_circuit_is_open() { + use cachekit::error::BackendErrorKind; + use cachekit::reliability::{CircuitBreakerConfig, ReliabilityConfig}; + + let cache = CacheKit::builder() + .backend(DownBackend::shared()) + .reliability(ReliabilityConfig { + retry: None, + circuit_breaker: Some(CircuitBreakerConfig { + failure_threshold: 1, + open_timeout: Duration::from_secs(60), + ..CircuitBreakerConfig::default() + }), + }) + .no_l1() + .build() + .expect("client builds"); + + // First call: the transient get counts a breaker failure (threshold 1 → + // the circuit opens) and falls open — the body runs uncached. + let user = circuit_open_op(&cache, 1) + .await + .expect("transient fail-open"); + assert_eq!(user.name, "breaker-open 1"); + + // The circuit is now open: direct calls fail fast without the backend. + let err = cache + .get::("probe") + .await + .expect_err("circuit is open"); + match err { + CachekitError::Backend(be) => assert_eq!(be.kind, BackendErrorKind::CircuitOpen), + other => panic!("expected a circuit-open backend error, got: {other:?}"), + } + + // Macro call against the OPEN circuit: the CircuitOpen branch of the + // fail-open arm must run the body uncached, not surface the error. + let user = circuit_open_op(&cache, 1) + .await + .expect("circuit-open fail-open"); + assert_eq!(user.name, "breaker-open 1"); + assert_eq!( + CIRCUIT_OPEN_RUNS.load(std::sync::atomic::Ordering::SeqCst), + 2, + "both outage classes (transient, circuit-open) fall open to the body" + ); +}