cachekit-rs 0.13.1

//! `tracing` feature: every completed cache operation, every failed
//! fill-lock call and every failed `#[cachekit]` store emits one debug event
//! on the `cachekit` target carrying `key_hash` — never the key — and
//! circuit-breaker transitions emit on `cachekit::reliability`.
//!
//! Run with:
//!   cargo test --test tracing_tests --features tracing

#![cfg(all(feature = "tracing", not(target_arch = "wasm32")))]

mod common;

use std::fmt;
use std::sync::{Arc, Mutex};

use tracing::field::{Field, Visit};
use tracing::span::{Attributes, Id, Record};
use tracing::{Event, Metadata, Subscriber};

use crate::common::MockBackend;
use cachekit::metrics::key_hash;
use cachekit::CacheKit;

// ── Capturing subscriber ─────────────────────────────────────────────────────

/// A reaction a test installs on the subscriber: runs synchronously inside
/// `event()` for the first event whose rendering contains the trigger, before
/// that event is recorded — the shape of a real alerting layer that reads the
/// breaker back when it sees a transition.
type Hook = Box<dyn Fn(&str) + Send + Sync>;

/// Renders every event as `"<target> <LEVEL> field=value ..."`.
#[derive(Default, Clone)]
struct Capture {
    lines: Arc<Mutex<Vec<String>>>,
    on_event: Arc<Mutex<Option<(&'static str, Hook)>>>,
}

impl Capture {
    fn lines(&self) -> Vec<String> {
        self.lines.lock().expect("capture lock").clone()
    }

    fn on_event_containing(&self, trigger: &'static str, hook: Hook) {
        *self.on_event.lock().expect("hook lock") = Some((trigger, hook));
    }

    /// Take the hook if this line triggers it — taken before running, so a
    /// hook that emits nested events cannot re-enter itself, and no lock is
    /// held while it runs.
    fn take_hook_for(&self, line: &str) -> Option<Hook> {
        let mut slot = self.on_event.lock().expect("hook lock");
        match &*slot {
            Some((trigger, _)) if line.contains(trigger) => slot.take().map(|(_, hook)| hook),
            _ => None,
        }
    }
}

struct Render(String);

impl Visit for Render {
    fn record_debug(&mut self, field: &Field, value: &dyn fmt::Debug) {
        self.0.push_str(&format!(" {}={value:?}", field.name()));
    }

    fn record_str(&mut self, field: &Field, value: &str) {
        self.0.push_str(&format!(" {}={value}", field.name()));
    }
}

impl Subscriber for Capture {
    fn enabled(&self, _: &Metadata<'_>) -> bool {
        true
    }

    fn new_span(&self, _: &Attributes<'_>) -> Id {
        Id::from_u64(1)
    }

    fn record(&self, _: &Id, _: &Record<'_>) {}

    fn record_follows_from(&self, _: &Id, _: &Id) {}

    fn event(&self, event: &Event<'_>) {
        let meta = event.metadata();
        let mut render = Render(format!("{} {}", meta.target(), meta.level()));
        event.record(&mut render);
        if let Some(hook) = self.take_hook_for(&render.0) {
            hook(&render.0);
        }
        self.lines.lock().expect("capture lock").push(render.0);
    }

    fn enter(&self, _: &Id) {}

    fn exit(&self, _: &Id) {}
}

fn has(lines: &[String], target: &str, needles: &[&str]) -> bool {
    lines
        .iter()
        .any(|l| l.starts_with(target) && needles.iter().all(|n| l.contains(n)))
}

// ── Operation events ─────────────────────────────────────────────────────────

#[tokio::test]
async fn operations_emit_hashed_key_events_and_never_the_key() {
    let capture = Capture::default();
    let _guard = tracing::subscriber::set_default(capture.clone());

    let client = CacheKit::builder()
        .backend(MockBackend::shared())
        .namespace("ns")
        .build()
        .expect("client builds");
    let raw_key = "user:42";

    let _: Option<u32> = client.get(raw_key).await.expect("get");
    client.set(raw_key, &1u32).await.expect("set");
    let _: Option<u32> = client.get(raw_key).await.expect("get");
    assert!(client.delete(raw_key).await.expect("delete"));

    let lines = capture.lines();
    assert!(
        !lines.iter().any(|l| l.contains(raw_key)),
        "raw key leaked into an event: {lines:?}"
    );

    let hash = format!("key_hash={}", key_hash("ns:user:42"));
    let hit = if cfg!(feature = "l1") {
        "outcome=l1_hit"
    } else {
        "outcome=l2_hit"
    };
    for needles in [
        ["op=get", "outcome=miss"],
        ["op=set", "ttl_secs=300"],
        ["op=get", hit],
        ["op=delete", "existed=true"],
    ] {
        assert!(
            has(&lines, "cachekit DEBUG", &[needles[0], needles[1], &hash]),
            "no event with {needles:?} and the namespaced key hash: {lines:?}"
        );
    }
    assert_eq!(
        lines.iter().filter(|l| l.starts_with("cachekit ")).count(),
        4,
        "exactly one event per operation: {lines:?}"
    );
}

// ── Breaker transitions ──────────────────────────────────────────────────────

#[cfg(feature = "reliability")]
mod breaker {
    use std::time::Duration;

