Commit `6e92f133db`

6e92f133dbb18b6615861c8d47859626fd37a6ab

parent: e8edd54d8f

Verified · cmc

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

Add CollectorUp and key the cache by source

SignalName::CollectorUp at discriminant 11; SCHEMA_VERSION is 5, the 1.0
contract. Value is 1.0 up, 0.0 down, matching Charging. Untagged: allows_tag
stays the four git names.

The hub keyed its last-value cache by (name, tag), so two signals differing
only by source overwrote each other. CollectorUp is published once per Source,
which makes that latent flaw load-bearing, so the key is now (name, source,
tag). terminal-garden's latest map keys the same way.

Nothing observable regresses: no existing name/source pair collided, so a
subscriber sees the same entries it saw before.

Swift SCHEMA_VERSION, both canonical frames and the macos-collector byte table
move to v5. The collector emits no health of its own — the daemon's supervisor
publishes it — so the Swift change is version-only.

Closes #8

crates/signal-schema/src/lib.rs +25 −10

		@@ -32,12 +32,14 @@
32	32	/// v3 (0.3.0): added the aggregate [`SignalName::CpuLoad`] emitted by the macOS
33	33	/// IOKit hardware collector (`macos-collector/`, a sibling Swift package).
34	34	///
35		/// v4 (0.5.0) is the 1.0 contract. [`SignalName`] was cut to the eleven
36		/// metrics that have a producer and its discriminants renumbered from zero.
37		/// From 1.0 onward discriminants are append-only and removing one is a
38		/// breaking change, so this is the last version in which renumbering is
39		/// possible.
40		pub const SCHEMA_VERSION: u16 = 4;
	35	/// v4 (0.5.0): [`SignalName`] was cut to the metrics that have a producer and
	36	/// its discriminants renumbered from zero — the last version in which
	37	/// renumbering was possible.
	38	///
	39	/// v5 (1.0.0) is the 1.0 contract. Appends [`SignalName::CollectorUp`], the
	40	/// daemon's own health. From 1.0 onward discriminants are append-only and
	41	/// removing one is a breaking change.
	42	pub const SCHEMA_VERSION: u16 = 5;
41	43
42	44	/// The collector domain a signal originated from.
43	45	///
		@@ -108,6 +110,15 @@ pub enum SignalName {
108	110	/// Aggregate CPU busy fraction across all cores in `[0.0, 1.0]`, from Mach
109	111	/// `host_processor_info` tick deltas.
110	112	CpuLoad,
	113	// --- the daemon's own health (appended in v5) ---
	114	/// Whether the collector named by [`Signal::source`] is running: `1.0` up,
	115	/// `0.0` down. Published by the daemon's supervisor on every state change,
	116	/// and once at startup so a late subscriber reads the truth from the
	117	/// last-value cache.
	118	///
	119	/// Liveness is not freshness. A collector can be up and still stalled; a
	120	/// reader that cares should also look at how old [`Signal::ts`] is.
	121	CollectorUp,
111	122	}
112	123
113	124	impl SignalName {
		@@ -141,6 +152,7 @@ impl SignalName {
141	152	SignalName::ThermalState => 8,
142	153	SignalName::BatteryDrawW => 9,
143	154	SignalName::CpuLoad => 10,
	155	SignalName::CollectorUp => 11,
144	156	}
145	157	}
146	158
		@@ -159,6 +171,7 @@ impl SignalName {
159	171	8 => SignalName::ThermalState,
160	172	9 => SignalName::BatteryDrawW,
161	173	10 => SignalName::CpuLoad,
	174	11 => SignalName::CollectorUp,
162	175	_ => return None,
163	176	})
164	177	}
		@@ -490,12 +503,12 @@ mod wire_tests {
490	503	assert!(wire::decode(&[0, 1, 2]).is_none());
491	504	}
492	505
493		/// The v4 contract: eleven names, discriminants 0..=10, each round-tripping
	506	/// The v5 contract: twelve names, discriminants 0..=11, each round-tripping
