crates/signald/src/supervisor.rs
264 lines · 9576 bytes
1//! # Collector supervision
2//!
3//! The daemon runs unattended for weeks, so every way a collector can stop has
4//! to be a way it can start again — and when it cannot, that has to reach the
5//! bus rather than a log file nobody reads.
6//!
7//! Two failure shapes need two mechanisms. The git and terminal collectors are
8//! function calls on one producer thread: a panic in either would take the
9//! thread down while `publish::serve` kept handing subscribers a frozen cache,
10//! so each call is wrapped and a panic costs one tick. The hardware collector
11//! is a child process that can die on its own, so it gets a respawn loop.
12//!
13//! Health is published as [`SignalName::CollectorUp`], once per [`Source`], on
14//! transitions only — publishing `up` every tick would put three redundant
15//! frames per tick on the bus for no reader's benefit.
16
17use std::collections::BTreeMap;
18use std::panic::{catch_unwind, AssertUnwindSafe};
19use std::path::PathBuf;
20use std::sync::{Arc, Mutex};
21use std::time::{Duration, Instant};
22
23use signal_schema::{Signal, SignalName, Source, Value, SCHEMA_VERSION};
24
25use crate::collectors;
26use crate::hub::Hub;
27use crate::now_millis;
28
29/// First delay after a collector exits.
30pub const BACKOFF_START: Duration = Duration::from_secs(1);
31/// Ceiling on the retry delay. There is no retry limit: a transient failure at
32/// hour three must not disable a collector until a human notices.
33pub const BACKOFF_CAP: Duration = Duration::from_secs(60);
34/// A run lasting at least this long is treated as a success, and resets the
35/// delay. Without it a child that flaps ratchets the delay to the ceiling and
36/// stays there.
37pub const BACKOFF_RESET_AFTER: Duration = Duration::from_secs(30);
38
39/// Retry delays for a collector that keeps exiting.
40#[derive(Debug, Clone, Copy)]
41pub struct Backoff {
42 next: Duration,
43}
44
45impl Default for Backoff {
46 fn default() -> Self {
47 Backoff { next: BACKOFF_START }
48 }
49}
50
51impl Backoff {
52 /// Record that a run ended after `ran_for`, and return how long to wait
53 /// before the next attempt.
54 pub fn record_exit(&mut self, ran_for: Duration) -> Duration {
55 if ran_for >= BACKOFF_RESET_AFTER {
56 self.next = BACKOFF_START;
57 }
58 let delay = self.next;
59 self.next = (self.next * 2).min(BACKOFF_CAP);
60 delay
61 }
62}
63
64/// Publishes [`SignalName::CollectorUp`] when a collector's state changes.
65///
66/// Shared between the producer thread and the hardware supervisor, so both
67/// report through one place and neither can publish a redundant frame.
68#[derive(Clone, Default)]
69pub struct Health {
70 last: Arc<Mutex<BTreeMap<u8, bool>>>,
71}
72
73impl Health {
74 pub fn new() -> Health {
75 Health::default()
76 }
77
78 /// Publish `up` for `source` if it differs from what was last published.
79 /// The first call for a source always publishes, so a subscriber that
80 /// connects later reads the truth out of the last-value cache.
81 pub fn set(&self, hub: &Hub, source: Source, up: bool) {
82 let mut last = self.last.lock().unwrap();
83 if last.get(&source.to_u8()) == Some(&up) {
84 return;
85 }
86 last.insert(source.to_u8(), up);
87 drop(last);
88 hub.publish(Signal {
89 schema_version: SCHEMA_VERSION,
90 ts: now_millis(),
91 source,
92 name: SignalName::CollectorUp,
93 value: Value(if up { 1.0 } else { 0.0 }),
94 tag: None,
95 });
96 }
97}
98
99/// Run `f`, catching a panic and reporting it as `source` going down.
100///
101/// A panic inside a collector used to kill the producer thread outright. The
102/// daemon stayed up and the socket kept serving, so a face went on rendering
103/// the last values it had — plausible numbers that had stopped being true.
104fn guarded<T>(hub: &Hub, health: &Health, source: Source, f: impl FnOnce() -> T) -> Option<T> {
105 match catch_unwind(AssertUnwindSafe(f)) {
106 Ok(v) => {
107 health.set(hub, source, true);
108 Some(v)
109 }
110 Err(_) => {
111 eprintln!("signald: {source:?} collector panicked; continuing");
112 health.set(hub, source, false);
113 None
114 }
115 }
116}
117
118/// The producer loop: poll the git and terminal collectors on a tick and
119/// publish into `hub`. Runs for the life of the daemon.
120///
121/// A panic in either collector costs that collector one tick, not the thread.
122/// The terminal collector holds state across ticks, so a panic replaces it with
123/// a fresh one — that costs one window of rate data and nothing else.
124pub fn run_producer(
125 hub: Hub,
126 health: Health,
127 repos: Vec<PathBuf>,
128 spool: PathBuf,
129 interval: Duration,
130) -> ! {
131 let mut terminal = collectors::terminal::Collector::new();
132 loop {
133 let signals = guarded(&hub, &health, Source::Git, || {
134 repos
135 .iter()
136 .flat_map(|repo| collectors::git::collect(repo))
137 .filter(|sig| crate::tag_within_roots(sig, &repos))
138 .collect::<Vec<_>>()
139 });
140 for sig in signals.unwrap_or_default() {
141 hub.publish(sig);
142 }
143
144 match guarded(&hub, &health, Source::Terminal, || terminal.collect(&spool)) {
145 Some(signals) => {
146 for sig in signals {
147 hub.publish(sig);
148 }
149 }
150 None => terminal = collectors::terminal::Collector::new(),
151 }
152
153 std::thread::sleep(interval);
154 }
155}
156
157/// Keep the hardware collector running, restarting it with backoff.
158///
159/// The child used to be spawned once and never again: if it died the daemon
160/// stayed up, so launchd's `KeepAlive` never fired and hardware signals stopped
161/// for good.
162pub fn supervise_hardware(hub: Hub, health: Health, program: PathBuf, interval_ms: u64) -> ! {
163 let mut backoff = Backoff::default();
164 loop {
165 let started = Instant::now();
166 let up = || report_hardware_up(&hub, &health);
167 match collectors::hardware::run(&program, interval_ms, &hub, up) {
168 Ok(n) => eprintln!("signald: hardware collector ended after {n} frame(s)"),
169 Err(e) => eprintln!("signald: hardware collector failed: {e}"),
170 }
171 health.set(&hub, Source::Hardware, false);
172 health.set(&hub, Source::Macos, false);
173 let delay = backoff.record_exit(started.elapsed());
174 eprintln!("signald: restarting hardware collector in {}s", delay.as_secs());
175 std::thread::sleep(delay);
176 }
177}
178
179/// Report a collector that was never configured, once.
180///
181/// Absent is not the same as failed: there is nothing to retry, and spinning on
182/// a binary that does not exist is not resilience.
183pub fn report_unconfigured(hub: &Hub, health: &Health, program: &str) {
184 eprintln!("signald: {program} not found; hardware signals disabled");
185 health.set(hub, Source::Hardware, false);
186 health.set(hub, Source::Macos, false);
187}
188
189/// Mark the hardware collector up. Called once the child is producing frames.
190pub fn report_hardware_up(hub: &Hub, health: &Health) {
191 health.set(hub, Source::Hardware, true);
192 health.set(hub, Source::Macos, true);
193}
194
195#[cfg(test)]
196mod tests {
197 use super::*;
198
199 #[test]
200 fn backoff_doubles_to_the_cap() {
201 let mut b = Backoff::default();
202 let brief = Duration::from_secs(1);
203 assert_eq!(b.record_exit(brief), Duration::from_secs(1));
204 assert_eq!(b.record_exit(brief), Duration::from_secs(2));
205 assert_eq!(b.record_exit(brief), Duration::from_secs(4));
206 for _ in 0..10 {
207 b.record_exit(brief);
208 }
209 assert_eq!(b.record_exit(brief), BACKOFF_CAP, "delay is capped");
210 }
211
212 #[test]
213 fn a_run_that_survives_resets_the_delay() {
214 let mut b = Backoff::default();
215 let brief = Duration::from_secs(1);
216 for _ in 0..5 {
217 b.record_exit(brief);
218 }
219 assert_eq!(
220 b.record_exit(BACKOFF_RESET_AFTER),
221 BACKOFF_START,
222 "a child that ran long enough starts over rather than staying at the ceiling"
223 );
224 }
225
226 #[test]
227 fn health_publishes_only_on_change() {
228 let hub = Hub::new();
229 let health = Health::new();
230 health.set(&hub, Source::Git, true);
231 health.set(&hub, Source::Git, true);
232 health.set(&hub, Source::Git, true);
233 assert_eq!(hub.snapshot().len(), 1, "no redundant frames");
234
235 health.set(&hub, Source::Git, false);
236 let snap = hub.snapshot();
237 assert_eq!(snap.len(), 1, "same name and source is keep-latest");
238 assert_eq!(snap[0].value, Value(0.0));
239 }
240
241 #[test]
242 fn health_is_per_source() {
243 let hub = Hub::new();
244 let health = Health::new();
245 health.set(&hub, Source::Git, true);
246 health.set(&hub, Source::Hardware, false);
247 assert_eq!(hub.snapshot().len(), 2, "one entry per source");
248 }
249
250 #[test]
251 fn a_panicking_collector_reports_down_and_does_not_unwind() {
252 let hub = Hub::new();
253 let health = Health::new();
254 let out = guarded(&hub, &health, Source::Terminal, || panic!("collector broke"));
255 assert!(out.is_none());
256 let snap = hub.snapshot();
257 assert_eq!(snap[0].value, Value(0.0), "terminal reported down");
258 assert_eq!(snap[0].source, Source::Terminal);
259
260 let out = guarded(&hub, &health, Source::Terminal, || 7);
261 assert_eq!(out, Some(7), "the next tick still runs");
262 assert_eq!(hub.snapshot()[0].value, Value(1.0), "and reports up again");
263 }
264}