krz/ambient-companions

Supervise the collectors and bound the ways they hang !16

merged merged by cmc on 2026-09-04 15:56 UTC · krz/ambient-companions:feat/supervisor into main

7 files changed, +561 −51

crates/signal-schema/src/lib.rs +6 −1
@@ -225,7 +225,12 @@ impl Tag {
225225pub struct Signal {
226226 /// Bump on any field change; see [`SCHEMA_VERSION`].
227227 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.
229234 pub ts: u64,
230235 /// Originating collector domain.
231236 pub source: Source,
crates/signald/src/lib.rs +181 −19
@@ -19,6 +19,7 @@ use signal_schema::Signal;
1919
2020pub mod history;
2121pub mod hub;
22pub mod supervisor;
2223
2324/// Whether `sig` may be published: any tag it carries must name a path under
2425/// one of the roots the daemon was told to watch.
@@ -40,8 +41,12 @@ pub mod collectors {
4041 /// branch/age scalars only — never diff content. This is the cleanest
4142 /// signal in the suite and the first collector wired.
4243 pub mod git {
44 use std::io::Read;
4345 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;
4550
4651 use signal_schema::{Signal, SignalName, Source, Tag, Value, SCHEMA_VERSION};
4752
@@ -120,17 +125,115 @@ pub mod collectors {
120125 }
121126
122127 /// 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
123138 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)
125152 .arg("-C")
126153 .arg(repo)
127154 .args(args)
128 .output()
155 .stdin(Stdio::null())
156 .stdout(Stdio::piped())
157 .stderr(Stdio::null())
158 .spawn()
129159 .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"));
132236 }
133 Some(String::from_utf8_lossy(&out.stdout).trim().to_string())
134237 }
135238 }
136239
@@ -163,6 +266,9 @@ pub mod collectors {
163266 /// aggregates (its shell is idle or gone).
164267 pub const ACTIVE_WINDOW_MS: u64 = 5 * 60_000;
165268
269 /// Shortest gap between two flushes that is treated as a rate sample.
270 pub const MIN_RATE_WINDOW_MS: u64 = 1_000;
271
166272 /// One aggregate flush parsed from the spool: a timestamp, counts, and
167273 /// the writing shell's session id. Every field is a number — there is
168274 /// nowhere to put content.
@@ -206,7 +312,12 @@ pub mod collectors {
206312 fn keys_per_min(&self) -> f64 {
207313 if let Some(prev) = self.prev {
208314 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 {
210321 return self.last.keys as f64 * 60_000.0 / dt_ms as f64;
211322 }
212323 }
@@ -287,6 +398,47 @@ pub mod collectors {
287398 }
288399 }
289400
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
290442 fn reading_path(spool: &Path) -> PathBuf {
291443 let mut p = spool.as_os_str().to_os_string();
292444 p.push(".reading");
@@ -376,8 +528,7 @@ pub mod collectors {
376528 pub mod hardware {
377529 use std::io::{self, Read};
378530 use std::path::{Path, PathBuf};
379 use std::process::{Child, Command, Stdio};
380 use std::thread;
531 use std::process::{Command, Stdio};
381532
382533 use signal_schema::wire;
383534
@@ -406,11 +557,22 @@ pub mod collectors {
406557 Ok(n)
407558 }
408559
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> {
414576 let mut child = Command::new(program)
415577 .arg("--interval-ms")
416578 .arg(interval_ms.to_string())
@@ -419,11 +581,11 @@ pub mod collectors {
419581 .stderr(Stdio::inherit())
420582 .spawn()?;
421583 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
427589 }
428590
429591 /// Find [`PROGRAM`] on `PATH`.
crates/signald/src/main.rs +24 −30
@@ -43,6 +43,7 @@ use std::time::Duration;
4343use signal_schema::Source;
4444use signald::collectors;
4545use signald::history::History;
46use signald::supervisor;
4647use signald::hub::Hub;
4748use signald::publish;
4849
@@ -76,39 +77,32 @@ fn main() {
7677 };
7778 let hub = Hub::with_history(history);
7879
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));
8593 }
94 None => supervisor::report_unconfigured(&hub, &health, collectors::hardware::PROGRAM),
8695 }
8796
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 }
112106
113107 eprintln!("signald: watching {} repo(s), spool {}", cfg.repos.len(), cfg.spool.display());
114108 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
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}
crates/signald/tests/git_collector.rs +10 −1
@@ -7,6 +7,7 @@
77
88use std::path::{Path, PathBuf};
99use std::process::Command;
10use std::sync::atomic::{AtomicU64, Ordering};
1011
1112use signal_schema::{Signal, SignalName};
1213use signald::collectors::git;
@@ -118,9 +119,17 @@ fn run(repo: &Path, args: &[&str], envs: &[(&str, &str)]) {
118119}
119120
120121fn 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);
121129 let nanos = std::time::SystemTime::now()
122130 .duration_since(std::time::UNIX_EPOCH)
123131 .unwrap()
124132 .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()))
126135}
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
9use std::path::PathBuf;
10use std::process::Command;
11
12fn 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]
22fn 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
5151
5252# precmd: the previous command finished. Append ONE aggregate record (numbers
5353# 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
60zmodload -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
5468_signald_precmd() {
5569 local now_ms=$(( ${EPOCHREALTIME:-$EPOCHSECONDS} * 1000 ))
5670 local session=$(( ${EPOCHSECONDS:-0} - _SIGNALD_SESSION_START ))
5771 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
5876 print -r -- "${now_ms%.*} ${_SIGNALD_KEYS} ${session} $$" >> $SIGNALD_SPOOL
5977 _SIGNALD_KEYS=0
6078}