494	507	/// through the wire byte. Adding a variant without a discriminant, or
495	508	/// reusing one, fails here.
496	509	#[test]
497		fn v4_names_are_exactly_zero_through_ten() {
498		const NAMES: [SignalName; 11] = [
	510	fn v5_names_are_exactly_zero_through_eleven() {
	511	const NAMES: [SignalName; 12] = [
499	512	SignalName::KeysPerMin,
500	513	SignalName::SessionSeconds,
501	514	SignalName::CommitsWindow,
		@@ -507,12 +520,13 @@ mod wire_tests {
507	520	SignalName::ThermalState,
508	521	SignalName::BatteryDrawW,
509	522	SignalName::CpuLoad,
	523	SignalName::CollectorUp,
510	524	];
511	525	for (i, name) in NAMES.iter().enumerate() {
512	526	assert_eq!(name.to_u8(), i as u8, "{name:?} discriminant");
513	527	assert_eq!(SignalName::from_u8(i as u8), Some(*name));
514	528	}
515		assert_eq!(SignalName::from_u8(11), None, "11 is past the frozen set");
	529	assert_eq!(SignalName::from_u8(12), None, "12 is past the frozen set");
516	530	}
517	531
518	532	/// Only the four git metrics may carry a tag in v4.
		@@ -534,6 +548,7 @@ mod wire_tests {
534	548	SignalName::ThermalState,
535	549	SignalName::BatteryDrawW,
536	550	SignalName::CpuLoad,
	551	SignalName::CollectorUp,
537	552	] {
538	553	assert!(!name.allows_tag(), "{name:?} must not allow a tag");
539	554	}

crates/signal-schema/tests/hardware_wire.rs +4 −4

		@@ -1,4 +1,4 @@
1		//! # The Swift↔Rust hardware wire contract (v4)
	1	//! # The Swift↔Rust hardware wire contract (v5)
2	2	//!
3	3	//! The macOS IOKit collector is a sibling Swift package (`macos-collector/`)
4	4	//! that speaks this crate's wire format. Swift and Rust are two independent
		@@ -26,13 +26,13 @@ use signal_schema::{wire, Signal, SignalName, Source, Value, SCHEMA_VERSION};
26	26
27	27	/// The canonical hardware test vector, shared verbatim with the Swift side.
28	28	///
29		/// Signal: schema_version=4, ts=0, source=Hardware(3), name=CpuLoad(10),
	29	/// Signal: schema_version=5, ts=0, source=Hardware(3), name=CpuLoad(10),
30	30	/// value=0.5 (`f64`), tag=None.
31	31	///
32	32	/// Frame bytes (little-endian throughout):
33	33	/// ```text
34	34	/// 15 00 00 00 body_len = 21 (u32)
35		/// 04 00 schema_version = 4 (u16)
	35	/// 05 00 schema_version = 5 (u16)
36	36	/// 00 00 00 00 00 00 00 00 ts = 0 (u64)
37	37	/// 03 source = Hardware (u8)
38	38	/// 0A name = CpuLoad = 10 (u8)
		@@ -41,7 +41,7 @@ use signal_schema::{wire, Signal, SignalName, Source, Value, SCHEMA_VERSION};
41	41	/// ```
42	42	const CANONICAL_FRAME: [u8; 25] = [
43	43	0x15, 0x00, 0x00, 0x00, // body_len = 21
44		0x04, 0x00, // schema_version = 4
	44	0x05, 0x00, // schema_version = 5
45	45	0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // ts = 0
46	46	0x03, // source = Hardware
47	47	0x0A, // name = CpuLoad (10)

crates/signald/src/hub.rs +35 −7

		@@ -22,10 +22,13 @@ use signal_schema::Signal;
22	22
23	23	use crate::history::History;
24	24
25		/// Cache key: metric name discriminant + optional audited tag. Distinct
26		/// per-repo / per-core signals coexist; a new value for the same key replaces
27		/// the old one (keep-latest).
28		type Key = (u8, Option<String>);
	25	/// Cache key: metric name discriminant, source discriminant, and optional
	26	/// audited tag. Distinct per-repo signals coexist, and so do two signals that
	27	/// share a name but come from different collectors — `SignalName::CollectorUp`
	28	/// is published once per `Source`, and without `source` in the key those
	29	/// would overwrite each other. A new value for the same key replaces the old
	30	/// one (keep-latest).
	31	type Key = (u8, u8, Option<String>);
29	32
30	33	/// A cloneable handle to the shared hub. Clones share one cache + subscriber
31	34	/// set behind a mutex.
		@@ -77,7 +80,11 @@ impl Hub {
77	80	// A history write failing must not stop the live stream.
78	81	let _ = h.record(&s);
79	82	}
80		let key = (s.name.to_u8(), s.tag.as_ref().map(\|t\| t.as_str().to_string()));
	83	let key = (
	84	s.name.to_u8(),
	85	s.source.to_u8(),
	86	s.tag.as_ref().map(\|t\| t.as_str().to_string()),
	87	);
81	88	cache.insert(key, s.clone());
82	89	subs.retain(\|tx\| tx.send(s.clone()).is_ok());
83	90	}
		@@ -138,12 +145,33 @@ mod tests {
138	145	}
139	146
140	147	#[test]
141		fn cache_keeps_latest_per_name() {
	148	fn cache_keeps_latest_per_name_and_source() {
142	149	let hub = Hub::new();
143	150	hub.publish(sig(SignalName::KeysPerMin, 1.0));
144	151	hub.publish(sig(SignalName::KeysPerMin, 2.0));
145	152	let snap = hub.snapshot();
146		assert_eq!(snap.len(), 1, "same name collapses to keep-latest");
	153	assert_eq!(snap.len(), 1, "same name and source collapses to keep-latest");
147	154	assert_eq!(snap[0].value, Value(2.0));
148	155	}
	156
	157	/// `CollectorUp` is published once per source, so the cache must not treat
	158	/// the name alone as the identity. Without `source` in the key the second
	159	/// publish would evict the first and a subscriber would see one collector's
	160	/// health standing in for all three.
	161	#[test]
	162	fn same_name_from_two_sources_coexists() {
	163	let hub = Hub::new();
	164	for (source, value) in [(Source::Git, 1.0), (Source::Hardware, 0.0)] {
	165	hub.publish(Signal {
	166	source,
	167	..sig(SignalName::CollectorUp, value)
	168	});
	169	}
	170	let snap = hub.snapshot();
	171	assert_eq!(snap.len(), 2, "one entry per source");
	172	let git = snap.iter().find(\|s\| s.source == Source::Git).expect("git health");
	173	let hw = snap.iter().find(\|s\| s.source == Source::Hardware).expect("hardware health");
	174	assert_eq!(git.value, Value(1.0));
	175	assert_eq!(hw.value, Value(0.0));
	176	}
149	177	}

crates/terminal-garden/src/main.rs +9 −4

		@@ -52,9 +52,10 @@ fn run(socket: &PathBuf) -> std::io::Result<()> {
52	52	let stream = UnixStream::connect(socket)?;
53	53	let mut reader = BufReader::new(stream);
54	54
55		// Keyed by (metric name, repo tag) so per-repo signals coexist and updates
56		// replace prior values rather than accumulating.
57		let mut latest: BTreeMap<(u8, Option<String>), Signal> = BTreeMap::new();
	55	// Keyed by (metric name, source, repo tag) so per-repo signals coexist,
	56	// two collectors reporting the same name stay distinct, and updates replace
	57	// prior values rather than accumulating.
	58	let mut latest: BTreeMap<(u8, u8, Option<String>), Signal> = BTreeMap::new();
58	59	loop {
59	60	let sig = match wire::read_frame(&mut reader)? {
60	61	wire::Frame::Signal(sig) => sig,
		@@ -66,7 +67,11 @@ fn run(socket: &PathBuf) -> std::io::Result<()> {
66	67	if !SUBSCRIBE.contains(&sig.name) {
67	68	continue;
68	69	}
69		let key = (sig.name.to_u8(), sig.tag.as_ref().map(\|t\| t.as_str().to_string()));
	70	let key = (
	71	sig.name.to_u8(),
	72	sig.source.to_u8(),
	73	sig.tag.as_ref().map(\|t\| t.as_str().to_string()),
	74	);
70	75	latest.insert(key, sig);
71	76
72	77	let signals: Vec<Signal> = latest.values().cloned().collect();

macos-collector/README.md +3 −3

		@@ -62,7 +62,7 @@ multi-byte integers and the `f64` (via its IEEE-754 bit pattern) are
62	62	frame:
63	63	[u32 body_len] little-endian length of body
64	64	body:
65		[u16 schema_version] must equal 4 (v4); a reader skips other versions
	65	[u16 schema_version] must equal 5 (v5); a reader skips other versions
66	66	[u64 ts] unix milliseconds
67	67	[u8 source] 0=terminal 1=git 2=macos 3=hardware
68	68	[u8 name] SignalName discriminant (see table above)
		@@ -74,12 +74,12 @@ frame:
74	74
75	75	### Canonical frame (pinned by tests on both sides)
76	76
77		`schema_version=4, ts=0, source=Hardware(3), name=CpuLoad(10), value=0.5,
	77	`schema_version=5, ts=0, source=Hardware(3), name=CpuLoad(10), value=0.5,
78	78	tag=none` encodes to these exact 25 bytes:
79	79
80	80	```text
81	81	15 00 00 00 body_len = 21
82		04 00 schema_version = 4
	82	05 00 schema_version = 5
83	83	00 00 00 00 00 00 00 00 ts = 0
84	84	03 source = Hardware
85	85	0A name = CpuLoad (10)

macos-collector/Sources/CollectorCore/Wire.swift +2 −2

		@@ -1,9 +1,9 @@
1	1	import Foundation
2	2
3	3	/// The wire schema version. Must equal `signal_schema::SCHEMA_VERSION` on the
4		/// Rust side (v4 as of 0.5.0, the 1.0 contract). A reader skips frames whose
	4	/// Rust side (v5 as of 1.0.0, the 1.0 contract). A reader skips frames whose
5	5	/// version it does not understand, so this is a hard cross-language contract.
6		public let SCHEMA_VERSION: UInt16 = 4
	6	public let SCHEMA_VERSION: UInt16 = 5
7	7
8	8	/// The collector domain. Discriminants must match Rust `Source::to_u8`.
9	9	public enum Source: UInt8 {

macos-collector/Tests/CollectorCoreTests/WireTests.swift +2 −2

		@@ -14,7 +14,7 @@ final class WireTests: XCTestCase {
14	14	let signal = Signal(ts: 0, source: .hardware, name: .cpuLoad, value: 0.5)
15	15	let expected: [UInt8] = [
16	16	0x15, 0x00, 0x00, 0x00, // body_len = 21
17		0x04, 0x00, // schema_version = 4
	17	0x05, 0x00, // schema_version = 5
18	18	0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // ts = 0
19	19	0x03, // source = Hardware
20	20	0x0A, // name = CpuLoad (10)
		@@ -27,7 +27,7 @@ final class WireTests: XCTestCase {
27	27	/// The schema version must equal the Rust `SCHEMA_VERSION`, or the daemon
28	28	/// drops every frame this collector sends.
29	29	func testSchemaVersionMatchesRust() {
30		XCTAssertEqual(SCHEMA_VERSION, 4)
	30	XCTAssertEqual(SCHEMA_VERSION, 5)
31	31	}
32	32
33	33	/// The tagged-frame layout matches Rust too (hardware signals are untagged,

Commit 6e92f133db

Commit `6e92f133db`