crates/signald/src/lib.rs
712 lines · 28913 bytes
1//! # signald (library)
2//!
3//! The collector + transport internals of the ambient-companions daemon, split
4//! into a library so the collectors are unit-testable and the thin `signald`
5//! binary (`src/main.rs`) just wires arguments to the runtime.
6//!
7//! v0.2 implements the **git path** and the **terminal path** end to end, adds a
8//! **SQLite (WAL) history store** ([`history`]), and turns the socket into a
9//! **live pub/sub bus with a last-value cache** ([`hub`], [`publish`]). The
10//! sensitive terminal collector is aggregate-only and ships behind the now-active
11//! differential secret-typing test (see `signal-schema/tests/privacy_invariant.rs`).
12//! v0.4 adds the **hardware path**: the daemon spawns the sibling
13//! `macos-collector` and ingests its wire frames ([`collectors::hardware`]).
14
15use std::path::{Path, PathBuf};
16use std::time::{SystemTime, UNIX_EPOCH};
17
18use signal_schema::Signal;
19
20pub mod history;
21pub mod hub;
22pub mod supervisor;
23
24/// Whether `sig` may be published: any tag it carries must name a path under
25/// one of the roots the daemon was told to watch.
26///
27/// `Tag::repo_path` validates shape and nothing else — the watched roots live
28/// here, not in the schema crate. Comparison is by path component, so `/a/bc`
29/// is not under `/a/b`.
30pub fn tag_within_roots(sig: &Signal, roots: &[PathBuf]) -> bool {
31 let Some(tag) = &sig.tag else {
32 return true;
33 };
34 let path = Path::new(tag.as_str());
35 roots.iter().any(|root| path.starts_with(root))
36}
37
38/// Collectors: each reduces its domain to schema scalars.
39pub mod collectors {
40 /// Git collector: shell out to `git` for counts and ages. Counts and
41 /// branch/age scalars only — never diff content. This is the cleanest
42 /// signal in the suite and the first collector wired.
43 pub mod git {
44 use std::io::Read;
45 use std::path::Path;
46 use std::process::{Command, Stdio};
47 use std::sync::mpsc;
48 use std::thread;
49 use std::time::Duration;
50
51 use signal_schema::{Signal, SignalName, Source, Tag, Value, SCHEMA_VERSION};
52
53 use super::super::now_millis;
54
55 /// Rolling window (days) for [`SignalName::CommitsWindow`].
56 pub const WINDOW_DAYS: u64 = 7;
57
58 /// Derive the aggregate git signals for one repo. Every value is a
59 /// scalar `f64`; the only string on the wire is the repo path carried
60 /// in the audited `tag`. Returns an empty vec if `repo` is
61 /// not a git repo.
62 pub fn collect(repo: &Path) -> Vec<Signal> {
63 if !is_git_repo(repo) {
64 return Vec::new();
65 }
66 let ts = now_millis();
67 let now_secs = ts / 1000;
68 let tag = Tag::repo_path(&repo.to_string_lossy());
69
70 let mk = |name: SignalName, value: f64| Signal {
71 schema_version: SCHEMA_VERSION,
72 ts,
73 source: Source::Git,
74 name,
75 value: Value(value),
76 tag: tag.clone(),
77 };
78
79 vec![
80 mk(
81 SignalName::CommitsWindow,
82 commits_since(repo, &format!("{WINDOW_DAYS} days ago")),
83 ),
84 mk(SignalName::CommitsToday, commits_since(repo, "midnight")),
85 mk(SignalName::BranchCount, branch_count(repo)),
86 mk(
87 SignalName::DaysSinceLastCommit,
88 days_since_last_commit(repo, now_secs),
89 ),
90 ]
91 }
92
93 fn is_git_repo(repo: &Path) -> bool {
94 run_git(repo, &["rev-parse", "--is-inside-work-tree"])
95 .map(|s| s == "true")
96 .unwrap_or(false)
97 }
98
99 /// Count commits reachable from `HEAD` newer than `since` (a git
100 /// approxidate, e.g. "midnight" or "7 days ago").
101 fn commits_since(repo: &Path, since: &str) -> f64 {
102 run_git(
103 repo,
104 &["rev-list", "--count", &format!("--since={since}"), "HEAD"],
105 )
106 .and_then(|s| s.parse::<f64>().ok())
107 .unwrap_or(0.0)
108 }
109
110 fn branch_count(repo: &Path) -> f64 {
111 run_git(repo, &["for-each-ref", "--format=%(refname)", "refs/heads/"])
112 .map(|s| if s.is_empty() { 0 } else { s.lines().count() })
113 .unwrap_or(0) as f64
114 }
115
116 /// Whole days between the most recent `HEAD` commit and now. `0.0` when
117 /// the repo has no commits yet.
118 fn days_since_last_commit(repo: &Path, now_secs: u64) -> f64 {
119 match run_git(repo, &["log", "-1", "--format=%ct", "HEAD"])
120 .and_then(|s| s.parse::<u64>().ok())
121 {
122 Some(last) => (now_secs.saturating_sub(last) / 86_400) as f64,
123 None => 0.0,
124 }
125 }
126
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
138 fn run_git(repo: &Path, args: &[&str]) -> Option<String> {
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)
152 .arg("-C")
153 .arg(repo)
154 .args(args)
155 .stdin(Stdio::null())
156 .stdout(Stdio::piped())
157 .stderr(Stdio::null())
158 .spawn()
159 .ok()?;
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"));
236 }
237 }
238 }
239
240 /// Terminal collector (the sensitive one): **no input tap, no PTY sniff.**
241 ///
242 /// The shell side (`shell-hooks/signald-hooks.zsh`) counts keystrokes with a
243 /// `zle` widget that increments a *number* and discards the key, and appends
244 /// aggregate count records — `<epoch_ms> <keys> <session_seconds> <session_id>`,
245 /// numbers only — to a spool file on each `precmd`. Every shell appends to
246 /// the same spool; the session id (the shell's pid) keeps them apart.
247 ///
248 /// [`Collector`](terminal::Collector) consumes the spool each tick (renames
249 /// it aside, parses it, deletes it, so it never regrows) and keeps the last
250 /// two flushes of each session in memory. It derives
251 /// [`SignalName::KeysPerMin`](signal_schema::SignalName::KeysPerMin) as the
252 /// sum of each active session's rate and
253 /// [`SignalName::SessionSeconds`](signal_schema::SignalName::SessionSeconds)
254 /// as the longest active session. A line that is not four integers is
255 /// dropped, so the spool can carry nothing but counts (the privacy contract,
256 /// enforced by the differential secret-typing test).
257 pub mod terminal {
258 use std::collections::BTreeMap;
259 use std::path::{Path, PathBuf};
260
261 use signal_schema::{Signal, SignalName, Source, Value, SCHEMA_VERSION};
262
263 use super::super::now_millis;
264
265 /// A session with no flush inside this window is dropped from the
266 /// aggregates (its shell is idle or gone).
267 pub const ACTIVE_WINDOW_MS: u64 = 5 * 60_000;
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
272 /// One aggregate flush parsed from the spool: a timestamp, counts, and
273 /// the writing shell's session id. Every field is a number — there is
274 /// nowhere to put content.
275 #[derive(Debug, Clone, Copy, PartialEq)]
276 pub struct Flush {
277 pub ts_ms: u64,
278 pub keys: u64,
279 pub session_s: u64,
280 pub session: u64,
281 }
282
283 /// Parse one spool line. Returns `None` for anything that is not exactly
284 /// four integers, so non-count lines can never survive into a signal.
285 pub fn parse_flush(line: &str) -> Option<Flush> {
286 let mut it = line.split_whitespace();
287 let ts_ms = it.next()?.parse().ok()?;
288 let keys = it.next()?.parse().ok()?;
289 let session_s = it.next()?.parse().ok()?;
290 let session = it.next()?.parse().ok()?;
291 if it.next().is_some() {
292 return None; // extra fields => not a well-formed count record
293 }
294 Some(Flush {
295 ts_ms,
296 keys,
297 session_s,
298 session,
299 })
300 }
301
302 /// The previous and latest flush of one session.
303 #[derive(Debug, Clone, Copy)]
304 struct Session {
305 prev: Option<Flush>,
306 last: Flush,
307 }
308
309 impl Session {
310 /// Keys/min over the latest flush interval; the raw count when only
311 /// one flush (or a zero interval) is available.
312 fn keys_per_min(&self) -> f64 {
313 if let Some(prev) = self.prev {
314 let dt_ms = self.last.ts_ms.saturating_sub(prev.ts_ms);
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 {
321 return self.last.keys as f64 * 60_000.0 / dt_ms as f64;
322 }
323 }
324 self.last.keys as f64
325 }
326 }
327
328 /// Per-session state carried across ticks, since the spool is consumed.
329 #[derive(Debug, Default)]
330 pub struct Collector {
331 sessions: BTreeMap<u64, Session>,
332 }
333
334 impl Collector {
335 pub fn new() -> Collector {
336 Collector::default()
337 }
338
339 /// What the daemon does each tick: consume the spool, then derive
340 /// the current aggregates. Empty if no session is active.
341 pub fn collect(&mut self, spool: &Path) -> Vec<Signal> {
342 self.ingest(spool);
343 self.signals(now_millis())
344 }
345
346 /// Consume the spool: rename it aside (the hook opens it with `>>`
347 /// per write, so later appends land in a fresh file), parse, delete.
348 /// A leftover `.reading` file from an interrupted tick is read
349 /// first. Returns the number of records absorbed.
350 pub fn ingest(&mut self, spool: &Path) -> usize {
351 let reading = reading_path(spool);
352 if !reading.exists() && std::fs::rename(spool, &reading).is_err() {
353 return 0; // no spool yet
354 }
355 let text = std::fs::read_to_string(&reading).unwrap_or_default();
356 let _ = std::fs::remove_file(&reading);
357 self.absorb(text.lines().filter_map(parse_flush))
358 }
359
360 /// Fold flushes into per-session state (pure, unit-testable).
361 pub fn absorb(&mut self, flushes: impl IntoIterator<Item = Flush>) -> usize {
362 let mut n = 0;
363 for f in flushes {
364 n += 1;
365 self.sessions
366 .entry(f.session)
367 .and_modify(|s| {
368 s.prev = Some(s.last);
369 s.last = f;
370 })
371 .or_insert(Session { prev: None, last: f });
372 }
373 n
374 }
375
376 /// The aggregates at `now_ms`, dropping sessions idle longer than
377 /// [`ACTIVE_WINDOW_MS`]. Empty if no session is active.
378 pub fn signals(&mut self, now_ms: u64) -> Vec<Signal> {
379 self.sessions
380 .retain(|_, s| now_ms.saturating_sub(s.last.ts_ms) <= ACTIVE_WINDOW_MS);
381 if self.sessions.is_empty() {
382 return Vec::new();
383 }
384 let keys_per_min: f64 = self.sessions.values().map(Session::keys_per_min).sum();
385 let session_s = self.sessions.values().map(|s| s.last.session_s).max().unwrap_or(0);
386 let mk = |name: SignalName, value: f64| Signal {
387 schema_version: SCHEMA_VERSION,
388 ts: now_ms,
389 source: Source::Terminal,
390 name,
391 value: Value(value),
392 tag: None, // terminal aggregates are never tagged
393 };
394 vec![
395 mk(SignalName::KeysPerMin, keys_per_min),
396 mk(SignalName::SessionSeconds, session_s as f64),
397 ]
398 }
399 }
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
442 fn reading_path(spool: &Path) -> PathBuf {
443 let mut p = spool.as_os_str().to_os_string();
444 p.push(".reading");
445 PathBuf::from(p)
446 }
447
448 #[cfg(test)]
449 mod tests {
450 use super::*;
451
452 fn flush(ts_ms: u64, keys: u64, session_s: u64, session: u64) -> Flush {
453 Flush {
454 ts_ms,
455 keys,
456 session_s,
457 session,
458 }
459 }
460
461 #[test]
462 fn parse_requires_exactly_four_integers() {
463 assert_eq!(parse_flush("1000 12 30 4242"), Some(flush(1000, 12, 30, 4242)));
464 assert_eq!(parse_flush("1000 12 30"), None);
465 assert_eq!(parse_flush("1000 12 30 4242 extra"), None);
466 assert_eq!(parse_flush("1000 twelve 30 4242"), None);
467 }
468
469 #[test]
470 fn rates_are_per_session_then_summed() {
471 let mut c = Collector::new();
472 // Shell 1: 60 keys over 30 s = 120/min. Shell 2: 10 keys over
473 // 60 s = 10/min. Interleaved in the spool as they would be.
474 c.absorb([
475 flush(0, 0, 0, 1),
476 flush(0, 0, 0, 2),
477 flush(60_000, 10, 60, 2),
478 flush(30_000, 60, 30, 1),
479 ]);
480 let sigs = c.signals(60_000);
481 let kpm = sigs.iter().find(|s| s.name == SignalName::KeysPerMin).unwrap();
482 assert_eq!(kpm.value, Value(130.0));
483 let ss = sigs.iter().find(|s| s.name == SignalName::SessionSeconds).unwrap();
484 assert_eq!(ss.value, Value(60.0), "longest active session");
485 }
486
487 #[test]
488 fn idle_sessions_age_out() {
489 let mut c = Collector::new();
490 c.absorb([flush(0, 5, 10, 1), flush(1_000, 5, 10, 1)]);
491 assert_eq!(c.signals(1_000).len(), 2);
492 assert!(c.signals(1_000 + ACTIVE_WINDOW_MS + 1).is_empty());
493 }
494
495 #[test]
496 fn ingest_consumes_the_spool() {
497 let dir = std::env::temp_dir().join(format!("signald-spool-{}", std::process::id()));
498 std::fs::create_dir_all(&dir).unwrap();
499 let spool = dir.join("terminal.spool");
500 std::fs::write(&spool, "1000 3 1 7\nnot a record\n2000 4 2 7\n").unwrap();
501
502 let mut c = Collector::new();
503 assert_eq!(c.ingest(&spool), 2);
504 assert!(!spool.exists(), "spool is consumed, not left to grow");
505 assert!(!reading_path(&spool).exists());
506 assert_eq!(c.ingest(&spool), 0, "nothing until the hook appends again");
507
508 // State survives consumption: the rate uses both flushes.
509 let sigs = c.signals(2_000);
510 let kpm = sigs.iter().find(|s| s.name == SignalName::KeysPerMin).unwrap();
511 assert_eq!(kpm.value, Value(240.0)); // 4 keys over 1 s
512 let _ = std::fs::remove_dir_all(&dir);
513 }
514 }
515 }
516
517 /// System + hardware collector: **IOKit-only, no powermetrics, no root**
518 /// Native macOS.
519 ///
520 /// The reads live out of process in the sibling Swift package
521 /// `macos-collector/` (SwiftPM, not in this cargo workspace), which writes
522 /// `signal-schema` wire frames to stdout. This module spawns it and publishes
523 /// every frame it emits, so hardware signals reach the hub, the history
524 /// store, and subscribers by the same path as every other collector. Frames
525 /// are decoded by the same `wire::read_frame` the socket uses, so one that
526 /// fails the schema's structural checks (version, tag rule) is dropped at
527 /// this boundary rather than published.
528 pub mod hardware {
529 use std::io::{self, Read};
530 use std::path::{Path, PathBuf};
531 use std::process::{Command, Stdio};
532
533 use signal_schema::wire;
534
535 use crate::hub::Hub;
536
537 /// The collector binary's name, looked up on `PATH` when no explicit
538 /// path is configured.
539 pub const PROGRAM: &str = "macos-collector";
540
541 /// Read frames from `reader` until EOF, publishing each into `hub`.
542 /// Returns the number of frames published. A frame this build cannot
543 /// decode is skipped and not published; only a stream that can no
544 /// longer be framed ends with an error.
545 pub fn ingest(reader: &mut impl Read, hub: &Hub) -> io::Result<usize> {
546 let mut n = 0;
547 loop {
548 match wire::read_frame(reader)? {
549 wire::Frame::Signal(sig) => {
550 hub.publish(sig);
551 n += 1;
552 }
553 wire::Frame::Skipped => {}
554 wire::Frame::Eof => break,
555 }
556 }
557 Ok(n)
558 }
559
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> {
576 let mut child = Command::new(program)
577 .arg("--interval-ms")
578 .arg(interval_ms.to_string())
579 .stdin(Stdio::null())
580 .stdout(Stdio::piped())
581 .stderr(Stdio::inherit())
582 .spawn()?;
583 let mut stdout = child.stdout.take().expect("stdout is piped");
584 on_spawn();
585 let result = ingest(&mut stdout, hub);
586 let _ = child.kill();
587 let _ = child.wait();
588 result
589 }
590
591 /// Find [`PROGRAM`] on `PATH`.
592 pub fn find_on_path() -> Option<PathBuf> {
593 let path = std::env::var_os("PATH")?;
594 std::env::split_paths(&path)
595 .map(|dir| dir.join(PROGRAM))
596 .find(|p| p.is_file())
597 }
598 }
599}
600
601/// The publish side: a live pub/sub fan-out with a last-value cache.
602///
603/// On connect a subscriber is handed the current value of every cached `name`
604/// immediately (so a renderer paints correct state at once), then streams
605/// updates as they change. Each subscriber runs on its own thread with its own
606/// unbounded channel, so a slow subscriber never blocks the daemon: one stuck
607/// renderer must not stall any other subscriber's feed.
608pub mod publish {
609 use std::io;
610 use std::io::Write;
611 use std::os::unix::net::{UnixListener, UnixStream};
612 use std::path::Path;
613 use std::thread;
614
615 use signal_schema::wire;
616
617 use crate::hub::Hub;
618
619 /// Bind `socket_path` and serve the live stream from `hub` to each
620 /// subscriber. Blocks, accepting connections until the listener errors.
621 pub fn serve(socket_path: &Path, hub: Hub) -> io::Result<()> {
622 // Clear any stale socket file from a previous run.
623 let _ = std::fs::remove_file(socket_path);
624 if let Some(parent) = socket_path.parent() {
625 std::fs::create_dir_all(parent)?;
626 }
627 let listener = UnixListener::bind(socket_path)?;
628 eprintln!("signald: listening on {}", socket_path.display());
629
630 for stream in listener.incoming() {
631 match stream {
632 Ok(stream) => {
633 let hub = hub.clone();
634 // One subscriber per thread; a misbehaving one cannot take
635 // the daemon or the other subscribers down.
636 thread::spawn(move || {
637 if let Err(e) = serve_subscriber(stream, hub) {
638 eprintln!("signald: subscriber dropped: {e}");
639 }
640 });
641 }
642 Err(e) => eprintln!("signald: accept error: {e}"),
643 }
644 }
645 Ok(())
646 }
647
648 /// Replay the last-value cache to a freshly connected subscriber, then
649 /// stream live updates until it disconnects.
650 fn serve_subscriber(mut stream: UnixStream, hub: Hub) -> io::Result<()> {
651 let (snapshot, rx) = hub.subscribe();
652 for sig in snapshot {
653 wire::write_frame(&mut stream, &sig)?;
654 }
655 stream.flush()?;
656 // `rx` yields every signal published after we subscribed. A write error
657 // means the subscriber went away; returning drops `rx`, and the hub
658 // prunes the dead sender on its next publish.
659 for sig in rx.iter() {
660 wire::write_frame(&mut stream, &sig)?;
661 stream.flush()?;
662 }
663 Ok(())
664 }
665}
666
667/// Unix millis, best-effort (spec: monotonic-corrected later).
668pub fn now_millis() -> u64 {
669 SystemTime::now()
670 .duration_since(UNIX_EPOCH)
671 .map(|d| d.as_millis() as u64)
672 .unwrap_or(0)
673}
674
675#[cfg(test)]
676mod root_tests {
677 use super::*;
678 use signal_schema::{SignalName, Source, Tag, Value, SCHEMA_VERSION};
679
680 fn tagged(path: &str) -> Signal {
681 Signal {
682 schema_version: SCHEMA_VERSION,
683 ts: 0,
684 source: Source::Git,
685 name: SignalName::CommitsToday,
686 value: Value(1.0),
687 tag: Tag::repo_path(path),
688 }
689 }
690
691 #[test]
692 fn a_tag_under_a_watched_root_is_allowed() {
693 let roots = vec![PathBuf::from("/home/x/git")];
694 assert!(tag_within_roots(&tagged("/home/x/git/repo"), &roots));
695 assert!(tag_within_roots(&tagged("/home/x/git"), &roots));
696 }
697
698 #[test]
699 fn a_tag_outside_every_watched_root_is_refused() {
700 let roots = vec![PathBuf::from("/home/x/git")];
701 assert!(!tag_within_roots(&tagged("/etc/passwd"), &roots));
702 // Component-wise, so a shared string prefix is not a shared path.
703 assert!(!tag_within_roots(&tagged("/home/x/gitsecrets"), &roots));
704 }
705
706 #[test]
707 fn an_untagged_signal_needs_no_root() {
708 let mut s = tagged("/home/x/git/repo");
709 s.tag = None;
710 assert!(tag_within_roots(&s, &[]));
711 }
712}