fix(json): end a stream track when a record cannot be written

rs/moq-binary/src/stream/producer.rs

@@ -44,36 +44,32 @@ 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<moq_net::track::Subscriber> {
-		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.
 	///
 	/// 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<Mutex>`.
 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<moq_net::track::Producer>,
+	track: moq_net::track::Producer,
 
 	/// Opened on the first append and never rolled.
 	group: Option<moq_net::group::Producer>,
@@ -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?;

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,22 +155,29 @@ 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.
 		let track = moq_net::broadcast::Info::new()
 			.produce()
 			.create_track("test", None)
 			.unwrap();
-		let subscriber = track.subscribe(None);
+		let mut subscriber = track.subscribe(None);
 		let mut producer = Producer::<std::collections::BTreeMap<(u8, u8), u8>>::new(track, ProducerConfig::default());
 
 		let mut bad = std::collections::BTreeMap::new();
 		bad.insert((1, 2), 3);
 		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::<Value>::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::<Value>::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]

rs/moq-json/src/stream/producer.rs

@@ -31,6 +31,9 @@ impl<T> Clone for Producer<T> {
 
 impl<T> Producer<T> {
 	/// 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<T: Serialize> Producer<T> {
 	}
 
 	/// 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,69 +89,106 @@ impl<T: Serialize> Inner<T> {
 		// 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();
 		Ok(())
 	}
 
 	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<moq_net::group::Producer>,
 }
 
 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()?);
 		}
 		Ok(())
 	}
 
 	/// 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(())
 	}
 }

rs/moq-mux/src/binary.rs

@@ -173,8 +173,11 @@ impl<E: CatalogExt> Stream<E> {
 		&self.name
 	}
 
-	/// Create a subscriber for the underlying track, or `None` once a failed write has aborted it.
-	pub fn consume(&self) -> Option<moq_net::track::Subscriber> {
+	/// 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<Bytes> = (0..3u8).map(|n| Bytes::from(vec![n; 8])).collect();
 		for payload in &expected {
 			samples.append(payload.clone()).unwrap();