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

cachekit/
metrics.rs
