pharos_plugins/
ready.rs

1//! Readiness endpoint configuration and state.
2
3use std::{collections::BTreeMap, sync::Mutex};
4
5use pharos_corefile::{Cursor, Error};
6
7#[derive(Debug)]
8pub struct Ready {
9    pub address: String,
10    pub continuous: bool,
11}
12
13pub(crate) fn parse(cursor: &mut Cursor<'_>) -> Result<Ready, Error> {
14    cursor.next();
15    let args = cursor.remaining_args();
16    if args.len() > 1 {
17        return Err(cursor.arg_err());
18    }
19    let address = args.first().map_or(":8181", String::as_str);
20    if !crate::valid_http_address(address) {
21        return Err(cursor.error("invalid ready address"));
22    }
23    let mut continuous = false;
24    while cursor.next_block() {
25        if cursor.val() != "monitor" {
26            return Err(cursor.error(format!("unknown option: {}", cursor.val())));
27        }
28        let args = cursor.remaining_args();
29        if args.len() != 1 {
30            return Err(cursor.arg_err());
31        }
32        continuous = match args[0].as_str() {
33            "until-ready" => false,
34            "continuously" => true,
35            _ => return Err(cursor.error("unsupported monitor type")),
36        };
37    }
38    Ok(Ready {
39        address: address.to_owned(),
40        continuous,
41    })
42}
43
44type Check = Box<dyn Fn() -> bool + Send + Sync>;
45
46#[derive(Default)]
47struct Inner {
48    checks: BTreeMap<&'static str, Option<Check>>,
49    continuous: bool,
50    shutting_down: bool,
51}
52
53/// Shared process readiness. Shutdown is marked before block shutdown hooks run.
54#[derive(Default)]
55pub struct State(Mutex<Inner>);
56
57impl State {
58    pub fn append(&self, name: &'static str, check: impl Fn() -> bool + Send + Sync + 'static) {
59        self.0
60            .lock()
61            .unwrap()
62            .checks
63            .insert(name, Some(Box::new(check)));
64    }
65
66    pub fn set_continuous(&self, continuous: bool) {
67        self.0.lock().unwrap().continuous = continuous;
68    }
69
70    pub fn begin_shutdown(&self) {
71        self.0.lock().unwrap().shutting_down = true;
72    }
73
74    pub fn response(&self) -> (u16, String) {
75        let mut state = self.0.lock().unwrap();
76        // RFC 9110 §15.6.4: 503 indicates temporary inability to handle requests.
77        if state.shutting_down {
78            return (503, "Shutting down".into());
79        }
80        let continuous = state.continuous;
81        let mut not_ready = Vec::new();
82        for (name, check) in &mut state.checks {
83            if let Some(current) = check {
84                if current() {
85                    if !continuous {
86                        *check = None;
87                    }
88                } else {
89                    not_ready.push(*name);
90                }
91            }
92        }
93        if not_ready.is_empty() {
94            (200, "OK".into())
95        } else {
96            (503, not_ready.join(","))
97        }
98    }
99}
100
101#[cfg(test)]
102mod tests {
103    use super::State;
104    use pharos_server::Readiness;
105    use std::sync::{
106        Arc,
107        atomic::{AtomicBool, Ordering},
108    };
109
110    struct Fake(AtomicBool);
111    impl Readiness for Fake {
112        fn ready(&self) -> bool {
113            self.0.load(Ordering::SeqCst)
114        }
115    }
116
117    #[test]
118    fn monitor_modes_sort_names_and_report_shutdown() {
119        for continuous in [false, true] {
120            let state = State::default();
121            state.set_continuous(continuous);
122            let fake = Arc::new(Fake(AtomicBool::new(false)));
123            for name in ["zebra", "alpha"] {
124                let fake = Arc::clone(&fake);
125                state.append(name, move || fake.ready());
126            }
127            assert_eq!(state.response(), (503, "alpha,zebra".into()));
128            fake.0.store(true, Ordering::SeqCst);
129            assert_eq!(state.response(), (200, "OK".into()));
130            fake.0.store(false, Ordering::SeqCst);
131            assert_eq!(
132                state.response(),
133                if continuous {
134                    (503, "alpha,zebra".into())
135                } else {
136                    (200, "OK".into())
137                }
138            );
139            state.begin_shutdown();
140            assert_eq!(state.response(), (503, "Shutting down".into()));
141        }
142    }
143
144    #[test]
145    fn later_plugin_with_same_name_replaces_earlier_one() {
146        let state = State::default();
147        state.append("same", || false);
148        state.append("same", || true);
149        assert_eq!(state.response(), (200, "OK".into()));
150    }
151}
pharos_plugins/ready.rs

pharos_plugins/
ready.rs
