Commit `e5a591f952`

e5a591f952c55da9e81ea124e88be1795fafc6ea

parent: 88da2814ee

Verified · cmc

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

Skip frames the build cannot decode instead of killing the stream

read_frame returns wire::Frame — Signal, Skipped or Eof — instead of
Option<Signal>. A body that is framed correctly but cannot be decoded (unknown
schema version, unassigned name discriminant, tag-rule violation) is consumed
in full and reported as Skipped, leaving the stream in sync. Err is now only
for a stream that can no longer be framed: a truncated length prefix or a body
shorter than its prefix promised.

terminal-garden and the hardware ingest loop continue past skipped frames, so
a renderer built before a metric was appended no longer exits on the first
frame it does not understand.

Closes #3

crates/signal-schema/src/lib.rs +92 −13

		@@ -338,23 +338,49 @@ pub mod wire {
338	338	decode_body(body)
339	339	}
340	340
341		/// Read exactly one frame from a stream. `Ok(None)` signals a clean EOF at a
342		/// frame boundary; a partial or malformed frame is an error.
343		pub fn read_frame(r: &mut impl std::io::Read) -> std::io::Result<Option<Signal>> {
344		use std::io::{Error, ErrorKind};
	341	/// The outcome of reading one frame.
	342	///
	343	/// A renderer built against an older schema, or before a metric was
	344	/// appended, must not die on the first record it does not recognise. The
	345	/// length prefix makes that possible: the body can be consumed whole and
	346	/// discarded, leaving the stream in sync for the next frame.
	347	#[derive(Debug, Clone, PartialEq)]
	348	pub enum Frame {
	349	/// A record this build understands.
	350	Signal(Signal),
	351	/// A well-framed body this build cannot decode: a schema version it
	352	/// does not know, an unassigned [`SignalName`] discriminant, or a body
	353	/// that breaks the tag rule. The frame was consumed in full and the
	354	/// stream is still in sync, so the caller should carry on reading.
	355	Skipped,
	356	/// Clean EOF at a frame boundary.
	357	Eof,
	358	}
	359
	360	/// Read exactly one frame from a stream.
	361	///
	362	/// `Err` is reserved for a stream that can no longer be framed: a length
	363	/// prefix that ends mid-way, or a body shorter than its prefix promised.
	364	/// A body that is framed correctly but cannot be decoded is
	365	/// [`Frame::Skipped`], not an error.
	366	pub fn read_frame(r: &mut impl std::io::Read) -> std::io::Result<Frame> {
	367	use std::io::ErrorKind;
345	368
346	369	let mut len_buf = [0u8; 4];
347	370	match r.read_exact(&mut len_buf) {
348	371	Ok(()) => {}
349		Err(e) if e.kind() == ErrorKind::UnexpectedEof => return Ok(None),
	372	Err(e) if e.kind() == ErrorKind::UnexpectedEof => return Ok(Frame::Eof),
350	373	Err(e) => return Err(e),
351	374	}
352	375	let body_len = u32::from_le_bytes(len_buf) as usize;
353	376	let mut body = vec![0u8; body_len];
	377	// A short read here means the stream is truncated: the next bytes are
	378	// not a length prefix, so there is no way to resynchronise.
354	379	r.read_exact(&mut body)?;
355		decode_body(&body)
356		.map(Some)
357		.ok_or_else(\|\| Error::new(ErrorKind::InvalidData, "malformed signal frame"))
	380	Ok(match decode_body(&body) {
	381	Some(signal) => Frame::Signal(signal),
	382	None => Frame::Skipped,
	383	})
358	384	}
359	385
360	386	/// Encode `s` and write the whole frame to a stream.
		@@ -450,12 +476,65 @@ mod wire_tests {
450	476	wire::write_frame(&mut buf, &b).unwrap();
451	477
452	478	let mut cursor = std::io::Cursor::new(buf);
453		let ra = wire::read_frame(&mut cursor).unwrap().unwrap();
454		let rb = wire::read_frame(&mut cursor).unwrap().unwrap();
	479	let ra = wire::read_frame(&mut cursor).unwrap();
	480	let rb = wire::read_frame(&mut cursor).unwrap();
455	481	let end = wire::read_frame(&mut cursor).unwrap();
456		assert_eq!(ra, a);
457		assert_eq!(rb, b);
458		assert!(end.is_none(), "clean EOF at frame boundary");
	482	assert_eq!(ra, wire::Frame::Signal(a));
	483	assert_eq!(rb, wire::Frame::Signal(b));
	484	assert_eq!(end, wire::Frame::Eof, "clean EOF at frame boundary");
	485	}
	486
	487	/// Re-frame `bytes` after overwriting one body byte, so the frame stays
	488	/// well-formed at the framing layer but undecodable at the schema layer.
	489	fn frame_with_body_byte(s: &Signal, offset: usize, value: u8) -> Vec<u8> {
	490	let mut frame = wire::encode(s);
	491	frame[4 + offset] = value;
	492	frame
	493	}
	494
	495	#[test]
	496	fn newer_schema_version_is_skipped_not_fatal() {
	497	let good = sig(SignalName::CommitsToday, 7.0, None);
	498	let mut buf = Vec::new();
	499	// schema_version is the first two bytes of the body.
	500	buf.extend_from_slice(&frame_with_body_byte(&good, 0, SCHEMA_VERSION as u8 + 1));
	501	wire::write_frame(&mut buf, &good).unwrap();
	502
	503	let mut cursor = std::io::Cursor::new(buf);
	504	assert_eq!(wire::read_frame(&mut cursor).unwrap(), wire::Frame::Skipped);
	505	assert_eq!(
	506	wire::read_frame(&mut cursor).unwrap(),
	507	wire::Frame::Signal(good),
	508	"the valid frame after an unknown version is still read"
	509	);
	510	}
	511
	512	#[test]
	513	fn unassigned_name_discriminant_is_skipped_not_fatal() {
	514	let good = sig(SignalName::CommitsToday, 7.0, None);
	515	let mut buf = Vec::new();
	516	// name is body byte 11; 200 is not assigned to any variant.
	517	buf.extend_from_slice(&frame_with_body_byte(&good, 11, 200));
	518	wire::write_frame(&mut buf, &good).unwrap();
	519
	520	let mut cursor = std::io::Cursor::new(buf);
	521	assert_eq!(wire::read_frame(&mut cursor).unwrap(), wire::Frame::Skipped);
	522	assert_eq!(
	523	wire::read_frame(&mut cursor).unwrap(),
	524	wire::Frame::Signal(good),
	525	"the valid frame after an unknown name is still read"
	526	);
	527	}
	528
	529	#[test]
	530	fn truncated_body_is_an_error_not_a_skip() {
	531	let good = sig(SignalName::CommitsToday, 7.0, None);
	532	let mut frame = wire::encode(&good);
	533	frame.pop(); // body one byte shorter than its length prefix promises
	534
	535	let mut cursor = std::io::Cursor::new(frame);
	536	let err = wire::read_frame(&mut cursor).expect_err("truncated frame cannot be resynced");
	537	assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);
459	538	}
460	539
461	540	#[test]

