fix(net): preserve aborted source errors through origins

doc/lib/rs/moq-net.md

@@ -18,6 +18,7 @@ above ([hang](/lib/rs/hang)); relays and CDNs implement only this.
 - **Origins** scope what a session can see, and merge duplicate subscriptions so a broadcast is pulled upstream once no matter how many local readers.
 - **Broadcasts** are created unannounced and invisible to everyone, then announced as an exact route, or served below a prefix with `dynamic`. A consumer of the same origin sees exactly what a peer sees. Discovery accepts pattern unions; events carry the advertised prefix and captures for a complete match.
 - **Patterns** (`Pattern`, `Patterns`) are re-exported from [`moq-pattern`](https://docs.rs/moq-pattern). Literal `Path` stays a coordinate.
+- **Broadcast errors** preserve the source abort cause through origin routing, including requests made after closure. Existing tracks keep their independent ends. IETF `PUBLISH_DONE` Unauthorized is reported as `Error::Unauthorized`.
 - **Tracks** carry groups with a priority, a retention window, and a timescale. Subscribers set their own priority and max age and can change them live.
 - **Groups** are written frame by frame and delivered on independent streams. Old groups are cached for fetch-by-sequence; stale groups are skipped per the subscriber's budget.
 - **Track ends**: `finish()` ends a track at its live edge, while `finish_at(n)` declares the exclusive end ahead of it and still accepts the groups below. A subscriber awaits it with `finished()`. A remote track ends only once every group below its end has arrived or was dropped; one reset before its header arrived is skipped after the subscription's max age on moq-lite (one second without one), or after one second on IETF.

quest/m1/dropped-sources.md

@@ -5,23 +5,29 @@
 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`.
+The open #4179 fixes revoked upstream subscriptions. The merged #4120 preserves
+session death and resumed-track errors; remaining paths still need verification.
 
 ## 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.
+- Origin broadcasts now preserve an aborted source's cause through local and
+  routed fronts, including a concurrent route withdrawal and later track lookup.
+  A closed source's standing route is excluded from that front's failover.
+- Rust maps IETF `PUBLISH_DONE` Unauthorized to `Error::Unauthorized`.
+- After #4179 lands, verify the combined track paths locally and over
+  mock sessions, and map JS `PUBLISH_DONE` Unauthorized to the shared error from
+  #4179. Do not duplicate the resume changes those PRs own.
+
+- Complete the remaining source-close, route-removal, and broadcast-withdrawal
+  track regressions locally and over a mock session. 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/benches/origin.rs

@@ -377,6 +377,54 @@ fn bench_handoff(c: &mut Criterion) {
 	group.finish();
 }
 
+/// Closing one routed source with a standing route, swept over unrelated routes
+/// and consumers of that source. Only the touched front should reselect.
+fn bench_source_close(c: &mut Criterion) {
+	let mut group = c.benchmark_group("origin/source_close");
+	for (routes, subscribers) in [(1, 1), (1_000, 1), (1, 100), (1_000, 100)] {
+		group.bench_function(BenchmarkId::from_parameter(format!("{routes}r_{subscribers}s")), |b| {
+			let runtime = tokio::runtime::Builder::new_current_thread()
+				.enable_all()
+				.build()
+				.unwrap();
+			let (producer, driver) = origin::Producer::new(origin::Config::default());
+			runtime.spawn(moq_net::time::run(driver));
+			let consumer = producer.consume();
+			let _others: Vec<_> = (0..routes)
+				.map(|i| {
+					producer
+						.dynamic(format!("other/{i}"), origin::Route::default())
+						.unwrap()
+				})
+				.collect();
+			b.iter_custom(|iterations| {
+				runtime.block_on(async {
+					let mut total = Duration::ZERO;
+					for _ in 0..iterations {
+						let server = producer.dynamic("live", peer_route(1, 0)).unwrap();
+						let source = broadcast::Info::new().produce();
+						let pending = consumer.request_broadcast("live");
+						server.requested_broadcast().await.unwrap().accept(&source);
+						let first = pending.await.unwrap();
+						let mut readers = vec![first];
+						for _ in 1..subscribers {
+							readers.push(consumer.request_broadcast("live").await.unwrap());
+						}
+						let started = std::time::Instant::now();
+						source.abort(moq_net::Error::Unauthorized).unwrap();
+						for reader in readers {
+							assert!(matches!(reader.closed().await, moq_net::Error::Unauthorized));
+						}
+						total += started.elapsed();
+					}
+					total
+				})
+			});
+		});
+	}
+	group.finish();
+}
+
 criterion_group!(
 	benches,
 	bench_announce,
@@ -386,6 +434,7 @@ criterion_group!(
 	bench_serve_idle,
 	bench_subscribe,
 	bench_request,
-	bench_handoff
+	bench_handoff,
+	bench_source_close
 );
 criterion_main!(benches);

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

@@ -163,6 +163,7 @@ impl PublishDone<'_> {
 	/// How the publisher ended the subscription: cleanly, or with the error its status names.
 	pub(crate) fn end(&self, version: Version) -> Result<(), crate::Error> {
 		match self.status_code {
+			0x1 => Err(crate::Error::Unauthorized),
 			code if code == PublishDoneStatus::TrackEnded.code(version) => Ok(()),
 			// SUBSCRIPTION_ENDED: the subscription reached the end its filter asked for.
 			// Draft-20 removed it and left 0x3 unassigned.
@@ -731,6 +732,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/broadcast.rs

@@ -818,7 +818,7 @@ impl Consumer {
 	fn track_inner(&self, name: &str) -> Result<track::Consumer, Error> {
 		// A closed broadcast (every producer and handler gone) serves nothing.
 		if self.is_closed() {
-			return Err(Error::Dropped);
+			return Err(self.error());
 		}
 
 		let mut state = self.state.lock();
@@ -927,6 +927,11 @@ impl Consumer {
 	/// tell those apart).
 	pub async fn closed(&self) -> Error {
 		self.alive.closed().await;
+		self.error()
+	}
+
+	/// The recorded source error, or `Dropped` when no abort was recorded.
+	pub(crate) fn error(&self) -> Error {
 		self.state.read().abort.clone().unwrap_or(Error::Dropped)
 	}
 

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

@@ -52,7 +52,7 @@ pub(super) enum Event {
 	/// given id by the driver) or was refused.
 	Resolved { route: u64, result: Result<u64, Refusal> },
 	/// A source closed: it will never serve again.
-	SourceClosed { source: u64 },
+	SourceClosed { source: u64, err: Error },
 	/// The spliced broadcast handed out a new logical track to serve.
 	TrackAssigned { track: Arc<str> },
 	/// A source answered a track query: its copy's metadata, or a refusal.
@@ -274,7 +274,7 @@ impl Front {
 		match event {
 			Event::Selected { best, serving_closing } => self.selected(best, serving_closing, &mut actions),
 			Event::Resolved { route, result } => self.resolved(route, result, &mut actions),
-			Event::SourceClosed { source } => self.source_closed(source, &mut actions),
+			Event::SourceClosed { source, err } => self.source_closed(source, err, &mut actions),
 			// A fresh logical track, even under a name served before: an earlier
 			// verdict belonged to that request, and a later one asks afresh. The
 			// broadcast's track metadata is what persists.
@@ -428,13 +428,16 @@ impl Front {
 		};
 	}
 
-	fn source_closed(&mut self, source: u64, actions: &mut Vec<Action>) {
-		let Some((serving, _)) = self.serving else {
+	fn source_closed(&mut self, source: u64, err: Error, actions: &mut Vec<Action>) {
+		let Some((serving, route)) = self.serving else {
 			return;
 		};
 		if serving != source {
 			return;
 		}
+		// A standing route can still advertise a source that ended. This front
+		// must not ask it for new content; another route may resume the old source.
+		self.refused.insert(route);
 		self.serving = None;
 		self.serving_closing = false;
 		actions.push(Action::Detach { source });
@@ -444,11 +447,11 @@ impl Front {
 				track.state = TrackState::Idle;
 			}
 		}
-		self.last_err = Some(Error::Dropped);
+		self.last_err = Some(err.clone());
 		match self.identity {
 			// A local publisher ending ends its broadcast; a newcomer at the
 			// path gets a fresh one. An anonymous source can never be resumed.
-			Identity::Local | Identity::Anonymous { .. } | Identity::Undetermined => self.end(Error::Dropped, actions),
+			Identity::Local | Identity::Anonymous { .. } | Identity::Undetermined => self.end(err, actions),
 			Identity::Publisher(_) => actions.push(Action::Reselect),
 		}
 	}
@@ -810,9 +813,13 @@ mod tests {
 	fn dead_source_reselects_through_the_same_publisher() {
 		let mut front = serving(remote(1, 10), 100);
 		assert_actions(
-			front.step(Event::SourceClosed { source: 100 }),
+			front.step(Event::SourceClosed {
+				source: 100,
+				err: Error::Dropped,
+			}),
 			&[Action::Detach { source: 100 }, Action::Reselect],
 		);
+		assert!(front.refused_routes().contains(&1));
 		assert_actions(
 			front.step(Event::Selected {
 				best: Some(remote(3, 10)),
@@ -875,7 +882,10 @@ mod tests {
 	#[test]
 	fn dead_source_with_no_replacement_ends() {
 		let mut front = serving(remote(1, 10), 100);
-		front.step(Event::SourceClosed { source: 100 });
+		front.step(Event::SourceClosed {
+			source: 100,
+			err: Error::Dropped,
+		});
 		assert_actions(
 			front.step(Event::Selected {
 				best: None,
@@ -895,7 +905,10 @@ mod tests {
 		let mut front = serving(candidate, 100);
 		assert_eq!(front.pin(), Pin::Route(1));
 		assert_actions(
-			front.step(Event::SourceClosed { source: 100 }),
+			front.step(Event::SourceClosed {
+				source: 100,
+				err: Error::Dropped,
+			}),
 			&[Action::Detach { source: 100 }, Action::End { err: Error::Dropped }],
 		);
 	}
@@ -997,7 +1010,10 @@ mod tests {
 		);
 		assert!(front.tracks[&name("audio")].refused.is_empty());
 		// The replacement is asked afresh.
-		front.step(Event::SourceClosed { source: 100 });
+		front.step(Event::SourceClosed {
+			source: 100,
+			err: Error::Dropped,
+		});
 		front.step(Event::Selected {
 			best: Some(remote(2, 10)),
 			serving_closing: false,
@@ -1186,7 +1202,10 @@ mod tests {
 	fn local_incumbent_ending_ends_the_front() {
 		let mut front = serving(local(1), 100);
 		assert_actions(
-			front.step(Event::SourceClosed { source: 100 }),
+			front.step(Event::SourceClosed {
+				source: 100,
+				err: Error::Dropped,
+			}),
 			&[Action::Detach { source: 100 }, Action::End { err: Error::Dropped }],
 		);
 	}
@@ -1230,7 +1249,10 @@ mod tests {
 					standing: true,
 				}),
 			},
-			Event::SourceClosed { source: 100 },
+			Event::SourceClosed {
+				source: 100,
+				err: Error::Dropped,
+			},
 			Event::TrackAssigned { track: name("v") },
 			Event::Used { track: name("v") },
 			Event::Unused {

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

@@ -2025,7 +2025,7 @@ async fn run_front(task: FrontTask) {
 	enum Step {
 		Assigned(Arc<str>, super::resume::Producer),
 		Resolved(u64, Result<broadcast::Consumer, Error>),
-		SourceClosed(u64),
+		SourceClosed(u64, Error),
 		Info(Arc<str>, u64, Result<track::Info, Error>),
 		Ended(Arc<str>, u64, Result<(), Error>),
 		Demand(Arc<str>),
@@ -2051,6 +2051,17 @@ async fn run_front(task: FrontTask) {
 		if table.closed {
 			return Event::Closed;
 		}
+		// A table withdrawal can race the close wakeup. Preserve the source's cause
+		// before selecting a replacement or ending a front with no remaining route.
+		if let Some(source) = front.serving()
+			&& let Some(broadcast) = sources.get(&source)
+			&& broadcast.is_closed()
+		{
+			return Event::SourceClosed {
+				source,
+				err: broadcast.error(),
+			};
+		}
 		// Read alongside the decision, under the lock a poke takes first.
 		*seen = watch.seen();
 		front.retain_routes(|route| table.routes.covers(&path.as_path(), route));
@@ -2286,7 +2297,11 @@ async fn run_front(task: FrontTask) {
 						// subscriptions already in flight): dropping their producers
 						// leaves each reader on the copy it was spliced from, ending
 						// when and as that copy ends.
-						broadcast.finish();
+						if matches!(err, Error::Dropped) {
+							broadcast.finish();
+						} else {
+							let _ = broadcast.clone().abort(err.clone());
+						}
 						broadcast.release_spliced(err.clone());
 						for (_, mut io) in tracks.drain() {
 							// A reader still waiting on its source's answer is in flight
@@ -2336,7 +2351,7 @@ async fn run_front(task: FrontTask) {
 				&& let Some(source) = sources.get(&id)
 				&& source.poll_closed(waiter).is_ready()
 			{
-				return Poll::Ready(Step::SourceClosed(id));
+				return Poll::Ready(Step::SourceClosed(id, source.error()));
 			}
 			for (name, io) in &tracks {
 				if let Some((source, _, query)) = &io.query
@@ -2404,7 +2419,7 @@ async fn run_front(task: FrontTask) {
 					}
 				}
 			}
-			Step::SourceClosed(source) => Event::SourceClosed { source },
+			Step::SourceClosed(source, err) => Event::SourceClosed { source, err },
 			Step::Info(name, source, result) => {
 				let closing = sources.get(&source).is_some_and(|s| s.is_closing());
 				let Some(io) = tracks.get_mut(&name) else { continue };
@@ -6228,6 +6243,36 @@ mod tests {
 		assert!(end.is_none(), "a group followed the final one");
 	}
 
+	#[tokio::test]
+	async fn an_aborted_source_preserves_its_broadcast_error() {
+		let producer = origin(1).produce();
+		let broadcast = producer.publish("room/alice", Route::default()).unwrap();
+		let resolved = producer.consume().request_broadcast("room/alice").await.unwrap();
+		broadcast.abort(Error::Unauthorized).unwrap();
+		assert!(matches!(resolved.closed().await, Error::Unauthorized));
+		assert!(matches!(resolved.track("video"), Err(Error::Unauthorized)));
+	}
+
+	#[tokio::test]
+	async fn an_aborted_routed_source_preserves_its_broadcast_error() {
+		for withdraw in [false, true] {
+			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.abort(Error::Unauthorized).unwrap();
+			if withdraw {
+				drop(server);
+			}
+			assert!(matches!(resolved.closed().await, Error::Unauthorized));
+			assert!(matches!(resolved.track("video"), Err(Error::Unauthorized)));
+		}
+	}
+
 	/// 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