    use super::{has, Capture};
    use crate::common::FailingBackend;
    use cachekit::reliability::{CircuitBreakerConfig, ReliabilityConfig};
    use cachekit::{CacheKit, CircuitState};

    #[tokio::test]
    async fn breaker_transitions_emit_on_the_reliability_target() {
        let capture = Capture::default();
        let _guard = tracing::subscriber::set_default(capture.clone());

        let client = CacheKit::builder()
            .backend(FailingBackend::shared())
            .no_l1()
            .reliability(ReliabilityConfig {
                retry: None,
                circuit_breaker: Some(CircuitBreakerConfig {
                    failure_threshold: 1,
                    open_timeout: Duration::from_millis(20),
                    ..CircuitBreakerConfig::default()
                }),
                backpressure: None,
            })
            .build()
            .expect("client builds");

        client
            .get::<u32>("k")
            .await
            .expect_err("backend is down: opens the circuit");
        assert!(
            has(
                &capture.lines(),
                "cachekit::reliability WARN",
                &[
                    "breaker=",
                    "seq=1",
                    "from=Closed",
                    "to=Open",
                    "circuit breaker opened"
                ],
            ),
            "{:?}",
            capture.lines()
        );

        tokio::time::sleep(Duration::from_millis(60)).await;
        assert_eq!(client.circuit_state(), Some(CircuitState::HalfOpen));
        assert!(
            has(
                &capture.lines(),
                "cachekit::reliability INFO",
                &["breaker=", "seq=2", "from=Open", "to=HalfOpen"],
            ),
            "{:?}",
            capture.lines()
        );
    }

    /// Several clients under one subscriber: `seq` restarts at 1 per breaker,
    /// so `seq` alone cannot tell whose `seq=2` it is. `breaker` is a
    /// process-unique id; grouping by it and ordering by `seq` reconstructs
    /// each timeline.
    #[tokio::test]
    async fn breaker_field_groups_events_when_several_breakers_share_a_subscriber() {
        let capture = Capture::default();
        let _guard = tracing::subscriber::set_default(capture.clone());

        let build = |open_timeout: Duration| {
            CacheKit::builder()
                .backend(FailingBackend::shared())
                .no_l1()
                .reliability(ReliabilityConfig {
                    retry: None,
                    circuit_breaker: Some(CircuitBreakerConfig {
                        failure_threshold: 1,
                        open_timeout,
                        ..CircuitBreakerConfig::default()
                    }),
                    backpressure: None,
                })
                .build()
                .expect("client builds")
        };
        let a = build(Duration::from_secs(60));
        let b = build(Duration::ZERO);

        a.get::<u32>("k").await.expect_err("A opens"); // A: seq=1
        b.get::<u32>("k").await.expect_err("B opens"); // B: seq=1
        assert_eq!(b.circuit_state(), Some(CircuitState::HalfOpen)); // B: seq=2
        assert_eq!(a.circuit_state(), Some(CircuitState::Open));

        let lines: Vec<String> = capture
            .lines()
            .into_iter()
            .filter(|l| l.starts_with("cachekit::reliability"))
            .collect();
        assert_eq!(lines.len(), 3, "{lines:?}");

        let field = |line: &str, name: &str| -> String {
            line.split_whitespace()
                .find_map(|kv| kv.strip_prefix(&format!("{name}=")))
                .unwrap_or_else(|| panic!("no {name}= in {line:?}"))
                .to_owned()
        };
        let (a_id, b_id) = (field(&lines[0], "breaker"), field(&lines[1], "breaker"));
        assert_ne!(a_id, b_id, "each breaker has its own id");
        // Two `seq=1` events: only the pair disambiguates them.
        assert_eq!(
            [&lines[0], &lines[1], &lines[2]].map(|l| field(l, "seq")),
            ["1", "1", "2"]
        );

        let timeline = |id: &str| -> Vec<(u64, String)> {
            let mut t: Vec<(u64, String)> = lines
                .iter()
                .filter(|l| field(l, "breaker") == id)
                .map(|l| (field(l, "seq").parse().expect("seq is u64"), field(l, "to")))
                .collect();
            t.sort();
            t
        };
        assert_eq!(timeline(&a_id), vec![(1, "Open".to_owned())]);
        assert_eq!(
            timeline(&b_id),
            vec![(1, "Open".to_owned()), (2, "HalfOpen".to_owned())]
        );
    }

