diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index 2ec967127dd..acfce7a3f07 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -37,6 +37,16 @@ how a conference room learns who joined, how a player learns a stream came online without polling, and how [relay clusters](/bin/relay/cluster) discover each other. +Once that first reply is read, the reader has **synced**: a broadcast not +announced by then is absent rather than late, so a player can say "offline" +instead of waiting. A relay replies from what it holds now and never waits on +its own upstreams. moq-lite 05 and later, and moq-transport with the +[active-count](/draft/moq-active-count) extension, mark exactly where the first +reply ends. Older versions sync on the peer's answer, or as soon as the +request is sent, as a best effort. Reading an origin fed by sessions, a reader +has synced once every current session's first reply is in, so a reconnect +takes it back until the new reply lands. + An announcement is a **route**: a claim that broadcasts at a path prefix, and every path beneath it, can be served. By convention a publisher announces each broadcast's exact path, so subscribers enumerate broadcasts by listing routes. diff --git a/doc/concept/standard.md b/doc/concept/standard.md index a4f21412a9f..bd120d2b6eb 100644 --- a/doc/concept/standard.md +++ b/doc/concept/standard.md @@ -42,7 +42,7 @@ ignores unknown parameters: [cluster](/draft/moq-cluster) routing hop lists, [hidden](/draft/moq-hidden) to keep `.`-named namespaces out of discovery, [auth](/draft/moq-auth) to tell each peer what it may publish and subscribe to, [active-count](/draft/moq-active-count) to count the `NAMESPACE` messages -before a `SUBSCRIBE_NAMESPACE` is caught up, and +before a `SUBSCRIBE_NAMESPACE` is synced, and [probe](/draft/moq-probe) for bandwidth estimation. Both libraries offer solicit and auth (draft 17 and later) unless a side's `extensions` setting declines one, which connects it as a peer that does not diff --git a/drafts/draft-lcurley-moq-active-count.md b/drafts/draft-lcurley-moq-active-count.md index 3be4d75db7b..cc28ea0c6dd 100644 --- a/drafts/draft-lcurley-moq-active-count.md +++ b/drafts/draft-lcurley-moq-active-count.md @@ -24,7 +24,7 @@ informative: --- abstract This document defines an extension for MoQ Transport {{moqt}} that tells a subscriber how many NAMESPACE messages answering its SUBSCRIBE_NAMESPACE come before the first change. -The subscriber then knows when it has caught up, instead of guessing from a quiet stream. +The subscriber then knows when it has synced, instead of guessing from a quiet stream. --- note_Note_to_Readers @@ -37,7 +37,7 @@ Submit an [issue](https://github.com/moq-dev/moq/issues) or [PR](https://github. {::boilerplate bcp14-tagged} A suffix is **active** on a SUBSCRIBE_NAMESPACE stream from a NAMESPACE message carrying it until a NAMESPACE_DONE message carrying it. -A subscriber is **caught up** on a SUBSCRIBE_NAMESPACE once it has read Count ({{count}}) NAMESPACE messages after REQUEST_OK, immediately on a Count of 0. +A subscriber is **synced** on a SUBSCRIBE_NAMESPACE once it has read Count ({{count}}) NAMESPACE messages after REQUEST_OK, immediately on a Count of 0. # Introduction @@ -87,7 +87,7 @@ An endpoint MUST close the session with a PROTOCOL_VIOLATION if a REQUEST_OK ans # Security Considerations The count reveals no more than the NAMESPACE messages that follow it. -A publisher that promises more than it sends can only delay the subscriber's notion of having caught up, which it could equally do by withholding NAMESPACE messages. +A publisher that promises more than it sends can only delay the subscriber's notion of having synced, which it could equally do by withholding NAMESPACE messages. # IANA Considerations diff --git a/js/net/src/announced.test.ts b/js/net/src/announced.test.ts index 2142abff528..70569c6f12d 100644 --- a/js/net/src/announced.test.ts +++ b/js/net/src/announced.test.ts @@ -128,3 +128,131 @@ test("the local route wins over a blind request on a no-discovery origin", async broadcast.close(); origin.close(); }); + +const tick = () => new Promise((resolve) => setTimeout(resolve, 0)); +const start = (prefix: string): Announce.Event => ({ + prefix: p(prefix), + captures: undefined, + kind: "start", + route: Route.default, +}); + +test("a stream syncs once the reader is past what was queued at the boundary", async () => { + const producer = new Announce.Producer(); + const consumer = producer.consume(); + producer.append(start("a")); + producer.append(start("b")); + expect(consumer.synced.peek()).toBe(false); + + producer.synced(); + // Appended after the boundary, so not owed. + producer.append(start("c")); + expect(consumer.synced.peek()).toBe(false); + + await consumer.next(); + await consumer.next(); + // Still handling b. + expect(consumer.synced.peek()).toBe(false); + expect(await consumer.next()).toMatchObject({ prefix: p("c") }); + expect(consumer.synced.peek()).toBe(true); +}); + +test("an event taken before the boundary is handled before syncing", async () => { + const producer = new Announce.Producer(); + const consumer = producer.consume(); + producer.append(start("a")); + await consumer.next(); + producer.synced(); + expect(consumer.synced.peek()).toBe(false); + + void consumer.next(); + expect(consumer.synced.peek()).toBe(true); + producer.close(); +}); + +test("behind takes sync back, and closing hands it back for good", () => { + const producer = new Announce.Producer(); + const consumer = producer.consume(); + producer.synced(); + expect(consumer.synced.peek()).toBe(true); + producer.behind(); + expect(consumer.synced.peek()).toBe(false); + producer.close(); + expect(consumer.synced.peek()).toBe(true); + producer.behind(); + expect(consumer.synced.peek()).toBe(true); +}); + +test("an origin with no session delivering syncs on its own replay", async () => { + const origin = new OriginProducer(); + const room = publish(origin, p("room")); + const announced = origin.consume().announced(); + expect(await announced.next()).toMatchObject({ prefix: p("room"), kind: "start" }); + void announced.next(); + await tick(); + expect(announced.synced.peek()).toBe(true); + room.close(); + announced.close(); +}); + +test("an origin stream waits on every overlapping initial set in flight", async () => { + const origin = new OriginProducer(); + const wire = wireOf(origin); + const room = wire.snapshot(p("room")); + const other = wire.snapshot(p("other")); + + const scoped = origin.consume().announced(Path.Pattern.subtree(p("room"))); + const all = origin.consume().announced(); + await tick(); + expect(scoped.synced.peek()).toBe(false); + expect(all.synced.peek()).toBe(false); + + other(); + await tick(); + // Still held by the room snapshot; the other one never held the scoped stream. + expect(scoped.synced.peek()).toBe(false); + expect(all.synced.peek()).toBe(false); + + // The initial set lands, and so does its snapshot. + const route = wire.receive(p("room/cam")); + room(); + expect(await scoped.next()).toMatchObject({ prefix: p("room/cam"), kind: "start" }); + void scoped.next(); + await tick(); + expect(scoped.synced.peek()).toBe(true); + + // A new session, as a reconnect makes, takes it back until its own set lands. + const second = wire.snapshot(p("")); + await tick(); + expect(scoped.synced.peek()).toBe(false); + second(); + await tick(); + expect(scoped.synced.peek()).toBe(true); + + route.close(); + scoped.close(); + all.close(); +}); + +test("a snapshot elsewhere never holds a scoped stream", async () => { + const origin = new OriginProducer(); + const other = wireOf(origin).snapshot(p("other")); + const scoped = origin.consume().announced(Path.Pattern.subtree(p("room"))); + await tick(); + expect(scoped.synced.peek()).toBe(true); + other(); + scoped.close(); +}); + +test("a snapshot under a hidden segment holds only a reader that opts in", async () => { + const origin = new OriginProducer(); + const stats = wireOf(origin).snapshot(p(".stats")); + const all = origin.consume().announced(); + const hidden = origin.consume().announced(Path.Pattern.all(), { hidden: true }); + await tick(); + expect(all.synced.peek()).toBe(true); + expect(hidden.synced.peek()).toBe(false); + stats(); + all.close(); + hidden.close(); +}); diff --git a/js/net/src/announced.ts b/js/net/src/announced.ts index 0bda8779352..689a2a26d79 100644 --- a/js/net/src/announced.ts +++ b/js/net/src/announced.ts @@ -3,7 +3,7 @@ * * @module */ -import { type GetPromise, Once, Signal } from "@moq/signals"; +import { type GetPromise, type Getter, Once, Signal } from "@moq/signals"; import type { Route } from "./hop.js"; import type * as Path from "./path.js"; @@ -62,6 +62,18 @@ export interface Options { class AnnounceState { queue = new Signal([]); closed = new Once(); + synced = new Signal(false); + // The queued events still to read before the stream counts as synced, or undefined + // while the initial set is still arriving. + owed: number | undefined; + // A reader holds the last event it took before syncing until it asks for the next + // one, so whatever it does with that event lands before synced flips. + reading = false; +} + +// Synced once nothing is owed and the reader has come back for more. +function settle(state: AnnounceState) { + if (state.owed === 0 && !state.reading) state.synced.set(true); } // Once.set throws on a second settle, and both ends of a stream can close independently. @@ -71,6 +83,10 @@ function closeState(state: AnnounceState, abort?: Error) { state.queue.mutate((queue) => { queue.length = 0; }); + // Nothing more is coming, so nothing waiting on the boundary should hang. + state.owed = 0; + state.reading = false; + state.synced.set(true); } // pending: /quest/m1/public-model/answer.md @@ -103,6 +119,33 @@ export class Producer { }); } + // pending: /quest/m1/public-model/answer.md + /** + * Marks the initial set as delivered: the stream syncs once the reader has taken + * everything queued so far. A no-op until {@link behind} once marked. + * + * @internal + */ + synced() { + const state = this.#state; + if (state.owed !== undefined) return; + state.owed = state.queue.peek().length; + settle(state); + } + + // pending: /quest/m1/public-model/answer.md + /** + * Takes sync back until the next {@link synced}, as when another initial set starts. + * + * @internal + */ + behind() { + const state = this.#state; + if (state.closed.peek() !== undefined) return; + state.owed = undefined; + state.synced.set(false); + } + /** Closes the writer. Idempotent. */ close(abort?: Error) { closeState(this.#state, abort); @@ -137,6 +180,19 @@ export class Consumer { makeConsumer = (state) => new Consumer(state); } + /** + * Whether the stream has delivered its source's initial announcements and the reader has + * taken them, so a prefix not announced by now is absent rather than not yet heard of. + * + * Flips once the reader asks for the event after the last of them, and is true once + * closed. Through a session, the initial set is the peer's view when it answered: exact + * on versions that count it, best-effort on the rest. Reading an origin, it is every + * attached session's initial set, so a reconnect takes it back until the new one lands. + */ + get synced(): Getter { + return this.#state.synced; + } + /** The events as they arrive, until the stream closes. */ async *[Symbol.asyncIterator](): AsyncGenerator { for (;;) { @@ -148,9 +204,16 @@ export class Consumer { /** Returns the next event, or undefined once the stream closes. */ async next(): Promise { + const state = this.#state; + state.reading = false; + settle(state); for (;;) { - const announce = this.#state.queue.peek().shift(); - if (announce) return announce; + const announce = state.queue.peek().shift(); + if (announce) { + if (state.owed) state.owed -= 1; + if (!state.synced.peek()) state.reading = true; + return announce; + } const closed = this.#state.closed.peek(); if (closed instanceof Error) throw closed; diff --git a/js/net/src/connection/accept.ts b/js/net/src/connection/accept.ts index 15ce14da542..4b2edaa8300 100644 --- a/js/net/src/connection/accept.ts +++ b/js/net/src/connection/accept.ts @@ -118,7 +118,7 @@ async function acceptAlpn( version: Ietf.IetfVersion, wiring: SessionProps, ): Promise { - const { control, early, solicit, hidden, cluster, auth } = await exchangeSetup( + const { control, early, solicit, hidden, activeCount, cluster, auth } = await exchangeSetup( transport, version, "moq-lite-js", @@ -134,6 +134,7 @@ async function acceptAlpn( early, solicit, hidden, + activeCount, cluster, auth, // v17+ uses NativeSession which manages its own request IDs; maxRequestId is unused. @@ -171,6 +172,7 @@ async function acceptSetup( params.setBytes(Ietf.SetupOption.Implementation, encoder.encode("moq-lite-js")); Extensions.intoSetup(params, wiring.extensions, version); Ietf.hiddenIntoSetup(params); + Ietf.activeCountIntoSetup(params, version); const server = new Ietf.ServerSetup({ version, parameters: params }); await server.encode(stream.writer, version); @@ -187,6 +189,7 @@ async function acceptSetup( version, solicit: Ietf.solicitFromSetup(client.parameters), hidden: Ietf.hiddenFromSetup(client.parameters), + activeCount: Ietf.activeCountFromSetup(client.parameters, version), }); } @@ -231,6 +234,7 @@ async function acceptNegotiated( params.setBytes(Ietf.SetupOption.Implementation, encoder.encode("moq-lite-js")); Extensions.intoSetup(params, wiring.extensions, setupVersion); Ietf.hiddenIntoSetup(params); + Ietf.activeCountIntoSetup(params, setupVersion); const server = new Ietf.ServerSetup({ version: selectedVersion, parameters: params }); await server.encode(stream.writer, setupVersion); @@ -255,6 +259,7 @@ async function acceptNegotiated( version: selectedVersion as Ietf.IetfVersion, solicit: Ietf.solicitFromSetup(client.parameters), hidden: Ietf.hiddenFromSetup(client.parameters), + activeCount: Ietf.activeCountFromSetup(client.parameters, selectedVersion as Ietf.IetfVersion), }); } else { throw new Error(`unsupported version: ${selectedVersion.toString(16)}`); diff --git a/js/net/src/connection/connect.ts b/js/net/src/connection/connect.ts index 957a0cb5972..5a73415146b 100644 --- a/js/net/src/connection/connect.ts +++ b/js/net/src/connection/connect.ts @@ -313,6 +313,7 @@ async function negotiate(url: URL, session: WebTransport, wiring: SessionProps): params.setBytes(Ietf.SetupOption.Implementation, encoder.encode("moq-lite-js")); Extensions.intoSetup(params, wiring.extensions, setupVersion); Ietf.hiddenIntoSetup(params); + Ietf.activeCountIntoSetup(params, setupVersion); const client = new Ietf.ClientSetup({ versions: @@ -352,6 +353,7 @@ async function negotiate(url: URL, session: WebTransport, wiring: SessionProps): version: server.version as Ietf.IetfVersion, solicit: Ietf.solicitFromSetup(server.parameters), hidden: Ietf.hiddenFromSetup(server.parameters), + activeCount: Ietf.activeCountFromSetup(server.parameters, server.version as Ietf.IetfVersion), }); } else { throw new Error(`unsupported server version: ${server.version.toString()}`); @@ -368,7 +370,7 @@ async function handshakeAlpn( version: Ietf.IetfVersion, wiring: SessionProps, ): Promise { - const { control, early, solicit, hidden, cluster, auth } = await exchangeSetup( + const { control, early, solicit, hidden, activeCount, cluster, auth } = await exchangeSetup( session, version, "moq-lite-js", @@ -384,6 +386,7 @@ async function handshakeAlpn( early, solicit, hidden, + activeCount, cluster, auth, // v17+ uses NativeSession which manages its own request IDs; maxRequestId is unused. diff --git a/js/net/src/connection/forward.test.ts b/js/net/src/connection/forward.test.ts index 1a5bba0fc8a..ed3350c345a 100644 --- a/js/net/src/connection/forward.test.ts +++ b/js/net/src/connection/forward.test.ts @@ -42,6 +42,9 @@ class FakeSession { */ readonly consumed = new Map(); + /** Whether each request is synced as it opens, as one with no initial set to wait for is. */ + syncedAtOnce = false; + closed: Promise; #die!: (reason: Error | null) => void; @@ -55,6 +58,7 @@ class FakeSession { announced(scope: Path.Pattern = Path.Pattern.all(), options?: Announce.Options): Announce.Consumer { const stream = new Announce.Producer(); this.requests.push({ scope: scope.text, hidden: options?.hidden, stream }); + if (this.syncedAtOnce) stream.synced(); return stream.consume(); } @@ -610,3 +614,64 @@ test("one failed request leaves the other routes live until the session ends", a b.close(); origin.close(); }); + +test("an origin watcher syncs with the session's initial set, not before", async () => { + const origin = new OriginProducer(); + const session = new FakeSession(); + forwardAnnounced(session.session, origin); + + const watch = origin.announced(Path.Pattern.subtree(Path.from("room"))); + const events = record(watch); + await settle(); + expect(watch.synced.peek()).toBe(false); + + const stream = session.stream("room/**"); + start(stream, "room/cam"); + await settle(); + // The route is in, but the peer has not said the set is complete. + expect(events).toEqual(["start room/cam"]); + expect(watch.synced.peek()).toBe(false); + + stream.synced(); + await settle(); + expect(watch.synced.peek()).toBe(true); + + watch.close(); + session.die(); + origin.close(); +}); + +test("a session ending before its initial set lands stops holding the watcher", async () => { + const origin = new OriginProducer(); + const session = new FakeSession(); + forwardAnnounced(session.session, origin); + + const watch = origin.announced(); + record(watch); + await settle(); + expect(watch.synced.peek()).toBe(false); + + session.die(); + await settle(); + expect(watch.synced.peek()).toBe(true); + + watch.close(); + origin.close(); +}); + +test("a request synced as it opens releases the watcher", async () => { + const origin = new OriginProducer(); + const session = new FakeSession(); + session.syncedAtOnce = true; + forwardAnnounced(session.session, origin); + + const watch = origin.announced(); + record(watch); + await settle(); + expect(session.requests).toHaveLength(1); + expect(watch.synced.peek()).toBe(true); + + watch.close(); + session.die(); + origin.close(); +}); diff --git a/js/net/src/connection/forward.ts b/js/net/src/connection/forward.ts index d237bd9fe03..932533666da 100644 --- a/js/net/src/connection/forward.ts +++ b/js/net/src/connection/forward.ts @@ -53,8 +53,8 @@ interface Reported { * watches (`announced(scope)` and `request(path, { announced: true })`) that no open request * already covers, and closes {@link LINGER} after nothing at or beneath its prefix is watched. * A narrower watch rides an open broader request, and a narrower request already open stays - * open until the broader one covering it goes, so neither takes the other's routes over: - * nothing tells when a fresh request has caught up. A route both report is inserted once and + * open until the broader one covering it goes, so neither takes the other's routes over + * before it has synced, which not every version can tell. A route both report is inserted once and * driven by its first reporter; if it changes hands anyway (a request failing), it restarts * unless one epoch proves both requests saw one instance. * @@ -164,6 +164,16 @@ export function forwardAnnounced(conn: Established, origin: OriginProducer): voi }; interests.set(prefix, interest); + // The origin's readers under the prefix wait on this request's initial set. Released + // once the session stream has synced, which it reports only after the last of those + // events is reported here. + const snapshot = originWire.snapshot(prefix); + const synced = interest.announced.synced.subscribe((synced) => { + if (synced) snapshot(); + }); + // A request with no initial set to wait for is synced before the listener exists. + if (interest.announced.synced.peek()) snapshot(); + void (async () => { let failure: unknown; try { @@ -176,7 +186,10 @@ export function forwardAnnounced(conn: Established, origin: OriginProducer): voi // The session died mid-stream, or the peer refused or reset the request. failure = err; } finally { + synced(); + snapshot(); for (const reported of [...interest.reported]) unreport(interest, reported); + interest.announced.close(); interest.linger?.(); if (interests.get(prefix) === interest) interests.delete(prefix); @@ -198,7 +211,7 @@ export function forwardAnnounced(conn: Established, origin: OriginProducer): voi /** * Settle a request nothing beneath it watches any more. A broader open request that is * still needed holds it until that one goes, so the broader never takes its routes over - * before it has caught up; otherwise it lingers out. + * before it has synced; otherwise it lingers out. */ const release = (prefix: Path.Valid, interest: Interest) => { for (const above of coveringPrefixes(prefix)) { diff --git a/js/net/src/connection/handshake.ts b/js/net/src/connection/handshake.ts index f7dd9eb8d90..218ee518e45 100644 --- a/js/net/src/connection/handshake.ts +++ b/js/net/src/connection/handshake.ts @@ -29,6 +29,7 @@ export async function exchangeSetup( early: Reader[]; solicit: boolean | undefined; hidden: boolean; + activeCount: boolean; cluster: Ietf.Cluster.Hops; auth: boolean; }> { @@ -37,6 +38,7 @@ export async function exchangeSetup( params.setBytes(Ietf.SetupOption.Implementation, encoder.encode(implementation)); Extensions.intoSetup(params, extensions, version); Ietf.hiddenIntoSetup(params); + Ietf.activeCountIntoSetup(params, version); // One id per session, like the moq-lite connection: nothing in this process forwards // between sessions, so there is nothing for a shared id to detect. @@ -64,6 +66,7 @@ export async function exchangeSetup( early: received.early, solicit: received.solicit, hidden: received.hidden, + activeCount: received.activeCount, cluster: { self, peer: received.cluster }, // Negotiated only when both sides offer it. auth: received.auth && extensions.auth, @@ -94,6 +97,7 @@ async function receiveSetup( early: Reader[]; solicit: boolean | undefined; hidden: boolean; + activeCount: boolean; cluster: Hop | undefined; auth: boolean; }> { @@ -147,6 +151,7 @@ async function receiveSetup( early, solicit: Ietf.solicitFromSetup(setup.parameters), hidden: Ietf.hiddenFromSetup(setup.parameters), + activeCount: Ietf.activeCountFromSetup(setup.parameters, version), cluster: Ietf.Cluster.fromSetup(setup.parameters, version), auth: Ietf.Auth.fromSetup(setup.parameters, version) === true, }; diff --git a/js/net/src/connection/pool.test.ts b/js/net/src/connection/pool.test.ts index 9a22eb8aee6..3a8a4451c74 100644 --- a/js/net/src/connection/pool.test.ts +++ b/js/net/src/connection/pool.test.ts @@ -607,3 +607,20 @@ test("pooled probe readers share demand and carry it through reconnects", async await settle(); } }); + +test("a handle's announcements sync with the shared origin", async () => { + stubTransports(); + + const handle = new Connection({ url }); + const watch = handle.announced(); + expect(watch.synced.peek()).toBe(false); + + await advance(); + expect(handle.status.peek()).toBe("connected"); + // The session's announce request answers after it establishes. + await advance(); + expect(watch.synced.peek()).toBe(true); + + watch.close(); + handle.close(); +}); diff --git a/js/net/src/connection/pool.ts b/js/net/src/connection/pool.ts index 252f1e60202..ac37be11839 100644 --- a/js/net/src/connection/pool.ts +++ b/js/net/src/connection/pool.ts @@ -26,7 +26,7 @@ import type { Established as EstablishedType } from "./established.ts"; import type { Extensions as ExtensionsType } from "./extensions.ts"; import { DISCOVERY_REMOVED } from "./forward.ts"; import type { GoawayProps, Redirect as RedirectType } from "./goaway.ts"; -import { Reload, type ReloadDelay, type ReloadStatus } from "./reload.ts"; +import { pumpAnnounced, Reload, type ReloadDelay, type ReloadStatus } from "./reload.ts"; import type { Probe as ProbeType, Stats as StatsType } from "./stats.ts"; import type { Transport as TransportType } from "./transport.ts"; @@ -281,44 +281,20 @@ export class Connection { const producer = new Announce.Producer(); const consumer = producer.consume(); - // Closing the consumer closes the shared state, so stop appending after that. - let closed = false; - void consumer.closed.then(() => { - closed = true; - }); - // A child of this handle's scope rather than a standalone Effect it merely cleans up // after: the disposer run() hands back also drops itself from the parent, so opening // and closing announcement streams repeatedly does not pile up dead pumps that live // until the whole handle closes. const stop = this.#signals.run((effect) => { const origin = effect.get(this.#origin); - if (!origin) return; - - const upstream = origin.announced(scope, options); - effect.cleanup(() => upstream.close()); - - // Track what this origin announced so a URL switch retracts it; the last - // event rides along for the retraction. - const active = new Map(); - - effect.spawn(async () => { - try { - for (;;) { - const entry = await effect.race(upstream.next()); - if (!entry) break; - if (entry.kind === "end") active.delete(entry.prefix); - else active.set(entry.prefix, entry); - producer.append(entry); - } - } finally { - if (!closed) { - for (const entry of active.values()) { - producer.append({ ...entry, kind: "end" }); - } - } - } - }); + if (!origin) { + // Like a reconnecting loop's origin: behind while enabled with nothing to deliver. + if (effect.get(this.enabled)) producer.behind(); + else producer.synced(); + return; + } + // A URL switch retracts what the old origin announced. + pumpAnnounced(effect, origin.announced(scope, options), producer); }); void consumer.closed.then(stop); diff --git a/js/net/src/connection/reload.test.ts b/js/net/src/connection/reload.test.ts index 19ce8d11038..2385d760dcb 100644 --- a/js/net/src/connection/reload.test.ts +++ b/js/net/src/connection/reload.test.ts @@ -1,16 +1,18 @@ import { expect, spyOn, test } from "bun:test"; import { Milli } from "@moq/clock"; import { Effect, Signal } from "@moq/signals"; +import * as Announce from "../announced.ts"; import type { Producer as BroadcastProducer } from "../broadcast.ts"; import { SessionCode, SessionError, StreamCode, toTransport } from "../error.ts"; import { Route } from "../hop.ts"; +import { ALPN as IetfALPN } from "../ietf/version.ts"; import * as Lite from "../lite/index.ts"; import { createMockTransportPair } from "../mock.ts"; import { Producer as OriginProducer } from "../origin.ts"; import * as Path from "../path.ts"; import { wireOf } from "../wire.ts"; import { type AcceptProps, accept as acceptSession } from "./index.ts"; -import { Reload, type ReloadProps } from "./reload.ts"; +import { pumpAnnounced, Reload, type ReloadProps } from "./reload.ts"; function accept(transport: WebTransport, url: URL, props: Omit = {}) { return acceptSession({ transport, url, ...props }); @@ -600,3 +602,103 @@ test("announced accepts a non-prefix scope before any session exists", () => { reload.close(); } }); + +/** + * A watcher on the consume origin syncs only once the session has delivered the + * server's initial set, exactly on versions that count it (lite-05+, MoQ Active Count), and + * a reconnect takes it back until the next session's set lands. + */ +test("a watcher syncs with the server's initial set, and again after a reconnect", async () => { + // Through the consume origin, and through the handle's own stream without one. + for (const [alpn, viaOrigin] of [ + [Lite.ALPN_06, true], + [IetfALPN.DRAFT_19, true], + [Lite.ALPN_06, false], + [IetfALPN.DRAFT_19, false], + ] as const) { + const original = globalThis.WebTransport; + const url = new URL("https://example.com/"); + + const serverOrigin = new OriginProducer(); + const published = [publish(serverOrigin, Path.from("a")), publish(serverOrigin, Path.from("b"))]; + const sessions: { abort: () => void }[] = []; + const stub = function StubWebTransport() { + const pair = createMockTransportPair(alpn); + void accept(pair.server, url, { publish: serverOrigin.consume() }).then((server) => { + sessions.push(server); + }); + return pair.client; + }; + globalThis.WebTransport = stub as unknown as typeof WebTransport; + + const clientOrigin = new OriginProducer(); + const reload = new Reload({ + enabled: true, + url, + websocket: { enabled: false }, + delay: { initial: Milli(10), multiplier: 1, max: Milli(10) }, + consume: viaOrigin ? clientOrigin : undefined, + }); + const watching = viaOrigin ? clientOrigin.announced() : reload.announced(); + const seen = new Set(); + void (async () => { + for await (const event of watching) { + if (event.kind === "end") seen.delete(event.prefix); + else seen.add(event.prefix); + } + })(); + + try { + // Nothing is connected yet, which is not the same as nothing being announced. + expect(watching.synced.peek()).toBe(false); + await waitUntil(() => watching.synced.peek()); + expect([...seen].sort()).toEqual(["a", "b"]); + + sessions[0]?.abort(); + await waitUntil(() => !watching.synced.peek()); + await waitUntil(() => sessions.length > 1 && watching.synced.peek()); + expect([...seen].sort()).toEqual(["a", "b"]); + } finally { + watching.close(); + reload.close(); + clientOrigin.close(); + for (const broadcast of published) broadcast.close(); + serverOrigin.close(); + for (const session of sessions) session.abort(); + globalThis.WebTransport = original; + } + } +}); + +/** Replacing a synced upstream with one already synced still owes the old routes' ends. */ +test("a pump owes the retractions of the upstream it replaces", async () => { + const tick = async () => { + for (let i = 0; i < 16; i++) await Promise.resolve(); + }; + const producer = new Announce.Producer(); + const consumer = producer.consume(); + + const first = new Announce.Producer(); + first.append({ prefix: Path.from("a"), captures: undefined, kind: "start", route: Route.default }); + first.synced(); + const upstream = new Signal(first); + const pump = new Effect((effect) => pumpAnnounced(effect, effect.get(upstream).consume(), producer)); + + expect(await consumer.next()).toMatchObject({ prefix: Path.from("a"), kind: "start" }); + const parked = consumer.next(); + await tick(); + expect(consumer.synced.peek()).toBe(true); + + const second = new Announce.Producer(); + second.synced(); + upstream.set(second); + expect(await parked).toMatchObject({ prefix: Path.from("a"), kind: "end" }); + await tick(); + expect(consumer.synced.peek()).toBe(false); + void consumer.next(); + await tick(); + expect(consumer.synced.peek()).toBe(true); + + pump.close(); + producer.close(); +}); diff --git a/js/net/src/connection/reload.ts b/js/net/src/connection/reload.ts index 1a52747966e..9570ddb5927 100644 --- a/js/net/src/connection/reload.ts +++ b/js/net/src/connection/reload.ts @@ -243,7 +243,23 @@ export class Reload { // is recoverable (a new URL or a disable/re-enable starts another sequence), so a // request must keep waiting rather than go unroutable in the gap. if (this.consume) { - this.#signals.cleanup(wireOf(this.consume).expect()); + const consume = wireOf(this.consume); + this.#signals.cleanup(consume.expect()); + // Likewise, its announcement streams are not synced while no session is there to + // deliver an initial set, so an empty table before the first connect, or during a + // reconnect, never reads as nothing being announced. The next session takes its own + // hold before this one is released. Taken now rather than in the effect, which runs + // a microtask later, after a watcher created alongside would already have synced. + let hold = this.enabled.peek() ? consume.snapshot(Path.empty()) : undefined; + this.#signals.run((effect) => { + if (effect.get(this.enabled) && !effect.get(this.established)) { + hold ??= consume.snapshot(Path.empty()); + } else { + hold?.(); + hold = undefined; + } + }); + this.#signals.cleanup(() => hold?.()); } this.error = this.#error; @@ -582,36 +598,16 @@ export class Reload { const pump = new Effect(); pump.run((effect) => { const conn = effect.get(this.established); - if (!conn) return; - - const upstream = conn.announced(scope, options); - effect.cleanup(() => upstream.close()); - - // Track what this connection announced so we can retract it if the connection - // drops; the last event rides along for the retraction. - const active = new Map(); + if (!conn) { + // The same hold as a consume origin's: behind while enabled with no session. + if (effect.get(this.enabled)) producer.behind(); + else producer.synced(); + return; + } - effect.spawn(async () => { - try { - for (;;) { - const entry = await effect.race(upstream.next()); - if (!entry) break; - if (entry.kind === "end") active.delete(entry.prefix); - else active.set(entry.prefix, entry); - producer.append(entry); - } - } catch { - // A dropped connection resets the announce stream; the retractions below cover it. - } finally { - // Retract everything from the connection that just went away, so a per-broadcast - // watcher tears down instead of clinging to the dead route. - if (consumer.closed.peek() === undefined) { - for (const entry of active.values()) { - producer.append({ ...entry, kind: "end" }); - } - } - } - }); + // A dropped connection retracts what it announced, so a per-broadcast watcher tears + // down instead of clinging to the dead route. + pumpAnnounced(effect, conn.announced(scope, options), producer); }); this.#signals.cleanup(() => pump.close()); @@ -636,6 +632,60 @@ export class Reload { } } +/** + * Pump `upstream` into `producer` for the rest of `effect`'s run: its events, then an `end` + * for each route still active once it closes or the run ends, so nothing clings to a source + * that went away. Synced follows the upstream's, and owes those retractions too. + * + * @internal + */ +export function pumpAnnounced(effect: Effect, upstream: Announce.Consumer, producer: Announce.Producer) { + // The last event per active route rides along for its retraction. + const active = new Map(); + let live = true; + const retract = () => { + live = false; + if (producer.closed.peek() !== undefined || active.size === 0) return; + // Behind first, so a reader parked on the stream takes these as owed, not as news. + producer.behind(); + for (const entry of active.values()) producer.append({ ...entry, kind: "end" }); + active.clear(); + }; + + // Torn down in reverse: the mirror stops, the retractions land, then the upstream closes, + // so its closing never reads as synced. + effect.cleanup(() => upstream.close()); + effect.cleanup(retract); + // A new upstream owes its own initial set, and the retractions of the one before. + producer.behind(); + effect.subscribe(upstream.synced, (synced) => { + if (!synced) producer.behind(); + else if (upstream.closed.peek() === undefined) producer.synced(); + }); + + effect.spawn(async () => { + try { + for (;;) { + const entry = await effect.race(upstream.next()); + if (!entry || !live || producer.closed.peek() !== undefined) break; + if (entry.kind === "end") active.delete(entry.prefix); + else active.set(entry.prefix, entry); + producer.append(entry); + } + } catch { + // A dropped connection resets the announce stream; the retractions below cover it. + } finally { + if (live) { + // The upstream ended on its own: nothing more comes from it, so once its + // retractions are read there is nothing left to wait for. + retract(); + producer.behind(); + producer.synced(); + } + } + }); +} + /** * A session the peer sent GOAWAY on, left serving so its groups in flight finish. It retires * when it closes on its own or overstays its handover window, and `onRetire` runs once either way. diff --git a/js/net/src/ietf/active_count.ts b/js/net/src/ietf/active_count.ts new file mode 100644 index 00000000000..361d98c9777 --- /dev/null +++ b/js/net/src/ietf/active_count.ts @@ -0,0 +1,38 @@ +import { SetupOption, type SetupOptions } from "./parameters.ts"; +import { type IetfVersion, Version } from "./version.ts"; + +/** + * The MoQ Active Count extension (draft-lcurley-moq-active-count-00). + * + * A publisher answers SUBSCRIBE_NAMESPACE with a NAMESPACE for every suffix it already + * advertises, then reports changes on the same stream. Nothing in moq-transport says where + * the first part ends, so when both endpoints declare the ACTIVE_COUNT option, the + * publisher puts on its REQUEST_OK how many NAMESPACE messages come before the + * subscription is synced. + * + * @module + * @internal + */ + +// Draft-14/15 answer SUBSCRIBE_NAMESPACE with PUBLISH_NAMESPACE requests, which the stream cannot count. +function supported(version: IetfVersion): boolean { + return version !== Version.DRAFT_14 && version !== Version.DRAFT_15; +} + +/** + * Whether the extension is negotiated. We always declare it, so the peer's SETUP decides. + * + * @internal + */ +export function activeCountFromSetup(params: SetupOptions, version: IetfVersion): boolean { + return supported(version) && params.getVarint(SetupOption.ActiveCount) !== undefined; +} + +/** + * Declare that we count and read the peer's count. + * + * @internal + */ +export function activeCountIntoSetup(params: SetupOptions, version: IetfVersion) { + if (supported(version)) params.setVarint(SetupOption.ActiveCount, 1n); +} diff --git a/js/net/src/ietf/connection.ts b/js/net/src/ietf/connection.ts index 7291ee9389d..27efad21c18 100644 --- a/js/net/src/ietf/connection.ts +++ b/js/net/src/ietf/connection.ts @@ -123,6 +123,7 @@ export class Connection implements Established { publish, solicit, hidden = false, + activeCount = false, cluster, auth = false, early = [], @@ -145,6 +146,8 @@ export class Connection implements Established { solicit?: boolean; /** Whether the peer understands the HIDDEN parameter (MoQ Hidden). */ hidden?: boolean; + /** Whether both SETUPs declared MoQ Active Count. */ + activeCount?: boolean; /** * The Hop IDs this session declared (MoQ Cluster). `undefined` on a version that * cannot negotiate the extension, as is a `peer` the peer never declared. @@ -210,6 +213,7 @@ export class Connection implements Established { publish, requiresSolicitation: solicit ?? false, hidden, + activeCount, cluster, grant: this.#auth.grant, auth: this.#auth, @@ -223,6 +227,7 @@ export class Connection implements Established { quic, cluster, hidden, + activeCount, solicit, goaway: this.#goaway, grant: this.#auth.grant, diff --git a/js/net/src/ietf/index.ts b/js/net/src/ietf/index.ts index 269f01b1b9c..784a064100d 100644 --- a/js/net/src/ietf/index.ts +++ b/js/net/src/ietf/index.ts @@ -1,3 +1,4 @@ +export * from "./active_count.ts"; export * from "./adapter.ts"; export * as Auth from "./auth.ts"; export * as Cluster from "./cluster.ts"; diff --git a/js/net/src/ietf/parameters.ts b/js/net/src/ietf/parameters.ts index dd51b8f3877..13f9d5f4cca 100644 --- a/js/net/src/ietf/parameters.ts +++ b/js/net/src/ietf/parameters.ts @@ -25,6 +25,8 @@ export const SetupOption = { Hidden: 0x40b5cn, /** AUTH, from the MoQ Auth extension. See `auth.ts`. */ Auth: 0x40b60n, + /** ACTIVE_COUNT, from the MoQ Active Count extension. See `active_count.ts`. */ + ActiveCount: 0x40b64n, } as const; // Unknown SETUP options may repeat, including GREASE (draft-21 section 9.1). @@ -552,6 +554,19 @@ export class Parameters { else this.vars.delete(MSG_PARAM_HIDDEN); } + /** How many NAMESPACE messages follow a REQUEST_OK before the subscription is synced (MoQ Active Count). */ + get activeCount(): number | undefined { + const v = this.vars.get(MSG_PARAM_ACTIVE_COUNT); + if (v === undefined) return undefined; + if (v > BigInt(Number.MAX_SAFE_INTEGER)) throw new ProtocolViolation(`ACTIVE_COUNT too large: ${v}`); + return Number(v); + } + + set activeCount(v: number | undefined) { + if (v === undefined) this.vars.delete(MSG_PARAM_ACTIVE_COUNT); + else this.vars.set(MSG_PARAM_ACTIVE_COUNT, BigInt(v)); + } + // --- Bytes accessors --- get largest(): MessageLocation | undefined { diff --git a/js/net/src/ietf/publisher.test.ts b/js/net/src/ietf/publisher.test.ts index e7c1e70665e..5e29d8ce54d 100644 --- a/js/net/src/ietf/publisher.test.ts +++ b/js/net/src/ietf/publisher.test.ts @@ -1,4 +1,4 @@ -import { expect, jest, spyOn, test } from "bun:test"; +import { expect, jest, onTestFinished, spyOn, test } from "bun:test"; import { Milli } from "@moq/clock"; import { Signal } from "@moq/signals"; import type { Auth, Grant } from "../auth.ts"; @@ -19,6 +19,7 @@ import type * as Cluster from "./cluster.ts"; import { toRequestCode } from "./error.ts"; import { Fetch, FetchHeader } from "./fetch.ts"; import { Frame, Group as GroupMessage } from "./object.ts"; +import { Parameters } from "./parameters.ts"; import { toWire } from "./priority.ts"; import { PublishDone } from "./publish.ts"; import { PublishNamespace, PublishNamespaceUpdate } from "./publish_namespace.ts"; @@ -120,12 +121,14 @@ function publisher( session, cluster, grant, + activeCount, auth, }: { requiresSolicitation?: boolean; session?: Session; cluster?: Cluster.Hops; grant?: Signal; + activeCount?: boolean; auth?: Tokens; } = {}, ): { pub: Publisher; origin: OriginProducer } { @@ -139,6 +142,7 @@ function publisher( requiresSolicitation, cluster, grant, + activeCount, auth, }), origin, @@ -3218,3 +3222,59 @@ test("namespace updates move the prefix after acknowledgement", async () => { } } }); + +/** MoQ Active Count: the REQUEST_OK counts the NAMESPACEs sent ahead of any change. */ +test("a counted answer names the NAMESPACEs that follow it", async () => { + for (const activeCount of [true, false]) { + const pair = createMockTransportPair(ALPN.DRAFT_19); + const { pub, origin } = publisher(pair.server, { requiresSolicitation: true, activeCount }); + publish(origin, Path.from("a")); + publish(origin, Path.from("b")); + + const subscription = await Stream.open(pair.client, { version: VERSION }); + const accepted = await Stream.accept(pair.server, VERSION); + if (!accepted) throw new Error("the subscription stream was never accepted"); + void pub.runSubscribeNamespace(new SubscribeNamespace({ requestId: 0n, namespace: Path.empty() }), accepted); + + expect(await subscription.reader.u53()).toBe(RequestOk.id); + const ok = await RequestOk.decode(subscription.reader, VERSION); + expect(ok.parameters.activeCount).toBe(activeCount ? 2 : undefined); + const suffixes = []; + for (let i = 0; i < 2; i++) { + expect(await subscription.reader.u53()).toBe(SubscribeNamespaceEntry.id); + suffixes.push((await SubscribeNamespaceEntry.decode(subscription.reader, VERSION)).suffix); + } + expect(suffixes.sort()).toEqual([Path.from("a"), Path.from("b")]); + + subscription.close(); + origin.close(); + } +}); + +/** ACTIVE_COUNT only answers SUBSCRIBE_NAMESPACE, so one on any other REQUEST_OK is a violation. */ +test.each([0, Number.MAX_SAFE_INTEGER + 2])( + "an ACTIVE_COUNT of %p answering a PUBLISH_NAMESPACE closes the session", + async (count) => { + const pair = createMockTransportPair(ALPN.DRAFT_19); + const { pub, origin } = publisher(pair.server, { activeCount: true }); + const warn = spyOn(console, "warn").mockImplementation(() => void 0); + onTestFinished(() => warn.mockRestore()); + void pub.runPublishNamespaces(); + publish(origin, Path.from("a")); + + const stream = await nextStream(pair.client); + if (!stream) throw new Error("no PUBLISH_NAMESPACE was sent"); + expect(await readPublishNamespace(stream)).toBe(Path.from("a")); + const parameters = new Parameters(); + parameters.activeCount = count; + await stream.writer.u53(RequestOk.id); + await new RequestOk({ requestId: undefined, parameters }).encode(stream.writer, VERSION); + + const closed = await Promise.race([ + pair.server.closed, + new Promise((_resolve, reject) => setTimeout(() => reject(new Error("session stayed up")), 500)), + ]); + expect(closed.closeCode).toBe(new ProtocolViolation("").code); + origin.close(); + }, +); diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts index da8f0b86290..9db63dff242 100644 --- a/js/net/src/ietf/publisher.ts +++ b/js/net/src/ietf/publisher.ts @@ -286,6 +286,8 @@ export class Publisher { // turns them off: there is no stream fallback. Released in close(). #datagramWriter?: WritableStreamDefaultWriter; #hidden: boolean; + // Whether a SUBSCRIBE_NAMESPACE's REQUEST_OK counts the NAMESPACEs that follow it (MoQ Active Count). + #activeCount: boolean; // The origin this session serves, borrowed: it outlives the session, and closing the // session leaves its broadcasts alone. Namespaces go out as an unsolicited @@ -323,6 +325,7 @@ export class Publisher { publish, requiresSolicitation, hidden = false, + activeCount = false, cluster, grant = new Signal(undefined), auth, @@ -339,6 +342,8 @@ export class Publisher { requiresSolicitation: boolean; /** Whether the peer declared MoQ Hidden. */ hidden?: boolean; + /** Whether both SETUPs declared MoQ Active Count. */ + activeCount?: boolean; /** The Hop IDs the SETUP exchange settled (MoQ Cluster). */ cluster?: Cluster.Hops; /** The union of our tokens' grants (MoQ Auth), which bounds what we publish. */ @@ -362,6 +367,7 @@ export class Publisher { this.#publish = publish; this.#requiresSolicitation = requiresSolicitation; this.#hidden = hidden; + this.#activeCount = activeCount; this.#advert = Cluster.advertise(cluster); this.#datagramWriter = DatagramStream.datagramWriter(quic); } @@ -1199,18 +1205,27 @@ export class Publisher { ); let updating = version >= Version.DRAFT_18 ? readUpdate() : undefined; try { - // Send OK response - if (version === Version.DRAFT_14) { - await stream.writer.u53(SubscribeNamespaceOk.id); - const ok = new SubscribeNamespaceOk({ requestId: msg.requestId }); - await ok.encode(stream.writer, version); - } else { + const answer = async (activeCount?: number) => { + if (version === Version.DRAFT_14) { + await stream.writer.u53(SubscribeNamespaceOk.id); + const ok = new SubscribeNamespaceOk({ requestId: msg.requestId }); + await ok.encode(stream.writer, version); + return; + } + const parameters = new Parameters(); + parameters.activeCount = activeCount; await stream.writer.u53(RequestOk.id); const ok = new RequestOk({ requestId: version === Version.DRAFT_15 || version === Version.DRAFT_16 ? msg.requestId : undefined, + parameters, }); await ok.encode(stream.writer, version); - } + }; + + // MoQ Active Count: the OK counts what the first pass below sends, so it waits for + // that pass. Otherwise answer now. + let answered = !this.#activeCount; + if (answered) await answer(); // Hidden namespaces are left out unless the peer opted in (MoQ Hidden). Draft-16 // and later carry every remaining match. Draft-14/15 only add what the @@ -1278,12 +1293,19 @@ export class Publisher { }); const advertised = this.#advertised.peek(); + const updated = advertised ? presented(prefix, advertised, carries) : new Map(); + // A fresh pass sends a NAMESPACE for every entry, ahead of any change. + if (!answered) { + answered = true; + await answer(updated.size); + } if (!advertised) { dispose(); break; } - const outstanding = await this.#reconcile(ns, presented(prefix, advertised, carries), target); + const outstanding = await this.#reconcile(ns, updated, target); + retry = outstanding ? Math.min(retry ? retry * 2 : RETRY_BASE, RETRY_MAX) : 0; // Wait for the next change, or for the peer to unsubscribe. @@ -1656,7 +1678,7 @@ export class Publisher { if (this.#session.version === Version.DRAFT_14) { await PublishNamespaceOk.decode(stream.reader, this.#session.version); } else { - await RequestOk.decode(stream.reader, this.#session.version); + this.#answered(await RequestOk.decode(stream.reader, this.#session.version)); } return "held"; })(), @@ -1675,6 +1697,25 @@ export class Publisher { } } + /** + * Check a REQUEST_OK answering a PUBLISH_NAMESPACE. ACTIVE_COUNT only answers + * SUBSCRIBE_NAMESPACE (MoQ Active Count), and the session closes over it here, with the + * violation's code, since the caller only drops the advertisement. + */ + #answered(ok: RequestOk) { + try { + // Reading it throws on a count too large to hold, a violation all the same. + if (ok.parameters.activeCount === undefined) return; + throw new ProtocolViolation("ACTIVE_COUNT on a PUBLISH_NAMESPACE answer"); + } catch (err) { + if (err instanceof ProtocolViolation) { + this.#quic.close({ closeCode: err.code, reason: err.message }); + this.#session.close(); + } + throw err; + } + } + /** * Update a namespace the peer holds: REQUEST_UPDATE on the request that carries it, * with only the parameters that changed, then its answer. @@ -1720,7 +1761,7 @@ export class Publisher { if (typeId !== RequestOk.id) { throw new Error(`unexpected answer to the update: typeId=0x${typeId.toString(16)}`); } - await RequestOk.decode(reader, version); + this.#answered(await RequestOk.decode(reader, version)); return "held"; })(), ADVERTISE_TIMEOUT, diff --git a/js/net/src/ietf/subscriber.test.ts b/js/net/src/ietf/subscriber.test.ts index 8d57fff13c2..d1c584963c6 100644 --- a/js/net/src/ietf/subscriber.test.ts +++ b/js/net/src/ietf/subscriber.test.ts @@ -16,6 +16,7 @@ import type * as Cluster from "./cluster.ts"; import { Connection } from "./connection.ts"; import { ObjectDatagram } from "./datagram.ts"; import { EndOfTrackProperties, encodeObjectExtensions, type GroupFlags, Group as GroupMessage } from "./object.ts"; +import { Parameters } from "./parameters.ts"; import type { Properties } from "./properties.ts"; import { PublishNamespace, PublishNamespaceCancel, PublishNamespaceUpdate } from "./publish_namespace.ts"; import { MaxRequestId, RequestError, RequestOk } from "./request.ts"; @@ -2093,3 +2094,112 @@ test("an announce consumer closed while waiting on request-ID credit takes no ID session.close(); void running; }); + +/** Answer the next SUBSCRIBE_NAMESPACE with REQUEST_OK, carrying `activeCount` when given. */ +async function answerCounted(transport: WebTransport, activeCount: number | undefined): Promise { + const stream = await nextStream(transport); + if (!stream) throw new Error("no SUBSCRIBE_NAMESPACE was sent"); + await stream.reader.u53(); + await SubscribeNamespace.decode(stream.reader, VERSION); + const parameters = new Parameters(); + parameters.activeCount = activeCount; + await stream.writer.u53(RequestOk.id); + await new RequestOk({ requestId: undefined, parameters }).encode(stream.writer, VERSION); + return stream; +} + +const sessionClosed = (pair: ReturnType) => + Promise.race([ + pair.server.closed, + new Promise((_resolve, reject) => setTimeout(() => reject(new Error("session stayed up")), STREAM_WAIT)), + ]); + +/** The session closes with PROTOCOL_VIOLATION, not a clean code. */ +async function violationClosed(pair: ReturnType) { + expect((await sessionClosed(pair)).closeCode).toBe(new ProtocolViolation("").code); +} + +const yieldTask = () => new Promise((resolve) => setTimeout(resolve, 0)); + +/** MoQ Active Count: the consumer syncs once the counted NAMESPACEs are read. */ +test("counted NAMESPACEs sync", async () => { + const pair = createMockTransportPair(ALPN.DRAFT_19); + const session = new NativeSession(pair.server, VERSION, true); + const subscriber = new Subscriber({ session, activeCount: true }); + + const announced = subscriber.announced(); + const subscription = await answerCounted(pair.client, 2); + await inlineNamespace(subscription, Path.from("a")); + expect(await announced.next()).toMatchObject({ prefix: Path.from("a"), kind: "start" }); + const b = announced.next(); + await yieldTask(); + expect(announced.synced.peek()).toBe(false); + + await inlineNamespace(subscription, Path.from("b")); + expect(await b).toMatchObject({ prefix: Path.from("b"), kind: "start" }); + await yieldTask(); + // Still handling the last of them. + expect(announced.synced.peek()).toBe(false); + void announced.next(); + await yieldTask(); + expect(announced.synced.peek()).toBe(true); + announced.close(); +}); + +/** Without MoQ Active Count, REQUEST_OK is the best boundary there is. */ +test("an uncounted answer syncs on the OK", async () => { + const pair = createMockTransportPair(ALPN.DRAFT_19); + const session = new NativeSession(pair.server, VERSION, true); + const subscriber = new Subscriber({ session }); + + const announced = subscriber.announced(); + await yieldTask(); + expect(announced.synced.peek()).toBe(false); + await acceptSubscribeNamespace(pair.client); + await yieldTask(); + expect(announced.synced.peek()).toBe(true); + announced.close(); +}); + +/** A count the negotiation did not promise, or a missing one it did, breaks the extension. */ +test("an ACTIVE_COUNT that does not match the negotiation closes the session", async () => { + for (const [negotiated, count] of [ + [true, undefined], + [false, 0], + ] as const) { + const pair = createMockTransportPair(ALPN.DRAFT_19); + const session = new NativeSession(pair.server, VERSION, true); + const subscriber = new Subscriber({ session, quic: pair.server, activeCount: negotiated }); + const warn = spyOn(console, "warn").mockImplementation(() => void 0); + onTestFinished(() => warn.mockRestore()); + + subscriber.announced(); + await answerCounted(pair.client, count); + await violationClosed(pair); + } +}); + +/** A counted set cut short, by a NAMESPACE_DONE or the stream ending, is the peer's fault. */ +test.each(["done", "fin"] as const)("a counted set cut short by %s closes the session", async (cut) => { + { + const pair = createMockTransportPair(ALPN.DRAFT_19); + const session = new NativeSession(pair.server, VERSION, true); + const subscriber = new Subscriber({ session, quic: pair.server, activeCount: true }); + const warn = spyOn(console, "warn").mockImplementation(() => void 0); + onTestFinished(() => warn.mockRestore()); + + subscriber.announced(); + const subscription = await answerCounted(pair.client, 2); + await inlineNamespace(subscription, Path.from("a")); + if (cut === "done") { + // The type alone is the violation, so the session may close before the rest is written. + await subscription.writer.u53(SubscribeNamespaceEntryDone.id); + await new SubscribeNamespaceEntryDone({ suffix: Path.from("a") }) + .encode(subscription.writer, VERSION) + .catch(() => undefined); + } else { + subscription.writer.close(); + } + await violationClosed(pair); + } +}); diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index ddfd992f0fa..76d2f22ce8b 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -168,6 +168,9 @@ export class Subscriber { // Whether the peer understands the HIDDEN parameter (MoQ Hidden). #hidden: boolean; + // Whether a SUBSCRIBE_NAMESPACE's REQUEST_OK counts the NAMESPACEs that follow it (MoQ Active Count). + #activeCount: boolean; + // What the peer's SETUP declared about being solicited (MoQ Solicit), `undefined` // when it declared nothing. #solicit?: boolean; @@ -193,6 +196,7 @@ export class Subscriber { quic, cluster, hidden = false, + activeCount = false, solicit, goaway, grant, @@ -206,6 +210,8 @@ export class Subscriber { cluster?: Cluster.Hops; /** Whether the peer understands the HIDDEN parameter (MoQ Hidden). */ hidden?: boolean; + /** Whether both SETUPs declared MoQ Active Count. */ + activeCount?: boolean; /** What the peer's SETUP declared about being solicited (MoQ Solicit). */ solicit?: boolean; /** Settles when the peer sends GOAWAY. */ @@ -220,6 +226,7 @@ export class Subscriber { this.#announced = new SlotMap(new Slots(limits.announces, { quic, reason: "too many announcements" })); this.#cluster = cluster; this.#hidden = hidden; + this.#activeCount = activeCount; this.#solicit = solicit; this.#goaway = goaway; this.#grant = grant; @@ -391,6 +398,17 @@ export class Subscriber { } } + // The peer broke the protocol: close with the violation's code before the session's own + // close, which would send a clean one, as the connection does for its own violations. + #violated(err: ProtocolViolation) { + try { + this.#quic?.close({ closeCode: err.code, reason: err.message }); + } catch { + // Already closed. + } + this.#session.close(); + } + async #runAnnounced(announced: announce.Producer, prefix: Path.Valid, hidden: boolean) { const version = this.#session.version; @@ -401,6 +419,8 @@ export class Subscriber { // could land. A peer that declared Solicit only tells when asked, so it still is. const legacy = version === Version.DRAFT_14 || version === Version.DRAFT_15; if (legacy && prefix.length === 0 && this.#solicit === undefined) { + // Nothing marks where what it tells unasked starts or ends, so there is nothing to wait for. + announced.synced(); // No request stream ends this wait, so the session's end has to. const ends: PromiseLike[] = [announced.closed]; if (this.#quic) ends.push(this.#quic.closed.catch(() => undefined)); @@ -483,39 +503,63 @@ export class Subscriber { stream.close(); }; + // Resolves to how many NAMESPACEs make up the initial set (MoQ Active Count). const response = (async () => { const respTypeId = await stream.reader.u53(); if (respTypeId === RequestOk.id) { - await RequestOk.decode(stream.reader, version); - } else if (respTypeId === SubscribeNamespaceOk.id) { + const ok = await RequestOk.decode(stream.reader, version); + // A count the negotiation did not promise, or a missing one it did, is + // the peer breaking the extension. Without it, the answer is the best + // boundary there is. + const count = ok.parameters.activeCount; + if (this.#activeCount !== (count !== undefined)) { + throw new ProtocolViolation("ACTIVE_COUNT does not match the negotiation"); + } + return count ?? 0; + } + if (respTypeId === SubscribeNamespaceOk.id) { // v14: SubscribeNamespaceOk const size = await stream.reader.u16(); await stream.reader.read(size); - } else { - throw new Error(`SubscribeNamespace rejected: typeId=0x${respTypeId.toString(16)}`); + return 0; } - return true; + throw new Error(`SubscribeNamespace rejected: typeId=0x${respTypeId.toString(16)}`); })(); // A consumer that leaves before the peer answers ends the request now, not on the answer. - if (!(await race([response, announced.closed.then(() => false)]))) { + const answered = await race([response, announced.closed.then(() => undefined)]); + if (answered === undefined) { await end(); return; } + let initial = answered; // Loop reading Namespace/NamespaceDone entries const readLoop = (async () => { for (;;) { + if (initial === 0) announced.synced(); + const done = await stream.reader.done(); - if (done) break; + if (done) { + if (initial > 0) + throw new ProtocolViolation("namespace stream ended inside the initial set"); + break; + } const msgType = await stream.reader.u53(); + // The counted initial set is NAMESPACEs only, so anything else cuts it short. + if (initial > 0 && msgType !== SubscribeNamespaceEntry.id) { + throw new ProtocolViolation(`message 0x${msgType.toString(16)} inside the initial set`); + } if (msgType === SubscribeNamespaceEntry.id) { const entry = await SubscribeNamespaceEntry.decode( stream.reader, version, Cluster.negotiated(this.#cluster), ); + // Counted even when discarded below. + if (initial > 0) initial -= 1; if (released) break; + const path = Path.join(prefix, entry.suffix); // A repeat updates the advertisement in place, so one that now @@ -561,7 +605,7 @@ export class Subscriber { // rejection would go unhandled. Idempotent with the catch below, which is // what runs when the loop is the one that ends the race. readLoop.catch((err: unknown) => { - if (err instanceof ProtocolViolation) this.#session.close(); + if (err instanceof ProtocolViolation) this.#violated(err); }); // Wait for either the read loop or the announced to close @@ -579,7 +623,7 @@ export class Subscriber { // is the only place a malformed one surfaces. The cluster draft requires closing // the session over those, and the stream alone is not enough: the peer would // just repeat it on the next SUBSCRIBE_NAMESPACE. - if (e instanceof ProtocolViolation) this.#session.close(); + if (e instanceof ProtocolViolation) this.#violated(e); // Abort the stream rather than letting the caller's `finally` close it cleanly. // A rejected namespace subscription is a failure, and a consumer that can't tell diff --git a/js/net/src/lite/subscriber.test.ts b/js/net/src/lite/subscriber.test.ts index 8789e8fb2f9..46d1606d539 100644 --- a/js/net/src/lite/subscriber.test.ts +++ b/js/net/src/lite/subscriber.test.ts @@ -124,6 +124,8 @@ function announceHarness(version: Version, origin = 1n) { return { subscriber, send, + // The peer FINs the announce stream. + finish: () => inbound.close(), abortReason, sessionEnded, settle: () => new Promise((resolve) => setTimeout(resolve, 0)), @@ -1828,6 +1830,93 @@ test("closing a broadcast while TRACK_INFO is pending leaves no unhandled reject } }); +const announceStart = (suffix: string) => (w: Writer) => + encodeAnnounceBroadcast(w, { status: "active", suffix: Path.from(suffix), hops: [PUBLISHER_A] }, Version.DRAFT_06); + +test("a counted initial set syncs once the reader is past its last announce", async () => { + const { subscriber, send, settle } = announceHarness(Version.DRAFT_06); + const announced = subscriber.announced(); + await settle(); + + await send((w) => new AnnounceOk(PEER, 2).encode(w, Version.DRAFT_06)); + await send(announceStart("a")); + expect(await announced.next()).toMatchObject({ prefix: Path.from("a"), kind: "start" }); + expect(announced.synced.peek()).toBe(false); + + await send(announceStart("b")); + expect(await announced.next()).toMatchObject({ prefix: Path.from("b"), kind: "start" }); + await settle(); + // Still handling the last of them. + expect(announced.synced.peek()).toBe(false); + + void announced.next(); + await settle(); + expect(announced.synced.peek()).toBe(true); + + announced.close(); + subscriber.close(); +}); + +test("an empty initial set syncs on the answer", async () => { + const { subscriber, send, settle } = announceHarness(Version.DRAFT_06); + const announced = subscriber.announced(); + await settle(); + expect(announced.synced.peek()).toBe(false); + + await send((w) => new AnnounceOk(PEER, 0).encode(w, Version.DRAFT_06)); + await settle(); + expect(announced.synced.peek()).toBe(true); + + announced.close(); + subscriber.close(); +}); + +test("a counted initial set cut short is a protocol violation", async () => { + for (const cut of ["ended", "fin"] as const) { + const { subscriber, send, finish, sessionEnded, settle } = announceHarness(Version.DRAFT_06); + const announced = subscriber.announced(); + await settle(); + + await send((w) => new AnnounceOk(PEER, 2).encode(w, Version.DRAFT_06)); + await send(announceStart("a")); + if (cut === "ended") { + await send((w) => encodeAnnounceBroadcast(w, { status: "endedId", id: 0n }, Version.DRAFT_06)); + } else { + finish(); + } + + expect(await announced.next()).toMatchObject({ prefix: Path.from("a"), kind: "start" }); + await expect(announced.next()).rejects.toThrow("inside the initial set"); + expect((await sessionEnded())?.closeCode).toBe(SessionCode.ProtocolViolation); + subscriber.close(); + } +}); + +test("a version without a boundary syncs once the request is out", async () => { + const { subscriber, settle } = announceHarness(Version.DRAFT_03); + const announced = subscriber.announced(); + await settle(); + expect(announced.synced.peek()).toBe(true); + + announced.close(); + subscriber.close(); +}); + +test("announce init is the initial set", async () => { + const { subscriber, send, settle } = announceHarness(Version.DRAFT_02); + const announced = subscriber.announced(); + await settle(); + + await send((w) => new AnnounceInit([Path.from("room")]).encode(w, Version.DRAFT_02)); + expect(await announced.next()).toMatchObject({ prefix: Path.from("room"), kind: "start" }); + void announced.next(); + await settle(); + expect(announced.synced.peek()).toBe(true); + + announced.close(); + subscriber.close(); +}); + test.each(["answer", "credit"] as const)("FETCH shares the TRACK deadline while waiting for %s", async (stage) => { const clock = new Manual(); const restore = clock.install(); diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index c43c88f4b01..fe1fe710e85 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -307,6 +307,9 @@ export class Subscriber { // It no longer stamps itself onto each hop chain, so we append it here to // keep the reflected-announce loop check seeing the full chain. let responderOrigin: Hop | undefined; + // The ANNOUNCEs still owed of the peer's initial set (lite-05+). Older versions + // mark no boundary, so they sync once the request is out, as a best effort. + let initial = 0; if (hasAnnounceOk(this.version)) { const ok = await race([AnnounceOk.decode(stream.reader, this.version), announced.closed]); // A consumer that leaves before the peer answers ends the request now, not on the answer. @@ -318,6 +321,7 @@ export class Subscriber { // anonymous mark and must travel the reconstructed chain. Assigned identities // stay off this hop and are never forwarded. responderOrigin = ok.hop; + initial = ok.active; } switch (this.version) { @@ -381,14 +385,28 @@ export class Subscriber { // Receive announce updates (for Draft03+, this includes initial state) for (;;) { + if (initial === 0) announced.synced(); + const announce = await race([ decodeAnnounceBroadcastMaybe(stream.reader, this.version), announced.closed, ]); // undefined: the stream ended. null: the consumer closed cleanly. + if (announce === undefined && initial > 0) { + throw new ProtocolViolation("announce stream ended inside the initial set"); + } if (!announce) break; if (announce instanceof Error) throw announce; + // The counted initial set is ANNOUNCEs only, so anything else cuts it short. An + // unknown message counts for nothing either way. + if (initial > 0 && announce.status !== "skipped") { + if (announce.status !== "active") { + throw new ProtocolViolation(`${announce.status} inside the initial set`); + } + initial -= 1; + } + let path: Path.Valid; let active: boolean; // Present on active/update/restart; ended messages never carry hops worth checking. diff --git a/js/net/src/origin.ts b/js/net/src/origin.ts index bd3611d38d0..48d14b0bcaf 100644 --- a/js/net/src/origin.ts +++ b/js/net/src/origin.ts @@ -610,6 +610,27 @@ class OriginState { // What backs the public `discovery` getter. sessions = new Signal({ total: 0, discovery: 0 }); + // The prefixes whose initial announcements a session is still delivering, counted. An + // announcement stream overlapping one has not synced yet. + snapshots = new Signal(new Map()); + + /** Hold an initial set in flight under the absolute `prefix` until the returned dispose. */ + addSnapshot(prefix: Path.Valid): Dispose { + this.snapshots.mutate((counts) => { + counts.set(prefix, (counts.get(prefix) ?? 0) + 1); + }); + let landed = false; + return () => { + if (landed) return; + landed = true; + this.snapshots.mutate((counts) => { + const count = (counts.get(prefix) ?? 0) - 1; + if (count > 0) counts.set(prefix, count); + else counts.delete(prefix); + }); + }; + } + // The literal heads readers are watching, counted, as a trie: announcement streams and // announced requests. Sessions ask their peer for announcements under the broadest of // these and nothing else, and hear each change to that covering set as it happens, so a @@ -1128,6 +1149,7 @@ export class Producer implements Table { accepts: (prefix) => this.#scope.overlaps(Path.join(this.#scope.root, prefix)), attach: (discovery) => this.#attach(discovery), expect: () => this.#expect(), + snapshot: (prefix) => this.#state.addSnapshot(this.#scope.prefix(prefix)), get requests() { if (thisProducer.#scope === Scope.all) return thisProducer.#state.requests; thisProducer.#requests ??= new Derived([thisProducer.#state.requests], (requests) => @@ -2032,7 +2054,26 @@ export class Consumer { } active = next; - await Signal.race(this.#state.local, this.#state.advertisedLocal, this.#state.routes, producer.closed); + // Synced once no session is still delivering an initial set this stream could + // list: one under a hidden segment holds only a reader that opts in. + let behind = false; + for (const prefix of this.#state.snapshots.peek().keys()) { + for (const pattern of patterns) { + if (!pattern.overlaps(Path.Pattern.subtree(prefix))) continue; + if (hidden || !hiddenBelow(scopeHead(pattern), prefix)) behind = true; + } + } + if (behind) producer.behind(); + else producer.synced(); + + await Signal.race( + this.#state.local, + this.#state.advertisedLocal, + this.#state.routes, + this.#state.snapshots, + producer.closed, + ); + if (producer.closed.peek() !== undefined) return; } } catch { diff --git a/js/net/src/wire.ts b/js/net/src/wire.ts index e072779c667..1ae5b386c53 100644 --- a/js/net/src/wire.ts +++ b/js/net/src/wire.ts @@ -65,6 +65,12 @@ export interface OriginProducer { // pending: /quest/m1/public-model/answer.md expect(): Dispose; // pending: /quest/m1/public-model/answer.md + /** + * Hold the origin's announcement streams overlapping `prefix` unsynced until + * the returned dispose: a session holds one while its peer's initial set is in flight. + */ + snapshot(prefix: Path.Valid): Dispose; + // pending: /quest/m1/public-model/answer.md readonly requests: Getter | undefined>; // pending: /quest/m1/public-model/answer.md changed(): GetPromise; diff --git a/quest/m1/README.md b/quest/m1/README.md index b66b5446a41..2981e184933 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -70,7 +70,6 @@ lands before the cut. - [More tests under load](/quest/m1/test-flakes-2/README.md) - the second round of load-only failures, one quest per flake, fixed at the cause - [A public model](/quest/m1/public-model/README.md) - every model item a built-in session reaches is public, unnecessary, or marked deliberately private, so a custom publisher, subscriber, or relay builds from the public model alone - [Lazy discovery](/quest/m1/lazy-discovery/README.md) - a `consume` session requests announcements only for prefixes the origin is watching, in JS then Rust; `follow`, `broadcasts`, and the `discovery` option go away -- [Announce caught up](/quest/m1/announce-live.md) - Rust and JS announce consumers expose when a session's initial announce snapshot has arrived, from the counts the wire already carries where a version has them - [A spinning loop fails a sim test](/quest/m1/spinning-loops.md) - a sim test names any loop that holds a task poll too long, and each one yields through a budget - [Data consumer timestamps](/quest/m1/data-consumer-timestamps.md) - json and binary consumers return each value's timestamp, in Rust and every binding; snapshots add `latest()` beside an in-order `next()` - [JSON window timestamps](/quest/m1/json-window-timestamps.md) - Rust and JS window consumers return each event's frame timestamp like the snapshot and stream consumers diff --git a/quest/m1/announce-live.md b/quest/m1/announce-live.md deleted file mode 100644 index 9b04cfc8421..00000000000 --- a/quest/m1/announce-live.md +++ /dev/null @@ -1,103 +0,0 @@ -# [M] An announce consumer knows when it has caught up - -## Goal - -The Rust and JS announce consumers expose when a session's initial announce -snapshot has arrived. A consumer can tell "not announced" from "not heard -yet": a player shows offline instead of loading, and lazy discovery can hand -a watch from a broad announce request to a narrower one only once the -narrower one has caught up. - -Non-goals: global completeness across a cluster, and any new wire format. -Every version that can carry the boundary already does. - -## Plan - -Raised 2026-10-13 on [#5466](https://github.com/moq-dev/moq/pull/5466). -Hand-over between a broad and a narrower announce request had two gaps when -the narrower request answered slowly, so #5466 keeps the broad request open -while anything beneath it is watched. The maintainer suggested this signal -as what would make a narrower hand-over safe. - -The boundary is already on the wire, and the subscribers drop it: - -- lite-01/02: the initial set is the single ANNOUNCE_INIT message. -- lite-05+, including lite-07: ANNOUNCE_OK's `Active Count` is the number of - ANNOUNCE_START messages in the initial set. Publishers fill it from their - current view; the Rust lite subscriber reads and drops it. -- IETF: the MoQ Active Count extension - (`drafts/draft-lcurley-moq-active-count.md`) puts the count on the - REQUEST_OK answering SUBSCRIBE_NAMESPACE. The Rust IETF subscriber checks - it and drops it. JS only decodes the parameter: it neither declares the - setup option nor sends the count. -- lite-03/04, and IETF without the extension (always on draft-14/15): no - boundary. lite-03/04 have no OK either, and an unscoped draft-14/15 peer - without MoQ Solicit is sent no request at all. - -Decided with the maintainer, 2026-10-13: - -- ✅ The boundary plus a public API. Rejected: wire only; not now. The - maintainer chose a new lite-07 marker before the existing counts were - found; this plan reuses them instead. -- ✅ m1. Rejected: m2. -- ✅ A relay reports caught up once it has sent its own current view of the - prefix. It never waits for its upstream peers to catch up, so caught up - means "caught up with the sender", not global completeness. Rejected: - waiting for upstream markers. ANNOUNCE_OK's count already has this meaning. -- ✅ Where the session has no boundary, it reports caught up on the OK, as a - best effort. Document the gap. Rejected: never caught up, which hangs - consumers that wait on it. With MoQ Active Count negotiated, IETF waits - for the counted NAMESPACE messages like lite does. Where there is no OK - to wait on (lite-03/04, the unsolicited draft-14/15 case), it reports - caught up once the request is sent, or at once when none is. -- Open, for the implementer to bring back with a recommendation: the API - shape. The maintainer found an `Event::Live` enum variant ugly. The - candidate is separate `poll_next` (route events only) and `poll_live` - (ready once, then stays ready), with JS mirroring as `announced.live`. - To avoid a race, the consumer is caught up only after it has drained - every snapshot event before the boundary. Name the method by role. - The shape must also say what caught up means for a consumer that spans - sources (a reconnecting or pooled connection, an origin fed by several - sessions or prefix requests): reset per source generation, or wait for - every current source. "Ready once, then stays ready" is wrong after a - reconnect. Readiness is a watermark over the coalesced queue, since an - unseen start and end cancel out, and the origin keeps it so a consumer - created after the boundary is caught up once it drains its replay. - -Work: - -- Rust `rs/moq-net` and JS `js/net`: the lite and IETF subscribers keep the - count and mark the point where the initial set has been read, and the - origin's announce consumers expose it. A counted snapshot cut short, by - another message or the stream ending, is a protocol violation. Mirror - names across languages. -- JS MoQ Active Count: declare the setup option, record the negotiation, - validate the parameter against it (and refuse it on any REQUEST_OK but - SUBSCRIBE_NAMESPACE's), and send the count as a publisher, so JS IETF - sessions get the exact boundary too. -- `doc/concept`: what caught up means to a subscriber, including the - best-effort versions. -- A benchmark sweeping announce streams and announce consumers, so - propagating the boundary costs the touched path, not the table. -- Use it in lazy discovery: once both halves have landed, a narrower - request may take over from a broad one after it reports caught up. Only - an exact boundary qualifies; a best-effort session keeps the broad - request. Do that here or in a follow-up, as the implementer judges. -- Regressions with mocked time: caught up after the counted snapshot and - never before an earlier event is read; a count of 0 is caught up at once; - a snapshot cut short is refused; a start and end that cancel before the - poll still reach caught up; a consumer created after the boundary is - caught up after its replay; - a relay reports caught up without waiting on a silent upstream; IETF - waits for the counted NAMESPACE messages, or reports on the OK without - the extension; a session with no OK still reports caught up; a reconnect - and a second feed follow the chosen multi-source rule. Run - `just test interop --all`. - -Public API: new caught-up surface on Rust and JS announce consumers. Wire: -no new format or draft change; JS adopts the existing MoQ Active Count -extension. - -## Related - -- [Lazy discovery](/quest/m1/lazy-discovery/README.md) - its broad-request hold is what this lets a narrower request replace diff --git a/quest/m1/lazy-discovery/rust.md b/quest/m1/lazy-discovery/rust.md index f5caece745c..87e248b5c22 100644 --- a/quest/m1/lazy-discovery/rust.md +++ b/quest/m1/lazy-discovery/rust.md @@ -33,7 +33,7 @@ covering-set changes a watch causes, so a change costs the prefixes it covers or uncovers and stays flat over unrelated held watches (`js/net/bench/forward.ts`); an announce request stays open while anything at or beneath its prefix is watched, so a narrower watch rides it and never takes -its routes over (nothing tells when a fresh request has caught up); a narrower +its routes over (nothing tells when a fresh request has synced); a narrower request already open when a broader watch arrives stays open until the broader request goes; a route reported by overlapping requests is inserted once, driven by the first reporter, and if it changes hands anyway (a request diff --git a/rs/moq-net/Cargo.toml b/rs/moq-net/Cargo.toml index 29c155112d2..817d74e89bd 100644 --- a/rs/moq-net/Cargo.toml +++ b/rs/moq-net/Cargo.toml @@ -95,6 +95,12 @@ name = "announce" harness = false required-features = ["fuzz"] +# Reaches the crate-private session snapshot through the hidden `fuzz` module. +[[bench]] +name = "synced" +harness = false +required-features = ["fuzz"] + # Reaches the private message codecs through the hidden `fuzz` module. [[bench]] name = "codec" diff --git a/rs/moq-net/benches/synced.rs b/rs/moq-net/benches/synced.rs new file mode 100644 index 00000000000..a648e25131e --- /dev/null +++ b/rs/moq-net/benches/synced.rs @@ -0,0 +1,90 @@ +//! What it costs an origin to track whether its announce consumers have synced. +//! +//! Every session requesting announcements holds a snapshot on the origin until its +//! peer's initial set is in, and the announce consumers its prefix overlaps wait on +//! it. Both shapes are swept over the snapshots in flight (announce streams) and the +//! consumers watching elsewhere, so a cost that grows with the table rather than the +//! touched path shows up as a slope. +//! +//! - land: a snapshot starts and lands under one prefix, releasing its one consumer. +//! - register: a consumer of one prefix registers, counting the snapshots it waits on. +//! +//! The snapshot is crate-private, so this reaches it through the hidden `fuzz` module. +//! +//! Run with `cargo bench -p moq-net --features fuzz --bench synced`. + +use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main}; +use moq_net::{Hop, Pattern, Patterns, announce, fuzz, origin}; + +/// `(streams, consumers)` shapes. +const SHAPES: [(usize, usize); 4] = [(10, 10), (1_000, 10), (10, 1_000), (1_000, 1_000)]; + +/// An origin with `streams` snapshots in flight under `room/` and `consumers` cursors +/// on prefixes of their own under `other/`. The handles are held for the whole run. +struct Table { + producer: origin::Producer, + _driver: origin::Driver, + _streams: Vec, + _consumers: Vec, +} + +fn table(streams: usize, consumers: usize) -> Table { + let (producer, driver) = origin::Producer::new(origin::Config::new(Hop::new(1).unwrap())); + let streams = (0..streams) + .map(|i| fuzz::snapshot(&producer, &format!("room/{i}"))) + .collect(); + let consumers = (0..consumers) + .map(|i| scoped(&producer, &format!("other/{i}")).announced()) + .collect(); + Table { + producer, + _driver: driver, + _streams: streams, + _consumers: consumers, + } +} + +fn scoped(producer: &origin::Producer, prefix: &str) -> origin::Consumer { + producer + .consume() + .scope(prefix, &Patterns::from(Pattern::all())) + .unwrap() +} + +fn bench_land(c: &mut Criterion) { + let mut group = c.benchmark_group("synced/land"); + for (streams, consumers) in SHAPES { + let id = BenchmarkId::from_parameter(format!("{streams}st_{consumers}c")); + group.bench_function(id, |b| { + let table = table(streams, consumers); + let watcher = scoped(&table.producer, "room/incoming").announced(); + b.iter(|| { + let snapshot = fuzz::snapshot(&table.producer, "room/incoming"); + assert!(!watcher.is_synced()); + drop(snapshot); + assert!(watcher.is_synced()); + }); + }); + } + group.finish(); +} + +fn bench_register(c: &mut Criterion) { + let mut group = c.benchmark_group("synced/register"); + for (streams, consumers) in SHAPES { + let id = BenchmarkId::from_parameter(format!("{streams}st_{consumers}c")); + group.bench_function(id, |b| { + let table = table(streams, consumers); + let incoming = scoped(&table.producer, "room/incoming"); + let _held = fuzz::snapshot(&table.producer, "room/incoming"); + b.iter(|| { + let watcher = incoming.announced(); + assert!(!watcher.is_synced()); + }); + }); + } + group.finish(); +} + +criterion_group!(benches, bench_land, bench_register); +criterion_main!(benches); diff --git a/rs/moq-net/src/fuzz.rs b/rs/moq-net/src/fuzz.rs index 8b9ff9f52f9..c7b1b8ed26c 100644 --- a/rs/moq-net/src/fuzz.rs +++ b/rs/moq-net/src/fuzz.rs @@ -268,6 +268,15 @@ pub fn excluding(origin: crate::origin::Consumer, peer: crate::Hop) -> crate::or origin.excluding(peer) } +/// A session's snapshot of its peer's initial announcements, for the synced bench. +/// Drop it to land the snapshot. +pub struct Snapshot(#[allow(dead_code)] crate::model::Snapshot); + +/// Start a snapshot under `prefix`, as a session requesting announcements there does. +pub fn snapshot(origin: &crate::origin::Producer, prefix: &str) -> Snapshot { + Snapshot(origin.snapshot(prefix)) +} + /// Add `n` publisher-side payload bytes on `(path, tier)`, creating the entry /// when the path is tracked. The stats producer bench marks a share of held /// paths changed this way, without a media frame. A later [`crate::stats::Registry::report`] diff --git a/rs/moq-net/src/ietf/active_count.rs b/rs/moq-net/src/ietf/active_count.rs index 37138f4fb65..24b55d1a096 100644 --- a/rs/moq-net/src/ietf/active_count.rs +++ b/rs/moq-net/src/ietf/active_count.rs @@ -4,16 +4,16 @@ //! advertises, then reports changes on the same stream. Nothing in moq-transport says //! where the first part ends, so a subscriber can only guess from the stream going //! quiet. When both endpoints declare the ACTIVE_COUNT option, the publisher puts on its -//! REQUEST_OK how many NAMESPACE messages come before the subscription is caught up. +//! REQUEST_OK how many NAMESPACE messages come before the subscription is synced. use super::Version; /// ACTIVE_COUNT Setup Option: the sender counts the NAMESPACE messages before a -/// subscription is caught up, and reads the peer's count. Even, so the value is a bare varint. +/// subscription is synced, and reads the peer's count. Even, so the value is a bare varint. pub const ACTIVE_COUNT: u64 = 0x40B64; /// ACTIVE_COUNT REQUEST_OK parameter: how many NAMESPACE messages immediately follow -/// before the subscription is caught up. Only answers SUBSCRIBE_NAMESPACE. +/// before the subscription is synced. Only answers SUBSCRIBE_NAMESPACE. /// Even, so the value is a bare varint. pub const ACTIVE_COUNT_PARAM: u64 = 0x40B66; diff --git a/rs/moq-net/src/ietf/peer.rs b/rs/moq-net/src/ietf/peer.rs index fcf4014369c..3f0e920789c 100644 --- a/rs/moq-net/src/ietf/peer.rs +++ b/rs/moq-net/src/ietf/peer.rs @@ -24,7 +24,7 @@ pub(crate) struct Peer { pub auth: bool, /// MoQ Active Count: whether the REQUEST_OK answering SUBSCRIBE_NAMESPACE counts the - /// NAMESPACE messages before the subscription is caught up, in both directions. + /// NAMESPACE messages before the subscription is synced, in both directions. pub active_count: bool, } diff --git a/rs/moq-net/src/ietf/request.rs b/rs/moq-net/src/ietf/request.rs index ea970b7df62..b5d461869dd 100644 --- a/rs/moq-net/src/ietf/request.rs +++ b/rs/moq-net/src/ietf/request.rs @@ -87,7 +87,7 @@ impl Message for RequestsBlocked { pub struct RequestOk { pub request_id: Option, /// MoQ Active Count: how many NAMESPACE messages follow before the subscription is - /// caught up. Only on the answer to a SUBSCRIBE_NAMESPACE, when negotiated. + /// synced. Only on the answer to a SUBSCRIBE_NAMESPACE, when negotiated. pub active: Option, } diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index c04e850beef..633e149790b 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -155,7 +155,17 @@ where // One SUBSCRIBE_NAMESPACE per permitted prefix, like `lite::Subscriber`: the // scope is what we may ask for, and it is not the origin's root. - let namespaces = subscribe.as_ref().map(subscribe_prefixes).unwrap_or_default(); + // Each holds a snapshot from now, so the origin's announce consumers never read as + // synced before the request goes out. + let namespaces: Vec<_> = subscribe + .as_ref() + .map(|origin| { + subscribe_prefixes(origin) + .into_iter() + .map(|prefix| (origin.snapshot(&prefix), prefix)) + .collect() + }) + .unwrap_or_default(); let withdrawal = crate::session::Withdrawal::default(); let withdrawing = withdrawal.clone(); @@ -311,7 +321,7 @@ where let mut pub_ns_run = std::pin::pin!(err_only(publisher.clone().run_publish_namespaces())); let mut sub_ns_run = std::pin::pin!(err_only(async { let mut prefixes = futures::stream::FuturesUnordered::new(); - for prefix in namespaces { + for (snapshot, prefix) in namespaces { let mut sub_ns = sub_ns.clone(); let sub_ns_adapter = sub_ns_adapter.clone(); prefixes.push(async move { @@ -326,7 +336,7 @@ where } _ => Stream::open(&mut sub_ns_adapter.clone(), version).await?, }; - if let Err(err) = sub_ns.run_subscribe_namespace(stream, prefix).await { + if let Err(err) = sub_ns.run_subscribe_namespace(stream, prefix, snapshot).await { // The peer breaking the protocol is fatal, and the driver // below turns this into the session close the draft wants. if is_protocol_violation(&err) { @@ -525,13 +535,13 @@ where let mut pub_ns_run = std::pin::pin!(err_only(publisher.clone().run_publish_namespaces())); let mut sub_ns_run = std::pin::pin!(err_only(async { let mut prefixes = futures::stream::FuturesUnordered::new(); - for prefix in namespaces { + for (snapshot, prefix) in namespaces { let mut sub_ns = sub_ns.clone(); let sub_ns_session = sub_ns_session.clone(); prefixes.push(async move { let mut sub_ns_session = sub_ns_session; let stream = Stream::open(&mut sub_ns_session, version).await?; - if let Err(err) = sub_ns.run_subscribe_namespace(stream, prefix).await { + if let Err(err) = sub_ns.run_subscribe_namespace(stream, prefix, snapshot).await { // The peer breaking the protocol is fatal, and the driver // below turns this into the session close the draft wants. if is_protocol_violation(&err) { diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index d237a236743..99bbd8c736d 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -715,6 +715,14 @@ pub(super) fn subscribe_prefixes(origin: &origin::Producer) -> Vec { crate::model::interest_prefixes(&origin.allowed()) } +/// The initial set of a SUBSCRIBE_NAMESPACE still to read, under MoQ Active Count. +struct Initial { + /// NAMESPACE messages still owed before the first change. + initial: u64, + /// Released once `initial` reaches zero. + snapshot: Option, +} + /// Resolve the subscription a data stream belongs to. /// /// SUBSCRIBE_OK can be reordered behind the stream it describes, so an alias we have not @@ -1006,10 +1014,15 @@ where /// A failure here is per-prefix, so the caller decides what it means for the /// session: [`is_protocol_violation`] separates the peer's fault (fatal) from a /// stream of ours that simply died (survivable). + /// + /// `snapshot` holds the origin's announce consumers under the prefix short of + /// synced. It drops once the NAMESPACE messages MoQ Active Count promised are + /// in, on REQUEST_OK without the extension, or when no request goes out at all. pub async fn run_subscribe_namespace( &mut self, mut stream: Stream, prefix: PathOwned, + snapshot: crate::model::Snapshot, ) -> Result<(), Error> { // Hidden namespaces are requested too, as on moq-lite: the session mirrors the // peer into the origin and each local reader opts in on its own. The parameter @@ -1086,13 +1099,14 @@ where tracing::debug!(%prefix, ?count, "subscribe_namespace ok"); - // MoQ Active Count says how many NAMESPACE messages make up the initial set; - // nothing here needs that boundary, so it is only checked. A count the - // negotiation did not promise, or a missing one it did, is the peer breaking - // the extension. + // MoQ Active Count says how many NAMESPACE messages make up the initial set. A + // count the negotiation did not promise, or a missing one it did, is the peer + // breaking the extension. Without it, the answer is the best boundary there is. if declared.active_count != count.is_some() { return Err(Error::ProtocolViolation); } + let initial = count.unwrap_or(0); + let snapshot = (initial > 0).then_some(snapshot); // The extension changes the NAMESPACE encoding, so we can't parse one until // the peer's SETUP says whether it negotiated. @@ -1108,7 +1122,9 @@ where // // This is what moq-lite already does, where the equivalent map is a local whose // guards drop. - let res = self.run_namespace_entries(&mut stream, &prefix, &peer, &mut live).await; + let res = self + .run_namespace_entries(&mut stream, &prefix, &peer, &mut live, Initial { initial, snapshot }) + .await; for path in live { let _ = self.stop_announce(path, false); } @@ -1126,11 +1142,18 @@ where prefix: &PathOwned, peer: &cluster::Peer, live: &mut std::collections::HashSet, + mut initial: Initial, ) -> Result<(), Error> { loop { + // Every counted NAMESPACE is in the origin by now, dropped ones included. + if initial.initial == 0 { + initial.snapshot = None; + } let type_id = match stream.reader.varint_maybe().await? { Some(id) => id, - None => break, // Stream closed + // A stream ending inside the counted set cut it short. + None if initial.initial > 0 => return Err(Error::ProtocolViolation), + None => break, }; let body: ietf::Body = stream.reader.decode().await?; let mut data = body.decoder(self.version); @@ -1144,6 +1167,7 @@ where if !data.is_empty() { return Err(Error::WrongSize); } + initial.initial = initial.initial.saturating_sub(1); let path = prefix.join(&msg.suffix); let Some(advert) = self.route(msg.cluster.as_ref(), peer) else { // Looped back through us: forwarding it would extend the loop and @@ -1179,6 +1203,8 @@ where } } } + // Only NAMESPACE may come before the counted set is in. + ietf::NamespaceDone::ID if initial.initial > 0 => return Err(Error::ProtocolViolation), ietf::NamespaceDone::ID => { let msg = ietf::NamespaceDone::decode_msg(&mut data, self.version)?; let path = prefix.join(&msg.suffix); @@ -5673,7 +5699,8 @@ mod tests { let prefix = namespaces.pop().unwrap(); let stream = Stream::open(&mut session.clone(), Version::Draft16).await.unwrap(); - let mut run = std::pin::pin!(subscriber.run_subscribe_namespace(stream, prefix)); + let mut run = + std::pin::pin!(subscriber.run_subscribe_namespace(stream, prefix, crate::model::Snapshot::landed())); // Parks awaiting the peer's response; the request is already on the wire. assert!(futures::poll!(run.as_mut()).is_pending()); @@ -5747,7 +5774,8 @@ mod tests { let prefix = subscribe_prefixes(&subscriber.origin).pop().expect("one prefix"); let stream = Stream::open(&mut session.clone(), VERSION).await.unwrap(); // Parks on the read after the scripted NAMESPACE is consumed. - let mut run = std::pin::pin!(subscriber.run_subscribe_namespace(stream, prefix)); + let mut run = + std::pin::pin!(subscriber.run_subscribe_namespace(stream, prefix, crate::model::Snapshot::landed())); for _ in 0..100 { // The result is deliberately ignored: a regressed mount lands out of scope // and errors here, which the assertions below name far better than a poll @@ -5814,7 +5842,11 @@ mod tests { let stream = Stream::open(&mut session.clone(), VERSION).await.unwrap(); let res = subscriber - .run_subscribe_namespace(stream, crate::Path::new("").to_owned()) + .run_subscribe_namespace( + stream, + crate::Path::new("").to_owned(), + crate::model::Snapshot::landed(), + ) .await; assert!( matches!(res, Err(Error::ProtocolViolation)), @@ -5823,6 +5855,142 @@ mod tests { } } + /// The peer's REQUEST_OK carrying `active`, then a NAMESPACE per suffix, or a + /// NAMESPACE_DONE for one prefixed with `-`. + async fn counted_response(version: Version, active: Option, entries: &[&str]) -> Vec { + let log = crate::lite::test_transport::Log::default(); + let mut writer = crate::coding::Writer::new(crate::lite::test_transport::SinkSend::new(log.clone()), version); + writer.varint(ietf::RequestOk::ID).await.unwrap(); + writer + .encode(&ietf::RequestOk { + request_id: None, + active, + }) + .await + .unwrap(); + for entry in entries { + match entry.strip_prefix('-') { + Some(suffix) => { + writer.varint(ietf::NamespaceDone::ID).await.unwrap(); + writer + .encode(&ietf::NamespaceDone { + suffix: crate::Path::new(suffix), + }) + .await + .unwrap(); + } + None => { + writer.varint(ietf::Namespace::ID).await.unwrap(); + writer + .encode(&ietf::Namespace { + suffix: crate::Path::new(entry), + cluster: None, + }) + .await + .unwrap(); + } + } + } + let writes = log.writes.lock().unwrap(); + writes.clone() + } + + /// A root SUBSCRIBE_NAMESPACE on a fresh origin whose peer replays `session`, holding + /// the origin's snapshot as the session does. + fn counted_subscriber( + version: Version, + active_count: bool, + session: crate::lite::test_transport::ScriptedSession, + ) -> ( + Subscriber, + crate::origin::Consumer, + crate::model::Snapshot, + ) { + let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); + let consumer = origin.consume(); + let snapshot = origin.snapshot(""); + let (tasks, _task_set) = crate::util::TaskSet::new(); + let peer_setup = peer::PeerSetup::default(); + peer_setup.set(peer::Peer { + active_count, + ..Default::default() + }); + let subscriber = Subscriber::new( + crate::time::sim(), + session, + origin, + Control::new(None, false), + None, + peer_setup, + crate::Hop::new(1).unwrap(), + None, + version, + tasks, + Default::default(), + ); + (subscriber, consumer, snapshot) + } + + /// With MoQ Active Count, the origin syncs once the counted NAMESPACEs are in. + #[moq_clock::test] + async fn counted_namespaces_catch_up() { + const VERSION: Version = Version::Draft18; + for (entries, synced) in [(&["a"][..], false), (&["a", "b"][..], true)] { + let session = + crate::lite::test_transport::ScriptedSession::new(counted_response(VERSION, Some(2), entries).await); + let (mut subscriber, consumer, snapshot) = counted_subscriber(VERSION, true, session.clone()); + let mut cursor = consumer.announced(); + let stream = Stream::open(&mut session.clone(), VERSION).await.unwrap(); + let mut run = + std::pin::pin!(subscriber.run_subscribe_namespace(stream, crate::Path::new("").to_owned(), snapshot)); + for _ in 0..10 { + assert!(futures::poll!(run.as_mut()).is_pending()); + settle().await; + } + for entry in entries { + cursor.assert_next_active(*entry); + } + assert_eq!(cursor.is_synced(), synced, "{entries:?}"); + } + } + + /// Without MoQ Active Count, REQUEST_OK is the best boundary there is. + #[moq_clock::test] + async fn an_uncounted_answer_catches_up_on_the_ok() { + const VERSION: Version = Version::Draft18; + let session = crate::lite::test_transport::ScriptedSession::new(counted_response(VERSION, None, &[]).await); + let (mut subscriber, consumer, snapshot) = counted_subscriber(VERSION, false, session.clone()); + let cursor = consumer.announced(); + assert!(!cursor.is_synced()); + let stream = Stream::open(&mut session.clone(), VERSION).await.unwrap(); + let mut run = + std::pin::pin!(subscriber.run_subscribe_namespace(stream, crate::Path::new("").to_owned(), snapshot)); + for _ in 0..10 { + assert!(futures::poll!(run.as_mut()).is_pending()); + settle().await; + } + assert!(cursor.is_synced()); + } + + /// A counted set cut short, by a NAMESPACE_DONE or the stream ending, is the peer's fault. + #[moq_clock::test] + async fn a_counted_set_cut_short_is_a_violation() { + const VERSION: Version = Version::Draft18; + for (entries, fin) in [(&["a", "-a"][..], false), (&["a"][..], true)] { + let session = + crate::lite::test_transport::ScriptedSession::new(counted_response(VERSION, Some(2), entries).await); + if fin { + session.close(crate::lite::test_transport::Close::Fin); + } + let (mut subscriber, _consumer, snapshot) = counted_subscriber(VERSION, true, session.clone()); + let stream = Stream::open(&mut session.clone(), VERSION).await.unwrap(); + let res = subscriber + .run_subscribe_namespace(stream, crate::Path::new("").to_owned(), snapshot) + .await; + assert!(matches!(res, Err(Error::ProtocolViolation)), "{entries:?}: {res:?}"); + } + } + #[test] fn retiring_old_track_does_not_retire_reused_alias() { let aliases = TrackAliases::default(); @@ -7375,7 +7543,11 @@ mod tests { let stream = Stream::open(&mut session.clone(), VERSION).await.unwrap(); subscriber - .run_subscribe_namespace(stream, crate::Path::new("").to_owned()) + .run_subscribe_namespace( + stream, + crate::Path::new("").to_owned(), + crate::model::Snapshot::landed(), + ) .await .expect("a clean FIN is not an error"); settle().await; @@ -7451,7 +7623,11 @@ mod tests { ); let stream = Stream::open(&mut session.clone(), VERSION).await.unwrap(); - let mut run = std::pin::pin!(subscriber.run_subscribe_namespace(stream, crate::Path::new("").to_owned())); + let mut run = std::pin::pin!(subscriber.run_subscribe_namespace( + stream, + crate::Path::new("").to_owned(), + crate::model::Snapshot::landed() + )); for _ in 0..100 { assert!( futures::poll!(run.as_mut()).is_pending(), diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 292d49d18c8..741db1edae5 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -2592,6 +2592,39 @@ mod announce_test { harness_with(version, None).await } + /// A relay answers with what it holds now: an upstream still delivering its own + /// initial set never delays the ANNOUNCE_OK, so a peer syncs with this relay. + #[moq_clock::test] + async fn a_silent_upstream_never_delays_the_answer() { + let origin = Hop::new(1).unwrap().produce(); + let _route = origin.announce("cam", crate::origin::Route::default()).unwrap(); + // An upstream that never answers its announce request. + let _silent = origin.clone().upstream().snapshot(""); + + let log = Log::default(); + let writes = log.writes.clone(); + let consumer = origin.consume(); + assert!(!consumer.announced().is_synced(), "the relay itself is still syncing"); + let mut stream = Stream:: { + writer: Writer::new(SinkSend::new(log), VERSION), + reader: Reader::new(PendingRecv, VERSION), + }; + let _task = moq_clock::sim::spawn(async move { + let mut announced = consumer.announced(); + let self_origin = consumer.hop(); + TestPublisher::run_announce(&mut stream, &consumer, &mut announced, self_origin, VERSION, None).await + }); + settle().await; + + let mut wire = Wire { + writes, + cursor: 0, + version: VERSION, + }; + assert_eq!(wire.take_ok().active, 1, "answered with the current view"); + assert_eq!(wire.take_announces().len(), 1); + } + /// [`harness_on`], announcing under `auth`'s grant. async fn harness_with(version: Version, auth: Option) -> Harness { let origin = Hop::new(1).unwrap().produce(); diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 5f59f20715d..7d5f26368cc 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -1285,6 +1285,10 @@ struct AnnouncePrefix { subscriber: Subscriber, prefix: PathOwned, state: PrefixState, + /// Holds the origin's announce consumers under the prefix unsynced + /// until the peer's initial set is in. Dropped at the boundary the version + /// marks, or with the stream. + snapshot: Option, } enum PrefixState { @@ -1298,6 +1302,8 @@ enum PrefixState { Cost { stream: Stream, responder_origin: Option, + /// Lite05+: the ANNOUNCE_STARTs the publisher counted into its initial set. + initial: u64, }, /// Lite01/02: reading the ANNOUNCE_INIT set. ReadInit { stream: Stream, run: PrefixRun }, @@ -1320,14 +1326,21 @@ struct PrefixRun { decoder: lite::AnnounceDecoder, /// The auth epoch last applied, so a new limit holds back or brings back routes. auth_epoch: u64, + /// Lite05+: the ANNOUNCE_STARTs of the initial set still to read. Anything else + /// before them, or the stream ending, cuts the counted set short. + initial: u64, } impl AnnouncePrefix { fn new(subscriber: Subscriber, prefix: PathOwned) -> Self { + // Taken now, while the session is being set up, so a consumer of the origin never + // reads as synced before the request even goes out. + let snapshot = Some(subscriber.origin.snapshot(&prefix)); Self { subscriber, prefix, state: PrefixState::Open, + snapshot, } } @@ -1369,20 +1382,29 @@ impl AnnouncePrefix { let PrefixState::Send { stream } = std::mem::replace(&mut self.state, PrefixState::Open) else { unreachable!() }; - self.state = match self.subscriber.version.has_announce_ok() { + let version = self.subscriber.version; + // Lite03/04 mark no initial set, so the request going out is the best + // boundary. Lite01/02 mark it with ANNOUNCE_INIT, newer ones with a count. + if !version.has_announce_ok() && !matches!(version, Version::Lite01 | Version::Lite02) { + self.snapshot = None; + } + self.state = match version.has_announce_ok() { true => PrefixState::ReadOk { stream }, false => PrefixState::Cost { stream, responder_origin: None, + initial: 0, }, }; } PrefixState::ReadOk { stream } => { // Lite05+: the publisher reports its own origin id, which we stamp onto // every received Announce's hop chain since it no longer does so itself. - // Its `active` count marks where the initial set ends; nothing here needs - // that boundary, so it is read and dropped. + // Its `active` count marks where the initial set ends. let ok = ready!(stream.reader.poll_decode::(&mut cx))?; + if ok.active == 0 { + self.snapshot = None; + } // A peer may legally report id 0 (no identity). Keep it: the assigned // identity stays on `via` and is never forwarded as a hop. let origin = ok.origin; @@ -1392,6 +1414,7 @@ impl AnnouncePrefix { self.state = PrefixState::Cost { stream, responder_origin: Some(origin), + initial: ok.active, }; } PrefixState::Cost { .. } => { @@ -1399,6 +1422,7 @@ impl AnnouncePrefix { let PrefixState::Cost { stream, responder_origin, + initial, } = std::mem::replace(&mut self.state, PrefixState::Open) else { unreachable!() @@ -1410,6 +1434,7 @@ impl AnnouncePrefix { announced: Announced::new(self.subscriber.announces.clone()), decoder: lite::AnnounceDecoder::default(), auth_epoch: 0, + initial, }; // Lite01/02 send the initial set as one ANNOUNCE_INIT message, so they @@ -1441,6 +1466,7 @@ impl AnnouncePrefix { else { unreachable!() }; + self.snapshot = None; self.state = PrefixState::Run { stream, run }; } PrefixState::Run { stream, run } => { @@ -1464,7 +1490,22 @@ impl AnnouncePrefix { loop { match stream.reader.poll_decode_maybe::(&mut cx) { Poll::Ready(Ok(Some(announce))) => { + if run.initial > 0 { + match announce { + lite::AnnounceBroadcast::Active { .. } => run.initial -= 1, + // An unknown type is no change, so it neither counts nor cuts. + lite::AnnounceBroadcast::Skipped => {} + _ => return Poll::Ready(Err(Error::ProtocolViolation)), + } + } self.subscriber.handle_announce(&self.prefix, announce, run)?; + // Released only once the last counted route is in the origin. + if run.initial == 0 { + self.snapshot = None; + } + } + Poll::Ready(Ok(None)) if run.initial > 0 => { + return Poll::Ready(Err(Error::ProtocolViolation)); } Poll::Ready(Ok(None)) => { // The publisher FINed: it has nothing (more) to announce for this @@ -4175,6 +4216,185 @@ mod tests { cursor.expect_next_active("b").await; } + /// A peer's announce stream for the root, encoded for `version`. + fn announce_script(version: Version, messages: &[&dyn Fn(&mut crate::coding::Encoder<'_>)]) -> Vec { + let mut script = Vec::new(); + for message in messages { + message(&mut crate::coding::Encoder::new(&mut script, version.into())); + } + script + } + + /// An ANNOUNCE_OK counting `active` starts into the initial set. + fn announce_ok(version: Version, active: u64) -> impl Fn(&mut crate::coding::Encoder<'_>) { + move |w| { + lite::AnnounceOk { + origin: crate::Hop::new(9).unwrap(), + active, + } + .encode(w, version) + .unwrap() + } + } + + /// An ANNOUNCE_START for `suffix`. + fn announce_start(version: Version, suffix: &'static str) -> impl Fn(&mut crate::coding::Encoder<'_>) { + move |w| { + lite::AnnounceBroadcast::Active { + epoch: None, + suffix: lite::PathRef::literal(Path::new(suffix)), + hops: lite::HopsRef::literal(crate::Hops::new()), + cost: crate::origin::Cost::default(), + } + .encode(w, version) + .unwrap() + } + } + + /// The announce stream for the root of a fresh origin, its peer replaying `session`. + fn root_prefix( + version: Version, + session: crate::lite::test_transport::ScriptedSession, + ) -> ( + AnnouncePrefix, + origin::Consumer, + ) { + let origin = origin::Config::new(crate::Hop::new(1).unwrap()).produce(); + let consumer = origin.consume(); + let subscriber = Subscriber::new(SubscriberConfig { + runtime: crate::time::sim(), + session, + origin, + recv_bandwidth: None, + version, + peer_setup: crate::lite::PeerSetup::new(false, crate::transport::Binding::InBand), + cost: Some(1), + peer_hop: None, + going_away: Default::default(), + auth: crate::auth::Handle::new(false), + }); + (AnnouncePrefix::new(subscriber, Path::new("").to_owned()), consumer) + } + + /// The origin's consumers sync once the counted starts are in and read, never before. + #[moq_clock::test] + async fn a_counted_initial_set_catches_up() { + const VERSION: Version = Version::Lite06; + let script = announce_script( + VERSION, + &[ + &announce_ok(VERSION, 2), + &announce_start(VERSION, "a"), + &announce_start(VERSION, "b"), + ], + ); + let (mut prefix, consumer) = root_prefix(VERSION, crate::lite::test_transport::ScriptedSession::new(script)); + let mut cursor = consumer.announced(); + assert!(!cursor.is_synced(), "held from the request on"); + + let mut run = std::pin::pin!(kio::wait(|waiter| prefix.poll(waiter))); + assert!(futures::poll!(run.as_mut()).is_pending()); + + cursor.assert_next_active("a"); + assert!(!cursor.is_synced(), "never before an earlier event is read"); + cursor.assert_next_active("b"); + assert!(cursor.is_synced()); + // A consumer created after the boundary syncs on its replay. + let mut late = consumer.announced(); + late.assert_next_active("a"); + late.assert_next_active("b"); + assert!(late.is_synced()); + } + + /// Part of a counted set is not synced, however much of it was read. + #[moq_clock::test] + async fn a_partial_initial_set_holds() { + const VERSION: Version = Version::Lite06; + let script = announce_script(VERSION, &[&announce_ok(VERSION, 2), &announce_start(VERSION, "a")]); + let (mut prefix, consumer) = root_prefix(VERSION, crate::lite::test_transport::ScriptedSession::new(script)); + let mut cursor = consumer.announced(); + let mut run = std::pin::pin!(kio::wait(|waiter| prefix.poll(waiter))); + assert!(futures::poll!(run.as_mut()).is_pending()); + + cursor.assert_next_active("a"); + assert!(!cursor.is_synced()); + } + + /// A count of zero is synced on the ANNOUNCE_OK. + #[moq_clock::test] + async fn an_empty_initial_set_catches_up_at_once() { + const VERSION: Version = Version::Lite07; + let script = announce_script(VERSION, &[&announce_ok(VERSION, 0)]); + let (mut prefix, consumer) = root_prefix(VERSION, crate::lite::test_transport::ScriptedSession::new(script)); + let cursor = consumer.announced(); + let mut run = std::pin::pin!(kio::wait(|waiter| prefix.poll(waiter))); + assert!(futures::poll!(run.as_mut()).is_pending()); + assert!(cursor.is_synced()); + } + + /// Anything but a start inside the counted set cuts it short: a protocol violation. + #[moq_clock::test] + async fn an_end_inside_the_initial_set_is_refused() { + const VERSION: Version = Version::Lite06; + let end = + |w: &mut crate::coding::Encoder<'_>| lite::AnnounceBroadcast::EndedId { id: 0 }.encode(w, VERSION).unwrap(); + let script = announce_script( + VERSION, + &[&announce_ok(VERSION, 2), &announce_start(VERSION, "a"), &end], + ); + let (mut prefix, _consumer) = root_prefix(VERSION, crate::lite::test_transport::ScriptedSession::new(script)); + let err = kio::wait(|waiter| prefix.poll_run(waiter)).await.unwrap_err(); + assert!(matches!(err, Error::ProtocolViolation), "{err}"); + } + + /// The stream ending inside the counted set cuts it short too. + #[moq_clock::test] + async fn a_fin_inside_the_initial_set_is_refused() { + const VERSION: Version = Version::Lite06; + let script = announce_script(VERSION, &[&announce_ok(VERSION, 2), &announce_start(VERSION, "a")]); + let session = crate::lite::test_transport::ScriptedSession::new(script); + session.close(crate::lite::test_transport::Close::Fin); + let (mut prefix, _consumer) = root_prefix(VERSION, session); + let err = kio::wait(|waiter| prefix.poll_run(waiter)).await.unwrap_err(); + assert!(matches!(err, Error::ProtocolViolation), "{err}"); + } + + /// Lite03/04 mark no initial set: synced, best effort, once the request is sent. + #[moq_clock::test] + async fn an_unmarked_version_catches_up_on_the_request() { + const VERSION: Version = Version::Lite04; + let (mut prefix, consumer) = + root_prefix(VERSION, crate::lite::test_transport::ScriptedSession::new(Vec::new())); + let cursor = consumer.announced(); + assert!(!cursor.is_synced()); + let mut run = std::pin::pin!(kio::wait(|waiter| prefix.poll(waiter))); + assert!(futures::poll!(run.as_mut()).is_pending()); + assert!(cursor.is_synced()); + } + + /// Lite01/02 mark the initial set with ANNOUNCE_INIT. + #[moq_clock::test] + async fn announce_init_is_the_initial_set() { + const VERSION: Version = Version::Lite02; + let init = |w: &mut crate::coding::Encoder<'_>| { + lite::AnnounceInit { + suffixes: vec![Path::new("a")], + } + .encode(w, VERSION) + .unwrap() + }; + let (mut prefix, consumer) = root_prefix( + VERSION, + crate::lite::test_transport::ScriptedSession::new(announce_script(VERSION, &[&init])), + ); + let mut cursor = consumer.announced(); + let mut run = std::pin::pin!(kio::wait(|waiter| prefix.poll(waiter))); + assert!(futures::poll!(run.as_mut()).is_pending()); + assert!(!cursor.is_synced()); + cursor.assert_next_active("a"); + assert!(cursor.is_synced()); + } + /// An announce past the session's cap fails the session with TOO_MANY_REQUESTS, and /// the stream gives its slots back once the session drops it. #[moq_clock::test] diff --git a/rs/moq-net/src/model/mod.rs b/rs/moq-net/src/model/mod.rs index a1b744f0084..04e7d797958 100644 --- a/rs/moq-net/src/model/mod.rs +++ b/rs/moq-net/src/model/mod.rs @@ -75,6 +75,11 @@ pub use origin_impl::{Consume, Hop, Hops, InvalidHop}; // pending: /quest/m1/public-model/answer.md pub(crate) use origin_impl::interest_prefixes; +// Held by a session while its peer's initial announcements arrive. +#[cfg(feature = "async")] +// pending: /quest/m1/public-model/answer.md +pub(crate) use origin_impl::Snapshot; + // The advertise-only route guard, for tests shaping the route table. #[cfg(all(test, feature = "async"))] // private: test diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 0c0edb27772..a1f7aacc69d 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -5,7 +5,7 @@ use std::{ collections::{BTreeMap, BTreeSet, HashMap, HashSet, VecDeque}, fmt, sync::Arc, - sync::atomic::{AtomicBool, AtomicU64, Ordering}, + sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}, task::{Context, Poll, Wake, Waker, ready}, time::Duration, }; @@ -821,9 +821,43 @@ struct OriginConsumerState { /// Set by the origin's teardown: the cursor drains `pending`, then reports /// the end instead of parking forever on a table that can never fire again. ended: bool, + /// Snapshots in flight that can present on this consumer's cursors, counted + /// once per cursor they touch. See [`AnnounceConsumer::poll_synced`]. + /// + /// Changed through a read guard, so a session joining wakes none of the cursors it + /// holds back (nothing waits on losing sync), except the last landing, which + /// takes the write guard to owe the queue too. Only the table lock changes it. + waiting: AtomicUsize, + /// The pending prefixes the consumer still owes reading from before the last + /// snapshot landed. Synced once empty and nothing is `waiting`. + owed: BTreeSet, } impl OriginConsumerState { + /// Whether the consumer has read everything its sources announced before they + /// synced. A torn-down origin has no sources left to wait on. + fn synced(&self) -> bool { + (self.waiting.load(Ordering::Relaxed) == 0 || self.ended) && self.owed.is_empty() + } + + /// Owe every pending update that changes which prefixes are announced: a start + /// the consumer has not seen, or an end. A re-price or restart of a delivered + /// route leaves the set as it is, and may be held, so it is not owed. + fn owe(&mut self) { + self.owed = self + .pending + .iter() + .filter(|(prefix, update)| { + !self.delivered.contains(*prefix) + || matches!( + update, + PendingUpdate::Unannounce(_) | PendingUpdate::UnannounceAnnounce { .. } + ) + }) + .map(|(prefix, _)| prefix.clone()) + .collect(); + } + fn apply_announce(&mut self, prefix: PathOwned, meta: RouteMeta, captures: Option>) { let meta = (meta, captures); let new = match self.pending.remove(&prefix) { @@ -868,8 +902,10 @@ impl OriginConsumerState { let last = (last, captures); match self.pending.remove(&prefix) { // The pending announce was never delivered and neither was any earlier - // one, so the pair cancels entirely. - Some(PendingUpdate::Announce(_)) if !self.delivered.contains(&prefix) => {} + // one, so the pair cancels entirely, and nothing is owed for it. + Some(PendingUpdate::Announce(_)) if !self.delivered.contains(&prefix) => { + self.owed.remove(&prefix); + } // Either nothing is pending or the pending announce was a metadata // update on a delivered route; the consumer still owes a retraction. None | Some(PendingUpdate::Announce(_) | PendingUpdate::Restart(..) | PendingUpdate::Unannounce(_)) => { @@ -950,6 +986,19 @@ impl OriginConsumerState { /// Take the pending update at `prefix` for delivery. fn take_prefix(&mut self, prefix: PathOwned) -> AnnounceEvent { + let event = self.take_update(prefix); + let (AnnounceEvent::Start(announce) + | AnnounceEvent::Update(announce) + | AnnounceEvent::Restart(announce) + | AnnounceEvent::End(announce)) = &event; + // A trailing announce left pending is still owed. + if !self.pending.contains_key(&announce.prefix) { + self.owed.remove(&announce.prefix); + } + event + } + + fn take_update(&mut self, prefix: PathOwned) -> AnnounceEvent { let ((meta, captures), kind): (_, fn(Announce) -> AnnounceEvent) = match self.pending.remove(&prefix).unwrap() { PendingUpdate::Announce(meta) => { // The consumer has seen this prefix before, so it is a metadata update. @@ -1176,9 +1225,12 @@ struct Horizon { impl Horizon { /// Whether `entry` may be observed or served through this horizon. fn admits(&self, entry: &RouteEntry) -> bool { - !(self.local && entry.link != Link::Local) - && !(self.upstream && entry.link == Link::Upstream) - && entry.visible_to(self.exclude) + self.admits_link(entry.link) && entry.visible_to(self.exclude) + } + + /// Whether routes arriving on `link` may be observed through this horizon. + fn admits_link(&self, link: Link) -> bool { + !(self.local && link != Link::Local) && !(self.upstream && link == Link::Upstream) } /// This horizon as it applies to the routes covering `path`: an excluded peer @@ -1307,6 +1359,17 @@ impl TableCursor { .map(|(_, captures)| captures) } + /// Whether this cursor waits on `snapshot`: what it delivers could present here, + /// as [`Self::visible`] judges a route at or beneath its prefix. + fn awaits(&self, snapshot: &SnapshotEntry) -> bool { + self.horizon.admits_link(snapshot.link) + && self.allowed.overlaps(&snapshot.claim) + && self.hidden.reaches(&self.heads, &snapshot.prefix) + && !self.holes.iter().any(|hole| snapshot.prefix.has_prefix(hole)) + // Too deep for the mount to name means everything beneath is too. + && self.named.as_ref().is_none_or(|(mount, _)| mount.names(&snapshot.prefix)) + } + /// Whether this cursor may observe `entry` at all: advertised, not behind /// the excluded peer (split horizon), within the cursor's patterns, and /// nameable through its mount. @@ -1565,8 +1628,14 @@ pub(super) struct Hidden { impl Hidden { /// Whether a cursor hanging at `heads` reports a route at `prefix`. fn discovers(&self, heads: &[PathOwned], prefix: &Path) -> bool { - (self.include || !hides(self.from.as_ref().map(std::slice::from_ref).unwrap_or(heads), prefix)) - && self.beyond.as_ref().is_none_or(|outer| hides(outer, prefix)) + self.reaches(heads, prefix) && self.beyond.as_ref().is_none_or(|outer| hides(outer, prefix)) + } + + /// Whether a cursor hanging at `heads` could report a route at or beneath `prefix`. + /// A hidden segment hides everything beneath it, while a top-up feed's rule turns + /// on the rest of the path, so only the first can rule a prefix out. + fn reaches(&self, heads: &[PathOwned], prefix: &Path) -> bool { + self.include || !hides(self.from.as_ref().map(std::slice::from_ref).unwrap_or(heads), prefix) } /// This rule for a cursor reading through `mount`, whose heads sit on the @@ -1948,6 +2017,39 @@ impl Producer { ) } + /// Start a [`Snapshot`] under `prefix`: the announce consumers it overlaps are not + /// synced until the returned guard drops. Taken before the request goes out, so + /// nothing reads as synced in between. + #[cfg_attr(not(feature = "async"), allow(dead_code))] + // pending: /quest/m1/public-model/answer.md + pub(crate) fn snapshot(&self, prefix: impl AsPath) -> Snapshot { + let prefix = self.root.join(prefix.as_path()).to_owned(); + // A prefix past the depth limit can carry no route, so nothing waits on it. + let Ok(claim) = prefix_claim(&prefix) else { + return Snapshot::landed(); + }; + let mut state = self.shared.lock(); + if state.closed { + return Snapshot::landed(); + } + let id = state.next_snapshot; + state.next_snapshot += 1; + state.add_snapshot( + &prefix, + SnapshotEntry { + id, + link: self.link, + claim, + prefix: prefix.clone(), + }, + ); + Snapshot { + shared: Some(self.shared.clone()), + prefix, + id, + } + } + /// Advertise a route over `prefix` and serve the requests beneath it. /// /// A route is always a prefix: it claims `prefix` and every path beneath it @@ -2527,10 +2629,14 @@ impl DriverState { } // End the announce cursors: each drains its pending updates, then reports - // the end. Registrations stay (the cursors remove themselves on drop). + // the end. Registrations stay (the cursors remove themselves on drop). A + // snapshot still in flight is released like a landing, owing what is queued. for state in cursors { if let Ok(mut state) = state.write() { state.ended = true; + if state.waiting.swap(0, Ordering::Relaxed) > 0 { + state.owe(); + } } } } @@ -3481,6 +3587,11 @@ struct RouteNode { /// Watches at this node or below, so a route change walks only the subtrees /// holding one. watches_below: usize, + /// Snapshots in flight at exactly this prefix; see [`Snapshot`]. + snapshots: Vec, + /// Snapshots at this node or below, so a cursor registering walks only the + /// subtrees holding one. + snapshots_below: usize, children: HashMap, } @@ -3637,23 +3748,89 @@ impl Drop for Watch { } } +/// A snapshot in flight, as the route table holds it. +struct SnapshotEntry { + id: u64, + /// The link its routes arrive on, so a horizon hiding them never waits on it. + link: Link, + /// The prefix's [`prefix_claim`], so a cursor whose patterns miss it never waits on it. + claim: Pattern, + /// The absolute prefix, so a cursor that could see nothing beneath it never waits on it. + prefix: PathOwned, +} + +/// A session still delivering its peer's initial announcements under a prefix. +/// +/// Every announce consumer whose scope the prefix overlaps is unsynced +/// (see [`AnnounceConsumer::poll_synced`]) until this drops: once the session +/// has inserted the announcements its peer counted, or it gave up on them. +// pending: /quest/m1/public-model/answer.md +pub(crate) struct Snapshot { + shared: Option>>, + prefix: PathOwned, + id: u64, +} + +impl Snapshot { + /// A snapshot nothing waits on. + #[cfg_attr(not(feature = "async"), allow(dead_code))] + // pending: /quest/m1/public-model/answer.md + pub(crate) fn landed() -> Self { + Self { + shared: None, + prefix: PathOwned::default(), + id: 0, + } + } +} + +impl Drop for Snapshot { + fn drop(&mut self) { + if let Some(shared) = self.shared.take() { + shared.lock().remove_snapshot(&self.prefix, self.id); + } + } +} + /// What a registration adds to the subtree counts on its walk. #[derive(Clone, Copy)] struct Below { cursors: usize, watches: usize, + snapshots: usize, } impl Below { - const NONE: Self = Self { cursors: 0, watches: 0 }; - const CURSOR: Self = Self { cursors: 1, watches: 0 }; - const WATCH: Self = Self { cursors: 0, watches: 1 }; + const NONE: Self = Self { + cursors: 0, + watches: 0, + snapshots: 0, + }; + const CURSOR: Self = Self { + cursors: 1, + watches: 0, + snapshots: 0, + }; + const WATCH: Self = Self { + cursors: 0, + watches: 1, + snapshots: 0, + }; + const SNAPSHOT: Self = Self { + cursors: 0, + watches: 0, + snapshots: 1, + }; } impl RouteNode { /// Nothing here and nothing below: the node can be pruned. fn is_empty(&self) -> bool { - self.entries.is_empty() && self.cursors.is_empty() && self.watches.is_empty() && self.children.is_empty() + self.entries.is_empty() + && self.cursors.is_empty() + && self.watches.is_empty() + && self.snapshots.is_empty() + && self.children.is_empty() } /// The node `parts` below this one, if the table has it. @@ -3669,6 +3846,7 @@ impl RouteNode { fn reach<'a>(&mut self, mut parts: impl Iterator, below: Below) -> &mut Self { self.cursors_below += below.cursors; self.watches_below += below.watches; + self.snapshots_below += below.snapshots; match parts.next() { None => self, Some(part) => self.children.entry(part.to_string()).or_default().reach(parts, below), @@ -3697,6 +3875,7 @@ impl RouteNode { }; self.cursors_below -= below.cursors; self.watches_below -= below.watches; + self.snapshots_below -= below.snapshots; Some(result) } @@ -3756,6 +3935,17 @@ impl RouteNode { } } + /// Collect the snapshots at this node and below, skipping subtrees with none. + fn collect_snapshots<'a>(&'a self, out: &mut Vec<&'a SnapshotEntry>) { + if self.snapshots_below == 0 { + return; + } + out.extend(&self.snapshots); + for child in self.children.values() { + child.collect_snapshots(out); + } + } + /// Collect the cursors at this node and below, skipping subtrees with none. fn collect_cursors(&self, out: &mut Vec) { if self.cursors_below == 0 { @@ -3964,6 +4154,7 @@ struct OriginState { // read. A join takes a value of its own, so a start before it reads older. next_route: u64, next_watch: u64, + next_snapshot: u64, // The registered announce cursors, each with its own coalescing buffer. Each // also hangs in the route table at its heads, which is how an announcement @@ -3993,6 +4184,7 @@ impl OriginState { routes: RouteTable::default(), next_route: 0, next_watch: 0, + next_snapshot: 0, cursors: HashMap::new(), fronts: WeakCache::default(), withdrawn: HashMap::new(), @@ -4192,9 +4384,80 @@ impl OriginState { for head in &cursor.heads { self.routes.add_cursor(head, id); } + + // The snapshots in flight it can see sit where its routes would, each counted once. + let mut snapshots = Vec::new(); + for head in &cursor.heads { + let (above, at) = self.routes.split(head); + snapshots.extend(above.into_iter().flat_map(|node| node.snapshots.iter())); + if let Some(node) = at { + node.collect_snapshots(&mut snapshots); + } + } + snapshots.retain(|snapshot| cursor.awaits(snapshot)); + snapshots.sort_unstable_by_key(|snapshot| snapshot.id); + snapshots.dedup_by_key(|snapshot| snapshot.id); + if !snapshots.is_empty() { + cursor + .state + .read() + .waiting + .fetch_add(snapshots.len(), Ordering::Relaxed); + } + self.cursors.insert(id, cursor); } + /// Hold every cursor `snapshot` can present on unsynced until it lands. + fn add_snapshot(&mut self, prefix: &Path, snapshot: SnapshotEntry) { + for id in self.routes.cursors_touching(prefix) { + if let Some(cursor) = self.cursors.get(&id) + && cursor.awaits(&snapshot) + { + cursor.state.read().waiting.fetch_add(1, Ordering::Relaxed); + } + } + self.routes + .root + .reach(prefix.parts(), Below::SNAPSHOT) + .snapshots + .push(snapshot); + } + + /// The snapshot `id` at `prefix` landed: release the cursors it held. + fn remove_snapshot(&mut self, prefix: &Path, id: u64) { + let removed = self.routes.root.edit(prefix.parts(), Below::SNAPSHOT, |node| { + let index = node.snapshots.iter().position(|snapshot| snapshot.id == id)?; + Some(node.snapshots.swap_remove(index)) + }); + let Some(Some(snapshot)) = removed else { + return; + }; + for id in self.routes.cursors_touching(prefix) { + let Some(cursor) = self.cursors.get(&id).filter(|cursor| cursor.awaits(&snapshot)) else { + continue; + }; + // Only the last landing changes what a synced waiter sees, so only it takes the + // write guard and wakes. It zeroes the count and owes the queue in one critical + // section, so no poll sees nothing waiting before the owed set is in. The table + // lock keeps the count from moving between the two guards. + { + let state = cursor.state.read(); + if state.waiting.load(Ordering::Relaxed) > 1 { + state.waiting.fetch_sub(1, Ordering::Relaxed); + continue; + } + } + match cursor.state.write() { + Ok(mut state) => { + state.waiting.store(0, Ordering::Relaxed); + state.owe(); + } + Err(state) => state.waiting.store(0, Ordering::Relaxed), + } + } + } + /// The best served route covering `path` (absolute) for a requester seeing /// `horizon`, skipping the `excluded` entry ids. /// @@ -5537,6 +5800,12 @@ impl AnnounceConsumer { ids.push(id); } } + // Under the table lock, so no snapshot lands between the count and this. + if let Ok(mut state) = state.write() + && state.waiting.load(Ordering::Relaxed) == 0 + { + state.owe(); + } } Self { @@ -5747,6 +6016,42 @@ impl AnnounceConsumer { state.is_closed() || state.ended } + /// Whether this consumer has synced; see [`Self::poll_synced`]. + pub fn is_synced(&self) -> bool { + let state = self.state.read(); + state.is_closed() || state.synced() + } + + /// Wait until this consumer has synced with what its sources announced. + /// + /// Synced means every session feeding this scope has delivered its peer's + /// announcements as of when it answered, and this consumer has read every event + /// queued before then. A prefix not announced by then is not announced, rather + /// than not heard yet. A session joining the scope later, such as a reconnect, + /// takes this back until it syncs too; routes published on this origin + /// itself never wait. + /// + /// Synced is relative to the session's peer, never global: a relay answers with + /// what it holds without waiting on its own upstreams. Versions that cannot mark + /// the boundary (moq-lite-03/04, and moq-transport without MoQ Active Count) catch + /// up once the peer accepts the request, or once it is sent when nothing answers. + /// + /// Level-triggered: ready whenever synced, and once the cursor is closed. + pub fn poll_synced(&self, waiter: &kio::Waiter) -> Poll<()> { + self.state + .poll(waiter, |state| match state.synced() { + true => Poll::Ready(()), + false => Poll::Pending, + }) + .map(|_| ()) + } + + /// Wait until this consumer has synced; see [`Self::poll_synced`]. + #[cfg(feature = "async")] + pub async fn synced(&self) { + kio::wait(|waiter| self.poll_synced(waiter)).await + } + /// Returns the root that is automatically stripped from emitted prefixes. pub fn root(&self) -> &Path<'_> { &self.root @@ -6047,6 +6352,190 @@ mod tests { announced.assert_next_wait(); } + /// A consumer is synced once a snapshot lands and it has read what came before. + #[test] + fn synced_after_the_snapshot_and_its_events() { + let producer = origin(1).produce(); + let snapshot = producer.snapshot("room"); + let mut announced = producer.consume().announced(); + assert!(!announced.is_synced(), "a snapshot in flight holds it"); + + let _alice = producer.announce("room/alice", Route::default()).unwrap(); + drop(snapshot); + assert!(!announced.is_synced(), "never before an earlier event is read"); + + announced.assert_next_active("room/alice"); + assert!(announced.is_synced()); + + // A later announcement is news, not part of what it synced with. + let _bob = producer.announce("room/bob", Route::default()).unwrap(); + assert!(announced.is_synced()); + } + + /// Without any session feeding it, a consumer syncs on its own replay. + #[test] + fn a_local_origin_catches_up_on_its_replay() { + let producer = origin(1).produce(); + let mut empty = producer.consume().announced(); + assert!(empty.is_synced(), "nothing to wait on"); + empty.assert_next_wait(); + + let _alice = producer.announce("room/alice", Route::default()).unwrap(); + let mut announced = producer.consume().announced(); + assert!(!announced.is_synced(), "the replay is owed"); + announced.assert_next_active("room/alice"); + assert!(announced.is_synced()); + } + + /// A start and end that cancel before being read are owed nothing. + #[test] + fn a_cancelled_start_still_catches_up() { + let producer = origin(1).produce(); + let snapshot = producer.snapshot(""); + let mut announced = producer.consume().announced(); + + let alice = producer.announce("room/alice", Route::default()).unwrap(); + drop(snapshot); + assert!(!announced.is_synced()); + drop(alice); + assert!(announced.is_synced(), "the pair cancelled"); + announced.assert_next_wait(); + } + + /// A re-price of a route already read is not owed: the set of prefixes is unchanged. + #[test] + fn an_update_is_not_owed() { + let producer = origin(1).produce(); + let mut announced = producer.consume().announced(); + let alice = producer.announce("room/alice", Route::default()).unwrap(); + announced.assert_next_active("room/alice"); + + let snapshot = producer.snapshot(""); + alice.update(Route::default().with_cost(5)).unwrap(); + drop(snapshot); + assert!(announced.is_synced()); + } + + /// A snapshot taken after syncing (a reconnect) holds it again until it lands. + #[test] + fn a_new_snapshot_takes_sync_back() { + let producer = origin(1).produce(); + let announced = producer.consume().announced(); + assert!(announced.is_synced()); + + let snapshot = producer.snapshot("room/alice"); + assert!(!announced.is_synced(), "a snapshot beneath the scope holds it"); + let second = producer.snapshot(""); + drop(snapshot); + assert!(!announced.is_synced(), "every current snapshot must land"); + drop(second); + assert!(announced.is_synced()); + } + + /// Only snapshots that could present on a consumer hold it. + #[test] + fn only_overlapping_snapshots_hold() { + let producer = origin(1).produce(); + let room = producer + .consume() + .scope("room", &Patterns::from(Pattern::all())) + .unwrap(); + + let other = producer.snapshot("other"); + assert!(room.announced().is_synced(), "a disjoint prefix"); + + let above = producer.snapshot(""); + let announced = room.announced(); + assert!(!announced.is_synced(), "a snapshot above the root covers it"); + drop(above); + assert!(announced.is_synced()); + + drop(other); + + // A local view never waits on a peer's routes. + let _peer = producer.clone().peer().snapshot("room"); + assert!(producer.consume().local().announced().is_synced()); + assert!(!producer.consume().announced().is_synced()); + } + + /// A source shadowed by a mount can present nothing on the handle, so it never holds it. + #[test] + fn a_shadowed_snapshot_never_holds() { + let producer = origin(1).produce(); + let project = mounted(&producer, "p1", &["**"]).with_hidden(true); + let announced = project.announced(); + assert!(announced.is_synced()); + + let shadowed = producer.snapshot("p1/.svc"); + assert!(announced.is_synced(), "a snapshot in flight under the hole"); + assert!(project.announced().is_synced(), "a reader registering past one"); + drop(shadowed); + + let _target = producer.snapshot(".svc/p1"); + assert!(!announced.is_synced(), "the mount's target does hold it"); + } + + /// A source under a hidden segment holds only the readers that opt into hidden routes. + #[test] + fn a_hidden_snapshot_holds_only_hidden_readers() { + let producer = origin(1).produce(); + let announced = producer.consume().announced(); + let _stats = producer.snapshot(".stats"); + assert!(announced.is_synced(), "a snapshot in flight under a hidden segment"); + assert!( + producer.consume().announced().is_synced(), + "a reader registering past one" + ); + assert!(!producer.consume().with_hidden(true).announced().is_synced()); + } + + /// A waiter parked on syncing is woken when it does. + #[test] + fn catching_up_wakes_the_waiter() { + struct Woken(AtomicBool); + impl Wake for Woken { + fn wake(self: Arc) { + self.0.store(true, Ordering::SeqCst); + } + } + + let producer = origin(1).produce(); + let snapshot = producer.snapshot(""); + let announced = producer.consume().announced(); + let woken = Arc::new(Woken(AtomicBool::new(false))); + let waiter = kio::Waiter::new(std::task::Waker::from(woken.clone())); + assert!(announced.poll_synced(&waiter).is_pending()); + drop(snapshot); + assert!(woken.0.load(Ordering::SeqCst), "not woken"); + assert!(announced.poll_synced(&waiter).is_ready()); + } + + /// A session joining or landing short of the last one wakes no parked cursor, so a + /// relay's announce feeds stay idle while peers connect. + #[test] + fn holding_back_wakes_nobody() { + struct Woken(AtomicBool); + impl Wake for Woken { + fn wake(self: Arc) { + self.0.store(true, Ordering::SeqCst); + } + } + + let producer = origin(1).produce(); + let mut announced = producer.consume().announced(); + let woken = Arc::new(Woken(AtomicBool::new(false))); + let waiter = kio::Waiter::new(std::task::Waker::from(woken.clone())); + assert!(announced.poll_next(&waiter).is_pending()); + assert!(announced.poll_synced(&waiter).is_ready()); + + let first = producer.snapshot(""); + let second = producer.snapshot("room"); + drop(first); + assert!(!woken.0.load(Ordering::SeqCst), "woken while still held back"); + drop(second); + assert!(woken.0.load(Ordering::SeqCst), "the last landing wakes"); + } + /// A `.`-prefixed segment below the requested prefix hides a route from /// discovery unless the reader opts in; one inside the prefix does not. #[moq_clock::test] @@ -9232,6 +9721,22 @@ mod tests { assert!(late.next().now_or_never().expect("ended").is_none()); } + /// A teardown with a snapshot in flight still owes what the cursor has not read. + #[test] + #[cfg(feature = "async")] + fn teardown_owes_what_a_snapshot_held() { + let (producer, driver) = Producer::new(Config::new(origin(1))); + let snapshot = producer.snapshot("room"); + let mut announced = producer.consume().announced(); + let _alice = producer.announce("room/alice", Route::default()).unwrap(); + + drop(driver); + assert!(!announced.is_synced(), "the queue is still owed"); + announced.assert_next_active("room/alice"); + assert!(announced.is_synced()); + drop(snapshot); + } + #[test] fn internal_route_requests_are_closed_on_teardown() { let (producer, mut driver) = Producer::new(Config::new(origin(1))); diff --git a/rs/moq-net/tests/integration/main.rs b/rs/moq-net/tests/integration/main.rs index d155bb218f5..0a814a8af27 100644 --- a/rs/moq-net/tests/integration/main.rs +++ b/rs/moq-net/tests/integration/main.rs @@ -45,6 +45,7 @@ mod settled_end_survives_abort; mod subscribe_tracks; mod subscription_end_integrity; mod support; +mod synced; mod track_abort_over_session; mod track_info_cancel; mod track_stream_demand; diff --git a/rs/moq-net/tests/integration/synced.rs b/rs/moq-net/tests/integration/synced.rs new file mode 100644 index 00000000000..47a05873b41 --- /dev/null +++ b/rs/moq-net/tests/integration/synced.rs @@ -0,0 +1,122 @@ +#![cfg(feature = "async")] + +//! A subscriber's announce consumers sync once the session has delivered its +//! peer's initial announcements: exactly on versions that mark the boundary, and +//! on the answer, as a best effort, on the rest. + +use std::collections::BTreeSet; +use std::time::Duration; + +use moq_net::{Hop, Version, announce, origin}; + +use crate::support; +use support::harness::{MockConnectOptions, connect_mock}; + +/// Long enough, in virtual time, for anything in flight to reach the far side. +const SETTLE: Duration = Duration::from_secs(1); + +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 +} + +/// Read events until the consumer syncs, returning the prefixes then announced. +async fn read_until_synced(announced: &mut announce::Consumer) -> BTreeSet { + let mut seen = BTreeSet::new(); + while !announced.is_synced() { + let event = moq_clock::timeout(SETTLE, announced.next()) + .await + .expect("synced in time") + .expect("the cursor stays open"); + match event { + announce::Event::Start(announce) | announce::Event::Restart(announce) => { + seen.insert(announce.prefix.to_string()); + } + announce::Event::End(announce) => { + seen.remove(announce.prefix.as_str()); + } + announce::Event::Update(_) => {} + } + } + seen +} + +/// Connect a subscriber to a publisher holding two broadcasts, and return what its +/// origin's consumer had read when it synced. +async fn synced(version: &str) -> BTreeSet { + let publisher = produce_origin(1); + let a = publisher.create_broadcast("a").unwrap(); + a.announce(Default::default()).unwrap(); + let b = publisher.create_broadcast("b").unwrap(); + b.announce(Default::default()).unwrap(); + + let subscriber = produce_origin(2); + let mut options = MockConnectOptions::new(version.parse::().unwrap()); + options.server_publish = Some(publisher.consume()); + options.client_subscribe = Some(subscriber.clone()); + let _pair = connect_mock(options).await; + + let mut announced = subscriber.consume().announced(); + read_until_synced(&mut announced).await +} + +fn both() -> BTreeSet { + ["a", "b"].into_iter().map(String::from).collect() +} + +/// Every version that marks the boundary syncs with the whole initial set. +#[moq_clock::test] +async fn an_exact_boundary_catches_up_with_the_initial_set() { + for version in [ + "moq-lite-05", + "moq-lite-06", + "moq-lite-07-wip", + "moq-transport-16", + "moq-transport-17", + "moq-transport-18", + "moq-transport-19", + "moq-transport-20", + "moq-transport-21", + "moq-transport-22", + ] { + assert_eq!(synced(version).await, both(), "{version}"); + } +} + +/// Versions without a boundary still sync, so nothing waiting on it hangs. +#[moq_clock::test] +async fn a_best_effort_boundary_still_catches_up() { + for version in ["moq-lite-03", "moq-lite-04", "moq-transport-14", "moq-transport-15"] { + synced(version).await; + } +} + +/// A second session into the same origin, as a reconnect makes, syncs too: the +/// consumer waits on every current session. +#[moq_clock::test] +async fn a_second_session_catches_up() { + let publisher = produce_origin(1); + let a = publisher.create_broadcast("a").unwrap(); + a.announce(Default::default()).unwrap(); + + let subscriber = produce_origin(2); + let connect = || { + let mut options = MockConnectOptions::new("moq-lite-07-wip".parse::().unwrap()); + options.server_publish = Some(publisher.consume()); + options.client_subscribe = Some(subscriber.clone()); + connect_mock(options) + }; + let first = connect().await; + let mut announced = subscriber.consume().announced(); + assert_eq!( + read_until_synced(&mut announced).await, + ["a"].into_iter().map(String::from).collect() + ); + + let second = connect().await; + drop(first); + moq_clock::sleep(SETTLE).await; + read_until_synced(&mut announced).await; + drop(second); +}