headless_lms_server/domain/credit_registration_phases/
breaker.rs

1//! The circuit breaker the study-registry phases share within one worker process.
2//!
3//! `BREAKERS` is a process-local static: `credit-registrar` and `suotar-syncer` are separate OS
4//! processes (see `programs/credit_registrar.rs` and `programs/suotar_syncer.rs`), each with its own
5//! map, so an outage tripping the breaker in one does not pause the study-registry phases of the
6//! other. Only the phases within the same process actually share a breaker per scope key.
7//!
8//! Keyed by scope rather than global: a test driving a deliberate outage for its own course must not
9//! silence the pipeline for every other test running at the same moment. Production only ever uses
10//! the global key.
11
12use std::collections::HashMap;
13use std::sync::{LazyLock, Mutex};
14use std::time::{Duration, Instant};
15
16use uuid::Uuid;
17
18use super::PhaseScope;
19
20pub const MAX_CONSECUTIVE_SUOTAR_FAILURES: u32 = 5;
21pub const SUOTAR_COOLDOWN_SECS: u64 = 300;
22/// Playwright's per-test budget is 100 s, which the production cooldown does not fit inside: a test
23/// that trips the breaker deliberately has to be able to watch it recover.
24pub const TEST_SUOTAR_COOLDOWN_SECS: u64 = 5;
25
26/// What one breaker counts failures for.
27#[derive(Debug, Clone, PartialEq, Eq, Hash)]
28pub enum ScopeKey {
29    /// Production, and any unscoped run.
30    Global,
31    Course(Uuid),
32    User(Uuid),
33    Registrations(Vec<Uuid>),
34}
35
36impl ScopeKey {
37    pub fn of(scope: &PhaseScope) -> Self {
38        if let Some(course_id) = scope.course_id {
39            Self::Course(course_id)
40        } else if let Some(user_id) = scope.user_id {
41            Self::User(user_id)
42        } else if !scope.credit_registration_ids.is_empty() {
43            let mut ids = scope.credit_registration_ids.clone();
44            ids.sort();
45            Self::Registrations(ids)
46        } else {
47            Self::Global
48        }
49    }
50}
51
52/// How long a run of failures that never tripped the breaker is remembered, so a scope never run
53/// again leaves the map. Much longer than a worker tick, so a real outage never loses its count.
54const FAILURE_RUN_MEMORY: Duration = Duration::from_secs(SUOTAR_COOLDOWN_SECS);
55
56#[derive(Debug, Clone)]
57struct BreakerState {
58    consecutive_failures: u32,
59    open_until: Option<Instant>,
60    last_failure_at: Instant,
61}
62
63impl BreakerState {
64    /// Whether the entry still says anything: an open cooldown, or a recent enough run of failures.
65    fn is_live(&self, now: Instant) -> bool {
66        self.open_until.is_some_and(|until| now < until)
67            || now.duration_since(self.last_failure_at) < FAILURE_RUN_MEMORY
68    }
69}
70
71static BREAKERS: LazyLock<Mutex<HashMap<ScopeKey, BreakerState>>> =
72    LazyLock::new(|| Mutex::new(HashMap::new()));
73
74pub fn cooldown(test_mode: bool) -> Duration {
75    Duration::from_secs(if test_mode {
76        TEST_SUOTAR_COOLDOWN_SECS
77    } else {
78        SUOTAR_COOLDOWN_SECS
79    })
80}
81
82/// Whether the phases that call the study registry should skip this iteration.
83pub fn is_open(key: &ScopeKey) -> bool {
84    let now = Instant::now();
85    let mut breakers = lock();
86    let Some(state) = breakers.get(key) else {
87        return false;
88    };
89    if state.open_until.is_some_and(|until| now < until) {
90        return true;
91    }
92    if state.is_live(now) {
93        return false;
94    }
95    // Dropped rather than reset in place so an idle scope leaves the map; the fresh entry the next
96    // failure creates is the state a reset would have left behind anyway.
97    breakers.remove(key);
98    false
99}
100
101/// What one breaker holds right now, in this process.
102#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
103pub struct BreakerSnapshot {
104    pub open: bool,
105    pub consecutive_failures: u32,
106    /// How much of the cooldown is left, in seconds.
107    pub open_for_secs: Option<u64>,
108}
109
110/// Reads a breaker without touching it, for the dashboard. Not [`is_open`], which clears an elapsed
111/// cooldown as a side effect.
112pub fn snapshot(key: &ScopeKey) -> BreakerSnapshot {
113    let breakers = lock();
114    let Some(state) = breakers
115        .get(key)
116        .filter(|state| state.is_live(Instant::now()))
117    else {
118        return BreakerSnapshot::default();
119    };
120    let remaining = state
121        .open_until
122        .and_then(|until| until.checked_duration_since(Instant::now()));
123    BreakerSnapshot {
124        open: remaining.is_some(),
125        consecutive_failures: state.consecutive_failures,
126        open_for_secs: remaining.map(|left| left.as_secs()),
127    }
128}
129
130pub fn record_success(key: &ScopeKey) {
131    let mut breakers = lock();
132    breakers.remove(key);
133}
134
135/// Returns whether this failure opened the breaker.
136pub fn record_failure(key: &ScopeKey, cooldown: Duration) -> bool {
137    let now = Instant::now();
138    let mut breakers = lock();
139    breakers.retain(|_, state| state.is_live(now));
140    let state = breakers.entry(key.clone()).or_insert(BreakerState {
141        consecutive_failures: 0,
142        open_until: None,
143        last_failure_at: now,
144    });
145    state.last_failure_at = now;
146    state.consecutive_failures = state.consecutive_failures.saturating_add(1);
147    if state.consecutive_failures >= MAX_CONSECUTIVE_SUOTAR_FAILURES {
148        state.open_until = Some(now + cooldown);
149        return true;
150    }
151    false
152}
153
154#[cfg(test)]
155pub fn reset(key: &ScopeKey) {
156    lock().remove(key);
157}
158
159fn lock() -> std::sync::MutexGuard<'static, HashMap<ScopeKey, BreakerState>> {
160    // The counters are advisory, so recovering a poisoned lock beats taking the worker down.
161    BREAKERS
162        .lock()
163        .unwrap_or_else(|poisoned| poisoned.into_inner())
164}
165
166#[cfg(test)]
167mod tests {
168    use super::*;
169
170    fn key() -> ScopeKey {
171        ScopeKey::Course(Uuid::new_v4())
172    }
173
174    #[test]
175    fn the_breaker_opens_only_after_the_documented_run_of_failures() {
176        let key = key();
177        for _ in 1..MAX_CONSECUTIVE_SUOTAR_FAILURES {
178            assert!(!record_failure(&key, cooldown(false)));
179            assert!(!is_open(&key));
180        }
181        assert!(record_failure(&key, cooldown(false)));
182        assert!(is_open(&key));
183        reset(&key);
184    }
185
186    #[test]
187    fn one_success_puts_the_run_of_failures_back_to_zero() {
188        let key = key();
189        for _ in 1..MAX_CONSECUTIVE_SUOTAR_FAILURES {
190            record_failure(&key, cooldown(false));
191        }
192        record_success(&key);
193        assert!(!record_failure(&key, cooldown(false)));
194        assert!(!is_open(&key));
195        reset(&key);
196    }
197
198    #[test]
199    fn two_scopes_do_not_trip_each_other() {
200        let storm = key();
201        let bystander = key();
202        for _ in 0..MAX_CONSECUTIVE_SUOTAR_FAILURES {
203            record_failure(&storm, cooldown(false));
204        }
205        assert!(is_open(&storm));
206        assert!(!is_open(&bystander));
207        reset(&storm);
208        reset(&bystander);
209    }
210
211    #[test]
212    fn a_scoped_run_gets_its_own_key_and_an_unscoped_one_gets_the_global_key() {
213        let course = Uuid::new_v4();
214        let user = Uuid::new_v4();
215        assert_eq!(ScopeKey::of(&PhaseScope::default()), ScopeKey::Global);
216        assert_eq!(
217            ScopeKey::of(&PhaseScope::for_course(course)),
218            ScopeKey::Course(course)
219        );
220        assert_eq!(
221            ScopeKey::of(&PhaseScope {
222                user_id: Some(user),
223                ..PhaseScope::default()
224            }),
225            ScopeKey::User(user)
226        );
227    }
228
229    #[test]
230    fn a_registration_scope_is_order_independent() {
231        let first = Uuid::new_v4();
232        let second = Uuid::new_v4();
233        let one = PhaseScope {
234            credit_registration_ids: vec![first, second],
235            ..PhaseScope::default()
236        };
237        let other = PhaseScope {
238            credit_registration_ids: vec![second, first],
239            ..PhaseScope::default()
240        };
241        assert_eq!(ScopeKey::of(&one), ScopeKey::of(&other));
242    }
243
244    #[test]
245    fn a_tripped_breaker_closes_once_its_cooldown_has_elapsed() {
246        let key = key();
247        for _ in 0..MAX_CONSECUTIVE_SUOTAR_FAILURES {
248            record_failure(&key, Duration::ZERO);
249        }
250        assert!(!is_open(&key));
251        reset(&key);
252    }
253}
headless_lms_server/domain/credit_registration_phases/breaker.rs

headless_lms_server/domain/credit_registration_phases/
breaker.rs