    /// Events are emitted after the breaker lock is released, so while the
    /// subscriber handles `Closed → Open` another thread may read the breaker
    /// (an alerting layer confirming live state) — the deadlock that order
    /// prevents. With `open_timeout = 0` that read commits and emits
    /// `Open → HalfOpen` before the first callback returns, so the later
    /// transition reaches the subscriber first. Arrival order is wrong; `seq`,
    /// assigned under the lock, is the true order.
    ///
    /// The reader runs on its own thread with its own subscriber default:
    /// `tracing` drops events emitted from *inside* a callback on the same
    /// thread (re-entrancy guard), which is also why the race is cross-thread
    /// in production.
    #[cfg(not(feature = "unsync"))]
    #[tokio::test]
    async fn subscriber_may_read_the_breaker_and_seq_orders_out_of_order_events() {
        use std::sync::mpsc;
        use std::sync::{Arc, Mutex};

        let capture = Capture::default();
        let _guard = tracing::subscriber::set_default(capture.clone());

        let client = CacheKit::builder()
            .backend(FailingBackend::shared())
            .no_l1()
            .reliability(ReliabilityConfig {
                retry: None,
                circuit_breaker: Some(CircuitBreakerConfig {
                    failure_threshold: 1,
                    open_timeout: Duration::ZERO,
                    ..CircuitBreakerConfig::default()
                }),
                backpressure: None,
            })
            .build()
            .expect("client builds");

        // The "alerting layer": on the open event, another thread reads the
        // live breaker while this callback waits for its answer. Were the
        // breaker lock still held during emission, the reader would block and
        // the wait below would time out.
        let observed = Arc::new(Mutex::new(None));
        let (probe, recorder, sink) = (client.clone(), capture.clone(), Arc::clone(&observed));
        capture.on_event_containing(
            "to=Open",
            Box::new(move |_| {
                let (tx, rx) = mpsc::channel();
                let (probe, recorder) = (probe.clone(), recorder.clone());
                std::thread::spawn(move || {
                    let state =
                        tracing::subscriber::with_default(recorder, || probe.circuit_state());
                    let _ = tx.send(state);
                });
                *sink.lock().expect("observed lock") = rx
                    .recv_timeout(Duration::from_secs(5))
                    .expect("the breaker lock is released during emission");
            }),
        );

        client
            .get::<u32>("k")
            .await
            .expect_err("backend is down: opens the circuit");

        assert_eq!(
            *observed.lock().expect("observed lock"),
            Some(CircuitState::HalfOpen),
            "the concurrent read returned and saw the zero-timeout half-open"
        );
        let lines: Vec<String> = capture
            .lines()
            .into_iter()
            .filter(|l| l.starts_with("cachekit::reliability"))
            .collect();
        assert_eq!(lines.len(), 2, "{lines:?}");
        assert!(
            lines[0].contains("to=HalfOpen") && lines[0].contains("seq=2"),
            "the nested, later transition arrives first: {lines:?}"
        );
        assert!(
            lines[1].contains("to=Open") && lines[1].contains("seq=1"),
            "the earlier transition arrives second: {lines:?}"
        );
    }
}

// ── Fill-lock failures ───────────────────────────────────────────────────────

#[cfg(all(feature = "reliability", not(feature = "unsync")))]
mod fill_lock {
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::sync::Arc;
    use std::time::Duration;

