Commit `d1942a01f0`

d1942a01f0ba0dc26a87f6ff948ca0ea4a58a95e

parent: 52a0b45f52

Verified · cmc

cmc <hello@cleberg.net> · 2026-09-04 15:25 UTC

Supervise the collectors and bound the ways they hang

signald/src/supervisor.rs owns collector liveness and is the only publisher of
CollectorUp.

The git and terminal collectors run on one producer thread, so a panic in
either used to kill the thread while publish::serve kept handing subscribers a
frozen cache — faces went on rendering numbers that had stopped being true.
Each call is now wrapped: a panic publishes that source down, the tick
continues, and the next tick retries. The terminal collector holds state across
ticks, so a panic replaces it with a fresh one, costing one window of rate data.

The hardware collector was spawned once and never again. If the child died,
signald stayed up, so launchd's KeepAlive never fired and hardware signals
stopped for good. It now runs under a respawn loop: backoff from 1s doubling to
a 60s cap, reset after a run surviving 30s so a flapping child cannot sit at
the ceiling, and no retry limit. A collector that was never configured is
reported down once and not retried — absent is not failed.

run_git used .output() with no timeout, so a held index lock or a vanished
network mount hung the producer thread forever while health still read up. It
now kills the child on a 10s budget. A Rust thread cannot be safely killed, so
killing the subprocess is the only thing that gets the thread back.

Session::keys_per_min had no floor on the sample window: two prompts a
millisecond apart extrapolated a handful of keys into a five-figure rate. Below
a second, report the count. Signal.ts claimed to be monotonic-corrected and
never was; it cannot be, since it is persisted, retention prunes on it, and
CommitsToday means since local midnight. The claim is gone and the constraint
is documented instead.

The zsh hook caps the spool at 1 MiB, checked with zstat so the prompt does not
fork. Nothing consumes the spool while the daemon is away, and a megabyte of
backlog describes sessions the active window would discard anyway.

Also fixes the git_collector flake: unique_dir keyed temp directories by pid
and a nanosecond timestamp, but tests run in parallel threads of one process,
so the pid is shared and the clock is coarser than a nanosecond. Two git inits
into one directory failed with "File exists". A process-wide counter makes the
name unique.

Closes #10

