cacheflight 0.2.0

Cache-backed deduplication for async Rust services with stale-while-revalidate support
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130

#![allow(dead_code)]

use cacheflight::{CacheMissReason, Error, MetricsHooks, RecomputeOutcome, RecomputeReason};
use std::{
    sync::{Arc, Mutex},
    time::{Duration, Instant},
};
use tokio::time::sleep;

#[derive(Clone, Default)]
pub struct TestMetrics {
    inner: Arc<Mutex<MetricsSnapshot>>,
}

#[derive(Debug, Clone, Default)]
pub struct MetricsSnapshot {
    pub hits: usize,
    pub stale_hits: usize,
    pub cache_read_failures: usize,
    pub cache_write_failures: usize,
    pub misses: Vec<CacheMissReason>,
    pub deduplicated: Vec<RecomputeReason>,
    pub recompute_started: Vec<RecomputeReason>,
    pub recompute_finished: Vec<(RecomputeReason, RecomputeOutcome)>,
    pub xfetch_early_refreshes: usize,
    pub background_refresh_failures: usize,
}

impl TestMetrics {
    pub fn snapshot(&self) -> MetricsSnapshot {
        self.inner.lock().expect("metrics mutex poisoned").clone()
    }
}

impl MetricsHooks for TestMetrics {
    fn on_background_refresh_failed(&self, _key: &str, _reason: RecomputeReason, _error: &Error) {
        self.inner
            .lock()
            .expect("metrics mutex poisoned")
            .background_refresh_failures += 1;
    }

    fn on_cache_hit(&self, _key: &str) {
        self.inner.lock().expect("metrics mutex poisoned").hits += 1;
    }

    fn on_cache_stale_hit(&self, _key: &str) {
        self.inner
            .lock()
            .expect("metrics mutex poisoned")
            .stale_hits += 1;
    }

    fn on_cache_miss(&self, _key: &str, reason: CacheMissReason) {
        self.inner
            .lock()
            .expect("metrics mutex poisoned")
            .misses
            .push(reason);
    }

    fn on_cache_read_failed(&self, _key: &str, _error: &Error) {
        self.inner
            .lock()
            .expect("metrics mutex poisoned")
            .cache_read_failures += 1;
    }

    fn on_deduplicated(&self, _key: &str, reason: RecomputeReason) {
        self.inner
            .lock()
            .expect("metrics mutex poisoned")
            .deduplicated
            .push(reason);
    }

    fn on_recompute_started(&self, _key: &str, reason: RecomputeReason) {
        self.inner
            .lock()
            .expect("metrics mutex poisoned")
            .recompute_started
            .push(reason);
    }

    fn on_recompute_finished(
        &self,
        _key: &str,
        reason: RecomputeReason,
        outcome: RecomputeOutcome,
        _duration: Duration,
    ) {
        self.inner
            .lock()
            .expect("metrics mutex poisoned")
            .recompute_finished
            .push((reason, outcome));
    }

    fn on_cache_write_failed(&self, _key: &str, _error: &Error) {
        self.inner
            .lock()
            .expect("metrics mutex poisoned")
            .cache_write_failures += 1;
    }

    fn on_xfetch_early_refresh(&self, _key: &str) {
        self.inner
            .lock()
            .expect("metrics mutex poisoned")
            .xfetch_early_refreshes += 1;
    }
}

pub async fn wait_until(
    timeout: Duration,
    interval: Duration,
    mut condition: impl FnMut() -> bool,
) {
    let started = Instant::now();

    while started.elapsed() < timeout {
        if condition() {
            return;
        }

        sleep(interval).await;
    }

    panic!("condition was not satisfied within {:?}", timeout);
}