doiget_core/
repeat.rs

1//! Repeat suppression: a session that already asked about a ref, and was
2//! told something a retry cannot change yet, is not sent back to the network
3//! for the same answer (#507, ADR-0057).
4//!
5//! The rate cap bounds how fast doiget asks. It did not bound how often it
6//! asks the same thing: an agent retrying 500 refused DOIs ten times each is
7//! 5,000 requests without ever exceeding 5/s. The provenance log already
8//! records, per call, what the caller was told (`session_end` rows carry
9//! `ref` and `error_code`, #507 step 1), so the data to answer "has this
10//! session been told this already?" exists before the network is touched.
11//! [`RepeatIndex`] is that answer, fed from the log as it is written.
12//!
13//! Scope, and why each bound is what it is:
14//!
15//! - **One session**: a `doiget serve` process, or one CLI run (`batch`). The
16//!   capability profile is fixed for its lifetime, so "the configuration has
17//!   not changed" holds by construction -- except for `config.toml`, which
18//!   the HTTP client reads per call. Each entry therefore carries a
19//!   fingerprint of that file, and a changed file lifts the replay.
20//! - **By disposition** (ADR-0055): `terminal` and `needs_config` answers are
21//!   replayed for [`REPLAY_WINDOW`]; a `retry_after` answer is let through
22//!   once [`RETRY_AFTER_GAP`] has passed and refused with the true remaining
23//!   time before that. Backoff keeps working; hammering does not.
24//! - **Never silent**: a replay is an error the caller can see is a replay,
25//!   with the time of the original answer.
26//! - **Always overridable by the caller, never by configuration**: `force`
27//!   on the request asks anyway, and the log records that it did. There is no
28//!   setting that turns suppression off, which is what LEGAL.md 6a's
29//!   "cannot be overridden by configuration" requires of a safeguard.
30
31use std::collections::HashMap;
32use std::sync::Mutex;
33use std::time::{Duration, Instant};
34
35use chrono::{DateTime, Utc};
36
37use crate::{Disposition, ErrorCode, Ref};
38
39/// How long a `terminal` or `needs_config` answer is replayed.
40pub const REPLAY_WINDOW: Duration = Duration::from_secs(10 * 60);
41
42/// The minimum gap before a `retry_after` answer is asked again.
43pub const RETRY_AFTER_GAP: Duration = Duration::from_secs(30);
44
45#[derive(Debug, Clone)]
46struct Entry {
47    code: ErrorCode,
48    at: Instant,
49    at_utc: DateTime<Utc>,
50    fingerprint: u64,
51}
52
53/// What a new request for a ref should do.
54#[derive(Debug, Clone, PartialEq, Eq)]
55pub enum Verdict {
56    /// Ask the network.
57    Proceed,
58    /// Report the earlier answer again without asking.
59    Replay {
60        /// The code the caller was given.
61        code: ErrorCode,
62        /// When, RFC 3339.
63        at: String,
64    },
65    /// The earlier answer said to back off, and not long enough ago.
66    Wait {
67        /// The earlier code.
68        code: ErrorCode,
69        /// When it was given, RFC 3339.
70        at: String,
71        /// Seconds until a retry is let through.
72        remaining_secs: u64,
73    },
74}
75
76/// Recent answers, per ref, for one session.
77#[derive(Debug, Default)]
78pub struct RepeatIndex {
79    entries: Mutex<HashMap<String, Entry>>,
80    /// The map's size after the last sweep; the next one waits until it has
81    /// doubled, so sweeping stays amortised O(1) per insert.
82    swept_len: std::sync::atomic::AtomicUsize,
83}
84
85/// Below this many entries the map is never swept.
86const SWEEP_FLOOR: usize = 64;
87
88impl RepeatIndex {
89    /// Record what a call told its caller about `ref_input`: an error code,
90    /// or `None` for a clean success (which clears the ref).
91    ///
92    /// An answer with the same disposition under the same configuration
93    /// keeps the FIRST time: a caller looping on a refusal must not slide
94    /// the window forward. That rule is also what makes a replay's own
95    /// bookend harmless -- it repeats the disposition it replayed (a `Wait`
96    /// reports `RATE_LIMITED`, which is `retry_after` like what it waits on)
97    /// -- without marking requests, so a real answer arriving concurrently
98    /// is never mistaken for one. A different disposition, or a success,
99    /// replaces the entry.
100    pub fn observe(&self, ref_input: &str, code: Option<ErrorCode>) {
101        self.observe_at(ref_input, code, Instant::now(), config_fingerprint());
102    }
103
104    /// [`RepeatIndex::observe`] with an explicit clock and fingerprint.
105    pub fn observe_at(
106        &self,
107        ref_input: &str,
108        code: Option<ErrorCode>,
109        now: Instant,
110        fingerprint: u64,
111    ) {
112        let key = key(ref_input);
113        let mut map = self.lock();
114        match code {
115            None => {
116                map.remove(&key);
117            }
118            Some(code) => {
119                let same = |e: &Entry| {
120                    e.fingerprint == fingerprint && e.code.disposition() == code.disposition()
121                };
122                if let Some(e) = map.get_mut(&key).filter(|e| same(e)) {
123                    // Keep the time. A terminal answer takes the newer code
124                    // (what the caller was last told); a wait keeps the code
125                    // it is waiting on, since its replay reports RATE_LIMITED.
126                    if code.disposition() != Disposition::RetryAfter {
127                        e.code = code;
128                    }
129                } else {
130                    // A new entry sweeps the ones no rule can act on any
131                    // more (#649 review): nothing replays or waits past
132                    // REPLAY_WINDOW, and a `doiget serve` session otherwise
133                    // kept one entry per refused ref for its whole life.
134                    // Only once the map has doubled since the last sweep, so
135                    // a stream of new refusals is not a scan each.
136                    use std::sync::atomic::Ordering;
137                    let due = SWEEP_FLOOR.max(2 * self.swept_len.load(Ordering::Relaxed));
138                    if map.len() >= due {
139                        map.retain(|_, e| now.saturating_duration_since(e.at) < REPLAY_WINDOW);
140                        self.swept_len.store(map.len(), Ordering::Relaxed);
141                    }
142                    map.insert(
143                        key,
144                        Entry {
145                            code,
146                            at: now,
147                            at_utc: Utc::now(),
148                            fingerprint,
149                        },
150                    );
151                }
152            }
153        }
154    }
155
156    /// The map, recovered from a poisoned lock: a panic elsewhere must not
157    /// silently switch suppression off for the rest of the session, which
158    /// would be the configuration-free off switch ADR-0057 rules out. Every
159    /// write leaves the map consistent, so the data is still sound.
160    fn lock(&self) -> std::sync::MutexGuard<'_, HashMap<String, Entry>> {
161        self.entries
162            .lock()
163            .unwrap_or_else(std::sync::PoisonError::into_inner)
164    }
165
166    /// What a request for `ref_input` should do now.
167    #[must_use]
168    pub fn check(&self, ref_input: &str) -> Verdict {
169        self.check_at(ref_input, Instant::now(), config_fingerprint())
170    }
171
172    /// [`RepeatIndex::check`] with an explicit clock and fingerprint.
173    #[must_use]
174    pub fn check_at(&self, ref_input: &str, now: Instant, fingerprint: u64) -> Verdict {
175        let map = self.lock();
176        let Some(e) = map.get(&key(ref_input)) else {
177            return Verdict::Proceed;
178        };
179        if e.fingerprint != fingerprint {
180            return Verdict::Proceed;
181        }
182        let elapsed = now.saturating_duration_since(e.at);
183        let at = e.at_utc.to_rfc3339_opts(chrono::SecondsFormat::Secs, true);
184        match e.code.disposition() {
185            Disposition::Terminal | Disposition::NeedsConfig if elapsed < REPLAY_WINDOW => {
186                Verdict::Replay { code: e.code, at }
187            }
188            Disposition::RetryAfter if elapsed < RETRY_AFTER_GAP => Verdict::Wait {
189                code: e.code,
190                at,
191                remaining_secs: (RETRY_AFTER_GAP - elapsed).as_secs().max(1),
192            },
193            _ => Verdict::Proceed,
194        }
195    }
196}
197
198/// One key per work, however it was written (`doi:10.1/x`, `10.1/X`).
199fn key(ref_input: &str) -> String {
200    Ref::parse(ref_input).map_or_else(
201        |_| ref_input.trim().to_string(),
202        |r| r.safekey().as_str().to_string(),
203    )
204}
205
206/// A hash of `config.toml`'s bytes, or 0 when there is none. Read per call:
207/// it is the one input the HTTP client re-reads within a session. The read
208/// is a [`crate::store::blocking_section`], as store reads are, since it
209/// runs inside every fetch (#649 review).
210#[must_use]
211pub fn config_fingerprint() -> u64 {
212    use std::hash::{Hash, Hasher};
213    let bytes = crate::user_extension::config_path()
214        .ok()
215        .and_then(|p| crate::store::blocking_section(|| std::fs::read(p)).ok())
216        .unwrap_or_default();
217    let mut h = std::collections::hash_map::DefaultHasher::new();
218    bytes.hash(&mut h);
219    h.finish()
220}
221
222#[cfg(test)]
223#[allow(clippy::expect_used, clippy::unwrap_used, clippy::panic)]
224mod tests {
225    use super::*;
226
227    const R: &str = "10.1137/0117004";
228
229    #[test]
230    fn entries_past_every_window_are_swept_when_a_new_one_lands() {
231        let idx = RepeatIndex::default();
232        let t0 = Instant::now();
233        idx.observe_at(R, Some(ErrorCode::NotFound), t0, 7);
234        for i in 1..SWEEP_FLOOR {
235            idx.observe_at(&format!("10.1/b{i}"), Some(ErrorCode::RateLimited), t0, 7);
236        }
237        assert_eq!(idx.lock().len(), SWEEP_FLOOR, "no sweep below the floor");
238        let later = t0 + REPLAY_WINDOW + Duration::from_secs(1);
239        idx.observe_at("10.1/c", Some(ErrorCode::NotFound), later, 7);
240        assert_eq!(idx.lock().len(), 1, "only the new entry is left");
241        assert!(matches!(
242            idx.check_at("10.1/c", later, 7),
243            Verdict::Replay { .. }
244        ));
245        assert_eq!(idx.check_at(R, later, 7), Verdict::Proceed);
246    }
247
248    #[test]
249    fn a_terminal_answer_is_replayed_within_the_window_and_not_after() {
250        let idx = RepeatIndex::default();
251        let t0 = Instant::now();
252        idx.observe_at(R, Some(ErrorCode::NotFound), t0, 7);
253        assert!(matches!(
254            idx.check_at(R, t0 + Duration::from_secs(5), 7),
255            Verdict::Replay {
256                code: ErrorCode::NotFound,
257                ..
258            }
259        ));
260        assert_eq!(idx.check_at(R, t0 + REPLAY_WINDOW, 7), Verdict::Proceed);
261    }
262
263    #[test]
264    fn the_same_work_written_differently_is_one_key() {
265        let idx = RepeatIndex::default();
266        let t0 = Instant::now();
267        idx.observe_at(
268            "doi:10.1137/0117004",
269            Some(ErrorCode::CapabilityDenied),
270            t0,
271            7,
272        );
273        assert!(matches!(idx.check_at(R, t0, 7), Verdict::Replay { .. }));
274    }
275
276    #[test]
277    fn a_retry_after_answer_is_let_through_after_the_gap_and_timed_before_it() {
278        let idx = RepeatIndex::default();
279        let t0 = Instant::now();
280        idx.observe_at(R, Some(ErrorCode::RateLimited), t0, 7);
281        match idx.check_at(R, t0 + Duration::from_secs(10), 7) {
282            Verdict::Wait { remaining_secs, .. } => assert_eq!(remaining_secs, 20),
283            v => panic!("expected Wait, got {v:?}"),
284        }
285        assert_eq!(idx.check_at(R, t0 + RETRY_AFTER_GAP, 7), Verdict::Proceed);
286    }
287
288    #[test]
289    fn a_changed_config_or_a_success_lifts_the_replay() {
290        let idx = RepeatIndex::default();
291        let t0 = Instant::now();
292        idx.observe_at(R, Some(ErrorCode::CapabilityDenied), t0, 7);
293        assert_eq!(
294            idx.check_at(R, t0, 8),
295            Verdict::Proceed,
296            "config.toml changed"
297        );
298        idx.observe_at(R, None, t0, 7);
299        assert_eq!(
300            idx.check_at(R, t0, 7),
301            Verdict::Proceed,
302            "a later success clears it"
303        );
304    }
305
306    #[test]
307    fn a_replayed_wait_does_not_restart_the_clock_it_enforces() {
308        let idx = RepeatIndex::default();
309        let t0 = Instant::now();
310        idx.observe_at(R, Some(ErrorCode::NetworkError), t0, 7);
311        assert!(matches!(
312            idx.check_at(R, t0 + Duration::from_secs(10), 7),
313            Verdict::Wait { .. }
314        ));
315        // The replay reports RATE_LIMITED; its bookend must not reset `at`.
316        idx.observe_at(
317            R,
318            Some(ErrorCode::RateLimited),
319            t0 + Duration::from_secs(10),
320            7,
321        );
322        assert_eq!(idx.check_at(R, t0 + RETRY_AFTER_GAP, 7), Verdict::Proceed);
323    }
324
325    #[test]
326    fn a_looping_caller_does_not_slide_the_window() {
327        let idx = RepeatIndex::default();
328        let t0 = Instant::now();
329        idx.observe_at(R, Some(ErrorCode::NotFound), t0, 7);
330        // The replayed call's own bookend records the same code again.
331        idx.observe_at(R, Some(ErrorCode::NotFound), t0 + REPLAY_WINDOW / 2, 7);
332        assert_eq!(idx.check_at(R, t0 + REPLAY_WINDOW, 7), Verdict::Proceed);
333    }
334
335    #[test]
336    fn a_different_answer_replaces_the_entry_and_a_terminal_code_is_updated() {
337        let idx = RepeatIndex::default();
338        let t0 = Instant::now();
339        idx.observe_at(R, Some(ErrorCode::RateLimited), t0, 7);
340        // A concurrent real request comes back NOT_FOUND: a new answer.
341        idx.observe_at(R, Some(ErrorCode::NotFound), t0 + Duration::from_secs(1), 7);
342        assert!(matches!(
343            idx.check_at(R, t0 + Duration::from_secs(40), 7),
344            Verdict::Replay {
345                code: ErrorCode::NotFound,
346                ..
347            }
348        ));
349        // Another code of the same disposition keeps the clock but reports
350        // the newer code.
351        let other = ErrorCode::ALL
352            .iter()
353            .copied()
354            .find(|c| {
355                *c != ErrorCode::NotFound && c.disposition() == ErrorCode::NotFound.disposition()
356            })
357            .expect("a second terminal code");
358        idx.observe_at(R, Some(other), t0 + Duration::from_secs(50), 7);
359        match idx.check_at(R, t0 + Duration::from_secs(60), 7) {
360            Verdict::Replay { code, .. } => assert_eq!(code, other),
361            v => panic!("expected Replay, got {v:?}"),
362        }
363        assert_eq!(
364            idx.check_at(R, t0 + Duration::from_secs(1) + REPLAY_WINDOW, 7),
365            Verdict::Proceed,
366            "the window runs from the first terminal answer"
367        );
368    }
369
370    #[test]
371    fn a_poisoned_lock_does_not_switch_suppression_off() {
372        let idx = std::sync::Arc::new(RepeatIndex::default());
373        let t0 = Instant::now();
374        idx.observe_at(R, Some(ErrorCode::NotFound), t0, 7);
375        let poisoner = std::sync::Arc::clone(&idx);
376        let _ = std::thread::spawn(move || {
377            let _guard = poisoner.entries.lock().unwrap();
378            panic!("poison the lock");
379        })
380        .join();
381        assert!(idx.entries.is_poisoned());
382        assert!(matches!(idx.check_at(R, t0, 7), Verdict::Replay { .. }));
383    }
384}
doiget_core/repeat.rs

doiget_core/
repeat.rs
