diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index a2292686fb..4de86de0c2 100644 --- a/doc/concept/moq-lite.md +++ b/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 diff --git a/doc/concept/standard.md b/doc/concept/standard.md index 9b4730a795..8ef0005764 100644 --- a/doc/concept/standard.md +++ b/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 diff --git a/doc/lib/c/index.md b/doc/lib/c/index.md index f49b96850e..9777accff1 100644 --- a/doc/lib/c/index.md +++ b/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. diff --git a/js/net/src/ietf/subscriber.test.ts b/js/net/src/ietf/subscriber.test.ts index b6d33f83f6..4c062396c2 100644 --- a/js/net/src/ietf/subscriber.test.ts +++ b/js/net/src/ietf/subscriber.test.ts @@ -1,4 +1,4 @@ -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"; @@ -6,6 +6,7 @@ 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[] = []; diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index 45e53461aa..c49c180f0c 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/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; diff --git a/js/net/src/integration.test.ts b/js/net/src/integration.test.ts index 722bcac4d3..3fdf0baf9d 100644 --- a/js/net/src/integration.test.ts +++ b/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 { 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 { + 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(); diff --git a/js/net/src/lite/subscriber.test.ts b/js/net/src/lite/subscriber.test.ts index 7b668639bb..99421e1595 100644 --- a/js/net/src/lite/subscriber.test.ts +++ b/js/net/src/lite/subscriber.test.ts @@ -1,4 +1,4 @@ -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"; @@ -6,6 +6,7 @@ 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 { + for (let i = 0; i < MAX_DRAIN_TURNS; i++) await Promise.resolve(); + jest.advanceTimersByTime(REQUEST_LINGER_MS); +} + async function drainUntil(predicate: () => boolean): Promise { 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); diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index 6f59818192..f7b1b6cb47 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -13,6 +13,7 @@ import { TAIL_GRACE_MS, Tail } from "../tail.ts"; import * as Time from "../time.ts"; import type * as track from "../track.ts"; import { untilAborted } from "../util/abort.ts"; +import { abandoned } from "../util/linger.ts"; import { TimeoutError, withTimeout } from "../util/timeout.ts"; import { overrideBroadcastWire, wireOf } from "../wire.ts"; import { @@ -612,13 +613,13 @@ export class Subscriber { if (subscriptionUpdates !== undefined) terminal.push(subscriptionUpdates.then(() => closed)); const done = race(terminal); - // 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 (e.g. a quickly unmuted tile) resumes on the same subscription. + // 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 (e.g. a quickly unmuted tile, a seek) resumes on the same subscription. const idle = Symbol("idle"); 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; break; } @@ -860,16 +861,13 @@ export class Subscriber { try { const decode = frameDecoder(timescale); - // Serve until the stream FINs, the group closes, or every reader leaves. A group can - // stay open indefinitely (a catalog or JSON stream), so an abandoned fetch is stopped by - // demand, not by the stream ending. `unused` is watched across frames as one stable - // promise; the check is level-triggered, so a coalesced fetch that arrives before we - // cancel re-arms and resumes. + // Serve until the stream FINs, the group closes, or every reader stays gone through the + // linger. A group can stay open indefinitely (a catalog or JSON stream), so an abandoned + // fetch is stopped by demand, not by the stream ending. `unused` is watched across frames + // as one stable 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 - .demand() - .unused() - .then((): typeof idle => idle); + let unused = abandoned(group.demand()).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 | undefined; @@ -882,10 +880,7 @@ export class Subscriber { if (next === idle) { if (group.isClosed) break; if (group.demand().used.peek()) { - unused = group - .demand() - .unused() - .then((): typeof idle => idle); + unused = abandoned(group.demand()).then((): typeof idle => idle); continue; } // Abandoned mid-group: the truncated group must never end clean. @@ -1272,21 +1267,12 @@ async function untilClosed(group: netGroup.Producer, step: Promise): Promi return value as T; } -// Like untilClosed, but also cancels once every reader has left. Demand is level-triggered, so a -// caller that coalesces onto the group before the check re-arms it. +// Like untilClosed, but also cancels once every reader stays gone through the linger. Demand is +// level-triggered, so a caller that coalesces onto the group before the check re-arms it. async function untilAbandoned(group: netGroup.Producer, step: Promise): Promise { const idle: unique symbol = Symbol("idle"); for (;;) { - const value = await untilClosed( - group, - race([ - step, - group - .demand() - .unused() - .then((): typeof idle => idle), - ]), - ); + const value = await untilClosed(group, race([step, abandoned(group.demand()).then((): typeof idle => idle)])); if (value !== idle) return value as T; if (!group.demand().used.peek()) { // Close here rather than where the error lands, so no fetch coalesces onto the group diff --git a/js/net/src/util/linger.test.ts b/js/net/src/util/linger.test.ts new file mode 100644 index 0000000000..2c2fa2e108 --- /dev/null +++ b/js/net/src/util/linger.test.ts @@ -0,0 +1,77 @@ +import { expect, jest, test } from "bun:test"; +import * as Track from "../track.ts"; +import { abandoned, REQUEST_LINGER_MS } from "./linger.ts"; + +function fakeTime(): Disposable { + jest.useFakeTimers(); + return { [Symbol.dispose]: () => jest.useRealTimers() }; +} + +async function settle(): Promise { + for (let i = 0; i < 10; i++) await Promise.resolve(); +} + +function watch(promise: Promise): { done: boolean } { + const state = { done: false }; + void promise.then(() => { + state.done = true; + }); + return state; +} + +test("abandoned: resolves once demand stays unused through the linger", async () => { + using _time = fakeTime(); + const producer = new Track.Producer("video"); + const subscriber = producer.subscribe(); + const watched = watch(abandoned(producer.demand())); + + subscriber.close(); + await settle(); + jest.advanceTimersByTime(REQUEST_LINGER_MS - 1); + await settle(); + expect(watched.done).toBe(false); + + jest.advanceTimersByTime(1); + await settle(); + expect(watched.done).toBe(true); + producer.close(); +}); + +test("abandoned: a returning reader restarts the linger", async () => { + using _time = fakeTime(); + const producer = new Track.Producer("video"); + const first = producer.subscribe(); + const watched = watch(abandoned(producer.demand())); + + first.close(); + await settle(); + jest.advanceTimersByTime(REQUEST_LINGER_MS / 2); + const second = producer.subscribe(); + await settle(); + jest.advanceTimersByTime(REQUEST_LINGER_MS); + await settle(); + expect(watched.done).toBe(false); + + // The window counts from the second reader leaving. + second.close(); + await settle(); + jest.advanceTimersByTime(REQUEST_LINGER_MS - 1); + await settle(); + expect(watched.done).toBe(false); + jest.advanceTimersByTime(1); + await settle(); + expect(watched.done).toBe(true); + producer.close(); +}); + +test("abandoned: resolves at once when the track closes", async () => { + using _time = fakeTime(); + const producer = new Track.Producer("video"); + const subscriber = producer.subscribe(); + const watched = watch(abandoned(producer.demand())); + + producer.close(); + subscriber.close(); + await settle(); + expect(watched.done).toBe(true); +}); diff --git a/js/net/src/util/linger.ts b/js/net/src/util/linger.ts new file mode 100644 index 0000000000..91b7803554 --- /dev/null +++ b/js/net/src/util/linger.ts @@ -0,0 +1,39 @@ +import { type GetPromise, type Getter, race, Signal } from "@moq/signals"; + +/** + * How long an upstream SUBSCRIBE or FETCH outlives the last reader wanting it, mirroring Rust's + * `track::REQUEST_LINGER`. A reader that re-subscribes, seeks, or blips is back within a round + * trip, and rides the request still in flight instead of a cancel and a fresh request. + */ +export const REQUEST_LINGER_MS = 1000; + +/** The demand a request is held for: a track's subscribers or a group's readers. */ +export interface Demand { + readonly used: Getter; + readonly closed: GetPromise; + unused(): Promise; +} + +/** + * Resolves once `demand` stays unused through {@link REQUEST_LINGER_MS}, or closes. A reader + * returning restarts the linger for the next time it leaves. + */ +export async function abandoned(demand: Demand): Promise { + const lapsed = Symbol("lapsed"); + for (;;) { + await demand.unused(); + if (demand.closed.peek() !== undefined) return; + + let timer: ReturnType | undefined; + const expired = new Promise((resolve) => { + timer = setTimeout(() => resolve(lapsed), REQUEST_LINGER_MS); + }); + try { + while (!demand.used.peek() && demand.closed.peek() === undefined) { + if ((await race([expired, Signal.race(demand.used, demand.closed)])) === lapsed) return; + } + } finally { + clearTimeout(timer); + } + } +} diff --git a/quest/m1/README.md b/quest/m1/README.md index 849bcbd842..f5ff2b196a 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -40,7 +40,6 @@ blocks. The quests that gated m0 lines moved under them. - [Cluster routing](/quest/m1/cluster-routing/README.md) - any node routes toward a broadcast's origin over CDN and P2P links alike, with per-origin routes and path-less announces - [One route cost](/quest/m1/route-cost.md) - Warm and Cold collapse to one static route cost - [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` -- [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 - [Flat questlines](/quest/m1/quest-flat-lines.md) - moq pins the current quest CLI, lands its questline branches on main, and retires them - [moq-bot may push workflow changes](/quest/m1/bot-workflows-permission.md) - condition: the maintainer grants moq-bot's GitHub App the `workflows` permission, so back-merges carrying workflow changes go through - [Binding audio delay](/quest/m1/binding-surface.md) - moq-ffi and every wrapper configure and observe audio playout delay diff --git a/quest/m1/request-linger.md b/quest/m1/request-linger.md deleted file mode 100644 index 3d89ba3276..0000000000 --- a/quest/m1/request-linger.md +++ /dev/null @@ -1,27 +0,0 @@ -# [M] Upstream requests linger past their demand - -## Goal - -An upstream FETCH or SUBSCRIBE outlives the last reader wanting it by a short -linger, so a request issued and abandoned in quick succession (a reader that -re-subscribes, seeks, or blips) does not churn upstream. A reader returning -within the linger rides the request already in flight. - -## Plan - -Decided in planning (2026-10-03, from [#4741](https://github.com/moq-dev/moq/pull/4741)): - -- **Both request types.** The session's upstream SUBSCRIBE, today canceled the - moment the last subscriber leaves, and FETCH, today cut short (all the way - upstream) once no fetcher waits and no reader holds its group. -- **Demand decides, per request type, unsplit.** A FETCH is canceled once its - demand stays unused through the linger. A subscription's groups ignore group - demand: a front's reader still gives up a stale group a fetcher happens to - hold, as a relay's subscription would. Group demand is not split into fetch - and subscription halves. -- **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. 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/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index e972304807..3aea80c390 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -1922,6 +1922,8 @@ where let setup = { let mut response = std::pin::pin!(self.read_subscribe_response(&mut stream)); + // A subscriber returning within the linger rides the SUBSCRIBE in flight. + let mut cancel = crate::time::Linger::new(&self.runtime, track::REQUEST_LINGER); loop { let setup = kio::wait(|waiter| { // An answer that has already arrived wins over the local terminals. Both can @@ -1937,10 +1939,11 @@ where if !state.subscribes.contains_key(&request_id) { return Poll::Ready(Setup::Gone); } - return match idle.demand.poll_unused(waiter) { - Poll::Ready(_) => Poll::Ready(Setup::Unused), - Poll::Pending => Poll::Pending, - }; + let unused = idle.demand.poll_unused(waiter).is_ready(); + if unused { + let _ = idle.demand.poll_used(waiter); + } + return cancel.poll(unused, waiter).map(|()| Setup::Unused); } let Some(pending) = state .subscribes @@ -1949,10 +1952,12 @@ where else { return Poll::Ready(Setup::Gone); }; - if pending.demand().poll_unused(waiter).is_ready() { - return Poll::Ready(Setup::Unused); + let demand = pending.demand(); + let unused = demand.poll_unused(waiter).is_ready(); + if unused { + let _ = demand.poll_used(waiter); } - Poll::Pending + cancel.poll(unused, waiter).map(|()| Setup::Unused) }) .await; @@ -2117,6 +2122,7 @@ where let mut fetch_done = fetching.is_none(); let demand = track.demand(); + let mut cancel = crate::time::Linger::new(&self.runtime, track::REQUEST_LINGER); // Nobody subscribing at all (only fetches asked) needs no subscription. let mut subscribed = track.subscription().is_some(); let idle = { @@ -2143,12 +2149,17 @@ where track.set_live(largest); resuming = None; } - // The last subscriber left: the upstream subscription goes with it, as on - // lite, and the copy lingers for fetches and a returning subscriber. + // The last subscriber left: the upstream subscription goes with it once + // nobody returns within the linger, as on lite, and the copy lingers longer + // for fetches and a returning subscriber. while let Poll::Ready(Ok(subscription)) = track.poll_subscription_changed(waiter) { subscribed = subscription.is_some(); } - if !subscribed || demand.poll_unused(waiter).is_ready() { + let unused = !subscribed || demand.poll_unused(waiter).is_ready(); + if unused { + let _ = demand.poll_used(waiter); + } + if cancel.poll(unused, waiter).is_ready() { return Poll::Ready(End::Idle); } waiter.poll_future(done.as_mut()).map(End::Done) @@ -2261,7 +2272,7 @@ where state.lingering.insert(id, idle.track.clone()); id }; - let mut linger = crate::time::Deadline::new(&self.runtime); + let mut linger = crate::time::Linger::new(&self.runtime, track::IDLE_LINGER); let mut subscribed = idle.track.subscription().is_some(); enum Step { Fetch(group::Request), @@ -2286,18 +2297,11 @@ where } // Nobody holds it: let it go after the linger. A reader waiting on a fetch holds // it too. A holder returning restarts the countdown when it next leaves. - if idle.demand.poll_unused(waiter).is_ready() { - if linger.deadline().is_none() { - linger.set(self.runtime.now().checked_add(track::IDLE_LINGER)); - } - if linger.poll(waiter).is_ready() { - return Poll::Ready(Step::Expired); - } + let unused = idle.demand.poll_unused(waiter).is_ready(); + if unused { let _ = idle.demand.poll_used(waiter); - } else { - linger.set(None); } - Poll::Pending + linger.poll(unused, waiter).map(|()| Step::Expired) }) .await; @@ -2318,10 +2322,8 @@ where self.state.lock().lingering.remove(&id); return None; } - Err(used) => { - idle.track = used; - linger.set(None); - } + // The next poll finds it held and restarts the linger. + Err(used) => idle.track = used, }, Step::Closed => break false, } @@ -3435,7 +3437,7 @@ where self, broadcast: PathOwned, name: String, - request: group::Request, + mut request: group::Request, timescale: Option, ) { let sequence = request.sequence(); @@ -3492,6 +3494,8 @@ where Ok::<_, Error>(()) } .await; + // A caller returning within the linger rides the FETCH in flight. + let mut linger = crate::time::Linger::new(&self.runtime, track::REQUEST_LINGER); let res = match res { Err(err) => Some(Err(err)), Ok(()) => { @@ -3501,7 +3505,7 @@ where if let Poll::Ready(res) = waiter.poll_future(response.as_mut()) { return Poll::Ready(Some(res)); } - request.demand().poll_unused(waiter).map(|_| None) + request.poll_abandon(&mut linger, waiter).map(|()| None) }) .await } @@ -3509,8 +3513,9 @@ where let ok = match res { Some(Ok(ok)) => ok, + // `poll_abandon` already rejected it. None => { - request.reject(Error::Cancel); + drop(request); drop(registered); self.cancel_group_fetch(stream, fetch_id).await; return; @@ -3593,16 +3598,21 @@ where Poll::Pending => {} } } - slot.poll(waiter, |state| match &**state { - GroupFetch::Done => Poll::Ready(()), - _ if joined.poll_unused(waiter).is_ready() - && demand.poll_unused(waiter).is_ready() - && demand.abort_unused(Error::Cancel) => - { + slot.poll(waiter, |state| { + if matches!(&**state, GroupFetch::Done) { + return Poll::Ready(()); + } + let unused = joined.poll_unused(waiter).is_ready() && demand.poll_unused(waiter).is_ready(); + if unused { + let _ = joined.poll_used(waiter); + let _ = demand.demand().poll_used(waiter); + } + // A reader returning in the gap wins the abort; the next poll restarts the linger. + if linger.poll(unused, waiter).is_ready() && demand.abort_unused(Error::Cancel) { abandoned = true; - Poll::Ready(()) + return Poll::Ready(()); } - _ => Poll::Pending, + Poll::Pending }) .map(|_| None) }) @@ -4813,7 +4823,7 @@ mod tests { subscriber.run_subscribe(Path::new("broadcast"), dynamic, request).await; }); - moq_net_sim::timeout(std::time::Duration::from_secs(1), serving) + moq_net_sim::timeout(crate::track::REQUEST_LINGER * 2, serving) .await .expect("run_subscribe did not finish") .unwrap(); @@ -4871,7 +4881,7 @@ mod tests { drop(track); drop(consumer); - moq_net_sim::timeout(std::time::Duration::from_secs(1), serving) + moq_net_sim::timeout(crate::track::REQUEST_LINGER * 2, serving) .await .expect("run_subscribe parked waiting for a response that never came") .unwrap(); @@ -4945,7 +4955,7 @@ mod tests { drop(subscription); drop(track); drop(consumer); - moq_net_sim::timeout(std::time::Duration::from_secs(1), serving) + moq_net_sim::timeout(crate::track::REQUEST_LINGER * 2, serving) .await .expect("run_subscribe parked after its reader left") .unwrap(); @@ -5034,7 +5044,7 @@ mod tests { drop(track); drop(consumer); - moq_net_sim::timeout(std::time::Duration::from_secs(1), serving) + moq_net_sim::timeout(crate::track::REQUEST_LINGER * 2, serving) .await .expect("run_subscribe did not finish") .unwrap(); @@ -8570,6 +8580,11 @@ mod joining_fetch_tests { if stage == FetchStage::Unanswered { drop(fetch); + assert!( + futures::poll!(run.as_mut()).is_pending(), + "{version:?}: the FETCH lingers for a returning reader" + ); + moq_net_sim::advance(crate::track::REQUEST_LINGER).await; assert!( futures::poll!(run.as_mut()).is_ready(), "{version:?}: unanswered FETCH stays alive" @@ -8602,6 +8617,11 @@ mod joining_fetch_tests { }; if stage == FetchStage::Accepted { drop(group); + assert!( + futures::poll!(run.as_mut()).is_pending(), + "{version:?}: the FETCH lingers for a returning reader" + ); + moq_net_sim::advance(crate::track::REQUEST_LINGER).await; assert!( futures::poll!(run.as_mut()).is_ready(), "{version:?}: accepted FETCH stays alive" @@ -8663,6 +8683,11 @@ mod joining_fetch_tests { assert_eq!(cached.read_frame().await.unwrap().unwrap().payload.as_ref(), b"x"); assert!(cached.read_frame().await.unwrap().is_none()); } else { + assert!( + futures::poll!(run.as_mut()).is_pending(), + "{version:?}: the FETCH lingers for a returning reader" + ); + moq_net_sim::advance(crate::track::REQUEST_LINGER).await; assert!( futures::poll!(run.as_mut()).is_ready(), "{version:?}: partial FETCH stays alive" diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index d04e0d9a71..12e0aad615 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -2016,6 +2016,11 @@ mod tests { ); drop(waiting); + assert!( + kio::Task::poll(&mut run, &waiter).is_pending(), + "{stage}: lingers for a returning subscriber" + ); + moq_net_sim::advance(track::REQUEST_LINGER).await; assert!( kio::Task::poll(&mut run, &waiter).is_ready(), "{stage}: an unused track kept waiting on TRACK_INFO" @@ -3454,8 +3459,8 @@ enum ServeEnd { /// Serves one requested track for a relay: owns this session's copy of the /// track (pumped into the origin's logical track), driving the single upstream -/// subscription (opened lazily on the first downstream subscriber, canceled when -/// the last one leaves) concurrently with any number of one-shot fetches. +/// subscription (opened lazily on the first downstream subscriber, canceled once +/// the last one stays gone) concurrently with any number of one-shot fetches. #[derive(Clone)] struct TrackServe { subscriber: Subscriber, @@ -3501,7 +3506,7 @@ impl TrackServe { /// Apply a subscription-demand change: hand back an [`Establish`] to open the /// upstream SUBSCRIBE on the first subscriber, buffer a SUBSCRIBE_UPDATE while - /// live (the caller flushes), or cancel outright when the last one leaves. + /// live (the caller flushes), or [`Begin::Idle`] when the last one leaves. fn begin_subscription( &self, producer: &mut track::Producer, @@ -3569,21 +3574,23 @@ impl TrackServe { } } } - None => { - // Last subscriber left: cancel the upstream subscription outright. An - // idle subscription still streams every group into a cache nobody - // reads, and the upstream counts it as a live viewer of the broadcast. - // A returning subscriber re-establishes from the current demand. - if let Sub::Active(active) = sub { - self.subscriber.remove_subscribe(active.id); - let _ = active.stream.writer.finish(); - tracing::info!(track = %self.name, "subscribe canceled (idle)"); - *sub = Sub::None; - // The copy is still held: what it cached goes stale from here. - producer.set_idle(); - } - Ok(Begin::None) - } + // The serve loop lingers before canceling, for a subscriber returning soon. + None => Ok(Begin::Idle), + } + } + + /// Cancel the upstream subscription once nobody subscribes through the linger. An + /// idle subscription still streams every group into a cache nobody reads, and the + /// upstream counts it as a live viewer of the broadcast. A returning subscriber + /// re-establishes from the current demand. + fn cancel_subscription(&self, producer: &mut track::Producer, sub: &mut Sub) { + if let Sub::Active(active) = sub { + self.subscriber.remove_subscribe(active.id); + let _ = active.stream.writer.finish(); + tracing::info!(track = %self.name, "subscribe canceled (idle)"); + *sub = Sub::None; + // The copy is still held: what it cached goes stale from here. + producer.set_idle(); } } @@ -3693,6 +3700,8 @@ impl TrackServe { std::future::poll_fn(|cx| active.stream.writer.poll_flush(cx)).await?; } } + // As if the linger already ran out. + Begin::Idle => self.cancel_subscription(producer, sub), } Ok(()) } @@ -3708,6 +3717,9 @@ enum Begin { None, /// Open an upstream SUBSCRIBE for the first subscriber. Establish(Establish), + /// The last subscriber left: cancel the upstream subscription, unless one returns + /// within [`track::REQUEST_LINGER`]. + Idle, } /// Buffer a SUBSCRIBE_UPDATE echoing the current params, varying only the end @@ -3846,6 +3858,9 @@ enum TrackRunState { Info { request: Option, info: TrackInfoFetch, + /// Runs while nobody wants the track: a subscriber returning soon rides this + /// request rather than a fresh one. + cancel: crate::time::Linger, }, Serve(ServeLoop), /// Preserve the lite07 completion FIN until the publisher acknowledges it. @@ -3859,6 +3874,7 @@ impl TrackServeRun { TrackRunState::Info { request: Some(request), info: TrackInfoFetch::new(&serve), + cancel: crate::time::Linger::new(&serve.subscriber.runtime, track::REQUEST_LINGER), } } else { // Older wires declare no publisher retention limit. @@ -3875,20 +3891,22 @@ impl kio::Task for TrackServeRun { fn poll(&mut self, waiter: &kio::Waiter) -> Poll<()> { loop { match &mut self.state { - TrackRunState::Info { request, info } => { - // Nobody wants the track anymore (the origin failed over, the reader - // left): stop waiting on a peer that may never answer, which would - // otherwise hold this task, its TRACK stream, and the track for good. - // Dropping the fetch resets the stream. + TrackRunState::Info { request, info, cancel } => { + // Nobody wants the track anymore through the linger (the origin failed + // over, the reader left): stop waiting on a peer that may never answer, + // which would 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.demand().poll_unused(waiter).is_ready() { - if pending.reject_unused(Error::Cancel) { - self.state = TrackRunState::Done; - return Poll::Ready(()); - } - // Demand returned in the gap and won inside `reject_unused`: poll - // again so the unused wait is armed for when it leaves. - continue; + let demand = pending.demand(); + let unused = demand.poll_unused(waiter).is_ready(); + if unused { + let _ = demand.poll_used(waiter); + } + // Demand returning in the gap wins inside `reject_unused`, and already + // woke the `used` wait above to restart the linger. + if cancel.poll(unused, waiter).is_ready() && pending.reject_unused(Error::Cancel) { + self.state = TrackRunState::Done; + return Poll::Ready(()); } let res = ready!(info.poll_fetch(&self.serve, waiter)); let request = request.take().expect("request pending"); @@ -4048,8 +4066,9 @@ impl TrackInfoFetch { /// The serve loop proper: owns this session's copy of the track (pumped into /// the origin's logical track), driving the single upstream subscription -/// (opened lazily on the first downstream subscriber, canceled when the last -/// one leaves) concurrently with any number of one-shot fetches. +/// (opened lazily on the first downstream subscriber, canceled once the last +/// one stays gone through [`track::REQUEST_LINGER`]) concurrently with any number +/// of one-shot fetches. struct ServeLoop { /// This session's copy, accepted with the resolved info. The origin pumps /// it into the logical track; demand from the logical subscribers arrives @@ -4070,9 +4089,14 @@ struct ServeLoop { supports_fetch: bool, timescale: Option, mode: ServeMode, - /// Armed while nobody holds the copy: it stays, cache and all, for a returning + /// Runs while nobody holds the copy: it stays, cache and all, for a returning /// reader until this fires. - linger: crate::time::Deadline, + linger: crate::time::Linger, + /// Set while the upstream subscription has no subscriber left. + unsubscribed: bool, + /// Runs while `unsubscribed`: the upstream subscription stays, for a subscriber + /// returning soon, until this fires. + cancel: crate::time::Linger, } // A state machine's enum is its storage: one transient instance per stream, so the @@ -4119,7 +4143,9 @@ impl ServeLoop { supports_fetch: serve.subscriber.version.has_track_stream(), timescale, mode: ServeMode::Select, - linger: crate::time::Deadline::new(&serve.subscriber.runtime), + linger: crate::time::Linger::new(&serve.subscriber.runtime, track::IDLE_LINGER), + unsubscribed: false, + cancel: crate::time::Linger::new(&serve.subscriber.runtime, track::REQUEST_LINGER), } } @@ -4208,8 +4234,12 @@ impl ServeLoop { self.supports_update, self.timescale, ) { - Ok(Begin::Establish(est)) => self.mode = ServeMode::Establish(est), - Ok(Begin::None) => {} + Ok(Begin::Establish(est)) => { + self.unsubscribed = false; + self.mode = ServeMode::Establish(est); + } + Ok(Begin::None) => self.unsubscribed = false, + Ok(Begin::Idle) => self.unsubscribed = true, // Updating the upstream failed: hand the track back for // another route to resume. Err(err) => return Poll::Ready(ServeEnd::GiveBack(err)), @@ -4220,6 +4250,15 @@ impl ServeLoop { Poll::Pending => {} } + // The last subscriber left. The upstream subscription lingers, so one + // returning soon (a re-subscribe or a seek) rides it rather than a cancel + // and a fresh SUBSCRIBE. Its return is the change polled above. + let unsubscribed = self.unsubscribed && matches!(self.sub, Sub::Active(_)); + if self.cancel.poll(unsubscribed, waiter).is_ready() { + serve.cancel_subscription(&mut self.serving, &mut self.sub); + self.unsubscribed = false; + } + // (2) In-flight fetches; completions just retire. let _ = self.fetches.poll(waiter); @@ -4227,18 +4266,13 @@ impl ServeLoop { // went with the last subscriber, so it lingers, cache and all, for a // reader or fetch that asks again soon, then is dropped. In-flight // fetches keep it alive: work already accepted still gets finished. - if self.fetches.is_empty() && self.demand.poll_unused(waiter).is_ready() { - if self.linger.deadline().is_none() { - let now = serve.subscriber.runtime.now(); - self.linger.set(now.checked_add(track::IDLE_LINGER)); - } - if self.linger.poll(waiter).is_ready() { - return Poll::Ready(ServeEnd::Idle); - } - // A reader returning restarts the countdown when it next leaves. + let unused = self.fetches.is_empty() && self.demand.poll_unused(waiter).is_ready(); + if self.linger.poll(unused, waiter).is_ready() { + return Poll::Ready(ServeEnd::Idle); + } + // A reader returning restarts the countdown when it next leaves. + if unused { let _ = self.demand.poll_used(waiter); - } else { - self.linger.set(None); } // (4) The upstream subscribe stream closed, or carried a START/END/DROP. @@ -4391,6 +4425,9 @@ struct FetchServeRun { timescale: Option, group: u64, state: FetchRunState, + /// Runs while nobody wants the fetch: it stays in flight, for a caller returning + /// soon, until this fires. + linger: crate::time::Linger, } enum FetchRunState { @@ -4422,9 +4459,9 @@ enum FetchRunState { impl FetchRunState { /// The downstream request, while it waits on the publisher's answer. - fn request(&self) -> Option<&group::Request> { + fn request(&mut self) -> Option<&mut group::Request> { match self { - Self::Open { request } | Self::Send { request, .. } | Self::Answer { request, .. } => request.as_ref(), + Self::Open { request } | Self::Send { request, .. } | Self::Answer { request, .. } => request.as_mut(), Self::Ingest { .. } | Self::Done => None, } } @@ -4434,12 +4471,14 @@ impl FetchServeRun { fn new(serve: TrackServe, request: group::Request, timescale: Option) -> Self { let session = serve.subscriber.session.clone(); let group = request.sequence(); + let linger = crate::time::Linger::new(&serve.subscriber.runtime, track::REQUEST_LINGER); Self { serve, session, timescale, group, state: FetchRunState::Open { request: Some(request) }, + linger, } } } @@ -4450,10 +4489,11 @@ impl kio::Task for FetchServeRun { fn poll(&mut self, waiter: &kio::Waiter) -> Poll<()> { let mut cx = waiter.context(); loop { - // 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. + // A fetch nobody waits on through the linger is cancelled upstream, so the + // publisher stops serving it (and a relay there releases its own FETCH). A caller + // returning meanwhile joins it instead. Ingest has its own check. if let Some(request) = self.state.request() - && request.demand().poll_unused(waiter).is_ready() + && request.poll_abandon(&mut self.linger, waiter).is_ready() { tracing::debug!(track = %self.serve.name, group = self.group, "fetch abandoned"); if let FetchRunState::Send { stream, .. } | FetchRunState::Answer { stream, .. } = @@ -4619,10 +4659,17 @@ impl kio::Task for FetchServeRun { } => { let Poll::Ready(res) = ingest.poll(&mut stream.reader, producer, waiter) else { // Still short of its end, so completion always wins. Once nobody - // wants the rest (no joined fetch left to pick it up, no reader), - // cancel upstream and abort the truncated group, never caching it - // as complete. The abort is atomic with a new reader arriving. - if joined.poll_unused(waiter).is_pending() || producer.poll_unused(waiter).is_pending() { + // wants the rest (no joined fetch left to pick it up, no reader) + // through the linger, cancel upstream and abort the truncated group, + // never caching it as complete. The abort is atomic with a new reader + // arriving. + let unused = + joined.poll_unused(waiter).is_ready() && producer.poll_unused(waiter).is_ready(); + if unused { + let _ = joined.poll_used(waiter); + let _ = producer.demand().poll_used(waiter); + } + if self.linger.poll(unused, waiter).is_pending() { return Poll::Pending; } if !producer.abort_unused(Error::Cancel) { diff --git a/rs/moq-net/src/model/requests.rs b/rs/moq-net/src/model/requests.rs index c33d60fc10..99134cb839 100644 --- a/rs/moq-net/src/model/requests.rs +++ b/rs/moq-net/src/model/requests.rs @@ -130,6 +130,15 @@ impl Requests { self.take(key) } + /// Returns `true` if the request for `key` is still queued, not yet popped. + pub fn is_queued(&self, key: &Q) -> bool + where + K: Borrow, + Q: Eq + Hash + ?Sized, + { + self.pending.get(key).is_some_and(|entry| entry.queued) + } + /// Returns `true` if a queued (not yet popped) request exists. pub fn has_queued(&self) -> bool { !self.order.is_empty() diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 899bd96164..0f0801b784 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -74,6 +74,14 @@ pub(super) struct ExpiryScan { /// viewer. pub(crate) const IDLE_LINGER: Duration = Duration::from_secs(30); +/// How long an upstream SUBSCRIBE or FETCH outlives the last reader wanting it. +/// +/// A reader that re-subscribes, seeks, or blips is back within a round trip, and +/// rides the request still in flight instead of churning a cancel and a fresh +/// request upstream. Kept far below [`IDLE_LINGER`]: unlike a lingering copy, a +/// lingering request still pulls media nobody reads. +pub(crate) const REQUEST_LINGER: Duration = Duration::from_secs(1); + /// Publisher-side properties of a track. /// /// These are fixed by the publisher when the track is created and don't change @@ -2949,9 +2957,8 @@ 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, demand never - /// returns, so drop the request. + /// The attempt stays joinable after the last caller leaves, so demand can return + /// until the request is dropped, accepted, or rejected. pub fn demand(&self) -> group::Demand { group::Demand::fetch(self.sequence, self.result.weak()) } @@ -3028,6 +3035,41 @@ impl group::Request { } } + /// Reject only while no caller waits on this fetch. Joins take the same lock, so a + /// caller returning after [`group::Demand::poll_unused`] wins the race and `false` + /// leaves the request to serve it. + pub(crate) fn reject_unused(&mut self, err: Error) -> bool { + let mut fetch = self.fetch.lock(); + if self.result.is_used() { + return false; + } + withdraw(&mut fetch, self.sequence, |pending| { + pending.result.same_channel(&self.result) + }); + drop(fetch); + self.done = true; + if let Ok(mut outcome) = self.result.write() { + outcome.rejected = Some(err); + } + true + } + + /// Poll until no caller has wanted this fetch through `linger`, then reject it with + /// [`Error::Cancel`]. A caller returning meanwhile joins it and restarts the linger. + pub(crate) fn poll_abandon(&mut self, linger: &mut crate::time::Linger, waiter: &kio::Waiter) -> Poll<()> { + let demand = self.demand(); + let unused = demand.poll_unused(waiter).is_ready(); + if unused { + let _ = demand.poll_used(waiter); + } + ready!(linger.poll(unused, waiter)); + // A caller returning in the gap already woke the `used` wait above. + match self.reject_unused(Error::Cancel) { + true => Poll::Ready(()), + false => Poll::Pending, + } + } + /// Remove this attempt from the fetch state, unless a newer attempt for the same /// sequence has already replaced it. fn remove(&self) { @@ -3108,11 +3150,13 @@ impl Drop for Fetching { }; 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. + // either keeps the attempt wanted, or finds it withdrawn and queues a fresh one. + // Only a queued attempt is withdrawn here, having cost nothing yet. One a handler + // took stays joinable: the handler lingers on it for a caller returning soon, and + // gives it up with `group::Request::reject_unused`. let mut fetch = self.fetch.lock(); drop(outcome); - if !attempt.is_used() { + if !attempt.is_used() && fetch.is_queued(&self.sequence) { withdraw(&mut fetch, self.sequence, |pending| { pending.result.weak().same_channel(&attempt) }); @@ -7916,10 +7960,11 @@ mod test { assert!(matches!(pending.await, Err(Error::Dropped))); } - /// A request stays wanted while any joined fetch waits, and once the last leaves it - /// is withdrawn: a later fetch starts a fresh request instead of joining it. + /// A request stays wanted while any joined fetch waits. Once the last leaves, a + /// handler that took it keeps it joinable: a fetch returning before the handler gives + /// up rides it rather than queuing a fresh request. #[moq_net_sim::test] - async fn fetch_request_unused_once_every_fetch_leaves() { + async fn fetch_request_rejoinable_after_every_fetch_leaves() { let producer = track_producer("test", None); let dynamic = producer.dynamic(); let consumer = producer.consume(); @@ -7938,14 +7983,12 @@ mod test { assert!(req.demand().poll_unused(&kio::Waiter::noop()).is_ready()); let retry = consumer.fetch_group(5, None); - let fresh = dynamic - .requested_group() - .now_or_never() - .expect("the retry queues a fresh request") - .unwrap(); - drop(req); - assert!(fresh.demand().poll_unused(&kio::Waiter::noop()).is_pending()); - fresh.accept(None).unwrap().finish().unwrap(); + assert!(req.demand().poll_unused(&kio::Waiter::noop()).is_pending()); + assert!( + dynamic.requested_group().now_or_never().is_none(), + "the retry joined the request in flight" + ); + req.accept(None).unwrap().finish().unwrap(); assert_eq!(retry.await.unwrap().sequence, 5); } @@ -7963,26 +8006,31 @@ mod test { 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. + /// Giving up an unused request is atomic with a caller joining it: a fetch that + /// joined first keeps the request, and one arriving after queues a fresh request + /// rather than reading the rejection. #[moq_net_sim::test] - async fn fetch_after_the_last_caller_left_is_not_dropped() { + async fn fetch_reject_unused_races_a_returning_caller() { let producer = track_producer("test", None); let dynamic = producer.dynamic(); let consumer = producer.consume(); let first = consumer.fetch_group(5, None); - let req = dynamic + let mut req = dynamic .requested_group() .now_or_never() .expect("should not block") .unwrap(); drop(first); + let returned = consumer.fetch_group(5, None); + assert!(!req.reject_unused(Error::Cancel), "the returning caller wants it"); + drop(returned); + assert!(req.reject_unused(Error::Cancel)); + 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() @@ -7992,41 +8040,6 @@ mod test { 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. - #[moq_net_sim::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/src/time.rs b/rs/moq-net/src/time.rs index 2bfcd5fffd..6aa0e1274e 100644 --- a/rs/moq-net/src/time.rs +++ b/rs/moq-net/src/time.rs @@ -261,6 +261,37 @@ impl std::fmt::Debug for Deadline { } } +/// Keeps something nobody wants for a fixed window, for a holder returning soon. +pub(crate) struct Linger { + window: std::time::Duration, + deadline: Deadline, +} + +impl Linger { + /// A linger that lets go once nothing wants it for `window`. + pub(crate) fn new(clock: &Clock, window: std::time::Duration) -> Self { + Self { + window, + deadline: Deadline::new(clock), + } + } + + /// `Ready` once every poll for the whole window found it `unused`. + /// + /// A poll finding it wanted restarts the window for the next time it is not. The caller + /// registers its own wake for demand returning, so that poll happens. + pub(crate) fn poll(&mut self, unused: bool, waiter: &kio::Waiter) -> Poll<()> { + if !unused { + self.deadline.set(None); + return Poll::Pending; + } + if self.deadline.deadline().is_none() { + self.deadline.set(self.deadline.clock.now().checked_add(self.window)); + } + self.deadline.poll(waiter) + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/rs/moq-net/tests/request_linger.rs b/rs/moq-net/tests/request_linger.rs new file mode 100644 index 0000000000..cdd3908518 --- /dev/null +++ b/rs/moq-net/tests/request_linger.rs @@ -0,0 +1,170 @@ +//! An upstream SUBSCRIBE or FETCH outlives its last reader by a short linger. +//! +//! A reader that re-subscribes, seeks, or blips is back within it and rides the request +//! still in flight, so the relay neither cancels upstream nor asks again. + +mod support; + +use std::time::Duration; + +use bytes::Bytes; +use moq_net::{Hop, Timestamp, Version, origin, track}; +use support::harness::peer; + +/// Inside the relay's one second request linger. +const WITHIN: Duration = Duration::from_millis(500); +/// Past it. +const PAST: Duration = Duration::from_secs(2); + +fn produce_origin(hop: u64) -> origin::Producer { + let (producer, driver) = origin::Producer::new(origin::Config::new(Hop::new(hop).unwrap())); + support::harness::spawn(driver); + producer +} + +async fn a_resubscribe_within_the_linger_rides_the_subscription(version: &str) { + let version: Version = version.parse().unwrap(); + let publisher = produce_origin(1); + let relay = produce_origin(2); + let _pair = peer(version, &publisher, &relay).await; + + let broadcast = publisher.create_broadcast("room").unwrap(); + let track = broadcast.create_track("video", None).unwrap(); + broadcast.announce(Default::default()).unwrap(); + moq_net_sim::sleep(Duration::from_secs(1)).await; + + let remote = relay.consume().request_broadcast("room").await.unwrap(); + let mut sub = remote.track("video").unwrap().subscribe(None).await.unwrap(); + let mut group = track.append_group().unwrap(); + group.write_frame(Timestamp::ZERO, Bytes::from_static(b"a")).unwrap(); + group.finish().unwrap(); + sub.recv_group().await.unwrap().expect("the first group"); + + // The reader blips: the relay keeps its upstream subscription meanwhile. + drop(sub); + moq_net_sim::sleep(WITHIN).await; + assert!( + track.demand().is_used(), + "{version}: the relay canceled upstream inside the linger" + ); + + let mut sub = remote.track("video").unwrap().subscribe(None).await.unwrap(); + moq_net_sim::sleep(WITHIN).await; + assert!( + track.demand().is_used(), + "{version}: the returning reader rides the subscription" + ); + let mut group = track.append_group().unwrap(); + group.write_frame(Timestamp::ZERO, Bytes::from_static(b"b")).unwrap(); + group.finish().unwrap(); + // The relay's cached group may come first. + let mut next = moq_net_sim::timeout(PAST, async { + loop { + let group = sub.recv_group().await.unwrap().unwrap(); + if group.sequence == 1 { + return group; + } + } + }) + .await + .unwrap_or_else(|_| panic!("{version}: the returning reader stalled")); + assert_eq!(&next.read_frame().await.unwrap().unwrap().payload[..], b"b"); + + // Gone for good: canceled upstream once the linger runs out, not before. + drop(sub); + let left = moq_net_sim::now(); + moq_net_sim::timeout(PAST * 2, track.demand().unused()) + .await + .unwrap_or_else(|_| panic!("{version}: the relay never canceled upstream")) + .unwrap(); + assert!( + moq_net_sim::now() - left > WITHIN, + "{version}: canceled after {:?}, before the linger ran out", + moq_net_sim::now() - left + ); +} + +#[moq_net_sim::test] +async fn a_resubscribe_within_the_linger_rides_the_subscription_lite06() { + a_resubscribe_within_the_linger_rides_the_subscription("moq-lite-06").await; +} + +#[moq_net_sim::test] +async fn a_resubscribe_within_the_linger_rides_the_subscription_lite07() { + a_resubscribe_within_the_linger_rides_the_subscription("moq-lite-07-wip").await; +} + +#[moq_net_sim::test] +async fn a_resubscribe_within_the_linger_rides_the_subscription_ietf19() { + a_resubscribe_within_the_linger_rides_the_subscription("moq-transport-19").await; +} + +async fn a_refetch_within_the_linger_rides_the_fetch(version: &str) { + let version: Version = version.parse().unwrap(); + let publisher = produce_origin(1); + let relay = produce_origin(2); + let _pair = peer(version, &publisher, &relay).await; + + let broadcast = publisher.create_broadcast("room").unwrap(); + let mut dynamic = broadcast.dynamic(); + broadcast.announce(Default::default()).unwrap(); + moq_net_sim::sleep(Duration::from_secs(1)).await; + + let consumer = relay.consume().request_broadcast("room").await.unwrap(); + let mut waiting = Box::pin(consumer.track("video").unwrap().fetch_group(0, None)); + let request = match futures::future::select(std::pin::pin!(dynamic.requested_track()), &mut waiting).await { + futures::future::Either::Left((request, _)) => request.expect("the track request reaches the publisher"), + futures::future::Either::Right(_) => panic!("nothing answered the track"), + }; + let groups = request.dynamic(); + let _track = request.accept(track::Info::default()); + let request = match futures::future::select(std::pin::pin!(groups.requested_group()), &mut waiting).await { + futures::future::Either::Left((request, _)) => request.expect("the fetch reaches the publisher"), + futures::future::Either::Right(_) => panic!("nothing answered the fetch"), + }; + + // The reader gives up before the publisher answers, then asks again. + drop(waiting); + moq_net_sim::sleep(WITHIN).await; + assert!( + request.demand().is_used(), + "{version}: the relay canceled the FETCH inside the linger" + ); + let mut refetch = Box::pin(consumer.track("video").unwrap().fetch_group(0, None)); + let again = std::pin::pin!(groups.requested_group()); + let past = std::pin::pin!(moq_net_sim::sleep(PAST)); + let waited = futures::future::select(&mut refetch, past); + match futures::future::select(again, waited).await { + futures::future::Either::Left(_) => panic!("{version}: the relay fetched the group again"), + futures::future::Either::Right((futures::future::Either::Left(_), _)) => { + panic!("{version}: nothing was served yet") + } + futures::future::Either::Right((futures::future::Either::Right(_), _)) => {} + } + + let mut group = request.accept(None).unwrap(); + group + .write_frame(Timestamp::ZERO, Bytes::from_static(b"segment")) + .unwrap(); + group.finish().unwrap(); + let mut fetched = moq_net_sim::timeout(PAST, refetch) + .await + .unwrap_or_else(|_| panic!("{version}: the refetch stalled")) + .expect("the in-flight FETCH serves the returning reader"); + assert_eq!(&fetched.read_frame().await.unwrap().unwrap().payload[..], b"segment"); +} + +#[moq_net_sim::test] +async fn a_refetch_within_the_linger_rides_the_fetch_lite06() { + a_refetch_within_the_linger_rides_the_fetch("moq-lite-06").await; +} + +#[moq_net_sim::test] +async fn a_refetch_within_the_linger_rides_the_fetch_lite07() { + a_refetch_within_the_linger_rides_the_fetch("moq-lite-07-wip").await; +} + +#[moq_net_sim::test] +async fn a_refetch_within_the_linger_rides_the_fetch_ietf19() { + a_refetch_within_the_linger_rides_the_fetch("moq-transport-19").await; +} diff --git a/rs/moq-tokio/tests/broadcast.rs b/rs/moq-tokio/tests/broadcast.rs index 4b28669558..0e8e005dfb 100644 --- a/rs/moq-tokio/tests/broadcast.rs +++ b/rs/moq-tokio/tests/broadcast.rs @@ -1292,11 +1292,10 @@ async fn rejoin_skips_a_stale_warm_cache(version: &str) { #[tracing_test::traced_test] #[tokio::test] async fn broadcast_rejoin_replays_a_current_warm_cache() { - for version in moq_net::Version::names() { - for open in [false, true] { - rejoin_replays_a_current_warm_cache(version, open).await; - } - } + // Concurrently: each round waits out the relay's request linger on the wall clock. + let runs = moq_net::Version::names() + .flat_map(|version| [false, true].map(move |open| rejoin_replays_a_current_warm_cache(version, open))); + futures::future::join_all(runs).await; } async fn rejoin_replays_a_current_warm_cache(version: &str, open: bool) {