crates/signal-schema/src/lib.rs +6 −1

		@@ -225,7 +225,12 @@ impl Tag {
225	225	pub struct Signal {
226	226	/// Bump on any field change; see [`SCHEMA_VERSION`].
227	227	pub schema_version: u16,
228		/// Unix millis, monotonic-corrected.
	228	/// Wall-clock Unix milliseconds.
	229	///
	230	/// Not monotonic, and it cannot be: this value is persisted, retention
	231	/// prunes on it, and `CommitsToday` means since local midnight. An NTP
	232	/// step can move it backwards, so a consumer computing a rate from two
	233	/// timestamps must handle a non-positive interval.
229	234	pub ts: u64,
230	235	/// Originating collector domain.
231	236	pub source: Source,

crates/signald/src/lib.rs +181 −19

		@@ -19,6 +19,7 @@ use signal_schema::Signal;
19	19
20	20	pub mod history;
21	21	pub mod hub;
	22	pub mod supervisor;
22	23
23	24	/// Whether `sig` may be published: any tag it carries must name a path under
24	25	/// one of the roots the daemon was told to watch.
		@@ -40,8 +41,12 @@ pub mod collectors {
40	41	/// branch/age scalars only — never diff content. This is the cleanest
41	42	/// signal in the suite and the first collector wired.
42	43	pub mod git {
	44	use std::io::Read;
43	45	use std::path::Path;
44		use std::process::Command;
	46	use std::process::{Command, Stdio};
	47	use std::sync::mpsc;
	48	use std::thread;
	49	use std::time::Duration;
45	50
46	51	use signal_schema::{Signal, SignalName, Source, Tag, Value, SCHEMA_VERSION};
47	52
		@@ -120,17 +125,115 @@ pub mod collectors {
120	125	}
121	126
122	127	/// Run `git -C <repo> <args>` and return trimmed stdout on success.
	128	/// How long a single `git` invocation may take before it is killed.
	129	///
	130	/// These are local, aggregate queries that finish in milliseconds. The
	131	/// budget exists for the cases where `git` does not return at all — an
	132	/// index lock held by another process, a repository on a network mount
	133	/// that has gone away. `.output()` would block the producer thread
	134	/// forever, and a Rust thread cannot be safely killed, so the only way
	135	/// to get the thread back is to kill the child.
	136	pub const GIT_TIMEOUT: Duration = Duration::from_secs(10);
	137
123	138	fn run_git(repo: &Path, args: &[&str]) -> Option<String> {
124		let out = Command::new("git")
	139	run_git_with(Path::new("git"), GIT_TIMEOUT, repo, args)
	140	}
	141
	142	/// [`run_git`] with the program and budget named, so the timeout path
	143	/// can be tested against a command that hangs on purpose without
	144	/// hijacking `PATH` for the whole process.
	145	fn run_git_with(
	146	program: &Path,
	147	timeout: Duration,
	148	repo: &Path,
	149	args: &[&str],
	150	) -> Option<String> {
	151	let mut child = Command::new(program)
125	152	.arg("-C")
126	153	.arg(repo)
127	154	.args(args)
128		.output()
	155	.stdin(Stdio::null())
	156	.stdout(Stdio::piped())
	157	.stderr(Stdio::null())
	158	.spawn()
129	159	.ok()?;
130		if !out.status.success() {
131		return None;
	160
	161	// wait_timeout is not in std, so read the pipe on a thread and let
	162	// the channel carry the deadline.
	163	let mut stdout = child.stdout.take()?;
	164	let (tx, rx) = mpsc::channel();
	165	thread::spawn(move \|\| {
	166	let mut buf = String::new();
	167	let read = stdout.read_to_string(&mut buf);
	168	let _ = tx.send(read.map(\|_\| buf));
	169	});
	170
	171	match rx.recv_timeout(timeout) {
	172	Ok(Ok(out)) => match child.wait() {
	173	Ok(status) if status.success() => Some(out.trim().to_string()),
	174	_ => None,
	175	},
	176	Ok(Err(_)) => {
	177	let _ = child.kill();
	178	let _ = child.wait();
	179	None
	180	}
	181	Err(_) => {
	182	eprintln!(
	183	"signald: git {args:?} in {} exceeded {}s; killed",
	184	repo.display(),
	185	timeout.as_secs()
	186	);
	187	let _ = child.kill();
	188	let _ = child.wait();
	189	None
	190	}
	191	}
	192	}
	193	#[cfg(test)]
	194	mod timeout_tests {
	195	use super::*;
	196
	197	/// A `git` that never returns must not hold the producer thread.
	198	#[test]
	199	fn a_hanging_git_is_killed_and_reported_as_a_failure() {
	200	let dir = std::env::temp_dir().join(format!("signald-githang-{}", std::process::id()));
	201	let _ = std::fs::remove_dir_all(&dir);
	202	std::fs::create_dir_all(&dir).unwrap();
	203	let shim = dir.join("hanging-git");
	204	// exec, so the shell is replaced rather than forking: killing
	205	// the child must kill the sleep, not orphan it holding the
	206	// inherited descriptors. A short sleep bounds the damage if a
	207	// future change breaks that.
	208	std::fs::write(&shim, "#!/bin/sh\nexec sleep 30\n").unwrap();
	209	use std::os::unix::fs::PermissionsExt;
	210	std::fs::set_permissions(&shim, std::fs::Permissions::from_mode(0o755)).unwrap();
	211
	212	let budget = Duration::from_secs(1);
	213	let started = std::time::Instant::now();
	214	let out = run_git_with(&shim, budget, &dir, &["rev-parse"]);
	215	let elapsed = started.elapsed();
	216
	217	assert!(out.is_none(), "a killed git is a failed call, not a value");
	218	assert!(
	219	elapsed < budget * 5,
	220	"returned in {elapsed:?}, budget is {budget:?}"
	221	);
	222	let _ = std::fs::remove_dir_all(&dir);
	223	}
	224
	225	/// The ordinary path still returns output.
	226	#[test]
	227	fn a_command_that_returns_is_read() {
	228	let out = run_git_with(
	229	Path::new("/bin/echo"),
	230	Duration::from_secs(5),
	231	Path::new("."),
	232	&["hello"],
	233	);
	234	// `-C .` is passed before the args, so echo prints it too.
	235	assert!(out.expect("echo returns").contains("hello"));
132	236	}
133		Some(String::from_utf8_lossy(&out.stdout).trim().to_string())
134	237	}
135	238	}
136	239
		@@ -163,6 +266,9 @@ pub mod collectors {
163	266	/// aggregates (its shell is idle or gone).
164	267	pub const ACTIVE_WINDOW_MS: u64 = 5 * 60_000;
165	268
	269	/// Shortest gap between two flushes that is treated as a rate sample.
	270	pub const MIN_RATE_WINDOW_MS: u64 = 1_000;
	271
166	272	/// One aggregate flush parsed from the spool: a timestamp, counts, and
167	273	/// the writing shell's session id. Every field is a number — there is
168	274	/// nowhere to put content.
		@@ -206,7 +312,12 @@ pub mod collectors {
206	312	fn keys_per_min(&self) -> f64 {
207	313	if let Some(prev) = self.prev {
208	314	let dt_ms = self.last.ts_ms.saturating_sub(prev.ts_ms);
209		if dt_ms > 0 {
	315	// Two prompts a millisecond apart would extrapolate a
	316	// handful of keys into a five-figure rate. Below the floor
	317	// the sample is too short to be a rate, so report the count
	318	// itself. saturating_sub also collapses a backwards clock
	319	// step to zero and lands here.
	320	if dt_ms >= MIN_RATE_WINDOW_MS {
210	321	return self.last.keys as f64 * 60_000.0 / dt_ms as f64;
211	322	}
212	323	}
		@@ -287,6 +398,47 @@ pub mod collectors {
287	398	}
288	399	}
289	400
	401	#[cfg(test)]
	402	mod rate_tests {
	403	use super::*;
	404
	405	fn flush_at(ts_ms: u64, keys: u64) -> Flush {
	406	Flush { ts_ms, keys, session_s: 1, session: 1 }
	407	}
	408
	409	/// Two prompts in the same millisecond used to extrapolate to a
	410	/// five-figure keys-per-minute. Below the floor, report the count.
	411	#[test]
	412	fn a_sub_second_gap_is_not_extrapolated() {
	413	let s = Session {
	414	prev: Some(flush_at(1_000, 0)),
	415	last: flush_at(1_001, 5),
	416	};
	417	assert_eq!(s.keys_per_min(), 5.0, "1ms apart is a count, not a rate");
	418	}
	419
	420	#[test]
	421	fn a_real_gap_is_a_rate() {
	422	let s = Session {
	423	prev: Some(flush_at(0, 0)),
	424	last: flush_at(60_000, 90),
	425	};
	426	assert_eq!(s.keys_per_min(), 90.0, "90 keys over a minute");
	427	}
	428
	429	/// saturating_sub collapses a backwards clock step to zero, which
	430	/// is below the floor, so it degrades to the count rather than
	431	/// producing a negative or absurd rate.
	432	#[test]
	433	fn a_backwards_clock_step_degrades_to_the_count() {
	434	let s = Session {
	435	prev: Some(flush_at(10_000, 0)),
	436	last: flush_at(9_000, 4),
	437	};
	438	assert_eq!(s.keys_per_min(), 4.0);
	439	}
	440	}
	441
290	442	fn reading_path(spool: &Path) -> PathBuf {
291	443	let mut p = spool.as_os_str().to_os_string();
292	444	p.push(".reading");
		@@ -376,8 +528,7 @@ pub mod collectors {
376	528	pub mod hardware {
377	529	use std::io::{self, Read};
378	530	use std::path::{Path, PathBuf};
379		use std::process::{Child, Command, Stdio};
380		use std::thread;
	531	use std::process::{Command, Stdio};
381	532
382	533	use signal_schema::wire;
383	534
		@@ -406,11 +557,22 @@ pub mod collectors {
406	557	Ok(n)
407	558	}
408	559
409		/// Spawn `program` streaming frames every `interval_ms` and ingest its
410		/// stdout on a background thread. Returns once the child is running;
411		/// the thread logs when the child's stream ends. The child's stdout is
412		/// a pipe, so it exits on its next write after the daemon goes away.
413		pub fn spawn(program: &Path, interval_ms: u64, hub: Hub) -> io::Result<Child> {
	560	/// Spawn `program` and ingest its stdout until the stream ends,
	561	/// blocking the caller. Returns the number of frames published.
	562	///
	563	/// `on_spawn` runs once the child exists, which is where a supervisor
	564	/// marks the collector up — a spawn that fails returns `Err` without
	565	/// calling it.
	566	///
	567	/// The child is killed and reaped before returning, so a supervisor
	568	/// that restarts in a loop cannot accumulate zombies or leave an orphan
	569	/// writing into a pipe nobody reads.
	570	pub fn run(
	571	program: &Path,
	572	interval_ms: u64,
	573	hub: &Hub,
	574	on_spawn: impl FnOnce(),
	575	) -> io::Result<usize> {
414	576	let mut child = Command::new(program)
415	577	.arg("--interval-ms")
416	578	.arg(interval_ms.to_string())
		@@ -419,11 +581,11 @@ pub mod collectors {
419	581	.stderr(Stdio::inherit())
420	582	.spawn()?;
421	583	let mut stdout = child.stdout.take().expect("stdout is piped");
422		thread::spawn(move \|\| match ingest(&mut stdout, &hub) {
423		Ok(n) => eprintln!("signald: hardware collector exited after {n} frame(s)"),
424		Err(e) => eprintln!("signald: hardware collector stream error: {e}"),
425		});
426		Ok(child)
	584	on_spawn();
	585	let result = ingest(&mut stdout, hub);
	586	let _ = child.kill();
	587	let _ = child.wait();
	588	result
427	589	}
428	590
429	591	/// Find [`PROGRAM`] on `PATH`.

crates/signald/src/main.rs +24 −30

		@@ -43,6 +43,7 @@ use std::time::Duration;
43	43	use signal_schema::Source;
44	44	use signald::collectors;
45	45	use signald::history::History;
	46	use signald::supervisor;
46	47	use signald::hub::Hub;
47	48	use signald::publish;
48	49
		@@ -76,39 +77,32 @@ fn main() {
76	77	};
77	78	let hub = Hub::with_history(history);
78	79
79		// Hardware: spawn the out-of-process collector and ingest its frames. Not
80		// having one (Linux, or a dev build not on PATH) is not fatal.
81		if let Some(collector) = &cfg.collector {
82		match collectors::hardware::spawn(collector, cfg.interval.as_millis() as u64, hub.clone()) {
83		Ok(_) => eprintln!("signald: hardware collector {}", collector.display()),
84		Err(e) => eprintln!("signald: cannot start hardware collector {}: {e}", collector.display()),
	80	// Every collector reports through one Health, so a state change reaches the
	81	// bus once and only when it is a change.
	82	let health = supervisor::Health::new();
	83
	84	// Hardware: keep the out-of-process collector running. Not having one
	85	// (Linux, or a dev build not on PATH) is not fatal and is not retried.
	86	match &cfg.collector {
	87	Some(collector) => {
	88	eprintln!("signald: hardware collector {}", collector.display());
	89	let (hub, health) = (hub.clone(), health.clone());
	90	let collector = collector.clone();
	91	let interval_ms = cfg.interval.as_millis() as u64;
	92	thread::spawn(move \|\| supervisor::supervise_hardware(hub, health, collector, interval_ms));
85	93	}
	94	None => supervisor::report_unconfigured(&hub, &health, collectors::hardware::PROGRAM),
86	95	}
87	96
88		// Producer: poll the collectors on a tick and publish into the hub. Runs
89		// for the life of the daemon, independent of any subscriber.
90		let producer = hub.clone();
91		let repos = cfg.repos.clone();
92		let spool = cfg.spool.clone();
93		let interval = cfg.interval;
94		thread::spawn(move \|\| {
95		let mut terminal = collectors::terminal::Collector::new();
96		loop {
97		for repo in &repos {
98		for sig in collectors::git::collect(repo) {
99		// The audited repo-path tag never leaves the watched roots.
100		if !signald::tag_within_roots(&sig, &repos) {
101		continue;
102		}
103		producer.publish(sig);
104		}
105		}
106		for sig in terminal.collect(&spool) {
107		producer.publish(sig);
108		}
109		thread::sleep(interval);
110		}
111		});
	97	// Producer: poll the git and terminal collectors on a tick and publish into
	98	// the hub. Runs for the life of the daemon, independent of any subscriber.
	99	{
	100	let (hub, health) = (hub.clone(), health.clone());
	101	let repos = cfg.repos.clone();
	102	let spool = cfg.spool.clone();
	103	let interval = cfg.interval;
	104	thread::spawn(move \|\| supervisor::run_producer(hub, health, repos, spool, interval));
	105	}
112	106
113	107	eprintln!("signald: watching {} repo(s), spool {}", cfg.repos.len(), cfg.spool.display());
114	108	if let Err(e) = publish::serve(&cfg.socket, hub) {

crates/signald/src/supervisor.rs added +264

		@@ -0,0 +1,264 @@
	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
	17	use std::collections::BTreeMap;
	18	use std::panic::{catch_unwind, AssertUnwindSafe};
	19	use std::path::PathBuf;
	20	use std::sync::{Arc, Mutex};
	21	use std::time::{Duration, Instant};
	22
	23	use signal_schema::{Signal, SignalName, Source, Value, SCHEMA_VERSION};
	24
	25	use crate::collectors;
	26	use crate::hub::Hub;
	27	use crate::now_millis;
	28
	29	/// First delay after a collector exits.
	30	pub 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.
	33	pub 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.
	37	pub const BACKOFF_RESET_AFTER: Duration = Duration::from_secs(30);
	38
	39	/// Retry delays for a collector that keeps exiting.
	40	#[derive(Debug, Clone, Copy)]
	41	pub struct Backoff {
	42	next: Duration,
	43	}
	44
	45	impl Default for Backoff {
	46	fn default() -> Self {
	47	Backoff { next: BACKOFF_START }
	48	}
	49	}
	50
	51	impl 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)]
	69	pub struct Health {
	70	last: Arc<Mutex<BTreeMap<u8, bool>>>,
	71	}
	72
	73	impl 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.
	104	fn 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.
	124	pub 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.
	162	pub 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.
	183	pub 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.
	190	pub 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)]
	196	mod 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	}

crates/signald/tests/git_collector.rs +10 −1

		@@ -7,6 +7,7 @@
7	7
8	8	use std::path::{Path, PathBuf};
9	9	use std::process::Command;
	10	use std::sync::atomic::{AtomicU64, Ordering};
10	11
11	12	use signal_schema::{Signal, SignalName};
12	13	use signald::collectors::git;
		@@ -118,9 +119,17 @@ fn run(repo: &Path, args: &[&str], envs: &[(&str, &str)]) {
118	119	}
119	120
120	121	fn unique_dir(prefix: &str) -> PathBuf {
	122	// pid and a timestamp are not enough on their own: tests run in parallel
	123	// threads of one process, so the pid is shared, and the clock is coarser
	124	// than a nanosecond, so two tests starting together can read the same
	125	// value. Two `git init`s into one directory then fail with "File exists".
	126	// The counter makes the name unique within the process; pid keeps it
	127	// unique across concurrent cargo runs.
	128	static SEQ: AtomicU64 = AtomicU64::new(0);
121	129	let nanos = std::time::SystemTime::now()
122	130	.duration_since(std::time::UNIX_EPOCH)
123	131	.unwrap()
124	132	.as_nanos();
125		std::env::temp_dir().join(format!("{prefix}-{}-{nanos}", std::process::id()))
	133	let seq = SEQ.fetch_add(1, Ordering::Relaxed);
	134	std::env::temp_dir().join(format!("{prefix}-{}-{nanos}-{seq}", std::process::id()))
126	135	}

crates/signald/tests/spool_bound.rs added +58

		@@ -0,0 +1,58 @@
	1	//! The spool cannot grow without bound while the daemon is away.
	2	//!
	3	//! The hook appends one record per prompt and the daemon consumes the spool
	4	//! each tick. When the daemon is not running nothing consumes it, so the file
	5	//! grew forever and the daemon then read the whole thing into memory on
	6	//! recovery. The hook caps it at the source, which is the only place that
	7	//! actually bounds growth.
	8
	9	use std::path::PathBuf;
	10	use std::process::Command;
	11
	12	fn hook_path() -> PathBuf {
	13	PathBuf::from(env!("CARGO_MANIFEST_DIR"))
	14	.join("../../shell-hooks/signald-hooks.zsh")
	15	.canonicalize()
	16	.expect("hook script")
	17	}
	18
	19	/// Source the real hook, drive `precmd` past a small cap, and check the file
	20	/// was truncated rather than left to grow.
	21	#[test]
	22	fn the_hook_truncates_a_spool_past_the_cap() {
	23	let dir = std::env::temp_dir().join(format!("signald-spoolcap-{}", std::process::id()));
	24	let _ = std::fs::remove_dir_all(&dir);
	25	std::fs::create_dir_all(&dir).unwrap();
	26	let spool = dir.join("terminal.spool");
	27
	28	let script = format!(
	29	r#"
	30	export SIGNALD_SPOOL={spool}
	31	export SIGNALD_SPOOL_MAX_BYTES=200
	32	source {hook}
	33	for i in {{1..40}}; do _signald_precmd; done
	34	wc -c < $SIGNALD_SPOOL
	35	"#,
	36	spool = spool.display(),
	37	hook = hook_path().display(),
	38	);
	39	let out = Command::new("zsh")
	40	.arg("-c")
	41	.arg(&script)
	42	.output()
	43	.expect("zsh runs the hook");
	44	assert!(
	45	out.status.success(),
	46	"hook failed: {}",
	47	String::from_utf8_lossy(&out.stderr)
	48	);
	49
	50	let size: u64 = String::from_utf8_lossy(&out.stdout)
	51	.trim()
	52	.parse()
	53	.expect("wc prints a size");
	54	assert!(size <= 200, "spool grew to {size} bytes past a 200-byte cap");
	55	assert!(size > 0, "truncation must not stop the current record landing");
	56
	57	let _ = std::fs::remove_dir_all(&dir);
	58	}

shell-hooks/signald-hooks.zsh +18

		@@ -51,10 +51,28 @@ zle -N self-insert _signald_self_insert
51	51
52	52	# precmd: the previous command finished. Append ONE aggregate record (numbers
53	53	# only) and reset the per-flush key counter.
	54	# Largest spool we will keep. The daemon consumes the spool every tick, so this
	55	# only fills up while it is not running. Records are worth nothing once the
	56	# daemon has been away for long — the sessions they describe fall outside its
	57	# active window anyway — so the file is truncated rather than rotated.
	58	: ${SIGNALD_SPOOL_MAX_BYTES:=1048576}
	59
	60	zmodload -F zsh/stat b:zstat 2>/dev/null
	61
	62	_signald_spool_too_big() {
	63	local -a st
	64	zstat -A st +size $SIGNALD_SPOOL 2>/dev/null \|\| return 1
	65	(( st[1] > SIGNALD_SPOOL_MAX_BYTES ))
	66	}
	67
54	68	_signald_precmd() {
55	69	local now_ms=$(( ${EPOCHREALTIME:-$EPOCHSECONDS} * 1000 ))
56	70	local session=$(( ${EPOCHSECONDS:-0} - _SIGNALD_SESSION_START ))
57	71	mkdir -p ${SIGNALD_SPOOL:h} 2>/dev/null
	72	# zstat rather than `wc -c`: this runs at every prompt and must not fork.
	73	if _signald_spool_too_big; then
	74	: > $SIGNALD_SPOOL
	75	fi
58	76	print -r -- "${now_ms%.*} ${_SIGNALD_KEYS} ${session} $$" >> $SIGNALD_SPOOL
59	77	_SIGNALD_KEYS=0
60	78	}

Commit d1942a01f0

Commit `d1942a01f0`