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}