ecr_store/index/
freshness.rs

1//! When the index may be trusted to answer instead of notmuch.
2//!
3//! Three facts decide it, and the interesting part is how they interact rather
4//! than any one of them, which is why they live together here with tests
5//! instead of as loose fields on the store.
6
7use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
8use std::sync::Mutex;
9use std::time::{Duration, Instant};
10
11/// How long a read trusts the index without asking notmuch whether the database
12/// has moved. Every writer ecr knows about says so directly, so this window
13/// only bounds how long a *stranger's* `notmuch tag` can go unnoticed — at one
14/// cheap process per window rather than one per request.
15pub const REVALIDATE_AFTER: Duration = Duration::from_secs(2);
16
17pub struct Freshness {
18    window: Duration,
19    /// When the index was last known to stand where notmuch does.
20    verified: Mutex<Option<Instant>>,
21    building: AtomicBool,
22    /// Writes this server has made, counted so a refresh can tell whether one
23    /// landed while it was working.
24    writes: AtomicU64,
25}
26
27impl Default for Freshness {
28    fn default() -> Self {
29        Self::new(REVALIDATE_AFTER)
30    }
31}
32
33impl Freshness {
34    pub fn new(window: Duration) -> Self {
35        Self {
36            window,
37            verified: Mutex::new(None),
38            building: AtomicBool::new(false),
39            writes: AtomicU64::new(0),
40        }
41    }
42
43    /// Whether a refresh is writing to the index right now.
44    ///
45    /// A read checks this rather than queueing behind the write: the index is
46    /// one connection behind one mutex, and a chunk of a rebuild takes far
47    /// longer than the notmuch call the reader would otherwise be waiting on,
48    /// so blocking on it would make the index *slower* than not having one.
49    pub fn building(&self) -> bool {
50        self.building.load(Ordering::SeqCst)
51    }
52
53    pub fn begin_build(&self) {
54        self.building.store(true, Ordering::SeqCst);
55    }
56
57    pub fn end_build(&self) {
58        self.building.store(false, Ordering::SeqCst);
59    }
60
61    /// Whether the index was confirmed current recently enough to be believed.
62    pub fn fresh(&self) -> bool {
63        self.verified
64            .lock()
65            .ok()
66            .and_then(|at| *at)
67            .is_some_and(|at| at.elapsed() < self.window)
68    }
69
70    /// Read before doing the work that makes the index current, and handed back
71    /// to [`Self::vouch`] afterwards.
72    pub fn generation(&self) -> u64 {
73        self.writes.load(Ordering::SeqCst)
74    }
75
76    /// Declares the index current — but only if nothing was written since
77    /// `generation` was taken.
78    ///
79    /// A write that lands *during* a refresh is not in what that refresh read,
80    /// so vouching for it unconditionally hides the write for a whole window:
81    /// the tag is in notmuch, the list does not have it, and nothing anywhere
82    /// is in an error state. It takes a build finishing in the same moment as a
83    /// write, which is why it survived every run of the suite but one.
84    pub fn vouch(&self, generation: u64) {
85        if self.generation() != generation {
86            return;
87        }
88        if let Ok(mut at) = self.verified.lock() {
89            *at = Some(Instant::now());
90        }
91    }
92
93    /// A write of ours moved the database, so the next read revalidates — and
94    /// any refresh already in flight can no longer vouch for what it built.
95    pub fn note_write(&self) {
96        self.writes.fetch_add(1, Ordering::SeqCst);
97        if let Ok(mut at) = self.verified.lock() {
98            *at = None;
99        }
100    }
101}
102
103#[cfg(test)]
104mod tests {
105    use super::*;
106
107    fn freshness() -> Freshness {
108        Freshness::new(Duration::from_secs(60))
109    }
110
111    #[test]
112    fn nothing_is_trusted_before_a_refresh_has_vouched() {
113        assert!(!freshness().fresh());
114    }
115
116    #[test]
117    fn a_refresh_that_ran_alone_is_trusted() {
118        let f = freshness();
119        let generation = f.generation();
120        f.vouch(generation);
121
122        assert!(f.fresh());
123    }
124
125    #[test]
126    fn a_write_during_a_refresh_stops_it_vouching() {
127        let f = freshness();
128        let generation = f.generation();
129
130        // The refresh is under way; the write lands before it finishes.
131        f.note_write();
132        f.vouch(generation);
133
134        assert!(!f.fresh(), "the refresh vouched for mail it had not read");
135    }
136
137    #[test]
138    fn a_write_after_a_refresh_retracts_the_vouching() {
139        let f = freshness();
140        f.vouch(f.generation());
141        f.note_write();
142
143        assert!(!f.fresh());
144    }
145
146    #[test]
147    fn a_later_refresh_can_vouch_again_after_a_write() {
148        let f = freshness();
149        f.note_write();
150        f.vouch(f.generation());
151
152        assert!(f.fresh());
153    }
154
155    #[test]
156    fn the_window_expires() {
157        let f = Freshness::new(Duration::ZERO);
158        f.vouch(f.generation());
159
160        assert!(!f.fresh());
161    }
162
163    #[test]
164    fn building_is_reported_while_it_lasts() {
165        let f = freshness();
166        assert!(!f.building());
167
168        f.begin_build();
169        assert!(f.building());
170
171        f.end_build();
172        assert!(!f.building());
173    }
174}
ecr_store/index/freshness.rs

ecr_store/index/
freshness.rs
