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