fix(net): decode PUBLISH_DONE Unauthorized and skip a dead source's standing route

quest/m1/README.md

@@ -20,7 +20,7 @@ transport, benchmark tooling); worktrees isolate commits, not semantics.
 - [BBR classic ECN](/quest/m1/bbr-classic-ecn.md) - Startup and bandwidth probing respond to CE marks before the bottleneck drops packets
 - [libmoq hidden opt-in](/quest/m1/libmoq-hidden.md) - `moq_origin_announced` takes a `hidden` flag so C callers can list `.`-named broadcasts
 - [lite-07 count settle](/quest/m1/lite-count-settle.md) - moq-lite-07 subscribers stop waiting for a subscription's tail once SUBSCRIBE_END's stream count is reached
-- [Dropped sources](/quest/m1/dropped-sources.md) - consumers see the producer's real error on every end path, never `Dropped`
+- [Dropped sources](/quest/m1/dropped-sources.md) - track consumers see the producer's real error on every end path, never `Dropped`
 - [JS group guard](/quest/m1/js-group-guard.md) - a `@moq/net` publisher abandons a group past its max age without an unhandled rejection
 - [Track tail interop](/quest/m1/track-tail-interop.md) - a Rust publisher ending a track with a group in flight is read to its end by the JS subscriber, and the reverse, in `just test interop`
 - [Interop flakes](/quest/m1/interop-flakes.md) - the interop harness passes with other runs sharing the machine

quest/m1/dropped-sources.md

@@ -2,26 +2,28 @@
 
 ## Goal
 
-A track or broadcast that ends because its source ended reports the source's
-own error to every consumer, locally and across a relay. `Dropped` means only
-that a handle was dropped without an end, which a correct producer never does.
-#4179 fixed one path (a revoked upstream subscription now reads
-`Unauthorized`); the rest still surface `Dropped`.
+A track that ends because its source ended reports the source's own error to
+every consumer, locally and across a relay. `Dropped` means only that a handle
+was dropped without an end, which a correct producer never does. A broadcast
+end carries no cause since #4031, so only track errors are in scope.
 
 ## Plan
 
-- Known sources, from #4179: a source closing, a route leaving the origin's
-  table, and a withdrawn source broadcast. Find each place a consumer can
-  observe `Dropped` and make the ending side carry its real error (an explicit
-  `abort` or a preserved cause), at the source rather than by remapping at the
-  consumer.
-- moq-transport: a `PUBLISH_DONE` carrying Unauthorized arrives as
-  `Error::Remote(1)`. Map it to the same error lite reports.
-- Regression tests per path, each failing on `Dropped` today, in-process and
-  over a mock session.
+- Rust already maps IETF `PUBLISH_DONE` Unauthorized to `Error::Unauthorized`,
+  and a closed source's standing route no longer re-requests it.
+- #4179 fixes revoked upstream subscriptions and #4120 preserves session death
+  and resumed-track errors. After #4179 reaches `main`, verify the remaining
+  track paths (source close, route removal, broadcast withdrawal) locally and
+  over a mock session, and map JS `PUBLISH_DONE` Unauthorized to #4179's shared
+  error. Preserve causes at the source rather than remapping `Dropped` at
+  consumers.
 
 Public API: none expected; error values consumers observe change. Wire: none.
 
+## Required
+
+- [Unauthorized](/quest/m1/auth/unauthorized.md) - #4179 supplies shared Unauthorized errors and revoked-stream handling
+
 ## Related
 
