From de42bb31d84934b4ebbc7d2ba14c94cee4a8cc5e Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Thu, 1 Oct 2026 21:55:58 -0700
Subject: [PATCH 1/5] chore: claim demand everywhere quest


From 37f8ff9035a5ce09d46f38ef771f5b82a6b06812 Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Thu, 1 Oct 2026 22:08:59 -0700
Subject: [PATCH 2/5] refactor(net): watch subscriber demand through per-level
 handles

Co-Authored-By: GPT-6 <noreply@openai.com>
---
 doc/lib/js/net.md                     |  10 +++
 doc/lib/rs/moq-net.md                 |  12 +++
 js/flate/src/stream/stream.test.ts    |   4 +-
 js/json/src/stream/stream.test.ts     |   4 +-
 js/net/bench/broadcasts.ts            |  35 +++++++++
 js/net/src/broadcast.test.ts          |  42 ++++++++++
 js/net/src/broadcast.ts               |  78 ++++++++++++++++++-
 js/net/src/group.test.ts              |  10 +--
 js/net/src/group.ts                   |  66 ++++++++++------
 js/net/src/integration.test.ts        |  10 +--
 js/net/src/lite/subscriber.ts         |  25 ++++--
 js/net/src/track.ts                   |   7 +-
 rs/moq-ffi/src/producer.rs            |   2 +-
 rs/moq-gst/src/source/imp.rs          |   5 +-
 rs/moq-net/src/ietf/subscriber.rs     |   2 +-
 rs/moq-net/src/lite/subscriber.rs     |   2 +-
 rs/moq-net/src/model/group.rs         | 108 ++++++++++++++++++++++++--
 rs/moq-net/src/model/track.rs         |  65 +++++++++++++---
 rs/moq-net/tests/track_info_cancel.rs |  10 ++-
 rs/moq-stats/src/produce.rs           |   2 +-
 20 files changed, 431 insertions(+), 68 deletions(-)

