cachekit/
metrics.rs

1//! Telemetry: live hit/miss counters, the `X-CacheKit-*` SaaS telemetry
2//! headers built from them, and — behind the `tracing` feature — structured
3//! events for cache operations.
4//!
5//! Counters are owned by a [`CacheKit`](crate::CacheKit) and shared by its
6//! clones; read a snapshot with [`CacheKit::stats`](crate::CacheKit::stats).
7//! At build time the client hands a
8//! [`MetricsProvider`](crate::metrics::MetricsProvider) over those counters to
9//! the backend ([`Backend::attach_metrics`](crate::backend::Backend::attach_metrics)),
10//! so the cachekit.io backends' telemetry headers report real numbers with no
11//! user plumbing. A provider given to the backend's own builder takes
12//! precedence.
13
14use std::sync::atomic::{AtomicU64, Ordering};
15use std::sync::Arc;
16use std::time::Duration;
17
18/// Snapshot of cache hit statistics for one [`CacheKit`](crate::CacheKit)
19/// and all of its clones.
20///
21/// Counts cover the value read path — `get`, `interop_get`, their SWR
22/// variants, and the `SecureCache` equivalents. `exists` and reads that end
23/// in a backend error are not counted.
24#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
25pub struct L1Stats {
26    /// Number of reads served from the L1 in-process cache.
27    pub l1_hits: u64,
28    /// Number of reads served from the L2 backend.
29    pub l2_hits: u64,
30    /// Number of reads that found nothing in either layer.
31    pub misses: u64,
32    /// Whether the L1 cache is enabled on this client.
33    pub l1_enabled: bool,
34}
35
36impl L1Stats {
37    /// Fraction of reads served from L1, in `[0.0, 1.0]` (`0.0` before any read).
38    pub fn l1_hit_rate(&self) -> f64 {
39        let total = self.l1_hits + self.l2_hits + self.misses;
40        if total == 0 {
41            0.0
42        } else {
43            self.l1_hits as f64 / total as f64
44        }
45    }
46}
47
48/// Thread-safe closure that produces an optional [`L1Stats`] snapshot.
49pub type MetricsProvider = Arc<dyn Fn() -> Option<L1Stats> + Send + Sync>;
50
51/// Names of the `X-CacheKit-*` telemetry headers, in the order
52/// [`metrics_headers`] emits them: L1 status, L1 hits, L2 hits, misses, L1 hit
53/// rate.
54pub(crate) const METRICS_HEADER_NAMES: [&str; 5] = [
55    "X-CacheKit-L1-Status",
56    "X-CacheKit-L1-Hits",
57    "X-CacheKit-L2-Hits",
58    "X-CacheKit-Misses",
59    "X-CacheKit-L1-Hit-Rate",
60];
61
62/// Build `X-CacheKit-*` HTTP headers from the current L1 stats provider.
63pub fn metrics_headers(provider: Option<&MetricsProvider>) -> Vec<(&'static str, String)> {
64    metrics_headers_named(provider, &METRICS_HEADER_NAMES)
65}
66
67/// [`metrics_headers`] with the names supplied in [`METRICS_HEADER_NAMES`]
68/// order, so the native CachekitIO backend can pass prebuilt `HeaderName`s
69/// instead of having reqwest parse each name on every request.
70pub(crate) fn metrics_headers_named<N: Clone>(
71    provider: Option<&MetricsProvider>,
72    names: &[N; 5],
73) -> Vec<(N, String)> {
74    let [status, l1_hits, l2_hits, misses, hit_rate] = names.clone();
75    let disabled = vec![(status.clone(), "disabled".to_string())];
76
77    let provider = match provider {
78        Some(p) => p,
79        None => return disabled,
80    };
81
82    let stats = match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| (provider)())) {
83        Ok(Some(s)) => s,
84        _ => return disabled,
85    };
86
87    if !stats.l1_enabled {
88        return disabled;
89    }
90
91    vec![
92        // The server accepts exactly `hit`, `miss`, or `disabled` and answers
93        // 400 to an SDK key sending anything else (spec/saas-api.md, "Optional
94        // Metrics Headers"). Like the Python and TypeScript SDKs we report the
95        // aggregate `miss` whenever L1 is on — per-request classification is
96        // not something an L2 request can know about itself.
97        (status, "miss".to_string()),
98        (l1_hits, stats.l1_hits.to_string()),
99        (l2_hits, stats.l2_hits.to_string()),
100        (misses, stats.misses.to_string()),
101        (hit_rate, format!("{:.3}", stats.l1_hit_rate())),
102    ]
103}
104
105// ── Key hashing ──────────────────────────────────────────────────────────────
106
107/// Blake2b-128 digest of a storage key as 32 lowercase hex characters.
108///
109/// Cache keys routinely embed user identifiers, so `tracing` events carry this
110/// digest instead of the key (CWE-532). It is a **correlator, not a
111/// redaction**: it is unkeyed and deterministic, so the same key hashes
112/// identically in every process and matches the File backend's on-disk
113/// filename (and cachekit-py's `hashlib.blake2b(key, digest_size=16)`) — and
114/// for that same reason a low-entropy key such as `user:42` can be recovered
115/// from its digest by enumeration. Treat `cachekit=debug` output with the
116/// care you give the keys themselves.
117pub fn key_hash(key: &str) -> String {
118    use blake2::{digest::consts::U16, Blake2b, Digest};
119
120    let mut hasher = Blake2b::<U16>::new();
121    hasher.update(key.as_bytes());
122    hex::encode(hasher.finalize())
123}
124
125// ── Live counters ────────────────────────────────────────────────────────────
126
127/// How a value read was served. Counted by [`CacheCounters`] and, with the
128/// `tracing` feature, emitted as the `outcome` field of the read event.
129///
130/// The L1 variants are only constructed by the L1 read paths (`L1Stale` by
131/// the native SWR path), so builds without them would flag dead variants.
132#[cfg_attr(
133    not(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32"))),
134    allow(dead_code)
135)]
136#[derive(Debug, Clone, Copy, PartialEq, Eq)]
137pub(crate) enum ReadOutcome {
138    /// Served from the in-process L1 cache, within its freshness window.
139    L1Hit,
140    /// Served from L1 past the SWR freshness threshold: still an L1 hit for
141    /// the counters, but the caller is about to refresh it in the background.
142    L1Stale,
143    /// Served from the L2 backend (and backfilled into L1 when enabled, unless
144    /// the server marked the read stale or its freshness exhausted).
145    L2Hit,
146    /// Neither layer had the key.
147    Miss,
148}
149
150#[cfg(feature = "tracing")]
151impl ReadOutcome {
152    fn as_str(self) -> &'static str {
153        match self {
154            Self::L1Hit => "l1_hit",
155            Self::L1Stale => "l1_stale",
156            Self::L2Hit => "l2_hit",
157            Self::Miss => "miss",
158        }
159    }
160}
161
162/// Live counters behind [`L1Stats`], owned by a client and shared with its
163/// clones and its backend's [`MetricsProvider`].
164///
165/// Relaxed atomics: each counter is an independent monotonic tally and a
166/// snapshot may observe one counter a step ahead of another — fine for
167/// telemetry, and the cost of a stronger ordering on the read hot path is not.
168#[derive(Debug)]
169pub(crate) struct CacheCounters {
170    l1_hits: AtomicU64,
171    l2_hits: AtomicU64,
172    misses: AtomicU64,
173    /// Fixed at build time: whether the owning client has an L1.
174    l1_enabled: bool,
175}
176
177impl CacheCounters {
178    pub(crate) fn new(l1_enabled: bool) -> Self {
179        Self {
180            l1_hits: AtomicU64::new(0),
181            l2_hits: AtomicU64::new(0),
182            misses: AtomicU64::new(0),
183            l1_enabled,
184        }
185    }
186
187    /// Count one read outcome and emit its `tracing` event (feature `tracing`).
188    pub(crate) fn record(&self, outcome: ReadOutcome, full_key: &str) {
189        let counter = match outcome {
190            ReadOutcome::L1Hit | ReadOutcome::L1Stale => &self.l1_hits,
191            ReadOutcome::L2Hit => &self.l2_hits,
192            ReadOutcome::Miss => &self.misses,
193        };
194        counter.fetch_add(1, Ordering::Relaxed);
195        trace_read(full_key, outcome);
196    }
197
198    /// Current totals.
199    pub(crate) fn snapshot(&self) -> L1Stats {
200        L1Stats {
201            l1_hits: self.l1_hits.load(Ordering::Relaxed),
202            l2_hits: self.l2_hits.load(Ordering::Relaxed),
203            misses: self.misses.load(Ordering::Relaxed),
204            l1_enabled: self.l1_enabled,
205        }
206    }
207
208    /// A [`MetricsProvider`] over these counters, for
209    /// [`Backend::attach_metrics`](crate::backend::Backend::attach_metrics).
210    ///
211    /// Holds the counters weakly: a backend can outlive the client that was
212    /// built over it, and once every clone of that client is gone the provider
213    /// reports `None` — the headers fall back to `disabled` — rather than the
214    /// frozen numbers of a dead client.
215    pub(crate) fn provider(self: &Arc<Self>) -> MetricsProvider {
216        let counters = Arc::downgrade(self);
217        Arc::new(move || counters.upgrade().map(|c| c.snapshot()))
218    }
219}
220
221// ── tracing events ───────────────────────────────────────────────────────────
222//
223// One `debug` event per completed cache operation on the `cachekit` target
224// (`RUST_LOG=cachekit=debug`). Fields: `op` (`get` | `set` | `delete`),
225// `key_hash` (see `key_hash`), and per op `outcome` / `ttl_secs` / `existed`.
226// Field expressions only run when a subscriber is enabled for the callsite,
227// so the hash is never computed for an uninterested process. Without the
228// feature these are empty functions the optimizer removes.
229
230/// Target for cache-operation events.
231#[cfg(feature = "tracing")]
232const TARGET: &str = "cachekit";
233
234fn trace_read(full_key: &str, outcome: ReadOutcome) {
235    #[cfg(feature = "tracing")]
236    tracing::debug!(
237        target: TARGET,
238        op = "get",
239        key_hash = %key_hash(full_key),
240        outcome = outcome.as_str(),
241    );
242    #[cfg(not(feature = "tracing"))]
243    let _ = (full_key, outcome);
244}
245
246/// Emit the event for a completed write (`set` and SWR refresh commits).
247pub(crate) fn trace_write(full_key: &str, ttl: Duration) {
248    #[cfg(feature = "tracing")]
249    tracing::debug!(
250        target: TARGET,
251        op = "set",
252        key_hash = %key_hash(full_key),
253        ttl_secs = crate::backend::ttl_wire_secs(ttl),
254    );
255    #[cfg(not(feature = "tracing"))]
256    let _ = (full_key, ttl);
257}
258
259/// Emit the event for a completed delete.
260pub(crate) fn trace_delete(full_key: &str, existed: bool) {
261    #[cfg(feature = "tracing")]
262    tracing::debug!(
263        target: TARGET,
264        op = "delete",
265        key_hash = %key_hash(full_key),
266        existed,
267    );
268    #[cfg(not(feature = "tracing"))]
269    let _ = (full_key, existed);
270}
271
272#[cfg(test)]
273#[allow(clippy::panic)] // test-only: a missing header on a fixture should panic loudly
274mod tests {
275    use super::*;
276
277    fn header<'a>(headers: &'a [(&str, String)], name: &str) -> &'a str {
278        headers
279            .iter()
280            .find(|h| h.0 == name)
281            .map(|h| h.1.as_str())
282            .unwrap_or_else(|| panic!("header {name} missing from {headers:?}"))
283    }
284
285    fn provider(l1_hits: u64, l2_hits: u64, misses: u64, l1_enabled: bool) -> MetricsProvider {
286        Arc::new(move || {
287            Some(L1Stats {
288                l1_hits,
289                l2_hits,
290                misses,
291                l1_enabled,
292            })
293        })
294    }
295
296    #[test]
297    fn disabled_when_no_provider() {
298        let headers = metrics_headers(None);
299        assert_eq!(headers.len(), 1);
300        assert_eq!(headers[0], ("X-CacheKit-L1-Status", "disabled".to_string()));
301    }
302
303    #[test]
304    fn disabled_when_l1_not_enabled() {
305        let headers = metrics_headers(Some(&provider(0, 0, 0, false)));
306        assert_eq!(headers[0].1, "disabled");
307    }
308
309    /// The server rejects any status outside `hit|miss|disabled` with a 400
310    /// for SDK keys — this pins the value the SDK sends when L1 is on.
311    #[test]
312    fn status_is_a_server_accepted_value_when_l1_enabled() {
313        let headers = metrics_headers(Some(&provider(3, 2, 5, true)));
314        let status = header(&headers, "X-CacheKit-L1-Status");
315        assert_eq!(status, "miss");
316        assert!(["hit", "miss", "disabled"].contains(&status));
317    }
318
319    #[test]
320    fn correct_hit_rate_calculation() {
321        let headers = metrics_headers(Some(&provider(3, 2, 5, true)));
322        assert_eq!(header(&headers, "X-CacheKit-L1-Hit-Rate"), "0.300"); // 3 / (3+2+5)
323        assert_eq!(header(&headers, "X-CacheKit-L1-Hits"), "3");
324        assert_eq!(header(&headers, "X-CacheKit-L2-Hits"), "2");
325        assert_eq!(header(&headers, "X-CacheKit-Misses"), "5");
326    }
327
328    #[test]
329    fn zero_division_guard() {
330        let headers = metrics_headers(Some(&provider(0, 0, 0, true)));
331        assert_eq!(header(&headers, "X-CacheKit-L1-Hit-Rate"), "0.000");
332        assert_eq!(L1Stats::default().l1_hit_rate(), 0.0);
333    }
334
335    #[test]
336    fn disabled_when_provider_panics() {
337        #[allow(clippy::panic)]
338        let provider: MetricsProvider = Arc::new(|| panic!("boom"));
339        let headers = metrics_headers(Some(&provider));
340        assert_eq!(headers[0].1, "disabled");
341    }
342
343    #[test]
344    fn counters_record_and_snapshot() {
345        let counters = Arc::new(CacheCounters::new(true));
346        counters.record(ReadOutcome::L1Hit, "k");
347        counters.record(ReadOutcome::L1Stale, "k");
348        counters.record(ReadOutcome::L2Hit, "k");
349        counters.record(ReadOutcome::Miss, "k");
350        assert_eq!(
351            counters.snapshot(),
352            L1Stats {
353                l1_hits: 2,
354                l2_hits: 1,
355                misses: 1,
356                l1_enabled: true,
357            }
358        );
359        // The provider is a live view, not a copy...
360        let provider = counters.provider();
361        counters.record(ReadOutcome::Miss, "k");
362        assert_eq!(provider().map(|s| s.misses), Some(2));
363        // ...and a weak one: a dead client reports nothing, not stale numbers.
364        drop(counters);
365        assert_eq!(provider(), None);
366        assert_eq!(metrics_headers(Some(&provider))[0].1, "disabled");
367    }
368
369    /// Same digest as the File backend filename and cachekit-py's
370    /// `hashlib.blake2b(key, digest_size=16).hexdigest()`.
371    #[test]
372    fn key_hash_is_py_blake2b16_hex() {
373        assert_eq!(
374            key_hash("ns:app:func:m.f:args:abc:v1"),
375            "bcb35ae6f64fa65b2770ab3af631b1ce" // pragma: allowlist secret
376        );
377        assert_eq!(key_hash("").len(), 32);
378    }
379}
cachekit/metrics.rs

cachekit/
metrics.rs
