From a44d61dcd8ad9790bfb7719e2e896eb18507d635 Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Sat, 25 Jul 2026 21:28:58 +1000 Subject: [PATCH] fix(l1): guard LAB-728 SWR refresh commits --- README.md | 26 ++- crates/cachekit-macros/src/lib.rs | 33 +++- crates/cachekit/src/client.rs | 227 ++++++++++++++++++++++-- crates/cachekit/src/flight.rs | 80 +++++++++ crates/cachekit/src/l1/mod.rs | 24 ++- crates/cachekit/src/lib.rs | 6 +- crates/cachekit/tests/l1_tests.rs | 18 ++ crates/cachekit/tests/swr_tests.rs | 270 +++++++++++++++++++++++++++++ 8 files changed, 647 insertions(+), 37 deletions(-) diff --git a/README.md b/README.md index 0cee415..efbade9 100644 --- a/README.md +++ b/README.md @@ -333,12 +333,17 @@ 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. +function. If the same-key mutation token is still current, the task rewrites +both cache layers and renews L1 hard expiry with the full write-path TTL; if a +newer `set()` or `delete()` landed through the same client (or one of its +clones) while the origin ran, that explicit mutation wins and the older +refresh result is discarded before it can touch L2. Entry expiry and capacity +eviction do not invalidate the token, so a valid slow refresh can still +repopulate both layers. 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() @@ -356,6 +361,15 @@ fraction; enabled by default) and cachekit-ts (`getWithSwr`). Worth knowing: 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. +- **Refresh completion is version-guarded and same-key ordered.** Each L1 + stale read receives a mutation token. A concurrent explicit write or delete + through that client or a clone invalidates the token, so an older origin + result cannot clobber the new value or resurrect the deletion in either + layer. The guard is intentionally process-local; cross-instance invalidation + is outside SWR's serving-policy scope. +- **Jitter is fixed per entry.** The ±10% threshold jitter is drawn when an + entry is inserted, not on every hit; hot L1 reads do not perform entropy + work and the entry's freshness boundary stays stable for its lifetime. - **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. diff --git a/crates/cachekit-macros/src/lib.rs b/crates/cachekit-macros/src/lib.rs index 7e5bb24..9b0ec74 100644 --- a/crates/cachekit-macros/src/lib.rs +++ b/crates/cachekit-macros/src/lib.rs @@ -215,8 +215,10 @@ fn extract_ok_type(ret: &ReturnType) -> syn::Result { /// `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 +/// merely-stale entry — while one background task re-executes the function. +/// Its version-checked commit rewrites both layers only if no newer set or +/// delete replaced the stale entry; a successful commit renews the L1 hard +/// expiry with the full write-path TTL. 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. @@ -326,7 +328,7 @@ fn expand(args: &MacroArgs, mut func: ItemFn) -> syn::Result { }); // Generate cache get/set calls depending on `secure` flag. - let (get_expr, get_swr_expr, set_expr) = if args.secure { + let (get_expr, get_swr_expr, set_expr, complete_refresh_expr) = if args.secure { ( quote! { { @@ -344,6 +346,15 @@ fn expand(args: &MacroArgs, mut func: ItemFn) -> syn::Result { let __ck_sec = #client_ident.secure()?; let _ = __ck_sec.set_with_ttl(&__ck_key, __ck_val, std::time::Duration::from_secs(#ttl_secs)).await; }, + quote! { + let __ck_sec = #client_ident.secure()?; + let _ = __ck_sec.__complete_swr_refresh( + &__ck_key, + __ck_val, + std::time::Duration::from_secs(#ttl_secs), + __ck_swr_token, + ).await; + }, ) } else { ( @@ -352,6 +363,14 @@ fn expand(args: &MacroArgs, mut func: ItemFn) -> syn::Result { quote! { let _ = #client_ident.set_with_ttl(&__ck_key, __ck_val, std::time::Duration::from_secs(#ttl_secs)).await; }, + quote! { + let _ = #client_ident.__complete_swr_refresh( + &__ck_key, + __ck_val, + std::time::Duration::from_secs(#ttl_secs), + __ck_swr_token, + ).await; + }, ) }; @@ -472,7 +491,9 @@ fn expand(args: &MacroArgs, mut func: ItemFn) -> syn::Result { // 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)) => { + // Completion is version-checked, so a newer explicit set or + // delete always wins over this older origin computation. + Ok(cachekit::SwrRead::Stale(__ck_cached, __ck_swr_token)) => { let __ck_swr_client = ::std::clone::Clone::clone(#client_ident); let __ck_swr_key = __ck_key.clone(); #swr_captures @@ -493,7 +514,9 @@ fn expand(args: &MacroArgs, mut func: ItemFn) -> syn::Result { } let __ck_result: #ret_ty = (async #original_body).await; if let Ok(ref __ck_val) = __ck_result { - #set_expr + // Commit only if no newer set/delete replaced + // the stale entry while the origin ran. + #complete_refresh_expr } __ck_flight.release().await; __ck_result.map(|_| ()) diff --git a/crates/cachekit/src/client.rs b/crates/cachekit/src/client.rs index bf2d86e..3b7383c 100644 --- a/crates/cachekit/src/client.rs +++ b/crates/cachekit/src/client.rs @@ -30,6 +30,12 @@ type SharedFlight = std::sync::Arc; #[cfg(any(target_arch = "wasm32", feature = "unsync"))] type SharedFlight = std::rc::Rc; +/// Separate same-key ordering from single-flight: a refresh holds the flight +/// lock while computing, then takes this lock only for its version-checked +/// commit. Direct reads/writes/deletes take the same mutation lock. +#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] +type SharedMutations = std::sync::Arc; + // ── SharedEncryption type alias ────────────────────────────────────────────── /// Reference-counted pointer to the encryption layer. @@ -61,9 +67,10 @@ const MAX_KEY_BYTES: usize = 1024; /// 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. +/// background refresh re-executes the origin. If no newer mutation replaced +/// the entry, that refresh writes both layers and renews L1 hard expiry 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> { @@ -102,12 +109,48 @@ pub enum SwrRead { 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), + /// does this via [`CacheKit::single_flight`] + re-execution). The token + /// makes refresh completion conditional: a newer set or delete wins. + Stale(T, SwrToken), /// No usable entry: fall through to a normal blocking miss + fill. Miss, } +/// Mutation version captured by an SWR stale read. +/// +/// Pass this back only through the `#[cachekit]`-generated refresh path. It +/// prevents an older background computation from overwriting a newer write +/// or resurrecting a deleted entry on this client or any of its clones. +#[derive(Clone)] +pub struct SwrToken { + #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] + state: std::sync::Arc, + version: u64, +} + +impl std::fmt::Debug for SwrToken { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("SwrToken") + .field("version", &self.version) + .finish_non_exhaustive() + } +} + +impl PartialEq for SwrToken { + fn eq(&self, other: &Self) -> bool { + #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] + { + self.version == other.version && std::sync::Arc::ptr_eq(&self.state, &other.state) + } + #[cfg(not(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32"))))] + { + self.version == other.version + } + } +} + +impl Eq for SwrToken {} + // ── CacheKit ───────────────────────────────────────────────────────────────── /// Production-ready cache client with optional L1 in-process cache layer. @@ -124,6 +167,9 @@ pub struct CacheKit { max_payload_bytes: usize, flight: SharedFlight, + #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] + mutations: SharedMutations, + #[cfg(feature = "l1")] l1: Option, @@ -237,6 +283,56 @@ impl CacheKit { } } + #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] + async fn lock_l1_mutation(&self, full_key: &str) -> Option { + if self.l1.is_some() { + Some(self.mutations.lock(full_key).await) + } else { + None + } + } + + #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] + async fn complete_swr_bytes( + &self, + key: &str, + bytes: Vec, + ttl: Duration, + token: SwrToken, + ) -> Result { + Self::validate_ttl(ttl)?; + self.check_payload_size(bytes.len())?; + let full_key = self.resolve_key(key)?; + let Some(mutation) = self.lock_l1_mutation(&full_key).await else { + return Ok(false); + }; + + // Check before touching L2. The mutation state lives independently of + // the moka entry, so hard expiry or capacity eviction does not look + // like an explicit set/delete and waste a valid origin result. + if !mutation.is_current(&token.state, token.version) { + return Ok(false); + } + let l1_bytes = bytes.clone(); + self.backend.set(&full_key, bytes, Some(ttl)).await?; + self.l1_set(&full_key, &l1_bytes, ttl); + mutation.advance(); + Ok(true) + } + + #[cfg(not(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32"))))] + async fn complete_swr_bytes( + &self, + _key: &str, + _bytes: Vec, + _ttl: Duration, + _token: SwrToken, + ) -> Result { + // `SwrRead::Stale` is unreachable on this build, so there is no + // versioned entry a caller could legitimately complete. + Ok(false) + } + /// Validate TTL is at least 1 second. fn validate_ttl(ttl: Duration) -> Result<(), CachekitError> { if ttl < Duration::from_secs(1) { @@ -318,7 +414,8 @@ impl CacheKit { /// 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). + /// `#[cachekit]` macro generates). The accompanying [`SwrToken`] makes + /// completion conditional, so a newer set/delete always wins. /// - [`SwrRead::Miss`] — nothing usable anywhere: normal blocking miss. /// /// A hard-expired L1 entry is a [`SwrRead::Miss`], never `Stale` — moka @@ -340,7 +437,7 @@ impl CacheKit { 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::Stale(b, token) => Ok(SwrRead::Stale(crate::interop::deserialize(&b)?, token)), SwrRead::Miss => Ok(SwrRead::Miss), } } @@ -359,9 +456,24 @@ impl CacheKit { 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)); + crate::l1::L1SwrRead::Stale(_) => { + // Token capture and the stale snapshot must be atomic + // relative to explicit mutations. Re-check after + // taking the per-key guard; a write may have landed + // between the optimistic classification and here. + let mutation = self.mutations.lock(&full_key).await; + 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())?; + let (state, version) = mutation.snapshot(); + return Ok(SwrRead::Stale(bytes, SwrToken { state, version })); + } + crate::l1::L1SwrRead::Miss => {} + } } // Absent or hard-expired: fall through to the normal // read path (the redundant L1 re-check there is a cheap @@ -388,6 +500,18 @@ impl CacheKit { return Ok(Some(bytes)); } + // Serialize an L2 read/backfill with same-key writes. Re-check L1 + // after taking the lock because another operation may have filled it + // between the optimistic read above and lock acquisition. + #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] + let _mutation = self.lock_l1_mutation(&full_key).await; + + #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] + if let Some(bytes) = self.l1_get(&full_key) { + self.check_payload_size(bytes.len())?; + return Ok(Some(bytes)); + } + // L2 backend let bytes = match self.backend.get(&full_key).await? { Some(b) => b, @@ -424,6 +548,17 @@ impl CacheKit { let full_key = self.resolve_key(key)?; + #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] + let mutation = self.lock_l1_mutation(&full_key).await; + + // Invalidate older refresh tokens before the first backend await, so + // cancellation cannot leave an applied/attempted write vulnerable to + // a stale background result. + #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] + if let Some(ref mutation) = mutation { + mutation.advance(); + } + // Only clone bytes when L1 needs a copy after the backend consumes them. #[cfg(feature = "l1")] { @@ -439,12 +574,36 @@ impl CacheKit { Ok(()) } + /// Commit an unencrypted SWR refresh only if its stale-read token is + /// still current. Macro plumbing; ordinary writes use [`Self::set_with_ttl`]. + #[doc(hidden)] + pub async fn __complete_swr_refresh( + &self, + key: &str, + value: &T, + ttl: Duration, + token: SwrToken, + ) -> Result { + let bytes = serializer::serialize(value)?; + self.complete_swr_bytes(key, bytes, ttl, token).await + } + /// Delete `key` and return `true` if it existed. /// /// Invalidates the L1 entry regardless of the backend result. pub async fn delete(&self, key: &str) -> Result { let full_key = self.resolve_key(key)?; + #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] + let mutation = self.lock_l1_mutation(&full_key).await; + + // Invalidate before L1 changes or backend I/O so task cancellation + // cannot let an older refresh resurrect this key. + #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] + if let Some(ref mutation) = mutation { + mutation.advance(); + } + // Invalidate L1 first so callers never read a stale value even if the // backend delete fails partway through. #[cfg(feature = "l1")] @@ -569,6 +728,16 @@ impl SecureCache<'_> { let full_key = self.client.resolve_key(key)?; + #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] + let mutation = self.client.lock_l1_mutation(&full_key).await; + + // Match the plain write path: invalidate older refresh tokens before + // the first backend await, including on cancellation or failure. + #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] + if let Some(ref mutation) = mutation { + mutation.advance(); + } + // Only clone when L1 needs a copy after the backend consumes the data. #[cfg(feature = "l1")] { @@ -590,6 +759,23 @@ impl SecureCache<'_> { Ok(()) } + /// Commit an encrypted SWR refresh only if its stale-read token is still + /// current. Macro plumbing; ordinary writes use [`Self::set_with_ttl`]. + #[doc(hidden)] + pub async fn __complete_swr_refresh( + &self, + key: &str, + value: &T, + ttl: Duration, + token: SwrToken, + ) -> Result { + let plaintext = serializer::serialize(value)?; + let ciphertext = self.encryption.encrypt(&plaintext, key)?; + self.client + .complete_swr_bytes(key, ciphertext, ttl, token) + .await + } + /// Retrieve, decrypt, and deserialize a value stored under `key`. /// /// Checks L1 (which holds ciphertext) before the backend. @@ -643,9 +829,10 @@ impl SecureCache<'_> { 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::Stale(ct, token) => Ok(SwrRead::Stale( + crate::interop::deserialize(&self.encryption.decrypt(&ct, key)?)?, + token, + )), SwrRead::Miss => Ok(SwrRead::Miss), } } @@ -764,11 +951,12 @@ impl CacheKitBuilder { /// 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`]. + /// An entry is *fresh* until it has lived `ratio × TTL` (±10% jitter, + /// drawn once when the entry is inserted, 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); @@ -902,6 +1090,9 @@ impl CacheKitBuilder { max_payload_bytes: self.max_payload_bytes.unwrap_or(5 * 1024 * 1024), flight: SharedFlight::default(), + #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] + mutations: SharedMutations::default(), + #[cfg(feature = "l1")] l1, diff --git a/crates/cachekit/src/flight.rs b/crates/cachekit/src/flight.rs index 5256829..a40baa1 100644 --- a/crates/cachekit/src/flight.rs +++ b/crates/cachekit/src/flight.rs @@ -40,6 +40,9 @@ use std::collections::HashMap; use std::sync::{Arc, Mutex, PoisonError, Weak}; +#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] +use std::sync::atomic::{AtomicU64, Ordering}; + use crate::client::SharedBackend; /// Above this many live entries, dead map slots are swept opportunistically. @@ -91,6 +94,83 @@ impl FlightMap { } } +// ── SWR mutation ordering ──────────────────────────────────────────────────── + +/// State shared by a stale-read token and same-key mutations while that token +/// is alive. The weak map can discard idle keys without losing correctness: +/// an outstanding token itself keeps this state alive. +#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] +pub(crate) struct MutationState { + lock: Arc>, + version: AtomicU64, +} + +#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] +impl MutationState { + fn new() -> Self { + Self { + lock: Arc::new(tokio::sync::Mutex::new(())), + version: AtomicU64::new(0), + } + } + + pub(crate) fn version(&self) -> u64 { + self.version.load(Ordering::Acquire) + } +} + +/// Per-key mutation versions for conditional SWR commits. Cloned clients +/// share this map. Each entry is weak so unrelated keys do not accumulate. +#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] +#[derive(Default)] +pub(crate) struct MutationMap { + entries: Mutex>>, +} + +#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] +impl MutationMap { + pub(crate) fn state(&self, key: &str) -> Arc { + let mut map = self.entries.lock().unwrap_or_else(PoisonError::into_inner); + if map.len() > SWEEP_THRESHOLD { + map.retain(|_, weak| weak.strong_count() > 0); + } + if let Some(existing) = map.get(key).and_then(Weak::upgrade) { + return existing; + } + let fresh = Arc::new(MutationState::new()); + map.insert(key.to_owned(), Arc::downgrade(&fresh)); + fresh + } + + pub(crate) async fn lock(&self, key: &str) -> MutationGuard { + let state = self.state(key); + let lock = Arc::clone(&state.lock).lock_owned().await; + MutationGuard { state, _lock: lock } + } +} + +/// Holds same-key ordering across an L2 mutation and its L1 update. +#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] +pub(crate) struct MutationGuard { + state: Arc, + _lock: tokio::sync::OwnedMutexGuard<()>, +} + +#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))] +impl MutationGuard { + pub(crate) fn snapshot(&self) -> (Arc, u64) { + (Arc::clone(&self.state), self.state.version()) + } + + pub(crate) fn is_current(&self, state: &Arc, version: u64) -> bool { + Arc::ptr_eq(&self.state, state) && self.state.version() == version + } + + pub(crate) fn advance(&self) { + self.state.version.fetch_add(1, Ordering::Release); + } +} + // ── SingleFlight guard ─────────────────────────────────────────────────────── enum Role { diff --git a/crates/cachekit/src/l1/mod.rs b/crates/cachekit/src/l1/mod.rs index 62923f0..0965b87 100644 --- a/crates/cachekit/src/l1/mod.rs +++ b/crates/cachekit/src/l1/mod.rs @@ -7,6 +7,7 @@ struct L1Entry { data: Vec, ttl: Duration, created_at: Instant, + freshness_jitter: f64, } struct L1Expiry; @@ -20,6 +21,19 @@ impl Expiry for L1Expiry { ) -> Option { Some(value.ttl) } + + fn expire_after_update( + &self, + _key: &String, + value: &L1Entry, + _updated_at: std::time::Instant, + _duration_until_expiry: Option, + ) -> Option { + // A successful SWR refresh is an update of the existing moka entry. + // Returning the replacement value's TTL renews hard expiry from the + // refresh commit instead of retaining the original create deadline. + Some(value.ttl) + } } /// Outcome of an SWR-aware L1 read — see [`L1Cache::get_with_swr`]. @@ -63,9 +77,9 @@ impl L1Cache { /// 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 + /// TTL (±10% jitter, drawn once when the entry is inserted so a hot read + /// never touches the entropy source), *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. @@ -81,8 +95,7 @@ impl L1Cache { 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; + let threshold = entry.ttl.as_secs_f64() * threshold_ratio * entry.freshness_jitter; if entry.created_at.elapsed().as_secs_f64() > threshold { L1SwrRead::Stale(entry.data.clone()) } else { @@ -98,6 +111,7 @@ impl L1Cache { data: value.to_vec(), ttl, created_at: Instant::now(), + freshness_jitter: 0.9 + crate::random_unit() * 0.2, }, ); } diff --git a/crates/cachekit/src/lib.rs b/crates/cachekit/src/lib.rs index 9c0d290..f78572c 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, SwrRead}; +pub use client::{CacheKit, CacheKitBuilder, SharedBackend, SwrRead, SwrToken}; pub use config::CachekitConfig; pub use error::{BackendError, BackendErrorKind, CachekitError}; @@ -88,7 +88,7 @@ pub use reliability::{CircuitBreakerConfig, ReliabilityConfig, RetryConfig}; /// 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`). +/// L1 SWR freshness threshold at entry insertion (`l1`). #[cfg(any( feature = "l1", all(feature = "reliability", not(target_arch = "wasm32")) @@ -134,7 +134,7 @@ where pub mod prelude { pub use crate::{ BackendError, BackendErrorKind, CacheKit, CacheKitBuilder, CachekitConfig, CachekitError, - SwrRead, + SwrRead, SwrToken, }; #[cfg(feature = "encryption")] diff --git a/crates/cachekit/tests/l1_tests.rs b/crates/cachekit/tests/l1_tests.rs index 854ebe5..94dc8a0 100644 --- a/crates/cachekit/tests/l1_tests.rs +++ b/crates/cachekit/tests/l1_tests.rs @@ -37,6 +37,24 @@ mod tests { assert_eq!(cache.get("exp_key"), None); } + #[test] + fn l1_update_renews_hard_expiry_from_the_update() { + let cache = L1Cache::new(16); + cache.set("refresh", b"old", Duration::from_millis(100)); + std::thread::sleep(Duration::from_millis(75)); + + // Updating an existing moka entry used to retain the original + // create deadline. The replacement must receive its full TTL. + cache.set("refresh", b"new", Duration::from_millis(300)); + std::thread::sleep(Duration::from_millis(100)); + cache.run_pending_tasks(); + assert_eq!(cache.get("refresh"), Some(b"new".to_vec())); + + std::thread::sleep(Duration::from_millis(250)); + cache.run_pending_tasks(); + assert_eq!(cache.get("refresh"), None); + } + #[test] fn l1_capacity_eviction() { let cache = L1Cache::new(2); diff --git a/crates/cachekit/tests/swr_tests.rs b/crates/cachekit/tests/swr_tests.rs index a6aa88b..91121b6 100644 --- a/crates/cachekit/tests/swr_tests.rs +++ b/crates/cachekit/tests/swr_tests.rs @@ -23,7 +23,9 @@ use std::time::{Duration, Instant}; use common::MockBackend; +use cachekit::interop::{interop_key, InteropValue}; use cachekit::{cachekit, CacheKit, CachekitError}; +use tokio::sync::Notify; /// Origin latency: long enough that a blocking call is unmistakable next to /// a served-from-L1 stale read. @@ -39,6 +41,10 @@ fn client(backend: cachekit::SharedBackend) -> CacheKit { .expect("client builds") } +fn key(operation: &str, id: u64) -> String { + interop_key("swrtest", operation, &[InteropValue::from(id)]).expect("test key is valid") +} + // ── serve stale + exactly-one refresh ──────────────────────────────────────── static SWR_CALLS: AtomicU32 = AtomicU32::new(0); @@ -104,6 +110,270 @@ async fn stale_reads_serve_immediately_and_refresh_exactly_once() { assert_eq!(SWR_CALLS.load(Ordering::SeqCst), 2); } +// ── version-guarded refresh commit ─────────────────────────────────────────── + +static DELETE_RACE_CALLS: AtomicU32 = AtomicU32::new(0); +static DELETE_REFRESH_STARTED: Notify = Notify::const_new(); +static DELETE_REFRESH_RELEASE: Notify = Notify::const_new(); + +#[cachekit( + client = cache, + ttl = 4, + interop = "swr_delete_race", + namespace = "swrtest" +)] +async fn swr_delete_race(cache: &CacheKit, id: u64) -> Result { + let n = DELETE_RACE_CALLS.fetch_add(1, Ordering::SeqCst) + 1; + if n == 2 { + DELETE_REFRESH_STARTED.notify_one(); + DELETE_REFRESH_RELEASE.notified().await; + } + Ok(format!("d{id}-c{n}")) +} + +/// A delete that lands while the origin is recomputing invalidates the stale +/// token. The refresh must be discarded in both L1 and L2 — never resurrected. +#[tokio::test] +async fn concurrent_delete_wins_over_an_older_refresh() { + let cache = client(MockBackend::shared()); + let storage_key = key("swr_delete_race", 11); + + assert_eq!(swr_delete_race(&cache, 11).await.unwrap(), "d11-c1"); + tokio::time::sleep(Duration::from_millis(1400)).await; + + // Serve stale and wait until the background origin is definitely in + // flight before deleting the same key. + assert_eq!(swr_delete_race(&cache, 11).await.unwrap(), "d11-c1"); + tokio::time::timeout(Duration::from_secs(2), DELETE_REFRESH_STARTED.notified()) + .await + .expect("refresh origin started"); + assert!(cache.delete(&storage_key).await.unwrap()); + + DELETE_REFRESH_RELEASE.notify_one(); + let flight = cache.single_flight(&storage_key).await; + flight.release().await; + + let value: Option = cache.interop_get(&storage_key).await.unwrap(); + assert_eq!(value, None, "completed refresh must not resurrect a delete"); + assert_eq!(DELETE_RACE_CALLS.load(Ordering::SeqCst), 2); +} + +#[cfg(feature = "encryption")] +static SECURE_DELETE_RACE_CALLS: AtomicU32 = AtomicU32::new(0); +#[cfg(feature = "encryption")] +static SECURE_DELETE_REFRESH_STARTED: Notify = Notify::const_new(); +#[cfg(feature = "encryption")] +static SECURE_DELETE_REFRESH_RELEASE: Notify = Notify::const_new(); + +#[cfg(feature = "encryption")] +#[cachekit( + client = cache, + ttl = 4, + interop = "swr_secure_delete_race", + namespace = "swrtest", + secure +)] +async fn swr_secure_delete_race(cache: &CacheKit, id: u64) -> Result { + let n = SECURE_DELETE_RACE_CALLS.fetch_add(1, Ordering::SeqCst) + 1; + if n == 2 { + SECURE_DELETE_REFRESH_STARTED.notify_one(); + SECURE_DELETE_REFRESH_RELEASE.notified().await; + } + Ok(format!("sd{id}-c{n}")) +} + +/// The encrypted macro path uses the same version check before it persists a +/// freshly encrypted value, so ciphertext cannot resurrect a deleted entry. +#[cfg(feature = "encryption")] +#[tokio::test] +async fn concurrent_delete_wins_over_an_older_secure_refresh() { + let cache = CacheKit::builder() + .backend(MockBackend::shared()) + .swr_threshold_ratio(0.25) + .encryption_from_bytes(&[7_u8; 32], "swr-race-tenant") + .expect("encryption configures") + .build() + .expect("client builds"); + let storage_key = key("swr_secure_delete_race", 13); + + assert_eq!(swr_secure_delete_race(&cache, 13).await.unwrap(), "sd13-c1"); + tokio::time::sleep(Duration::from_millis(1400)).await; + + assert_eq!(swr_secure_delete_race(&cache, 13).await.unwrap(), "sd13-c1"); + tokio::time::timeout( + Duration::from_secs(2), + SECURE_DELETE_REFRESH_STARTED.notified(), + ) + .await + .expect("secure refresh origin started"); + assert!(cache.secure().unwrap().delete(&storage_key).await.unwrap()); + + SECURE_DELETE_REFRESH_RELEASE.notify_one(); + let flight = cache.single_flight(&storage_key).await; + flight.release().await; + + let value: Option = cache + .secure() + .unwrap() + .interop_get(&storage_key) + .await + .unwrap(); + assert_eq!(value, None, "secure refresh must not resurrect a delete"); + assert_eq!(SECURE_DELETE_RACE_CALLS.load(Ordering::SeqCst), 2); +} + +static SET_RACE_CALLS: AtomicU32 = AtomicU32::new(0); +static SET_REFRESH_STARTED: Notify = Notify::const_new(); +static SET_REFRESH_RELEASE: Notify = Notify::const_new(); + +#[cachekit( + client = cache, + ttl = 4, + interop = "swr_set_race", + namespace = "swrtest" +)] +async fn swr_set_race(cache: &CacheKit, id: u64) -> Result { + let n = SET_RACE_CALLS.fetch_add(1, Ordering::SeqCst) + 1; + if n == 2 { + SET_REFRESH_STARTED.notify_one(); + SET_REFRESH_RELEASE.notified().await; + } + Ok(format!("s{id}-c{n}")) +} + +/// A newer explicit set wins over an origin result computed from the stale +/// snapshot, even when that refresh completes later. +#[tokio::test] +async fn concurrent_set_wins_over_an_older_refresh() { + let cache = client(MockBackend::shared()); + let storage_key = key("swr_set_race", 12); + + assert_eq!(swr_set_race(&cache, 12).await.unwrap(), "s12-c1"); + tokio::time::sleep(Duration::from_millis(1400)).await; + + assert_eq!(swr_set_race(&cache, 12).await.unwrap(), "s12-c1"); + tokio::time::timeout(Duration::from_secs(2), SET_REFRESH_STARTED.notified()) + .await + .expect("refresh origin started"); + cache + .clone() + .set_with_ttl( + &storage_key, + &"manual-write".to_owned(), + Duration::from_secs(4), + ) + .await + .unwrap(); + + SET_REFRESH_RELEASE.notify_one(); + let flight = cache.single_flight(&storage_key).await; + flight.release().await; + + let value: Option = cache.interop_get(&storage_key).await.unwrap(); + assert_eq!(value.as_deref(), Some("manual-write")); + assert_eq!(SET_RACE_CALLS.load(Ordering::SeqCst), 2); +} + +static RENEW_CALLS: AtomicU32 = AtomicU32::new(0); +static RENEW_REFRESH_STARTED: Notify = Notify::const_new(); + +#[cachekit( + client = cache, + ttl = 2, + interop = "swr_expiry_renewal", + namespace = "swrtest" +)] +async fn swr_expiry_renewal(cache: &CacheKit, id: u64) -> Result { + let n = RENEW_CALLS.fetch_add(1, Ordering::SeqCst) + 1; + if n == 2 { + RENEW_REFRESH_STARTED.notify_one(); + } + Ok(format!("r{id}-c{n}")) +} + +/// A successful refresh renews moka's hard-expiry deadline from the commit, +/// rather than retaining the original entry's create deadline. +#[tokio::test] +async fn refresh_renews_the_full_l1_ttl() { + let (backend, handle) = MockBackend::new_with_handle(); + let cache = client(backend); + let storage_key = key("swr_expiry_renewal", 14); + + assert_eq!(swr_expiry_renewal(&cache, 14).await.unwrap(), "r14-c1"); + tokio::time::sleep(Duration::from_millis(700)).await; + assert_eq!(swr_expiry_renewal(&cache, 14).await.unwrap(), "r14-c1"); + tokio::time::timeout(Duration::from_secs(2), RENEW_REFRESH_STARTED.notified()) + .await + .expect("refresh origin started"); + + // Join the refresh flight, then remove L2 so the later assertion can + // succeed only through the renewed L1 entry. + let flight = cache.single_flight(&storage_key).await; + flight.release().await; + assert_eq!(RENEW_CALLS.load(Ordering::SeqCst), 2); + handle.store.lock().await.clear(); + + // Past the original t+2s deadline, but comfortably before the refresh's + // t+2s deadline (the refresh committed near t+700ms). + tokio::time::sleep(Duration::from_millis(1500)).await; + let value: Option = cache.interop_get(&storage_key).await.unwrap(); + assert_eq!(value.as_deref(), Some("r14-c2")); +} + +static SLOW_RENEW_CALLS: AtomicU32 = AtomicU32::new(0); +static SLOW_RENEW_STARTED: Notify = Notify::const_new(); +static SLOW_RENEW_RELEASE: Notify = Notify::const_new(); + +#[cachekit( + client = cache, + ttl = 2, + interop = "swr_slow_expiry_renewal", + namespace = "swrtest" +)] +async fn swr_slow_expiry_renewal(cache: &CacheKit, id: u64) -> Result { + let n = SLOW_RENEW_CALLS.fetch_add(1, Ordering::SeqCst) + 1; + if n == 2 { + SLOW_RENEW_STARTED.notify_one(); + SLOW_RENEW_RELEASE.notified().await; + } + Ok(format!("sr{id}-c{n}")) +} + +/// Hard expiry controls serving, not mutation identity. A valid origin result +/// that finishes after the stale entry expires must still repopulate L1/L2. +#[tokio::test] +async fn slow_refresh_can_commit_after_the_original_entry_expires() { + let (backend, handle) = MockBackend::new_with_handle(); + let cache = client(backend); + let storage_key = key("swr_slow_expiry_renewal", 15); + + assert_eq!( + swr_slow_expiry_renewal(&cache, 15).await.unwrap(), + "sr15-c1" + ); + tokio::time::sleep(Duration::from_millis(700)).await; + assert_eq!( + swr_slow_expiry_renewal(&cache, 15).await.unwrap(), + "sr15-c1" + ); + tokio::time::timeout(Duration::from_secs(2), SLOW_RENEW_STARTED.notified()) + .await + .expect("slow refresh origin started"); + + // Cross the original t+2s hard-expiry deadline while origin work remains + // in flight, then allow its still-current token to commit. + tokio::time::sleep(Duration::from_millis(1500)).await; + SLOW_RENEW_RELEASE.notify_one(); + let flight = cache.single_flight(&storage_key).await; + flight.release().await; + assert_eq!(SLOW_RENEW_CALLS.load(Ordering::SeqCst), 2); + + // Remove L2: the refreshed value must now be coming from the renewed L1. + handle.store.lock().await.clear(); + let value: Option = cache.interop_get(&storage_key).await.unwrap(); + assert_eq!(value.as_deref(), Some("sr15-c2")); +} + // ── hard expiry falls through to a blocking miss ───────────────────────────── static EXPIRY_CALLS: AtomicU32 = AtomicU32::new(0);