diff --git a/doc/lib/js/net.md b/doc/lib/js/net.md
index 4e77fa2659..22db07d1a9 100644
--- a/doc/lib/js/net.md
+++ b/doc/lib/js/net.md
@@ -124,3 +124,13 @@ Examples in
 [`js/net/examples/`](https://github.com/moq-dev/moq/tree/main/js/net/examples).
 Runs in the browser and, over WebSocket, in Node, Bun, and Deno; see
 [server-side](/lib/js/#server-side).
+
+## Subscriber demand
+
+Call `demand()` on a track producer or pending track request, a group producer,
+or a broadcast producer. The returned `Track.Demand`, `Group.Demand`, or
+`Broadcast.Demand` exposes a read-only `used` signal, `unused()`, and `closed`.
+A broadcast watches subscribers to its tracks; holding a broadcast consumer
+alone is not demand. A group's demand counts its mirror readers, preserving
+fetch coalescing: the shared download becomes unused only after every reader
+leaves. Demand handles cannot write or close the producer.
diff --git a/doc/lib/rs/moq-net.md b/doc/lib/rs/moq-net.md
index cb2975a206..5790cf30b7 100644
--- a/doc/lib/rs/moq-net.md
+++ b/doc/lib/rs/moq-net.md
@@ -175,3 +175,15 @@ subscription's upstream request.
 
 Inside a group, `set_frames(...)` applies the same range syntax to frame
 indices.
+
+## Subscriber demand
+
+Track producers, pending track requests, and dynamic track handlers return a
+`track::Demand` from `demand()`. Group producers and pending group requests
+return `group::Demand`; broadcast producers return `broadcast::Demand`.
+Watch `used()` / `unused()` (or `poll_used` / `poll_unused`) on these handles,
+or take a snapshot with `is_used()`. Demand handles cannot write or abort and
+do not keep a producer or request alive. A pending group request counts every
+caller sharing its fetch, so one caller leaving does not cancel the others.
+Use the producer's `abort_unused` to atomically end an idle track; observing
+unused demand alone does not protect against a new subscriber arriving.
diff --git a/js/flate/src/stream/stream.test.ts b/js/flate/src/stream/stream.test.ts
index 014a69a495..ebe4278cac 100644
--- a/js/flate/src/stream/stream.test.ts
+++ b/js/flate/src/stream/stream.test.ts
@@ -121,8 +121,8 @@ test("a second group is reported while the first is still open", async () => {
 
 	// Both mirrors are released. The read that lost the race would otherwise stay registered on the
 	// first group, keeping this consumer's subscription reachable after the caller drops it.
-	expect(first.used.peek()).toBe(false);
-	expect(second.used.peek()).toBe(false);
+	expect(first.demand().used.peek()).toBe(false);
+	expect(second.demand().used.peek()).toBe(false);
 
 	// Sticky: a later read must not report the rest of the first group as a whole log.
 	first.writeFrame({ payload: payloads(3)[2], timestamp: Time.Timestamp.now() });
diff --git a/js/json/src/stream/stream.test.ts b/js/json/src/stream/stream.test.ts
index 92fa0fe2b2..1e14c3cb2a 100644
--- a/js/json/src/stream/stream.test.ts
+++ b/js/json/src/stream/stream.test.ts
@@ -89,8 +89,8 @@ test("a second group is reported while the first is still open", async () => {
 
 	// Both mirrors are released. The read that lost the race would otherwise stay registered on the
 	// first group, keeping this consumer's subscription reachable after the caller drops it.
-	expect(first.used.peek()).toBe(false);
-	expect(second.used.peek()).toBe(false);
+	expect(first.demand().used.peek()).toBe(false);
+	expect(second.demand().used.peek()).toBe(false);
 
 	// Sticky: a later read must not report the rest of the first group as a whole log.
 	first.writeFrame({ payload: encode({ n: 2 }), timestamp: Time.Timestamp.now() });
diff --git a/js/net/bench/broadcasts.ts b/js/net/bench/broadcasts.ts
index 903ab89762..29525c739a 100644
--- a/js/net/bench/broadcasts.ts
+++ b/js/net/bench/broadcasts.ts
@@ -1,3 +1,4 @@
+import { Producer as BroadcastProducer } from "../src/broadcast.ts";
 /** Sweep route and observer counts for a single touched-path route update. */
 import { Producer } from "../src/origin.ts";
 import * as Path from "../src/path.ts";
@@ -49,3 +50,37 @@ for (const scope of ["unscoped", "distinct-scopes", "producer-scopes", "rooted-p
 	}
 }
 if (checksum === 0) throw new Error("benchmark did no work");
+
+// Sweep both axes while only one track's demand changes.
+const demandUpdates = 1024;
+console.log("tracks,observers,demand_update_us");
+for (const trackCount of [8, 32, 128]) {
+	for (const observerCount of [1, 8, 32]) {
+		const broadcast = new BroadcastProducer();
+		const tracks = Array.from({ length: trackCount }, (_, index) => broadcast.createTrack(`track-${index}`));
+		const demand = broadcast.demand();
+		let notifications = 0;
+		const disposes = Array.from({ length: observerCount }, () =>
+			demand.used.subscribe(() => {
+				notifications++;
+			}),
+		);
+		const start = performance.now();
+		for (let index = 0; index < demandUpdates; index++) {
+			const subscriber = tracks[0].subscribe();
+			await Promise.resolve();
+			await Promise.resolve();
+			if (!demand.used.peek()) throw new Error("missing demand");
+			subscriber.close();
+			await demand.unused();
+			await Promise.resolve();
+		}
+		if (notifications !== demandUpdates * 2 * observerCount) throw new Error(`lost demand edge: ${notifications}`);
+		console.log(
+			`${trackCount},${observerCount},${(((performance.now() - start) * 1000) / demandUpdates).toFixed(1)}`,
+		);
+		for (const dispose of disposes) dispose();
+		for (const track of tracks) track.close();
+		broadcast.close();
+	}
+}
diff --git a/js/net/src/broadcast.test.ts b/js/net/src/broadcast.test.ts
index 9c77d51b06..1895c3365e 100644
--- a/js/net/src/broadcast.test.ts
+++ b/js/net/src/broadcast.test.ts
@@ -503,3 +503,45 @@ test("aborting a fetch rejects with the signal's reason", async () => {
 
 	broadcast.close();
 });
+
+test("broadcast demand watches static and pending tracks rather than broadcast handles", async () => {
+	const broadcast = new BroadcastProducer();
+	const demand = broadcast.demand();
+	const consumer = broadcast.consume();
+	expect(demand.used.peek()).toBe(false);
+	const video = broadcast.createTrack("video");
+	const first = video.subscribe();
+	await Promise.resolve();
+	expect(demand.used.peek()).toBe(true);
+	const second = consumer.track("audio").subscribe();
+	const request = await wireOf(broadcast).requested();
+	if (!request) throw new Error("expected request");
+	const requested = request.demand();
+	expect(requested.used.peek()).toBe(true);
+	first.close();
+	await Promise.resolve();
+	expect(demand.used.peek()).toBe(true);
+	second.close();
+	await requested.unused();
+	await demand.unused();
+	expect(demand.used.peek()).toBe(false);
+	request.reject();
+	video.close();
+	broadcast.close();
+	expect(await demand.closed).toBeNull();
+	consumer.close();
+});
+
+test("broadcast closure clears active demand and subscriptions", async () => {
+	const broadcast = new BroadcastProducer();
+	const track = broadcast.createTrack("video");
+	const subscriber = track.subscribe();
+	const demand = broadcast.demand();
+	await Promise.resolve();
+	expect(demand.used.peek()).toBe(true);
+	broadcast.close();
+	expect(demand.used.peek()).toBe(false);
+	await demand.unused();
+	subscriber.close();
+	track.close();
+});
diff --git a/js/net/src/broadcast.ts b/js/net/src/broadcast.ts
index d477986c99..99efcd09c3 100644
--- a/js/net/src/broadcast.ts
+++ b/js/net/src/broadcast.ts
@@ -3,7 +3,7 @@
  *
  * @module
  */
-import { type GetPromise, Once, Signal } from "@moq/signals";
+import { type Dispose, type GetPromise, type Getter, Once, Signal } from "@moq/signals";
 import { NotFound } from "./error.ts";
 import type { Consumer as GroupConsumer } from "./group.ts";
 import { Route } from "./hop.ts";
@@ -34,6 +34,42 @@ class BroadcastState {
 	// Live consumer handles sharing this state (see {@link Consumer.clone}). The broadcast
 	// closes once the last one closes, so a shared consumer can be handed to several callers.
 	consumers = 0;
+	used = new Signal(false);
+	active = 0;
+	watched = new WeakSet<track.Producer>();
+	demandCleanup = new Set<Dispose>();
+}
+
+// Each track updates the aggregate on an edge, so a demand change touches only its track.
+function watchDemand(state: BroadcastState, producer: track.Producer): void {
+	if (state.watched.has(producer)) return;
+	state.watched.add(producer);
+	const demand = producer.demand();
+	let active = false;
+	const update = () => {
+		const used = state.closed.peek() === undefined && demand.closed.peek() === undefined && demand.used.peek();
+		if (active === used) return;
+		state.active += used ? 1 : -1;
+		active = used;
+		state.used.set(state.active > 0);
+	};
+	const disposeUsed = demand.used.subscribe(update);
+	const disposeClosed = demand.closed.subscribe(() => {
+		if (demand.closed.peek() === undefined) return;
+		cleanup();
+	});
+	const cleanup = () => {
+		disposeUsed();
+		disposeClosed();
+		if (active) {
+			state.active--;
+			active = false;
+			state.used.set(state.active > 0);
+		}
+		state.demandCleanup.delete(cleanup);
+	};
+	state.demandCleanup.add(cleanup);
+	update();
 }
 
 function dequeueRequest(state: BroadcastState): track.Request | undefined {
@@ -51,6 +87,7 @@ function dequeueRequest(state: BroadcastState): track.Request | undefined {
 function closeState(state: BroadcastState) {
 	if (state.closed.peek() !== undefined) return;
 	state.closed.set(null);
+	for (const cleanup of state.demandCleanup) cleanup();
 	for (const request of state.pending) request.reject();
 	state.requested.mutate((requests) => {
 		requests.length = 0;
@@ -81,6 +118,7 @@ function subscribe(
 	}
 
 	const producer = new track.Producer(name);
+	watchDemand(state, producer);
 	const subscriber = producer.subscribe(options);
 
 	if (register) {
@@ -110,6 +148,7 @@ async function resolveTrackInfo(state: BroadcastState, name: string): Promise<tr
 	}
 
 	const producer = new track.Producer(name);
+	watchDemand(state, producer);
 	state.requested.mutate((requested) => {
 		requested.push(hooks.makeRequest({ name, producer, sequences: state.sequences, pending: state.pending }));
 	});
@@ -154,6 +193,36 @@ async function fetchGroup(
 	}
 }
 
+let makeDemand: (state: BroadcastState) => Demand;
+
+/** A watch-only view of demand for any track in a broadcast. */
+export class Demand {
+	#state: BroadcastState;
+	private constructor(state: BroadcastState) {
+		this.#state = state;
+	}
+	static {
+		makeDemand = (state) => new Demand(state);
+	}
+
+	/** Whether any track currently has subscribers. */
+	get used(): Getter<boolean> {
+		return this.#state.used;
+	}
+
+	/** Wait until every track becomes unused or the broadcast closes. */
+	async unused(): Promise<void> {
+		while (this.#state.used.peek() && this.#state.closed.peek() === undefined) {
+			await Signal.race(this.#state.used, this.#state.closed);
+		}
+	}
+
+	/** The broadcast's clean close. */
+	get closed(): GetPromise<null> {
+		return this.#state.closed;
+	}
+}
+
 /**
  * The write side of a broadcast.
  *
@@ -161,6 +230,7 @@ async function fetchGroup(
  */
 export class Producer {
 	#state = new BroadcastState();
+	#demand = makeDemand(this.#state);
 	#announcer?: Announcer;
 	#path = Path.empty();
 
@@ -186,6 +256,11 @@ export class Producer {
 		return this.#state.closed;
 	}
 
+	/** Watch demand for the broadcast's tracks. */
+	demand(): Demand {
+		return this.#demand;
+	}
+
 	/** A read handle for this broadcast, named by the path the origin created it at. */
 	consume(): Consumer {
 		return makeConsumer({ state: this.#state, path: this.#path });
@@ -213,6 +288,7 @@ export class Producer {
 			throw new Error(`duplicate track: ${track.name}`);
 		}
 
+		watchDemand(this.#state, track);
 		this.#state.tracks.set(track.name, track);
 
 		void track.closed.then(() => {
diff --git a/js/net/src/group.test.ts b/js/net/src/group.test.ts
index 5bb5187199..d7b8b5733e 100644
--- a/js/net/src/group.test.ts
+++ b/js/net/src/group.test.ts
@@ -9,21 +9,21 @@ test("used reflects mirror demand and unused resolves when the last reader leave
 	const producer = new Producer(0);
 
 	// No mirror readers: no demand.
-	expect(producer.used.peek()).toBe(false);
+	expect(producer.demand().used.peek()).toBe(false);
 
 	const a = producer.mirror();
 	const b = producer.mirror();
-	expect(producer.used.peek()).toBe(true);
+	expect(producer.demand().used.peek()).toBe(true);
 
 	// Closing one of two keeps demand, so unused() stays pending.
 	a.close();
-	expect(producer.used.peek()).toBe(true);
+	expect(producer.demand().used.peek()).toBe(true);
 
 	// Closing the last reader drops demand; unused() resolves. Fetch coalescing awaits this to
 	// cancel a download that everyone has abandoned (a group may never end on its own).
 	b.close();
-	await producer.unused();
-	expect(producer.used.peek()).toBe(false);
+	await producer.demand().unused();
+	expect(producer.demand().used.peek()).toBe(false);
 });
 
 function pair(sequence: number) {
diff --git a/js/net/src/group.ts b/js/net/src/group.ts
index f1c47f082c..aa6990d4bd 100644
--- a/js/net/src/group.ts
+++ b/js/net/src/group.ts
@@ -82,6 +82,43 @@ function appendFrame(state: GroupState, frame: Frame) {
 	state.total.update((total) => total + 1);
 }
 
+let makeDemand: (sequence: number, used: Getter<boolean>, state: GroupState) => Demand;
+
+/** A watch-only view of a group's mirror readers, including coalesced fetches. */
+export class Demand {
+	/** The group sequence this handle watches. */
+	readonly sequence: number;
+	#used: Getter<boolean>;
+	#state: GroupState;
+
+	private constructor(sequence: number, used: Getter<boolean>, state: GroupState) {
+		this.sequence = sequence;
+		this.#used = used;
+		this.#state = state;
+	}
+
+	static {
+		makeDemand = (sequence, used, state) => new Demand(sequence, used, state);
+	}
+
+	/** Whether any mirror reader is attached. */
+	get used(): Getter<boolean> {
+		return this.#used;
+	}
+
+	/** Wait until all mirror readers leave or the group closes. */
+	async unused(): Promise<void> {
+		while (this.#used.peek() && this.#state.closed.peek() === undefined) {
+			await Signal.race(this.#used, this.#state.closed);
+		}
+	}
+
+	/** The group's clean close or abort error. */
+	get closed(): GetPromise<Error | null> {
+		return this.#state.closed;
+	}
+}
+
 /**
  * The write side of an ordered stream of frames within a track.
  *
@@ -94,10 +131,11 @@ export class Producer {
 	#state: GroupState;
 	#mirrors?: Set<GroupState>;
 
-	// Whether any mirror reader is attached (see {@link used}). Fetch coalescing watches it to
+	// Whether any mirror reader is attached (see {@link Demand.used}). Fetch coalescing watches it to
 	// cancel an abandoned download; a group can stay open indefinitely (a catalog or JSON stream),
 	// so this is what stops a reader-less fetch instead of the stream ending on its own.
 	#used = new Signal<boolean>(false);
+	#demand: Demand;
 
 	// When the group was created or last written. Retention ages a group from this,
 	// not from its creation, so a group still being filled is never reclaimed.
@@ -106,6 +144,7 @@ export class Producer {
 	constructor(sequence: number) {
 		this.#state = new GroupState(sequence);
 		this.sequence = sequence;
+		this.#demand = makeDemand(sequence, this.#used, this.#state);
 	}
 
 	/**
@@ -171,28 +210,9 @@ export class Producer {
 		return makeConsumer(dst);
 	}
 
-	/**
-	 * Whether any mirror reader is currently attached.
-	 *
-	 * Pairs with {@link unused}. Fetch coalescing watches it to cancel a download once every reader
-	 * has gone: a group can stay open indefinitely (a catalog track, a JSON stream), so it can't
-	 * rely on the stream ending on its own.
-	 *
-	 * @internal Track fan-out and fetch coalescing only.
-	 */
-	get used(): Getter<boolean> {
-		return this.#used;
-	}
-
-	/**
-	 * Resolves once no mirror reader remains (or the group closes).
-	 *
-	 * @internal Track fan-out and fetch coalescing only.
-	 */
-	async unused(): Promise<void> {
-		while (this.#used.peek() && this.#state.closed.peek() === undefined) {
-			await Signal.race(this.#used, this.#state.closed);
-		}
+	/** Watch the mirror readers attached to this group. */
+	demand(): Demand {
+		return this.#demand;
 	}
 
 	/** Writes a frame to the group. */
diff --git a/js/net/src/integration.test.ts b/js/net/src/integration.test.ts
index fcde98ca10..c86548cdef 100644
--- a/js/net/src/integration.test.ts
+++ b/js/net/src/integration.test.ts
@@ -1323,12 +1323,12 @@ test("integration: lite fetch teardown when the reader abandons an open group",
 	expect(await fetched.readString()).toBe("hello");
 
 	// The publisher is now serving the still-open group.
-	await waitUntil(() => group.used.peek());
+	await waitUntil(() => group.demand().used.peek());
 
 	// Abandoning the fetch cancels the FETCH stream, so the publisher stops serving instead of
 	// pumping an open group to a reader that left.
 	fetched.close();
-	await waitUntil(() => !group.used.peek());
+	await waitUntil(() => !group.demand().used.peek());
 
 	group.close();
 	broadcast.close();
@@ -1431,17 +1431,17 @@ test("integration: lite coalesced fetch stays until every reader abandons the op
 	const f2 = await remote.track("video").fetchGroup(group.sequence);
 	expect(await f1.readString()).toBe("hello");
 	expect(await f2.readString()).toBe("hello");
-	await waitUntil(() => group.used.peek());
+	await waitUntil(() => group.demand().used.peek());
 
 	// Closing one coalesced reader keeps the shared FETCH flowing for the other.
 	f1.close();
 	group.writeString("more");
 	expect(await f2.readString()).toBe("more");
-	expect(group.used.peek()).toBe(true);
+	expect(group.demand().used.peek()).toBe(true);
 
 	// The last abandon cancels the FETCH.
 	f2.close();
-	await waitUntil(() => !group.used.peek());
+	await waitUntil(() => !group.demand().used.peek());
 
 	group.close();
 	broadcast.close();
diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts
index 228ec19d49..8b7d8cc1a4 100644
--- a/js/net/src/lite/subscriber.ts
+++ b/js/net/src/lite/subscriber.ts
@@ -868,7 +868,10 @@ export class Subscriber {
 			// promise; the check is level-triggered, so a coalesced fetch that arrives before we
 			// cancel re-arms and resumes.
 			const idle: unique symbol = Symbol("idle");
-			let unused = group.unused().then((): typeof idle => idle);
+			let unused = group
+				.demand()
+				.unused()
+				.then((): typeof idle => idle);
 			// A decode consumes its frame whenever it lands, so one outstanding across a re-arm is
 			// kept and awaited again rather than abandoned with its frame.
 			let pending: Promise<netGroup.Frame | undefined> | undefined;
@@ -879,8 +882,11 @@ export class Subscriber {
 					pending ??= stream.reader.decodeMaybe(decode);
 					const next = await race([pending, group.closed, unused]);
 					if (next === idle) {
-						if (!group.isClosed && group.used.peek()) {
-							unused = group.unused().then((): typeof idle => idle);
+						if (!group.isClosed && group.demand().used.peek()) {
+							unused = group
+								.demand()
+								.unused()
+								.then((): typeof idle => idle);
 							continue;
 						}
 						break;
@@ -1271,9 +1277,18 @@ async function untilClosed<T>(group: netGroup.Producer, step: Promise<T>): Promi
 async function untilAbandoned<T>(group: netGroup.Producer, step: Promise<T>): Promise<T> {
 	const idle: unique symbol = Symbol("idle");
 	for (;;) {
-		const value = await untilClosed(group, race([step, group.unused().then((): typeof idle => idle)]));
+		const value = await untilClosed(
+			group,
+			race([
+				step,
+				group
+					.demand()
+					.unused()
+					.then((): typeof idle => idle),
+			]),
+		);
 		if (value !== idle) return value as T;
-		if (!group.used.peek()) throw new StreamError(StreamCode.Cancel, { message: "cancel" });
+		if (!group.demand().used.peek()) throw new StreamError(StreamCode.Cancel, { message: "cancel" });
 	}
 }
 
diff --git a/js/net/src/track.ts b/js/net/src/track.ts
index e4d6a63ba1..cb53fbf779 100644
--- a/js/net/src/track.ts
+++ b/js/net/src/track.ts
@@ -217,6 +217,11 @@ export class Request {
 		hooks.pendingTrackProducer = (request) => request.#producer;
 	}
 
+	/** Watch the subscribers waiting for this request. */
+	demand(): Demand {
+		return this.#producer.demand();
+	}
+
 	/** The aggregate subscription requested for this track. */
 	get subscription(): Readonly<Subscription> {
 		return this.#producer.subscription.peek() ?? subscriptionDefaults();
@@ -415,7 +420,7 @@ let ordered_: {
 let makeDemand: (name: string, used: Getter<boolean>, state: TrackState) => Demand;
 
 /**
- * A watch-only view of a track's subscriber demand, from {@link Producer.demand}. Mirrors the
+ * A watch-only view of a track's subscriber demand, from {@link Producer.demand} or {@link Request.demand}. Mirrors the
  * Rust `Demand`.
  *
  * Lets a publisher gate work (on-demand capture or encoding) on whether anyone is subscribed,
diff --git a/rs/moq-ffi/src/producer.rs b/rs/moq-ffi/src/producer.rs
index dfe9049474..59de359329 100644
--- a/rs/moq-ffi/src/producer.rs
+++ b/rs/moq-ffi/src/producer.rs
@@ -881,7 +881,7 @@ impl MoqGroupProducer {
 	/// Wait until a consumer has this group.
 	pub(crate) async fn used(&self) -> Result<(), MoqError> {
 		let producer = self.inner.lock().unwrap().as_ref().ok_or(MoqError::Closed)?.clone();
-		Ok(producer.used().await?)
+		Ok(producer.demand().used().await?)
 	}
 }
 
diff --git a/rs/moq-gst/src/source/imp.rs b/rs/moq-gst/src/source/imp.rs
index 3d773be299..d309567641 100644
--- a/rs/moq-gst/src/source/imp.rs
+++ b/rs/moq-gst/src/source/imp.rs
@@ -1278,11 +1278,12 @@ mod session_tests {
 			.block_on(async {
 				tokio::time::timeout(
 					Duration::from_secs(10),
-					moq_net::kio::wait(|waiter| request.poll_unused(waiter)),
+					moq_net::kio::wait(|waiter| request.demand().poll_unused(waiter)),
 				)
 				.await
 			})
-			.expect("the cancelled pump never dropped its subscription");
+			.expect("the cancelled pump never dropped its subscription")
+			.expect("the pending request is still open");
 
 		// Only now answer it. The pump is gone and its state is terminal, so no later scheduling
 		// can produce a pad.
diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs
index 6165ea47a5..0f2bf90522 100644
--- a/rs/moq-net/src/ietf/subscriber.rs
+++ b/rs/moq-net/src/ietf/subscriber.rs
@@ -1768,7 +1768,7 @@ where
 					else {
 						return Poll::Ready(Setup::Gone);
 					};
-					if pending.poll_unused(waiter).is_ready() {
+					if pending.demand().poll_unused(waiter).is_ready() {
 						return Poll::Ready(Setup::Unused);
 					}
 					Poll::Pending
diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs
index 4e85ed56ff..b870528817 100644
--- a/rs/moq-net/src/lite/subscriber.rs
+++ b/rs/moq-net/src/lite/subscriber.rs
@@ -3719,7 +3719,7 @@ impl<S: crate::transport::poll::Session> kio::Task for TrackServeRun<S> {
 					// otherwise hold this task, its TRACK stream, and the track for good.
 					// Dropping the fetch resets the stream.
 					let pending = request.as_ref().expect("request pending");
-					if pending.poll_unused(waiter).is_ready() {
+					if pending.demand().poll_unused(waiter).is_ready() {
 						if pending.reject_unused(Error::Cancel) {
 							self.state = TrackRunState::Done;
 							return Poll::Ready(());
diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs
index e9f3a4bbda..e4a2c4aca2 100644
--- a/rs/moq-net/src/model/group.rs
+++ b/rs/moq-net/src/model/group.rs
@@ -949,19 +949,97 @@ impl Producer {
 		self.state.poll_closed(waiter).map(|()| self.abort_reason())
 	}
 
-	/// Block until there is at least one active consumer.
+	/// Watch reader demand without keeping the group alive.
+	pub fn demand(&self) -> Demand {
+		Demand {
+			sequence: self.info.sequence,
+			state: DemandSource::Group(self.state.weak()),
+		}
+	}
+
+	/// The recorded abort reason, or [`Error::Dropped`] if the group closed without one.
+	fn abort_reason(&self) -> Error {
+		self.state.read().abort.clone().unwrap_or(Error::Dropped)
+	}
+}
+
+/// A cloneable, watch-only handle to a group's readers or pending fetch callers.
+#[derive(Clone)]
+pub struct Demand {
+	sequence: u64,
+	state: DemandSource,
+}
+
+#[derive(Clone)]
+enum DemandSource {
+	Group(kio::ProducerWeak<GroupState>),
+	Fetch(kio::ProducerWeak<track::FetchOutcome>),
+}
+
+impl Demand {
+	pub(crate) fn fetch(sequence: u64, state: kio::ProducerWeak<track::FetchOutcome>) -> Self {
+		Self {
+			sequence,
+			state: DemandSource::Fetch(state),
+		}
+	}
+
+	/// The group sequence this handle watches.
+	pub fn sequence(&self) -> u64 {
+		self.sequence
+	}
+
+	/// Whether the group or fetch is open and has readers right now.
+	pub fn is_used(&self) -> bool {
+		match &self.state {
+			DemandSource::Group(state) => !state.is_closed() && state.is_used(),
+			DemandSource::Fetch(state) => !state.is_closed() && state.is_used(),
+		}
+	}
+
+	/// Wait until at least one reader needs the group.
 	pub async fn used(&self) -> Result<()> {
-		self.state.used().await.map_err(|_| self.abort_reason())
+		kio::wait(|waiter| self.poll_used(waiter)).await
 	}
 
-	/// Block until there are no active consumers.
+	/// Wait until no reader needs the group.
 	pub async fn unused(&self) -> Result<()> {
-		self.state.unused().await.map_err(|_| self.abort_reason())
+		kio::wait(|waiter| self.poll_unused(waiter)).await
+	}
+
+	/// Poll until at least one reader needs the group.
+	pub fn poll_used(&self, waiter: &kio::Waiter) -> Poll<Result<()>> {
+		match &self.state {
+			DemandSource::Group(state) => state.poll_used(waiter),
+			DemandSource::Fetch(state) => state.poll_used(waiter),
+		}
+		.map_err(|_| self.abort_reason())
+	}
+
+	/// Poll until no reader needs the group.
+	pub fn poll_unused(&self, waiter: &kio::Waiter) -> Poll<Result<()>> {
+		match &self.state {
+			DemandSource::Group(state) => state.poll_unused(waiter),
+			DemandSource::Fetch(state) => state.poll_unused(waiter),
+		}
+		.map_err(|_| self.abort_reason())
+	}
+
+	/// Wait until the group or fetch closes, returning its cause.
+	pub async fn closed(&self) -> Error {
+		match &self.state {
+			DemandSource::Group(state) => state.closed().await,
+			DemandSource::Fetch(state) => state.closed().await,
+		}
+		self.abort_reason()
 	}
 
-	/// The recorded abort reason, or [`Error::Dropped`] if the group closed without one.
 	fn abort_reason(&self) -> Error {
-		self.state.read().abort.clone().unwrap_or(Error::Dropped)
+		match &self.state {
+			DemandSource::Group(state) => state.read().abort.clone(),
+			DemandSource::Fetch(state) => state.read().rejected.clone(),
+		}
+		.unwrap_or(Error::Dropped)
 	}
 }
 
@@ -1825,6 +1903,24 @@ mod test {
 	use bytes::Bytes;
 	use futures::FutureExt;
 
+	#[test]
+	fn demand_watches_readers_without_keeping_the_group_alive() {
+		let producer = Info { sequence: 7 }.produce();
+		let demand = producer.demand();
+		assert_eq!(demand.sequence(), 7);
+		assert!(!demand.is_used());
+		let first = producer.consume();
+		let second = first.clone();
+		assert!(demand.is_used());
+		drop(first);
+		assert!(demand.poll_unused(&kio::Waiter::noop()).is_pending());
+		drop(second);
+		assert!(matches!(demand.poll_unused(&kio::Waiter::noop()), Poll::Ready(Ok(()))));
+		drop(producer);
+		assert!(!demand.is_used());
+		assert!(matches!(demand.used().now_or_never(), Some(Err(Error::Dropped))));
+	}
+
 	/// [`FRAME_SLOTS`] is std's rounding, not ours, so measure it: a larger real value
 	/// would undercharge every cached group without touching a line of this crate.
 	#[test]
diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs
index 921c72af34..de07522335 100644
--- a/rs/moq-net/src/model/track.rs
+++ b/rs/moq-net/src/model/track.rs
@@ -1836,9 +1836,12 @@ impl Dynamic {
 		poll_requested_group(&self.state, &self.fetch, waiter)
 	}
 
-	/// Poll for the track becoming unused (every consumer dropped).
-	pub fn poll_unused(&self, waiter: &kio::Waiter) -> Poll<()> {
-		self.state.poll_unused(waiter).map(|_| ())
+	/// Watch subscriber demand without keeping the track alive.
+	pub fn demand(&self) -> Demand {
+		Demand {
+			name: self.name.clone(),
+			state: self.state.weak(),
+		}
 	}
 }
 
@@ -2199,7 +2202,8 @@ impl super::WeakEntry for TrackWeak {
 
 /// A cloneable, watch-only handle to a track's subscriber demand.
 ///
-/// Obtained from [`Producer::demand`]. A publisher uses it to react to
+/// Obtained from [`Producer::demand`], [`Request::demand`], or [`Dynamic::demand`].
+/// A publisher uses it to react to
 /// whether anyone is subscribed (on-demand capture / encoding) without being able
 /// to publish frames or close the track. It's a weak handle, so it neither keeps
 /// the track alive nor pins its cached groups; once the owning [`Producer`]
@@ -2961,6 +2965,11 @@ impl kio::Task for Querying {
 }
 
 impl group::Request {
+	/// Watch the callers waiting for this fetch without keeping the attempt alive.
+	pub fn demand(&self) -> group::Demand {
+		group::Demand::fetch(self.sequence, self.result.weak())
+	}
+
 	/// The group sequence the consumer wants.
 	pub fn sequence(&self) -> u64 {
 		self.sequence
@@ -4347,10 +4356,12 @@ impl Request {
 		Dynamic::new(self.name.clone(), self.state.clone(), self.alive.clone())
 	}
 
-	/// Poll for the request becoming unused (every consumer dropped), so a relay can
-	/// stop serving and drop the request.
-	pub fn poll_unused(&self, waiter: &kio::Waiter) -> Poll<()> {
-		self.state.poll_unused(waiter).map(|_| ())
+	/// Watch subscriber demand without keeping the track alive.
+	pub fn demand(&self) -> Demand {
+		Demand {
+			name: self.name.clone(),
+			state: self.state.weak(),
+		}
 	}
 
 	/// Mark this request as taken by a dynamic handler, which alone decides its answer.
@@ -4362,7 +4373,7 @@ impl Request {
 	}
 
 	/// Reject only while no consumer needs this pending track. Demand and the check
-	/// share one lock, so demand returning after `poll_unused` wins the race, and the
+	/// share one lock, so demand returning after `Demand::poll_unused` wins the race, and the
 	/// close under that lock stops a later consumer attaching to a dead request.
 	pub(crate) fn reject_unused(&self, err: Error) -> bool {
 		match self.state.write_unused() {
@@ -8933,6 +8944,42 @@ mod test {
 		assert!(pool.used() < used, "its bytes are released");
 	}
 
+	#[test]
+	fn request_and_dynamic_share_weak_track_demand() {
+		let request = Request::new(Arc::new(broadcast::Info::default()), "demand");
+		let dynamic = request.dynamic();
+		let demand = request.demand();
+		let consumer = request.consume();
+		assert!(demand.is_used());
+		assert!(dynamic.demand().is_used());
+		drop(consumer);
+		assert!(!demand.is_used());
+		let producer = request.accept(None);
+		assert_eq!(producer.demand().name(), demand.name());
+		drop(producer);
+		drop(dynamic);
+		assert!(matches!(demand.used().now_or_never(), Some(Err(Error::Dropped))));
+	}
+
+	#[test]
+	fn fetch_request_demand_counts_every_joined_caller() {
+		let producer = track_producer("test", None);
+		let dynamic = producer.dynamic();
+		let consumer = producer.consume();
+		let first = consumer.fetch_group(3, None);
+		let second = consumer.fetch_group(3, None);
+		let request = dynamic.requested_group().now_or_never().unwrap().unwrap();
+		let demand = request.demand();
+		assert_eq!(demand.sequence(), 3);
+		assert!(demand.is_used());
+		drop(first);
+		assert!(demand.poll_unused(&kio::Waiter::noop()).is_pending());
+		drop(second);
+		assert!(matches!(demand.poll_unused(&kio::Waiter::noop()), Poll::Ready(Ok(()))));
+		request.reject(Error::Cancel);
+		assert!(matches!(demand.used().now_or_never(), Some(Err(Error::Cancel))));
+	}
+
 	#[tokio::test]
 	async fn fetch_aborts_with_track() {
 		let producer = track_producer("test", None);
diff --git a/rs/moq-net/tests/track_info_cancel.rs b/rs/moq-net/tests/track_info_cancel.rs
index 1f8be9be93..5ae79fb57a 100644
--- a/rs/moq-net/tests/track_info_cancel.rs
+++ b/rs/moq-net/tests/track_info_cancel.rs
@@ -46,9 +46,13 @@ async fn abandoned_request_reaches_the_publisher(version: &str, relays: u64) {
 
 	// The publisher never answers; the reader gives up.
 	drop(waiting);
-	tokio::time::timeout(Duration::from_secs(5), kio::wait(|waiter| request.poll_unused(waiter)))
-		.await
-		.unwrap_or_else(|_| panic!("{version:?} over {relays} relays: the publisher's request is still wanted"));
+	tokio::time::timeout(
+		Duration::from_secs(5),
+		kio::wait(|waiter| request.demand().poll_unused(waiter)),
+	)
+	.await
+	.unwrap_or_else(|_| panic!("{version:?} over {relays} relays: the publisher's request is still wanted"))
+	.expect("the pending request is still open");
 }
 
 #[tokio::test(start_paused = true)]
diff --git a/rs/moq-stats/src/produce.rs b/rs/moq-stats/src/produce.rs
index 427e255bc6..0d4887032b 100644
--- a/rs/moq-stats/src/produce.rs
+++ b/rs/moq-stats/src/produce.rs
@@ -480,7 +480,7 @@ impl PendingPair {
 		self.plain
 			.iter()
 			.chain(self.compressed.iter())
-			.any(|request| request.poll_unused(waiter).is_pending())
+			.any(|request| request.demand().poll_unused(waiter).is_pending())
 	}
 }
 

From 1d82ae442972e584270c1e88ab2c7a0277acc0ad Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Mon, 5 Oct 2026 10:59:40 -0700
Subject: [PATCH 3/5] fix(net): withdraw a fetch when its last caller leaves

The last fetch_group caller dropping its Fetching removes the attempt from
the fetch table under the fetch lock, atomic with a join, purging a still
queued attempt too. group::Request::poll_unused is gone; handlers watch
demand() and drop the request. JS lite closes an abandoned fetch's group at
the cancel decision so no fetch coalesces onto it in between.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 doc/lib/rs/moq-net.md                 |   2 +
 doc/setup/upgrade.md                  |  16 ++-
 js/net/bench/broadcasts.ts            |   3 +-
 js/net/src/lite/subscriber.test.ts    |  39 ++++++
 js/net/src/lite/subscriber.ts         |   8 +-
 quest/m1/README.md                    |   1 -
 quest/m1/request-linger.md            |   9 +-
 quest/m1/unannounce-demand-release.md |   1 -
 rs/moq-net/src/ietf/subscriber.rs     |   2 +-
 rs/moq-net/src/lite/publisher.rs      |   7 +-
 rs/moq-net/src/lite/subscriber.rs     |   2 +-
 rs/moq-net/src/model/requests.rs      |   7 +-
 rs/moq-net/src/model/track.rs         | 164 ++++++++++++++++++++------
 rs/moq-net/tests/fetch_cancel.rs      |  12 +-
 rs/moq-net/tests/loom.rs              |  38 ++++++
 15 files changed, 254 insertions(+), 57 deletions(-)

diff --git a/doc/lib/rs/moq-net.md b/doc/lib/rs/moq-net.md
index 4baa1fbbdf..5fa5089f8b 100644
--- a/doc/lib/rs/moq-net.md
+++ b/doc/lib/rs/moq-net.md
@@ -188,5 +188,7 @@ Watch `used()` / `unused()` (or `poll_used` / `poll_unused`) on these handles,
 or take a snapshot with `is_used()`. Demand handles cannot write or abort and
 do not keep a producer or request alive. A pending group request counts every
 caller sharing its fetch, so one caller leaving does not cancel the others.
+The last caller to leave withdraws the fetch, so a later fetch of that group
+starts a fresh request; a handler that sees the request unused just drops it.
 Use the producer's `abort_unused` to atomically end an idle track; observing
 unused demand alone does not protect against a new subscriber arriving.
diff --git a/doc/setup/upgrade.md b/doc/setup/upgrade.md
index 88ee9c0c3a..1bf57db3c9 100644
--- a/doc/setup/upgrade.md
+++ b/doc/setup/upgrade.md
@@ -31,12 +31,20 @@ These land with the next breaking release, not the 2026-09-23 train.
   are `publish_flate_snapshot` / `publish_flate_stream`, taking
   `MoqFlateConfig` and returning `MoqFlateSnapshotProducer` /
   `MoqFlateStreamProducer`. The C `moq_publish_binary_*` calls are unchanged.
-- **Track demand is read through `demand()`.** In Rust, `track::Producer`'s
-  `is_used`, `used`, `unused`, and `poll_unused` are `producer.demand().X`.
+- **Demand is read through `demand()`.** In Rust, `track::Producer`'s
+  `is_used`, `used`, `unused`, and `poll_unused` are `producer.demand().X`,
+  and so are `group::Producer`'s `used` and `unused`.
+  `track::Request::poll_unused`, `track::Dynamic::poll_unused`, and
+  `group::Request::poll_unused` are `demand().poll_unused`, which returns
+  `Poll<Result<()>>` instead of `Poll<()>`: an `Err` means the request closed,
+  so treat it as unused too. A `group::Request` no longer needs polling to be
+  withdrawn; the last `fetch_group` caller leaving does it, so a handler that
+  sees it unused just drops it.
   The moq-json snapshot and moq-flate `is_used()` is `demand().is_used()`.
   In TypeScript, `Track.Producer`'s `used` and `unused()` are
-  `producer.demand().used` and `.unused()`, and `Allocator.reserve` takes
-  `producer.demand()`, replacing the `Bandwidth.Demand` interface.
+  `producer.demand().used` and `.unused()`, as are `Group.Producer`'s, and
+  `Allocator.reserve` takes `producer.demand()`, replacing the
+  `Bandwidth.Demand` interface.
 - **moq-mux has no clock translators.** `clock::Anchor`, `clock::Lane`, and
   `SourceMap` (#4667) are gone, along with the importers' `live()`. Publish the
   source's own timestamps and let the catalog clock map them to wall time;
diff --git a/js/net/bench/broadcasts.ts b/js/net/bench/broadcasts.ts
index 29525c739a..80a37a3488 100644
--- a/js/net/bench/broadcasts.ts
+++ b/js/net/bench/broadcasts.ts
@@ -1,5 +1,6 @@
+/** Sweep observers against routes for one route update, and against tracks for one demand edge. */
+
 import { Producer as BroadcastProducer } from "../src/broadcast.ts";
-/** Sweep route and observer counts for a single touched-path route update. */
 import { Producer } from "../src/origin.ts";
 import * as Path from "../src/path.ts";
 
diff --git a/js/net/src/lite/subscriber.test.ts b/js/net/src/lite/subscriber.test.ts
index 888eb44472..7b668639bb 100644
--- a/js/net/src/lite/subscriber.test.ts
+++ b/js/net/src/lite/subscriber.test.ts
@@ -1006,6 +1006,45 @@ test.each([
 	subscriber.close();
 });
 
+// The fetch is withdrawn the moment it is cancelled, so a fetch arriving at any point after the
+// last sharer left either revives it or starts a fresh one, never joins the cancelled one.
+test("a fetch after the last sharer left is never failed with the cancelled one", async () => {
+	for (let depth = 0; depth < 32; depth++) {
+		const { quic, streams } = fakeSession();
+		const subscriber = new Subscriber(quic, Version.DRAFT_05, HopSchema.parse(1n));
+
+		const controller = new AbortController();
+		const a = subscriber.fetchGroup(Path.from("room"), "video", 0, { signal: controller.signal });
+		void a.catch(() => undefined);
+		await drainUntil(() => streams.length === 1);
+		await answerTrackInfo(streams[0]);
+		await drainUntil(() => streams.length === 2);
+		await streams[1].reading;
+
+		controller.abort(new Error("gone"));
+		for (let i = 0; i < depth; i++) await Promise.resolve();
+		const b = subscriber.fetchGroup(Path.from("room"), "video", 0).catch((err: unknown) => err);
+		for (let i = 0; i < 64; i++) await Promise.resolve();
+
+		// An empty-group FIN accepts whichever FETCH the late caller is waiting on.
+		if (streams.length > 2) {
+			await answerTrackInfo(streams[2]);
+			await drainUntil(() => streams.length === 4);
+			await streams[3].reading;
+			streams[3].inbound.close();
+		} else {
+			// Only open if the late caller revived it.
+			try {
+				streams[1].inbound.close();
+			} catch {}
+		}
+		const res = await b;
+		expect(res instanceof Error ? `depth ${depth}: ${res.message}` : undefined).toBeUndefined();
+
+		subscriber.close();
+	}
+});
+
 // A reader leaving partway through a fetched group cancels the FETCH: the truncated group
 // must not end clean, as a FIN would make it read whole.
 test("the last reader leaving mid-response cancels the fetch", async () => {
diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts
index 8d5b67bc33..9a6236b581 100644
--- a/js/net/src/lite/subscriber.ts
+++ b/js/net/src/lite/subscriber.ts
@@ -1288,7 +1288,13 @@ async function untilAbandoned<T>(group: netGroup.Producer, step: Promise<T>): Pr
 			]),
 		);
 		if (value !== idle) return value as T;
-		if (!group.demand().used.peek()) throw new StreamError(StreamCode.Cancel, { message: "cancel" });
+		if (!group.demand().used.peek()) {
+			// Close here rather than where the error lands, so no fetch coalesces onto the group
+			// in between only to fail with it.
+			const err = new StreamError(StreamCode.Cancel, { message: "cancel" });
+			group.close(err);
+			throw err;
+		}
 	}
 }
 
diff --git a/quest/m1/README.md b/quest/m1/README.md
index fef5e2a0e8..b36f90ff89 100644
--- a/quest/m1/README.md
+++ b/quest/m1/README.md
@@ -30,7 +30,6 @@ QUIC studies there on that rule.
 - [One route cost](/quest/m1/route-cost.md) - Warm and Cold collapse to one static route cost
 - [Delete removed cluster flags](/quest/m1/cluster-shims.md) - `mesh` and `linger` leave `cluster::Config` and their flags become unknown
 - [Track tail interop](/quest/m1/track-tail-interop.md) - a Rust publisher ending a track with a group in flight is read to its end by the JS subscriber, and the reverse, in `just test interop`
-- [Demand everywhere](/quest/m1/demand-everywhere.md) - requests and producers at every level watch subscribers through `demand()` alone, ahead of the bindings exposing it
 - [Request linger](/quest/m1/request-linger.md) - an upstream FETCH or SUBSCRIBE outlives its last reader by a short linger, so quick re-requests don't churn upstream
 - [Merge queue](/quest/m1/merge-queue.md) - the required checks run on `merge_group`, so a stale green check can no longer break main
 - [Flat questlines](/quest/m1/quest-flat-lines.md) - moq pins the current quest CLI, lands its questline branches on main, and retires them
diff --git a/quest/m1/request-linger.md b/quest/m1/request-linger.md
index 02c3561524..3d89ba3276 100644
--- a/quest/m1/request-linger.md
+++ b/quest/m1/request-linger.md
@@ -22,9 +22,6 @@ Decided in planning (2026-10-03, from [#4741](https://github.com/moq-dev/moq/pul
 - **A short fixed window**, around a second: enough to absorb a re-subscribe or
   a seek, far below the front's 30 s cache linger. A constant, measured during
   the quest.
-- Watches demand through the `demand()` handles
-  [Demand everywhere](/quest/m1/demand-everywhere.md) plumbs.
-
-## Required
-
-- [Demand everywhere](/quest/m1/demand-everywhere.md) - requests and producers at every level watch subscribers through `demand()` alone
+- Watches demand through the `demand()` handles. The last `fetch_group`
+  caller leaving withdraws its fetch at once, so a reader returning within the
+  linger starts a fresh request; the linger has to defer that withdrawal too.
diff --git a/quest/m1/unannounce-demand-release.md b/quest/m1/unannounce-demand-release.md
index 3e6712a08b..8a5b7002f5 100644
--- a/quest/m1/unannounce-demand-release.md
+++ b/quest/m1/unannounce-demand-release.md
@@ -74,6 +74,5 @@ out its linger, not forever.
 
 ## Related
 
-- [Demand everywhere](/quest/m1/demand-everywhere.md) - the `demand()` handles this watches
 - [Request linger](/quest/m1/request-linger.md) - bounds how long an abandoned upstream request may outlive its readers
 - [Broadcast epochs](/quest/m0/broadcast-epoch/README.md) - gates the release #4741 ships in
diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs
index 186e9cfcb2..e1b6bd34eb 100644
--- a/rs/moq-net/src/ietf/subscriber.rs
+++ b/rs/moq-net/src/ietf/subscriber.rs
@@ -3456,7 +3456,7 @@ where
 					if let Poll::Ready(res) = waiter.poll_future(response.as_mut()) {
 						return Poll::Ready(Some(res));
 					}
-					request.poll_unused(waiter).map(|()| None)
+					request.demand().poll_unused(waiter).map(|_| None)
 				})
 				.await
 			}
diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs
index d13acaae45..861b82abf1 100644
--- a/rs/moq-net/src/lite/publisher.rs
+++ b/rs/moq-net/src/lite/publisher.rs
@@ -3992,7 +3992,7 @@ mod tests {
 				.expect("track requested")
 				.unwrap();
 			assert!(
-				track.poll_unused(&waiter).is_pending(),
+				track.demand().poll_unused(&waiter).is_pending(),
 				"{case}: nobody wants the track"
 			);
 			track
@@ -4002,7 +4002,10 @@ mod tests {
 		assert!(drive(&mut serve).await, "{case}: still serving");
 		drop(serve);
 		if let Some(track) = track {
-			let unused = tokio::time::timeout(Duration::from_millis(1), kio::wait(|waiter| track.poll_unused(waiter)));
+			let unused = tokio::time::timeout(
+				Duration::from_millis(1),
+				kio::wait(|waiter| track.demand().poll_unused(waiter)),
+			);
 			assert!(unused.await.is_ok(), "{case}: the track is still wanted");
 		}
 
diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs
index d44a13fe6c..ea801d25c8 100644
--- a/rs/moq-net/src/lite/subscriber.rs
+++ b/rs/moq-net/src/lite/subscriber.rs
@@ -4432,7 +4432,7 @@ impl<S: crate::transport::poll::Session> kio::Task for FetchServeRun<S> {
 			// A fetch nobody waits on any more is cancelled upstream, so the publisher stops
 			// serving it (and a relay there releases its own FETCH). Ingest has its own check.
 			if let Some(request) = self.state.request()
-				&& request.poll_unused(waiter).is_ready()
+				&& request.demand().poll_unused(waiter).is_ready()
 			{
 				tracing::debug!(track = %self.serve.name, group = self.group, "fetch abandoned");
 				if let FetchRunState::Send { stream, .. } | FetchRunState::Answer { stream, .. } =
diff --git a/rs/moq-net/src/model/requests.rs b/rs/moq-net/src/model/requests.rs
index 1e6d74a416..7b915b5c65 100644
--- a/rs/moq-net/src/model/requests.rs
+++ b/rs/moq-net/src/model/requests.rs
@@ -102,7 +102,8 @@ impl<K: Clone + Eq + Hash, V> Requests<K, V> {
 	}
 
 	/// Remove the pending request for `key` if `f` says it's the caller's own,
-	/// leaving a newer entry that replaced it alone.
+	/// leaving a newer entry that replaced it alone. Purges a still-queued key as
+	/// [`Self::take`] does, so a handler never pops a withdrawn request.
 	pub fn remove_if<Q>(&mut self, key: &Q, f: impl FnOnce(&V) -> bool) -> Option<V>
 	where
 		K: Borrow<Q>,
@@ -111,7 +112,7 @@ impl<K: Clone + Eq + Hash, V> Requests<K, V> {
 		if !self.pending.get(key).is_some_and(f) {
 			return None;
 		}
-		self.pending.remove(key)
+		self.take(key)
 	}
 
 	/// Returns `true` if a queued (not yet popped) request exists.
@@ -202,6 +203,8 @@ mod test {
 		*requests.join(&1).unwrap() = "new";
 		assert_eq!(requests.remove_if(&1, |v| *v == "old"), None);
 		assert_eq!(requests.remove_if(&1, |v| *v == "new"), Some("new"));
+		// Still queued, so the removal also purged its queue entry.
+		assert!(!requests.has_queued());
 	}
 
 	#[test]
diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs
index 3eca2be2ef..79718d16f3 100644
--- a/rs/moq-net/src/model/track.rs
+++ b/rs/moq-net/src/model/track.rs
@@ -1858,7 +1858,8 @@ fn poll_requested_group(
 		let sequence = guard.pop().expect("predicate guaranteed a request");
 		// The popped attempt stays pending, so a fetch in the window between hand-off
 		// and accept joins it instead of queueing a duplicate.
-		// `group::Request::{accept, reject, drop}` removes the entry.
+		// `group::Request::{accept, reject, drop}` removes the entry, as does the last
+		// `Fetching` leaving.
 		let pending = guard.get(&sequence).expect("popped key must be pending");
 		let priority = pending.priority;
 		let frame_start = pending.frame_start;
@@ -2678,20 +2679,20 @@ impl Consumer {
 				// for, so it fails cleanly and its retry queues a fresh attempt.
 				pending.priority = pending.priority.max(options.priority);
 				pending.frame_start = pending.frame_start.min(options.frame_start);
-				result = Some(pending.result.consume());
+				result = Some(Joined::new(&pending.result));
 			} else {
 				// Queue a new attempt. The handler gate is atomic with a handler
 				// dropping (no fetch stranded on a queue nobody drains); with no
 				// handler, `Fetching::poll` fails fast instead.
 				let producer = kio::Producer::<FetchOutcome>::default();
-				let consumer = producer.consume();
+				let joined = Joined::new(&producer);
 				let attempt = PendingFetch {
 					priority: options.priority,
 					frame_start: options.frame_start,
 					result: producer,
 				};
 				if fetch.insert(sequence, attempt).is_ok() {
-					result = Some(consumer);
+					result = Some(joined);
 				}
 			}
 		}
@@ -2858,6 +2859,9 @@ impl kio::Task for Querying {
 
 impl group::Request {
 	/// Watch the callers waiting for this fetch without keeping the attempt alive.
+	///
+	/// The last caller to leave withdraws the attempt, so a later fetch of the group
+	/// queues a fresh request rather than joining this one: once unused, drop it.
 	pub fn demand(&self) -> group::Demand {
 		group::Demand::fetch(self.sequence, self.result.weak())
 	}
@@ -2892,6 +2896,10 @@ impl group::Request {
 	/// info if the track hasn't been accepted yet (a fetch with no live subscription),
 	/// and is ignored once accepted. Returns [`Error::Duplicate`] if the group is
 	/// already present, or the track's abort error if it closed while pending.
+	///
+	/// Accepting after every caller left still caches the group, which a fetch that
+	/// queued a fresh request meanwhile resolves from; that request's own accept is then
+	/// [`Error::Duplicate`].
 	pub fn accept(mut self, info: impl Into<Option<Info>>) -> Result<group::Producer> {
 		self.done = true;
 		// Cache the group before removing the attempt: the joined fetches resolve
@@ -2930,28 +2938,6 @@ impl group::Request {
 		}
 	}
 
-	/// Poll for the request becoming unused (every waiting [`Consumer::fetch_group`]
-	/// dropped), so a handler can stop serving and drop the request.
-	///
-	/// Once ready, a later fetch of the group starts a fresh request rather than
-	/// joining this abandoned one.
-	pub fn poll_unused(&self, waiter: &kio::Waiter) -> Poll<()> {
-		loop {
-			// A closed channel has nothing left to serve either.
-			if ready!(self.result.poll_unused(waiter)).is_err() {
-				return Poll::Ready(());
-			}
-			// Fetches join under the fetch lock, so re-checking under it makes the
-			// withdrawal atomic with a join racing the last one leaving.
-			let mut fetch = self.fetch.lock();
-			if self.result.is_used() {
-				continue;
-			}
-			fetch.remove_if(&self.sequence, |pending| pending.result.same_channel(&self.result));
-			return Poll::Ready(());
-		}
-	}
-
 	/// Remove this attempt from the fetch state, unless a newer attempt for the same
 	/// sequence has already replaced it.
 	fn remove(&self) {
@@ -2987,9 +2973,9 @@ pub struct Fetching {
 	frame_start: u64,
 	// This caller's priority, for a fetch that moves to a front's routes.
 	priority: u8,
-	// The joined attempt's result channel; `None` on a cache hit, or when no handler
-	// existed to queue on.
-	result: Option<kio::Consumer<FetchOutcome>>,
+	// The attempt this fetch joined; `None` on a cache hit, or when no handler
+	// existed to queue on. Boxed: a miss already allocates, and the handle stays small.
+	result: Option<Box<Joined>>,
 	// The group already cached when the fetch was made, held so it stays wanted.
 	// Boxed: a group consumer dwarfs the rest of the handle.
 	hit: Option<Box<group::Consumer>>,
@@ -3000,6 +2986,40 @@ pub struct Fetching {
 	resume: Option<Box<super::resume::Fetching>>,
 }
 
+/// A [`Fetching`]'s place in a queued or in-flight attempt.
+struct Joined {
+	// Counts this caller in the attempt's demand and carries its outcome.
+	outcome: kio::Consumer<FetchOutcome>,
+	// Names the attempt in the fetch state, so the last caller out withdraws only its own.
+	attempt: kio::ProducerWeak<FetchOutcome>,
+}
+
+impl Joined {
+	fn new(result: &kio::Producer<FetchOutcome>) -> Box<Self> {
+		Box::new(Self {
+			outcome: result.consume(),
+			attempt: result.weak(),
+		})
+	}
+}
+
+impl Drop for Fetching {
+	fn drop(&mut self) {
+		let Some(joined) = self.result.take() else {
+			return;
+		};
+		let Joined { outcome, attempt } = *joined;
+		// Joins take the fetch lock, so leaving under it is atomic with one: the joiner
+		// either keeps the attempt wanted, or finds it withdrawn and queues a fresh one
+		// rather than joining one its handler is about to drop.
+		let mut fetch = self.fetch.lock();
+		drop(outcome);
+		if !attempt.is_used() {
+			fetch.remove_if(&self.sequence, |pending| pending.result.weak().same_channel(&attempt));
+		}
+	}
+}
+
 impl kio::Task for Fetching {
 	type Output = Result<group::Consumer>;
 
@@ -3063,7 +3083,7 @@ impl kio::Task for Fetching {
 
 		// A written rejection fails every joined fetch. The channel closing without
 		// one means the attempt was dropped unserved (its handlers went away).
-		let err = match result.poll(waiter, |outcome| match &outcome.rejected {
+		let err = match result.outcome.poll(waiter, |outcome| match &outcome.rejected {
 			Some(err) => Poll::Ready(err.clone()),
 			None => Poll::Pending,
 		}) {
@@ -7703,9 +7723,9 @@ mod test {
 		let second = consumer.fetch_group(5, None);
 
 		drop(first);
-		assert!(req.poll_unused(&kio::Waiter::noop()).is_pending());
+		assert!(req.demand().poll_unused(&kio::Waiter::noop()).is_pending());
 		drop(second);
-		assert!(req.poll_unused(&kio::Waiter::noop()).is_ready());
+		assert!(req.demand().poll_unused(&kio::Waiter::noop()).is_ready());
 
 		let retry = consumer.fetch_group(5, None);
 		let fresh = dynamic
@@ -7714,11 +7734,89 @@ mod test {
 			.expect("the retry queues a fresh request")
 			.unwrap();
 		drop(req);
-		assert!(fresh.poll_unused(&kio::Waiter::noop()).is_pending());
+		assert!(fresh.demand().poll_unused(&kio::Waiter::noop()).is_pending());
+		fresh.accept(None).unwrap().finish().unwrap();
+		assert_eq!(retry.await.unwrap().sequence, 5);
+	}
+
+	/// A fetch abandoned before any handler took it is withdrawn from the queue too, so
+	/// no handler serves it.
+	#[tokio::test]
+	async fn fetch_abandoned_while_queued_never_reaches_the_handler() {
+		let producer = track_producer("test", None);
+		let dynamic = producer.dynamic();
+		let consumer = producer.consume();
+
+		drop(consumer.fetch_group(5, None));
+		assert!(dynamic.requested_group().now_or_never().is_none());
+		let fetch = producer.state.read().fetch.clone();
+		assert!(fetch.read().is_empty());
+	}
+
+	/// The withdrawal happens as the last caller leaves, not when the handler notices: a
+	/// fetch arriving before the handler drops the abandoned request is not failed with it.
+	#[tokio::test]
+	async fn fetch_after_the_last_caller_left_is_not_dropped() {
+		let producer = track_producer("test", None);
+		let dynamic = producer.dynamic();
+		let consumer = producer.consume();
+
+		let first = consumer.fetch_group(5, None);
+		let req = dynamic
+			.requested_group()
+			.now_or_never()
+			.expect("should not block")
+			.unwrap();
+		drop(first);
+
+		let mut retry = consumer.fetch_group(5, None);
+		drop(req);
+		assert!(kio::Task::poll(&mut *retry, &kio::Waiter::noop()).is_pending());
+
+		let fresh = dynamic
+			.requested_group()
+			.now_or_never()
+			.expect("the retry queues a fresh request")
+			.unwrap();
 		fresh.accept(None).unwrap().finish().unwrap();
 		assert_eq!(retry.await.unwrap().sequence, 5);
 	}
 
+	/// A handler that accepts after every caller left still caches the group, so a fetch
+	/// that queued a fresh request meanwhile resolves from it and that request is moot.
+	#[tokio::test]
+	async fn fetch_accept_after_withdrawal_caches_the_group() {
+		let producer = track_producer("test", None);
+		let dynamic = producer.dynamic();
+		let consumer = producer.consume();
+
+		let first = consumer.fetch_group(5, None);
+		let req = dynamic
+			.requested_group()
+			.now_or_never()
+			.expect("should not block")
+			.unwrap();
+		drop(first);
+
+		let retry = consumer.fetch_group(5, None);
+		let fresh = dynamic
+			.requested_group()
+			.now_or_never()
+			.expect("the retry queues a fresh request")
+			.unwrap();
+
+		let mut group = req.accept(None).expect("a withdrawn request still accepts");
+		group
+			.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"hi"))
+			.unwrap();
+		group.finish().unwrap();
+
+		let mut fetched = retry.await.unwrap();
+		assert_eq!(&fetched.read_frame().await.unwrap().unwrap().payload[..], b"hi");
+		assert!(!fresh.demand().is_used(), "the retry left the fresh request");
+		assert!(matches!(fresh.accept(None), Err(Error::Duplicate)));
+	}
+
 	/// Dropping an auto trait from a published type is a semver break, so the group
 	/// consumer a cached fetch holds must not cost `Fetching` its unwind safety.
 	#[test]
diff --git a/rs/moq-net/tests/fetch_cancel.rs b/rs/moq-net/tests/fetch_cancel.rs
index 53ab138626..9b2dcb3c63 100644
--- a/rs/moq-net/tests/fetch_cancel.rs
+++ b/rs/moq-net/tests/fetch_cancel.rs
@@ -68,9 +68,13 @@ async fn abandoned_fetch_reaches_the_publisher(version: &str, relays: u64, stage
 	match stage {
 		Stage::Unanswered => {
 			drop(waiting);
-			tokio::time::timeout(Duration::from_secs(5), kio::wait(|waiter| request.poll_unused(waiter)))
-				.await
-				.unwrap_or_else(|_| panic!("{ctx}: the publisher's fetch is still wanted"));
+			tokio::time::timeout(
+				Duration::from_secs(5),
+				kio::wait(|waiter| request.demand().poll_unused(waiter)),
+			)
+			.await
+			.unwrap_or_else(|_| panic!("{ctx}: the publisher's fetch is still wanted"))
+			.expect("the pending fetch is still open");
 		}
 		Stage::MidResponse => {
 			let mut group = request.accept(None).unwrap();
@@ -82,7 +86,7 @@ async fn abandoned_fetch_reaches_the_publisher(version: &str, relays: u64, stage
 			drop(fetched);
 
 			// The publisher stops serving the group: nobody reads it any more.
-			tokio::time::timeout(Duration::from_secs(5), group.unused())
+			tokio::time::timeout(Duration::from_secs(5), group.demand().unused())
 				.await
 				.unwrap_or_else(|_| panic!("{ctx}: the publisher still serves the fetch"))
 				.unwrap();
diff --git a/rs/moq-net/tests/loom.rs b/rs/moq-net/tests/loom.rs
index 4731533d4c..f124041860 100644
--- a/rs/moq-net/tests/loom.rs
+++ b/rs/moq-net/tests/loom.rs
@@ -292,3 +292,41 @@ fn publisher_drop_resolves_a_parked_subscriber() {
 		publisher.join().unwrap();
 	});
 }
+
+/// The last `fetch_group` caller leaving, racing a new fetch of the same group while
+/// the handler drops the request the moment it sees it unused.
+///
+/// The leaver withdraws the attempt under the fetch lock a join takes, so the new
+/// fetch either joined the live attempt (keeping its request wanted) or queued a fresh
+/// one. A fetch joining the withdrawn attempt and failing as its handler drops it is
+/// the bug, and shows up as `Err` here.
+#[test]
+fn a_fetch_never_joins_a_withdrawn_attempt() {
+	model(|| {
+		let broadcast = broadcast::Info::new().produce();
+		let track = broadcast.create_track("video", None).expect("create track");
+		let dynamic = track.dynamic();
+		let consumer = track.consume();
+
+		let first = consumer.fetch_group(5, None);
+		let request = block_on(dynamic.requested_group()).expect("requested");
+
+		let leaver = thread::spawn(move || {
+			drop(first);
+			request.demand().is_used().then_some(request)
+		});
+		let mut retry = consumer.fetch_group(5, None);
+		let request = match leaver.join().unwrap() {
+			Some(request) => request,
+			None => {
+				let polled = kio::Task::poll(&mut *retry, &kio::Waiter::noop());
+				if let std::task::Poll::Ready(Err(err)) = polled {
+					panic!("the retry joined a withdrawn attempt: {err}");
+				}
+				block_on(dynamic.requested_group()).expect("the retry queued a fresh request")
+			}
+		};
+		request.accept(None).expect("accept").finish().expect("finish");
+		block_on(retry).expect("the retry resolves");
+	});
+}

From c3ed73ac6656924eff46b7f57b6f5406cf901c70 Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Mon, 5 Oct 2026 11:19:00 -0700
Subject: [PATCH 4/5] perf(net): scan the request queue only for a still-queued
 entry

Requests tracks whether each entry is still queued, so removing a popped
entry stays O(1). A leaving fetch reads before writing, so it no longer
wakes parked handlers when its attempt is already gone. Document that a
withdrawn fetch request's demand never returns, and the Duplicate a
handler ignoring demand may now see.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 doc/setup/upgrade.md             |  4 +++-
 rs/moq-net/src/model/requests.rs | 37 +++++++++++++++++++++-----------
 rs/moq-net/src/model/track.rs    | 10 ++++++---
 3 files changed, 35 insertions(+), 16 deletions(-)

diff --git a/doc/setup/upgrade.md b/doc/setup/upgrade.md
index 1bf57db3c9..bd37fc3c20 100644
--- a/doc/setup/upgrade.md
+++ b/doc/setup/upgrade.md
@@ -39,7 +39,9 @@ These land with the next breaking release, not the 2026-09-23 train.
   `Poll<Result<()>>` instead of `Poll<()>`: an `Err` means the request closed,
   so treat it as unused too. A `group::Request` no longer needs polling to be
   withdrawn; the last `fetch_group` caller leaving does it, so a handler that
-  sees it unused just drops it.
+  sees it unused just drops it. A fetch arriving after that queues a fresh
+  request instead of joining the abandoned one, so a handler that keeps serving
+  without watching `demand()` may see its `accept` return `Error::Duplicate`.
   The moq-json snapshot and moq-flate `is_used()` is `demand().is_used()`.
   In TypeScript, `Track.Producer`'s `used` and `unused()` are
   `producer.demand().used` and `.unused()`, as are `Group.Producer`'s, and
diff --git a/rs/moq-net/src/model/requests.rs b/rs/moq-net/src/model/requests.rs
index 7b915b5c65..7f479389de 100644
--- a/rs/moq-net/src/model/requests.rs
+++ b/rs/moq-net/src/model/requests.rs
@@ -22,13 +22,20 @@ use std::{
 /// [`Self::drain_queued`] when the last handler leaves.
 pub(crate) struct Requests<K, V> {
 	// Every pending request: queued or already handed to a handler.
-	pending: HashMap<K, V>,
+	pending: HashMap<K, Entry<V>>,
 	// The queued (not yet popped) subset, FIFO. Every key here is in `pending`.
 	order: VecDeque<K>,
 	// Live handler count; gates `insert`.
 	handlers: usize,
 }
 
+/// A pending request, and whether its key still waits in `order`.
+struct Entry<V> {
+	value: V,
+	// Cleared by `pop`, so removing a popped entry skips scanning the queue.
+	queued: bool,
+}
+
 impl<K, V> Default for Requests<K, V> {
 	fn default() -> Self {
 		Self {
@@ -46,7 +53,7 @@ impl<K: Clone + Eq + Hash, V> Requests<K, V> {
 		K: Borrow<Q>,
 		Q: Eq + Hash + ?Sized,
 	{
-		self.pending.get_mut(key)
+		self.pending.get_mut(key).map(|entry| &mut entry.value)
 	}
 
 	/// Queue a new request, handing `value` back when no handler is alive to
@@ -58,7 +65,7 @@ impl<K: Clone + Eq + Hash, V> Requests<K, V> {
 		if self.handlers == 0 {
 			return Err(value);
 		}
-		let prev = self.pending.insert(key.clone(), value);
+		let prev = self.pending.insert(key.clone(), Entry { value, queued: true });
 		debug_assert!(prev.is_none(), "insert over a pending request; join it instead");
 		self.order.push_back(key);
 		Ok(())
@@ -67,7 +74,11 @@ impl<K: Clone + Eq + Hash, V> Requests<K, V> {
 	/// Pop the next queued key for a handler to serve. The entry stays pending
 	/// (joinable) until removed.
 	pub fn pop(&mut self) -> Option<K> {
-		self.order.pop_front()
+		let key = self.order.pop_front()?;
+		if let Some(entry) = self.pending.get_mut(&key) {
+			entry.queued = false;
+		}
+		Some(key)
 	}
 
 	/// The pending request for `key`, if any.
@@ -76,7 +87,7 @@ impl<K: Clone + Eq + Hash, V> Requests<K, V> {
 		K: Borrow<Q>,
 		Q: Eq + Hash + ?Sized,
 	{
-		self.pending.get(key)
+		self.pending.get(key).map(|entry| &entry.value)
 	}
 
 	/// Remove and return the pending request for `key`, if any.
@@ -85,7 +96,7 @@ impl<K: Clone + Eq + Hash, V> Requests<K, V> {
 		K: Borrow<Q>,
 		Q: Eq + Hash + ?Sized,
 	{
-		self.pending.remove(key)
+		self.pending.remove(key).map(|entry| entry.value)
 	}
 
 	/// Remove and return the pending request for `key`, purging its queue entry
@@ -96,9 +107,11 @@ impl<K: Clone + Eq + Hash, V> Requests<K, V> {
 		K: Borrow<Q>,
 		Q: Eq + Hash + ?Sized,
 	{
-		let value = self.pending.remove(key)?;
-		self.order.retain(|k| k.borrow() != key);
-		Some(value)
+		let entry = self.pending.remove(key)?;
+		if entry.queued {
+			self.order.retain(|k| k.borrow() != key);
+		}
+		Some(entry.value)
 	}
 
 	/// Remove the pending request for `key` if `f` says it's the caller's own,
@@ -109,7 +122,7 @@ impl<K: Clone + Eq + Hash, V> Requests<K, V> {
 		K: Borrow<Q>,
 		Q: Eq + Hash + ?Sized,
 	{
-		if !self.pending.get(key).is_some_and(f) {
+		if !self.get(key).is_some_and(f) {
 			return None;
 		}
 		self.take(key)
@@ -147,7 +160,7 @@ impl<K: Clone + Eq + Hash, V> Requests<K, V> {
 	pub fn drain_queued(&mut self) -> Vec<V> {
 		self.order
 			.drain(..)
-			.filter_map(|key| self.pending.remove(&key))
+			.filter_map(|key| self.pending.remove(&key).map(|entry| entry.value))
 			.collect()
 	}
 
@@ -155,7 +168,7 @@ impl<K: Clone + Eq + Hash, V> Requests<K, V> {
 	/// handler, so the caller can reject them all on a terminal teardown.
 	pub fn drain_all(&mut self) -> Vec<V> {
 		self.order.clear();
-		self.pending.drain().map(|(_, value)| value).collect()
+		self.pending.drain().map(|(_, entry)| entry.value).collect()
 	}
 }
 
diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs
index 79718d16f3..ef588d70f3 100644
--- a/rs/moq-net/src/model/track.rs
+++ b/rs/moq-net/src/model/track.rs
@@ -2861,7 +2861,8 @@ impl group::Request {
 	/// Watch the callers waiting for this fetch without keeping the attempt alive.
 	///
 	/// The last caller to leave withdraws the attempt, so a later fetch of the group
-	/// queues a fresh request rather than joining this one: once unused, drop it.
+	/// queues a fresh request rather than joining this one: once unused, demand never
+	/// returns, so drop the request.
 	pub fn demand(&self) -> group::Demand {
 		group::Demand::fetch(self.sequence, self.result.weak())
 	}
@@ -3014,8 +3015,11 @@ impl Drop for Fetching {
 		// rather than joining one its handler is about to drop.
 		let mut fetch = self.fetch.lock();
 		drop(outcome);
-		if !attempt.is_used() {
-			fetch.remove_if(&self.sequence, |pending| pending.result.weak().same_channel(&attempt));
+		// Read before writing: the attempt is usually already resolved and gone, and a
+		// write would wake every handler parked on the queue for nothing.
+		let ours = |pending: &PendingFetch| pending.result.weak().same_channel(&attempt);
+		if !attempt.is_used() && fetch.get(&self.sequence).is_some_and(ours) {
+			fetch.remove_if(&self.sequence, ours);
 		}
 	}
 }

From c2b7e3475ef6ce5a29f4290d41db637519733d3f Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Mon, 5 Oct 2026 11:51:34 -0700
Subject: [PATCH 5/5] refactor(net): share the read-first fetch withdrawal

group::Request::remove reads before it writes too, so finishing an attempt the last caller already withdrew wakes no parked handler. Requests::remove asserts its entry was popped.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 rs/moq-net/src/model/requests.rs | 18 +++++++++++++++++-
 rs/moq-net/src/model/track.rs    | 23 +++++++++++++++--------
 2 files changed, 32 insertions(+), 9 deletions(-)

diff --git a/rs/moq-net/src/model/requests.rs b/rs/moq-net/src/model/requests.rs
index 7f479389de..c33d60fc10 100644
--- a/rs/moq-net/src/model/requests.rs
+++ b/rs/moq-net/src/model/requests.rs
@@ -96,7 +96,9 @@ impl<K: Clone + Eq + Hash, V> Requests<K, V> {
 		K: Borrow<Q>,
 		Q: Eq + Hash + ?Sized,
 	{
-		self.pending.remove(key).map(|entry| entry.value)
+		let entry = self.pending.remove(key)?;
+		debug_assert!(!entry.queued, "remove of a queued request; take it instead");
+		Some(entry.value)
 	}
 
 	/// Remove and return the pending request for `key`, purging its queue entry
@@ -220,6 +222,20 @@ mod test {
 		assert!(!requests.has_queued());
 	}
 
+	#[test]
+	fn take_after_pop_keeps_the_queue_order() {
+		let mut requests = Requests::<u64, &str>::default();
+		requests.add_handler();
+		for (key, value) in [(1, "a"), (2, "b"), (3, "c")] {
+			assert!(requests.insert(key, value).is_ok());
+		}
+		assert_eq!(requests.pop(), Some(1));
+		assert_eq!(requests.take(&1), Some("a"));
+		assert_eq!(requests.take(&2), Some("b"));
+		assert_eq!(requests.pop(), Some(3));
+		assert_eq!(requests.pop(), None);
+	}
+
 	#[test]
 	fn drain_queued_spares_popped() {
 		let mut requests = Requests::<u64, &str>::default();
diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs
index ef588d70f3..95d0acae27 100644
--- a/rs/moq-net/src/model/track.rs
+++ b/rs/moq-net/src/model/track.rs
@@ -2942,9 +2942,17 @@ impl group::Request {
 	/// Remove this attempt from the fetch state, unless a newer attempt for the same
 	/// sequence has already replaced it.
 	fn remove(&self) {
-		self.fetch
-			.lock()
-			.remove_if(&self.sequence, |pending| pending.result.same_channel(&self.result));
+		withdraw(&mut self.fetch.lock(), self.sequence, |pending| {
+			pending.result.same_channel(&self.result)
+		});
+	}
+}
+
+/// Remove the attempt for `sequence` if it is still `ours`. Reads first: the attempt is
+/// often already gone, and a write would wake every handler parked on the queue.
+fn withdraw(fetch: &mut kio::Mut<'_, FetchState>, sequence: u64, ours: impl Fn(&PendingFetch) -> bool + Copy) {
+	if fetch.get(&sequence).is_some_and(ours) {
+		fetch.remove_if(&sequence, ours);
 	}
 }
 
@@ -3015,11 +3023,10 @@ impl Drop for Fetching {
 		// rather than joining one its handler is about to drop.
 		let mut fetch = self.fetch.lock();
 		drop(outcome);
-		// Read before writing: the attempt is usually already resolved and gone, and a
-		// write would wake every handler parked on the queue for nothing.
-		let ours = |pending: &PendingFetch| pending.result.weak().same_channel(&attempt);
-		if !attempt.is_used() && fetch.get(&self.sequence).is_some_and(ours) {
-			fetch.remove_if(&self.sequence, ours);
+		if !attempt.is_used() {
+			withdraw(&mut fetch, self.sequence, |pending| {
+				pending.result.weak().same_channel(&attempt)
+			});
 		}
 	}
 }