From 2ea2911535853930edf637f506507a57c0c3fa67 Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Sat, 7 Nov 2026 23:55:50 -0800
Subject: [PATCH 01/17] chore(quest): claim
 quest/m1/explicit-state/net-phases.md

Quest-Claim: 5d143ffb66b604a3

From 803125243187a0421e7015046c173cc8f140af36 Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Sun, 8 Nov 2026 00:19:12 -0800
Subject: [PATCH 02/17] fix(net): a drained IETF subscription serves the tail
 an update reopens

TrackServe owns its phase (serving, draining with the track's end, or
failed), so RunSubscription stops caching the drained result and polls it
every turn. An update that reopens a drained range now serves its fill
tail instead of ending the subscription without it.

The datagram forward, the session drivers' terminal result, IETF
dispatch, lite PeerSetup, and the listing queue each keep one phase.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 rs/moq-net/src/ietf/publisher.rs | 223 ++++++++++++++++++++-----------
 rs/moq-net/src/ietf/session.rs   | 206 ++++++++++++++--------------
 rs/moq-net/src/lite/session.rs   |  68 +++++-----
 rs/moq-net/src/lite/setup.rs     | 112 ++++++++++------
 rs/moq-net/src/model/listing.rs  |  73 +++++-----
 5 files changed, 398 insertions(+), 284 deletions(-)

diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs
index daab6e47e9d..223224ef35a 100644
--- a/rs/moq-net/src/ietf/publisher.rs
+++ b/rs/moq-net/src/ietf/publisher.rs
@@ -2007,8 +2007,6 @@ struct RunSubscription {
 	/// The fills updates asked for.
 	fills: crate::util::TaskSet,
 	fills_done: bool,
-	/// How serving the track ended, once it did.
-	served: Option<Result<(), Error>>,
 	/// An update asking for a fill, held until the fills and the group being opened finish.
 	waiting: Option<super::request_stream::Update>,
 	/// The Largest Location the REQUEST_OK for a renewal being verified reports.
@@ -2036,7 +2034,6 @@ impl RunSubscription {
 			fill,
 			fills: crate::util::TaskSet::owned(),
 			fills_done: true,
-			served: None,
 			waiting: None,
 			renewal_largest: None,
 			pausing: false,
@@ -2120,11 +2117,7 @@ impl RunSubscription {
 							}
 						}
 
-						if self.served.is_none()
-							&& let Poll::Ready(done) = serve.poll(waiter)
-						{
-							self.served = Some(done);
-						}
+						let served = serve.poll(waiter);
 						if self.fill.is_none()
 							&& self.fills_done && serve.tail.is_none()
 							&& serve.opening.is_none()
@@ -2132,12 +2125,10 @@ impl RunSubscription {
 						{
 							break 'turn Turn::Update(update);
 						}
-						match (
-							&self.served,
-							self.fill.is_none() && self.fills_done && self.waiting.is_none(),
-						) {
-							(Some(Ok(())), true) => Turn::Served(Ok(())),
-							(Some(Err(err)), true) => Turn::Served(Err(err.clone())),
+						match served {
+							Poll::Ready(res) if self.fill.is_none() && self.fills_done && self.waiting.is_none() => {
+								Turn::Served(res)
+							}
 							_ => return Poll::Pending,
 						}
 					};
@@ -6433,12 +6424,9 @@ struct TrackServe {
 	/// Where the group streams are ranked: the subscription's send group, or the session's
 	/// queue once [`Self::ranked`] on a transport without groups.
 	ranking: Ranking<GroupOf<Transport>>,
-	/// The track finished: the in-flight group machines drain, then FIN.
-	draining: bool,
+	phase: ServePhase,
 	/// Group streams opened, shared with the group machines.
 	opened: Arc<AtomicU64>,
-	/// The track's exclusive end, once its groups ran out because it finished.
-	end: Option<track::Position>,
 	terminal: Arc<Terminal>,
 	compression: bool,
 	/// A track that always has another group ready still lets the session's other tasks run.
@@ -6448,12 +6436,25 @@ struct TrackServe {
 	dynamic: bool,
 }
 
+/// Where serving a track's groups stands. Only [`TrackServe`] holds it, so an update that
+/// widens a drained range serves again on the next poll.
+enum ServePhase {
+	/// Handing out groups as they arrive.
+	Serving,
+	/// The groups ran out, so the group machines in flight drain, then FIN. Holds the track's
+	/// exclusive end when the track finishing is what ran them out. Entered only once no
+	/// group waits to open and no fill tail is owed.
+	Draining(Option<track::Position>),
+	/// Reading the track failed, which ends the subscription.
+	Failed(Error),
+}
+
+/// Forwards a broadcast's datagram track; dropped once that track ends or fails.
 struct DatagramForward {
 	consumer: crate::datagram::Consumer,
-	ready: bool,
-	ended: bool,
 	timescale: Option<Timescale>,
-	publisher: i8,
+	/// The datagram track's priority, once its info resolved.
+	publisher: Option<i8>,
 	demand: bool,
 }
 
@@ -6506,9 +6507,8 @@ impl TrackServe {
 			children: kio::Tasks::new(),
 			priority,
 			ranking,
-			draining: false,
+			phase: ServePhase::Serving,
 			opened: Default::default(),
-			end: None,
 			terminal: Default::default(),
 			budget: kio::coop::Budget::new(32),
 			dynamic,
@@ -6523,10 +6523,8 @@ impl TrackServe {
 		self.datagrams = Some(DatagramForward {
 			consumer: datagrams,
 			demand,
-			ready: false,
-			ended: false,
 			timescale: self.timescale.filter(|_| stamp),
-			publisher: 0,
+			publisher: None,
 		});
 		self
 	}
@@ -6617,7 +6615,9 @@ impl TrackServe {
 		let end = range.end.map_or(Bound::Unbounded, |end| Bound::Included(end.group));
 		self.track.set_range((Bound::Included(start), end));
 		self.range = range;
-		self.draining = false;
+		if let ServePhase::Draining(_) = self.phase {
+			self.phase = ServePhase::Serving;
+		}
 		Ok(fill.map(|fill| SubscriptionFill {
 			fill,
 			largest: edge.largest,
@@ -6686,7 +6686,9 @@ impl TrackServe {
 		if self.datagram_track.is_some() {
 			return None;
 		}
-		let end = self.end?;
+		let ServePhase::Draining(Some(end)) = self.phase else {
+			return None;
+		};
 		if self.terminal.sent.load(Ordering::Relaxed) {
 			return None;
 		}
@@ -6777,8 +6779,10 @@ impl TrackServe {
 		{
 			let _ = self.terminal.end.set(end);
 		}
-		if self.draining {
-			return self.children.poll(waiter).map(Ok);
+		match &self.phase {
+			ServePhase::Serving => {}
+			ServePhase::Draining(_) => return self.children.poll(waiter).map(Ok),
+			ServePhase::Failed(err) => return Poll::Ready(Err(err.clone())),
 		}
 
 		let _ = self.children.poll(waiter);
@@ -6844,13 +6848,17 @@ impl TrackServe {
 						let _ = self.children.poll(waiter);
 						return Poll::Pending;
 					}
-					self.draining = true;
-					if let Poll::Ready(Ok(end)) = self.track.poll_finished(waiter) {
-						self.end = Some(end);
-					}
+					let end = match self.track.poll_finished(waiter) {
+						Poll::Ready(Ok(end)) => Some(end),
+						_ => None,
+					};
+					self.phase = ServePhase::Draining(end);
 					return self.children.poll(waiter).map(Ok);
 				}
-				Poll::Ready(Err(err)) => return Poll::Ready(Err(err)),
+				Poll::Ready(Err(err)) => {
+					self.phase = ServePhase::Failed(err.clone());
+					return Poll::Ready(Err(err));
+				}
 				Poll::Pending => break,
 			}
 		}
@@ -6867,53 +6875,42 @@ impl TrackServe {
 			if self.budget.poll_yield(waiter).is_pending() {
 				return;
 			}
-			let datagram = {
-				let Some(forward) = self.datagrams.as_mut() else {
-					return;
-				};
-				if forward.ended {
-					return;
-				}
-				if !forward.ready {
-					match forward.consumer.poll_info(waiter) {
-						Poll::Pending => return,
-						Poll::Ready(Ok(info)) => {
-							// Object times use the stream subscription's advertised timescale.
-							forward.publisher = info.priority;
-							forward.ready = true;
-						}
-						Poll::Ready(Err(err)) => {
-							tracing::debug!(%err, "datagram track ended before it started");
-							forward.ended = true;
-							return;
-						}
-					}
-				}
-				match forward.consumer.poll_recv(waiter) {
+			let Some(forward) = self.datagrams.as_mut() else {
+				return;
+			};
+			let publisher = match forward.publisher {
+				Some(publisher) => publisher,
+				None => match forward.consumer.poll_info(waiter) {
 					Poll::Pending => return,
-					Poll::Ready(Ok(None)) => {
-						forward.ended = true;
-						if let Some(track) = &mut self.datagram_track {
-							let _ = track.finish();
-						}
-						return;
-					}
+					// Object times use the stream subscription's advertised timescale.
+					Poll::Ready(Ok(info)) => *forward.publisher.insert(info.priority),
 					Poll::Ready(Err(err)) => {
-						tracing::debug!(%err, "stopping datagrams on this subscription");
-						if let Some(track) = &self.datagram_track {
-							let _ = track.clone().abort(err);
-						}
-						forward.ended = true;
+						tracing::debug!(%err, "datagram track ended before it started");
+						self.datagrams = None;
 						return;
 					}
-					Poll::Ready(Ok(Some(datagram))) => {
-						let timescale = forward.timescale;
-						let publisher = forward.publisher;
-						(datagram, timescale, publisher)
+				},
+			};
+			let datagram = match forward.consumer.poll_recv(waiter) {
+				Poll::Pending => return,
+				Poll::Ready(Ok(Some(datagram))) => datagram,
+				Poll::Ready(Ok(None)) => {
+					self.datagrams = None;
+					if let Some(track) = &mut self.datagram_track {
+						let _ = track.finish();
+					}
+					return;
+				}
+				Poll::Ready(Err(err)) => {
+					tracing::debug!(%err, "stopping datagrams on this subscription");
+					self.datagrams = None;
+					if let Some(track) = &self.datagram_track {
+						let _ = track.clone().abort(err);
 					}
+					return;
 				}
 			};
-			let (datagram, timescale, publisher) = datagram;
+			let timescale = forward.timescale;
 			emit_datagram(
 				&self.session,
 				self.version,
@@ -12134,6 +12131,82 @@ mod serve_tests {
 		}
 	}
 
+	/// An update that reopens a drained subscription with a fill tail serves that tail.
+	///
+	/// The track finishes with its last group still open, outside the live range, so the
+	/// subscription drains while its fill waits for stream credit. The update behind that
+	/// fill moves the range back onto the open group: its fill covers the group's head and
+	/// its tail the frames still to come.
+	#[kio::test]
+	async fn an_update_reopening_a_drained_subscription_serves_its_tail() {
+		let version = Version::Draft22;
+		let credit = kio::Producer::new(false);
+		let session = ScriptedSession::per_stream(vec![Vec::new()]).with_open_uni_gate(credit.consume());
+		let h = serve_session(version, crate::track::Info::timed(), session);
+		let mut group = h.track.create_group(group::Info { sequence: 1 }).unwrap();
+		group.write_frame(timestamp(), b"head".as_slice()).unwrap();
+		kio::clock::sim::yield_now().await;
+		let mut request = ScriptedSession::new(Vec::new());
+		request.log = h.log.clone();
+		let stream = Stream::open(&crate::driver::Transport::new(request.clone()), version)
+			.await
+			.unwrap();
+		let at_head = Filter::Absolute {
+			start: Location { group: 1, object: 0 },
+			end: None,
+		};
+		let msg = subscribe(
+			Filter::Absolute {
+				start: Location { group: 5, object: 0 },
+				end: None,
+			},
+			Some(ietf::Fill {
+				group_order: None,
+				filter: Some(at_head),
+				range_filters: false,
+			}),
+		);
+		let mut serving = std::pin::pin!(h.publisher.clone().run_subscribe_stream(stream, msg));
+		serve_until(serving.as_mut(), "the subscription is accepted", || {
+			!h.log.trail().is_empty() && Wire::read(&h.log, version).sent(ietf::SubscribeOk::ID).is_some()
+		})
+		.await;
+		// Group 1 stays open: the end falls after it.
+		h.track.finish().unwrap();
+		let update = ietf::SubscribeUpdate {
+			request_id: RequestId(2),
+			subscriber_priority: None,
+			forward: None,
+			filter: Some(at_head),
+			authorization_token: None,
+			new_group: None,
+			max_age: None,
+			fill: Some(ietf::Fill {
+				group_order: None,
+				filter: None,
+				range_filters: false,
+			}),
+		};
+		let mut bytes = Vec::new();
+		let mut encoder = Encoder::new(&mut bytes, version.into());
+		encoder.varint(ietf::SubscribeUpdate::ID).unwrap();
+		update.encode(&mut encoder, version).unwrap();
+		request.push(&bytes);
+		for _ in 0..200 {
+			assert!(futures::poll!(serving.as_mut()).is_pending(), "the fill holds the request");
+		}
+
+		*credit.write().ok().expect("credit is live") = true;
+		serve_until(serving.as_mut(), "the update is accepted, its tail still owed", || {
+			Wire::read(&h.log, version).sent(ietf::RequestOk::ID).is_some()
+		})
+		.await;
+		group.write_frame(timestamp(), b"tail".as_slice()).unwrap();
+		group.finish().unwrap();
+		serving.await.unwrap();
+		assert_eq!(occurrences(&h.log, b"tail"), 1, "the update's tail is served");
+	}
+
 	/// A fill-only update keeps a live join's floor, so a fill short of the live edge does not
 	/// also hand the groups below that floor to the live cursor.
 	#[kio::test]
diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs
index 68551d383cb..91952be29a4 100644
--- a/rs/moq-net/src/ietf/session.rs
+++ b/rs/moq-net/src/ietf/session.rs
@@ -480,7 +480,7 @@ impl Launch {
 				tasks: task_set,
 				session,
 			},
-			ending: None,
+			ended: None,
 			auth,
 			auth_close,
 			setup_token,
@@ -491,8 +491,8 @@ impl Launch {
 /// A running session: its halves race until one fails or the task set drains.
 struct Running {
 	race: Race,
-	/// How the race ended, held while what already arrived drains.
-	ending: Option<Result<(), Error>>,
+	/// How the race ended, once it did: what already arrived drains, then this is the result.
+	ended: Option<Result<(), Error>>,
 	auth: crate::auth::Link,
 	auth_close: AuthClose,
 	/// Held for the life of the session.
@@ -501,17 +501,21 @@ struct Running {
 
 impl Running {
 	fn poll(&mut self, waiter: &kio::Waiter) -> Poll<Result<(), Error>> {
-		if self.ending.is_none() {
-			let res = ready!(self.race.poll(waiter));
-			close(&mut self.race.session, &res);
-			// Before the Auth serve tasks drop, so each settles with the session's error
-			// rather than a bare cancel.
-			self.auth.close(match &res {
-				Ok(()) => Error::Cancel,
-				Err(err) => err.clone(),
-			});
-			self.ending = Some(res);
-		}
+		let res = match &self.ended {
+			Some(res) => res.clone(),
+			None => {
+				let res = ready!(self.race.poll(waiter));
+				close(&mut self.race.session, &res);
+				// Before the Auth serve tasks drop, so each settles with the session's error
+				// rather than a bare cancel.
+				self.auth.close(match &res {
+					Ok(()) => Error::Cancel,
+					Err(err) => err.clone(),
+				});
+				self.ended = Some(res.clone());
+				res
+			}
+		};
 		// Whichever side closed it, read what already arrived.
 		ready!(self.race.unis.poll_drain(waiter));
 		let _ = self.race.tasks.poll(waiter);
@@ -523,7 +527,6 @@ impl Running {
 		}
 		self.race.subscriber.datagrams_routed();
 		let _ = self.race.tasks.poll(waiter);
-		let res = self.ending.take().expect("draining an ended session");
 		// Closed cleanly by dropping or draining: readers see what the peer's clean
 		// close would have shown them.
 		if self.race.session.closing().is_local() {
@@ -1441,10 +1444,7 @@ struct Dispatch {
 	session: Transport,
 	version: Version,
 	peer_setup: peer::PeerSetup,
-	/// What `bidi` is built from once the peer's SETUP arrives.
-	start: Option<(Publisher, Subscriber, Option<auth::Serve>)>,
-	/// What each request is served with, once the peer's SETUP says how they decode.
-	bidi: Option<Bidi>,
+	state: DispatchState,
 	tasks: TaskSet,
 	fatal: kio::Shared<Option<Error>>,
 	/// MAX_STREAMS credit alone lets the peer keep thousands of streams ready, and the
@@ -1452,12 +1452,19 @@ struct Dispatch {
 	budget: kio::coop::Budget,
 	/// One pass per accepted stream, not per wake.
 	charge: bool,
-	/// Each AUTH serve task settles `Issued::closed` from whatever the auth handle
-	/// holds when it drops, and the first reason wins. So the error that ends the
-	/// session has to reach the handle while `tasks` is still alive, or every grant
-	/// reports a bare cancel instead. `None` when the peer never negotiated AUTH,
-	/// which leaves no serve task to settle.
-	handle: Option<crate::auth::Link>,
+}
+
+/// Where [`Dispatch`] is: requests decode differently by what the peer's SETUP negotiates.
+enum DispatchState {
+	/// Waiting on the peer's SETUP, with what each request will be served by.
+	Setup {
+		publisher: Publisher,
+		subscriber: Subscriber,
+		/// Answers the peer's Auth requests, on the versions that can negotiate them.
+		serve: Option<auth::Serve>,
+	},
+	/// Accepting requests, each served with what the peer's SETUP negotiated.
+	Accepting(Bidi),
 }
 
 impl Dispatch {
@@ -1474,94 +1481,91 @@ impl Dispatch {
 			session,
 			version,
 			peer_setup,
-			start: Some((publisher, subscriber, serve)),
-			bidi: None,
+			state: DispatchState::Setup {
+				publisher,
+				subscriber,
+				serve,
+			},
 			tasks: TaskSet::owned(),
 			fatal: Default::default(),
 			budget: kio::coop::Budget::new(32),
 			charge: true,
-			handle: None,
 		}
 	}
 
-	/// Build what each request is served with, once the peer's SETUP arrives.
-	fn poll_setup(&mut self, waiter: &kio::Waiter) -> Poll<()> {
-		let (_, subscriber, _) = self.start.as_ref().expect("set up once");
-
-		// PUBLISH_NAMESPACE decodes differently once the MoQ Cluster extension is
-		// negotiated, so dispatch waits for SETUP, which `Unis` reads independently.
-		let peer = ready!(subscriber.poll_peer(waiter));
-
-		// An AUTH without MoQ Auth negotiated, offered by both sides, is an unknown request,
-		// which falls through to the protocol violation in `BidiServe`.
-		let negotiated = ready!(self.peer_setup.poll_get(waiter)).auth;
-
-		// From the same slot, so this costs nothing extra: it decides whether an unsolicited
-		// advertisement is the peer ignoring our own SETUP (MoQ Solicit).
-		let declared = ready!(subscriber.poll_solicit(waiter));
-
-		let (publisher, subscriber, serve) = self.start.take().expect("set up once");
-		let serve = match negotiated {
-			true => serve,
-			false => None,
-		};
-		self.handle = serve.as_ref().map(|serve| serve.handle.clone());
-		self.bidi = Some(Bidi {
-			publisher,
-			subscriber,
-			peer,
-			declared,
-			serve,
-			version: self.version,
-		});
-		Poll::Ready(())
-	}
-
 	/// `Ready` only with the error that ends the session.
 	fn poll(&mut self, waiter: &kio::Waiter) -> Poll<Error> {
-		if self.bidi.is_none() {
-			ready!(self.poll_setup(waiter));
-		}
-		let bidi = self.bidi.as_ref().expect("set up above");
 		loop {
-			let accepted = 'accept: {
-				let broken = self.fatal.poll(waiter, |fatal| match fatal.is_some() {
-					true => Poll::Ready(()),
-					false => Poll::Pending,
-				});
-				if let Poll::Ready(mut fatal) = broken
-					&& let Some(err) = fatal.take()
-				{
-					break 'accept Poll::Ready(Err(err));
-				}
-				if std::mem::take(&mut self.charge) && self.budget.poll_yield(waiter).is_pending() {
-					break 'accept Poll::Pending;
-				}
-				Stream::poll_accept(&self.session, self.version, waiter)
-			};
-			match accepted {
-				// QUIC may expose an earlier stream before its bytes arrive. Reading the
-				// header here could block later streams whose unread data holds flow control.
-				Poll::Ready(Ok(stream)) => {
-					self.charge = true;
-					self.tasks
-						.push(BidiServe::new(bidi.clone(), stream, self.fatal.clone()));
+			match &self.state {
+				DispatchState::Setup {
+					publisher,
+					subscriber,
+					serve,
+				} => {
+					// PUBLISH_NAMESPACE decodes differently once the MoQ Cluster extension is
+					// negotiated, so dispatch waits for SETUP, which `Unis` reads independently.
+					let peer = ready!(subscriber.poll_peer(waiter));
+
+					// An AUTH without MoQ Auth negotiated, offered by both sides, is an unknown
+					// request, which falls through to the protocol violation in `BidiServe`.
+					let negotiated = ready!(self.peer_setup.poll_get(waiter)).auth;
+
+					// From the same slot, so this costs nothing extra: it decides whether an
+					// unsolicited advertisement is the peer ignoring our own SETUP (MoQ Solicit).
+					let declared = ready!(subscriber.poll_solicit(waiter));
+
+					let bidi = Bidi {
+						publisher: publisher.clone(),
+						subscriber: subscriber.clone(),
+						peer,
+						declared,
+						serve: serve.clone().filter(|_| negotiated),
+						version: self.version,
+					};
+					self.state = DispatchState::Accepting(bidi);
 				}
-				Poll::Ready(Err(err)) => {
-					// `tasks` still holds the AUTH serve tasks, so the session's error lands on
-					// the handle before they drop. The driver closes it again with the same
-					// error, which then does nothing.
-					if let Some(handle) = &self.handle {
-						handle.close(err.clone());
+				DispatchState::Accepting(bidi) => loop {
+					let accepted = 'accept: {
+						let broken = self.fatal.poll(waiter, |fatal| match fatal.is_some() {
+							true => Poll::Ready(()),
+							false => Poll::Pending,
+						});
+						if let Poll::Ready(mut fatal) = broken
+							&& let Some(err) = fatal.take()
+						{
+							break 'accept Poll::Ready(Err(err));
+						}
+						if std::mem::take(&mut self.charge) && self.budget.poll_yield(waiter).is_pending() {
+							break 'accept Poll::Pending;
+						}
+						Stream::poll_accept(&self.session, self.version, waiter)
+					};
+					match accepted {
+						// QUIC may expose an earlier stream before its bytes arrive. Reading the header
+						// here could block later streams whose unread data holds flow control.
+						Poll::Ready(Ok(stream)) => {
+							self.charge = true;
+							self.tasks
+								.push(BidiServe::new(bidi.clone(), stream, self.fatal.clone()));
+						}
+						Poll::Ready(Err(err)) => {
+							// Each AUTH serve task settles `Issued::closed` from the handle as it
+							// drops, and the first reason wins: closing it while `tasks` still holds
+							// them keeps every grant from reporting a bare cancel. The driver closes
+							// it again with the same error, which then does nothing.
+							if let Some(serve) = &bidi.serve {
+								serve.handle.close(err.clone());
+							}
+							return Poll::Ready(err);
+						}
+						Poll::Pending => {
+							// The children never end the loop; a `Ready` here just means they're
+							// drained until the next stream.
+							let _ = self.tasks.poll(waiter);
+							return Poll::Pending;
+						}
 					}
-					return Poll::Ready(err);
-				}
-				Poll::Pending => {
-					// The children never end the loop; a `Ready` here just means they're
-					// drained until the next stream.
-					let _ = self.tasks.poll(waiter);
-					return Poll::Pending;
-				}
+				},
 			}
 		}
 	}
diff --git a/rs/moq-net/src/lite/session.rs b/rs/moq-net/src/lite/session.rs
index f5171b1cc3e..485e4be4e84 100644
--- a/rs/moq-net/src/lite/session.rs
+++ b/rs/moq-net/src/lite/session.rs
@@ -415,7 +415,7 @@ pub fn start(config: Config) -> Result<SessionStart, Error> {
 			.then(|| SendSetup::new(session.clone(), our_setup, version, stats.clone())),
 		goaway: Some(SendGoaway::new(session.clone(), goaway, version, stats.clone())),
 		session_stream: setup_stream.map(|stream| stream.with_stats(&stats)),
-		ending: None,
+		ended: None,
 		publisher,
 		subscriber: SubscriberDriver::new(subscriber, early, client),
 		session,
@@ -448,8 +448,8 @@ pub(crate) struct Driver {
 	/// The legacy session stream (pre-lite-03). Only its *error* ends the race, so
 	/// the publisher and subscriber keep running while it sits idle.
 	session_stream: Option<Stream<Transport, Version>>,
-	/// The first terminal outcome, retained while received streams drain.
-	ending: Option<Result<(), Error>>,
+	/// How the session ended, once it did: received streams drain, then this is the result.
+	ended: Option<Result<(), Error>>,
 	publisher: Publisher,
 	subscriber: SubscriberDriver,
 	/// For the terminal close: the machine's last act reports the outcome to
@@ -462,39 +462,18 @@ pub(crate) struct Driver {
 
 impl Driver {
 	pub(crate) fn poll(&mut self, waiter: &kio::Waiter) -> Poll<Result<(), Error>> {
-		if self.ending.is_none() {
-			let res = std::task::ready!(self.poll_protocol(waiter));
-			self.auth.handle.close(match &res {
-				Ok(()) => Error::Cancel,
-				Err(err) => err.clone(),
-			});
-			match &res {
-				Err(err @ (Error::Transport(_) | Error::IdleTimeout)) => {
-					tracing::info!("session terminated");
-					self.session.abort(err, SessionError::Internal.to_code(), "");
-				}
-				Err(err) => {
-					tracing::warn!(%err, "session error");
-					self.session.abort(
-						err,
-						crate::lite::error::to_session_code(&SessionError::from(err)),
-						err.to_string().as_ref(),
-					);
-				}
-				_ => {
-					tracing::info!("session closed");
-					self.session.abort(&Error::Cancel, SessionError::Cancel.to_code(), "");
-				}
+		let res = match &self.ended {
+			Some(res) => res.clone(),
+			None => {
+				let res = std::task::ready!(self.poll_protocol(waiter));
+				self.end(res)
 			}
-			self.ending = Some(res);
-			self.withholding = None;
-		}
+		};
 		// Closing the transport first bounds every read, including when a control
 		// stream ended the protocol while the connection itself was still live.
 		// Keep control machines and their subscription entries until data drains,
 		// whichever side closed it.
 		std::task::ready!(self.subscriber.poll_drain(waiter));
-		let res = self.ending.take().expect("draining a terminal session");
 		// A clean close, by dropping or draining here or by the peer, cuts off what is
 		// still arriving with the cancelled session either way.
 		self.subscriber.abort(match &res {
@@ -504,6 +483,35 @@ impl Driver {
 		Poll::Ready(res)
 	}
 
+	/// Close the transport over the protocol's result `res`, which ends the session.
+	fn end(&mut self, res: Result<(), Error>) -> Result<(), Error> {
+		self.auth.handle.close(match &res {
+			Ok(()) => Error::Cancel,
+			Err(err) => err.clone(),
+		});
+		match &res {
+			Err(err @ (Error::Transport(_) | Error::IdleTimeout)) => {
+				tracing::info!("session terminated");
+				self.session.abort(err, SessionError::Internal.to_code(), "");
+			}
+			Err(err) => {
+				tracing::warn!(%err, "session error");
+				self.session.abort(
+					err,
+					crate::lite::error::to_session_code(&SessionError::from(err)),
+					err.to_string().as_ref(),
+				);
+			}
+			_ => {
+				tracing::info!("session closed");
+				self.session.abort(&Error::Cancel, SessionError::Cancel.to_code(), "");
+			}
+		}
+		self.ended = Some(res.clone());
+		self.withholding = None;
+		res
+	}
+
 	/// Start withdrawing this session's announcements.
 	pub(crate) fn close(&self) {
 		self.publisher.close();
diff --git a/rs/moq-net/src/lite/setup.rs b/rs/moq-net/src/lite/setup.rs
index 712ae816dcc..1fa835e333e 100644
--- a/rs/moq-net/src/lite/setup.rs
+++ b/rs/moq-net/src/lite/setup.rs
@@ -408,8 +408,7 @@ struct PeerSetupState {
 	address: crate::transport::Address,
 	/// Whether our own SETUP offers the `ietf` extension.
 	ietf: bool,
-	seen: bool,
-	setup: Option<Setup>,
+	peer: Peer,
 	/// Where the peer's advertised implementation is published for the session handle.
 	implementation: crate::setup::PeerImplementation,
 	/// The session's auth link to tell once the peer's SETUP accepts the connection.
@@ -417,24 +416,62 @@ struct PeerSetupState {
 	/// What the peer's route key proof must sign, or `None` when the transport cannot bind
 	/// one to this session, so any proof leaves the peer unproven.
 	transcript: Option<crate::route::Transcript>,
-	/// The route key the peer proved.
-	route_key: Option<[u8; 32]>,
+}
+
+/// Where the peer's SETUP stands.
+enum Peer {
+	/// No Setup Stream yet.
+	Waiting,
+	/// The Setup Stream is claimed and its body still decoding, so no other may arrive.
+	Claimed,
+	/// The peer's SETUP, with the route key it proved, if any.
+	Received {
+		setup: Box<Setup>,
+		route_key: Option<[u8; 32]>,
+	},
 }
 
 impl PeerSetupState {
-	/// Keep the key the peer's SETUP proves, or refuse a proof whose signature fails.
-	fn prove(&mut self, setup: &Setup) -> Result<(), crate::Error> {
-		let (Some(proof), Some(transcript)) = (&setup.route_key, &self.transcript) else {
-			return Ok(());
-		};
-		// The peer holds the side of the session we do not.
-		if !transcript.verify(proof, !self.client) {
+	/// Record the peer's SETUP, refusing a second one or a route key proof whose signature
+	/// fails.
+	fn set(&mut self, setup: Setup) -> Result<(), crate::Error> {
+		if let Peer::Received { .. } = self.peer {
 			return Err(crate::Error::ProtocolViolation);
 		}
-		tracing::debug!(proof = ?proof, "peer proved its route key");
-		self.route_key = Some(proof.key);
+		let route_key = match (&setup.route_key, &self.transcript) {
+			(Some(proof), Some(transcript)) => {
+				// The peer holds the side of the session we do not.
+				if !transcript.verify(proof, !self.client) {
+					return Err(crate::Error::ProtocolViolation);
+				}
+				tracing::debug!(proof = ?proof, "peer proved its route key");
+				Some(proof.key)
+			}
+			_ => None,
+		};
+		self.implementation.set(setup.implementation.clone());
+		self.peer = Peer::Received {
+			setup: Box::new(setup),
+			route_key,
+		};
 		Ok(())
 	}
+
+	/// The peer's SETUP, once it arrived.
+	fn setup(&self) -> Option<&Setup> {
+		match &self.peer {
+			Peer::Received { setup, .. } => Some(setup),
+			Peer::Waiting | Peer::Claimed => None,
+		}
+	}
+
+	/// The route key the peer's SETUP proved, once it arrived.
+	fn route_key(&self) -> Option<[u8; 32]> {
+		match self.peer {
+			Peer::Received { route_key, .. } => route_key,
+			Peer::Waiting | Peer::Claimed => None,
+		}
+	}
 }
 
 impl PeerSetup {
@@ -443,12 +480,10 @@ impl PeerSetup {
 			client,
 			address,
 			ietf: false,
-			seen: false,
-			setup: None,
+			peer: Peer::Waiting,
 			implementation: Default::default(),
 			accept: None,
 			transcript: None,
-			route_key: None,
 		}))
 	}
 
@@ -478,10 +513,7 @@ impl PeerSetup {
 		{
 			return Err(crate::Error::ProtocolViolation);
 		}
-		state.prove(&received.setup)?;
-		state.seen = true;
-		state.implementation.set(received.setup.implementation.clone());
-		state.setup = Some(received.setup);
+		state.set(received.setup)?;
 		let accept = state.accept.take();
 		drop(state);
 		if let Some(link) = accept {
@@ -511,43 +543,40 @@ impl PeerSetup {
 		let state = self.0.lock();
 		match state.ietf {
 			false => Some(false),
-			true => state.setup.as_ref().map(|setup| setup.ietf),
+			true => state.setup().map(|setup| setup.ietf),
 		}
 	}
 
 	/// Record the peer's SETUP, refusing a route key proof that does not hold.
 	pub fn set(&self, setup: Setup) -> Result<(), crate::Error> {
-		let mut state = self.0.lock();
-		state.prove(&setup)?;
-		state.seen = true;
-		state.implementation.set(setup.implementation.clone());
-		state.setup = Some(setup);
-		Ok(())
+		self.0.lock().set(setup)
 	}
 
 	/// The route key the peer proved, once its SETUP arrived.
 	pub fn route_key(&self) -> Option<[u8; 32]> {
-		self.0.lock().route_key
+		self.0.lock().route_key()
 	}
 
 	/// Poll for the route key the peer proved, waiting until its SETUP arrives.
 	/// `None` when it proved none.
 	pub fn poll_route_key(&self, waiter: &kio::Waiter) -> std::task::Poll<Option<[u8; 32]>> {
-		let state = std::task::ready!(self.0.poll(waiter, |state| match state.setup.is_some() {
-			true => std::task::Poll::Ready(()),
-			false => std::task::Poll::Pending,
+		let state = std::task::ready!(self.0.poll(waiter, |state| match state.setup() {
+			Some(_) => std::task::Poll::Ready(()),
+			None => std::task::Poll::Pending,
 		}));
-		std::task::Poll::Ready(state.route_key)
+		std::task::Poll::Ready(state.route_key())
 	}
 
 	/// Claim the Setup Stream before decoding its body, refusing any second stream.
 	pub fn claim(&self) -> Result<(), crate::Error> {
 		let mut state = self.0.lock();
-		if state.seen {
-			return Err(crate::Error::ProtocolViolation);
+		match state.peer {
+			Peer::Waiting => {
+				state.peer = Peer::Claimed;
+				Ok(())
+			}
+			Peer::Claimed | Peer::Received { .. } => Err(crate::Error::ProtocolViolation),
 		}
-		state.seen = true;
-		Ok(())
 	}
 
 	/// Poll for the peer's advertised probe level, waiting until its SETUP arrives.
@@ -573,14 +602,11 @@ impl PeerSetup {
 	/// Pends forever if it never does; the caller is session state, dropped with the
 	/// driver.
 	fn poll_get<T>(&self, waiter: &kio::Waiter, f: impl FnOnce(&Setup) -> T) -> std::task::Poll<T> {
-		let slot = std::task::ready!(self.0.poll(waiter, |setup| {
-			if setup.setup.is_some() {
-				std::task::Poll::Ready(())
-			} else {
-				std::task::Poll::Pending
-			}
+		let state = std::task::ready!(self.0.poll(waiter, |state| match state.setup() {
+			Some(_) => std::task::Poll::Ready(()),
+			None => std::task::Poll::Pending,
 		}));
-		std::task::Poll::Ready(f(slot.setup.as_ref().expect("waited for Some")))
+		std::task::Poll::Ready(f(state.setup().expect("judged under this lock")))
 	}
 }
 
diff --git a/rs/moq-net/src/model/listing.rs b/rs/moq-net/src/model/listing.rs
index 209a5086ccf..f0a26891b54 100644
--- a/rs/moq-net/src/model/listing.rs
+++ b/rs/moq-net/src/model/listing.rs
@@ -85,9 +85,6 @@ pub(super) struct ListingState {
 	live: LinkedHashMap<Arc<str>, Entry>,
 	/// The queue of every reader of the current answer.
 	readers: slab::Slab<kio::Shared<Queue>>,
-	/// Bumped whenever every reader is detached at once, so a reader of an earlier
-	/// answer releases nothing on drop.
-	generation: u64,
 	/// Each owner's reader, by its key in `readers`.
 	owners: HashMap<Owner, usize>,
 	/// Whether a [`Lister`] exists. While one does, readers wait for its answer to sync.
@@ -109,31 +106,28 @@ impl ListingState {
 	}
 
 	/// Start a reader with every listed track queued, ending `owner`'s older one.
-	fn read(&mut self, owner: Option<Owner>) -> (kio::Shared<Queue>, usize, u64) {
+	fn read(&mut self, owner: Option<Owner>) -> (kio::Shared<Queue>, usize) {
 		let mut queue = Queue::default();
 		for (name, entry) in &self.live {
 			queue.pending.insert(name.clone(), Pending::Add(entry.info.clone()));
 		}
 		if self.synced() {
-			queue.bound = Some(queue.pending.len());
+			queue.sync = SyncPhase::Counting(queue.pending.len());
 		}
 		let queue = kio::Shared::new(queue);
 		let key = self.readers.insert(queue.clone());
 		if let Some(older) = owner.and_then(|owner| self.owners.insert(owner, key)) {
-			let older = self.readers.remove(older);
-			let mut older = older.lock();
-			older.end = Some(Err(Error::Cancel));
-			older.detached = true;
+			self.readers.remove(older).lock().end = Some(Err(Error::Cancel));
 		}
-		(queue, key, self.generation)
+		(queue, key)
 	}
 
-	fn release(&mut self, key: usize, generation: u64, owner: Option<Owner>, queue: &kio::Shared<Queue>) {
-		// A replaced reader already left, and its key may be another's now.
-		if generation != self.generation || queue.read().detached {
+	fn release(&mut self, key: usize, owner: Option<Owner>, queue: &kio::Shared<Queue>) {
+		// A replaced or ended reader already left, and its key may be another's now.
+		if !self.readers.get(key).is_some_and(|current| current.same_channel(queue)) {
 			return;
 		}
-		self.readers.try_remove(key);
+		self.readers.remove(key);
 		if let Some(owner) = owner {
 			self.owners.remove(&owner);
 		}
@@ -270,7 +264,6 @@ impl ListingState {
 		self.each(|queue| queue.end = Some(end.clone()));
 		self.readers.clear();
 		self.owners.clear();
-		self.generation += 1;
 	}
 
 	/// The broadcast ended: so does every listing of it, first unlisting each track its
@@ -364,13 +357,23 @@ enum Pending {
 #[derive(Default)]
 pub(super) struct Queue {
 	pending: LinkedHashMap<Arc<str>, Pending>,
-	/// Events left to send before SYNCED, once the listing has synced.
-	bound: Option<usize>,
-	/// SYNCED was sent.
-	synced: bool,
+	sync: SyncPhase,
+	/// How the listing ended for this reader, once it did: its broadcast ended, or a newer
+	/// reader of the same owner replaced it.
 	end: Option<Result<(), Error>>,
-	/// A newer reader of the same owner replaced this one.
-	detached: bool,
+}
+
+/// Where one reader's SYNCED stands.
+#[derive(Default)]
+enum SyncPhase {
+	/// The listing has not synced.
+	#[default]
+	Waiting,
+	/// The listing synced: SYNCED goes out after this many more events, or once the queue
+	/// drains if that comes first.
+	Counting(usize),
+	/// SYNCED went out.
+	Sent,
 }
 
 impl Queue {
@@ -415,20 +418,20 @@ impl Queue {
 	/// The listing synced: SYNCED goes out once what is queued now has been read, or
 	/// earlier if the queue drains first.
 	fn sync(&mut self) {
-		if !self.synced && self.bound.is_none() {
-			self.bound = Some(self.events());
+		if let SyncPhase::Waiting = self.sync {
+			self.sync = SyncPhase::Counting(self.events());
 		}
 	}
 
 	/// The listing's answer restarted before this reader was told it synced.
 	fn unsync(&mut self) {
-		if !self.synced {
-			self.bound = None;
+		if let SyncPhase::Counting(_) = self.sync {
+			self.sync = SyncPhase::Waiting;
 		}
 	}
 
 	fn is_ready(&self) -> bool {
-		self.end.is_some() || !self.pending.is_empty() || (!self.synced && self.bound.is_some())
+		self.end.is_some() || !self.pending.is_empty() || matches!(self.sync, SyncPhase::Counting(_))
 	}
 
 	fn next(&mut self) -> Result<Option<TracksEvent>, Error> {
@@ -449,13 +452,15 @@ impl Queue {
 			}
 			None => {}
 		}
-		if !self.synced && (self.bound == Some(0) || (self.bound.is_some() && self.pending.is_empty())) {
-			self.synced = true;
+		if let SyncPhase::Counting(left) = self.sync
+			&& (left == 0 || self.pending.is_empty())
+		{
+			self.sync = SyncPhase::Sent;
 			return Ok(Some(TracksEvent::Synced));
 		}
 		let (name, pending) = self.pending.pop_front().expect("a ready queue has a pending change");
-		if let Some(bound) = self.bound.as_mut().filter(|_| !self.synced) {
-			*bound -= 1;
+		if let SyncPhase::Counting(left) = &mut self.sync {
+			*left -= 1;
 		}
 		Ok(Some(match pending {
 			Pending::Add(info) => TracksEvent::Added { name, info },
@@ -486,7 +491,6 @@ pub struct Tracks {
 	broadcast: Consumer,
 	queue: kio::Shared<Queue>,
 	key: usize,
-	generation: u64,
 	owner: Option<Owner>,
 }
 
@@ -513,7 +517,7 @@ impl Drop for Tracks {
 		let mut state = self.broadcast.state.lock();
 		state
 			.listing
-			.release(self.key, self.generation, self.owner, &self.queue);
+			.release(self.key, self.owner, &self.queue);
 	}
 }
 
@@ -542,13 +546,12 @@ impl Consumer {
 		if state.closing {
 			return Err(Error::Unroutable);
 		}
-		let (queue, key, generation) = state.listing.read(owner);
+		let (queue, key) = state.listing.read(owner);
 		drop(state);
 		Ok(Tracks {
 			broadcast: self.clone(),
 			queue,
 			key,
-			generation,
 			owner,
 		})
 	}
@@ -1112,7 +1115,7 @@ mod tests {
 		let mut listing = ListingState::default();
 		let first = listing.list_own("a".into(), track::Info::default());
 		listing.answer_add("a".into(), track::Info::default()).unwrap();
-		let (queue, _, _) = listing.read(None);
+		let (queue, _) = listing.read(None);
 		assert!(matches!(queue.lock().next(), Ok(Some(TracksEvent::Added { .. }))));
 		// The new track of the name lists before the earlier one unlists.
 		listing.list_own("a".into(), track::Info::default());

From 5c24eb306ad9d4be4c57ce2217ea0460af4fc34a Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Sun, 8 Nov 2026 00:35:37 -0800
Subject: [PATCH 03/17] refactor(net): js/net connections hold one close phase

Open, closing with its drain, or closed: lite's abort() now runs once, and
the IETF connection's closing promise and closed flag become one value.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 js/net/src/ietf/connection.ts | 43 ++++++++++++++++++++---------------
 js/net/src/lite/connection.ts | 30 +++++++++++++++++-------
 2 files changed, 47 insertions(+), 26 deletions(-)

diff --git a/js/net/src/ietf/connection.ts b/js/net/src/ietf/connection.ts
index c332e1c41b9..9a5220a2753 100644
--- a/js/net/src/ietf/connection.ts
+++ b/js/net/src/ietf/connection.ts
@@ -41,13 +41,16 @@ function handed(task: Promise<void>): Promise<void> {
 	return task.catch((err: unknown) => console.error("error processing bidi stream", err));
 }
 
+/** Where a session is: open, delivering what it owes before it closes, or closed. */
+type Phase = { kind: "open" } | { kind: "closing"; done: Promise<void> } | { kind: "closed" };
+
 /**
  * Represents a connection to a MoQ server using moq-transport protocol.
  *
  * @internal
  */
 export class Connection implements Established {
-	#closing?: Promise<void>;
+	#phase: Phase = { kind: "open" };
 	// The URL of the connection.
 	readonly url: URL;
 
@@ -95,9 +98,6 @@ export class Connection implements Established {
 	// The peer's GOAWAY, read off its SETUP stream.
 	#goaway = new Once<Drain>();
 
-	// Just to avoid logging when `close()` is called.
-	#closed = false;
-
 	/** The implementation the peer advertised; see {@link Established.peerImplementation}. */
 	readonly peerImplementation: Promise<string | undefined>;
 
@@ -189,7 +189,7 @@ export class Connection implements Established {
 				subscribe: new Path.Patterns(publish ? [Path.Pattern.all()] : []),
 			},
 			handle,
-			draining: () => this.#goaway.peek() !== undefined || this.#closing !== undefined,
+			draining: () => this.#goaway.peek() !== undefined || this.#phase.kind !== "open",
 		});
 		// Left before anything observes the close, so a successor on the same handle never
 		// finds this session still active.
@@ -245,25 +245,32 @@ export class Connection implements Established {
 
 	/** Withdraw announcements and wait up to one second for delivery before closing. */
 	close(): Promise<void> {
-		this.#closing ??= withTimeout(
-			this.#publisher.withdraw(),
-			Nano.fromSecond(Second(1)),
-			"session close timed out",
-		).finally(() => this.abort());
-		return this.#closing;
+		switch (this.#phase.kind) {
+			case "open": {
+				const done = withTimeout(
+					this.#publisher.withdraw(),
+					Nano.fromSecond(Second(1)),
+					"session close timed out",
+				).finally(() => this.abort());
+				this.#phase = { kind: "closing", done };
+				return done;
+			}
+			case "closing":
+				return this.#phase.done;
+			case "closed":
+				return Promise.resolve();
+		}
 	}
 
 	/** End the session immediately without waiting for delivery. */
 	abort(): void {
-		if (this.#closed) return;
 		this.#close();
 	}
 
 	// Close with the session code the peer should see, a clean close by default.
 	#close(info?: WebTransportCloseInfo) {
-		if (this.#closed) return;
-
-		this.#closed = true;
+		if (this.#phase.kind === "closed") return;
+		this.#phase = { kind: "closed" };
 
 		this.#publisher.close();
 		this.#auth.close();
@@ -301,7 +308,7 @@ export class Connection implements Established {
 			if (this.#client) tasks.push(this.#publisher.runEnforce(this.#auth.setupAnswered(), this.#auth));
 			await Promise.all(tasks);
 		} catch (err) {
-			if (!this.#closed) {
+			if (this.#phase.kind !== "closed") {
 				console.error("fatal error running connection", err);
 			}
 		} finally {
@@ -309,7 +316,7 @@ export class Connection implements Established {
 			// a tracked withdrawal, so a failure in it ends this driver while close() is
 			// still waiting on the sibling loops; closing here would drop their
 			// withdrawals. close() aborts once its barrier settles or the deadline hits.
-			if (!this.#closing) this.#close();
+			if (this.#phase.kind !== "closing") this.#close();
 		}
 	}
 
@@ -573,7 +580,7 @@ export class Connection implements Established {
 			});
 			await race([read(), stopped]);
 		} catch (err) {
-			if (!this.#closed && !sessionClosed(err))
+			if (this.#phase.kind !== "closed" && !sessionClosed(err))
 				this.#violated(
 					err instanceof ProtocolViolation
 						? err
diff --git a/js/net/src/lite/connection.ts b/js/net/src/lite/connection.ts
index 5ea2d4a8068..abbb6c66ab8 100644
--- a/js/net/src/lite/connection.ts
+++ b/js/net/src/lite/connection.ts
@@ -82,13 +82,16 @@ export interface ConnectionProps {
 	probeTarget?: Getter<number>;
 }
 
+/** Where a session is: open, delivering what it owes before it closes, or closed. */
+type Phase = { kind: "open" } | { kind: "closing"; done: Promise<void> } | { kind: "closed" };
+
 /**
  * Represents a connection to a MoQ server.
  *
  * @internal
  */
 export class Connection implements Established {
-	#closing?: Promise<void>;
+	#phase: Phase = { kind: "open" };
 	// The URL of the connection.
 	readonly url: URL;
 
@@ -214,7 +217,7 @@ export class Connection implements Established {
 				subscribe: new Path.Patterns(publish ? [Path.Pattern.all()] : []),
 			},
 			handle,
-			draining: () => this.#goaway.peek() !== undefined || this.#closing !== undefined,
+			draining: () => this.#goaway.peek() !== undefined || this.#phase.kind !== "open",
 		});
 		// Left before anything observes the close, so a successor on the same handle never
 		// finds this session still active.
@@ -276,16 +279,27 @@ export class Connection implements Established {
 	 * its own, so it holds the close until the deadline.
 	 */
 	close(): Promise<void> {
-		this.#closing ??= withTimeout(
-			this.#publisher.drain(),
-			Nano.fromMilli(Milli(1000)),
-			"session close timed out",
-		).finally(() => this.abort());
-		return this.#closing;
+		switch (this.#phase.kind) {
+			case "open": {
+				const done = withTimeout(
+					this.#publisher.drain(),
+					Nano.fromMilli(Milli(1000)),
+					"session close timed out",
+				).finally(() => this.abort());
+				this.#phase = { kind: "closing", done };
+				return done;
+			}
+			case "closing":
+				return this.#phase.done;
+			case "closed":
+				return Promise.resolve();
+		}
 	}
 
 	/** End the session immediately without waiting for delivery. */
 	abort(): void {
+		if (this.#phase.kind === "closed") return;
+		this.#phase = { kind: "closed" };
 		this.#auth.close();
 		this.#publisher.close();
 		// Received tracks end in #run once what already arrived drains, as on the peer's close.

From 7fc7edda501e082651e9873720fb4b4400eb47d5 Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Sun, 8 Nov 2026 00:38:59 -0800
Subject: [PATCH 04/17] style: rustfmt

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 rs/moq-net/src/ietf/publisher.rs | 5 ++++-
 rs/moq-net/src/model/listing.rs  | 4 +---
 2 files changed, 5 insertions(+), 4 deletions(-)

diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs
index 223224ef35a..ad7c2ff9e04 100644
--- a/rs/moq-net/src/ietf/publisher.rs
+++ b/rs/moq-net/src/ietf/publisher.rs
@@ -12193,7 +12193,10 @@ mod serve_tests {
 		update.encode(&mut encoder, version).unwrap();
 		request.push(&bytes);
 		for _ in 0..200 {
-			assert!(futures::poll!(serving.as_mut()).is_pending(), "the fill holds the request");
+			assert!(
+				futures::poll!(serving.as_mut()).is_pending(),
+				"the fill holds the request"
+			);
 		}
 
 		*credit.write().ok().expect("credit is live") = true;
diff --git a/rs/moq-net/src/model/listing.rs b/rs/moq-net/src/model/listing.rs
index f0a26891b54..90147b41dfa 100644
--- a/rs/moq-net/src/model/listing.rs
+++ b/rs/moq-net/src/model/listing.rs
@@ -515,9 +515,7 @@ impl Tracks {
 impl Drop for Tracks {
 	fn drop(&mut self) {
 		let mut state = self.broadcast.state.lock();
-		state
-			.listing
-			.release(self.key, self.owner, &self.queue);
+		state.listing.release(self.key, self.owner, &self.queue);
 	}
 }
 

From 22bd22d5382903fa34270659c4f7b28fa9c07048 Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Sun, 8 Nov 2026 00:43:41 -0800
Subject: [PATCH 05/17] quest: net-phases keeps the items that overlapped open
 PRs

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 quest/m1/explicit-state/net-phases.md | 29 +++++++++++++--------------
 1 file changed, 14 insertions(+), 15 deletions(-)

diff --git a/quest/m1/explicit-state/net-phases.md b/quest/m1/explicit-state/net-phases.md
index 8732175d782..35d89e9b45c 100644
--- a/quest/m1/explicit-state/net-phases.md
+++ b/quest/m1/explicit-state/net-phases.md
@@ -11,9 +11,22 @@ or call ordering stands in for a phase. A placeholder such as
 
 The shared decisions are in the [epic](/quest/m1/explicit-state/README.md).
 
-No reachable bug is folded into this child. Check each item below while
+No reachable bug is folded into what remains. Check each item below while
 redesigning it, and reproduce any inconsistent state you find.
 
+[moq-dev/moq#8275](https://github.com/moq-dev/moq/pull/8275) landed the IETF
+publisher's `TrackServe` phase and its reopened-tail fix, the datagram
+forward, the session drivers' terminal result, IETF `Dispatch`, lite
+`PeerSetup`, the listing `Queue`, and the js/net connections' close phase.
+What remains overlapped open PRs when that one started:
+[moq-dev/moq#8220](https://github.com/moq-dev/moq/pull/8220) (lite publisher
+and subscriber, `fills.rs`, js/net lite),
+[moq-dev/moq#8063](https://github.com/moq-dev/moq/pull/8063) (lite announce
+and subscriber, `model/origin.rs`, `model/follow.rs`, js/net `origin.ts`),
+and [moq-dev/moq#8222](https://github.com/moq-dev/moq/pull/8222) (js/net
+`reload.ts` and lite `subscriber.ts`). Start each item once the PR touching
+its file has landed or closed.
+
 Rust:
 
 - **IETF `SubscribeOnce`/`Established`** (`rs/moq-net/src/ietf/subscriber.rs`):
@@ -24,13 +37,6 @@ Rust:
     duplicate one another.
   - `request_id: RequestId(0)` is a placeholder for an id that isn't
     assigned yet.
-- **IETF publisher** (`rs/moq-net/src/ietf/publisher.rs`):
-  - `RunSubscription.served` caches the track's drained state.
-  - An update that reopens `TrackServe` (`draining = false`, a new `tail`)
-    is never polled again. Reproduce this first; it is the audit's plausible
-    finding.
-  - `TrackServe.draining`/`end` and `DatagramForward` should become explicit
-    phases.
 - **Lite subscriber** (`rs/moq-net/src/lite/subscriber.rs`):
   - `ServeLoop.finishing` and `parked` belong inside `SubStream`.
   - `SubStream.served` plus `started` are really `Option<Option<u64>>`.
@@ -40,12 +46,7 @@ Rust:
 - **`Following`** (`rs/moq-net/src/model/follow.rs`): eight fields, of which
   `last`, `moved`, and `absorbed` are always reset together.
 - **Smaller ones**:
-  - `ietf/session.rs` `Dispatch` holds two `Option`s, exactly one of which is
-    set.
   - `model/origin.rs` `TrackIo` keeps `request` and `accepted` side by side.
-  - The driver `ending` field.
-  - The lite `PeerSetupState`.
-  - The listing `Queue`.
   - `fills.rs` `Recovery`.
 
 JS (`js/net`):
@@ -67,8 +68,6 @@ JS (`js/net`):
   encoded by about 40 `?.` guards.
 - **`Join`** (`ietf/join.ts`): a second fetch stream after the track ended
   leaks the first producer.
-- **Connections** (`lite/connection.ts`, `ietf/connection.ts`): lite's
-  `abort()` has no guard.
 
 Keep the lite `TrackRun` start state out of this child. It belongs to
 [Delete the old start bookkeeping](/quest/m2/subscribe-ranges/start-bookkeeping.md).

From 26f19c95db21b93dbc347cfde9b9a00b4af98815 Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Sun, 8 Nov 2026 00:59:04 -0800
Subject: [PATCH 06/17] fix(net): a drained js/net IETF subscription serves the
 tail an update reopens

The serving loop ended for good once the track ran out, and the end race
then waited only for the fill. An update parked behind that fill could move
the range back onto a group the cursor never handed out. The cursor still
owed that group, but nothing polled it, so PUBLISH_DONE went out without
the tail.

Serving now ends only once no update or fill is in flight, and polls the
cursor again after each one. This matches moq-net, whose TrackServe
leaves Draining on an update.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 js/net/src/ietf/publisher.test.ts |  50 +++++++
 js/net/src/ietf/publisher.ts      | 216 ++++++++++++++++--------------
 2 files changed, 165 insertions(+), 101 deletions(-)

diff --git a/js/net/src/ietf/publisher.test.ts b/js/net/src/ietf/publisher.test.ts
index 4e00bb0a26d..5655ac60375 100644
--- a/js/net/src/ietf/publisher.test.ts
+++ b/js/net/src/ietf/publisher.test.ts
@@ -4580,6 +4580,56 @@ test("draft-22: narrowing gives up an excluded group's open stream", async () =>
 	}
 });
 
+test("draft-22: an update reopening a drained subscription serves its tail", async () => {
+	const fx = fixture();
+	// The SUBSCRIBE's fill waits for stream credit, holding the request open past the track's end.
+	const credit = Promise.withResolvers<void>();
+	const createUni = fx.pair.server.createUnidirectionalStream.bind(fx.pair.server);
+	spyOn(fx.pair.server, "createUnidirectionalStream").mockImplementationOnce(async (options) => {
+		await credit.promise;
+		return createUni(options);
+	});
+	const track = fx.broadcast.createTrack("video", { timescale: Timescale.MILLI });
+	const open = track.appendGroup();
+	open.writeFrame({ payload: new TextEncoder().encode("head"), timestamp: Timestamp.now() });
+	const { client } = await runSubscribe(
+		fx,
+		new Subscribe({
+			requestId: 0n,
+			trackNamespace: Path.from("test"),
+			trackName: "video",
+			subscriberPriority: 0,
+			filter: { kind: "absolute", startGroup: 5n, startObject: 0n },
+			fill: { filter: { kind: "absolute", startGroup: 0n, startObject: 0n }, rangeFilters: false },
+		}),
+	);
+	try {
+		// The open group stays open: the end falls after it.
+		track.finish();
+		const parameters = new Parameters();
+		parameters.subscriptionFilter = { kind: "absolute", startGroup: 0n, startObject: 0n };
+		parameters.fillParameters = Filter.encodeFill({ rangeFilters: false }, fx.version);
+		await client.writer.u53(SubscribeUpdate.id);
+		await new SubscribeUpdate({ requestId: 2n, parameters }).encode(client.writer, fx.version);
+		credit.resolve();
+		expect(await client.reader.u53()).toBe(RequestOk.id);
+		await RequestOk.decode(client.reader, fx.version);
+		open.writeFrame({ payload: new TextEncoder().encode("tail"), timestamp: Timestamp.now() });
+		open.finish();
+		expect(await client.reader.u53()).toBe(PublishDone.id);
+		const done = await PublishDone.decode(client.reader, fx.version);
+		// Both fills, the tail's group, then END_OF_TRACK once they closed.
+		expect(done.streamCount).toBe(4n);
+		expect((await readServed(fx.uni, 3)).sort()).toEqual(["head", "head", "tail"]);
+		expect(await readEndOfTrack(await nextUni(fx.uni))).toBe(1n);
+	} finally {
+		open.finish();
+		client.close();
+		track.close();
+		fx.close();
+	}
+});
+
 test("draft-22: widening with a fill keeps the open live group", async () => {
 	using clock = manual();
 	const fx = fixture();
diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts
index 7da764fffef..f2e908f5081 100644
--- a/js/net/src/ietf/publisher.ts
+++ b/js/net/src/ietf/publisher.ts
@@ -912,9 +912,10 @@ export class Publisher {
 					resolve();
 				};
 			});
-			let filling = Promise.resolve();
+			// The fill in flight. It never rejects.
+			let filling: Promise<void> | undefined;
 			const startFill = (requestId: bigint, selected: FillServe, source: TrackSubscriber) => {
-				filling = this.#runFill({
+				const fill: Promise<void> = this.#runFill({
 					requestId,
 					priority: fromWire(msg.subscriberPriority),
 					publisherPriority,
@@ -925,8 +926,13 @@ export class Publisher {
 					unsubscribed,
 					cancelled: fillAbort.signal,
 					streams,
+				}).finally(() => {
+					if (filling === fill) filling = undefined;
 				});
+				filling = fill;
 			};
+			// The update being applied, from its decode until it is answered or fails.
+			let updating: Promise<void> | undefined;
 			if (fill && cache) startFill(answer?.requestId ?? msg.requestId, fill, cache);
 
 			// Every group started, until its stream finishes or resets, and the data streams
@@ -959,96 +965,104 @@ export class Publisher {
 					// The requester sent its last message, which ends the request only where FIN cancels.
 					if (framed === undefined) return request.ended;
 					const { params, requestId: ownRequestId } = await this.#decodeUpdate(request, framed);
-					// update() replaces every option, so carry the rest over as Rust does.
-					if (params.subscriberPriority !== undefined) {
-						const priority = fromWire(params.subscriberPriority);
-						track.update({ ...track.subscription.peek(), priority });
-						datagramPriority.set(priority);
-						if (datagramDemand) datagrams?.setPriority(priority);
-					}
-					const maxAge = params.maxAge;
-					if (maxAge !== undefined) {
-						const current = track.subscription.peek();
-						track.update({ ...current, delay: { ...current?.delay, max: Milli(maxAge) } });
-					}
-					// A fill holds one cache view. At most this decoded update waits behind
-					// it; request parsing then backpressures any further fill requests.
-					if (params.fillParameters) await filling;
-					const snapshot = liveEdge(track);
-					this.#applyPublished(msg, params);
-					const selected =
-						msg.fill && Filter.isDraft20(version)
-							? fillRange(msg.fill, msg.filter, snapshot.largest)
-							: undefined;
-					if (params.subscriptionFilter !== undefined || selected) {
-						// An update's unfiltered range explicitly lowers the floor to zero;
-						// an initial unfiltered subscription still joins the live group. A
-						// fill-only update keeps the range it already has.
-						const requested: ServeRange =
-							params.subscriptionFilter === undefined
-								? bounds.peek()
-								: msg.filter.kind === "unfiltered"
-									? { start: { group: 0n, object: 0n } }
-									: filterRange(msg.filter, snapshot);
-						const tail = afterFill(requested, selected, snapshot);
-						range.start = tail.start;
-						range.end = tail.end;
-						bounds.set(requested, true);
-						track.update({ ...track.subscription.peek(), ranges: serveRanges(range) });
-						// Below its floor the live cursor already passed every group or never owned it,
-						// so lowering it for a fill-only update would hand those groups out a second time.
-						let start = range.start ? range.start.group : 0n;
-						if (params.subscriptionFilter === undefined) start = max(start, floor);
-						floor = start;
-						track.withRange({ ...rangeGroups(range), start: { included: start } });
-						// The fill took the current group's head. The cursor never hands a group out
-						// twice, so what its stream no longer carries resumes from the cache.
-						const next = tail !== requested ? tail.start : undefined;
-						const at = next?.group;
-						if (next && at !== undefined && hooks.handedOut(track, at)) {
-							const slice = groupSlice(range, at);
-							// A stream still serving this group from the fill's edge keeps what it
-							// covers, so the tail resumes where that stream stops.
-							const streaming =
-								live?.sequence === at &&
-								live.slice.skip <= slice.skip &&
-								!live.retired &&
-								!excludes(requested, at, live.slice)
-									? live
-									: undefined;
-							const resume = streaming ? streaming.slice.until : slice.skip;
-							if (resume !== undefined) slice.skip = Math.max(slice.skip, resume);
-							const group =
-								resume !== undefined &&
-								(!range.end || range.end.group >= next.group) &&
-								(slice.until === undefined || slice.until > slice.skip)
-									? hooks.cachedGroup(track, at)
-									: undefined;
-							if (group) serveGroup(group, slice);
-						}
-					}
-					if (params.groupOrder !== undefined)
-						track.update({ ...track.subscription.peek(), order: msg.groupOrder });
-					// Every data stream closes before the pause is answered.
-					if (params.forward === false) throw new Paused(true);
-					const parameters = new Parameters();
-					if (snapshot.largest)
-						parameters.largest = { groupId: snapshot.largest.group, objectId: snapshot.largest.object };
-					const source =
-						selected && selected.kind !== "empty"
-							? track.fork({
-									priority: fromWire(msg.subscriberPriority),
-									order: msg.groupOrder,
-									delay: { max: Milli(Varint.MAX_U53), min: null },
-								})
-							: undefined;
+					// A drained track waits this out, since the update may reopen its range.
+					const applied = Promise.withResolvers<void>();
+					updating = applied.promise;
 					try {
-						await Message.write(request.writer, RequestOk.id, new RequestOk({ parameters }), version);
-					} catch (err) {
-						source?.close();
-						throw err;
+						// update() replaces every option, so carry the rest over as Rust does.
+						if (params.subscriberPriority !== undefined) {
+							const priority = fromWire(params.subscriberPriority);
+							track.update({ ...track.subscription.peek(), priority });
+							datagramPriority.set(priority);
+							if (datagramDemand) datagrams?.setPriority(priority);
+						}
+						const maxAge = params.maxAge;
+						if (maxAge !== undefined) {
+							const current = track.subscription.peek();
+							track.update({ ...current, delay: { ...current?.delay, max: Milli(maxAge) } });
+						}
+						// A fill holds one cache view. At most this decoded update waits behind
+						// it; request parsing then backpressures any further fill requests.
+						if (params.fillParameters) await filling;
+						const snapshot = liveEdge(track);
+						this.#applyPublished(msg, params);
+						const selected =
+							msg.fill && Filter.isDraft20(version)
+								? fillRange(msg.fill, msg.filter, snapshot.largest)
+								: undefined;
+						if (params.subscriptionFilter !== undefined || selected) {
+							// An update's unfiltered range explicitly lowers the floor to zero;
+							// an initial unfiltered subscription still joins the live group. A
+							// fill-only update keeps the range it already has.
+							const requested: ServeRange =
+								params.subscriptionFilter === undefined
+									? bounds.peek()
+									: msg.filter.kind === "unfiltered"
+										? { start: { group: 0n, object: 0n } }
+										: filterRange(msg.filter, snapshot);
+							const tail = afterFill(requested, selected, snapshot);
+							range.start = tail.start;
+							range.end = tail.end;
+							bounds.set(requested, true);
+							track.update({ ...track.subscription.peek(), ranges: serveRanges(range) });
+							// Below its floor the live cursor already passed every group or never owned it,
+							// so lowering it for a fill-only update would hand those groups out a second time.
+							let start = range.start ? range.start.group : 0n;
+							if (params.subscriptionFilter === undefined) start = max(start, floor);
+							floor = start;
+							track.withRange({ ...rangeGroups(range), start: { included: start } });
+							// The fill took the current group's head. The cursor never hands a group out
+							// twice, so what its stream no longer carries resumes from the cache.
+							const next = tail !== requested ? tail.start : undefined;
+							const at = next?.group;
+							if (next && at !== undefined && hooks.handedOut(track, at)) {
+								const slice = groupSlice(range, at);
+								// A stream still serving this group from the fill's edge keeps what it
+								// covers, so the tail resumes where that stream stops.
+								const streaming =
+									live?.sequence === at &&
+									live.slice.skip <= slice.skip &&
+									!live.retired &&
+									!excludes(requested, at, live.slice)
+										? live
+										: undefined;
+								const resume = streaming ? streaming.slice.until : slice.skip;
+								if (resume !== undefined) slice.skip = Math.max(slice.skip, resume);
+								const group =
+									resume !== undefined &&
+									(!range.end || range.end.group >= next.group) &&
+									(slice.until === undefined || slice.until > slice.skip)
+										? hooks.cachedGroup(track, at)
+										: undefined;
+								if (group) serveGroup(group, slice);
+							}
+						}
+						if (params.groupOrder !== undefined)
+							track.update({ ...track.subscription.peek(), order: msg.groupOrder });
+						// Every data stream closes before the pause is answered.
+						if (params.forward === false) throw new Paused(true);
+						const parameters = new Parameters();
+						if (snapshot.largest)
+							parameters.largest = { groupId: snapshot.largest.group, objectId: snapshot.largest.object };
+						const source =
+							selected && selected.kind !== "empty"
+								? track.fork({
+										priority: fromWire(msg.subscriberPriority),
+										order: msg.groupOrder,
+										delay: { max: Milli(Varint.MAX_U53), min: null },
+									})
+								: undefined;
+						try {
+							await Message.write(request.writer, RequestOk.id, new RequestOk({ parameters }), version);
+						} catch (err) {
+							source?.close();
+							throw err;
+						}
+						if (selected && source) startFill(ownRequestId, selected, source);
+					} finally {
+						updating = undefined;
+						applied.resolve();
 					}
-					if (selected && source) startFill(ownRequestId, selected, source);
 				}
 			})();
 			const requestEnded = race([updates, request.ended]);
@@ -1060,12 +1074,19 @@ export class Publisher {
 				() => unsubscribe(),
 			);
 
-			// Serve groups until the track ends or the requester cancels.
+			// Serve groups until the track ends or the requester cancels. Serving ends with the
+			// track only once no update or fill is in flight: an update may reopen the range, and
+			// a reopened group the cursor never handed out is still the cursor's to serve.
 			const serving = (async () => {
 				for (;;) {
 					const group = await servingGroup(track);
-					if (!group) return;
-					serveGroup(group, groupSlice(range, group.sequence));
+					if (group) {
+						serveGroup(group, groupSlice(range, group.sequence));
+						continue;
+					}
+					const pending = updating ?? filling;
+					if (!pending) return;
+					await pending;
 				}
 			})();
 
@@ -1083,14 +1104,7 @@ export class Publisher {
 			let ended = false;
 			try {
 				const served = Symbol("served");
-				const end = await race([
-					serving.then(async () => {
-						await filling;
-						return served;
-					}),
-					requestEnded,
-					revoked,
-				]);
+				const end = await race([serving.then(() => served), requestEnded, revoked]);
 				if (end === "revoked") {
 					console.info(`subscription no longer authorized: broadcast=${name} track=${track.name}`);
 					unauthorized = true;

From 1f50aa4e859363d607dd7a1f22c53bd72eeb43bf Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Sun, 8 Nov 2026 08:57:43 -0800
Subject: [PATCH 07/17] fix(net): an update's decode holds a drained
 subscription open, and a readd keeps SYNCED behind queued changes

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 js/net/src/ietf/publisher.ts    |  4 ++--
 rs/moq-net/src/model/listing.rs | 28 +++++++++++++++++++++++++++-
 2 files changed, 29 insertions(+), 3 deletions(-)

diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts
index f2e908f5081..e9c66faefae 100644
--- a/js/net/src/ietf/publisher.ts
+++ b/js/net/src/ietf/publisher.ts
@@ -964,11 +964,11 @@ export class Publisher {
 					const framed = await request.next();
 					// The requester sent its last message, which ends the request only where FIN cancels.
 					if (framed === undefined) return request.ended;
-					const { params, requestId: ownRequestId } = await this.#decodeUpdate(request, framed);
-					// A drained track waits this out, since the update may reopen its range.
+					// A drained track waits this out, decode included, since the update may reopen its range.
 					const applied = Promise.withResolvers<void>();
 					updating = applied.promise;
 					try {
+						const { params, requestId: ownRequestId } = await this.#decodeUpdate(request, framed);
 						// update() replaces every option, so carry the rest over as Rust does.
 						if (params.subscriberPriority !== undefined) {
 							const priority = fromWire(params.subscriberPriority);
diff --git a/rs/moq-net/src/model/listing.rs b/rs/moq-net/src/model/listing.rs
index 90147b41dfa..2cc91669371 100644
--- a/rs/moq-net/src/model/listing.rs
+++ b/rs/moq-net/src/model/listing.rs
@@ -380,7 +380,13 @@ impl Queue {
 	fn add(&mut self, name: Arc<str>, info: track::Info) {
 		match self.pending.get_mut(&name) {
 			Some(pending) => match *pending {
-				Pending::Remove { finished } => *pending = Pending::Readd { finished, info },
+				Pending::Remove { finished } => {
+					*pending = Pending::Readd { finished, info };
+					// The readd's extra event must not use up the count owed to changes queued behind it.
+					if let SyncPhase::Counting(left) = &mut self.sync {
+						*left += 1;
+					}
+				}
 				// A name is listed once at a time, so only its removal can be pending.
 				_ => debug_assert!(false, "a listed track was listed again"),
 			},
@@ -713,6 +719,26 @@ mod tests {
 
 	const NONE: [&str; 0] = [];
 
+	/// A removal re-added after SYNCED was owed still lets the changes queued behind it out first.
+	#[test]
+	fn a_readd_after_sync_keeps_synced_behind_queued_changes() {
+		let mut queue = Queue::default();
+		queue.remove(&"a".into(), false);
+		queue.add("b".into(), track::Info::default());
+		queue.sync();
+		queue.add("a".into(), track::Info::default());
+		let mut events = Vec::new();
+		while queue.is_ready() {
+			events.push(match queue.next().unwrap().unwrap() {
+				TracksEvent::Added { name, .. } => format!("+{name}"),
+				TracksEvent::Removed { name } => format!("-{name}"),
+				TracksEvent::Finished { name } => format!("~{name}"),
+				TracksEvent::Synced => "synced".into(),
+			});
+		}
+		assert_eq!(events, ["-a", "+a", "+b", "synced"]);
+	}
+
 	#[test]
 	fn tracks_present_arrive_before_synced() {
 		let broadcast = Info::new().produce();

From b66050524c66ce49478b2d7e1fa86fd943f3288b Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Sun, 8 Nov 2026 10:27:31 -0800
Subject: [PATCH 08/17] fix(net): a readd collapsing back to a removal gives
 back its SYNCED count

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 rs/moq-net/src/model/listing.rs | 13 ++++++++++++-
 1 file changed, 12 insertions(+), 1 deletion(-)

diff --git a/rs/moq-net/src/model/listing.rs b/rs/moq-net/src/model/listing.rs
index 2cc91669371..2a2763d5175 100644
--- a/rs/moq-net/src/model/listing.rs
+++ b/rs/moq-net/src/model/listing.rs
@@ -402,7 +402,13 @@ impl Queue {
 			Some(Pending::Add(_)) => {
 				self.pending.remove(name);
 			}
-			// Read, then re-added unread: the reader sees only the last removal.
+			// Read, then re-added unread: the reader sees only the last removal, one event fewer.
+			Some(pending @ Pending::Readd { .. }) => {
+				*pending = Pending::Remove { finished };
+				if let SyncPhase::Counting(left) = &mut self.sync {
+					*left = left.saturating_sub(1);
+				}
+			}
 			Some(pending) => *pending = Pending::Remove { finished },
 			None => {
 				self.pending.insert(name.clone(), Pending::Remove { finished });
@@ -726,6 +732,11 @@ mod tests {
 		queue.remove(&"a".into(), false);
 		queue.add("b".into(), track::Info::default());
 		queue.sync();
+		// Churn coalesces in place, so it owes SYNCED nothing extra.
+		for _ in 0..3 {
+			queue.add("a".into(), track::Info::default());
+			queue.remove(&"a".into(), false);
+		}
 		queue.add("a".into(), track::Info::default());
 		let mut events = Vec::new();
 		while queue.is_ready() {

From ed671bf61dbf91fa34d1476bf5522843087cda0d Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Sun, 8 Nov 2026 13:05:02 -0800
Subject: [PATCH 09/17] chore(quest): claim
 quest/m1/explicit-state/net-phases.md

Quest-Claim: 579b8c0736b859c9

From 7341b2666e00e2dbb4fc1c793cbf52bec360c94d Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Sun, 8 Nov 2026 13:08:13 -0800
Subject: [PATCH 10/17] fix(net): a second js/net fetch stream for a join never
 leaks or settles the first's head

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 js/net/src/ietf/join.test.ts  | 28 ++++++++++-
 js/net/src/ietf/join.ts       | 88 +++++++++++++++++++++--------------
 js/net/src/ietf/subscriber.ts |  6 +--
 3 files changed, 82 insertions(+), 40 deletions(-)

diff --git a/js/net/src/ietf/join.test.ts b/js/net/src/ietf/join.test.ts
index 6692127cc15..75bf83a10a9 100644
--- a/js/net/src/ietf/join.test.ts
+++ b/js/net/src/ietf/join.test.ts
@@ -564,7 +564,7 @@ test("a clean end finishes a head nothing claimed", () => {
 	expect(j.open(() => producer)).toBe(producer);
 	producer.writeFrame({ payload: text("c") });
 	j.wrote();
-	j.finish();
+	j.finish(producer);
 	expect(j.outstanding).toBe(true);
 	j.end();
 	expect(producer.closed.peek()).toBeNull();
@@ -671,3 +671,29 @@ test("an evicted joined group is fetched again as standalone history", async ()
 	group.close();
 	consumer.close();
 });
+
+test("a second fetch stream after the track ended leaves the first head to its own stream", () => {
+	const j = new Join(0n, new Tail());
+	j.end();
+	const first = new track.Producer("catalog").createGroup(0n);
+	expect(j.open(() => first)).toBe(first);
+	const second = new track.Producer("catalog").createGroup(0n);
+	expect(j.open(() => second)).toBeUndefined();
+	// The second stream, discarding, ends first: the head is still the first stream's.
+	j.finish(undefined);
+	expect(first.closed.peek()).toBeUndefined();
+	first.writeFrame({ payload: text("c") });
+	j.finish(first);
+	expect(first.closed.peek()).toBeNull();
+});
+
+test("a discarding second fetch stream leaves a head being written alone", () => {
+	const j = new Join(0n, new Tail());
+	const first = new track.Producer("catalog").createGroup(0n);
+	expect(j.open(() => first)).toBe(first);
+	expect(j.open(() => new track.Producer("catalog").createGroup(0n))).toBeUndefined();
+	j.fail(new Error("reset"), undefined);
+	expect(first.closed.peek()).toBeUndefined();
+	expect(j.pending).toBe(false);
+	expect(j.outstanding).toBe(true);
+});
diff --git a/js/net/src/ietf/join.ts b/js/net/src/ietf/join.ts
index 68adac2115a..5b4e254a3d2 100644
--- a/js/net/src/ietf/join.ts
+++ b/js/net/src/ietf/join.ts
@@ -16,8 +16,11 @@ type State =
 	| { kind: "ready"; producer: netGroup.Producer; next: number }
 	/** No head is coming: refused, empty, failed, claimed, or the subscription went. */
 	| { kind: "done" }
-	/** The track ended, so no tail is coming and a head is the whole group. */
-	| { kind: "ended" };
+	/**
+	 * The track ended, so no tail is coming and a head is the whole group: `producer` is a head
+	 * its fetch stream is still writing.
+	 */
+	| { kind: "ended"; producer?: netGroup.Producer };
 
 /**
  * The join of one live group: the head a draft-18 joining FETCH or a draft-22 fill brings and
@@ -47,8 +50,6 @@ export class Join {
 	#tail: Tail;
 	#settled: Promise<void>;
 	#settle!: () => void;
-	// A head still being written when the track ended, finished with its stream.
-	#ended?: netGroup.Producer;
 
 	/**
 	 * `counted` says the head's stream is one of the subscription's own, which PUBLISH_DONE's
@@ -87,16 +88,19 @@ export class Join {
 	 * sequence, the join gave up, or a second stream answers the one request.
 	 */
 	open(create: () => netGroup.Producer): netGroup.Producer | undefined {
-		const { kind } = this.#state;
+		const state = this.#state;
+		if (state.kind === "pending") {
+			const producer = create();
+			this.#state = { kind: "writing", producer, next: 0 };
+			return producer;
+		}
 		// After the track ended the head still lands, as the whole group.
-		if (kind !== "pending" && kind !== "ended") return undefined;
-		const producer = create();
-		if (kind === "ended") {
-			this.#ended = producer;
+		if (state.kind === "ended" && !state.producer) {
+			const producer = create();
+			this.#state = { kind: "ended", producer };
 			return producer;
 		}
-		this.#state = { kind: "writing", producer, next: 0 };
-		return producer;
+		return undefined;
 	}
 
 	/** The head wrote its next object. */
@@ -104,32 +108,38 @@ export class Join {
 		if (this.#state.kind === "writing") this.#state.next += 1;
 	}
 
-	/** The fetch stream ended cleanly: the head waits for its tail, or none came at all. */
-	finish(): void {
-		if (this.#ended) {
-			this.#ended.finish();
-			this.#ended = undefined;
-			return;
-		}
+	/**
+	 * A fetch stream ended cleanly, having written `producer` (what `open` gave it): the head
+	 * waits for its tail, or none came at all. A stream `open` turned away changes nothing.
+	 */
+	finish(producer: netGroup.Producer | undefined): void {
 		const state = this.#state;
-		if (state.kind === "writing") {
-			this.#transition({ kind: "ready", producer: state.producer, next: state.next });
-		} else if (state.kind === "pending") {
+		if (state.kind === "pending") {
 			// An empty fetch stream: there is no head.
 			this.#transition({ kind: "done" });
+		} else if (state.kind === "writing" && state.producer === producer) {
+			this.#transition({ kind: "ready", producer: state.producer, next: state.next });
+		} else if (state.kind === "ended" && state.producer && state.producer === producer) {
+			state.producer.finish();
+			this.#state = { kind: "done" };
 		}
 	}
 
-	/** The fetch stream failed: a half-written head is a group with no end. */
-	fail(err: Error): void {
-		if (this.#ended) {
-			this.#ended.abort(err);
-			this.#ended = undefined;
-			return;
-		}
+	/**
+	 * A fetch stream failed, having written `producer` (what `open` gave it, if anything): a
+	 * half-written head is a group with no end. A stream `open` turned away changes nothing.
+	 */
+	fail(err: Error, producer: netGroup.Producer | undefined): void {
 		const state = this.#state;
-		if (state.kind === "writing" || state.kind === "ready") state.producer.abort(err);
-		if (state.kind !== "ended") this.#transition({ kind: "done" });
+		if (state.kind === "pending") {
+			this.#transition({ kind: "done" });
+		} else if (state.kind === "writing" && state.producer === producer) {
+			state.producer.abort(err);
+			this.#transition({ kind: "done" });
+		} else if (state.kind === "ended" && state.producer && state.producer === producer) {
+			state.producer.abort(err);
+			this.#state = { kind: "done" };
+		}
 	}
 
 	/**
@@ -211,8 +221,8 @@ export class Join {
 	end(): void {
 		const state = this.#state;
 		if (state.kind === "ready") state.producer.finish();
-		if (state.kind === "writing") this.#ended = state.producer;
-		if (state.kind !== "done") this.#transition({ kind: "ended" });
+		if (state.kind === "pending" || state.kind === "ready") this.#transition({ kind: "ended" });
+		if (state.kind === "writing") this.#transition({ kind: "ended", producer: state.producer });
 	}
 
 	/**
@@ -221,10 +231,16 @@ export class Join {
 	 */
 	cancel(err: Error): void {
 		const state = this.#state;
-		if (state.kind === "writing" || state.kind === "ready") state.producer.abort(err);
-		this.#ended?.abort(err);
-		this.#ended = undefined;
-		if (state.kind !== "done" && state.kind !== "ended") this.#transition({ kind: "done" });
+		if (state.kind === "writing" || state.kind === "ready") {
+			state.producer.abort(err);
+			this.#transition({ kind: "done" });
+		} else if (state.kind === "pending") {
+			this.#transition({ kind: "done" });
+		} else if (state.kind === "ended") {
+			// Already settled: only a head still being written is left to drop.
+			state.producer?.abort(err);
+			this.#state = { kind: "done" };
+		}
 	}
 
 	#transition(next: State): void {
diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts
index 0368a398cf7..56944e46031 100644
--- a/js/net/src/ietf/subscriber.ts
+++ b/js/net/src/ietf/subscriber.ts
@@ -2127,7 +2127,7 @@ export class Subscriber {
 					stream.tryDecode(decode) ??
 					(await race([stream.decodeMaybe(decode), producer?.closed ?? hooks.trackReleased(track)]));
 				if (object === null || object instanceof Error) {
-					join.fail(object ?? new Error("the track closed under the head"));
+					join.fail(object ?? new Error("the track closed under the head"), producer);
 					stream.stop(object ?? new StreamError(StreamCode.Cancel));
 					return;
 				}
@@ -2186,11 +2186,11 @@ export class Subscriber {
 				next += 1;
 			}
 			console.debug(`join head: group=${join.sequence} objects=${next}`);
-			join.finish();
+			join.finish(producer);
 		} catch (err: unknown) {
 			const e = await sessionCause(this.#quic, err);
 			if (e instanceof JoinUnsupported) console.warn(`${e.message}, joining at the next whole group`);
-			join.fail(e);
+			join.fail(e, producer);
 			if (e instanceof ProtocolViolation) this.#fail(e.code, e.message);
 			// As for a subgroup object: the track is malformed, not just its head.
 			if (e instanceof StreamError && e.code === StreamCode.MalformedTrack) track.abort(e);

From 59a4967c1fd8f3499ae7225de14654a2c921e2e7 Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Sun, 8 Nov 2026 13:08:42 -0800
Subject: [PATCH 11/17] fix(net): a js/net refusal that fails to write resets
 its request stream

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 js/net/src/ietf/request_stream.test.ts | 32 ++++++++++++++++++++++++++
 js/net/src/ietf/request_stream.ts      | 12 ++++++----
 2 files changed, 40 insertions(+), 4 deletions(-)

diff --git a/js/net/src/ietf/request_stream.test.ts b/js/net/src/ietf/request_stream.test.ts
index 121adbba81e..56353ca3654 100644
--- a/js/net/src/ietf/request_stream.test.ts
+++ b/js/net/src/ietf/request_stream.test.ts
@@ -236,3 +236,35 @@ test("a request waiting for stream credit outlasts its deadline, which runs from
 		installed.close();
 	}
 });
+
+test("a refusal that fails to write still ends the request", async () => {
+	const failed = new Error("stop sending");
+	const stream = new Stream({
+		readable: new ReadableStream<Uint8Array>(),
+		writable: new WritableStream<Uint8Array>({
+			write: () => {
+				throw failed;
+			},
+		}),
+		version: Version.DRAFT_22,
+	});
+	const aborts = spyOn(stream, "abort");
+	const session = { version: Version.DRAFT_22, client: true } as unknown as Session;
+	const request = new RequestStream(stream, session);
+	await expect(request.refuse("fetch", refusal)).rejects.toBe(failed);
+	expect(aborts).toHaveBeenCalledTimes(1);
+	await request.ended;
+
+	const bare = new Stream({
+		readable: new ReadableStream<Uint8Array>(),
+		writable: new WritableStream<Uint8Array>({
+			write: () => {
+				throw failed;
+			},
+		}),
+		version: Version.DRAFT_22,
+	});
+	const bareAborts = spyOn(bare, "abort");
+	await expect(refuse(bare, Version.DRAFT_22, "fetch", refusal)).rejects.toBe(failed);
+	expect(bareAborts).toHaveBeenCalledTimes(1);
+});
diff --git a/js/net/src/ietf/request_stream.ts b/js/net/src/ietf/request_stream.ts
index 7f04a7e4f2f..56cc794a73b 100644
--- a/js/net/src/ietf/request_stream.ts
+++ b/js/net/src/ietf/request_stream.ts
@@ -55,7 +55,12 @@ async function writeRefusal(w: Writer, version: IetfVersion, kind: RequestKind,
  * @internal
  */
 export async function refuse(stream: Stream, version: IetfVersion, kind: RequestKind, refusal: Refusal): Promise<void> {
-	await writeRefusal(stream.writer, version, kind, refusal);
+	try {
+		await writeRefusal(stream.writer, version, kind, refusal);
+	} catch (err) {
+		stream.abort(error(err));
+		throw err;
+	}
 	stream.close();
 }
 
@@ -226,10 +231,9 @@ export class RequestStream {
 		}
 	}
 
-	/** Refuse the request and end it. */
+	/** Refuse the request and end it, resetting it if the refusal cannot be written. */
 	async refuse(kind: RequestKind, refusal: Refusal): Promise<void> {
-		await writeRefusal(this.writer, this.#session.version, kind, refusal);
-		this.close();
+		await this.serve(writeRefusal(this.writer, this.#session.version, kind, refusal));
 	}
 
 	/**

From 7222a33208222ced38da9986e091b1657e3b0f7d Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Sun, 8 Nov 2026 13:09:23 -0800
Subject: [PATCH 12/17] fix(net): a closed js/net Lister answers nothing more

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 js/net/src/listing.test.ts | 15 +++++++++++++++
 js/net/src/listing.ts      | 34 +++++++++++++++++++---------------
 2 files changed, 34 insertions(+), 15 deletions(-)

diff --git a/js/net/src/listing.test.ts b/js/net/src/listing.test.ts
index e3c478f2b78..10438b23079 100644
--- a/js/net/src/listing.test.ts
+++ b/js/net/src/listing.test.ts
@@ -351,6 +351,21 @@ test("a closed lister withdraws its answer and syncs", async () => {
 	expect(await tracks.drain()).toEqual(["-up", "synced"]);
 });
 
+test("a closed lister answers nothing more, even beside its successor", async () => {
+	const broadcast = new BroadcastProducer();
+	const stale = wireOf(broadcast).lister();
+	stale.close();
+	const lister = wireOf(broadcast).lister();
+	const tracks = list(broadcast);
+	stale.add("stale", { priority: 0 });
+	stale.synced();
+	stale.abort(new Error("stale"));
+	expect(await tracks.drain()).toEqual([]);
+	lister.add("up", { priority: 0 });
+	lister.synced();
+	expect(await tracks.drain()).toEqual(["+up", "synced"]);
+});
+
 test("an owner lists each broadcast once", async () => {
 	const broadcast = new BroadcastProducer();
 	const other = new BroadcastProducer();
diff --git a/js/net/src/listing.ts b/js/net/src/listing.ts
index 80e45d4a491..ec7ba1152d6 100644
--- a/js/net/src/listing.ts
+++ b/js/net/src/listing.ts
@@ -420,15 +420,18 @@ export let makeLister: (listing: Listing) => Lister;
  *
  * While it exists, readers wait for it to {@link synced} before reporting `synced`. Answer
  * while {@link wanted}: list what the publisher behind it holds and keep it current, then
- * {@link clear} once unwanted. {@link close} withdraws what it listed.
+ * {@link clear} once unwanted. {@link close} withdraws what it listed; a closed lister answers
+ * nothing more.
  */
 export class Lister {
-	#listing: Listing;
-	#closed = false;
+	readonly #listing: Listing;
+	// The listing this lister answers into, until it closes.
+	#answering?: Listing;
 
 	private constructor(listing: Listing) {
 		listing.attach();
 		this.#listing = listing;
+		this.#answering = listing;
 	}
 
 	static {
@@ -447,45 +450,46 @@ export class Lister {
 
 	/** List a track. Throws if this answer already lists it. */
 	add(name: string, info: track.Info): void {
-		this.#listing.answerAdd(name, info);
+		this.#answering?.answerAdd(name, info);
 	}
 
 	/** Stop listing a track that left unfinished, reported as `removed`. Throws if this answer does not list it. */
 	remove(name: string): void {
-		this.#listing.answerRemove(name, false);
+		this.#answering?.answerRemove(name, false);
 	}
 
 	/** Stop listing a track that finished, reported as `finished`. Throws if this answer does not list it. */
 	finish(name: string): void {
-		this.#listing.answerRemove(name, true);
+		this.#answering?.answerRemove(name, true);
 	}
 
 	/** The answer holds everything this hop knows now; later changes are live. */
 	synced(): void {
-		this.#listing.answerSynced();
+		this.#answering?.answerSynced();
 	}
 
 	/** Withdraw the answer: every track it listed, and its sync. */
 	clear(): void {
-		this.#listing.answerClear();
+		this.#answering?.answerClear();
 	}
 
 	/** End every current reader as if the broadcast ended, and withdraw the answer. A later reader asks again. */
 	end(): void {
-		this.#listing.detach(null);
-		this.#listing.answerClear();
+		const listing = this.#answering;
+		listing?.detach(null);
+		listing?.answerClear();
 	}
 
 	/** Fail every current reader with `err`, and withdraw the answer. A later reader asks again. */
 	abort(err: Error): void {
-		this.#listing.detach(err);
-		this.#listing.answerClear();
+		const listing = this.#answering;
+		listing?.detach(err);
+		listing?.answerClear();
 	}
 
 	/** Stop answering: the broadcast lists only its producer's tracks from here on. Idempotent. */
 	close(): void {
-		if (this.#closed) return;
-		this.#closed = true;
-		this.#listing.detachLister();
+		this.#answering?.detachLister();
+		this.#answering = undefined;
 	}
 }

From 686637aade013eec6f0a05efd1288a550232dff0 Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Sun, 8 Nov 2026 13:09:56 -0800
Subject: [PATCH 13/17] fix(signals): get() from a spawn task that outlived its
 run no longer subscribes the next run

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 js/signals/src/index.test.ts | 35 +++++++++++++++++++++++++++++++++++
 js/signals/src/index.ts      |  6 +++++-
 2 files changed, 40 insertions(+), 1 deletion(-)

diff --git a/js/signals/src/index.test.ts b/js/signals/src/index.test.ts
index 0453e62e8a4..05b9b8390a4 100644
--- a/js/signals/src/index.test.ts
+++ b/js/signals/src/index.test.ts
@@ -864,6 +864,41 @@ describe("Effect", () => {
 		expect(log).toEqual(["served", "released"]);
 	});
 
+	test("get() from a stale spawn does not subscribe the next run", async () => {
+		const tick = new Signal(0);
+		const other = new Signal(0);
+		const gate = Promise.withResolvers<void>();
+		let runs = 0;
+
+		const effect = new Effect((e) => {
+			e.get(tick);
+			if (++runs > 1) return;
+
+			e.spawn(async () => {
+				await gate.promise;
+				expect(e.get(other)).toBe(0);
+			});
+		});
+
+		try {
+			await settle();
+			tick.set(1);
+			await settle();
+			expect(runs).toBe(1); // stale window
+
+			gate.resolve();
+			await settle();
+			expect(runs).toBe(2);
+
+			// Only the dead run read `other`.
+			other.set(1);
+			await settle();
+			expect(runs).toBe(2);
+		} finally {
+			effect.close();
+		}
+	});
+
 	test("cleanup after close fires immediately", async () => {
 		const effect = new Effect((e) => {
 			e.get(new Signal(0));
diff --git a/js/signals/src/index.ts b/js/signals/src/index.ts
index 1eb72ecacf6..21be471d1d7 100644
--- a/js/signals/src/index.ts
+++ b/js/signals/src/index.ts
@@ -743,12 +743,16 @@ export class Effect {
 		}
 	}
 
-	/** Reads a signal and tracks it, rerunning the effect whenever it changes. */
+	/**
+	 * Reads a signal and tracks it, rerunning the effect whenever it changes. A {@link spawn} task
+	 * that outlived its run only reads: the next run tracks what it reads itself.
+	 */
 	get<T>(signal: Getter<T>): T {
 		if (this.#dispose === undefined) {
 			console.warn("Effect.get called when closed, returning current value");
 			return signal.peek();
 		}
+		if (this.#stale) return signal.peek();
 
 		const value = signal.peek();
 

From 0d797125d2aa0e8f05b88184b565a8650111b211 Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Sun, 8 Nov 2026 13:11:48 -0800
Subject: [PATCH 14/17] refactor(net): js/net auth recounts its pending tokens
 and keeps an issued grant's queue in its open state

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 js/net/src/auth_session.ts | 89 +++++++++++++++++++++-----------------
 1 file changed, 49 insertions(+), 40 deletions(-)

diff --git a/js/net/src/auth_session.ts b/js/net/src/auth_session.ts
index e2871d95149..12cd2aa9aa3 100644
--- a/js/net/src/auth_session.ts
+++ b/js/net/src/auth_session.ts
@@ -471,9 +471,11 @@ class IssuedGrant implements Issued {
 	#stream: Stream;
 	#wire: AuthWire;
 	#writes = Promise.resolve();
-	/** The latest grant, until its write starts: a newer one replaces it rather than queueing. */
-	#queued: Grant | undefined;
-	#done = false;
+	/**
+	 * Open, holding the latest grant until its write starts, so a newer one replaces it rather
+	 * than queueing; or done: closed, revoked, or a reply that could not be sent.
+	 */
+	#state: { kind: "open"; queued?: Grant } | { kind: "done" } = { kind: "open" };
 
 	constructor(stream: Stream, wire: AuthWire) {
 		this.#stream = stream;
@@ -495,41 +497,51 @@ class IssuedGrant implements Issued {
 			// The peer already closed the stream: nothing left to tell it.
 			if (err instanceof StreamError) return;
 			// A grant the wire cannot carry is withheld, never widened, and so is any other
-			// reply that fails to encode.
+			// reply that fails to encode. An update since then is dropped with it.
 			if (!(err instanceof Unsupported)) console.warn("auth reply not sent", err);
-			this.#done = true;
-			this.#queued = undefined;
+			this.#state = { kind: "done" };
 			return this.#wire.unsupported(this.#stream, error(err).message).catch(() => void 0);
 		});
 	}
 
 	update(grant: Grant): void {
-		if (this.#done) return;
+		const state = this.#state;
+		if (state.kind === "done") return;
 		// A grant still waiting to be written is replaced, so a peer withholding stream credit
 		// holds at most one.
-		const queued = this.#queued !== undefined;
-		this.#queued = grant;
+		const queued = state.queued !== undefined;
+		state.queued = grant;
 		if (queued) return;
 		this.#write(async () => {
-			const grant = this.#queued;
-			this.#queued = undefined;
-			if (grant === undefined) return;
-			const expires = grant.expires === undefined ? undefined : grant.expires - wall();
-			await this.#wire.grant(this.#stream, { publish: grant.publish, subscribe: grant.subscribe, expires });
+			// A close since then sends the grant itself.
+			const state = this.#state;
+			if (state.kind === "done" || state.queued === undefined) return;
+			const grant = state.queued;
+			state.queued = undefined;
+			await this.#send(grant);
 		});
 	}
 
+	async #send(grant: Grant): Promise<void> {
+		const expires = grant.expires === undefined ? undefined : grant.expires - wall();
+		await this.#wire.grant(this.#stream, { publish: grant.publish, subscribe: grant.subscribe, expires });
+	}
+
 	revoke(code: SessionCode, reason: string): void {
-		if (this.#done) return;
+		if (this.#state.kind === "done") return;
 		// The peer never saw a grant still queued, so the refusal replaces it.
-		this.#queued = undefined;
+		this.#state = { kind: "done" };
 		this.#write(() => this.#wire.refuse(this.#stream, code, reason));
-		this.close();
+		this.#writes = this.#writes.then(() => this.#stream.writer.close());
 	}
 
 	close(): void {
-		if (this.#done) return;
-		this.#done = true;
+		const state = this.#state;
+		if (state.kind === "done") return;
+		this.#state = { kind: "done" };
+		// The grant still queued is the last thing the peer hears before the end.
+		const last = state.queued;
+		if (last !== undefined) this.#write(() => this.#send(last));
 		this.#writes = this.#writes.then(() => this.#stream.writer.close());
 	}
 }
@@ -621,6 +633,8 @@ export class AuthSession {
 	#presented = new Set<Presented>();
 	// The handle's tokens as this session presents them, so a request carries only those.
 	#byToken = new Map<Token, Presented>();
+	// How many presented tokens await a reply: at setup, and with bytes. Recounted from
+	// #presented by #recount after every change to it or to an `isAnswered`.
 	#setupPending = new Signal(0);
 	#pending = new Signal(0);
 	#closed = false;
@@ -843,8 +857,6 @@ export class AuthSession {
 	close(err: Error = new Error("session closed")) {
 		if (this.#closed) return;
 		this.#closed = true;
-		this.#pending.set(0);
-		this.#setupPending.set(0);
 		for (const presented of this.#presented) {
 			presented.ended = true;
 			presented.isAnswered = true;
@@ -859,8 +871,7 @@ export class AuthSession {
 
 	#present(wire: AuthWire, presented: Presented) {
 		this.#presented.add(presented);
-		if (presented.setup) this.#setupPending.update((n) => n + 1);
-		if (presented.token.byteLength > 0) this.#pending.update((n) => n + 1);
+		this.#recount();
 		void this.#run(wire, presented);
 	}
 
@@ -967,7 +978,7 @@ export class AuthSession {
 			granted = false;
 			// Unanswered again, as on its first presentation: a request or publication the token
 			// might cover waits for the retry's verdict rather than being judged without it.
-			this.#unsettle(presented);
+			presented.isAnswered = false;
 			this.#recompute();
 			delay = delay === undefined ? RETRY_INITIAL : Milli(Math.min(delay * 2, RETRY_MAX));
 			presented.retry = new AbortController();
@@ -992,7 +1003,7 @@ export class AuthSession {
 		presented.grant = undefined;
 		// A token that ends unanswered counts as answered for enforcement: its refusal is
 		// the reply.
-		this.#settle(presented);
+		presented.isAnswered = true;
 		this.#presented.delete(presented);
 		const owner = presented.owner;
 		if (owner && this.#byToken.get(owner) === presented) this.#byToken.delete(owner);
@@ -1020,27 +1031,25 @@ export class AuthSession {
 
 	/** The peer granted `presented`: it counts as answered, and so does the handle's token. */
 	#answered(presented: Presented) {
-		this.#settle(presented);
-		presented.owner?.answer();
-	}
-
-	/** Stop counting `presented` as unanswered, for enforcement and the setup wait. */
-	#settle(presented: Presented) {
-		if (presented.isAnswered) return;
 		presented.isAnswered = true;
-		if (presented.setup) this.#setupPending.update((n) => n - 1);
-		if (presented.token.byteLength > 0 && !this.#closed) this.#pending.update((n) => n - 1);
+		presented.owner?.answer();
 	}
 
-	/** Count `presented` as unanswered again, as a retry presents it anew. */
-	#unsettle(presented: Presented) {
-		if (!presented.isAnswered) return;
-		presented.isAnswered = false;
-		if (presented.setup) this.#setupPending.update((n) => n + 1);
-		if (presented.token.byteLength > 0) this.#pending.update((n) => n + 1);
+	/** Count the presented tokens still awaiting a reply, for enforcement and the setup wait. */
+	#recount() {
+		let setup = 0;
+		let pending = 0;
+		for (const presented of this.#presented) {
+			if (presented.isAnswered) continue;
+			if (presented.setup) setup += 1;
+			if (presented.token.byteLength > 0) pending += 1;
+		}
+		this.#setupPending.set(setup);
+		this.#pending.set(pending);
 	}
 
 	#recompute() {
+		this.#recount();
 		const granted: Grant[] = [];
 		for (const presented of this.#presented) {
 			if (presented.grant) granted.push(presented.grant);

From bc3128bb2ab7b21872c69771c618727841ee1448 Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Sun, 8 Nov 2026 13:12:50 -0800
Subject: [PATCH 15/17] refactor(net): js/net Reload derives its status in one
 place

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 js/net/src/connection/reload.ts | 51 +++++++++++++++++++--------------
 1 file changed, 30 insertions(+), 21 deletions(-)

diff --git a/js/net/src/connection/reload.ts b/js/net/src/connection/reload.ts
index 6ca7327d64c..4beaad32499 100644
--- a/js/net/src/connection/reload.ts
+++ b/js/net/src/connection/reload.ts
@@ -329,7 +329,7 @@ export class Reload {
 	// outlives the capped delay, or a new sequence starts.
 	#redirects = new Redirects();
 
-	// Whether a connect attempt is in flight, so a retiring predecessor reports the right status.
+	// Whether a run is dialing: from its start until each connect attempt settles.
 	#dialing = false;
 
 	// The current wait between attempts, doubling per failure, and when the retry window expires.
@@ -524,8 +524,7 @@ export class Reload {
 			this.#givenUpHref = undefined;
 		}
 
-		// A drained predecessor still serves while its replacement dials.
-		if (!this.#draining) this.status.set("connecting");
+		this.#setDialing(true);
 
 		// This run's teardown, handed to connect() so a rerun cancels the attempt in flight.
 		const signal = effect.abort;
@@ -536,10 +535,10 @@ export class Reload {
 		effect.cleanup(() => {
 			if (current) {
 				this.#end(current);
-				if (this.established.peek() === current) this.established.set(undefined);
+				if (this.established.peek() === current) this.#serve(undefined);
 				current = undefined;
 			}
-			if (this.established.peek() === undefined) this.status.set("disconnected");
+			this.#setDialing(false);
 		});
 
 		effect.spawn(async () => {
@@ -556,7 +555,7 @@ export class Reload {
 					// spends it too.
 					this.#deadline ??= this.#window(now());
 
-					this.#dialing = true;
+					this.#setDialing(true);
 					this.#dialedTokens = this.tokens.peek();
 					let connection: Established;
 					try {
@@ -574,8 +573,9 @@ export class Reload {
 							viewers: this.viewers,
 							signal,
 						});
-					} finally {
-						this.#dialing = false;
+					} catch (err) {
+						this.#setDialing(false);
+						throw err;
 					}
 
 					// Hand the connection to the effect. One that lands after this run is over never
@@ -587,8 +587,8 @@ export class Reload {
 					current = connection;
 
 					// The replacement serves now; a predecessor keeps draining its groups in flight.
-					this.established.set(connection);
-					this.status.set("connected");
+					this.#serve(connection);
+					this.#setDialing(false);
 					connected = now();
 
 					// A cancelled effect resolves undefined, so the sentinel tells the session
@@ -600,7 +600,7 @@ export class Reload {
 						console.warn(`connection closed, reconnecting: ${ended ? reason(ended) : "no error"}`);
 						this.#redirects.reset();
 						current = undefined;
-						if (this.established.peek() === connection) this.established.set(undefined);
+						if (this.established.peek() === connection) this.#serve(undefined);
 						this.#retry(effect, connected, ended ?? undefined);
 						return;
 					}
@@ -682,8 +682,7 @@ export class Reload {
 			// would ignore it, so stop here.
 			console.warn("GOAWAY redirect refused:", err);
 			this.#end(connection);
-			this.established.set(undefined);
-			this.status.set("disconnected");
+			this.#serve(undefined);
 			this.#giveUp(error(err));
 			return false;
 		}
@@ -701,9 +700,7 @@ export class Reload {
 		const draining = new Draining(connection, cap, () => {
 			if (this.#draining === draining) this.#draining = undefined;
 			// If nothing replaced it yet, nothing is serving.
-			if (this.established.peek() !== connection) return;
-			this.established.set(undefined);
-			this.status.set(this.#dialing ? "connecting" : "disconnected");
+			if (this.established.peek() === connection) this.#serve(undefined);
 		});
 		this.#draining = draining;
 		return true;
@@ -725,6 +722,23 @@ export class Reload {
 		void ending.then(() => this.#ending.delete(connection));
 	}
 
+	/** Serve `connection`, or nothing. */
+	#serve(connection: Established | undefined): void {
+		this.established.set(connection);
+		this.#report();
+	}
+
+	#setDialing(dialing: boolean): void {
+		this.#dialing = dialing;
+		this.#report();
+	}
+
+	// The one derivation of status: serving, else dialing, else neither. A drained predecessor
+	// serves until it retires, so its replacement's dial reads as connected.
+	#report(): void {
+		this.status.set(this.established.peek() ? "connected" : this.#dialing ? "connecting" : "disconnected");
+	}
+
 	/** Stop serving from a drained predecessor, ending it if still open. */
 	#retire(): void {
 		const draining = this.#draining?.retire();
@@ -749,7 +763,6 @@ export class Reload {
 	 * window: the window restarts once the wait is over.
 	 */
 	#restart(effect: Effect): void {
-		if (this.established.peek() === undefined) this.status.set("disconnected");
 		const wait = this.#backoff();
 		const tick = this.#tick.peek() + 1;
 		effect.timer(() => {
@@ -774,10 +787,6 @@ export class Reload {
 	 * `cause` the error that killed it, if it died with one.
 	 */
 	#retry(effect: Effect, connected: Instant | undefined, cause?: unknown): void {
-		// Report disconnected during the backoff rather than when the retry reruns the
-		// effect, unless a drained predecessor still serves until it retires.
-		if (this.established.peek() === undefined) this.status.set("disconnected");
-
 		// A healthy session's end clears the backoff and starts a fresh retry window: a one-off
 		// drop should reconnect promptly. Anything shorter is a peer that accepts and soon
 		// severs, which has to keep escalating or we hammer it forever at the initial delay.

From f4fc150ef4b15f0377fbb4c64d3a25be0ea17d5a Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Sun, 8 Nov 2026 13:21:11 -0800
Subject: [PATCH 16/17] refactor(net): Following keeps the broadcast handed out
 last in one value

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 quest/m1/explicit-state/net-phases.md | 33 +++-------
 rs/moq-net/src/model/follow.rs        | 89 +++++++++++++++------------
 2 files changed, 60 insertions(+), 62 deletions(-)

diff --git a/quest/m1/explicit-state/net-phases.md b/quest/m1/explicit-state/net-phases.md
index 35d89e9b45c..31300e99327 100644
--- a/quest/m1/explicit-state/net-phases.md
+++ b/quest/m1/explicit-state/net-phases.md
@@ -18,14 +18,14 @@ redesigning it, and reproduce any inconsistent state you find.
 publisher's `TrackServe` phase and its reopened-tail fix, the datagram
 forward, the session drivers' terminal result, IETF `Dispatch`, lite
 `PeerSetup`, the listing `Queue`, and the js/net connections' close phase.
-What remains overlapped open PRs when that one started:
+[moq-dev/moq#8322](https://github.com/moq-dev/moq/pull/8322) landed js/net's
+`Join`, `RequestStream.refuse`, `Lister`, auth counters and `IssuedGrant`,
+`Reload.status`, the signals `Effect` stale read, and Rust's `Following`.
+
+The lite items overlap
 [moq-dev/moq#8220](https://github.com/moq-dev/moq/pull/8220) (lite publisher
-and subscriber, `fills.rs`, js/net lite),
-[moq-dev/moq#8063](https://github.com/moq-dev/moq/pull/8063) (lite announce
-and subscriber, `model/origin.rs`, `model/follow.rs`, js/net `origin.ts`),
-and [moq-dev/moq#8222](https://github.com/moq-dev/moq/pull/8222) (js/net
-`reload.ts` and lite `subscriber.ts`). Start each item once the PR touching
-its file has landed or closed.
+and subscriber, `fills.rs`, js/net lite). Start each once it has landed or
+closed.
 
 Rust:
 
@@ -42,32 +42,17 @@ Rust:
   - `SubStream.served` plus `started` are really `Option<Option<u64>>`.
 - **`auth::Slot`** (`rs/moq-net/src/auth.rs`): `streamed`, `refused`,
   `withdrawn`, `grant`, `answered`, `retry`, and `ended` are seven fields
-  coordinated by hand.
-- **`Following`** (`rs/moq-net/src/model/follow.rs`): eight fields, of which
-  `last`, `moved`, and `absorbed` are always reset together.
+  coordinated by hand. Its tests call `granted` on slots whose stream never
+  opened, so the phase enum has them open the stream first.
 - **Smaller ones**:
   - `model/origin.rs` `TrackIo` keeps `request` and `accepted` side by side.
   - `fills.rs` `Recovery`.
 
 JS (`js/net`):
 
-- **Auth** (`auth_session.ts`):
-  - `AuthSession`/`Presented` derive counters by hand at six sites.
-  - `IssuedGrant` sets `#done` asynchronously, so two updates in one tick
-    both pass its check.
-- **`Reload`** (`connection/reload.ts`): `status` is written at seven sites,
-  each re-deriving it from `established`, `#draining`, and `#dialing`.
 - **Lite `SubscribeEntry`** (`lite/subscriber.ts`): eleven optional fields.
-- **`Lister`** (`listing.ts`): it still writes into the listing after
-  `close()`.
-- **`RequestStream`** (`ietf/request_stream.ts`): `refuse()` leaves the stream
-  open if writing the refusal throws.
-- **Signals `Effect`** (`js/signals/src/index.ts`): `get()` during the stale
-  window pushes into the next run's unwatch list.
 - **Origin `Request` and `OriginState`** (`origin.ts`): the closed state is
   encoded by about 40 `?.` guards.
-- **`Join`** (`ietf/join.ts`): a second fetch stream after the track ended
-  leaks the first producer.
 
 Keep the lite `TrackRun` start state out of this child. It belongs to
 [Delete the old start bookkeeping](/quest/m2/subscribe-ranges/start-bookkeeping.md).
diff --git a/rs/moq-net/src/model/follow.rs b/rs/moq-net/src/model/follow.rs
index d4468c5bbc0..927cd0080b5 100644
--- a/rs/moq-net/src/model/follow.rs
+++ b/rs/moq-net/src/model/follow.rs
@@ -197,24 +197,38 @@ pub struct Following {
 	unseen: bool,
 	/// The resolution in flight, kept across a dropped wait so a retry picks it up.
 	resolving: Option<Requesting>,
-	/// The broadcast handed out last, until the path goes offline or the broadcast ends. Weak,
-	/// so it never keeps one the caller let go of from retiring.
-	last: Option<broadcast::WeakConsumer>,
-	/// Once the broadcast handed out last closed, waits for its instance to stop serving the
-	/// path.
-	moved: Option<Moved>,
-	/// A change resolved to the broadcast handed out last while it was still open, so it counts
-	/// as delivered only until that broadcast closes.
-	absorbed: bool,
-	/// A broadcast was handed out and its end not reported yet.
-	live: bool,
-	/// The broadcast handed out last stopped serving the path, and its end is owed.
-	ended: bool,
+	/// The broadcast handed out last, until the path goes offline, the broadcast ends, or a
+	/// resolution is refused.
+	last: Option<Last>,
+	/// Whether the end of what was handed out is still to be reported.
+	handed: Handed,
 	/// The last resolution was refused: any change to the serving route, even the same
 	/// instance over another route, is worth asking again.
 	refused: bool,
 }
 
+/// The broadcast a [`Following`] handed out last, and what it waits on once that closes.
+struct Last {
+	/// Weak, so it never keeps one the caller let go of from retiring.
+	broadcast: broadcast::WeakConsumer,
+	/// Once the broadcast closed, waits for its instance to stop serving the path.
+	moved: Option<Moved>,
+	/// A change resolved to this broadcast while it was still open, so it counts as delivered
+	/// only until the broadcast closes.
+	absorbed: bool,
+}
+
+/// Whether a [`Following`] owes the end of the broadcast it handed out.
+#[derive(Clone, Copy, PartialEq, Eq)]
+enum Handed {
+	/// Nothing handed out, or its end already reported.
+	Nothing,
+	/// A broadcast was handed out and still serves the path.
+	Live,
+	/// The broadcast handed out stopped serving the path, and its end is owed.
+	Ended,
+}
+
 impl Following {
 	pub(super) fn new(consumer: Consumer, path: Path, instance: Option<Instance>, follow: Follow) -> Self {
 		Self {
@@ -225,10 +239,7 @@ impl Following {
 			unseen: false,
 			resolving: None,
 			last: None,
-			moved: None,
-			absorbed: false,
-			live: false,
-			ended: false,
+			handed: Handed::Nothing,
 			refused: false,
 		}
 	}
@@ -253,9 +264,8 @@ impl Following {
 				self.resolving = None;
 			}
 			// Report an end before whatever serves next, so a reader drops the old broadcast first.
-			if self.ended {
-				self.ended = false;
-				self.live = false;
+			if self.handed == Handed::Ended {
+				self.handed = Handed::Nothing;
 				return Poll::Ready(Ok(None));
 			}
 			// The broadcast handed out ended while the path is still served: its instance's
@@ -265,12 +275,13 @@ impl Following {
 			// change it absorbed is owed. A refusal forgets it, so only a route change asks again.
 			if !self.unseen
 				&& self.follow.serving().is_some()
-				&& let Some(last) = self.last.as_ref().map(broadcast::WeakConsumer::consume)
-				&& last.poll_closed(waiter).is_ready()
+				&& let Some(last) = &mut self.last
+				&& let broadcast = last.broadcast.consume()
+				&& broadcast.poll_closed(waiter).is_ready()
 			{
-				let moved = match &last.info().instance {
-					_ if self.absorbed => true,
-					Some(instance) => self
+				let moved = match &broadcast.info().instance {
+					_ if last.absorbed => true,
+					Some(instance) => last
 						.moved
 						.get_or_insert_with(|| {
 							self.consumer
@@ -281,7 +292,7 @@ impl Following {
 					None => true,
 				};
 				if moved {
-					self.moved = None;
+					last.moved = None;
 					self.unseen = true;
 				}
 			}
@@ -301,12 +312,10 @@ impl Following {
 			self.refused = resolved.is_err();
 			let broadcast = match resolved {
 				Ok(broadcast) => broadcast,
-				// `live` stays: the end of what was handed out is still owed once the path goes
+				// `handed` stays: the end of what was handed out is still owed once the path goes
 				// offline.
 				Err(err) => {
 					self.last = None;
-					self.moved = None;
-					self.absorbed = false;
 					return Poll::Ready(Err(err));
 				}
 			};
@@ -315,14 +324,18 @@ impl Following {
 			// the origin holds its restart, and that restart then names the instance handed
 			// out. The caller freeing it before reading the restart is owed it again.
 			let weak = broadcast.weak();
-			if self.last.as_ref().is_some_and(|last| last.same_channel(&weak)) {
-				self.absorbed = true;
+			if let Some(last) = &mut self.last
+				&& last.broadcast.same_channel(&weak)
+			{
+				last.absorbed = true;
 				continue;
 			}
-			self.last = Some(weak);
-			self.moved = None;
-			self.absorbed = false;
-			self.live = true;
+			self.last = Some(Last {
+				broadcast: weak,
+				moved: None,
+				absorbed: false,
+			});
+			self.handed = Handed::Live;
 			return Poll::Ready(Ok(Some(broadcast)));
 		}
 	}
@@ -339,10 +352,10 @@ impl Following {
 	/// The broadcast handed out last stopped serving the path: owe its end, and hand out
 	/// whatever serves next, even the same instance coming back.
 	fn end(&mut self) {
-		self.ended |= self.live;
+		if self.handed == Handed::Live {
+			self.handed = Handed::Ended;
+		}
 		self.last = None;
-		self.moved = None;
-		self.absorbed = false;
 	}
 
 	/// Apply one change to the path, returning whether it is one.

From ce493fffd90cfb1a73d2754c32b3feecc008ee1e Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Sun, 8 Nov 2026 17:39:36 -0800
Subject: [PATCH 17/17] docs(net): Handed::Live says its end is not yet due,
 not that it still serves

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 rs/moq-net/src/model/follow.rs | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/rs/moq-net/src/model/follow.rs b/rs/moq-net/src/model/follow.rs
index 927cd0080b5..893d3c4bff3 100644
--- a/rs/moq-net/src/model/follow.rs
+++ b/rs/moq-net/src/model/follow.rs
@@ -223,7 +223,8 @@ struct Last {
 enum Handed {
 	/// Nothing handed out, or its end already reported.
 	Nothing,
-	/// A broadcast was handed out and still serves the path.
+	/// A broadcast was handed out and its end is not yet due, even if a refused takeover means
+	/// it no longer serves the path.
 	Live,
 	/// The broadcast handed out stopped serving the path, and its end is owed.
 	Ended,