-- [#4179](https://github.com/moq-dev/moq/pull/4179) - fixed the revoked-upstream path
+- [#4179](https://github.com/moq-dev/moq/pull/4179) - owns the revoked-upstream path

rs/moq-net/src/ietf/publish.rs

@@ -124,6 +124,8 @@ use super::Version;
 pub(crate) enum PublishDoneStatus {
 	/// An implementation-specific failure ended the subscription.
 	InternalError,
+	/// The subscriber is no longer authorized for the track.
+	Unauthorized,
 	/// The track is no longer being published.
 	TrackEnded,
 }
@@ -144,6 +146,7 @@ impl PublishDoneStatus {
 			| Version::Draft21
 			| Version::Draft22 => match self {
 				Self::InternalError => 0x0,
+				Self::Unauthorized => 0x1,
 				Self::TrackEnded => 0x2,
 			},
 		}
@@ -164,6 +167,7 @@ impl PublishDone<'_> {
 	pub(crate) fn end(&self, version: Version) -> Result<(), crate::Error> {
 		match self.status_code {
 			code if code == PublishDoneStatus::TrackEnded.code(version) => Ok(()),
+			code if code == PublishDoneStatus::Unauthorized.code(version) => Err(crate::Error::Unauthorized),
 			// SUBSCRIPTION_ENDED: the subscription reached the end its filter asked for.
 			// Draft-20 removed it and left 0x3 unassigned.
 			0x3 if matches!(
@@ -731,6 +735,10 @@ mod tests {
 
 		for version in [Version::Draft14, Version::Draft19, Version::Draft20, Version::Draft22] {
 			assert!(done(0x2).end(version).is_ok(), "{version:?}");
+			assert!(
+				matches!(done(0x1).end(version), Err(crate::Error::Unauthorized)),
+				"{version:?}"
+			);
 			assert!(
 				matches!(done(0x0).end(version), Err(crate::Error::Remote(0x0))),
 				"{version:?}"

rs/moq-net/src/model/front.rs

@@ -202,7 +202,8 @@ pub(super) struct Front {
 	serving_closing: bool,
 	/// The route an upstream request is in flight through.
 	upstream: Option<u64>,
-	/// Routes that refused the path while another source was serving.
+	/// Routes excluded from selection: they refused the path while another
+	/// source was serving, or their source ended while still advertised.
 	refused: HashSet<u64>,
 	/// Why the last candidate fell through, reported if the front ends unresolved.
 	last_err: Option<Error>,
@@ -244,7 +245,7 @@ impl Front {
 		self.identity.pin()
 	}
 
-	/// The routes that refused the path; the driver skips them when selecting.
+	/// The routes excluded from selection; the driver skips them.
 	pub(super) fn refused_routes(&self) -> &HashSet<u64> {
 		&self.refused
 	}
@@ -429,12 +430,16 @@ impl Front {
 	}
 
 	fn source_closed(&mut self, source: u64, actions: &mut Vec<Action>) {
-		let Some((serving, _)) = self.serving else {
+		let Some((serving, route)) = self.serving else {
 			return;
 		};
 		if serving != source {
 			return;
 		}
+		// A standing route can outlive the source it produced. Asking it again
+		// would re-request the broadcast that just ended; another route to the
+		// same publisher may still resume it.
+		self.refused.insert(route);
 		self.serving = None;
 		self.serving_closing = false;
 		actions.push(Action::Detach { source });
@@ -813,6 +818,7 @@ mod tests {
 			front.step(Event::SourceClosed { source: 100 }),
 			&[Action::Detach { source: 100 }, Action::Reselect],
 		);
+		assert!(front.refused_routes().contains(&1));
 		assert_actions(
 			front.step(Event::Selected {
 				best: Some(remote(3, 10)),

rs/moq-net/src/model/origin.rs

@@ -6225,6 +6225,27 @@ mod tests {
 		assert!(end.is_none(), "a group followed the final one");
 	}
 
+	/// A standing route outlives the source it produced: the front ends instead of
+	/// asking that route for the broadcast that just closed.
+	#[tokio::test]
+	async fn a_closed_source_is_not_requested_again_from_its_standing_route() {
+		let producer = origin(1).produce();
+		let server = producer
+			.dynamic("room", Route::default().with_hops(hops(&[10])))
+			.unwrap();
+		let pending = producer.consume().request_broadcast("room/alice");
+		let source = broadcast::Info::new().produce();
+		queued(&server).await.accept(&source);
+		let resolved = pending.await.unwrap();
+
+		source.close();
+		settle(|| resolved.is_closed()).await;
+		assert!(
+			server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending(),
+			"the closed source was requested again"
+		);
+	}
+
 	/// An origin front drops the source track as soon as its last reader leaves,
 	/// so the publisher's `unused()` resolves far below `TRACK_IDLE_LINGER`.
 	/// Cached groups stay on the front for the linger; a returning reader