Skip to main content

cachekit/
client.rs

1use std::sync::Arc;
2use std::time::Duration;
3
4use bytes::Bytes;
5
6use serde::{de::DeserializeOwned, Serialize};
7
8use crate::backend::Backend;
9#[cfg(feature = "l1")]
10use crate::backend::{ttl_wire_secs, Freshness};
11use crate::error::CachekitError;
12use crate::metrics::{CacheCounters, L1Stats, ReadOutcome};
13use crate::serializer;
14
15// ── SharedBackend type alias ──────────────────────────────────────────────────
16
17/// Reference-counted pointer to a heap-allocated backend.
18///
19/// On native targets (without `unsync`) we require `Send + Sync` via `Arc`.
20/// On `wasm32` or with the `unsync` feature, `Rc` is used instead — the runtime
21/// is single-threaded so `Send` bounds are unnecessary.
22#[cfg(not(any(target_arch = "wasm32", feature = "unsync")))]
23pub type SharedBackend = std::sync::Arc<dyn Backend>;
24
25/// Reference-counted pointer to a heap-allocated backend (`?Send` variant).
26#[cfg(any(target_arch = "wasm32", feature = "unsync"))]
27pub type SharedBackend = std::rc::Rc<dyn Backend>;
28
29// ── SharedFlight type alias ──────────────────────────────────────────────────
30
31/// Reference-counted pointer to the single-flight map, so client clones share
32/// fill-dedup state (two clones racing a cold miss must collapse to one fill).
33#[cfg(not(any(target_arch = "wasm32", feature = "unsync")))]
34type SharedFlight = std::sync::Arc<crate::flight::FlightMap>;
35
36#[cfg(any(target_arch = "wasm32", feature = "unsync"))]
37type SharedFlight = std::rc::Rc<crate::flight::FlightMap>;
38
39/// Separate same-key ordering from single-flight: a refresh holds the flight
40/// lock while computing, then takes this lock only for its version-checked
41/// commit. Direct reads/writes/deletes take the same mutation lock.
42#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
43type SharedMutations = std::sync::Arc<crate::flight::MutationMap>;
44
45// ── SharedEncryption type alias ──────────────────────────────────────────────
46
47/// Reference-counted pointer to the encryption layer.
48///
49/// On native targets (without `unsync`) `Arc` is used (requires `Sync`).
50/// On `wasm32` or with `unsync`, `Rc` is used — avoids the `!Sync` problem
51/// caused by `Cell<u64>` inside cachekit-core's nonce counter.
52#[cfg(all(
53    feature = "encryption",
54    not(any(target_arch = "wasm32", feature = "unsync"))
55))]
56type SharedEncryption = std::sync::Arc<crate::encryption::EncryptionLayer>;
57
58#[cfg(all(
59    feature = "encryption",
60    any(target_arch = "wasm32", feature = "unsync")
61))]
62type SharedEncryption = std::rc::Rc<crate::encryption::EncryptionLayer>;
63
64// ── Key validation ────────────────────────────────────────────────────────────
65
66const MAX_KEY_BYTES: usize = 1024;
67
68/// Maximum TTL for L1 entries populated from L2 cache hits.
69/// Uses a short ceiling to limit staleness when the original TTL is unknown.
70/// A server `X-CacheKit-Fresh-For` bounds it further, and a stale read or a
71/// zero bound is not backfilled at all — see [`l1_backfill_ttl`].
72///
73/// Reconciliation with stale-while-revalidate: a backfilled entry's SWR
74/// freshness window derives from this capped TTL (window = ratio × entry
75/// TTL), **not** from the write-path TTL — the cap is the staleness bound
76/// for L2-derived data, deliberately kept. SWR removes the cap's expiry
77/// cliff instead: past ~ratio × 30 s the entry is served stale while one
78/// background refresh re-executes the origin. If no newer mutation replaced
79/// the entry, that refresh writes both layers and renews L1 hard expiry with
80/// the caller's full TTL. Without SWR the entry simply hard-expires at the cap
81/// and the next read blocks on L2, as before.
82const L1_BACKFILL_TTL_SECS: u64 = 30;
83
84/// L1 lifetime for a backfill from an L2 hit, or `None` when the read must not
85/// be backfilled (`spec/saas-api.md` § Remaining Freshness). A stale label or a
86/// zero `Fresh-For` forbids it; a present `Fresh-For` is a hard bound under the
87/// 30 s cap, which SWR cannot outlive because moka drops the entry at its TTL.
88/// With no header the cap applies unchanged.
89#[cfg(feature = "l1")]
90fn l1_backfill_ttl(default_ttl: Duration, freshness: Freshness) -> Option<Duration> {
91    let cap = std::cmp::min(default_ttl, Duration::from_secs(L1_BACKFILL_TTL_SECS));
92    let ttl = freshness
93        .fresh_for
94        .map_or(cap, |fresh_for| cap.min(fresh_for));
95    (!freshness.is_stale && !ttl.is_zero()).then_some(ttl)
96}
97
98fn validate_key(key: &str) -> Result<(), CachekitError> {
99    if key.is_empty() {
100        return Err(CachekitError::InvalidKey(
101            "key must not be empty".to_owned(),
102        ));
103    }
104    if key.len() > MAX_KEY_BYTES {
105        return Err(CachekitError::InvalidKey(format!(
106            "key is {} bytes (limit: {MAX_KEY_BYTES})",
107            key.len()
108        )));
109    }
110    for b in key.bytes() {
111        if b < 0x20 || b == 0x7F {
112            return Err(CachekitError::InvalidKey(format!(
113                "key contains illegal control character 0x{b:02X}"
114            )));
115        }
116    }
117    Ok(())
118}
119
120// ── Stale-while-revalidate ───────────────────────────────────────────────────
121//
122// SWR needs an L1 to age entries in and a spawnable (`Send`) runtime for the
123// background refresh — native, non-`unsync` targets with the `l1` feature.
124// Everywhere else the SWR read path degrades to the plain read path and
125// `SwrRead::Stale` is never produced.
126
127/// Outcome of an SWR-aware typed read — see [`CacheKit::interop_get_swr`].
128#[derive(Debug, Clone, PartialEq)]
129pub enum SwrRead<T> {
130    /// Cache hit within the freshness window (or an L2 hit): use directly.
131    Fresh(T),
132    /// L1 hit past the freshness threshold but before hard expiry: use the
133    /// value now, and schedule a background refresh (the `#[cachekit]` macro
134    /// does this via [`CacheKit::single_flight`] + re-execution). The token
135    /// makes refresh completion conditional: a newer set or delete wins.
136    Stale(T, SwrToken),
137    /// No usable entry: fall through to a normal blocking miss + fill.
138    Miss,
139}
140
141/// Mutation version captured by an SWR stale read.
142///
143/// Pass this back only through the `#[cachekit]`-generated refresh path. It
144/// prevents an older background computation from overwriting a newer write
145/// or resurrecting a deleted entry on this client or any of its clones.
146#[derive(Clone)]
147pub struct SwrToken {
148    #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
149    state: std::sync::Arc<crate::flight::MutationState>,
150    version: u64,
151}
152
153impl std::fmt::Debug for SwrToken {
154    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
155        f.debug_struct("SwrToken")
156            .field("version", &self.version)
157            .finish_non_exhaustive()
158    }
159}
160
161impl PartialEq for SwrToken {
162    fn eq(&self, other: &Self) -> bool {
163        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
164        {
165            self.version == other.version && std::sync::Arc::ptr_eq(&self.state, &other.state)
166        }
167        #[cfg(not(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32"))))]
168        {
169            self.version == other.version
170        }
171    }
172}
173
174impl Eq for SwrToken {}
175
176// ── CacheKit ─────────────────────────────────────────────────────────────────
177
178/// Cache client with optional L1 in-process cache layer.
179///
180/// `Clone` is cheap and shares everything: backend, L1 cache, single-flight
181/// state, and encryption layer. Clones exist so `'static` background work
182/// (e.g. the SWR refresh spawned by `#[cachekit]`) can hold the client
183/// without borrowing it.
184///
185/// # Encryption
186///
187/// A client built with encryption configured (by any of the builder's
188/// encryption methods, the `secure` / `secure_from_env` preset, or
189/// [`CacheKit::from_env`] with `CACHEKIT_MASTER_KEY` set) encrypts every
190/// value it writes and decrypts every value it reads — [`Self::get`],
191/// [`Self::interop_get`], [`Self::interop_get_swr`], [`Self::set`] and
192/// [`Self::set_with_ttl`] — and L1 holds ciphertext. Such a client has no
193/// plaintext value path, and the `SecureCache` handle from `secure_cache()`
194/// reads and writes the same entries. [`Self::delete`] and [`Self::exists`]
195/// carry no value and are the same either way.
196#[derive(Clone)]
197pub struct CacheKit {
198    backend: SharedBackend,
199    default_ttl: Duration,
200    namespace: Option<String>,
201    max_payload_bytes: usize,
202    flight: SharedFlight,
203    /// Read counters behind [`Self::stats`]; the backend holds a
204    /// `MetricsProvider` over the same allocation.
205    counters: Arc<CacheCounters>,
206
207    /// The reliability stack's breaker, when configured — kept so
208    /// [`Self::circuit_state`] can read it through the type-erased backend.
209    #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
210    breaker: Option<Arc<crate::reliability::CircuitBreaker>>,
211
212    #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
213    mutations: SharedMutations,
214
215    #[cfg(feature = "l1")]
216    pub(crate) l1: Option<crate::l1::L1Cache>,
217
218    #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
219    pub(crate) swr_enabled: bool,
220
221    #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
222    swr_threshold_ratio: f64,
223
224    #[cfg(feature = "encryption")]
225    encryption: Option<SharedEncryption>,
226}
227
228impl CacheKit {
229    /// Create a new builder.
230    pub fn builder() -> CacheKitBuilder {
231        CacheKitBuilder::default()
232    }
233
234    /// Build from environment variables via [`crate::config::CachekitConfig::from_env`].
235    ///
236    /// Creates a [`crate::backend::cachekitio::CachekitIO`] backend from the
237    /// config. Requires the `cachekitio` feature. With the `encryption`
238    /// feature on and `CACHEKIT_MASTER_KEY` set, the client encrypts every
239    /// value read and write (see [Encryption](CacheKit#encryption)); without
240    /// the feature, a set `CACHEKIT_MASTER_KEY` is a [`CachekitError::Config`].
241    #[cfg(all(feature = "cachekitio", not(target_arch = "wasm32")))]
242    pub fn from_env() -> Result<CacheKitBuilder, CachekitError> {
243        use crate::backend::cachekitio::CachekitIO;
244        use crate::config::CachekitConfig;
245
246        let config = CachekitConfig::from_env()?;
247
248        // The same environment encrypts on a build with the feature, so
249        // ignoring the key here would be a silent plaintext downgrade.
250        #[cfg(not(feature = "encryption"))]
251        if config.master_key.is_some() {
252            return Err(encryption_feature_missing(
253                "CacheKit::from_env() with CACHEKIT_MASTER_KEY set",
254            ));
255        }
256
257        let api_key_z = config
258            .api_key
259            .ok_or_else(|| CachekitError::Config("CACHEKIT_API_KEY is required".to_owned()))?;
260
261        let backend = CachekitIO::builder()
262            .api_key(api_key_z.as_str())
263            .api_url(config.api_url)
264            .build()
265            .map_err(|e| CachekitError::Config(e.to_string()))?;
266
267        #[cfg(not(feature = "unsync"))]
268        let shared: SharedBackend = std::sync::Arc::new(backend);
269        #[cfg(feature = "unsync")]
270        let shared: SharedBackend = std::rc::Rc::new(backend);
271
272        let mut builder = CacheKitBuilder::default()
273            .backend(shared)
274            .default_ttl(config.default_ttl)
275            .max_payload_bytes(config.max_payload_bytes)
276            .l1_capacity(config.l1_capacity);
277
278        if let Some(ns) = config.namespace.clone() {
279            builder = builder.namespace(ns);
280        }
281
282        // Wire up encryption if master key is configured
283        #[cfg(feature = "encryption")]
284        if let Some(ref master_key) = config.master_key {
285            let namespace = config.namespace.as_deref().unwrap_or("default");
286            let previous: Vec<&[u8]> = config
287                .previous_master_keys
288                .iter()
289                .map(|key| key.as_slice())
290                .collect();
291            builder = builder.encryption_from_hex_decoded(master_key, &previous, namespace)?;
292        }
293
294        Ok(builder)
295    }
296
297    // ── Namespacing ───────────────────────────────────────────────────────────
298
299    fn namespaced_key(&self, key: &str) -> String {
300        match &self.namespace {
301            Some(ns) => format!("{ns}:{key}"),
302            None => key.to_owned(),
303        }
304    }
305
306    /// Validate key and return the namespaced version.
307    fn resolve_key(&self, key: &str) -> Result<String, CachekitError> {
308        validate_key(key)?;
309        Ok(self.namespaced_key(key))
310    }
311
312    // ── L1 helpers ───────────────────────────────────────────────────────────
313
314    /// Try L1 cache first. Returns Some(bytes) on hit.
315    #[cfg(feature = "l1")]
316    fn l1_get(&self, full_key: &str) -> Option<Bytes> {
317        self.l1.as_ref().and_then(|l1| l1.get_shared(full_key))
318    }
319
320    /// Populate L1 from an L2 hit, for no longer than [`l1_backfill_ttl`] allows.
321    #[cfg(feature = "l1")]
322    fn l1_backfill(&self, full_key: &str, bytes: Bytes, freshness: Freshness) {
323        if let (Some(l1), Some(ttl)) = (&self.l1, l1_backfill_ttl(self.default_ttl, freshness)) {
324            l1.set_shared(full_key, bytes, ttl);
325        }
326    }
327
328    /// Write-through to L1 after a write the backend accepted. The copy lives
329    /// no longer than the TTL the write sent on the wire ([`ttl_wire_secs`]):
330    /// `spec/saas-api.md` forbids serving it past `min(local_ttl, effective_ttl)`,
331    /// and the wire truncates a fractional TTL from 1 s up (1.5 s goes as 1).
332    #[cfg(feature = "l1")]
333    fn l1_set(&self, full_key: &str, bytes: Bytes, ttl: Duration) {
334        if let Some(ref l1) = self.l1 {
335            let wire = Duration::from_secs(ttl_wire_secs(ttl));
336            l1.set_shared(full_key, bytes, ttl.min(wire));
337        }
338    }
339
340    /// Invalidate L1 entry.
341    #[cfg(feature = "l1")]
342    fn l1_delete(&self, full_key: &str) {
343        if let Some(ref l1) = self.l1 {
344            l1.delete(full_key);
345        }
346    }
347
348    #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
349    async fn lock_l1_mutation(&self, full_key: &str) -> Option<crate::flight::MutationGuard> {
350        if self.l1.is_some() {
351            Some(self.mutations.lock(full_key).await)
352        } else {
353            None
354        }
355    }
356
357    #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
358    async fn complete_swr_bytes(
359        &self,
360        key: &str,
361        bytes: Vec<u8>,
362        ttl: Duration,
363        token: SwrToken,
364    ) -> Result<bool, CachekitError> {
365        Self::validate_ttl(ttl)?;
366        self.check_payload_size(bytes.len())?;
367        let full_key = self.resolve_key(key)?;
368        let Some(mutation) = self.lock_l1_mutation(&full_key).await else {
369            return Ok(false);
370        };
371
372        // Check before touching L2. The mutation state lives independently of
373        // the moka entry, so hard expiry or capacity eviction does not look
374        // like an explicit set/delete and waste a valid origin result.
375        if !mutation.is_current(&token.state, token.version) {
376            return Ok(false);
377        }
378        let l1_bytes = Bytes::copy_from_slice(&bytes);
379        self.backend.set(&full_key, bytes, Some(ttl)).await?;
380        self.l1_set(&full_key, l1_bytes, ttl);
381        mutation.advance();
382        crate::metrics::trace_write(&full_key, ttl);
383        Ok(true)
384    }
385
386    #[cfg(not(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32"))))]
387    async fn complete_swr_bytes(
388        &self,
389        _key: &str,
390        _bytes: Vec<u8>,
391        _ttl: Duration,
392        _token: SwrToken,
393    ) -> Result<bool, CachekitError> {
394        // `SwrRead::Stale` is unreachable on this build, so there is no
395        // versioned entry a caller could legitimately complete.
396        Ok(false)
397    }
398
399    /// Rejects a zero TTL.
400    fn validate_ttl(ttl: Duration) -> Result<(), CachekitError> {
401        if ttl.is_zero() {
402            return Err(CachekitError::Config(
403                "TTL must be greater than zero".to_owned(),
404            ));
405        }
406        Ok(())
407    }
408
409    // ── Public operations ─────────────────────────────────────────────────────
410
411    /// Retrieve and deserialize a value stored under `key`.
412    ///
413    /// Returns `None` if the key does not exist.
414    /// Checks L1 cache before hitting the backend.
415    ///
416    /// # Errors
417    ///
418    /// On a client with encryption configured, returns
419    /// [`CachekitError::Encryption`] — not a miss — if the stored entry fails
420    /// decryption. The key's L1 copy is dropped first and the backend entry
421    /// is left in place, so the next read reaches the backend: a hit once the
422    /// entry is replaced with ciphertext this client can decrypt, a miss once
423    /// it expires or is deleted. A decrypted entry that does not decode as `T`
424    /// is [`CachekitError::Serialization`] naming only `T`, never quoting the
425    /// value.
426    pub async fn get<T: DeserializeOwned>(&self, key: &str) -> Result<Option<T>, CachekitError> {
427        #[cfg(feature = "encryption")]
428        if let Some(secure) = self.encrypting() {
429            return secure.get(key).await;
430        }
431        match self.get_bytes(key).await? {
432            Some(bytes) => Ok(Some(serializer::deserialize(&bytes)?)),
433            None => Ok(None),
434        }
435    }
436
437    /// Retrieve and deserialize an interop-mode value stored under `key`.
438    ///
439    /// Identical to [`Self::get`] except the payload is decoded with
440    /// [`crate::interop::deserialize`], which consumes exactly one MessagePack
441    /// document and rejects trailing bytes (interop/v1 spec MUST). A
442    /// Python-SDK-internal CK frame is rejected with a specific diagnostic
443    /// instead of silently decoding as the integer 67.
444    ///
445    /// Use with keys from [`crate::interop::interop_key`] on a client
446    /// **without** a namespace prefix. There is no interop-specific write
447    /// method: [`Self::set`] already writes plain MessagePack (no ByteStorage
448    /// envelope), which is the interop value format. On a client with
449    /// encryption configured the stored bytes are AES-GCM ciphertext of that
450    /// MessagePack, decrypted here before decoding: in interop mode the
451    /// AES-GCM plaintext is the plain MessagePack value bytes, so the AAD
452    /// (v0x03, `format="msgpack"`, `compressed="False"`) verifies cross-SDK
453    /// unchanged.
454    ///
455    /// # Errors
456    ///
457    /// Returns [`CachekitError::Config`] if the client was built with
458    /// [`CacheKitBuilder::namespace`]: the prefix
459    /// would rewrite the storage key to `{prefix}:{interop_key}`, which no
460    /// other SDK computes — every cross-SDK entry would silently miss. Interop
461    /// keys carry their own namespace segment; failing loudly here beats a
462    /// 100% miss rate that looks like a cold cache. With encryption
463    /// configured, an entry that fails decryption is
464    /// [`CachekitError::Encryption`], with the key's L1 copy dropped as for
465    /// [`Self::get`].
466    pub async fn interop_get<T: DeserializeOwned>(
467        &self,
468        key: &str,
469    ) -> Result<Option<T>, CachekitError> {
470        #[cfg(feature = "encryption")]
471        if let Some(secure) = self.encrypting() {
472            return secure.interop_get(key).await;
473        }
474        self.reject_namespaced_interop()?;
475        match self.get_bytes(key).await? {
476            Some(bytes) => Ok(Some(crate::interop::deserialize(&bytes)?)),
477            None => Ok(None),
478        }
479    }
480
481    /// Interop keys must reach the backend verbatim; a client namespace prefix
482    /// would silently produce storage keys no other SDK computes.
483    fn reject_namespaced_interop(&self) -> Result<(), CachekitError> {
484        match self.namespace {
485            None => Ok(()),
486            Some(_) => Err(CachekitError::Config(
487                "interop reads require a client without a namespace prefix: .namespace() \
488                 would store interop entries under {prefix}:{interop_key}, \
489                 which other SDKs never compute (interop keys already carry a namespace \
490                 segment) — use a dedicated non-namespaced client for interop entries"
491                    .to_owned(),
492            )),
493        }
494    }
495
496    /// Retrieve and deserialize an interop-mode value with SWR classification.
497    ///
498    /// Identical to [`Self::interop_get`] except an L1 hit is classified
499    /// against the client's stale-while-revalidate freshness window:
500    ///
501    /// - [`SwrRead::Fresh`] — L1 hit within `swr_threshold_ratio` of the
502    ///   entry's TTL (±10% jitter), or any L2 hit. Use directly.
503    /// - [`SwrRead::Stale`] — L1 hit past the threshold but **before hard
504    ///   expiry**: the value is returned without touching the backend or
505    ///   origin, and the caller should schedule exactly one background
506    ///   refresh (dedup via [`Self::single_flight`] — this is what the
507    ///   `#[cachekit]` macro generates). The accompanying [`SwrToken`] makes
508    ///   completion conditional, so a newer set/delete always wins.
509    /// - [`SwrRead::Miss`] — nothing usable anywhere: normal blocking miss.
510    ///
511    /// A hard-expired L1 entry is a [`SwrRead::Miss`], never `Stale` — moka
512    /// drops entries at their TTL, so SWR cannot serve past hard expiry.
513    ///
514    /// With SWR disabled ([`CacheKitBuilder::swr_enabled`]`(false)`), without
515    /// the `l1` feature, on wasm32, or under `unsync`, this behaves exactly
516    /// like [`Self::interop_get`]: hits are `Fresh`, `Stale` is never
517    /// produced.
518    ///
519    /// On a client with encryption configured, staleness is judged on the L1
520    /// **ciphertext** entry (freshness metadata never exposes plaintext),
521    /// then the value is decrypted as for [`Self::interop_get`].
522    ///
523    /// # Errors
524    ///
525    /// Same as [`Self::interop_get`] (including the namespaced-client
526    /// rejection).
527    pub async fn interop_get_swr<T: DeserializeOwned>(
528        &self,
529        key: &str,
530    ) -> Result<SwrRead<T>, CachekitError> {
531        #[cfg(feature = "encryption")]
532        if let Some(secure) = self.encrypting() {
533            return secure.interop_get_swr(key).await;
534        }
535        self.reject_namespaced_interop()?;
536        match self.get_bytes_swr(key).await? {
537            SwrRead::Fresh(b) => Ok(SwrRead::Fresh(crate::interop::deserialize(&b)?)),
538            SwrRead::Stale(b, token) => Ok(SwrRead::Stale(crate::interop::deserialize(&b)?, token)),
539            SwrRead::Miss => Ok(SwrRead::Miss),
540        }
541    }
542
543    /// Fetch raw payload bytes with SWR classification: an L1 hit is split
544    /// into fresh vs stale against the configured freshness window; on L1
545    /// miss this defers to [`Self::get_bytes`] (L2 + backfill), whose hit is
546    /// always fresh.
547    async fn get_bytes_swr(&self, key: &str) -> Result<SwrRead<Bytes>, CachekitError> {
548        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
549        if self.swr_enabled {
550            if let Some(ref l1) = self.l1 {
551                let full_key = self.resolve_key(key)?;
552                match l1.get_with_swr_shared(&full_key, self.swr_threshold_ratio) {
553                    crate::l1::SharedSwrRead::Fresh(bytes) => {
554                        self.counters.record(ReadOutcome::L1Hit, &full_key);
555                        self.check_payload_size(bytes.len())?;
556                        return Ok(SwrRead::Fresh(bytes));
557                    }
558                    crate::l1::SharedSwrRead::Stale(_) => {
559                        // Token capture and the stale snapshot must be atomic
560                        // relative to explicit mutations. Re-check after
561                        // taking the per-key guard; a write may have landed
562                        // between the optimistic classification and here.
563                        let mutation = self.mutations.lock(&full_key).await;
564                        match l1.get_with_swr_shared(&full_key, self.swr_threshold_ratio) {
565                            crate::l1::SharedSwrRead::Fresh(bytes) => {
566                                self.counters.record(ReadOutcome::L1Hit, &full_key);
567                                self.check_payload_size(bytes.len())?;
568                                return Ok(SwrRead::Fresh(bytes));
569                            }
570                            crate::l1::SharedSwrRead::Stale(bytes) => {
571                                self.counters.record(ReadOutcome::L1Stale, &full_key);
572                                self.check_payload_size(bytes.len())?;
573                                let (state, version) = mutation.snapshot();
574                                return Ok(SwrRead::Stale(bytes, SwrToken { state, version }));
575                            }
576                            crate::l1::SharedSwrRead::Miss => {}
577                        }
578                    }
579                    // Absent or hard-expired: fall through to the normal
580                    // read path (the redundant L1 re-check there is a cheap
581                    // in-process miss).
582                    crate::l1::SharedSwrRead::Miss => {}
583                }
584            }
585        }
586
587        Ok(match self.get_bytes(key).await? {
588            Some(bytes) => SwrRead::Fresh(bytes),
589            None => SwrRead::Miss,
590        })
591    }
592
593    /// Fetch raw payload bytes for `key` (L1, then L2 with L1 backfill).
594    async fn get_bytes(&self, key: &str) -> Result<Option<Bytes>, CachekitError> {
595        let full_key = self.resolve_key(key)?;
596
597        // L1 hit
598        #[cfg(feature = "l1")]
599        if let Some(bytes) = self.l1_get(&full_key) {
600            self.counters.record(ReadOutcome::L1Hit, &full_key);
601            self.check_payload_size(bytes.len())?;
602            return Ok(Some(bytes));
603        }
604
605        // Serialize an L2 read/backfill with same-key writes. Re-check L1
606        // after taking the lock because another operation may have filled it
607        // between the optimistic read above and lock acquisition.
608        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
609        let _mutation = self.lock_l1_mutation(&full_key).await;
610
611        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
612        if let Some(bytes) = self.l1_get(&full_key) {
613            self.counters.record(ReadOutcome::L1Hit, &full_key);
614            self.check_payload_size(bytes.len())?;
615            return Ok(Some(bytes));
616        }
617
618        // L2 backend
619        let (bytes, freshness) = match self.backend.get_with_freshness(&full_key).await? {
620            Some(hit) => {
621                self.counters.record(ReadOutcome::L2Hit, &full_key);
622                hit
623            }
624            None => {
625                self.counters.record(ReadOutcome::Miss, &full_key);
626                return Ok(None);
627            }
628        };
629
630        self.check_payload_size(bytes.len())?;
631        // Takes over the Vec's allocation: no copy, and L1 shares it below.
632        let bytes = Bytes::from(bytes);
633
634        // Populate L1 on L2 hit, bounded by the cap and the server's freshness
635        #[cfg(feature = "l1")]
636        self.l1_backfill(&full_key, bytes.clone(), freshness);
637        #[cfg(not(feature = "l1"))]
638        let _ = freshness;
639
640        Ok(Some(bytes))
641    }
642
643    /// Serialize and store `value` under `key` using the client's default TTL.
644    ///
645    /// Encrypts first on a client with encryption configured — see
646    /// [`Self::set_with_ttl`].
647    pub async fn set<T: Serialize>(&self, key: &str, value: &T) -> Result<(), CachekitError> {
648        self.set_with_ttl(key, value, self.default_ttl).await
649    }
650
651    /// Serialize and store `value` under `key` with an explicit `ttl`.
652    ///
653    /// On a client with encryption configured the value is encrypted first:
654    /// the backend and L1 receive only the AES-256-GCM ciphertext, and the
655    /// payload limit applies to the ciphertext.
656    ///
657    /// A positive sub-second `ttl` is sent to the backend as 1 second.
658    ///
659    /// Returns [`CachekitError::Config`] if `ttl` is zero.
660    pub async fn set_with_ttl<T: Serialize>(
661        &self,
662        key: &str,
663        value: &T,
664        ttl: Duration,
665    ) -> Result<(), CachekitError> {
666        #[cfg(feature = "encryption")]
667        if let Some(secure) = self.encrypting() {
668            return secure.set_with_ttl(key, value, ttl).await;
669        }
670        Self::validate_ttl(ttl)?;
671
672        let bytes = serializer::serialize(value)?;
673        self.check_payload_size(bytes.len())?;
674
675        let full_key = self.resolve_key(key)?;
676
677        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
678        let mutation = self.lock_l1_mutation(&full_key).await;
679
680        // Invalidate older refresh tokens before the first backend await, so
681        // cancellation cannot leave an applied/attempted write vulnerable to
682        // a stale background result.
683        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
684        if let Some(ref mutation) = mutation {
685            mutation.advance();
686        }
687
688        // L1 needs its own copy: the backend consumes the Vec.
689        #[cfg(feature = "l1")]
690        {
691            let l1_bytes = Bytes::copy_from_slice(&bytes);
692            self.backend.set(&full_key, bytes, Some(ttl)).await?;
693            self.l1_set(&full_key, l1_bytes, ttl);
694        }
695        #[cfg(not(feature = "l1"))]
696        {
697            self.backend.set(&full_key, bytes, Some(ttl)).await?;
698        }
699
700        crate::metrics::trace_write(&full_key, ttl);
701        Ok(())
702    }
703
704    /// Commit an SWR refresh only if its stale-read token is still current,
705    /// encrypted when the client has encryption configured. Macro plumbing;
706    /// ordinary writes use [`Self::set_with_ttl`].
707    #[doc(hidden)]
708    pub async fn __complete_swr_refresh<T: Serialize>(
709        &self,
710        key: &str,
711        value: &T,
712        ttl: Duration,
713        token: SwrToken,
714    ) -> Result<bool, CachekitError> {
715        #[cfg(feature = "encryption")]
716        if let Some(secure) = self.encrypting() {
717            return secure.__complete_swr_refresh(key, value, ttl, token).await;
718        }
719        let bytes = serializer::serialize(value)?;
720        self.complete_swr_bytes(key, bytes, ttl, token).await
721    }
722
723    /// Delete `key` and return `true` if it existed.
724    ///
725    /// `CachekitIO` and `WorkersCachekitIO` return `true` on every successful
726    /// delete, whether or not the key existed: the server does not report
727    /// existence on `DELETE`.
728    ///
729    /// Invalidates the L1 entry regardless of the backend result.
730    pub async fn delete(&self, key: &str) -> Result<bool, CachekitError> {
731        let full_key = self.resolve_key(key)?;
732
733        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
734        let mutation = self.lock_l1_mutation(&full_key).await;
735
736        // Invalidate before L1 changes or backend I/O so task cancellation
737        // cannot let an older refresh resurrect this key.
738        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
739        if let Some(ref mutation) = mutation {
740            mutation.advance();
741        }
742
743        // Invalidate L1 first so callers never read a stale value even if the
744        // backend delete fails partway through.
745        #[cfg(feature = "l1")]
746        self.l1_delete(&full_key);
747
748        let existed = self.backend.delete(&full_key).await?;
749        crate::metrics::trace_delete(&full_key, existed);
750        Ok(existed)
751    }
752
753    /// Return `true` if `key` exists without fetching the value.
754    ///
755    /// A live L1 entry answers without a backend call. That check does not
756    /// count as a use of the entry for L1 eviction; only reads do.
757    pub async fn exists(&self, key: &str) -> Result<bool, CachekitError> {
758        let full_key = self.resolve_key(key)?;
759
760        // Check L1 first — avoids a network round-trip for warm entries.
761        #[cfg(feature = "l1")]
762        if self.l1.as_ref().is_some_and(|l1| l1.contains(&full_key)) {
763            return Ok(true);
764        }
765
766        Ok(self.backend.exists(&full_key).await?)
767    }
768
769    // ── Observability ─────────────────────────────────────────────────────────
770
771    /// Live hit/miss counters for this client and all of its clones.
772    ///
773    /// Every value read — `get`, `interop_get`, their SWR variants, and the
774    /// [`SecureCache`] equivalents — is counted by where it was served from.
775    /// `exists` and reads that fail with a backend error are not counted. The
776    /// same counters feed the cachekit.io backends' `X-CacheKit-*` telemetry
777    /// headers, so for a backend used by this client alone, what you see here
778    /// is what the SaaS sees (a backend shared by several clients reports the
779    /// first client built over it — see [`Backend::attach_metrics`]).
780    ///
781    /// ```no_run
782    /// # fn example(cache: &cachekit::CacheKit) {
783    /// let stats = cache.stats();
784    /// let reads = stats.l1_hits + stats.l2_hits + stats.misses;
785    /// eprintln!("L1 hit rate {:.1}% over {reads} reads", stats.l1_hit_rate() * 100.0);
786    /// # }
787    /// ```
788    pub fn stats(&self) -> L1Stats {
789        self.counters.snapshot()
790    }
791
792    /// Number of entries currently held in L1, or `None` when L1 is disabled
793    /// (or the `l1` feature is off).
794    ///
795    /// Exact at the time of the call: moka's pending expirations and
796    /// evictions are applied first, which makes this a housekeeping call, not
797    /// a free counter read — poll it, don't put it on a hot path.
798    pub fn l1_entry_count(&self) -> Option<u64> {
799        #[cfg(feature = "l1")]
800        let count = self.l1.as_ref().map(crate::l1::L1Cache::entry_count);
801        #[cfg(not(feature = "l1"))]
802        let count = None;
803        count
804    }
805
806    /// Current circuit-breaker state, or `None` when this client has no
807    /// breaker (reliability disabled, or configured with
808    /// `circuit_breaker: None`).
809    ///
810    /// Reads the live state: an open circuit whose `open_timeout` has elapsed
811    /// reports [`CircuitState::HalfOpen`](crate::CircuitState::HalfOpen). With
812    /// the `tracing` feature every transition is also emitted on the
813    /// `cachekit::reliability` target — `warn` when the circuit opens, `info`
814    /// for half-open and closed.
815    #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
816    pub fn circuit_state(&self) -> Option<crate::reliability::CircuitState> {
817        self.breaker.as_ref().map(|breaker| breaker.state())
818    }
819
820    // ── Single-flight ─────────────────────────────────────────────────────────
821
822    /// Begin a cold-miss single-flight for `key` (see [`crate::flight`]).
823    ///
824    /// Call after a cache miss, before computing the value. Concurrent
825    /// in-process fills of the same key are collapsed to one; with the
826    /// `reliability` feature and a lock-capable backend (CachekitIO, Redis),
827    /// fills are also suppressed across processes via a distributed fill
828    /// lock. The `#[cachekit]` macro does this automatically.
829    ///
830    /// The key is namespaced like every cache operation but not validated —
831    /// this call is infallible; an invalid key simply fails later at the
832    /// actual cache operation.
833    pub async fn single_flight(&self, key: &str) -> crate::flight::SingleFlight {
834        let full_key = self.namespaced_key(key);
835        crate::flight::SingleFlight::acquire(&self.flight, &self.backend, &full_key).await
836    }
837
838    // ── Secure cache ─────────────────────────────────────────────────────────
839
840    /// Return a [`SecureCache`] handle that encrypts all values before storage.
841    ///
842    /// L1 stores **ciphertext** (not plaintext) to preserve the zero-knowledge
843    /// property across all cache layers.
844    ///
845    /// The handle encrypts exactly what this client's own value methods
846    /// already encrypt (see [Encryption](CacheKit#encryption)), and the two
847    /// read each other's entries. What it adds is the guarantee: code that
848    /// must never run unencrypted asks for the handle, which fails on a
849    /// client without encryption, and it carries the
850    /// [`previous_key_hits`](SecureCache::previous_key_hits) rotation signal.
851    ///
852    /// # Errors
853    /// Returns `CachekitError::Config` if no encryption layer is configured.
854    /// Configure encryption via [`CacheKitBuilder::encryption`] /
855    /// [`CacheKitBuilder::encryption_from_bytes`], or start from the
856    /// `CacheKit::secure(url, master_key_hex)` /
857    /// `CacheKit::secure_from_env(url)` preset (`redis` feature).
858    #[cfg(feature = "encryption")]
859    pub fn secure_cache(&self) -> Result<SecureCache<'_>, CachekitError> {
860        self.encrypting().ok_or_else(|| {
861            CachekitError::Config(
862                "encryption not configured: use the CacheKit::secure(url, master_key_hex) \
863                 preset, CacheKit::secure_from_env(url) to read the hex key from \
864                 CACHEKIT_MASTER_KEY, or call .encryption() on the builder"
865                    .to_owned(),
866            )
867        })
868    }
869
870    /// The encrypting handle when encryption is configured. Every value
871    /// method routes through it first, so a configured key leaves no
872    /// plaintext path (protocol `spec/intent-presets.md` § Encryption
873    /// Activation, rule 1).
874    #[cfg(feature = "encryption")]
875    fn encrypting(&self) -> Option<SecureCache<'_>> {
876        self.encryption.as_deref().map(|encryption| SecureCache {
877            client: self,
878            encryption,
879        })
880    }
881
882    // ── Private helpers ───────────────────────────────────────────────────────
883
884    fn check_payload_size(&self, size: usize) -> Result<(), CachekitError> {
885        if size > self.max_payload_bytes {
886            return Err(CachekitError::PayloadTooLarge {
887                size,
888                limit: self.max_payload_bytes,
889            });
890        }
891        Ok(())
892    }
893}
894
895// ── SecureCache ──────────────────────────────────────────────────────────────
896
897/// Encrypted cache handle returned by [`CacheKit::secure_cache()`].
898///
899/// All values are serialized, then encrypted with AES-256-GCM before storage.
900/// L1 stores ciphertext to maintain zero-knowledge guarantees. The client's
901/// own value methods route through these same methods once encryption is
902/// configured, so the handle and the client read each other's entries.
903#[cfg(feature = "encryption")]
904pub struct SecureCache<'a> {
905    client: &'a CacheKit,
906    encryption: &'a crate::encryption::EncryptionLayer,
907}
908
909#[cfg(feature = "encryption")]
910impl std::fmt::Debug for SecureCache<'_> {
911    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
912        f.debug_struct("SecureCache")
913            .field("tenant_id", &self.encryption.tenant_id())
914            .finish()
915    }
916}
917
918#[cfg(feature = "encryption")]
919impl SecureCache<'_> {
920    /// Rotation drain signal; see
921    /// [`EncryptionLayer::previous_key_hits`](crate::EncryptionLayer::previous_key_hits).
922    pub fn previous_key_hits(&self) -> Vec<u64> {
923        self.encryption.previous_key_hits()
924    }
925
926    /// Whether AES is hardware-accelerated on this host (informational); see
927    /// [`EncryptionLayer::hardware_acceleration_enabled`](crate::EncryptionLayer::hardware_acceleration_enabled).
928    pub fn hardware_acceleration_enabled(&self) -> bool {
929        self.encryption.hardware_acceleration_enabled()
930    }
931
932    /// Encrypt and store `value` under `key` using the client's default TTL.
933    pub async fn set<T: Serialize>(&self, key: &str, value: &T) -> Result<(), CachekitError> {
934        self.set_with_ttl(key, value, self.client.default_ttl).await
935    }
936
937    /// Encrypt and store `value` under `key` with an explicit `ttl`.
938    pub async fn set_with_ttl<T: Serialize>(
939        &self,
940        key: &str,
941        value: &T,
942        ttl: Duration,
943    ) -> Result<(), CachekitError> {
944        CacheKit::validate_ttl(ttl)?;
945        let (full_key, ciphertext) = self.seal(key, value)?;
946        // Size-check what is actually persisted (nonce + ciphertext + tag).
947        // The get paths check the stored ciphertext length, so checking the
948        // plaintext here would let a value within 28 bytes of the limit write
949        // successfully and then fail EVERY subsequent read with PayloadTooLarge.
950        self.client.check_payload_size(ciphertext.len())?;
951
952        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
953        let mutation = self.client.lock_l1_mutation(&full_key).await;
954
955        // Match the plain write path: invalidate older refresh tokens before
956        // the first backend await, including on cancellation or failure.
957        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
958        if let Some(ref mutation) = mutation {
959            mutation.advance();
960        }
961
962        // L1 needs its own copy: the backend consumes the Vec.
963        #[cfg(feature = "l1")]
964        {
965            let l1_bytes = Bytes::copy_from_slice(&ciphertext);
966            self.client
967                .backend
968                .set(&full_key, ciphertext, Some(ttl))
969                .await?;
970            self.client.l1_set(&full_key, l1_bytes, ttl);
971        }
972        #[cfg(not(feature = "l1"))]
973        {
974            self.client
975                .backend
976                .set(&full_key, ciphertext, Some(ttl))
977                .await?;
978        }
979
980        crate::metrics::trace_write(&full_key, ttl);
981        Ok(())
982    }
983
984    /// Commit an encrypted SWR refresh only if its stale-read token is still
985    /// current. Macro plumbing; ordinary writes use [`Self::set_with_ttl`].
986    #[doc(hidden)]
987    pub async fn __complete_swr_refresh<T: Serialize>(
988        &self,
989        key: &str,
990        value: &T,
991        ttl: Duration,
992        token: SwrToken,
993    ) -> Result<bool, CachekitError> {
994        let (_, ciphertext) = self.seal(key, value)?;
995        self.client
996            .complete_swr_bytes(key, ciphertext, ttl, token)
997            .await
998    }
999
1000    /// Serialize and encrypt `value` for `key`, returning the key this client
1001    /// passes to its backend (`.namespace()` prefix included) with the
1002    /// ciphertext. Every write encrypts here, so the AAD binds that key and an
1003    /// entry copied to another namespace fails to decrypt.
1004    fn seal<T: Serialize>(&self, key: &str, value: &T) -> Result<(String, Vec<u8>), CachekitError> {
1005        let full_key = self.client.resolve_key(key)?;
1006        let plaintext = serializer::serialize(value)?;
1007        let ciphertext = self.encryption.encrypt(&plaintext, &full_key)?;
1008        Ok((full_key, ciphertext))
1009    }
1010
1011    /// Retrieve, decrypt, and deserialize a value stored under `key`.
1012    ///
1013    /// Checks L1 (which holds ciphertext) before the backend.
1014    ///
1015    /// # Errors
1016    ///
1017    /// Returns [`CachekitError::Encryption`] if the stored entry fails
1018    /// decryption. The key's L1 copy is dropped first and the backend entry
1019    /// is left in place, so the next read reaches the backend: a hit once the
1020    /// entry is replaced with ciphertext this client can decrypt, a miss once
1021    /// it expires or is deleted. A decrypted entry that does not decode as `T`
1022    /// is [`CachekitError::Serialization`] naming only `T`: the decoder's own
1023    /// message would quote the plaintext.
1024    pub async fn get<T: DeserializeOwned>(&self, key: &str) -> Result<Option<T>, CachekitError> {
1025        match self.get_plaintext(key).await? {
1026            Some(plaintext) => Ok(Some(redact_decode(serializer::deserialize(&plaintext))?)),
1027            None => Ok(None),
1028        }
1029    }
1030
1031    /// Retrieve, decrypt, and deserialize an interop-mode value stored under `key`.
1032    ///
1033    /// Identical to [`Self::get`] except the decrypted plaintext is decoded
1034    /// with [`crate::interop::deserialize`] — exactly one MessagePack document,
1035    /// trailing bytes rejected (interop/v1 spec MUST). In interop mode the
1036    /// AES-GCM plaintext is the plain MessagePack value bytes, so the AAD
1037    /// (v0x03, `format="msgpack"`, `compressed="False"`) verifies cross-SDK
1038    /// unchanged.
1039    ///
1040    /// # Errors
1041    ///
1042    /// Returns [`CachekitError::Config`] on a namespace-prefixed client — see
1043    /// [`CacheKit::interop_get`] — and [`CachekitError::Encryption`] if the
1044    /// stored entry fails decryption, with the key's L1 copy dropped as for
1045    /// [`Self::get`].
1046    pub async fn interop_get<T: DeserializeOwned>(
1047        &self,
1048        key: &str,
1049    ) -> Result<Option<T>, CachekitError> {
1050        self.client.reject_namespaced_interop()?;
1051        match self.get_plaintext(key).await? {
1052            Some(plaintext) => Ok(Some(redact_decode(crate::interop::deserialize(
1053                &plaintext,
1054            ))?)),
1055            None => Ok(None),
1056        }
1057    }
1058
1059    /// Retrieve, decrypt, and deserialize an interop-mode value with SWR
1060    /// classification. The secure twin of [`CacheKit::interop_get_swr`]:
1061    /// staleness is judged on the L1 **ciphertext** entry (zero-knowledge is
1062    /// preserved — freshness metadata never exposes plaintext), then the
1063    /// value is decrypted and decoded per [`Self::interop_get`].
1064    ///
1065    /// # Errors
1066    ///
1067    /// Same as [`Self::interop_get`] — the secure path fails closed on every
1068    /// backend and decryption error.
1069    pub async fn interop_get_swr<T: DeserializeOwned>(
1070        &self,
1071        key: &str,
1072    ) -> Result<SwrRead<T>, CachekitError> {
1073        self.client.reject_namespaced_interop()?;
1074        match self.client.get_bytes_swr(key).await? {
1075            SwrRead::Fresh(ct) => Ok(SwrRead::Fresh(redact_decode(crate::interop::deserialize(
1076                &self.decrypt_or_evict(&ct, key)?,
1077            ))?)),
1078            SwrRead::Stale(ct, token) => Ok(SwrRead::Stale(
1079                redact_decode(crate::interop::deserialize(
1080                    &self.decrypt_or_evict(&ct, key)?,
1081                ))?,
1082                token,
1083            )),
1084            SwrRead::Miss => Ok(SwrRead::Miss),
1085        }
1086    }
1087
1088    /// Fetch ciphertext (L1, then L2 with L1 backfill) and decrypt it.
1089    ///
1090    /// Ciphertext retrieval delegates to [`CacheKit::get_bytes`], which returns
1091    /// the stored bytes untransformed — for a secure cache exactly the AES-GCM
1092    /// ciphertext, so decrypt receives the same bytes the backend holds.
1093    async fn get_plaintext(&self, key: &str) -> Result<Option<Vec<u8>>, CachekitError> {
1094        match self.client.get_bytes(key).await? {
1095            Some(ciphertext) => Ok(Some(self.decrypt_or_evict(&ciphertext, key)?)),
1096            None => Ok(None),
1097        }
1098    }
1099
1100    /// Decrypt ciphertext read for `key` against the key this client passes to
1101    /// its backend (`.namespace()` prefix included); on failure, drop the key's L1 copy before propagating the
1102    /// error.
1103    ///
1104    /// An L2 hit is backfilled into L1 before it is decrypted, so an entry
1105    /// that fails authentication would otherwise keep failing from L1 after
1106    /// the backend entry is fixed or deleted. The backend entry itself is
1107    /// left untouched as evidence.
1108    fn decrypt_or_evict(&self, ciphertext: &[u8], key: &str) -> Result<Vec<u8>, CachekitError> {
1109        let full_key = self.client.resolve_key(key)?;
1110        self.encryption
1111            .decrypt(ciphertext, &full_key)
1112            .inspect_err(|_| {
1113                #[cfg(feature = "l1")]
1114                self.client.l1_delete(&full_key);
1115            })
1116    }
1117
1118    /// Delete an encrypted key. Behaves identically to [`CacheKit::delete`].
1119    pub async fn delete(&self, key: &str) -> Result<bool, CachekitError> {
1120        self.client.delete(key).await
1121    }
1122
1123    /// Check if an encrypted key exists. Behaves identically to [`CacheKit::exists`].
1124    pub async fn exists(&self, key: &str) -> Result<bool, CachekitError> {
1125        self.client.exists(key).await
1126    }
1127}
1128
1129/// Decoding decrypted plaintext keeps the `Serialization` class, which
1130/// `#[cachekit]` reads as a miss, but replaces the decoder's message: serde
1131/// quotes the offending value (`invalid type: string "…", expected struct
1132/// …`), and that value is the plaintext encryption exists to protect
1133/// (CWE-532).
1134#[cfg(feature = "encryption")]
1135fn redact_decode<T>(decoded: Result<T, CachekitError>) -> Result<T, CachekitError> {
1136    decoded.map_err(|e| match e {
1137        CachekitError::Serialization(_) => CachekitError::Serialization(format!(
1138            "decrypted entry does not decode as {}",
1139            std::any::type_name::<T>()
1140        )),
1141        other => other,
1142    })
1143}
1144
1145// ── CacheKitBuilder ───────────────────────────────────────────────────────────
1146
1147/// Fluent builder for [`CacheKit`].
1148#[derive(Default)]
1149#[must_use]
1150pub struct CacheKitBuilder {
1151    backend: Option<SharedBackend>,
1152    default_ttl: Option<Duration>,
1153    namespace: Option<String>,
1154    max_payload_bytes: Option<usize>,
1155
1156    #[cfg(feature = "l1")]
1157    l1_capacity: Option<usize>,
1158
1159    #[cfg(feature = "l1")]
1160    no_l1: bool,
1161
1162    #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
1163    swr_enabled: Option<bool>,
1164
1165    #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
1166    swr_threshold_ratio: Option<f64>,
1167
1168    // pub(crate): the `secure` preset's key-path tests read the layer.
1169    #[cfg(feature = "encryption")]
1170    pub(crate) encryption: Option<SharedEncryption>,
1171
1172    #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
1173    reliability: Option<crate::reliability::ReliabilityConfig>,
1174
1175    /// Retry deadlines for the backend `CacheKit::io` built; cleared by
1176    /// [`Self::backend`] so a replaced backend never inherits them.
1177    #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
1178    pub(crate) retry_deadlines: Option<crate::reliability::RetryDeadlines>,
1179}
1180
1181#[cfg(not(feature = "encryption"))]
1182fn encryption_feature_missing(what: &str) -> CachekitError {
1183    CachekitError::Config(format!(
1184        "{what} needs the `encryption` cargo feature, which this build of \
1185         cachekit-rs was compiled without; enable it to encrypt"
1186    ))
1187}
1188
1189impl CacheKitBuilder {
1190    /// Set the storage backend.
1191    pub fn backend(mut self, backend: SharedBackend) -> Self {
1192        self.backend = Some(backend);
1193        #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
1194        {
1195            self.retry_deadlines = None;
1196        }
1197        self
1198    }
1199
1200    /// Override the default TTL (used when no per-call TTL is specified).
1201    pub fn default_ttl(mut self, ttl: Duration) -> Self {
1202        self.default_ttl = Some(ttl);
1203        self
1204    }
1205
1206    /// Set a namespace prefix. All keys will be stored as `{namespace}:{key}`.
1207    pub fn namespace(mut self, ns: impl Into<String>) -> Self {
1208        self.namespace = Some(ns.into());
1209        self
1210    }
1211
1212    /// Set the maximum accepted payload size in bytes.
1213    pub fn max_payload_bytes(mut self, limit: usize) -> Self {
1214        self.max_payload_bytes = Some(limit);
1215        self
1216    }
1217
1218    /// Set the L1 cache capacity (max entries).
1219    #[cfg(feature = "l1")]
1220    pub fn l1_capacity(mut self, capacity: usize) -> Self {
1221        self.l1_capacity = Some(capacity);
1222        self
1223    }
1224
1225    /// Disable the L1 cache entirely.
1226    #[cfg(feature = "l1")]
1227    pub fn no_l1(mut self) -> Self {
1228        self.no_l1 = true;
1229        self
1230    }
1231
1232    /// Enable or disable L1 stale-while-revalidate (default: **enabled**,
1233    /// matching the Python and TypeScript SDKs).
1234    ///
1235    /// With SWR on, an L1 hit older than `swr_threshold_ratio` of its TTL is
1236    /// still served immediately, and the `#[cachekit]` macro schedules
1237    /// exactly one background refresh (deduplicated through
1238    /// [`CacheKit::single_flight`], in-process and — on lock-capable
1239    /// backends — across processes). A hard-expired entry is never served:
1240    /// it falls through to a normal blocking miss.
1241    ///
1242    /// Native targets only: this knob does not exist on wasm32, under the
1243    /// `unsync` feature, or without `l1` — calling it there is a compile
1244    /// error rather than a silent no-op.
1245    #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
1246    pub fn swr_enabled(mut self, enabled: bool) -> Self {
1247        self.swr_enabled = Some(enabled);
1248        self
1249    }
1250
1251    /// Set the SWR freshness threshold as a fraction of each L1 entry's TTL
1252    /// (default: **0.5**, matching the Python and TypeScript SDKs).
1253    ///
1254    /// An entry is *fresh* until it has lived `ratio × TTL` (±10% jitter,
1255    /// drawn once when the entry is inserted, to de-synchronise refreshes
1256    /// across processes), then *stale* — served immediately with a background
1257    /// refresh — until hard expiry. Mirrors cachekit-py's
1258    /// `swr_threshold_ratio` semantics (elapsed-lifetime fraction). Must be in
1259    /// `(0.0, 1.0]`; validated at [`Self::build`].
1260    #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
1261    pub fn swr_threshold_ratio(mut self, ratio: f64) -> Self {
1262        self.swr_threshold_ratio = Some(ratio);
1263        self
1264    }
1265
1266    // Stubs for when the l1 feature is disabled — still compile cleanly.
1267    #[cfg(not(feature = "l1"))]
1268    pub fn l1_capacity(self, _capacity: usize) -> Self {
1269        self
1270    }
1271
1272    #[cfg(not(feature = "l1"))]
1273    pub fn no_l1(self) -> Self {
1274        self
1275    }
1276
1277    /// Wrap the backend in the reliability stack (retry with exponential
1278    /// backoff + jitter, circuit breaker, backpressure) — see
1279    /// [`crate::reliability`].
1280    ///
1281    /// Enabled by default with production settings by the `production`,
1282    /// `secure`, and `io` intent presets; off for `minimal` and for
1283    /// manually-built clients. To opt a preset out, pass
1284    /// [`ReliabilityConfig::disabled()`](crate::reliability::ReliabilityConfig::disabled)
1285    /// — a disabled config applies no wrapping at all.
1286    #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
1287    pub fn reliability(mut self, config: crate::reliability::ReliabilityConfig) -> Self {
1288        self.reliability = Some(config);
1289        self
1290    }
1291
1292    /// Configure encryption from raw master key bytes and tenant ID.
1293    ///
1294    /// The master key must be exactly 32 raw bytes; anything else is a
1295    /// [`CachekitError::Config`]. Hex-encoded keys go to [`Self::encryption`]
1296    /// — their ASCII bytes are 64 bytes long and rejected here.
1297    /// Keys are derived per-tenant via HKDF-SHA256.
1298    ///
1299    /// The built client encrypts every value read and write, not only those
1300    /// through the [`secure_cache()`](CacheKit::secure_cache) handle; see
1301    /// [Encryption](CacheKit#encryption). Calling any encryption method
1302    /// again replaces the earlier configuration.
1303    ///
1304    /// Without the `encryption` cargo feature this method, and every other
1305    /// builder encryption method, returns [`CachekitError::Config`]: a build
1306    /// that cannot encrypt never accepts a key.
1307    #[cfg(feature = "encryption")]
1308    pub fn encryption_from_bytes(
1309        mut self,
1310        master_key: &[u8],
1311        tenant_id: &str,
1312    ) -> Result<Self, CachekitError> {
1313        let layer = crate::encryption::EncryptionLayer::new(master_key, tenant_id)?;
1314        self.encryption = Some(SharedEncryption::new(layer));
1315        Ok(self)
1316    }
1317
1318    /// Configure encryption with decrypt-only previous master keys for
1319    /// key rotation.
1320    ///
1321    /// Writes encrypt under `master_key`; reads attempt it first, then each
1322    /// key in `previous_keys` sequentially (attempt order = slice order).
1323    /// At most 3 previous keys; supplying more is a config error, never
1324    /// truncated. See [`crate::encryption::EncryptionLayer::with_previous_keys`].
1325    ///
1326    /// Every key, current and previous, must be exactly 32 raw bytes.
1327    ///
1328    /// Encrypts every value read and write, as for
1329    /// [`Self::encryption_from_bytes`].
1330    #[cfg(feature = "encryption")]
1331    pub fn encryption_from_bytes_with_previous(
1332        mut self,
1333        master_key: &[u8],
1334        previous_keys: &[&[u8]],
1335        tenant_id: &str,
1336    ) -> Result<Self, CachekitError> {
1337        let layer = crate::encryption::EncryptionLayer::with_previous_keys(
1338            master_key,
1339            previous_keys,
1340            tenant_id,
1341        )?;
1342        self.encryption = Some(SharedEncryption::new(layer));
1343        Ok(self)
1344    }
1345
1346    /// Configure encryption from a hex-encoded master key string.
1347    ///
1348    /// Hex-decodes with the same decoder as [`CacheKit::from_env`] and the
1349    /// `secure` preset. The key must decode to at least 32 bytes; use exactly
1350    /// 32 (64 hex chars), the only length every SDK accepts.
1351    ///
1352    /// Encrypts every value read and write, as for
1353    /// [`Self::encryption_from_bytes`].
1354    #[cfg(feature = "encryption")]
1355    pub fn encryption(self, hex_key: &str, tenant_id: &str) -> Result<Self, CachekitError> {
1356        let bytes = crate::config::decode_master_key_hex(hex_key, "master key")?;
1357        self.encryption_from_hex_decoded(&bytes, &[], tenant_id)
1358    }
1359
1360    /// Shared tail of the hex paths: keys already validated by
1361    /// `decode_master_key_hex`, so the `>= 32` floor applies, not the
1362    /// raw-bytes exactly-32 rule.
1363    #[cfg(feature = "encryption")]
1364    pub(crate) fn encryption_from_hex_decoded(
1365        mut self,
1366        master_key: &[u8],
1367        previous_keys: &[&[u8]],
1368        tenant_id: &str,
1369    ) -> Result<Self, CachekitError> {
1370        let layer = crate::encryption::EncryptionLayer::from_hex_decoded_keys(
1371            master_key,
1372            previous_keys,
1373            tenant_id,
1374        )?;
1375        self.encryption = Some(SharedEncryption::new(layer));
1376        Ok(self)
1377    }
1378
1379    // Without the `encryption` feature the builder methods still exist, so
1380    // feature-agnostic code compiles, but each one is an error: accepting a
1381    // key and building a plaintext client is the silent downgrade protocol
1382    // `spec/intent-presets.md` § Encryption Activation rule 1 forbids.
1383
1384    /// Always [`CachekitError::Config`]: this build has no `encryption`
1385    /// feature.
1386    #[cfg(not(feature = "encryption"))]
1387    pub fn encryption_from_bytes(
1388        self,
1389        _master_key: &[u8],
1390        _tenant_id: &str,
1391    ) -> Result<Self, CachekitError> {
1392        Err(encryption_feature_missing(".encryption_from_bytes()"))
1393    }
1394
1395    /// Always [`CachekitError::Config`]: this build has no `encryption`
1396    /// feature.
1397    #[cfg(not(feature = "encryption"))]
1398    pub fn encryption_from_bytes_with_previous(
1399        self,
1400        _master_key: &[u8],
1401        _previous_keys: &[&[u8]],
1402        _tenant_id: &str,
1403    ) -> Result<Self, CachekitError> {
1404        Err(encryption_feature_missing(
1405            ".encryption_from_bytes_with_previous()",
1406        ))
1407    }
1408
1409    /// Always [`CachekitError::Config`]: this build has no `encryption`
1410    /// feature.
1411    #[cfg(not(feature = "encryption"))]
1412    pub fn encryption(self, _hex_key: &str, _tenant_id: &str) -> Result<Self, CachekitError> {
1413        Err(encryption_feature_missing(".encryption()"))
1414    }
1415
1416    /// Finalise and build the [`CacheKit`] client.
1417    ///
1418    /// Returns an error if no backend was provided.
1419    pub fn build(self) -> Result<CacheKit, CachekitError> {
1420        let backend = self.backend.ok_or_else(|| {
1421            CachekitError::Config("a backend must be provided via .backend()".to_owned())
1422        })?;
1423
1424        // Validate namespace if provided
1425        if let Some(ref ns) = self.namespace {
1426            if ns.is_empty() {
1427                return Err(CachekitError::Config("namespace cannot be empty".into()));
1428            }
1429            if ns.len() > 255 {
1430                return Err(CachekitError::Config("namespace exceeds 255 bytes".into()));
1431            }
1432            if !ns.bytes().all(|b| (0x20..=0x7E).contains(&b)) {
1433                return Err(CachekitError::Config(
1434                    "namespace must be ASCII printable".into(),
1435                ));
1436            }
1437        }
1438
1439        #[cfg(feature = "l1")]
1440        let l1 = if self.no_l1 {
1441            None
1442        } else {
1443            let capacity = self.l1_capacity.unwrap_or(1000);
1444            Some(crate::l1::L1Cache::new(capacity))
1445        };
1446
1447        // SWR defaults mirror the sibling SDKs: enabled, threshold ratio 0.5,
1448        // ratio validated in (0.0, 1.0] exactly like py's L1CacheConfig.
1449        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
1450        let swr_threshold_ratio = {
1451            let ratio = self.swr_threshold_ratio.unwrap_or(0.5);
1452            if !(ratio > 0.0 && ratio <= 1.0) {
1453                return Err(CachekitError::Config(format!(
1454                    "swr_threshold_ratio must be in (0.0, 1.0]; got {ratio}"
1455                )));
1456            }
1457            ratio
1458        };
1459
1460        // Hand the backend a live view of this client's read counters so the
1461        // SaaS telemetry headers report real numbers with no user plumbing.
1462        // Attached to the raw backend, before the reliability decorator, so
1463        // no forwarding is needed; a provider set on the backend's own
1464        // builder wins (see `Backend::attach_metrics`).
1465        #[cfg(feature = "l1")]
1466        let l1_enabled = l1.is_some();
1467        #[cfg(not(feature = "l1"))]
1468        let l1_enabled = false;
1469        let counters = Arc::new(CacheCounters::new(l1_enabled));
1470        backend.attach_metrics(counters.provider());
1471
1472        // Apply the reliability stack last so it decorates the final backend.
1473        // A disabled config is the documented opt-out: skip the (no-op)
1474        // decorator entirely. The layer check lives on ReliabilityConfig
1475        // itself so a future layer can't be missed here (this gate
1476        // shipped that exact bug once already).
1477        #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
1478        let (backend, breaker) = match self.reliability {
1479            Some(config) if !config.is_disabled() => {
1480                crate::reliability::wrap_reliable(backend, config, self.retry_deadlines)
1481            }
1482            _ => (backend, None),
1483        };
1484
1485        Ok(CacheKit {
1486            backend,
1487            default_ttl: self.default_ttl.unwrap_or(Duration::from_secs(300)),
1488            namespace: self.namespace,
1489            max_payload_bytes: self.max_payload_bytes.unwrap_or(5 * 1024 * 1024),
1490            flight: SharedFlight::default(),
1491            counters,
1492
1493            #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
1494            breaker,
1495
1496            #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
1497            mutations: SharedMutations::default(),
1498
1499            #[cfg(feature = "l1")]
1500            l1,
1501
1502            #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
1503            swr_enabled: self.swr_enabled.unwrap_or(true),
1504
1505            #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
1506            swr_threshold_ratio,
1507
1508            #[cfg(feature = "encryption")]
1509            encryption: self.encryption,
1510        })
1511    }
1512}
1513
1514#[cfg(all(test, feature = "l1"))]
1515mod backfill_ttl_tests {
1516    use std::time::Duration;
1517
1518    use super::l1_backfill_ttl;
1519    use crate::backend::Freshness;
1520
1521    const DEFAULT_TTL: Duration = Duration::from_secs(300);
1522
1523    fn fresh_for(secs: u64) -> Freshness {
1524        Freshness {
1525            is_stale: false,
1526            fresh_for: Some(Duration::from_secs(secs)),
1527        }
1528    }
1529
1530    /// (e) No header: today's `min(default_ttl, 30 s)`, unchanged.
1531    #[test]
1532    fn absent_header_keeps_the_30s_cap() {
1533        let absent = Freshness::default();
1534        assert_eq!(
1535            l1_backfill_ttl(DEFAULT_TTL, absent),
1536            Some(Duration::from_secs(30))
1537        );
1538        assert_eq!(
1539            l1_backfill_ttl(Duration::from_secs(10), absent),
1540            Some(Duration::from_secs(10))
1541        );
1542    }
1543
1544    #[test]
1545    fn fresh_for_bounds_but_never_lifts_the_cap() {
1546        assert_eq!(
1547            l1_backfill_ttl(DEFAULT_TTL, fresh_for(5)),
1548            Some(Duration::from_secs(5))
1549        );
1550        assert_eq!(
1551            l1_backfill_ttl(DEFAULT_TTL, fresh_for(3600)),
1552            Some(Duration::from_secs(30))
1553        );
1554    }
1555
1556    #[test]
1557    fn stale_or_zero_forbids_the_backfill() {
1558        assert_eq!(l1_backfill_ttl(DEFAULT_TTL, fresh_for(0)), None);
1559        let stale = |fresh_for| Freshness {
1560            is_stale: true,
1561            fresh_for,
1562        };
1563        assert_eq!(l1_backfill_ttl(DEFAULT_TTL, stale(None)), None);
1564        assert_eq!(
1565            l1_backfill_ttl(DEFAULT_TTL, stale(Some(Duration::from_secs(60)))),
1566            None
1567        );
1568    }
1569}
1570
1571// SWR staleness exists only on native builds with L1 (see tests/swr_tests.rs).
1572#[cfg(all(
1573    test,
1574    feature = "encryption",
1575    feature = "l1",
1576    not(feature = "unsync"),
1577    not(target_arch = "wasm32")
1578))]
1579#[allow(clippy::unwrap_used)] // test-only: a failed setup step should panic loudly
1580mod secure_l1_tests {
1581    use std::time::Duration;
1582
1583    use super::{CacheKit, SharedBackend, SwrRead};
1584    use crate::backend::MemoryBackend;
1585    use crate::error::CachekitError;
1586
1587    /// The Stale arm reads straight from L1, and every public write on an
1588    /// encrypted client stores decryptable ciphertext there, so the
1589    /// undecryptable bytes are planted in L1 directly.
1590    #[tokio::test]
1591    async fn secure_interop_get_swr_evicts_stale_l1_after_decrypt_failure() {
1592        let backend = MemoryBackend::default();
1593        let shared: SharedBackend = std::sync::Arc::new(backend.clone());
1594        let client = CacheKit::builder()
1595            .backend(shared)
1596            .l1_capacity(100)
1597            .swr_threshold_ratio(0.001)
1598            .encryption_from_bytes(&[7u8; 32], "test-tenant")
1599            .unwrap()
1600            .build()
1601            .unwrap();
1602        client.l1_set(
1603            "poisoned-stale",
1604            bytes::Bytes::from_static(b"not ciphertext"),
1605            Duration::from_secs(60),
1606        );
1607        // Real time, not tokio's paused clock: L1 ages entries by std
1608        // Instant. threshold = 0.001 × 60 s = 60 ms (±10%); hard expiry 60 s.
1609        tokio::time::sleep(Duration::from_millis(200)).await;
1610
1611        let secure = client.secure_cache().unwrap();
1612        let err = secure
1613            .interop_get_swr::<String>("poisoned-stale")
1614            .await
1615            .unwrap_err();
1616        assert!(matches!(err, CachekitError::Encryption(_)), "got: {err:?}");
1617        // Served from L1, so the entry had not hard-expired into the L2 path.
1618        let stats = client.stats();
1619        assert_eq!((stats.l1_hits, stats.l2_hits), (1, 0), "{stats:?}");
1620
1621        // Nothing in the backend: a surviving L1 copy would keep failing; an
1622        // evicted one yields a miss.
1623        assert!(backend.store.lock().await.is_empty());
1624        assert!(matches!(
1625            secure.interop_get_swr::<String>("poisoned-stale").await,
1626            Ok(SwrRead::Miss)
1627        ));
1628    }
1629}