crates/signald/src/lib.rs +14 −7

		@@ -354,8 +354,8 @@ pub mod collectors {
354	354	/// every frame it emits, so hardware signals reach the hub, the history
355	355	/// store, and subscribers by the same path as every other collector. Frames
356	356	/// are decoded by the same `wire::read_frame` the socket uses, so one that
357		/// fails the schema's structural checks (version, tag rule) is rejected at
358		/// this boundary.
	357	/// fails the schema's structural checks (version, tag rule) is dropped at
	358	/// this boundary rather than published.
359	359	pub mod hardware {
360	360	use std::io::{self, Read};
361	361	use std::path::{Path, PathBuf};
		@@ -371,13 +371,20 @@ pub mod collectors {
371	371	pub const PROGRAM: &str = "macos-collector";
372	372
373	373	/// Read frames from `reader` until EOF, publishing each into `hub`.
374		/// Returns the number of frames published. A malformed frame ends the
375		/// stream with an error.
	374	/// Returns the number of frames published. A frame this build cannot
	375	/// decode is skipped and not published; only a stream that can no
	376	/// longer be framed ends with an error.
376	377	pub fn ingest(reader: &mut impl Read, hub: &Hub) -> io::Result<usize> {
377	378	let mut n = 0;
378		while let Some(sig) = wire::read_frame(reader)? {
379		hub.publish(sig);
380		n += 1;
	379	loop {
	380	match wire::read_frame(reader)? {
	381	wire::Frame::Signal(sig) => {
	382	hub.publish(sig);
	383	n += 1;
	384	}
	385	wire::Frame::Skipped => {}
	386	wire::Frame::Eof => break,
	387	}
381	388	}
382	389	Ok(n)
383	390	}

crates/signald/tests/hardware_ingest.rs +16 −5

		@@ -65,19 +65,30 @@ fn canned_frames_reach_a_subscriber_snapshot() {
65	65	}
66	66
67	67	#[test]
68		fn frame_that_breaks_the_tag_rule_is_rejected_at_the_boundary() {
	68	fn frame_that_breaks_the_tag_rule_is_dropped_and_ingest_continues() {
69	69	let hub = Hub::new();
70	70	let good = sig(Source::Hardware, SignalName::CpuLoad, 0.5);
71	71	// CpuLoad never allows a tag. Encode one anyway, bypassing the schema's
72	72	// well-formedness check, as a misbehaving collector could.
73	73	let mut bad = good.clone();
74	74	bad.tag = Some(Tag::repo_path("/not/allowed").unwrap());
	75	let after = sig(Source::Macos, SignalName::BatteryPct, 42.0);
	76
75	77	let mut buf = frames(&[good]);
76	78	buf.extend(wire::encode(&bad));
	79	buf.extend(frames(&[after]));
	80
	81	let n = hardware::ingest(&mut Cursor::new(buf), &hub).unwrap();
	82	assert_eq!(n, 2, "the tagged frame is dropped, the other two published");
77	83
78		let err = hardware::ingest(&mut Cursor::new(buf), &hub).unwrap_err();
79		assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
80	84	let snapshot = hub.snapshot();
81		assert_eq!(snapshot.len(), 1, "the good frame before it was published");
82		assert!(snapshot[0].tag.is_none());
	85	assert_eq!(snapshot.len(), 2);
	86	assert!(
	87	snapshot.iter().all(\|s\| s.tag.is_none()),
	88	"no tagged frame reaches the hub"
	89	);
	90	assert!(
	91	snapshot.iter().any(\|s\| s.name == SignalName::BatteryPct),
	92	"ingest kept reading past the rejected frame"
	93	);
83	94	}

crates/signald/tests/streaming.rs +6 −2

		@@ -42,13 +42,17 @@ fn subscriber_sees_cached_value_then_a_live_update() {
42	42	let mut reader = BufReader::new(stream);
43	43
44	44	// 1. On connect: the cached snapshot arrives immediately.
45		let first = wire::read_frame(&mut reader).unwrap().expect("snapshot frame");
	45	let wire::Frame::Signal(first) = wire::read_frame(&mut reader).unwrap() else {
	46	panic!("expected a snapshot frame");
	47	};
46	48	assert_eq!(first.name, SignalName::KeysPerMin);
47	49	assert_eq!(first.value, Value(12.0));
48	50
49	51	// 2. A change after subscribing is streamed live.
50	52	hub.publish(sig(SignalName::KeysPerMin, 99.0));
51		let update = wire::read_frame(&mut reader).unwrap().expect("live update frame");
	53	let wire::Frame::Signal(update) = wire::read_frame(&mut reader).unwrap() else {
	54	panic!("expected a live update frame");
	55	};
52	56	assert_eq!(update.name, SignalName::KeysPerMin);
53	57	assert_eq!(update.value, Value(99.0));
54	58

crates/terminal-garden/src/main.rs +10 −2

		@@ -45,7 +45,8 @@ fn main() {
45	45	/// Connect and render live: the daemon replays the last-value cache on connect,
46	46	/// then streams updates. We keep the latest value of each metric (keyed by name
47	47	/// and repo tag) and re-render the garden on every frame. `Ok(())` is a clean EOF
48		/// (the daemon closed the stream).
	48	/// (the daemon closed the stream). Frames this build cannot decode are skipped,
	49	/// so an older renderer keeps working against a newer daemon.
49	50	fn run(socket: &PathBuf) -> std::io::Result<()> {
50	51	let stream = UnixStream::connect(socket)?;
51	52	let mut reader = BufReader::new(stream);
		@@ -53,7 +54,14 @@ fn run(socket: &PathBuf) -> std::io::Result<()> {
53	54	// Keyed by (metric name, repo tag) so per-repo signals coexist and updates
54	55	// replace prior values rather than accumulating.
55	56	let mut latest: BTreeMap<(u8, Option<String>), Signal> = BTreeMap::new();
56		while let Some(sig) = wire::read_frame(&mut reader)? {
	57	loop {
	58	let sig = match wire::read_frame(&mut reader)? {
	59	wire::Frame::Signal(sig) => sig,
	60	// A record this build does not understand: a newer daemon, or a
	61	// metric appended after this renderer was built. Keep rendering.
	62	wire::Frame::Skipped => continue,
	63	wire::Frame::Eof => break,
	64	};
57	65	if !SUBSCRIBE.contains(&sig.name) {
58	66	continue;
59	67	}

Commit e5a591f952

Commit `e5a591f952`