From d74a66e4d58628c8aa575f1c3030b852b120d045 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 28 Aug 2026 08:45:44 -0700 Subject: [PATCH] fix(json): end a stream track when a record cannot be written `moq_json::stream::Producer::append` recovered from a failed frame write by closing the group and letting the next append open a fresh one. That contradicts the format: a `stream` is a single group, so the roll yields a log with a hole in it, and a subscriber joining afterwards starts at the latest group and never sees the records before it. Any failure now ends the track, matching `moq_binary::stream::Producer`. A write into the live group aborts it, and so does a record that never encodes: nothing was published either way, but the log is missing a record all the same. Opening the group stays the one recoverable case, and only because a failure there means the track had already ended. Both write paths reset the encoder, so the desync latch doesn't answer the next append ahead of the track. Both crates abort through a cloned track handle rather than consuming their own, so `consume` keeps returning a plain `Subscriber` the way every other producer in the workspace does. A subscriber taken after the abort surfaces it on the first read, which is what tells a late reader the log is truncated. `moq_mux::json::Stream::append` drops its workaround and simply retires the catalog entry, which is now correct for every error rather than only the ones that reached the wire. Co-Authored-By: Claude Opus 5 --- rs/moq-binary/src/stream/producer.rs | 48 ++++++-------- rs/moq-json/src/stream/mod.rs | 65 +++++++++++++----- rs/moq-json/src/stream/producer.rs | 98 ++++++++++++++++++++-------- rs/moq-mux/src/binary.rs | 9 ++- rs/moq-mux/src/json.rs | 53 +++++++++++---- 5 files changed, 186 insertions(+), 87 deletions(-) diff --git a/rs/moq-binary/src/stream/producer.rs b/rs/moq-binary/src/stream/producer.rs index 3314630401..1f6aa9ae65 100644 --- a/rs/moq-binary/src/stream/producer.rs +++ b/rs/moq-binary/src/stream/producer.rs @@ -44,24 +44,19 @@ impl Producer { pub fn new(track: moq_net::track::Producer, config: ProducerConfig) -> Self { Self { inner: Arc::new(Mutex::new(Inner { - track: Some(track), + track, group: None, flate: config.compression.then(moq_flate::Encoder::new), })), } } - /// Create a subscriber for the underlying track, or `None` once a failed write has aborted it. + /// Create a subscriber for the underlying track. /// - /// A cleanly finished track still yields a subscriber: the log is complete and readable. Only an - /// abort takes the track away, and at that point there is nothing coherent to subscribe to. - pub fn consume(&self) -> Option { - self.inner - .lock() - .unwrap() - .track - .as_ref() - .map(|track| track.subscribe(None)) + /// Still hands one back once a failed write has ended the log: the subscriber surfaces the abort + /// on its first read, which is what tells a late reader the log is truncated. + pub fn consume(&self) -> moq_net::track::Subscriber { + self.inner.lock().unwrap().track.subscribe(None) } /// Whether any consumer for the underlying track currently exists. @@ -69,11 +64,12 @@ impl Producer { /// The demand signal for a producer serving on request: an unused track is cached state nobody is /// watching, safe to drop and recreate on the next request. pub fn is_used(&self) -> bool { - match self.inner.lock().unwrap().track.as_mut() { - Some(track) => track.poll_unused(&kio::Waiter::noop()).is_pending(), - // An aborted track has no readers and can never gain one. - None => false, - } + self.inner + .lock() + .unwrap() + .track + .poll_unused(&kio::Waiter::noop()) + .is_pending() } /// Append one payload to the log. @@ -94,9 +90,7 @@ impl Producer { /// Shared publishing state behind [`Producer`]'s `Arc`. struct Inner { - /// `None` once a failed write has aborted the track, which is terminal. A clean finish keeps it, - /// since a completed log is still readable. - track: Option, + track: moq_net::track::Producer, /// Opened on the first append and never rolled. group: Option, @@ -119,8 +113,7 @@ impl Inner { // Open the group before compressing: a failure here must not leave the window ahead of a // consumer that never received the frame. if self.group.is_none() { - let track = self.track.as_mut().ok_or(moq_net::Error::Cancel)?; - self.group = Some(track.append_group()?); + self.group = Some(self.track.append_group()?); } let payload = match self.flate.as_mut() { @@ -151,9 +144,11 @@ impl Inner { fn abort(&mut self, err: moq_net::Error) { // The track abort closes its groups, so the handle only needs dropping. self.group = None; - if let Some(track) = self.track.take() { - let _ = track.abort(err); - } + + // Abort through a clone, since aborting consumes a handle and the state is shared. Keeping + // ours means `consume` still hands back a subscriber, which is how a reader learns the log + // ended badly rather than cleanly. + let _ = self.track.clone().abort(err); } fn finish(&mut self) -> Result<()> { @@ -165,10 +160,7 @@ impl Inner { Some(mut group) => group.finish(), None => Ok(()), }; - let track = match self.track.as_mut() { - Some(track) => track.finish(), - None => Ok(()), - }; + let track = self.track.finish(); group?; track?; diff --git a/rs/moq-json/src/stream/mod.rs b/rs/moq-json/src/stream/mod.rs index 06dcb1e609..58fb6c0a07 100644 --- a/rs/moq-json/src/stream/mod.rs +++ b/rs/moq-json/src/stream/mod.rs @@ -12,6 +12,10 @@ //! the source (e.g. the timeline's granularity); a consumer that finds a gap can fetch or //! extrapolate. //! +//! A record that cannot be encoded or written therefore ends the track rather than continuing in a +//! second group: a log missing a record is not lossless, and a gap dressed up as a complete log is +//! worse than a visible failure. A publisher with more to say opens a new track. +//! //! That single group is what bounds the log's history. moq-net caps a group's cached bytes, and a //! consumer always starts at frame 0, so once the log outgrows that budget and the earliest frames //! are evicted a new consumer fails with [`moq_net::Error::Lagged`] rather than reading a partial @@ -151,7 +155,8 @@ mod test { } /// A record the encoder rejects must not have published a group first: a live consumer would - /// advance into it and wait there even though nothing was ever appended. + /// advance into it and wait there even though nothing was ever appended. It still ends the + /// track, since the log is missing the record either way. #[test] fn a_rejected_record_does_not_open_a_group() { // A map with non-string keys can't be represented as JSON, so serialization fails. @@ -159,7 +164,7 @@ mod test { .produce() .create_track("test", None) .unwrap(); - let subscriber = track.subscribe(None); + let mut subscriber = track.subscribe(None); let mut producer = Producer::>::new(track, ProducerConfig::default()); let mut bad = std::collections::BTreeMap::new(); @@ -167,6 +172,12 @@ mod test { assert!(producer.append(&bad).is_err()); assert_eq!(subscriber.latest(), None, "a rejected record opened a group"); + + let waiter = kio::Waiter::noop(); + assert!( + matches!(subscriber.poll_recv_group(&waiter), Poll::Ready(Err(_))), + "the log is missing a record, so the track must end rather than stay writable" + ); } /// A track whose timescale is extreme enough that converting a wall-clock timestamp into it @@ -183,42 +194,60 @@ mod test { .unwrap() } - /// Same as the snapshot case: the log's group is published by `open`, so a record the track - /// rejects must not leave it open with nothing in it. + /// A failed write must reach the consumer, not just the caller. A clean close drains a reader to + /// `None`, which is exactly what a completed log looks like, so a truncated log would be + /// indistinguishable from a whole one. #[test] - fn a_rejected_record_does_not_strand_an_empty_group() { + fn a_failed_write_aborts_the_track() { let track = rejecting_track(); let mut subscriber = track.subscribe(None); let mut producer = Producer::::new(track, ProducerConfig::default()); - assert!(producer.append(&json!({ "n": 1 })).is_err()); + assert!(matches!(producer.append(&json!({ "n": 1 })), Err(crate::Error::Net(_)))); let waiter = kio::Waiter::noop(); - let Poll::Ready(Ok(Some(mut group))) = subscriber.poll_next_group(&waiter) else { - panic!("the group was published, so a subscriber sees it"); - }; assert!( - matches!(group.poll_read_frame(&waiter), Poll::Ready(Ok(None))), - "the empty group must be closed, not left open for a subscriber to wait in" + matches!(subscriber.poll_recv_group(&waiter), Poll::Ready(Err(_))), + "a truncated log must surface an error rather than read as a completed one" ); } - /// Closing the rejected group is only half the recovery. The record that never landed desyncs a - /// compressed encoder, so without a matching reset every later append fails with - /// [`Error::Desync`](crate::Error::Desync) before it can use the fresh group that closing prepared. + /// The track ends with the group, so nothing opens a second one and splits the log. The retry + /// reports the ended track rather than the [`Error::Desync`](crate::Error::Desync) the dropped + /// record left on the encoder, which says nothing about why the log stopped. #[test] - fn a_rejected_record_leaves_the_encoder_able_to_retry() { + fn a_failed_write_ends_the_track() { let track = rejecting_track(); let mut producer = Producer::::new(track, ProducerConfig::default().with_compression(true)); assert!(matches!(producer.append(&json!({ "n": 1 })), Err(crate::Error::Net(_)))); - // The retry fails on the same track, but it has to fail for the same reason: a desync here - // would mean the producer had latched itself shut instead of starting a new group. + // The retry reports the abort rather than the `Error::Desync` the dropped record left on the + // encoder, which says nothing about why the log stopped. assert!( matches!(producer.append(&json!({ "n": 2 })), Err(crate::Error::Net(_))), - "the encoder latched a desync instead of retrying into a fresh group" + "a second append must fail on the ended track rather than open another group" ); + + // A subscriber taken after the abort still exists; it surfaces the failure on its first read, + // which is how a late reader learns the log is truncated. + let waiter = kio::Waiter::noop(); + assert!(matches!( + producer.consume().poll_recv_group(&waiter), + Poll::Ready(Err(_)) + )); + } + + /// A completed log is still readable, so finishing must not end the track the way an abort does. + /// The append that follows fails on the closed track without turning it into a failure. + #[test] + fn appending_after_finish_fails_without_aborting() { + let (mut producer, _track) = producer(compressed()); + producer.append(&json!({ "n": 0 })).unwrap(); + producer.finish().unwrap(); + + assert!(producer.append(&json!({ "n": 1 })).is_err()); + assert_eq!(drain(consumer(producer.consume(), true)), vec![json!({ "n": 0 })]); } #[test] diff --git a/rs/moq-json/src/stream/producer.rs b/rs/moq-json/src/stream/producer.rs index b8b31667d5..eb72665a57 100644 --- a/rs/moq-json/src/stream/producer.rs +++ b/rs/moq-json/src/stream/producer.rs @@ -31,6 +31,9 @@ impl Clone for Producer { impl Producer { /// Create a subscriber for the underlying track. + /// + /// Still hands one back once a failed write has ended the log: the subscriber surfaces the abort + /// on its first read, which is what tells a late reader the log is truncated. pub fn consume(&self) -> moq_net::track::Subscriber { self.inner.lock().unwrap().track.inner.subscribe(None) } @@ -52,6 +55,12 @@ impl Producer { } /// Append one record to the log. + /// + /// Any failure ends the track. A log missing a record is not the lossless log a stream promises, + /// and that holds whether the group rejected the record or it never encoded at all, so the + /// failure is surfaced rather than papered over with a second group. The track is aborted rather + /// than closed cleanly, so a consumer sees the failure instead of a log that merely looks + /// complete. Every later append fails on the ended track. pub fn append(&mut self, value: &T) -> Result<()> { self.inner.lock().unwrap().append(value) } @@ -80,21 +89,44 @@ impl Inner { // Encode first, so a value that can't be serialized doesn't publish an empty group that // subscribers would advance into and wait on. Opening the group afterwards is safe because // `record` guards the window: any failure below drops it uncommitted. - let record = encoder.encode(value)?; + let record = match encoder.encode(value) { + Ok(record) => record, + Err(err) => { + // A record that can't be encoded is as lost as one the group rejects: the log is + // missing it either way, and carrying on would present that gap as a complete log. + // Nothing was published, so this only has to end the track. + track.abort(moq_net::Error::Cancel); + return Err(err); + } + }; - let result = match track.open() { + let opened = track.open(); + let published = opened.is_ok(); + let result = match opened { Ok(()) => track.write(record.payload()), Err(err) => Err(err), }; if let Err(err) = result { - // The record never reached the wire, so dropping it desyncs a compressed encoder. `Track` - // has already closed the group it published, which is the group roll that recovery needs; - // reset the encoder to finish it, or the desync latch refuses every later record even - // though the fresh group could carry one. + // The record never reached the wire either way, so the window is ahead of every consumer + // and has to be reset. Without this the desync latch answers the next append before the + // track does, masking the real reason the log stopped. drop(record); encoder.reset(); - return Err(err); + + // What differs is whether a consumer could have seen the group. A write failure means the + // group is already live, so the record is a hole in the log, and a second group would hand + // consumers that gap dressed up as a complete log. End the track, which is what keeps "a + // stream is one group" a real invariant rather than the usual case. + // + // An `open` failure published nothing (it only runs when there is no group), so a later + // append opens a fresh group whose decoder starts cold. That path also catches an append + // onto a track already ended this way, which keeps reporting the error it was aborted with. + if published { + track.abort(err.clone()); + } + + return Err(err.into()); } record.commit(); @@ -102,20 +134,21 @@ impl Inner { } fn finish(&mut self) -> Result<()> { - self.track.finish() + Ok(self.track.finish()?) } } /// The track half of [`Inner`]: the single group carrying the whole log. struct Track { inner: moq_net::track::Producer, - // Opened on the first append and never rolled. + + /// Opened on the first append and never rolled. group: Option, } impl Track { /// Open the log's group if it isn't already. - fn open(&mut self) -> Result<()> { + fn open(&mut self) -> std::result::Result<(), moq_net::Error> { if self.group.is_none() { self.group = Some(self.inner.append_group()?); } @@ -123,26 +156,39 @@ impl Track { } /// Append one encoded record to the log's group. - fn write(&mut self, payload: &bytes::Bytes) -> Result<()> { + fn write(&mut self, payload: &bytes::Bytes) -> std::result::Result<(), moq_net::Error> { let group = self.group.as_mut().expect("a group is open"); - let Err(err) = group.write_frame(moq_net::Timestamp::now(), payload.clone()) else { - return Ok(()); - }; + group.write_frame(moq_net::Timestamp::now(), payload.clone()) + } - // The group is already published and dropping the handle does not close it, so a subscriber - // that advanced into it would wait there with nothing to read. Close it and let a later append - // open a fresh one, which is what a caller recovering from the desync has to do anyway. - if let Some(mut group) = self.group.take() { - let _ = group.finish(); - } - Err(err.into()) + /// End the track with an error, so a consumer sees the failure rather than a clean end. + /// + /// Aborting the *track* is what a subscriber observes. Aborting only the group drops it from the + /// cache and the consumer still reads a clean end, which is exactly what a completed log looks + /// like, so a truncated log would be indistinguishable from a whole one. + fn abort(&mut self, err: moq_net::Error) { + // The track abort closes its groups, so the handle only needs dropping. + self.group = None; + + // Abort through a clone, since aborting consumes a handle and the state is shared. Keeping + // ours means `consume` still hands back a subscriber, which is how a reader learns the log + // ended badly rather than cleanly. + let _ = self.inner.clone().abort(err); } - fn finish(&mut self) -> Result<()> { - if let Some(mut group) = self.group.take() { - group.finish()?; - } - self.inner.finish()?; + fn finish(&mut self) -> std::result::Result<(), moq_net::Error> { + // Finalize both independently rather than short-circuiting on the group. Returning early + // would leave the track open with `group` already taken, so a later append would open a + // second group, and (with compression) write into it from a window the consumer never + // received. That is exactly the split log ending the track exists to prevent. + let group = match self.group.take() { + Some(mut group) => group.finish(), + None => Ok(()), + }; + let track = self.inner.finish(); + + group?; + track?; Ok(()) } } diff --git a/rs/moq-mux/src/binary.rs b/rs/moq-mux/src/binary.rs index 1e7ea70507..25c4797156 100644 --- a/rs/moq-mux/src/binary.rs +++ b/rs/moq-mux/src/binary.rs @@ -173,8 +173,11 @@ impl Stream { &self.name } - /// Create a subscriber for the underlying track, or `None` once a failed write has aborted it. - pub fn consume(&self) -> Option { + /// Create a subscriber for the underlying track. + /// + /// Still hands one back once a failed write has ended the log: the subscriber surfaces the abort + /// on its first read. + pub fn consume(&self) -> moq_net::track::Subscriber { self.inner.consume() } @@ -336,7 +339,7 @@ mod test { .binary_stream("samples", Config::default().with_compression(true)) .unwrap(); - let track = samples.consume().expect("the track is live"); + let track = samples.consume(); let expected: Vec = (0..3u8).map(|n| Bytes::from(vec![n; 8])).collect(); for payload in &expected { samples.append(payload.clone()).unwrap(); diff --git a/rs/moq-mux/src/json.rs b/rs/moq-mux/src/json.rs index 9991d55b66..7110448f65 100644 --- a/rs/moq-mux/src/json.rs +++ b/rs/moq-mux/src/json.rs @@ -178,29 +178,25 @@ impl Stream { } /// Create a subscriber for the underlying track. + /// + /// Still hands one back once a failed write has ended the log: the subscriber surfaces the abort + /// on its first read. pub fn consume(&self) -> moq_net::track::Subscriber { self.inner.consume() } /// Append one record to the log. /// - /// A record that cannot be written ends the track. [`moq_json::stream`] recovers from a failed - /// write by rolling a fresh group, but a catalog [`Mode::Stream`] track is a single group: a log - /// missing a record is not the lossless log the mode promises, and a second group would present - /// that gap as a complete log to a subscriber that joins afterwards. Every later append then - /// fails on the closed track. + /// Any failure ends the track (see [`moq_json::stream::Producer::append`]) and retires the + /// catalog entry with it. pub fn append(&mut self, value: &T) -> crate::Result<()> { let Err(err) = self.inner.append(value) else { return Ok(()); }; - // moq-json has already closed the group it was writing into; closing the track is what stops - // the next append from opening a second one. - let _ = self.inner.finish(); - - // Dropping the rendition retires the catalog entry. Waiting for the handle to drop would keep - // advertising a track that can no longer accept records, so a consumer discovering it now - // would subscribe to an already-ended log. + // The inner producer has already ended the track. Dropping the rendition retires the catalog + // entry: waiting for the handle to drop would keep advertising a track that can no longer + // accept records, so a consumer discovering it now would subscribe to an already-ended log. self.rendition = None; Err(err.into()) @@ -489,6 +485,39 @@ mod test { assert_eq!(consumer.next().await.unwrap(), None); } + /// A value that fails to serialize never reaches the track, but the log is missing it all the + /// same, so it is as terminal as a rejected write. The entry and the track have to agree: retiring + /// the entry while leaving the track writable would let a later valid record land somewhere no + /// consumer could discover. + #[test] + fn an_unserializable_record_ends_the_track_and_the_entry() { + struct Record(bool); + + impl Serialize for Record { + fn serialize(&self, serializer: S) -> Result { + match self.0 { + true => Err(serde::ser::Error::custom("cannot serialize")), + false => serializer.serialize_u8(0), + } + } + } + + let (_broadcast, catalog) = catalog(); + let mut chat = catalog.json_stream::("chat", Config::default()).unwrap(); + let mut track = chat.consume(); + + assert!(chat.append(&Record(true)).is_err()); + assert!(!catalog.snapshot().json.tracks.contains_key("chat")); + + assert!( + chat.append(&Record(false)).is_err(), + "a record after the failure would land on a track the retired entry no longer advertises" + ); + + let waiter = kio::Waiter::noop(); + assert!(matches!(track.poll_recv_group(&waiter), Poll::Ready(Err(_)))); + } + /// A consumer that can't tell a log from a latest-value document would silently drop records, /// so an unreadable entry is an error rather than a guess. #[test]