faucet_cli/serve/triggers/
watcher.rs

1//! The `Watcher` trait + the supervised polling loop. A watcher never dies on a
2//! transient error: it records health, backs off, and retries until shutdown.
3
4use super::health::TriggersHandle;
5use super::metrics;
6use crate::serve::state::ServerState;
7use async_trait::async_trait;
8use std::time::Duration;
9use tokio_util::sync::CancellationToken;
10
11/// Consecutive poll failures before a watcher is reported unhealthy on /readyz.
12pub const UNHEALTHY_THRESHOLD: u64 = 3;
13/// Backoff ceiling so a persistently-failing watcher still retries periodically.
14pub const MAX_BACKOFF: Duration = Duration::from_secs(60);
15
16/// One pollable watcher (object_arrival / queue_depth). Webhook is push, not
17/// polled, so it does not implement this.
18#[async_trait]
19pub trait Watcher: Send + Sync {
20    fn name(&self) -> &str;
21    fn kind(&self) -> &'static str;
22    fn poll_interval(&self) -> Duration;
23    /// Do one cycle. `Ok(true)` = something fired; `Ok(false)` = idle.
24    async fn poll(&mut self, state: &ServerState) -> Result<bool, String>;
25}
26
27/// Exponential backoff doubling from `base` (the poll interval) to `MAX_BACKOFF`.
28pub fn backoff(base: Duration, consecutive_failures: u64) -> Duration {
29    if consecutive_failures == 0 {
30        return base;
31    }
32    let shift = consecutive_failures.min(20) as u32;
33    let scaled = base.saturating_mul(1u32 << shift.min(16));
34    scaled.min(MAX_BACKOFF).max(base)
35}
36
37/// Run a watcher until `shutdown` fires. Never returns an error (logs instead).
38pub async fn run_supervised<W: Watcher>(
39    mut watcher: W,
40    state: ServerState,
41    health: TriggersHandle,
42    shutdown: CancellationToken,
43) {
44    let base = watcher.poll_interval();
45    let mut failures: u64 = 0;
46    metrics::healthy(watcher.name(), true);
47    loop {
48        let wait = backoff(base, failures);
49        tokio::select! {
50            biased;
51            _ = shutdown.cancelled() => {
52                tracing::info!(trigger = watcher.name(), "trigger watcher stopping");
53                break;
54            }
55            _ = tokio::time::sleep(wait) => {
56                match watcher.poll(&state).await {
57                    Ok(fired) => {
58                        failures = 0;
59                        let stamp = if fired {
60                            Some(chrono::Utc::now().to_rfc3339())
61                        } else {
62                            None
63                        };
64                        if fired {
65                            metrics::last_fire(watcher.name(), chrono::Utc::now().timestamp());
66                        }
67                        health.record_ok(watcher.name(), stamp);
68                        metrics::healthy(watcher.name(), true);
69                    }
70                    Err(e) => {
71                        failures += 1;
72                        tracing::warn!(trigger = watcher.name(), error = %e, "trigger poll failed");
73                        metrics::error(watcher.name(), watcher.kind());
74                        health.record_err(watcher.name(), e, UNHEALTHY_THRESHOLD);
75                        if failures >= UNHEALTHY_THRESHOLD {
76                            metrics::healthy(watcher.name(), false);
77                        }
78                    }
79                }
80            }
81        }
82    }
83}
84
85#[cfg(test)]
86mod tests {
87    use super::*;
88
89    #[test]
90    fn backoff_starts_at_base_and_caps() {
91        let base = Duration::from_secs(10);
92        assert_eq!(backoff(base, 0), base);
93        assert_eq!(backoff(base, 1), Duration::from_secs(20));
94        assert_eq!(backoff(base, 2), Duration::from_secs(40));
95        // Cap at MAX_BACKOFF.
96        assert_eq!(backoff(base, 50), MAX_BACKOFF);
97    }
98}
faucet_cli/serve/triggers/watcher.rs

faucet_cli/serve/triggers/
watcher.rs