    use cachekit::backend::{Backend, HealthStatus, LockableBackend};
    use cachekit::metrics::key_hash;
    use cachekit::{BackendError, CacheKit};

    use super::{has, Capture};
    use crate::common::MockBackend;

    /// Grants the fill lock only while `grant` is set; every other lock call
    /// fails with the key in its message, as a transport error's URL carries it.
    #[derive(Default)]
    struct FailingLock {
        mock: MockBackend,
        grant: AtomicBool,
    }

    #[async_trait::async_trait]
    impl Backend for FailingLock {
        async fn get(&self, key: &str) -> Result<Option<Vec<u8>>, BackendError> {
            self.mock.get(key).await
        }

        async fn set(
            &self,
            key: &str,
            value: Vec<u8>,
            ttl: Option<Duration>,
        ) -> Result<(), BackendError> {
            self.mock.set(key, value, ttl).await
        }

        async fn delete(&self, key: &str) -> Result<bool, BackendError> {
            self.mock.delete(key).await
        }

        async fn exists(&self, key: &str) -> Result<bool, BackendError> {
            self.mock.exists(key).await
        }

        async fn health(&self) -> Result<HealthStatus, BackendError> {
            self.mock.health().await
        }

        fn as_lockable(&self) -> Option<&dyn LockableBackend> {
            Some(self)
        }
    }

    #[async_trait::async_trait]
    impl LockableBackend for FailingLock {
        async fn acquire_lock(&self, key: &str, _: u64) -> Result<Option<String>, BackendError> {
            if self.grant.load(Ordering::SeqCst) {
                return Ok(Some("lock-1".to_owned()));
            }
            Err(BackendError::transient(format!(
                "POST /v1/cache/{key}/lock failed"
            )))
        }

        async fn release_lock(&self, key: &str, _: &str) -> Result<bool, BackendError> {
            Err(BackendError::timeout(format!(
                "DELETE /v1/cache/{key}/lock failed"
            )))
        }
    }

    /// A failed lock or unlock call is not silent: it emits the error's kind
    /// on the `cachekit` target, and never its message, which names the key.
    #[tokio::test]
    async fn failed_lock_calls_emit_the_error_kind_and_never_the_key() {
        let capture = Capture::default();
        let _guard = tracing::subscriber::set_default(capture.clone());

        let backend = Arc::new(FailingLock::default());
        let client = CacheKit::builder()
            .backend(backend.clone())
            .namespace("ns")
            .build()
            .expect("client builds");

        drop(client.single_flight("user:42").await);
        backend.grant.store(true, Ordering::SeqCst);
        client.single_flight("user:42").await.release().await;

        let lines = capture.lines();
        assert!(
            !lines.iter().any(|l| l.contains("user:42")),
            "raw key leaked into an event: {lines:?}"
        );
        let hash = format!("key_hash={}", key_hash("ns:user:42"));
        for needles in [
            ["op=lock", "error_kind=transient"],
            ["op=unlock", "error_kind=timeout"],
        ] {
            assert!(
                has(&lines, "cachekit DEBUG", &[needles[0], needles[1], &hash]),
                "no event with {needles:?} and the namespaced key hash: {lines:?}"
            );
        }
    }
}

// ── Macro store failures ─────────────────────────────────────────────────────

#[cfg(all(feature = "macros", not(feature = "unsync")))]
mod macro_store {
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::sync::Arc;
    use std::time::Duration;

    use cachekit::backend::{Backend, HealthStatus};
    use cachekit::interop::{interop_key, InteropValue};
    use cachekit::metrics::key_hash;
    use cachekit::{cachekit, BackendError, CacheKit, CachekitError};

    use super::{has, Capture};
    use crate::common::MockBackend;

    /// Refuses every write with the key in its message, as a transport
    /// error's URL carries it, unless `accepting` is set. Permanent, so no
    /// retry backoff. Reads pass through, so they miss until a write lands.
    #[derive(Default)]
    struct FailingSet {
        mock: MockBackend,
        accepting: AtomicBool,
    }

    #[async_trait::async_trait]
    impl Backend for FailingSet {
        async fn get(&self, key: &str) -> Result<Option<Vec<u8>>, BackendError> {
            self.mock.get(key).await
        }

