feat(net): linger upstream requests past their last reader

doc/concept/moq-lite.md

@@ -117,7 +117,7 @@ competes with remote routes to its path on cost like any other route, winning
 only a tie. Retracting a route (an unannounce, or the peer's `ANNOUNCE_END`)
 stops new requests from resolving through it but leaves subscriptions already
 in flight alone: each track runs to its own end or failure, or until its last
-subscriber leaves, which cancels it upstream. On moq-lite 05 and
+subscriber stays gone for a second, which cancels it upstream. On moq-lite 05 and
 newer, a clean end requires `SUBSCRIBE_END` before the publisher's FIN. A FIN
 without that declaration fails the subscription with `ProtocolViolation`; older
 moq-lite versions use FIN alone. moq-transport requires `PUBLISH_DONE` before FIN.
@@ -310,9 +310,11 @@ budget behind the new route's live edge. A successor group with no timestamp
 leaves the preceding group's reach unbounded until its first frame arrives; if
 it is dropped first, the next group takes its place.
 
-A relay cancels its upstream subscription once nobody subscribes, but keeps its
-copy of the track for 30 seconds after the last reader leaves, so a returning
-reader or the next fetch finds its cache. That copy is not live meanwhile:
+A relay keeps its upstream subscription or fetch for one second after the last
+reader leaves, so a reader that re-subscribes, seeks, or blips rides the request
+still in flight rather than a cancel and a fresh one. It then cancels upstream,
+but keeps its copy of the track for 30 seconds after the last reader leaves, so
+a returning reader or the next fetch finds its cache. That copy is not live meanwhile:
 readers get nothing from its cache until the source answers again, since how
 stale it is cannot be told. lite-07 and moq-transport answer with their largest
 position, and older lite versions are asked from the head of the newest cached

doc/concept/standard.md

@@ -52,8 +52,9 @@ on moq-lite.
 On drafts 14–19, the Rust publisher answers a standalone `FETCH` within one
 group from the cache. A relay fetches a missing group upstream with a `FETCH`
 of that one whole group, and an upstream refusal is the refusal the fetcher
-sees. Once its last reader leaves, the relay cancels the upstream fetch, even
-before `FETCH_OK`, and aborts an incomplete group instead of caching it as whole.
+sees. Once its last reader stays gone for a second, the relay cancels the upstream
+fetch, even before `FETCH_OK`, and aborts an incomplete group instead of caching it
+as whole. A reader returning within that second rides the fetch in flight.
 Drafts 14–16 use `FETCH_CANCEL`; drafts 17–19 stop and reset the request stream.
 A range touching several groups is refused with `NOT_SUPPORTED`, as is
 any `FETCH` on draft-20 and later, which moved the range into

doc/lib/c/index.md

@@ -50,7 +50,7 @@ pkg-config file included.
 - **Encoded video metadata.** `moq_video_init.hint` is a zero-initialized `moq_video_hint` with `has_*` flags for coded dimensions, bitrate (bits per second), frame rate, and latency preference. Hints seed a video codec track's catalog; detected dimensions take precedence.
 - **Client config.** A zeroed `moq_client_config` means the defaults for every knob, which is what lets a new one be appended without disturbing callers. Fields cover protocol (`versions`), TLS (`tls_fingerprints`, `tls_roots`, `tls_cert`/`_key`, `tls_host_name`), transport (`bind`, `connect_timeout_us`, `failover_delay_us`/`resolution_delay_us` for Happy Eyeballs, `websocket_enabled`/`_delay_us`), and tuning (reconnect backoff, `quic_*`). Every duration is in microseconds. A knob whose default isn't zero carries a `has_*` flag, so setting `backoff_timeout_us = 0` needs `has_backoff_timeout = true` to mean "retry forever" rather than "use the default". `moq_client_defaults()` reports what a NULL config dials with.
 - **Server.** `moq_server_listen` binds before it returns (a bad address or certificate fails there) and hands each incoming session to `on_request` as a request handle. Read `moq_session_request_path` and `_query` to route and authenticate, then `moq_session_request_accept` (a session handle, with origins like `moq_session_connect`) or `moq_session_request_reject` with an HTTP-style code (401 and 403 become the protocol's unauthorized close). An accepted session reports `1` once SETUP completes and never reconnects. `moq_server_addr` reports an ephemeral port and `moq_server_fingerprints` the hashes a client pins for a `tls_generate` certificate. `moq_server_close` stops listening; its terminal callback fires once the sockets are released.
-- **Demand.** A watcher on a published track (`moq_publish_track_demand`, `moq_publish_media_demand`, `moq_encode_video_demand`, `moq_encode_audio_demand`) calls `on_demand` with `MOQ_DEMAND_USED` or `MOQ_DEMAND_UNUSED` right away and again on every change, so an encoder on a battery-powered device runs only while someone is watching. The first call is the current state, so a track that went unused before the watcher existed still reports it. `moq_publish_demand_cancel` stops it; the terminal callback still fires. A container has no single demand and is refused. Demand follows the last real subscriber: an origin that served the track drops its source copy on the unused edge, so a cache linger never delays it. Only a relay keeps what it already delivered warm for 30 seconds, and a returning subscriber is served that cache only once the publisher confirms it is still current.
+- **Demand.** A watcher on a published track (`moq_publish_track_demand`, `moq_publish_media_demand`, `moq_encode_video_demand`, `moq_encode_audio_demand`) calls `on_demand` with `MOQ_DEMAND_USED` or `MOQ_DEMAND_UNUSED` right away and again on every change, so an encoder on a battery-powered device runs only while someone is watching. The first call is the current state, so a track that went unused before the watcher existed still reports it. `moq_publish_demand_cancel` stops it; the terminal callback still fires. A container has no single demand and is refused. Demand follows the last real subscriber: an origin that served the track drops its source copy on the unused edge, so a cache linger never delays it. A relay holds its upstream request one second past its last viewer, for one returning soon, and keeps what it already delivered warm for 30 seconds, and a returning subscriber is served that cache only once the publisher confirms it is still current.
 - **Requests.** `moq_publish_dynamic` serves subscriptions to tracks the broadcast never declared: each arrives as a request handle, read its name with `moq_track_request_name`, then `moq_track_request_accept` (a raw track handle), `moq_track_request_video` / `_audio` (the media handle `moq_publish_video` / `_audio` return), or `moq_track_request_abort` with an application code the subscriber sees. Without a live handler an unknown name is refused. `moq_publish_track_dynamic` does the same for fetches of groups a track no longer has cached, delivered as `moq_group_request_*` (`sequence`, `priority`, `frame_start`); `moq_group_request_accept` starts the producer at `frame_start` so written frames keep their group indices. Register it with `moq_track_request_dynamic` before accepting a track that was itself requested by a fetch, so that pending group survives the transition. Both handlers stop with `moq_publish_dynamic_cancel`.
 - **Everything the bindings can do** ([list](/lib/#what-every-binding-can-do)): media publish and consume with the catalog managed for you, raw pixels and PCM with the codec inside (`moq_encode_video`, `moq_encode_audio`, and the `moq_decode_*` mirrors), raw tracks with timestamps and datagrams, JSON and binary data tracks (snapshot or stream, each advertised in the catalog for as long as it lives), group fetch, catalog sections, shared video properties, and stalled hints. The three advertising operations are `moq_origin_create_broadcast` (unannounced producer, invisible to everyone), `moq_publish_announce` / `moq_publish_unannounce` (exact-path advertisement), `moq_publish_close` (ends the broadcast for good and releases its handle), and `moq_origin_dynamic` (a claim over a path prefix and everything beneath it; `""` for everything). A route is a capability, not an inventory. `moq_origin_announced` takes a `moq_announce_config` (NULL or zeroed for everything): a literal prefix and an optional relative pattern filter; each `moq_announce_event` has a `kind` (`MOQ_ANNOUNCE_KIND_START`, `_UPDATE`, `_END`, or `_LIVE` once every route live when the listener started has been delivered, with no prefix or captures); `prefix` stays relative to the origin, while `captures` reports what each wildcard matched when `has_captures` is true. Paths with a `.`-prefixed segment below the prefix are [hidden](/concept/moq-lite#hidden-broadcasts); set `hidden` or name the dot segment in `prefix` to list them.
 

js/net/src/ietf/subscriber.test.ts

@@ -1,11 +1,12 @@
-import { expect, spyOn, test } from "bun:test";
+import { expect, jest, spyOn, test } from "bun:test";
 import type * as announce from "../announced.ts";
 import { ProtocolViolation, StreamCode, Stream as StreamError } from "../error.ts";
 import { type Hop, HopSchema, UNKNOWN_HOP } from "../hop.ts";
 import { createMockTransportPair } from "../mock.ts";
 import * as Path from "../path.ts";
 import { Reader, Stream } from "../stream.ts";
 import type * as track from "../track.ts";
+import { REQUEST_LINGER_MS } from "../util/linger.ts";
 import { ControlStreamAdapter, NativeSession } from "./adapter.ts";
 import type * as Cluster from "./cluster.ts";
 import { Connection } from "./connection.ts";
@@ -769,9 +770,16 @@ test("a legacy cancel reaches the control stream", async () => {
 	const broadcast = subscriber.consume(Path.from("room"));
 	const track = broadcast.track("video").subscribe();
 
-	// Let the SUBSCRIBE reach the control stream before walking away.
+	// Let the SUBSCRIBE reach the control stream before walking away, then run the linger out.
 	await new Promise((resolve) => setTimeout(resolve, 50));
-	track.close();
+	jest.useFakeTimers();
+	try {
+		track.close();
+		for (let i = 0; i < 200; i++) await Promise.resolve();
+		jest.advanceTimersByTime(REQUEST_LINGER_MS);
+	} finally {
+		jest.useRealTimers();
+	}
 
 	// Everything the subscriber actually put on the wire.
 	const seen: bigint[] = [];

js/net/src/ietf/subscriber.ts

@@ -20,6 +20,7 @@ import type { Cursor, Reader, Stream } from "../stream.ts";
 import { Tail } from "../tail.ts";
 import { Milli, type Timescale, Timestamp } from "../time.ts";
 import type * as track from "../track.ts";
+import { abandoned } from "../util/linger.ts";
 import { TimeoutError, withTimeout } from "../util/timeout.ts";
 import { overrideBroadcastWire, wireOf } from "../wire.ts";
 import type { Session } from "./adapter.ts";
@@ -569,7 +570,7 @@ export class Subscriber {
 				await Signal.race(demand.used, demand.closed);
 			}
 			for (;;) {
-				await demand.unused();
+				await abandoned(demand);
 				if (demand.closed.peek() !== undefined || !demand.used.peek()) return null;
 			}
 		};
@@ -653,13 +654,13 @@ export class Subscriber {
 				producer.closed.then(() => localEnded),
 			]);
 
-			// Serve until a terminal condition fires or the last local subscriber leaves. The unused
-			// wake is level-triggered: re-check demand so a subscriber that returns before we tear
-			// down resumes on the same stream.
+			// Serve until a terminal condition fires or the last local subscriber stays gone through
+			// the linger. The wake is level-triggered: re-check demand so a subscriber that returns
+			// before we tear down resumes on the same stream.
 			let terminal = localEnded;
 			const demand = producer.demand();
 			for (;;) {
-				const reason = await race([done, demand.unused().then(() => idle)]);
+				const reason = await race([done, abandoned(demand).then(() => idle)]);
 				if (reason === idle && demand.closed.peek() === undefined && demand.used.peek()) continue;
 				terminal = reason;
 				break;

js/net/src/integration.test.ts

@@ -1,4 +1,4 @@
-import { expect, test } from "bun:test";
+import { expect, jest, test } from "bun:test";
 import type { Getter } from "@moq/signals";
 import type * as Announce from "./announce.ts";
 import { type Consumer as BroadcastConsumer, Producer as BroadcastProducer } from "./broadcast.ts";
@@ -20,6 +20,7 @@ import { Producer as OriginProducer } from "./origin.ts";
 import * as Path from "./path.ts";
 import { Milli, Timescale, Timestamp } from "./time.ts";
 import type { Producer as TrackProducer } from "./track.ts";
+import { REQUEST_LINGER_MS } from "./util/linger.ts";
 import { withTimeout } from "./util/timeout.ts";
 import { wireOf } from "./wire.ts";
 
@@ -1276,6 +1277,21 @@ async function waitUntil(predicate: () => boolean): Promise<void> {
 	throw new Error("condition not met within timeout");
 }
 
+// Drop the last reader with `leave`, then run the request linger out on fake time: upstream still
+// serves inside it, and lets go once it runs out.
+async function leaveForGood(leave: () => void, served: () => boolean): Promise<void> {
+	jest.useFakeTimers();
+	try {
+		leave();
+		for (let i = 0; i < 200; i++) await Promise.resolve();
+		expect(served()).toBe(true);
+		jest.advanceTimersByTime(REQUEST_LINGER_MS);
+	} finally {
+		jest.useRealTimers();
+	}
+	await waitUntil(() => !served());
+}
+
 // Closing the last subscriber to a track tears the wire subscription down, so the publisher stops
 // serving it (the muted-watch-tile case in #2355) instead of sending groups to a reader that left.
 async function runSubscriberTeardown(protocol: string, version?: number) {
@@ -1299,8 +1315,10 @@ async function runSubscriberTeardown(protocol: string, version?: number) {
 
 	// Closing the only subscriber must tear the wire subscription down, so demand drops on the
 	// publisher rather than the relay serving groups to nobody.
-	sub.close();
-	await waitUntil(() => !video.demand().used.peek());
+	await leaveForGood(
+		() => sub.close(),
+		() => video.demand().used.peek(),
+	);
 
 	broadcast.close();
 	remote.close();
@@ -1349,8 +1367,10 @@ test("integration: ietf draft-14 subscriber teardown on last unsubscribe", async
 	await waitUntil(() => served?.demand().used.peek() === true);
 
 	// Closing the subscriber sends Unsubscribe and tears the subscription down, so demand drops.
-	sub.close();
-	await waitUntil(() => served?.demand().used.peek() === false);
+	await leaveForGood(
+		() => sub.close(),
+		() => served?.demand().used.peek() === true,
+	);
 
 	broadcast.close();
 	await serving;
@@ -1384,8 +1404,10 @@ test("integration: lite fetch teardown when the reader abandons an open group",
 
 	// 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.demand().used.peek());
+	await leaveForGood(
+		() => fetched.close(),
+		() => group.demand().used.peek(),
+	);
 
 	group.close();
 	broadcast.close();
@@ -1428,8 +1450,10 @@ test("integration: lite fan-out keeps the upstream until the last subscriber lea
 	expect(video.demand().used.peek()).toBe(true);
 
 	// The last close tears it down.
-	b.close();
-	await waitUntil(() => !video.demand().used.peek());
+	await leaveForGood(
+		() => b.close(),
+		() => video.demand().used.peek(),
+	);
 
 	broadcast.close();
 	remote.close();
@@ -1458,8 +1482,10 @@ test("integration: lite re-subscribe re-opens the upstream after each teardown",
 		expect(await sub.readString()).toBe(`hello-${i}`);
 		await waitUntil(() => video.demand().used.peek());
 
-		sub.close();
-		await waitUntil(() => !video.demand().used.peek());
+		await leaveForGood(
+			() => sub.close(),
+			() => video.demand().used.peek(),
+		);
 	}
 
 	broadcast.close();
@@ -1497,8 +1523,10 @@ test("integration: lite coalesced fetch stays until every reader abandons the op
 	expect(group.demand().used.peek()).toBe(true);
 
 	// The last abandon cancels the FETCH.
-	f2.close();
-	await waitUntil(() => !group.demand().used.peek());
+	await leaveForGood(
+		() => f2.close(),
+		() => group.demand().used.peek(),
+	);
 
 	group.close();
 	broadcast.close();

js/net/src/lite/subscriber.test.ts

@@ -1,11 +1,12 @@
-import { expect, spyOn, test } from "bun:test";
+import { expect, jest, spyOn, test } from "bun:test";
 import { Signal } from "@moq/signals";
 import type { Probe as ProbeStats } from "../connection/stats.ts";
 import { error, fromTransport, reason, StreamCode, StreamError } from "../error.ts";
 import { HopSchema, isAnonymous, MAX_HOPS, Route, UNKNOWN_HOP } from "../hop.ts";
 import * as Path from "../path.ts";
 import { Writer } from "../stream.ts";
 import * as Time from "../time.ts";
+import { REQUEST_LINGER_MS } from "../util/linger.ts";
 import { type AnnounceBroadcast, AnnounceInit, AnnounceOk, encodeAnnounceBroadcast } from "./announce.ts";
 import { Probe } from "./probe.ts";
 import { Subscriber } from "./subscriber.ts";
@@ -420,6 +421,12 @@ async function probeBytes(probes: Probe[], version: Version): Promise<Uint8Array
 const MAX_DRAIN_TURNS = 1000;
 
 /** Yield until `predicate` holds, rather than guessing a fixed number of turns. */
+// Run the request linger out on fake time, once the microtasks that arm it have run.
+async function lingerOut(): Promise<void> {
+	for (let i = 0; i < MAX_DRAIN_TURNS; i++) await Promise.resolve();
+	jest.advanceTimersByTime(REQUEST_LINGER_MS);
+}
+
 async function drainUntil(predicate: () => boolean): Promise<void> {
 	for (let i = 0; i < MAX_DRAIN_TURNS; i++) {
 		if (predicate()) return;
@@ -972,6 +979,8 @@ test.each([
 	["the TRACK_INFO", "track"],
 	["the FETCH", "fetch"],
 ] as const)("the last fetch sharer aborting during %s cancels it", async (_, stage) => {
+	jest.useFakeTimers();
+	using _real = { [Symbol.dispose]: () => jest.useRealTimers() };
 	const { quic, streams } = fakeSession();
 	const subscriber = new Subscriber(quic, Version.DRAFT_05, HopSchema.parse(1n));
 
@@ -992,6 +1001,8 @@ test.each([
 	second.abort(new Error("second"));
 	expect(((await a.catch((err: unknown) => err)) as Error).message).toBe("first");
 	expect(((await b.catch((err: unknown) => err)) as Error).message).toBe("second");
+	// The request lingers for a sharer returning soon, then is let go.
+	await lingerOut();
 
 	if (stage === "track") {
 		// The TRACK_INFO still completes, but no FETCH is sent for the abandoned group.
@@ -1048,6 +1059,8 @@ test("a fetch after the last sharer left is never failed with the cancelled one"
 // 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 () => {
+	jest.useFakeTimers();
+	using _real = { [Symbol.dispose]: () => jest.useRealTimers() };
 	const { quic, streams } = fakeSession();
 	const subscriber = new Subscriber(quic, Version.DRAFT_05, HopSchema.parse(1n));
 
@@ -1062,6 +1075,7 @@ test("the last reader leaving mid-response cancels the fetch", async () => {
 	const group = await fetch;
 	expect(new TextDecoder().decode((await group.readFrame())?.payload)).toBe("head");
 	group.close();
+	await lingerOut();
 
 	const err = fromTransport(await streams[1].aborted) as StreamError;
 	expect(err.code).toBe(StreamCode.Cancel);