krz/ambient-companions

Skip frames the build cannot decode instead of killing the stream !6

merged merged by cmc on 2026-09-04 14:23 UTC · krz/ambient-companions:fix/skip-unknown-frames into main

5 files changed, +138 −29

crates/signal-schema/src/lib.rs +92 −13
@@ -338,23 +338,49 @@ pub mod wire {
338338 decode_body(body)
339339 }
340340
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;
345368
346369 let mut len_buf = [0u8; 4];
347370 match r.read_exact(&mut len_buf) {
348371 Ok(()) => {}
349 Err(e) if e.kind() == ErrorKind::UnexpectedEof => return Ok(None),
372 Err(e) if e.kind() == ErrorKind::UnexpectedEof => return Ok(Frame::Eof),
350373 Err(e) => return Err(e),
351374 }
352375 let body_len = u32::from_le_bytes(len_buf) as usize;
353376 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.
354379 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 })
358384 }
359385
360386 /// Encode `s` and write the whole frame to a stream.
@@ -450,12 +476,65 @@ mod wire_tests {
450476 wire::write_frame(&mut buf, &b).unwrap();
451477
452478 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();
455481 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);
459538 }
460539
461540 #[test]
crates/signald/src/lib.rs +14 −7
@@ -354,8 +354,8 @@ pub mod collectors {
354354 /// every frame it emits, so hardware signals reach the hub, the history
355355 /// store, and subscribers by the same path as every other collector. Frames
356356 /// 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.
359359 pub mod hardware {
360360 use std::io::{self, Read};
361361 use std::path::{Path, PathBuf};
@@ -371,13 +371,20 @@ pub mod collectors {
371371 pub const PROGRAM: &str = "macos-collector";
372372
373373 /// 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.
376377 pub fn ingest(reader: &mut impl Read, hub: &Hub) -> io::Result<usize> {
377378 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 }
381388 }
382389 Ok(n)
383390 }
crates/signald/tests/hardware_ingest.rs +16 −5
@@ -65,19 +65,30 @@ fn canned_frames_reach_a_subscriber_snapshot() {
6565}
6666
6767#[test]
68fn frame_that_breaks_the_tag_rule_is_rejected_at_the_boundary() {
68fn frame_that_breaks_the_tag_rule_is_dropped_and_ingest_continues() {
6969 let hub = Hub::new();
7070 let good = sig(Source::Hardware, SignalName::CpuLoad, 0.5);
7171 // CpuLoad never allows a tag. Encode one anyway, bypassing the schema's
7272 // well-formedness check, as a misbehaving collector could.
7373 let mut bad = good.clone();
7474 bad.tag = Some(Tag::repo_path("/not/allowed").unwrap());
75 let after = sig(Source::Macos, SignalName::BatteryPct, 42.0);
76
7577 let mut buf = frames(&[good]);
7678 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");
7783
78 let err = hardware::ingest(&mut Cursor::new(buf), &hub).unwrap_err();
79 assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
8084 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 );
8394}
crates/signald/tests/streaming.rs +6 −2
@@ -42,13 +42,17 @@ fn subscriber_sees_cached_value_then_a_live_update() {
4242 let mut reader = BufReader::new(stream);
4343
4444 // 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 };
4648 assert_eq!(first.name, SignalName::KeysPerMin);
4749 assert_eq!(first.value, Value(12.0));
4850
4951 // 2. A change after subscribing is streamed live.
5052 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 };
5256 assert_eq!(update.name, SignalName::KeysPerMin);
5357 assert_eq!(update.value, Value(99.0));
5458
crates/terminal-garden/src/main.rs +10 −2
@@ -45,7 +45,8 @@ fn main() {
4545/// Connect and render live: the daemon replays the last-value cache on connect,
4646/// then streams updates. We keep the latest value of each metric (keyed by name
4747/// 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.
4950fn run(socket: &PathBuf) -> std::io::Result<()> {
5051 let stream = UnixStream::connect(socket)?;
5152 let mut reader = BufReader::new(stream);
@@ -53,7 +54,14 @@ fn run(socket: &PathBuf) -> std::io::Result<()> {
5354 // Keyed by (metric name, repo tag) so per-repo signals coexist and updates
5455 // replace prior values rather than accumulating.
5556 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 };
5765 if !SUBSCRIBE.contains(&sig.name) {
5866 continue;
5967 }