        async fn set(
            &self,
            key: &str,
            value: Vec<u8>,
            ttl: Option<Duration>,
        ) -> Result<(), BackendError> {
            if self.accepting.load(Ordering::SeqCst) {
                return self.mock.set(key, value, ttl).await;
            }
            Err(BackendError::permanent(format!(
                "PUT /v1/cache/{key} failed"
            )))
        }

        async fn delete(&self, key: &str) -> Result<bool, BackendError> {
            self.mock.delete(key).await
        }

        async fn exists(&self, key: &str) -> Result<bool, BackendError> {
            self.mock.exists(key).await
        }

        async fn health(&self) -> Result<HealthStatus, BackendError> {
            self.mock.health().await
        }
    }

    #[cachekit(client = cache, ttl = 60, interop = "fill", namespace = "store")]
    async fn fill(cache: &CacheKit, id: u64) -> Result<u64, CachekitError> {
        Ok(id * 10)
    }

    /// A `#[cachekit]` fill the backend refuses to store still returns its
    /// result, and is not silent: it emits the error's kind on the `cachekit`
    /// target, and never its message, which names the key.
    #[tokio::test]
    async fn a_failed_fill_store_emits_the_error_kind_and_never_the_key() {
        let capture = Capture::default();
        let _guard = tracing::subscriber::set_default(capture.clone());

        let client = CacheKit::builder()
            .backend(Arc::new(FailingSet::default()))
            .build()
            .expect("client builds");

        assert_eq!(fill(&client, 42).await.expect("result returned"), 420);

        let key = interop_key("store", "fill", &[InteropValue::from(42u64)]).expect("key");
        let lines = capture.lines();
        assert!(
            !lines.iter().any(|l| l.contains(&key)),
            "raw key leaked into an event: {lines:?}"
        );
        let hash = format!("key_hash={}", key_hash(&key));
        assert!(
            has(
                &lines,
                "cachekit DEBUG",
                &["op=set", "error_kind=permanent", &hash]
            ),
            "no store-error event with the key hash: {lines:?}"
        );
    }

    #[cachekit(client = cache, ttl = 4, interop = "refresh", namespace = "store")]
    async fn refresh(cache: &CacheKit, id: u64) -> Result<u64, CachekitError> {
        Ok(id * 10)
    }

    /// An SWR refresh whose commit the backend refuses keeps serving the stale
    /// value, and is not silent either: the background task emits the same
    /// store-error event.
    #[cfg(feature = "l1")]
    #[tokio::test]
    async fn a_failed_refresh_commit_emits_the_error_kind_and_never_the_key() {
        let capture = Capture::default();
        let _guard = tracing::subscriber::set_default(capture.clone());

        let backend = Arc::new(FailingSet::default());
        backend.accepting.store(true, Ordering::SeqCst);
        let client = CacheKit::builder()
            .backend(backend.clone())
            // Stale from 25% of the 4 s TTL (±10%), hard expiry at 4 s.
            .swr_threshold_ratio(0.25)
            .build()
            .expect("client builds");
        assert_eq!(refresh(&client, 7).await.expect("cold fill"), 70);

        tokio::time::sleep(Duration::from_millis(1400)).await;
        backend.accepting.store(false, Ordering::SeqCst);
        assert_eq!(refresh(&client, 7).await.expect("stale hit served"), 70);

        let key = interop_key("store", "refresh", &[InteropValue::from(7u64)]).expect("key");
        let hash = format!("key_hash={}", key_hash(&key));
        // Pin the path: a cold-miss fill emits the same store-error line.
        assert!(
            has(
                &capture.lines(),
                "cachekit DEBUG",
                &["op=get", "outcome=l1_stale", &hash]
            ),
            "second call was not a stale hit: {:?}",
            capture.lines()
        );
        let needles = ["op=set", "error_kind=permanent", hash.as_str()];
        let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
        while !has(&capture.lines(), "cachekit DEBUG", &needles) {
            assert!(
                tokio::time::Instant::now() < deadline,
                "no store-error event from the refresh: {:?}",
                capture.lines()
            );
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        let lines = capture.lines();
        assert!(
            !lines.iter().any(|l| l.contains(&key)),
            "raw key leaked into an event: {lines:?}"
        );
    }
}