From c96f4bf21df58676d2ebcfb25e5696884b35ecfc Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Sat, 25 Jul 2026 02:40:35 +1000 Subject: [PATCH 1/2] =?UTF-8?q?feat(l1):=20stale-while-revalidate=20?= =?UTF-8?q?=E2=80=94=20serve=20stale=20+=20single-flight=20background=20re?= =?UTF-8?q?fresh=20(LAB-728)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An L1 hit past swr_threshold_ratio × entry TTL (default 0.5, ±10% jitter, mirroring py/ts) but before hard expiry is returned immediately while one background task re-executes the #[cachekit] function and rewrites both layers. Refresh dedup rides the existing cold-miss single_flight() — no SWR-specific dedup path — so N concurrent stale readers cost exactly one origin execution, in-process and (on lock-capable backends) per fleet. Hard-expired entries take the normal blocking miss path. - builder: swr_enabled (default on, sibling parity) + swr_threshold_ratio (validated (0.0, 1.0]); native-only surface — no silent no-op elsewhere - CacheKit is now Clone (all-Arc; clones share L1 + single-flight state) so the 'static refresh task can own the client - interop_get_swr on CacheKit + SecureCache (secure judges staleness on the L1 ciphertext entry — zero-knowledge preserved) - macro: sync fns rejected with a clear decoration-time error; refresh task re-materialises reference args via ToOwned (&str stays &str via as_str) - 30 s backfill cap reconciled: freshness window derives from each entry's own TTL, so backfilled entries background-refresh at ~ratio × 30 s and the refresh restores the full write-path TTL (documented, never silently clamped) - refresh needs a tokio runtime (Handle::try_current); other executors skip the refresh and keep serving until hard expiry — documented --- README.md | 50 +++++- crates/cachekit-macros/src/lib.rs | 187 ++++++++++++++++++-- crates/cachekit/Cargo.toml | 5 +- crates/cachekit/src/client.rs | 208 +++++++++++++++++++++- crates/cachekit/src/l1/mod.rs | 50 +++++- crates/cachekit/src/lib.rs | 49 +++++- crates/cachekit/src/reliability.rs | 11 +- crates/cachekit/tests/swr_tests.rs | 271 +++++++++++++++++++++++++++++ 8 files changed, 798 insertions(+), 33 deletions(-) create mode 100644 crates/cachekit/tests/swr_tests.rs diff --git a/README.md b/README.md index d73f82b..0cee415 100644 --- a/README.md +++ b/README.md @@ -38,7 +38,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) | +| `l1` | ✅ | In-process L1 cache via [moka](https://crates.io/crates/moka), with stale-while-revalidate (native) | | `reliability` | ✅ | Retry with backoff + jitter, circuit breaker, distributed fill locks (native only) | | `redis` | ❌ | Redis backend via [fred](https://crates.io/crates/fred) (native only) | | `memcached` | ❌ | Memcached backend via [rust-memcache](https://crates.io/crates/memcache) (native only) | @@ -301,7 +301,8 @@ When the `l1` feature is enabled (default), CacheKit maintains an in-process [mo ├─────────────────────────────────────────────────────────┤ │ │ │ GET path: │ -│ L1 hit (~50ns) ──► return immediately │ +│ L1 fresh hit (~50ns) ──► return immediately │ +│ L1 stale hit ──► return + background refresh (SWR) │ │ L1 miss ──► L2 backend ──► backfill L1 (30s cap) │ │ │ │ SET path: │ @@ -323,6 +324,48 @@ When the `l1` feature is enabled (default), CacheKit maintains an in-process [mo | **Invalidate-first** | `delete()` evicts L1 before touching L2 | | **Encrypted L1** | `SecureCache` stores ciphertext in L1 (never plaintext) | | **Default capacity** | 1,000 entries (configurable via `.l1_capacity()`) | +| **Stale-while-revalidate** | On by default (native): `#[cachekit]` serves an L1 hit past `swr_threshold_ratio` × entry TTL (default 0.5, ±10% jitter) immediately and refreshes it in the background — see below | + +### Stale-while-revalidate (SWR) + +With SWR (default when `l1` is on, native targets), an L1 entry has two phases +before it disappears: *fresh* until `swr_threshold_ratio` of its TTL has +elapsed, then *stale* until hard expiry. A `#[cachekit]`-wrapped call that +hits a stale entry returns it **immediately** — no caller ever blocks on a +merely-stale value — while exactly one background task re-executes the +function and rewrites both cache layers with a full TTL. Refresh dedup rides +the same single-flight as the cold-miss path (in-process, plus distributed +fill locks on lock-capable backends), so N concurrent stale readers cost one +origin execution — misses are billable; stampedes are not acceptable. A +hard-expired entry always takes the normal blocking miss path: SWR never +serves past hard expiry. + +```rust,ignore +let cache = CacheKit::builder() + .backend(backend) + .swr_threshold_ratio(0.25) // stale after 25% of entry TTL (default 0.5) + // .swr_enabled(false) // restore strict expire-or-serve behaviour + .build()?; +``` + +Semantics mirror cachekit-py (`swr_threshold_ratio` = elapsed-lifetime +fraction; enabled by default) and cachekit-ts (`getWithSwr`). Worth knowing: + +- **The freshness window derives from each entry's own TTL.** A backfilled + entry (L2 hit → L1, 30 s cap) goes stale at ~`ratio × 30 s` — the cap still + bounds staleness of L2-derived data, but SWR replaces its expiry cliff with + a background refresh that restores the full write-path TTL. The configured + ratio is never silently clamped; the window follows the entry. +- **The background refresh needs a tokio runtime** (`Handle::try_current`). + On other executors the stale value is still served and the refresh is + skipped — behaviourally SWR-off, never a panic. +- **Refresh failures are absorbed**: the stale value keeps serving, a later + stale read retries, and once the entry hard-expires the blocking path + surfaces errors normally. +- **Native only**: on wasm32 (`workers` excludes `l1`) and under `unsync` + there is no SWR; the builder knobs don't exist there, so misuse is a + compile error rather than a silent no-op. A sync function under + `#[cachekit]` is likewise a clear compile-time error. --- @@ -336,8 +379,9 @@ With the `reliability` feature (default, native only), the `production`, `encryp | **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 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 | +| **Stale-while-revalidate** | Stale-but-unexpired L1 hits are served immediately while one single-flight-deduplicated background task re-executes the function ([details](#stale-while-revalidate-swr)) | on by default with `l1` (native); threshold 0.5 × entry TTL ±10% jitter | -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. +Retry sits *inside* the breaker (one exhausted retry sequence = one breaker failure), degradation, single-flight, and SWR sit in the `#[cachekit]` macro around the read path — the same composition as the TypeScript SDK's `ReliabilityExecutor` and the Python decorator. ```rust,ignore use std::time::Duration; diff --git a/crates/cachekit-macros/src/lib.rs b/crates/cachekit-macros/src/lib.rs index 83a6378..fc72a19 100644 --- a/crates/cachekit-macros/src/lib.rs +++ b/crates/cachekit-macros/src/lib.rs @@ -203,8 +203,28 @@ 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. /// +/// # Requirements (continued) +/// +/// - The function must be `async` — the generated body awaits the cache +/// read, the cold-miss single-flight, and SWR refresh scheduling. A sync +/// function is a clear compile-time error, never a silent no-op. +/// /// # Reliability behaviour /// +/// - **Stale-while-revalidate** (client SWR enabled — the default with the +/// `l1` feature on native targets): an L1 hit past the client's freshness +/// threshold (`swr_threshold_ratio` × entry TTL, ±10% jitter) but before +/// hard expiry is returned **immediately** — the caller never blocks on a +/// merely-stale entry — while one background task re-executes the function +/// and rewrites both cache layers. Refresh dedup rides +/// [`CacheKit::single_flight`]: N concurrent stale readers trigger exactly +/// one re-execution per process (and, on lock-capable backends, per +/// fleet). Hard-expired entries always take the normal blocking miss path. +/// The refresh task needs a tokio runtime (skipped otherwise — the stale +/// value was already served) and captures arguments by owned copy +/// (`Clone`/`ToOwned` — already required for key derivation); the future +/// must be `Send`. Refresh failures are absorbed: the stale value keeps +/// serving until hard expiry, where the blocking path surfaces errors. /// - **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. @@ -244,14 +264,28 @@ fn expand(args: &MacroArgs, mut func: ItemFn) -> syn::Result { let namespace = args.namespace.as_str(); let operation = args.interop.as_str(); + // The generated body awaits (cache reads, single-flight, SWR refresh + // scheduling); a sync function cannot drive any of it. Fail with a clear + // decoration-time error instead of a cascade of "await is only allowed + // inside async functions" — no silent no-op (the sibling SDK posture). + if func.sig.asyncness.is_none() { + return Err(syn::Error::new_spanned( + func.sig.fn_token, + "#[cachekit] requires an `async fn`: the cache read, cold-miss single-flight, \ + and stale-while-revalidate background refresh all await — annotate the \ + function `async` (a sync function cannot be cached by this macro)", + )); + } + // Extract the Ok type from Result let ok_type = extract_ok_type(&func.sig.output)?; - // Collect non-client parameter idents for cache key derivation. Every - // parameter MUST contribute to the key — a silently skipped parameter - // means two different calls share one cache entry (wrong-data collision), - // so anything we can't name is a compile error, never a skip. - let mut non_client_idents: Vec<&Ident> = Vec::new(); + // Collect non-client parameters (ident + type) for cache key derivation + // and SWR refresh capture. Every parameter MUST contribute to the key — a + // silently skipped parameter means two different calls share one cache + // entry (wrong-data collision), so anything we can't name is a compile + // error, never a skip. + let mut non_client_params: Vec<(&Ident, &Type)> = Vec::new(); for arg in &func.sig.inputs { match arg { syn::FnArg::Receiver(recv) => { @@ -264,7 +298,7 @@ fn expand(args: &MacroArgs, mut func: ItemFn) -> syn::Result { syn::FnArg::Typed(pat) => match pat.pat.as_ref() { syn::Pat::Ident(pi) => { if pi.ident != *client_ident { - non_client_idents.push(&pi.ident); + non_client_params.push((&pi.ident, pat.ty.as_ref())); } } other => { @@ -278,6 +312,7 @@ fn expand(args: &MacroArgs, mut func: ItemFn) -> syn::Result { }, } } + let non_client_idents: Vec<&Ident> = non_client_params.iter().map(|(id, _)| *id).collect(); // Convert each non-client argument to an InteropValue. `.clone()` keeps // the original binding usable by the function body; for Copy types it @@ -291,7 +326,7 @@ fn expand(args: &MacroArgs, mut func: ItemFn) -> syn::Result { }); // Generate cache get/set calls depending on `secure` flag. - let (get_expr, set_expr) = if args.secure { + let (get_expr, get_swr_expr, set_expr) = if args.secure { ( quote! { { @@ -299,6 +334,12 @@ fn expand(args: &MacroArgs, mut func: ItemFn) -> syn::Result { __ck_sec.interop_get::<#ok_type>(&__ck_key).await } }, + quote! { + { + let __ck_sec = #client_ident.secure()?; + __ck_sec.interop_get_swr::<#ok_type>(&__ck_key).await + } + }, quote! { let __ck_sec = #client_ident.secure()?; let _ = __ck_sec.set_with_ttl(&__ck_key, __ck_val, std::time::Duration::from_secs(#ttl_secs)).await; @@ -307,12 +348,47 @@ fn expand(args: &MacroArgs, mut func: ItemFn) -> syn::Result { } else { ( quote! { #client_ident.interop_get::<#ok_type>(&__ck_key).await }, + quote! { #client_ident.interop_get_swr::<#ok_type>(&__ck_key).await }, quote! { let _ = #client_ident.set_with_ttl(&__ck_key, __ck_val, std::time::Duration::from_secs(#ttl_secs)).await; }, ) }; + // SWR refresh capture: the background task is `'static`, so every + // argument is re-materialised as an owned value before the spawn and + // rebound under its original name (and, for references, its original + // shape) inside the task — the function body runs unchanged: + // - `&str` → captured as `String`, rebound via `.as_str()` (exact) + // - other `&T` → captured as `T` via `ToOwned`, rebound as `&owned` + // - owned `T` → captured via `Clone`, rebound by value + let mut swr_captures = TokenStream2::new(); + let mut swr_rebinds = TokenStream2::new(); + for (i, (id, ty)) in non_client_params.iter().enumerate() { + let holder = quote::format_ident!("__ck_swr_arg_{i}"); + match ty { + Type::Reference(r) => { + swr_captures.extend(quote_spanned! {id.span()=> + let #holder = ::std::borrow::ToOwned::to_owned(#id); + }); + let is_str = matches!(r.elem.as_ref(), + Type::Path(p) if p.path.is_ident("str")); + if is_str { + swr_rebinds.extend(quote! { let #id = #holder.as_str(); }); + } else { + swr_rebinds.extend(quote! { let #id = &#holder; }); + } + } + _ => { + swr_captures.extend(quote_spanned! {id.span()=> + #[allow(clippy::clone_on_copy)] + let #holder = #id.clone(); + }); + swr_rebinds.extend(quote! { let #id = #holder; }); + } + } + } + // 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 @@ -355,24 +431,65 @@ fn expand(args: &MacroArgs, mut func: ItemFn) -> syn::Result { // interop/v1 key: {namespace}:{operation}:{args_hash} — the same // key is computable from any SDK (protocol spec/interop-mode.md). // The generated `.clone()` per argument keeps the binding usable - // by the body; for Copy types it is a copy — hence the allow. - #[allow(clippy::clone_on_copy)] + // by the body; for Copy types it is a copy and for `&str` it is + // a deliberate reference copy — hence the allows. + #[allow(clippy::clone_on_copy, noop_method_call)] let __ck_key = cachekit::interop::interop_key( #namespace, #operation, &[#(#interop_args),*], )?; - // Try cache hit. A Serialization error means the stored entry + // Try cache hit, classified against the stale-while-revalidate + // freshness window. A Serialization error means the stored entry // 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. 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) => {} + match #get_swr_expr { + Ok(cachekit::SwrRead::Fresh(__ck_cached)) => return Ok(__ck_cached), + // Stale (past the freshness threshold, before hard expiry): + // serve the cached value immediately and schedule ONE + // background refresh. Dedup rides the same single-flight as + // the cold-miss path: the first refresh task leads and + // re-executes the function; concurrent tasks queue, see the + // (still-present) entry, and exit without computing. Refresh + // failures are deliberately absorbed — the stale value keeps + // being served, a later stale read retries, and hard expiry + // falls through to the blocking path where errors surface. + Ok(cachekit::SwrRead::Stale(__ck_cached)) => { + let __ck_swr_client = ::std::clone::Clone::clone(#client_ident); + let __ck_swr_key = __ck_key.clone(); + #swr_captures + cachekit::__swr_spawn(async move { + let #client_ident = &__ck_swr_client; + let __ck_key = __ck_swr_key; + #swr_rebinds + let _: ::std::result::Result<(), cachekit::error::CachekitError> = async { + let mut __ck_flight = #client_ident.single_flight(&__ck_key).await; + while __ck_flight.wait_for_fill().await { + if matches!(#get_expr, Ok(Some(_))) { + // Another worker is already on it (the + // stale entry is still present, or the + // leader has refreshed) — stand down. + __ck_flight.release().await; + return Ok(()); + } + } + let __ck_result: #ret_ty = (async #original_body).await; + if let Ok(ref __ck_val) = __ck_result { + #set_expr + } + __ck_flight.release().await; + __ck_result.map(|_| ()) + } + .await; + }); + return Ok(__ck_cached); + } + Ok(cachekit::SwrRead::Miss) => {} Err(cachekit::error::CachekitError::Serialization(_)) => {} #fail_open_arm Err(__ck_err) => return Err(__ck_err.into()), @@ -417,7 +534,49 @@ fn expand(args: &MacroArgs, mut func: ItemFn) -> syn::Result { #[cfg(test)] mod tests { - use super::segment_is_valid; + use super::{expand, segment_is_valid, MacroArgs}; + + /// A sync fn must fail at decoration time with a clear, actionable error + /// — never a cascade of "await is only allowed inside async" diagnostics, + /// and never a silent no-op (LAB-728 acceptance criterion). + #[test] + fn sync_fn_is_a_clear_decoration_time_error() { + let args: MacroArgs = syn::parse_quote!( + client = cache, + ttl = 60, + interop = "get_user", + namespace = "users" + ); + let func: syn::ItemFn = syn::parse_quote! { + fn get_user(cache: &CacheKit, id: u64) -> Result { + Ok(format!("User {id}")) + } + }; + let err = expand(&args, func).expect_err("sync fn must be rejected"); + let msg = err.to_string(); + assert!( + msg.contains("requires an `async fn`"), + "error must name the fix, got: {msg}" + ); + } + + /// The same function with `async` added expands cleanly — proving the + /// rejection above is precisely about asyncness. + #[test] + fn async_fn_expands() { + let args: MacroArgs = syn::parse_quote!( + client = cache, + ttl = 60, + interop = "get_user", + namespace = "users" + ); + let func: syn::ItemFn = syn::parse_quote! { + async fn get_user(cache: &CacheKit, id: u64) -> Result { + Ok(format!("User {id}")) + } + }; + expand(&args, func).expect("async fn must expand"); + } /// Mirror of interop.rs's validate_segment acceptance table — the two /// implementations must not drift (see the NOTE on validate_segment). diff --git a/crates/cachekit/Cargo.toml b/crates/cachekit/Cargo.toml index 1e0db69..cbf1b94 100644 --- a/crates/cachekit/Cargo.toml +++ b/crates/cachekit/Cargo.toml @@ -30,7 +30,10 @@ workers = ["dep:worker", "dep:js-sys", "dep:getrandom", "dep:serde_json"] # cachekit-core 0.2 folds aes-gcm into the `encryption` feature automatically on wasm32. # No separate `cachekit-core/wasm` activation needed. encryption = ["cachekit-core/encryption"] -l1 = ["dep:moka"] +# tokio/rt provides Handle::try_current + spawn for the L1 stale-while- +# revalidate background refresh (native only — `workers`+`l1` is a compile +# error, so this never reaches wasm32). +l1 = ["dep:moka", "tokio/rt"] 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). diff --git a/crates/cachekit/src/client.rs b/crates/cachekit/src/client.rs index e856c9a..7c9aa9d 100644 --- a/crates/cachekit/src/client.rs +++ b/crates/cachekit/src/client.rs @@ -20,6 +20,16 @@ pub type SharedBackend = std::sync::Arc; #[cfg(any(target_arch = "wasm32", feature = "unsync"))] pub type SharedBackend = std::rc::Rc; +// ── SharedFlight type alias ────────────────────────────────────────────────── + +/// Reference-counted pointer to the single-flight map, so client clones share +/// fill-dedup state (two clones racing a cold miss must collapse to one fill). +#[cfg(not(any(target_arch = "wasm32", feature = "unsync")))] +type SharedFlight = std::sync::Arc; + +#[cfg(any(target_arch = "wasm32", feature = "unsync"))] +type SharedFlight = std::rc::Rc; + // ── SharedEncryption type alias ────────────────────────────────────────────── /// Reference-counted pointer to the encryption layer. @@ -45,6 +55,15 @@ const MAX_KEY_BYTES: usize = 1024; /// Maximum TTL for L1 entries populated from L2 cache hits. /// Uses a short ceiling to limit staleness when the original TTL is unknown. +/// +/// Reconciliation with stale-while-revalidate: a backfilled entry's SWR +/// freshness window derives from this capped TTL (window = ratio × entry +/// TTL), **not** from the write-path TTL — the cap is the staleness bound +/// for L2-derived data, deliberately kept. SWR removes the cap's expiry +/// cliff instead: past ~ratio × 30 s the entry is served stale while one +/// background refresh re-executes the origin, and that refresh writes both +/// layers with the caller's full TTL. Without SWR the entry simply +/// hard-expires at the cap and the next read blocks on L2, as before. const L1_BACKFILL_TTL_SECS: u64 = 30; fn validate_key(key: &str) -> Result<(), CachekitError> { @@ -69,19 +88,50 @@ fn validate_key(key: &str) -> Result<(), CachekitError> { Ok(()) } +// ── Stale-while-revalidate ─────────────────────────────────────────────────── +// +// SWR needs an L1 to age entries in and a spawnable (`Send`) runtime for the +// background refresh — native, non-`unsync` targets with the `l1` feature. +// Everywhere else the SWR read path degrades to the plain read path and +// `SwrRead::Stale` is never produced. + +/// Outcome of an SWR-aware typed read — see [`CacheKit::interop_get_swr`]. +pub enum SwrRead { + /// Cache hit within the freshness window (or an L2 hit): use directly. + Fresh(T), + /// L1 hit past the freshness threshold but before hard expiry: use the + /// value now, and schedule a background refresh (the `#[cachekit]` macro + /// does this via [`CacheKit::single_flight`] + re-execution). + Stale(T), + /// No usable entry: fall through to a normal blocking miss + fill. + Miss, +} + // ── CacheKit ───────────────────────────────────────────────────────────────── /// Production-ready cache client with optional L1 in-process cache layer. +/// +/// `Clone` is cheap and shares everything: backend, L1 cache, single-flight +/// state, and encryption layer. Clones exist so `'static` background work +/// (e.g. the SWR refresh spawned by `#[cachekit]`) can hold the client +/// without borrowing it. +#[derive(Clone)] pub struct CacheKit { backend: SharedBackend, default_ttl: Duration, namespace: Option, max_payload_bytes: usize, - flight: crate::flight::FlightMap, + flight: SharedFlight, #[cfg(feature = "l1")] l1: Option, + #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] + swr_enabled: bool, + + #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] + swr_threshold_ratio: f64, + #[cfg(feature = "encryption")] encryption: Option, } @@ -256,6 +306,76 @@ impl CacheKit { } } + /// Retrieve and deserialize an interop-mode value with SWR classification. + /// + /// Identical to [`Self::interop_get`] except an L1 hit is classified + /// against the client's stale-while-revalidate freshness window: + /// + /// - [`SwrRead::Fresh`] — L1 hit within `swr_threshold_ratio` of the + /// entry's TTL (±10% jitter), or any L2 hit. Use directly. + /// - [`SwrRead::Stale`] — L1 hit past the threshold but **before hard + /// expiry**: the value is returned without touching the backend or + /// origin, and the caller should schedule exactly one background + /// refresh (dedup via [`Self::single_flight`] — this is what the + /// `#[cachekit]` macro generates). + /// - [`SwrRead::Miss`] — nothing usable anywhere: normal blocking miss. + /// + /// A hard-expired L1 entry is a [`SwrRead::Miss`], never `Stale` — moka + /// drops entries at their TTL, so SWR cannot serve past hard expiry. + /// + /// With SWR disabled ([`CacheKitBuilder::swr_enabled`]`(false)`), without + /// the `l1` feature, on wasm32, or under `unsync`, this behaves exactly + /// like [`Self::interop_get`]: hits are `Fresh`, `Stale` is never + /// produced. + /// + /// # Errors + /// + /// Same as [`Self::interop_get`] (including the namespaced-client + /// rejection). + pub async fn interop_get_swr( + &self, + key: &str, + ) -> Result, CachekitError> { + self.reject_namespaced_interop()?; + match self.get_bytes_swr(key).await? { + SwrRead::Fresh(b) => Ok(SwrRead::Fresh(crate::interop::deserialize(&b)?)), + SwrRead::Stale(b) => Ok(SwrRead::Stale(crate::interop::deserialize(&b)?)), + SwrRead::Miss => Ok(SwrRead::Miss), + } + } + + /// Fetch raw payload bytes with SWR classification: an L1 hit is split + /// into fresh vs stale against the configured freshness window; on L1 + /// miss this defers to [`Self::get_bytes`] (L2 + backfill), whose hit is + /// always fresh. + async fn get_bytes_swr(&self, key: &str) -> Result>, CachekitError> { + #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] + if self.swr_enabled { + if let Some(ref l1) = self.l1 { + let full_key = self.resolve_key(key)?; + match l1.get_with_swr(&full_key, self.swr_threshold_ratio) { + crate::l1::L1SwrRead::Fresh(bytes) => { + self.check_payload_size(bytes.len())?; + return Ok(SwrRead::Fresh(bytes)); + } + crate::l1::L1SwrRead::Stale(bytes) => { + self.check_payload_size(bytes.len())?; + return Ok(SwrRead::Stale(bytes)); + } + // Absent or hard-expired: fall through to the normal + // read path (the redundant L1 re-check there is a cheap + // in-process miss). + crate::l1::L1SwrRead::Miss => {} + } + } + } + + Ok(match self.get_bytes(key).await? { + Some(bytes) => SwrRead::Fresh(bytes), + None => SwrRead::Miss, + }) + } + /// Fetch raw payload bytes for `key` (L1, then L2 with L1 backfill). async fn get_bytes(&self, key: &str) -> Result>, CachekitError> { let full_key = self.resolve_key(key)?; @@ -503,6 +623,32 @@ impl SecureCache<'_> { } } + /// Retrieve, decrypt, and deserialize an interop-mode value with SWR + /// classification. The secure twin of [`CacheKit::interop_get_swr`]: + /// staleness is judged on the L1 **ciphertext** entry (zero-knowledge is + /// preserved — freshness metadata never exposes plaintext), then the + /// value is decrypted and decoded per [`Self::interop_get`]. + /// + /// # Errors + /// + /// Same as [`Self::interop_get`] — the secure path fails closed on every + /// backend and decryption error. + pub async fn interop_get_swr( + &self, + key: &str, + ) -> Result, CachekitError> { + self.client.reject_namespaced_interop()?; + match self.client.get_bytes_swr(key).await? { + SwrRead::Fresh(ct) => Ok(SwrRead::Fresh(crate::interop::deserialize( + &self.encryption.decrypt(&ct, key)?, + )?)), + SwrRead::Stale(ct) => Ok(SwrRead::Stale(crate::interop::deserialize( + &self.encryption.decrypt(&ct, key)?, + )?)), + SwrRead::Miss => Ok(SwrRead::Miss), + } + } + /// Fetch ciphertext (L1, then L2 with L1 backfill) and decrypt it. /// /// Ciphertext retrieval delegates to [`CacheKit::get_bytes`], which returns @@ -543,6 +689,12 @@ pub struct CacheKitBuilder { #[cfg(feature = "l1")] no_l1: bool, + #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] + swr_enabled: Option, + + #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] + swr_threshold_ratio: Option, + #[cfg(feature = "encryption")] encryption: Option, @@ -589,6 +741,39 @@ impl CacheKitBuilder { self } + /// Enable or disable L1 stale-while-revalidate (default: **enabled**, + /// matching the Python and TypeScript SDKs). + /// + /// With SWR on, an L1 hit older than `swr_threshold_ratio` of its TTL is + /// still served immediately, and the `#[cachekit]` macro schedules + /// exactly one background refresh (deduplicated through + /// [`CacheKit::single_flight`], in-process and — on lock-capable + /// backends — across processes). A hard-expired entry is never served: + /// it falls through to a normal blocking miss. + /// + /// Native targets only: this knob does not exist on wasm32, under the + /// `unsync` feature, or without `l1` — calling it there is a compile + /// error rather than a silent no-op. + #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] + pub fn swr_enabled(mut self, enabled: bool) -> Self { + self.swr_enabled = Some(enabled); + self + } + + /// Set the SWR freshness threshold as a fraction of each L1 entry's TTL + /// (default: **0.5**, matching the Python and TypeScript SDKs). + /// + /// An entry is *fresh* until it has lived `ratio × TTL` (±10% jitter to + /// de-synchronise refreshes across processes), then *stale* — served + /// immediately with a background refresh — until hard expiry. Mirrors + /// cachekit-py's `swr_threshold_ratio` semantics (elapsed-lifetime + /// fraction). Must be in `(0.0, 1.0]`; validated at [`Self::build`]. + #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] + pub fn swr_threshold_ratio(mut self, ratio: f64) -> Self { + self.swr_threshold_ratio = Some(ratio); + self + } + // Stubs for when the l1 feature is disabled — still compile cleanly. #[cfg(not(feature = "l1"))] pub fn l1_capacity(self, _capacity: usize) -> Self { @@ -685,6 +870,19 @@ impl CacheKitBuilder { Some(crate::l1::L1Cache::new(capacity)) }; + // SWR defaults mirror the sibling SDKs: enabled, threshold ratio 0.5, + // ratio validated in (0.0, 1.0] exactly like py's L1CacheConfig. + #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] + let swr_threshold_ratio = { + let ratio = self.swr_threshold_ratio.unwrap_or(0.5); + if !(ratio > 0.0 && ratio <= 1.0) { + return Err(CachekitError::Config(format!( + "swr_threshold_ratio must be in (0.0, 1.0]; got {ratio}" + ))); + } + ratio + }; + // 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. @@ -701,11 +899,17 @@ impl CacheKitBuilder { 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(), + flight: SharedFlight::default(), #[cfg(feature = "l1")] l1, + #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] + swr_enabled: self.swr_enabled.unwrap_or(true), + + #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] + swr_threshold_ratio, + #[cfg(feature = "encryption")] encryption: self.encryption, }) diff --git a/crates/cachekit/src/l1/mod.rs b/crates/cachekit/src/l1/mod.rs index b2f4acb..84a8e81 100644 --- a/crates/cachekit/src/l1/mod.rs +++ b/crates/cachekit/src/l1/mod.rs @@ -1,11 +1,12 @@ use moka::sync::Cache; use moka::Expiry; -use std::time::Duration; +use std::time::{Duration, Instant}; #[derive(Clone)] struct L1Entry { data: Vec, ttl: Duration, + created_at: Instant, } struct L1Expiry; @@ -21,9 +22,23 @@ impl Expiry for L1Expiry { } } +/// Outcome of an SWR-aware L1 read — see [`L1Cache::get_with_swr`]. +pub enum L1SwrRead { + /// Entry present and within its freshness window: serve as-is. + Fresh(Vec), + /// Entry present but past the freshness threshold (and before hard + /// expiry): serve it, but the caller should schedule a background + /// refresh. + Stale(Vec), + /// Entry absent or hard-expired: a normal (blocking) miss. + Miss, +} + /// In-process LRU cache with per-entry TTL, backed by [`moka`]. /// -/// Used as the L1 layer in the dual-layer cache architecture. +/// Used as the L1 layer in the dual-layer cache architecture. `Clone` is +/// cheap and shares the underlying store (moka is internally referenced). +#[derive(Clone)] pub struct L1Cache { store: Cache, } @@ -44,6 +59,36 @@ impl L1Cache { self.store.get(key).map(|entry| entry.data.clone()) } + /// Retrieve cached bytes with stale-while-revalidate classification. + /// + /// An entry is *fresh* until it has lived `threshold_ratio` of its own + /// TTL (±10% jitter, so refreshes de-synchronise across processes — + /// same jitter as the Python and TypeScript SDKs), *stale* from then + /// until hard expiry, and a *miss* after that. The freshness window + /// derives from the TTL the entry was **inserted** with: a direct write + /// carries the caller's full TTL, an L2 backfill carries the capped + /// backfill TTL — see `CacheKit`'s L1 documentation. + /// + /// Semantics mirror cachekit-py's `swr_threshold_ratio` (elapsed + /// lifetime > ratio × TTL ⇒ stale). Hard expiry is enforced by moka: + /// an expired entry is never returned, so SWR can never serve past it. + /// + /// This is a pure read — it does not track refresh state. Callers own + /// refresh scheduling and deduplication (the `#[cachekit]` macro uses + /// `CacheKit::single_flight`). + pub fn get_with_swr(&self, key: &str, threshold_ratio: f64) -> L1SwrRead { + let Some(entry) = self.store.get(key) else { + return L1SwrRead::Miss; + }; + let jitter = 0.9 + crate::random_unit() * 0.2; + let threshold = entry.ttl.as_secs_f64() * threshold_ratio * jitter; + if entry.created_at.elapsed().as_secs_f64() > threshold { + L1SwrRead::Stale(entry.data.clone()) + } else { + L1SwrRead::Fresh(entry.data.clone()) + } + } + /// Insert or overwrite an entry with the given TTL. pub fn set(&self, key: &str, value: &[u8], ttl: Duration) { self.store.insert( @@ -51,6 +96,7 @@ impl L1Cache { L1Entry { data: value.to_vec(), ttl, + created_at: Instant::now(), }, ); } diff --git a/crates/cachekit/src/lib.rs b/crates/cachekit/src/lib.rs index a0ab85e..3747be0 100644 --- a/crates/cachekit/src/lib.rs +++ b/crates/cachekit/src/lib.rs @@ -66,7 +66,7 @@ pub mod l1; pub mod reliability; // Re-exports -pub use client::{CacheKit, CacheKitBuilder, SharedBackend}; +pub use client::{CacheKit, CacheKitBuilder, SharedBackend, SwrRead}; pub use config::CachekitConfig; pub use error::{BackendError, BackendErrorKind, CachekitError}; @@ -83,6 +83,53 @@ pub use flight::SingleFlight; #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))] pub use reliability::{CircuitBreakerConfig, ReliabilityConfig, RetryConfig}; +// ── Shared jitter source ───────────────────────────────────────────────────── + +/// 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. Used by retry backoff (`reliability`) and the +/// L1 SWR freshness threshold (`l1`). +#[cfg(any( + feature = "l1", + all(feature = "reliability", not(target_arch = "wasm32")) +))] +pub(crate) fn random_unit() -> f64 { + let bits = uuid::Uuid::new_v4().as_u128() & ((1u128 << 53) - 1); + (bits as f64) / ((1u64 << 53) as f64) +} + +// ── SWR background-refresh spawn (macro plumbing) ─────────────────────────── + +/// Spawn a stale-while-revalidate background refresh onto the ambient tokio +/// runtime. Macro plumbing for `#[cachekit]` — not public API. +/// +/// Without a tokio runtime on the current thread the refresh is skipped: the +/// caller has already been served the stale value, and a later stale read +/// simply retries. Panicking here would turn a cache optimisation into an +/// availability bug on non-tokio executors. +#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] +#[doc(hidden)] +pub fn __swr_spawn(fut: F) +where + F: std::future::Future + Send + 'static, +{ + if let Ok(handle) = tokio::runtime::Handle::try_current() { + drop(handle.spawn(fut)); + } +} + +/// No-op variant: under `unsync`, on wasm32, or without the `l1` feature the +/// client never classifies a hit as stale (`SwrRead::Stale` is unreachable), +/// so the refresh future handed here is dead code by construction. The stub +/// exists so `#[cachekit]`-generated code compiles under every configuration. +#[cfg(not(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32"))))] +#[doc(hidden)] +pub fn __swr_spawn(_fut: F) +where + F: std::future::Future + 'static, +{ +} + /// 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 index e4dd3d9..6d61751 100644 --- a/crates/cachekit/src/reliability.rs +++ b/crates/cachekit/src/reliability.rs @@ -30,6 +30,7 @@ use async_trait::async_trait; use crate::backend::{Backend, HealthStatus, LockableBackend}; use crate::client::SharedBackend; use crate::error::BackendError; +use crate::random_unit; // ── Configuration ──────────────────────────────────────────────────────────── @@ -120,16 +121,6 @@ impl Default for ReliabilityConfig { } } -// ── 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 diff --git a/crates/cachekit/tests/swr_tests.rs b/crates/cachekit/tests/swr_tests.rs new file mode 100644 index 0000000..52ec49b --- /dev/null +++ b/crates/cachekit/tests/swr_tests.rs @@ -0,0 +1,271 @@ +//! Integration tests for L1 stale-while-revalidate (LAB-728). +//! +//! Run with: +//! cargo test --test swr_tests --features macros,l1 +//! +//! SWR is native-only (no `unsync`, no wasm32): the background refresh needs +//! a spawnable `Send` future. These tests use real time — moka's clock is not +//! tokio-mockable — so windows are generous: threshold ~1 s (±10% jitter), +//! hard expiry 4 s, origin delay 400 ms, and every latency assertion leaves +//! two orders of magnitude of slack over an L1 read. + +#![cfg(all(feature = "macros", feature = "l1", not(feature = "unsync")))] + +mod common; + +use std::sync::atomic::{AtomicU32, Ordering}; +use std::time::{Duration, Instant}; + +use common::MockBackend; + +use cachekit::{cachekit, CacheKit, CachekitError}; + +/// Origin latency: long enough that a blocking call is unmistakable next to +/// a served-from-L1 stale read. +const ORIGIN_DELAY: Duration = Duration::from_millis(400); + +fn client(backend: cachekit::SharedBackend) -> CacheKit { + CacheKit::builder() + .backend(backend) + // threshold = 0.25 × 4 s = 1 s (±10%): stale from ≤1.1 s, hard + // expiry at 4 s — a comfortable mid-window probe point at 1.4 s. + .swr_threshold_ratio(0.25) + .build() + .expect("client builds") +} + +// ── serve stale + exactly-one refresh ──────────────────────────────────────── + +static SWR_CALLS: AtomicU32 = AtomicU32::new(0); + +#[cachekit(client = cache, ttl = 4, interop = "swr_probe", namespace = "swrtest")] +async fn swr_probe(cache: &CacheKit, id: u64) -> Result { + let n = SWR_CALLS.fetch_add(1, Ordering::SeqCst) + 1; + tokio::time::sleep(ORIGIN_DELAY).await; + Ok(format!("u{id}-c{n}")) +} + +/// The core LAB-728 contract in one scenario: a read past the SWR threshold +/// (but before hard expiry) returns the stale value without blocking on the +/// origin; N concurrent stale readers trigger exactly ONE background +/// re-execution (single-flight dedup); the next read sees the refreshed value +/// without recomputing. +#[tokio::test] +async fn stale_reads_serve_immediately_and_refresh_exactly_once() { + let cache = client(MockBackend::shared()); + + // Warm: one blocking origin call. + assert_eq!(swr_probe(&cache, 7).await.unwrap(), "u7-c1"); + assert_eq!(SWR_CALLS.load(Ordering::SeqCst), 1); + + // Age into the stale window (threshold ≤1.1 s, hard expiry 4 s). + tokio::time::sleep(Duration::from_millis(1400)).await; + + // 8 concurrent stale readers, each on its own client clone (clones share + // L1 and single-flight state, so this also exercises cross-clone dedup). + let started = Instant::now(); + let mut set = tokio::task::JoinSet::new(); + for _ in 0..8 { + let clone = cache.clone(); + set.spawn(async move { swr_probe(&clone, 7).await }); + } + let mut reads = Vec::new(); + while let Some(joined) = set.join_next().await { + reads.push(joined.expect("reader task panicked")); + } + let elapsed = started.elapsed(); + + // Every reader got the stale value, and none of them awaited the origin + // (which takes 400 ms): the whole batch is a set of L1 reads. + assert_eq!(reads.len(), 8); + for read in reads { + assert_eq!(read.unwrap(), "u7-c1", "stale value is served as-is"); + } + assert!( + elapsed < Duration::from_millis(300), + "stale reads must not block on the origin: took {elapsed:?}" + ); + + // Let the single background refresh finish. + tokio::time::sleep(ORIGIN_DELAY + Duration::from_millis(800)).await; + assert_eq!( + SWR_CALLS.load(Ordering::SeqCst), + 2, + "8 concurrent stale readers must trigger exactly one refresh" + ); + + // The refreshed value is now served fresh — no further origin calls. + assert_eq!(swr_probe(&cache, 7).await.unwrap(), "u7-c2"); + assert_eq!(SWR_CALLS.load(Ordering::SeqCst), 2); +} + +// ── hard expiry falls through to a blocking miss ───────────────────────────── + +static EXPIRY_CALLS: AtomicU32 = AtomicU32::new(0); + +#[cachekit(client = cache, ttl = 2, interop = "swr_expiry", namespace = "swrtest")] +async fn swr_expiry_probe(cache: &CacheKit, id: u64) -> Result { + let n = EXPIRY_CALLS.fetch_add(1, Ordering::SeqCst) + 1; + tokio::time::sleep(ORIGIN_DELAY).await; + Ok(format!("e{id}-c{n}")) +} + +/// SWR never serves past hard expiry: once the entry's TTL elapses, the read +/// is a normal blocking miss + fill, and the caller waits for the origin. +#[tokio::test] +async fn hard_expired_entry_takes_the_blocking_miss_path() { + let (backend, handle) = MockBackend::new_with_handle(); + let cache = client(backend); + + assert_eq!(swr_expiry_probe(&cache, 3).await.unwrap(), "e3-c1"); + + // Cross hard expiry (ttl = 2 s). The mock L2 has no TTL support, so + // clear it manually — this test is about the L1 expiry contract. + tokio::time::sleep(Duration::from_millis(2500)).await; + handle.store.lock().await.clear(); + + let started = Instant::now(); + let value = swr_expiry_probe(&cache, 3).await.unwrap(); + let elapsed = started.elapsed(); + + assert_eq!(value, "e3-c2", "hard-expired read must recompute"); + assert!( + elapsed >= ORIGIN_DELAY, + "hard-expired read must block on the origin: took {elapsed:?}" + ); + assert_eq!(EXPIRY_CALLS.load(Ordering::SeqCst), 2); +} + +// ── fresh reads schedule nothing ───────────────────────────────────────────── + +static FRESH_CALLS: AtomicU32 = AtomicU32::new(0); + +#[cachekit(client = cache, ttl = 60, interop = "swr_fresh", namespace = "swrtest")] +async fn swr_fresh_probe(cache: &CacheKit, id: u64) -> Result { + let n = FRESH_CALLS.fetch_add(1, Ordering::SeqCst) + 1; + Ok(format!("f{id}-c{n}")) +} + +/// A hit inside the freshness window is a plain hit: no background refresh +/// is scheduled, ever. +#[tokio::test] +async fn fresh_read_does_not_schedule_a_refresh() { + let cache = client(MockBackend::shared()); + + assert_eq!(swr_fresh_probe(&cache, 1).await.unwrap(), "f1-c1"); + assert_eq!(swr_fresh_probe(&cache, 1).await.unwrap(), "f1-c1"); + + tokio::time::sleep(Duration::from_millis(300)).await; + assert_eq!( + FRESH_CALLS.load(Ordering::SeqCst), + 1, + "a fresh hit must not spawn background work" + ); +} + +// ── the off switch ─────────────────────────────────────────────────────────── + +static OFF_CALLS: AtomicU32 = AtomicU32::new(0); + +#[cachekit(client = cache, ttl = 2, interop = "swr_off", namespace = "swrtest")] +async fn swr_off_probe(cache: &CacheKit, id: u64) -> Result { + let n = OFF_CALLS.fetch_add(1, Ordering::SeqCst) + 1; + Ok(format!("o{id}-c{n}")) +} + +/// `.swr_enabled(false)` restores the pre-SWR contract exactly: entries are +/// plain hits until hard expiry, and no background refresh ever runs. +#[tokio::test] +async fn disabled_swr_serves_until_hard_expiry_without_refreshing() { + let cache = CacheKit::builder() + .backend(MockBackend::shared()) + .swr_enabled(false) + .build() + .unwrap(); + + assert_eq!(swr_off_probe(&cache, 5).await.unwrap(), "o5-c1"); + + // Deep in what would be the stale window (threshold would be ~1 s). + tokio::time::sleep(Duration::from_millis(1500)).await; + assert_eq!(swr_off_probe(&cache, 5).await.unwrap(), "o5-c1"); + + tokio::time::sleep(Duration::from_millis(300)).await; + assert_eq!( + OFF_CALLS.load(Ordering::SeqCst), + 1, + "SWR off must mean zero background refreshes" + ); +} + +// ── reference-typed arguments survive the 'static refresh capture ──────────── + +static STR_CALLS: AtomicU32 = AtomicU32::new(0); + +#[cachekit(client = cache, ttl = 4, interop = "swr_str", namespace = "swrtest")] +async fn swr_str_probe(cache: &CacheKit, name: &str) -> Result { + let n = STR_CALLS.fetch_add(1, Ordering::SeqCst) + 1; + Ok(format!("{name}-c{n}")) +} + +/// A `&str` argument is re-materialised as an owned `String` for the +/// `'static` refresh task and rebound as `&str` — the background refresh +/// runs the unchanged function body with an identical-typed argument. +#[tokio::test] +async fn str_argument_refreshes_in_the_background() { + let cache = client(MockBackend::shared()); + + assert_eq!(swr_str_probe(&cache, "ada").await.unwrap(), "ada-c1"); + + tokio::time::sleep(Duration::from_millis(1400)).await; + assert_eq!(swr_str_probe(&cache, "ada").await.unwrap(), "ada-c1"); + + tokio::time::sleep(Duration::from_millis(500)).await; + assert_eq!(STR_CALLS.load(Ordering::SeqCst), 2, "one refresh ran"); + assert_eq!(swr_str_probe(&cache, "ada").await.unwrap(), "ada-c2"); +} + +// ── config validation ──────────────────────────────────────────────────────── + +#[tokio::test] +async fn threshold_ratio_is_validated_at_build() { + for bad in [0.0, -0.5, 1.5, f64::NAN] { + let Err(err) = CacheKit::builder() + .backend(MockBackend::shared()) + .swr_threshold_ratio(bad) + .build() + else { + panic!("out-of-range ratio {bad} must fail at build"); + }; + assert!(matches!(err, CachekitError::Config(_)), "got {err:?}"); + } + // Boundary: 1.0 is legal (stale only in the jitter margin before expiry). + assert!( + CacheKit::builder() + .backend(MockBackend::shared()) + .swr_threshold_ratio(1.0) + .build() + .is_ok(), + "ratio 1.0 is legal" + ); +} + +// ── client clones share cache state ────────────────────────────────────────── + +/// `CacheKit` clones share the L1 store (and single-flight map) — the clone +/// handed to a background refresh writes back into the same cache the +/// original reads from. +#[tokio::test] +async fn clones_share_l1_state() { + let (backend, handle) = MockBackend::new_with_handle(); + let cache = CacheKit::builder().backend(backend).build().unwrap(); + let clone = cache.clone(); + + cache.set("shared", &"value".to_owned()).await.unwrap(); + + // Remove the L2 copy: a hit through the clone can only come from the + // shared L1. + handle.store.lock().await.clear(); + + let via_clone: Option = clone.get("shared").await.unwrap(); + assert_eq!(via_clone.as_deref(), Some("value")); +} From 3a8e08d656c5f2abe33b57d6577113d768a08ade Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Sat, 25 Jul 2026 05:04:54 +1000 Subject: [PATCH 2/2] =?UTF-8?q?fix:=20address=20coderabbit=20review=20?= =?UTF-8?q?=E2=80=94=20reject=20&mut=20params,=20derives,=20prelude,=20tes?= =?UTF-8?q?t=20cfg=20+=20timing?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeRabbit-Resolved: cachekit-macros/src/lib.rs:390:Reject &mut parameters CodeRabbit-Resolved: client.rs:134:SWR enums lack common derives CodeRabbit-Resolved: lib.rs:69:SwrRead prelude export CodeRabbit-Resolved: swr_tests.rs:12:cfg gate missing not(wasm32) CodeRabbit-Resolved: swr_tests.rs:198:timing margin hard expiry --- crates/cachekit-macros/src/lib.rs | 36 ++++++++++++++++++++++++++++++ crates/cachekit/src/client.rs | 1 + crates/cachekit/src/l1/mod.rs | 1 + crates/cachekit/src/lib.rs | 1 + crates/cachekit/tests/swr_tests.rs | 12 ++++++++-- 5 files changed, 49 insertions(+), 2 deletions(-) diff --git a/crates/cachekit-macros/src/lib.rs b/crates/cachekit-macros/src/lib.rs index fc72a19..7e5bb24 100644 --- a/crates/cachekit-macros/src/lib.rs +++ b/crates/cachekit-macros/src/lib.rs @@ -368,6 +368,19 @@ fn expand(args: &MacroArgs, mut func: ItemFn) -> syn::Result { let holder = quote::format_ident!("__ck_swr_arg_{i}"); match ty { Type::Reference(r) => { + // The refresh task re-executes the body from an owned + // snapshot rebound as an immutable borrow — mutation through + // a `&mut` argument cannot be preserved, so reject it up + // front instead of surfacing a confusing downstream error. + if r.mutability.is_some() { + return Err(syn::Error::new_spanned( + ty, + "#[cachekit] does not support `&mut` parameters: the SWR background \ + refresh re-executes the function from an owned snapshot, so mutation \ + through the reference cannot be preserved — take the argument by \ + owned value instead", + )); + } swr_captures.extend(quote_spanned! {id.span()=> let #holder = ::std::borrow::ToOwned::to_owned(#id); }); @@ -560,6 +573,29 @@ mod tests { ); } + /// `&mut` parameters are rejected at decoration time: the SWR refresh + /// task rebinds references as immutable borrows of an owned snapshot, + /// so mutation through the argument cannot be modelled. + #[test] + fn mut_ref_param_is_a_clear_decoration_time_error() { + let args: MacroArgs = syn::parse_quote!( + client = cache, + ttl = 60, + interop = "get_user", + namespace = "users" + ); + let func: syn::ItemFn = syn::parse_quote! { + async fn get_user(cache: &CacheKit, id: &mut u64) -> Result { + Ok(format!("User {id}")) + } + }; + let err = expand(&args, func).expect_err("&mut param must be rejected"); + assert!( + err.to_string().contains("`&mut` parameters"), + "error must name the constraint, got: {err}" + ); + } + /// The same function with `async` added expands cleanly — proving the /// rejection above is precisely about asyncness. #[test] diff --git a/crates/cachekit/src/client.rs b/crates/cachekit/src/client.rs index 7c9aa9d..bf2d86e 100644 --- a/crates/cachekit/src/client.rs +++ b/crates/cachekit/src/client.rs @@ -96,6 +96,7 @@ fn validate_key(key: &str) -> Result<(), CachekitError> { // `SwrRead::Stale` is never produced. /// Outcome of an SWR-aware typed read — see [`CacheKit::interop_get_swr`]. +#[derive(Debug, Clone, PartialEq)] pub enum SwrRead { /// Cache hit within the freshness window (or an L2 hit): use directly. Fresh(T), diff --git a/crates/cachekit/src/l1/mod.rs b/crates/cachekit/src/l1/mod.rs index 84a8e81..62923f0 100644 --- a/crates/cachekit/src/l1/mod.rs +++ b/crates/cachekit/src/l1/mod.rs @@ -23,6 +23,7 @@ impl Expiry for L1Expiry { } /// Outcome of an SWR-aware L1 read — see [`L1Cache::get_with_swr`]. +#[derive(Debug, Clone, PartialEq)] pub enum L1SwrRead { /// Entry present and within its freshness window: serve as-is. Fresh(Vec), diff --git a/crates/cachekit/src/lib.rs b/crates/cachekit/src/lib.rs index 3747be0..9c0d290 100644 --- a/crates/cachekit/src/lib.rs +++ b/crates/cachekit/src/lib.rs @@ -134,6 +134,7 @@ where pub mod prelude { pub use crate::{ BackendError, BackendErrorKind, CacheKit, CacheKitBuilder, CachekitConfig, CachekitError, + SwrRead, }; #[cfg(feature = "encryption")] diff --git a/crates/cachekit/tests/swr_tests.rs b/crates/cachekit/tests/swr_tests.rs index 52ec49b..a6aa88b 100644 --- a/crates/cachekit/tests/swr_tests.rs +++ b/crates/cachekit/tests/swr_tests.rs @@ -9,7 +9,12 @@ //! hard expiry 4 s, origin delay 400 ms, and every latency assertion leaves //! two orders of magnitude of slack over an L1 read. -#![cfg(all(feature = "macros", feature = "l1", not(feature = "unsync")))] +#![cfg(all( + feature = "macros", + feature = "l1", + not(feature = "unsync"), + not(target_arch = "wasm32") +))] mod common; @@ -167,7 +172,7 @@ async fn fresh_read_does_not_schedule_a_refresh() { static OFF_CALLS: AtomicU32 = AtomicU32::new(0); -#[cachekit(client = cache, ttl = 2, interop = "swr_off", namespace = "swrtest")] +#[cachekit(client = cache, ttl = 4, interop = "swr_off", namespace = "swrtest")] async fn swr_off_probe(cache: &CacheKit, id: u64) -> Result { let n = OFF_CALLS.fetch_add(1, Ordering::SeqCst) + 1; Ok(format!("o{id}-c{n}")) @@ -180,6 +185,9 @@ async fn disabled_swr_serves_until_hard_expiry_without_refreshing() { let cache = CacheKit::builder() .backend(MockBackend::shared()) .swr_enabled(false) + // Would put the stale window at ~1 s of the 4 s TTL if SWR were on — + // the probe below lands deep inside it, 2.5 s clear of hard expiry. + .swr_threshold_ratio(0.25) .build() .unwrap();