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]