From 3b7d895d6b2b77dbfdf29f00e78befe7eb491cbd Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 26 Sep 2026 10:03:26 -0700 Subject: [PATCH 1/6] quest: claim js-announce-caught-up Co-Authored-By: Claude Opus 5.5 From d08847924636ce69d960d3293e995518956652fc Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 26 Sep 2026 10:18:47 -0700 Subject: [PATCH 2/6] feat(js/net)!: announce streams yield a live marker once caught up Mirror the Rust announce::Event: @moq/net's announcement streams yield Announce.Event, either a route event (announced, updated, retracted) carrying an Announce.Announce, or one { kind: "live" } marker once every route live at subscribe time has been delivered. Sessions land the peer's initial set per wire: lite-01/02 after ANNOUNCE_INIT, lite-05+ after ANNOUNCE_OK's count of ANNOUNCE_STARTs, and lite-03/04 and moq-transport once the stream goes quiet (500 ms before the first announce, 30 ms between). An origin's stream also waits for every session still replaying into it when the stream opened; a dead session releases its hold. Announce.Update, Announce.Kind, and Announce.isActive are removed. Co-Authored-By: Claude Opus 5.5 --- demo/web/src/index.ts | 9 +- demo/web/src/stats.ts | 3 +- doc/concept/moq-lite.md | 15 +-- doc/lib/js/net.md | 13 ++- js/moq-boy/src/element.tsx | 3 +- js/net/examples/discovery.ts | 4 + js/net/src/announce.ts | 2 +- js/net/src/announced.test.ts | 16 +++ js/net/src/announced.ts | 102 ++++++++++++----- js/net/src/connection/established.ts | 3 +- js/net/src/connection/forward.test.ts | 49 +++++++++ js/net/src/connection/forward.ts | 11 +- js/net/src/connection/pool.ts | 10 +- js/net/src/connection/reload.ts | 8 +- js/net/src/ietf/subscriber.test.ts | 60 +++++----- js/net/src/ietf/subscriber.ts | 10 ++ js/net/src/integration.test.ts | 92 +++++++++++++--- js/net/src/lite/subscriber.test.ts | 153 ++++++++++++++++++++------ js/net/src/lite/subscriber.ts | 29 ++++- js/net/src/origin.test.ts | 139 ++++++++++++++++++----- js/net/src/origin.ts | 51 ++++++++- js/net/src/wire.ts | 2 + js/room/src/room.test.ts | 2 +- js/room/src/room.ts | 3 +- js/watch/src/broadcast.ts | 7 +- quest/m1/README.md | 1 - quest/m1/announce-live-bindings.md | 4 - quest/m1/js-announce-caught-up.md | 16 --- quest/m1/js-origin-scope.md | 1 - 29 files changed, 628 insertions(+), 190 deletions(-) delete mode 100644 quest/m1/js-announce-caught-up.md diff --git a/demo/web/src/index.ts b/demo/web/src/index.ts index c70b95d171..dd0352f0c6 100644 --- a/demo/web/src/index.ts +++ b/demo/web/src/index.ts @@ -176,8 +176,8 @@ function createTile(name: string): WatchTile { // --------------------------------------------------------------------------- // // Subscribe to announcements under the prefix and keep a live set of active broadcasts. -// `announced.next()` drains the update stream, so we track membership ourselves: active=true adds the -// path, active=false removes it. `Connection.announced()` spans reconnects (it retracts everything on +// `announced.next()` drains the event stream, so we track membership ourselves: an announcement adds the +// path, a retraction removes it. `Connection.announced()` spans reconnects (it retracts everything on // disconnect and re-announces on reconnect), so the set self-heals without any extra wiring here. const discovery = new Signals.Effect(); discovery.run((effect) => { @@ -190,12 +190,13 @@ discovery.run((effect) => { for (;;) { const entry = await Promise.race([effect.cancel, announced.next()]); if (!entry) break; + if (entry.kind === "live") continue; const path = entry.prefix; // Only catalog-backed broadcasts are watchable streams; this skips the relay's // `.stats` broadcast (see the stats dashboard demo for that one). if (!path.endsWith(".hang") && !path.endsWith(".msf")) continue; - if (Net.Announce.isActive(entry.kind)) live.add(path); - else live.delete(path); + if (entry.kind === "retracted") live.delete(path); + else live.add(path); broadcasts.set([...live].sort()); } }); diff --git a/demo/web/src/stats.ts b/demo/web/src/stats.ts index a66d621617..ca1a937d0c 100644 --- a/demo/web/src/stats.ts +++ b/demo/web/src/stats.ts @@ -131,11 +131,12 @@ discovery.run((effect) => { for (;;) { const entry = await Promise.race([effect.cancel, announced.next()]); if (!entry) break; + if (entry.kind === "live") continue; const path = entry.prefix; const node = Net.Path.stripPrefix(prefix, path); if (!node) continue; - if (Net.Announce.isActive(entry.kind)) { + if (entry.kind !== "retracted") { if (subs.has(node)) continue; const ne = new Signals.Effect(); subs.set(node, ne); diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index 7de0e24bc8..7b87e1c8d5 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -136,13 +136,14 @@ root. The pattern scope filters which prefixes are visible without changing a route's prefix. Announce events carry the covered path, captures, and what happened to it: Rust `announce::Event::{Announced, Updated, Retracted}`, each holding an `announce::Announce { prefix, captures: Option>, route }`, -and TypeScript `Announce.Update { path, captures, route, kind }`, where the kind -is announced, updated (a reprice in place), or retracted. Captures are present -when the announced prefix pins every wildcard in the most-specific matching -scope member. The Rust consumer is a `Stream` and the TypeScript one an async -iterable. The Rust consumer also yields one `announce::Event::Live` once the -routes live at subscribe time have all been delivered, including those a peer -session was still sending: moq-lite-05+ counts them in `ANNOUNCE_OK`, +and TypeScript `Announce.Event`, whose `kind` is `"announced"`, `"updated"`, or +`"retracted"` alongside the same `Announce.Announce` fields. An update is a +reprice in place. Captures are present when the announced prefix pins every +wildcard in the most-specific matching scope member. The Rust consumer is a +`Stream` and the TypeScript one an async iterable. Both also yield one `Live` +marker (TypeScript `{ kind: "live" }`) once the routes live at subscribe time +have all been delivered, including those a peer session was still sending: +moq-lite-05+ counts them in `ANNOUNCE_OK`, moq-lite-01/02 send them in `ANNOUNCE_INIT`, and older or IETF sessions wait for the stream to go quiet. diff --git a/doc/lib/js/net.md b/doc/lib/js/net.md index 0ce30a89e2..c8b550dff8 100644 --- a/doc/lib/js/net.md +++ b/doc/lib/js/net.md @@ -100,11 +100,14 @@ Three operations, on an origin: route is always a prefix on every wire. A route is a capability, not an inventory. `origin.announced(scope)` yields -`Announce.Update` values: `prefix` is the covered prefix relative to the origin, -`captures` is one pattern per scope wildcard when the prefix pins a complete -match (otherwise `undefined`), `kind` is `"announced"`, `"updated"` (a -reprice in place), or `"retracted"`, and `route` carries hops and cost (on a -retraction, its last values). The consumer is an async iterable. A prefix is +`Announce.Event` values. `kind` is `"announced"`, `"updated"` (a reprice in +place), or `"retracted"`, each also carrying an `Announce.Announce`: `prefix` +is the covered prefix relative to the origin, `captures` is one pattern per +scope wildcard when the prefix pins a complete match (otherwise `undefined`), +and `route` carries hops and cost (on a retraction, its last values). A single +`{ kind: "live" }` follows the routes live at subscribe time, including every +route a connected peer was still sending, so a caller listing what is live +stops there. The consumer is an async iterable. A prefix is not a broadcast name; the scope filters locally while sessions request its literal head on the wire. Paths with a `.`-prefixed segment below that head are [hidden](/concept/moq-lite#hidden-broadcasts) unless `announced(scope, { hidden: true })` opts in; diff --git a/js/moq-boy/src/element.tsx b/js/moq-boy/src/element.tsx index 48346b6de1..bde64da8a5 100644 --- a/js/moq-boy/src/element.tsx +++ b/js/moq-boy/src/element.tsx @@ -137,13 +137,14 @@ export default class MoqBoy extends HTMLElement { for (;;) { const entry = await effect.race(announced.next()); if (!entry) break; + if (entry.kind === "live") continue; // A broad route that cannot pin the game id names nothing to open. const capture = entry.captures?.[0]; if (!capture?.isLiteral) continue; const id = capture.text; - if (Moq.Announce.isActive(entry.kind) && !this.#sessions.has(id)) { + if (entry.kind !== "retracted" && !this.#sessions.has(id)) { const config: GameConfig = { sessionId: id, connection: this.connection, diff --git a/js/net/examples/discovery.ts b/js/net/examples/discovery.ts index 5001a58326..12f29cf198 100644 --- a/js/net/examples/discovery.ts +++ b/js/net/examples/discovery.ts @@ -10,6 +10,10 @@ async function main() { // Discover broadcasts announced by the server for await (const announcement of announced) { + if (announcement.kind === "live") { + console.log("Caught up: everything live has been listed"); + continue; + } if (announcement.kind === "retracted") continue; console.log("New stream available:", announcement.prefix); diff --git a/js/net/src/announce.ts b/js/net/src/announce.ts index 91d3cee37c..ebc2adaf25 100644 --- a/js/net/src/announce.ts +++ b/js/net/src/announce.ts @@ -3,4 +3,4 @@ * * @module */ -export { Consumer, isActive, type Kind, type Update } from "./announced.ts"; +export { type Announce, Consumer, type Event } from "./announced.ts"; diff --git a/js/net/src/announced.test.ts b/js/net/src/announced.test.ts index 68adf0d212..90cac2bba7 100644 --- a/js/net/src/announced.test.ts +++ b/js/net/src/announced.test.ts @@ -57,6 +57,22 @@ test("a same-name re-announce is a distinct update", async () => { expect(await consumer.next()).toEqual({ prefix: p("a"), captures: undefined, kind: "announced", route }); }); +test("the live marker is delivered once", async () => { + const producer = new Announce.Producer(); + const consumer = producer.consume(); + + // A stream spanning sessions forwards each one's marker; only the first is news. + const route = Route.default; + producer.append({ kind: "live" }); + producer.append({ prefix: p("a"), captures: undefined, kind: "announced", route }); + producer.append({ kind: "live" }); + producer.append({ prefix: p("b"), captures: undefined, kind: "announced", route }); + + expect(await consumer.next()).toEqual({ kind: "live" }); + expect(await consumer.next()).toMatchObject({ prefix: p("a") }); + expect(await consumer.next()).toMatchObject({ prefix: p("b") }); +}); + test("closing resolves next with undefined", async () => { const producer = new Announce.Producer(); const consumer = producer.consume(); diff --git a/js/net/src/announced.ts b/js/net/src/announced.ts index 01056ae153..2bd5f82351 100644 --- a/js/net/src/announced.ts +++ b/js/net/src/announced.ts @@ -8,14 +8,7 @@ import type { Route } from "./hop.js"; import type * as Path from "./path.js"; /** - * What an {@link Update} reports about its prefix. - * - * @public - */ -export type Kind = "announced" | "updated" | "retracted"; - -/** - * A route announcement, update, or retraction. + * A route over a prefix, delivered inside an {@link Event}. * * An announcement is always a prefix, never a broadcast: a route claims that * {@link prefix} and every path beneath it can be served. By convention a publisher @@ -25,19 +18,30 @@ export type Kind = "announced" | "updated" | "retracted"; * * @public */ -export interface Update { +export interface Announce { /** * The prefix the route covers, relative to the origin (for a session, its URL path). */ prefix: Path.Valid; /** What the filter's wildcards stood for, when this prefix pins all of them. */ captures: Path.Pattern[] | undefined; - /** Whether the prefix was announced, re-priced, or retracted. */ - kind: Kind; /** Hops and cost of the route; on a retraction, its last advertised values. */ route: Route; } +/** + * What an announcement stream yields. + * + * `announced`: a route now covers the prefix. `updated`: the route covering it changed + * hops or cost, in place. `retracted`: no route covers it any more. `live`: every route + * live at subscribe time has been delivered, including those a connected peer was still + * sending, so what follows is live changes; yielded at most once, and a caller listing + * what is live stops there. + * + * @public + */ +export type Event = ({ kind: "announced" | "updated" | "retracted" } & Announce) | { kind: "live" }; + /** * Options for an announcement stream. * @@ -53,14 +57,11 @@ export interface Options { hidden?: boolean; } -/** Whether a route covers the path after an update of this {@link Kind}. */ -export function isActive(kind: Kind): boolean { - return kind !== "retracted"; -} - /** Reactive backing state shared by announcement producers and consumers. */ class AnnounceState { - queue = new Signal([]); + queue = new Signal([]); + // Whether the live marker was appended, so a repeat (from a stream spanning sessions) is dropped. + live = false; closed = new Once(); } @@ -94,11 +95,15 @@ export class Producer { return makeConsumer(this.#state); } - /** Writes an announcement to the queue. */ - append(update: Update) { + /** Writes an event to the queue. The `live` marker is written once; later ones are dropped. */ + append(event: Event) { if (this.#state.closed.peek() !== undefined) throw new Error("announcements are closed"); + if (event.kind === "live") { + if (this.#state.live) return; + this.#state.live = true; + } this.#state.queue.mutate((queue) => { - queue.push(update); + queue.push(event); }); } @@ -136,17 +141,17 @@ export class Consumer { makeConsumer = (state) => new Consumer(state); } - /** The announcements as they arrive, until the stream closes. */ - async *[Symbol.asyncIterator](): AsyncGenerator { + /** The events as they arrive, until the stream closes. */ + async *[Symbol.asyncIterator](): AsyncGenerator { for (;;) { - const update = await this.next(); - if (!update) return; - yield update; + const event = await this.next(); + if (!event) return; + yield event; } } - /** Returns the next announcement. */ - async next(): Promise { + /** Returns the next event, or undefined once the stream closes. */ + async next(): Promise { for (;;) { const announce = this.#state.queue.peek().shift(); if (announce) return announce; @@ -164,3 +169,46 @@ export class Consumer { closeState(this.#state, abort); } } + +/** + * When an initial set has landed, for wires that never say where it ends (lite-03/04, + * moq-transport): once its stream goes quiet. + * + * A peer writes its whole set back to back, so the first announcement gets a round trip's + * grace and each one after it only has to beat its siblings. Mirrors `Quiet` in `rs/moq-net`. + * + * @internal + */ +export class Quiet { + /** How long the stream may stay silent before its first announcement, in ms. */ + static readonly FIRST = 500; + /** How long the stream may stay silent between announcements, in ms. */ + static readonly GAP = 30; + + #landed: () => void; + #timer: ReturnType | undefined; + + /** Start counting from now; `landed` runs once the stream goes quiet. */ + constructor(landed: () => void) { + this.#landed = landed; + this.#timer = setTimeout(() => this.#land(), Quiet.FIRST); + } + + /** An announcement arrived: the set is still landing. */ + heard(): void { + if (this.#timer === undefined) return; + clearTimeout(this.#timer); + this.#timer = setTimeout(() => this.#land(), Quiet.GAP); + } + + /** Stop counting without landing, once the stream is gone. Idempotent. */ + close(): void { + clearTimeout(this.#timer); + this.#timer = undefined; + } + + #land(): void { + this.#timer = undefined; + this.#landed(); + } +} diff --git a/js/net/src/connection/established.ts b/js/net/src/connection/established.ts index f796892762..877db01154 100644 --- a/js/net/src/connection/established.ts +++ b/js/net/src/connection/established.ts @@ -37,7 +37,8 @@ export interface Established { * Subscribe to broadcast announcements matching `scope`, any pattern (`foo/**` * for a subtree, `room/* /chat` for each room's chat, default `**`). Paths are * relative to the session; captures report what the scope's wildcards stood for. - * Hidden routes are left out unless `options.hidden` opts in. + * Hidden routes are left out unless `options.hidden` opts in. The `live` marker follows + * once the peer's initial set has landed. */ announced(scope?: Path.Pattern, options?: announce.Options): announce.Consumer; diff --git a/js/net/src/connection/forward.test.ts b/js/net/src/connection/forward.test.ts index d9d9b74481..e644a65ea3 100644 --- a/js/net/src/connection/forward.test.ts +++ b/js/net/src/connection/forward.test.ts @@ -188,3 +188,52 @@ test("a request replaced across one coalesced wakeup still gets answered", async second.close(); origin.close(); }); + +/** Whether `next` is still waiting once everything queued has run. */ +async function pending(next: Promise): Promise { + const waiting = Symbol("waiting"); + const result = await Promise.race([next, settle().then(() => waiting)]); + return result === waiting; +} + +test("the origin's live marker waits for the session's initial set", async () => { + const origin = new OriginProducer(); + const session = new FakeSession(); + forwardAnnounced(session.session, origin); + + const announced = origin.announced(); + session.announces.append({ prefix: Path.from("a"), captures: undefined, kind: "announced", route: Route.default }); + expect(await announced.next()).toMatchObject({ prefix: Path.from("a"), kind: "announced" }); + + // The peer has not said its initial set is complete. + const next = announced.next(); + expect(await pending(next)).toBe(true); + + session.announces.append({ kind: "live" }); + expect(await next).toEqual({ kind: "live" }); + + // A stream opened after the set landed does not wait on it. + const later = origin.announced(); + expect(await later.next()).toMatchObject({ prefix: Path.from("a"), kind: "announced" }); + expect(await later.next()).toEqual({ kind: "live" }); + + announced.close(); + later.close(); + origin.close(); +}); + +test("a session dying before its initial set lands releases the marker", async () => { + const origin = new OriginProducer(); + const session = new FakeSession(); + forwardAnnounced(session.session, origin); + + const announced = origin.announced(); + const next = announced.next(); + expect(await pending(next)).toBe(true); + + session.die(); + expect(await next).toEqual({ kind: "live" }); + + announced.close(); + origin.close(); +}); diff --git a/js/net/src/connection/forward.ts b/js/net/src/connection/forward.ts index 66745398a9..31d4a168c0 100644 --- a/js/net/src/connection/forward.ts +++ b/js/net/src/connection/forward.ts @@ -4,7 +4,6 @@ * @module */ import { type Dispose, race } from "@moq/signals"; -import { isActive } from "../announced.ts"; import type { Dynamic, Producer as OriginProducer, RequestSlot } from "../origin.ts"; import type * as Path from "../path.ts"; import { wireOf } from "../wire.ts"; @@ -51,6 +50,10 @@ export function forwardAnnounced(conn: Established, origin: OriginProducer): voi const announced = conn.announced(undefined, { hidden: true }); const inserted = new Map(); + // Taken before the session is handed out, so no announcement stream on the origin misses + // it: each one opened now withholds its live marker until the peer's initial set lands. + const landed = originWire.replaying(); + // End the stream the moment the session closes rather than waiting for the wire to // error it, so the retractions below land promptly. void conn.closed.then(() => announced.close()); @@ -62,7 +65,9 @@ export function forwardAnnounced(conn: Established, origin: OriginProducer): voi const event = await announced.next(); if (!event) break; - if (isActive(event.kind)) { + if (event.kind === "live") { + landed(); + } else if (event.kind !== "retracted") { const existing = inserted.get(event.prefix); if (existing) { existing.update(event.route); @@ -82,6 +87,8 @@ export function forwardAnnounced(conn: Established, origin: OriginProducer): voi // cleanup below retracts everything this stream fed either way. failure = err; } finally { + // A dead stream cannot hold the marker back. + landed(); for (const handle of inserted.values()) handle.close(); inserted.clear(); announced.close(); diff --git a/js/net/src/connection/pool.ts b/js/net/src/connection/pool.ts index 959697ca09..d485a694dd 100644 --- a/js/net/src/connection/pool.ts +++ b/js/net/src/connection/pool.ts @@ -263,7 +263,8 @@ export class Connection { /** * Subscribe to broadcast announcements matching `scope`, spanning reconnects * and URL switches: a switch retracts everything from - * the old relay's origin, then the new one's arrivals stream in. + * the old relay's origin, then the new one's arrivals stream in. The `live` marker + * comes once, from the first origin; see {@link Origin.Consumer.announced}. */ announced(scope: Path.Pattern = Path.Pattern.all(), options?: Announce.Options): Announce.Consumer { const producer = new Announce.Producer(); @@ -288,15 +289,16 @@ export class Connection { // Track what this origin announced so a URL switch retracts it; the last // event rides along for the retraction. - const active = new Map(); + const active = new Map(); effect.spawn(async () => { try { for (;;) { const entry = await effect.race(upstream.next()); if (!entry) break; - if (Announce.isActive(entry.kind)) active.set(entry.prefix, entry); - else active.delete(entry.prefix); + if (entry.kind === "retracted") active.delete(entry.prefix); + else if (entry.kind !== "live") active.set(entry.prefix, entry); + // The stream delivers the marker once; a later session's is dropped. producer.append(entry); } } finally { diff --git a/js/net/src/connection/reload.ts b/js/net/src/connection/reload.ts index b7b77bcaef..21d9c4e54e 100644 --- a/js/net/src/connection/reload.ts +++ b/js/net/src/connection/reload.ts @@ -440,6 +440,7 @@ export class Reload { * The same {@link Announce.Consumer} stream as {@link Established.announced}, but everything active * is retracted (a `retracted` update) whenever the connection drops and re-announced on * reconnect, so a consumer draining `next()` never clings to a dead route across a reconnect. + * The `live` marker comes once, from the first session. * * Stays empty while the relay lacks {@link Established.discovery}. */ @@ -465,15 +466,16 @@ export class Reload { // 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(); + const active = new Map(); effect.spawn(async () => { try { for (;;) { const entry = await effect.race(upstream.next()); if (!entry) break; - if (Announce.isActive(entry.kind)) active.set(entry.prefix, entry); - else active.delete(entry.prefix); + if (entry.kind === "retracted") active.delete(entry.prefix); + else if (entry.kind !== "live") active.set(entry.prefix, entry); + // The stream delivers the marker once; a later session's is dropped. producer.append(entry); } } catch { diff --git a/js/net/src/ietf/subscriber.test.ts b/js/net/src/ietf/subscriber.test.ts index a0f560d42a..65bf252454 100644 --- a/js/net/src/ietf/subscriber.test.ts +++ b/js/net/src/ietf/subscriber.test.ts @@ -17,6 +17,16 @@ import { SubscribeNamespace, SubscribeNamespaceEntry, SubscribeNamespaceEntryDon import { Subscriber } from "./subscriber.ts"; import { ALPN, Version } from "./version.ts"; +/** The next route event, skipping the live marker: these tests pin routes, and the marker has its own. */ +async function nextRoute(announced: { + next(): Promise; +}): Promise | undefined> { + for (;;) { + const event = await announced.next(); + if (event?.kind !== "live") return event as Exclude | undefined; + } +} + const VERSION = Version.DRAFT_19; /** How long to wait for a stream before calling it absent. */ @@ -85,7 +95,7 @@ test("an unsolicited announcement lands", async () => { stream, ); - const next = await announced.next(); + const next = await nextRoute(announced); expect(next?.prefix).toBe(Path.from("surprise")); expect(next?.kind).toBe("announced"); @@ -95,7 +105,7 @@ test("an unsolicited announcement lands", async () => { if (!peer) throw new Error("no PUBLISH_NAMESPACE stream to close"); peer.close(); await handler; - expect(await announced.next()).toMatchObject({ + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("surprise"), kind: "retracted", }); @@ -131,7 +141,7 @@ async function inlineNamespace(stream: Stream, path: Path.Valid, cluster?: Clust */ async function syncInline(stream: Stream, announced: announce.Consumer, cluster?: Cluster.Advert): Promise { await inlineNamespace(stream, Path.from("sentinel"), cluster); - expect(await announced.next()).toMatchObject({ + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("sentinel"), kind: "announced", }); @@ -159,7 +169,7 @@ test("an announcement survives the first of its two sources ending", async () => new PublishNamespace({ requestId: 0n, trackNamespace: Path.from("both") }), request, ); - expect(await announced.next()).toMatchObject({ prefix: Path.from("both"), kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("both"), kind: "announced" }); await inlineNamespace(subscription, Path.from("both")); await syncInline(subscription, announced); @@ -172,7 +182,7 @@ test("an announcement survives the first of its two sources ending", async () => await handler; const next = await Promise.race([ - announced.next(), + nextRoute(announced), new Promise<"nothing">((resolve) => setTimeout(() => resolve("nothing"), 250)), ]); expect(next).toBe("nothing"); @@ -195,7 +205,7 @@ test("an announcement ends once its last source does", async () => { new PublishNamespace({ requestId: 0n, trackNamespace: Path.from("both") }), request, ); - expect(await announced.next()).toMatchObject({ prefix: Path.from("both"), kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("both"), kind: "announced" }); await inlineNamespace(subscription, Path.from("both")); await syncInline(subscription, announced); @@ -209,7 +219,7 @@ test("an announcement ends once its last source does", async () => { await subscription.writer.u53(SubscribeNamespaceEntryDone.id); await new SubscribeNamespaceEntryDone({ suffix: Path.from("both") }).encode(subscription.writer, VERSION); - expect(await announced.next()).toMatchObject({ prefix: Path.from("both"), kind: "retracted" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("both"), kind: "retracted" }); }); /** @@ -258,7 +268,7 @@ test("a duplicate legacy publish_namespace is still refused", async () => { new PublishNamespace({ requestId: 0n, trackNamespace: Path.from("twice") }), first, ); - expect(await announced.next()).toMatchObject({ prefix: Path.from("twice"), kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("twice"), kind: "announced" }); // The same namespace again, on its own request. const second = await Stream.open(pair.server, { version: Version.DRAFT_15 }); @@ -273,7 +283,7 @@ test("a duplicate legacy publish_namespace is still refused", async () => { peer.close(); await handler; - expect(await announced.next()).toMatchObject({ prefix: Path.from("twice"), kind: "retracted" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("twice"), kind: "retracted" }); }); /** @@ -332,7 +342,7 @@ test("concurrent legacy publish_namespace requests take one reference", async () second, ); - expect(await announced.next()).toMatchObject({ prefix: Path.from("raced"), kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("raced"), kind: "announced" }); await two; // Only one reference was taken, so the surviving request ending retracts the path. @@ -341,7 +351,7 @@ test("concurrent legacy publish_namespace requests take one reference", async () peer.close(); await one; - expect(await announced.next()).toMatchObject({ prefix: Path.from("raced"), kind: "retracted" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("raced"), kind: "retracted" }); }); /** The Hop IDs a cluster-negotiated session declared, ours first. */ @@ -380,10 +390,10 @@ test("an inline NAMESPACE that starts looping back is retracted", async () => { const subscription = await acceptSubscribeNamespace(pair.client); await inlineNamespace(subscription, Path.from("theirs"), { hops: [PEER], cost: 0n }); - expect(await announced.next()).toMatchObject({ prefix: Path.from("theirs"), kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("theirs"), kind: "announced" }); await inlineNamespace(subscription, Path.from("theirs"), { hops: [SELF, PEER], cost: 0n }); - expect(await announced.next()).toMatchObject({ prefix: Path.from("theirs"), kind: "retracted" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("theirs"), kind: "retracted" }); }); /** @@ -400,14 +410,14 @@ test("a repeated NAMESPACE reprices in place", async () => { const subscription = await acceptSubscribeNamespace(pair.client); await inlineNamespace(subscription, Path.from("theirs"), { hops: [PEER], cost: 4n }); - expect(await announced.next()).toMatchObject({ + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("theirs"), kind: "announced", route: { hops: [PEER], cost: { warm: 4n, cold: 4n } }, }); await inlineNamespace(subscription, Path.from("theirs"), { hops: [PEER], cost: 0n }); - expect(await announced.next()).toMatchObject({ + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("theirs"), kind: "updated", route: { hops: [PEER], cost: { warm: 0n, cold: 0n } }, @@ -533,7 +543,7 @@ test("a PUBLISH_NAMESPACE update that starts looping back is detached", async () }), request, ); - expect(await announced.next()).toMatchObject({ prefix: Path.from("theirs"), kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("theirs"), kind: "announced" }); const peer = await nextStream(pair.client); if (!peer) throw new Error("no PUBLISH_NAMESPACE stream"); @@ -547,7 +557,7 @@ test("a PUBLISH_NAMESPACE update that starts looping back is detached", async () VERSION, ); - expect(await announced.next()).toMatchObject({ prefix: Path.from("theirs"), kind: "retracted" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("theirs"), kind: "retracted" }); expect(await peer.reader.u53()).toBe(RequestOk.id); await RequestOk.decode(peer.reader, VERSION); @@ -557,7 +567,7 @@ test("a PUBLISH_NAMESPACE update that starts looping back is detached", async () peer.writer, VERSION, ); - expect(await announced.next()).toMatchObject({ prefix: Path.from("theirs"), kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("theirs"), kind: "announced" }); expect(await peer.reader.u53()).toBe(RequestOk.id); await RequestOk.decode(peer.reader, VERSION); @@ -587,7 +597,7 @@ test("a PUBLISH_NAMESPACE repricing is acknowledged in place", async () => { }), request, ); - expect(await announced.next()).toMatchObject({ + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("theirs"), kind: "announced", route: { hops: [PEER], cost: { warm: 4n, cold: 4n } }, @@ -602,7 +612,7 @@ test("a PUBLISH_NAMESPACE repricing is acknowledged in place", async () => { await new PublishNamespaceUpdate({ requestId: 3n, update: { cost: 0n } }).encode(peer.writer, VERSION); expect(await peer.reader.u53()).toBe(RequestOk.id); await RequestOk.decode(peer.reader, VERSION); - expect(await announced.next()).toMatchObject({ + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("theirs"), kind: "updated", route: { hops: [PEER], cost: { warm: 0n, cold: 0n } }, @@ -611,7 +621,7 @@ test("a PUBLISH_NAMESPACE repricing is acknowledged in place", async () => { // Still announced: the stream ending is what retracts it. peer.close(); await handler; - expect(await announced.next()).toMatchObject({ prefix: Path.from("theirs"), kind: "retracted" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("theirs"), kind: "retracted" }); }); /** @@ -636,7 +646,7 @@ test("a PUBLISH_NAMESPACE update that changes the publisher is refused", async ( }), request, ); - expect(await announced.next()).toMatchObject({ prefix: Path.from("theirs"), kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("theirs"), kind: "announced" }); const peer = await nextStream(pair.client); if (!peer) throw new Error("no PUBLISH_NAMESPACE stream"); @@ -653,7 +663,7 @@ test("a PUBLISH_NAMESPACE update that changes the publisher is refused", async ( // The refusal closed the stream, which withdrew the advertisement. await handler; - expect(await announced.next()).toMatchObject({ prefix: Path.from("theirs"), kind: "retracted" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("theirs"), kind: "retracted" }); }); /** @@ -675,7 +685,7 @@ test("a repeated PUBLISH_NAMESPACE is a protocol violation", async () => { }); const request = await Stream.open(pair.server, { version: VERSION }); const handler = subscriber.runPublishNamespace(advert, request); - expect(await announced.next()).toMatchObject({ prefix: Path.from("theirs"), kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("theirs"), kind: "announced" }); const peer = await nextStream(pair.client); if (!peer) throw new Error("no PUBLISH_NAMESPACE stream"); @@ -683,7 +693,7 @@ test("a repeated PUBLISH_NAMESPACE is a protocol violation", async () => { await advert.encode(peer.writer, VERSION); await expect(handler).rejects.toThrow(ProtocolViolation); - expect(await announced.next()).toMatchObject({ prefix: Path.from("theirs"), kind: "retracted" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("theirs"), kind: "retracted" }); }); /** diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index 914a4e8f4a..6c3a5b995c 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -303,6 +303,10 @@ export class Subscriber { // reference nobody is left to release, pinning the path for the session. let released = false; + // Nothing on this wire says where the initial set ends, so it has landed once the + // stream goes quiet. + let quiet: announce.Quiet | undefined; + // v14/v15: SubscribeNamespace on control stream (via adapter virtual stream) // v16+: SubscribeNamespace on its own real bidi stream @@ -351,6 +355,10 @@ export class Subscriber { throw new Error(`SubscribeNamespace rejected: typeId=0x${respTypeId.toString(16)}`); } + quiet = new announce.Quiet(() => { + if (announced.closed.peek() === undefined) announced.append({ kind: "live" }); + }); + // Loop reading Namespace/NamespaceDone entries const readLoop = (async () => { for (;;) { @@ -358,6 +366,7 @@ export class Subscriber { if (done) break; const msgType = await stream.reader.u53(); + quiet?.heard(); if (msgType === SubscribeNamespaceEntry.id) { const entry = await SubscribeNamespaceEntry.decode( stream.reader, @@ -453,6 +462,7 @@ export class Subscriber { // each namespace keeps its count and the source never detaches, which would // pin the path for the session even after the other source withdrew. released = true; + quiet?.close(); for (const path of live) { this.#detachAnnounce(path); } diff --git a/js/net/src/integration.test.ts b/js/net/src/integration.test.ts index 249f4b1bb9..b634435d30 100644 --- a/js/net/src/integration.test.ts +++ b/js/net/src/integration.test.ts @@ -1,5 +1,6 @@ import { expect, test } from "bun:test"; import type { Getter } from "@moq/signals"; +import type * as Announce from "./announce.ts"; import { type Consumer as BroadcastConsumer, Producer as BroadcastProducer } from "./broadcast.ts"; import { type AcceptProps, @@ -22,6 +23,16 @@ import type { Producer as TrackProducer } from "./track.ts"; import { withTimeout } from "./util/timeout.ts"; import { wireOf } from "./wire.ts"; +/** The next route event, skipping the live marker: these tests pin routes, and the marker has its own. */ +async function nextRoute(announced: { + next(): Promise; +}): Promise | undefined> { + for (;;) { + const event = await announced.next(); + if (event?.kind !== "live") return event as Exclude | undefined; + } +} + function publish(origin: OriginProducer, path: Path.Valid) { const broadcast = origin.createBroadcast(path); broadcast.announce(); @@ -90,14 +101,14 @@ async function runPublishSubscribeFlow(protocol: string, version?: number) { // Client discovers announced broadcast const announced = client.announced(); - const entry = await announced.next(); + const entry = await nextRoute(announced); if (!entry) throw new Error("expected entry"); expect(entry.prefix).toBe("test" as Path.Valid); expect(entry.kind).toBe("announced"); // Scoped discovery only echoes the suffix on the wire, but presents the whole path. const prefixed = client.announced(Path.Pattern.subtree(Path.from("root"))); - const prefixedEntry = await prefixed.next(); + const prefixedEntry = await nextRoute(prefixed); if (!prefixedEntry) throw new Error("expected prefixed entry"); expect(prefixedEntry.prefix).toBe("root/child" as Path.Valid); expect(prefixedEntry.kind).toBe("announced"); @@ -374,28 +385,28 @@ test("integration: lite draft-06 announce lifecycle", async () => { const first = publish(origin, Path.from("first")); const announced = client.announced(); - let entry = await announced.next(); + let entry = await nextRoute(announced); if (!entry) throw new Error("expected announce"); expect(entry.prefix).toBe("first" as Path.Valid); expect(entry.kind).toBe("announced"); // A live announce. const second = publish(origin, Path.from("second")); - entry = await announced.next(); + entry = await nextRoute(announced); if (!entry) throw new Error("expected announce"); expect(entry.prefix).toBe("second" as Path.Valid); expect(entry.kind).toBe("announced"); // Unannounce: retracted by announce id on the wire. second.close(); - entry = await announced.next(); + entry = await nextRoute(announced); if (!entry) throw new Error("expected unannounce"); expect(entry.prefix).toBe("second" as Path.Valid); expect(entry.kind).toBe("retracted"); // Re-announce the same path: a fresh announce assigning a fresh id. const secondAgain = publish(origin, Path.from("second")); - entry = await announced.next(); + entry = await nextRoute(announced); if (!entry) throw new Error("expected re-announce"); expect(entry.prefix).toBe("second" as Path.Valid); expect(entry.kind).toBe("announced"); @@ -409,10 +420,10 @@ test("integration: lite draft-06 announce lifecycle", async () => { }); /** Collect announced prefixes until `until` arrives. */ -async function announcedUntil(announced: { next(): Promise<{ prefix: Path.Valid } | undefined> }, until: string) { +async function announcedUntil(announced: Announce.Consumer, until: string) { const seen: string[] = []; while (!seen.includes(until)) { - const entry = await withTimeout(announced.next(), 1000, `waiting for ${until}`); + const entry = await withTimeout(nextRoute(announced), 1000, `waiting for ${until}`); if (!entry) throw new Error("announcements ended"); seen.push(entry.prefix); } @@ -1271,7 +1282,7 @@ test("integration: ietf draft-14 subscriber teardown on last unsubscribe", async // Draft-14 only completes SUBSCRIBE_OK once the session is warmed by an announce round-trip. const announced = client.announced(); - await announced.next(); + await nextRoute(announced); const remote = wireOf(client).consume(Path.from("test")); const sub = remote.track("video").subscribe().ordered(); @@ -1902,7 +1913,7 @@ async function runOriginFlow(protocol: string, version?: number) { // The announcement lands in the client's origin. const reader = clientOrigin.consume(); const announced = reader.announced(); - expect(await announced.next()).toMatchObject({ prefix: Path.from("test"), kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("test"), kind: "announced" }); // Consuming through the origin reaches the wire. const remote = await routed(reader, Path.from("test")); @@ -1912,7 +1923,7 @@ async function runOriginFlow(protocol: string, version?: number) { // Unpublishing retracts the entry over the wire and out of the origin. broadcast.close(); - expect(await announced.next()).toMatchObject({ prefix: Path.from("test"), kind: "retracted" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("test"), kind: "retracted" }); await until(() => !wireOf(reader).routes(Path.from("test"))); await serving; @@ -2259,7 +2270,7 @@ test("create then announce is discoverable on the wire", async () => { const broadcast = origin.createBroadcast(Path.from("later")); broadcast.createTrack("chat"); - const pending = announced.next(); + const pending = nextRoute(announced); broadcast.announce(); const entry = await pending; expect(entry?.prefix).toBe("later" as Path.Valid); @@ -2291,7 +2302,7 @@ test("a handle serves a request under live/** over the wire", async () => { ]); const announced = client.announced(); - const entry = await announced.next(); + const entry = await nextRoute(announced); expect(entry?.prefix).toBe("live" as Path.Valid); expect(entry?.kind).toBe("announced"); @@ -2405,3 +2416,58 @@ test("integration: lite draft-05 ends a track with the publisher's reset", async server.close(); origin.close(); }); + +/** + * Connect a client whose origin is fed by a peer publishing `paths`, then read the origin's + * announcement stream up to the live marker, returning the paths delivered before it. + */ +async function caughtUp(protocol: string, version: number | undefined, paths: string[]): Promise { + const pair = createMockTransportPair(protocol); + const serverOrigin = new OriginProducer(); + const clientOrigin = new OriginProducer(); + const broadcasts = paths.map((path) => publish(serverOrigin, Path.from(path))); + + const [client, server] = await Promise.all([ + connect(url, { transport: pair.client, consume: clientOrigin }), + accept(pair.server, url, { version, publish: serverOrigin.consume() }), + ]); + + const announced = clientOrigin.announced(); + const seen: string[] = []; + for (;;) { + const event = await withTimeout(announced.next(), 2000, `${protocol || version}: never caught up`); + if (!event) throw new Error("announcements ended"); + if (event.kind === "live") break; + if (event.kind !== "announced") throw new Error(`only announcements before the marker: got ${event.kind}`); + seen.push(event.prefix); + } + + announced.close(); + for (const broadcast of broadcasts) broadcast.close(); + client.close(); + server.close(); + serverOrigin.close(); + clientOrigin.close(); + return seen.sort(); +} + +// Every version reaches the marker after the whole initial set: lite-01/02 via ANNOUNCE_INIT, +// lite-05+ via ANNOUNCE_OK's count, and lite-03/04 and moq-transport once the stream goes quiet. +for (const [name, protocol, version] of [ + ["lite draft-01", "", Lite.Version.DRAFT_01], + ["lite draft-02", "", Lite.Version.DRAFT_02], + ["lite draft-03", Lite.ALPN_03, undefined], + ["lite draft-04", Lite.ALPN_04, undefined], + ["lite draft-05", Lite.ALPN_05, undefined], + ["lite draft-06", Lite.ALPN_06, undefined], + ["ietf draft-14", "", Ietf.Version.DRAFT_14], + ["ietf draft-19", Ietf.ALPN.DRAFT_19, undefined], +] as const) { + test(`integration: ${name} is live after the whole initial set`, async () => { + expect(await caughtUp(protocol, version, ["a", "b", "c"])).toEqual(["a", "b", "c"]); + }); + + test(`integration: ${name} is live with an empty peer`, async () => { + expect(await caughtUp(protocol, version, [])).toEqual([]); + }); +} diff --git a/js/net/src/lite/subscriber.test.ts b/js/net/src/lite/subscriber.test.ts index a26242cf66..1b8c1331ed 100644 --- a/js/net/src/lite/subscriber.test.ts +++ b/js/net/src/lite/subscriber.test.ts @@ -11,6 +11,16 @@ import { Probe } from "./probe.ts"; import { Subscriber } from "./subscriber.ts"; import { Version } from "./version.ts"; +/** The next route event, skipping the live marker: these tests pin routes, and the marker has its own. */ +async function nextRoute(announced: { + next(): Promise; +}): Promise | undefined> { + for (;;) { + const event = await announced.next(); + if (event?.kind !== "live") return event as Exclude | undefined; + } +} + test("closing the subscriber suppresses probe stream warnings", async () => { let readable!: ReadableStreamDefaultController; const quic = { @@ -150,7 +160,7 @@ test("a max-length chain plus withheld responder is dropped", async () => { Version.DRAFT_06, ), ); - expect(await announced.next()).toMatchObject({ + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room"), kind: "announced", route: { hops: [PUBLISHER_A, UNKNOWN_HOP] }, @@ -173,7 +183,7 @@ test("an unidentified responder keeps hop 0 on a nonempty chain", async () => { Version.DRAFT_06, ), ); - expect(await announced.next()).toMatchObject({ + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room"), kind: "announced", route: { hops: [PUBLISHER_A, UNKNOWN_HOP] }, @@ -192,7 +202,7 @@ test("a received empty hop list is filled with hop 0 and marked anonymous", asyn await send((w) => encodeAnnounceBroadcast(w, { status: "active", suffix: Path.from("room"), hops: [] }, Version.DRAFT_06), ); - expect(await announced.next()).toMatchObject({ + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room"), kind: "announced", route: { hops: [UNKNOWN_HOP] }, @@ -215,7 +225,7 @@ test("a received chain with hop 0 is marked anonymous", async () => { Version.DRAFT_06, ), ); - expect(await announced.next()).toMatchObject({ + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room"), kind: "announced", }); @@ -237,7 +247,7 @@ test("a restart from the same publisher is a route change, not a republish", asy Version.DRAFT_06, ), ); - expect(await announced.next()).toMatchObject({ prefix: Path.from("room"), kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room"), kind: "announced" }); // Same publisher over an identical route. In-flight subscriptions resume across it, // and there is no metadata to forward, so the subscriber must not surface a restart. @@ -246,19 +256,19 @@ test("a restart from the same publisher is a route change, not a republish", asy // A different publisher took the path: nothing carries over, so this one does surface, // as an end before the start. Reaching it proves the identical reroute above emitted nothing. await send((w) => encodeAnnounceBroadcast(w, { status: "restart", id: 0n, hops: [PUBLISHER_B] }, Version.DRAFT_06)); - expect(await announced.next()).toMatchObject({ prefix: Path.from("room"), kind: "retracted" }); - expect(await announced.next()).toMatchObject({ prefix: Path.from("room"), kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room"), kind: "retracted" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room"), kind: "announced" }); // A third publisher takes over. The replacement above has to leave its own publisher on // record, or this one reads as a first announcement and skips the end. await send((w) => encodeAnnounceBroadcast(w, { status: "restart", id: 0n, hops: [PUBLISHER_C] }, Version.DRAFT_06)); - expect(await announced.next()).toMatchObject({ prefix: Path.from("room"), kind: "retracted" }); - expect(await announced.next()).toMatchObject({ prefix: Path.from("room"), kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room"), kind: "retracted" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room"), kind: "announced" }); // And the new owner's own reroute is still transparent. await send((w) => encodeAnnounceBroadcast(w, { status: "restart", id: 0n, hops: [PUBLISHER_C] }, Version.DRAFT_06)); await send((w) => encodeAnnounceBroadcast(w, { status: "endedId", id: 0n }, Version.DRAFT_06)); - expect(await announced.next()).toMatchObject({ prefix: Path.from("room"), kind: "retracted" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room"), kind: "retracted" }); announced.close(); subscriber.close(); @@ -277,7 +287,7 @@ test("a restart that re-prices the same publisher emits the new route", async () Version.DRAFT_06, ), ); - expect(await announced.next()).toMatchObject({ + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room"), kind: "announced", route: { hops: [PUBLISHER_A, PEER], cost: { warm: 0n, cold: 0n } }, @@ -290,7 +300,7 @@ test("a restart that re-prices the same publisher emits the new route", async () Version.DRAFT_06, ), ); - expect(await announced.next()).toMatchObject({ + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room"), kind: "updated", route: { hops: [PUBLISHER_A, PEER], cost: { warm: 4n, cold: 4n } }, @@ -310,13 +320,13 @@ test("a lite-05 duplicate announce follows the same restart rule", async () => { encodeAnnounceBroadcast(w, { status: "active", suffix: Path.from("room"), hops }, Version.DRAFT_05); await send(active([PUBLISHER_A])); - expect(await announced.next()).toMatchObject({ prefix: Path.from("room"), kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room"), kind: "announced" }); // On lite-05 a restart travels as a duplicate ANNOUNCE rather than its own message. await send(active([PUBLISHER_A])); await send(active([PUBLISHER_B])); - expect(await announced.next()).toMatchObject({ prefix: Path.from("room"), kind: "retracted" }); - expect(await announced.next()).toMatchObject({ prefix: Path.from("room"), kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room"), kind: "retracted" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room"), kind: "announced" }); announced.close(); subscriber.close(); @@ -336,12 +346,12 @@ test("a restart from an unidentified publisher replaces rather than reroutes", a await send((w) => encodeAnnounceBroadcast(w, { status: "active", suffix: Path.from("room"), hops: [] }, Version.DRAFT_06), ); - expect(await announced.next()).toMatchObject({ prefix: Path.from("room"), kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room"), kind: "announced" }); // Nobody is named on either side, so this is not provably the same content. await send((w) => encodeAnnounceBroadcast(w, { status: "restart", id: 0n, hops: [] }, Version.DRAFT_06)); - expect(await announced.next()).toMatchObject({ prefix: Path.from("room"), kind: "retracted" }); - expect(await announced.next()).toMatchObject({ prefix: Path.from("room"), kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room"), kind: "retracted" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room"), kind: "announced" }); announced.close(); subscriber.close(); @@ -469,7 +479,7 @@ test("an announce skipped as a reflected loop still holds its path", async () => ), ); - await expect(announced.next()).rejects.toThrow("duplicate announce"); + await expect(nextRoute(announced)).rejects.toThrow("duplicate announce"); // The peer has to hear about it: closing only our side would leave it announcing // into a stream nobody reads. expect(await abortReason()).toContain("duplicate announce"); @@ -502,11 +512,11 @@ test("a restart replaces an announce that was skipped as a reflected loop", asyn // The id stays live, so the peer may restart it into a route that is usable here. await send((w) => encodeAnnounceBroadcast(w, { status: "restart", id: 0n, hops: [PUBLISHER_A] }, Version.DRAFT_06)); - expect(await announced.next()).toMatchObject({ prefix: Path.from("room"), kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room"), kind: "announced" }); // Retiring the id ends what the restart attached, and nothing else. await send((w) => encodeAnnounceBroadcast(w, { status: "endedId", id: 0n }, Version.DRAFT_06)); - expect(await announced.next()).toMatchObject({ prefix: Path.from("room"), kind: "retracted" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room"), kind: "retracted" }); announced.close(); subscriber.close(); @@ -536,13 +546,13 @@ test("retiring an id whose announce was skipped ends nothing", async () => { Version.DRAFT_06, ), ); - expect(await announced.next()).toMatchObject({ prefix: Path.from("lobby"), kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("lobby"), kind: "announced" }); // Retire the skipped one, then the live one. The first must surface nothing, so the // only end a consumer sees is "lobby". await send((w) => encodeAnnounceBroadcast(w, { status: "endedId", id: 0n }, Version.DRAFT_06)); await send((w) => encodeAnnounceBroadcast(w, { status: "endedId", id: 1n }, Version.DRAFT_06)); - expect(await announced.next()).toMatchObject({ prefix: Path.from("lobby"), kind: "retracted" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("lobby"), kind: "retracted" }); announced.close(); subscriber.close(); @@ -556,10 +566,10 @@ test("a draft-02 initial announcement can still be retracted", async () => { // ANNOUNCE_INIT carries the initial set. These are advertisements like any other, so // the peer may retract one later and the consumer has to hear about it. await send((w) => new AnnounceInit([Path.from("room")]).encode(w, Version.DRAFT_02)); - expect(await announced.next()).toMatchObject({ prefix: Path.from("room"), kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room"), kind: "announced" }); await send((w) => encodeAnnounceBroadcast(w, { status: "ended", suffix: Path.from("room") }, Version.DRAFT_02)); - expect(await announced.next()).toMatchObject({ prefix: Path.from("room"), kind: "retracted" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room"), kind: "retracted" }); announced.close(); subscriber.close(); @@ -581,7 +591,7 @@ test("a duplicate start is reported even when its own route reflects", async () Version.DRAFT_06, ), ); - expect(await announced.next()).toMatchObject({ prefix: Path.from("room"), kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room"), kind: "announced" }); // A second start for that path, carrying a route that loops back through us. Skipping // it must not pre-empt the violation: the peer sent two starts with no end between @@ -594,7 +604,7 @@ test("a duplicate start is reported even when its own route reflects", async () ), ); - await expect(announced.next()).rejects.toThrow("duplicate announce"); + await expect(nextRoute(announced)).rejects.toThrow("duplicate announce"); announced.close(); subscriber.close(); @@ -612,7 +622,7 @@ test("a draft-02 initial set naming a path twice is refused", async () => { // Erroring the stream discards what it had already queued, so the consumer sees the // violation rather than the first entry followed by it. - await expect(announced.next()).rejects.toThrow("duplicate announce"); + await expect(nextRoute(announced)).rejects.toThrow("duplicate announce"); expect(await abortReason()).toContain("duplicate announce"); // The draft makes this session-fatal: resetting only the stream would let the peer // repeat the violation on the next one. The code has to say so too, or the peer reads @@ -637,10 +647,10 @@ test("a violation on a very long path still ends the session", async () => { const long = Path.from("x".repeat(2000)); const active: AnnounceBroadcast = { status: "active", suffix: long, hops: [PUBLISHER_A] }; await send((w) => encodeAnnounceBroadcast(w, active, Version.DRAFT_06)); - expect(await announced.next()).toMatchObject({ prefix: long, kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: long, kind: "announced" }); await send((w) => encodeAnnounceBroadcast(w, active, Version.DRAFT_06)); - await expect(announced.next()).rejects.toThrow("duplicate announce"); + await expect(nextRoute(announced)).rejects.toThrow("duplicate announce"); const info = await sessionEnded(); expect(info?.closeCode).toBe(15); @@ -658,14 +668,14 @@ test("a draft-04 duplicate start is a violation, not a restart", async () => { const active: AnnounceBroadcast = { status: "active", suffix: Path.from("room"), hops: [PUBLISHER_A] }; await send((w) => encodeAnnounceBroadcast(w, active, Version.DRAFT_04)); - expect(await announced.next()).toMatchObject({ prefix: Path.from("room"), kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room"), kind: "announced" }); await send((w) => encodeAnnounceBroadcast(w, active, Version.DRAFT_04)); // Session first: it is the bounded assertion, so a version that treats this as a // restart fails here in 250ms instead of hanging on a rejection that never comes. expect((await sessionEnded())?.closeCode).toBe(15); - await expect(announced.next()).rejects.toThrow("duplicate announce"); + await expect(nextRoute(announced)).rejects.toThrow("duplicate announce"); }); test("a draft-05 duplicate start is still a restart", async () => { @@ -677,7 +687,7 @@ test("a draft-05 duplicate start is still a restart", async () => { const active: AnnounceBroadcast = { status: "active", suffix: Path.from("room"), hops: [PUBLISHER_A] }; await send((w) => encodeAnnounceBroadcast(w, active, Version.DRAFT_05)); - expect(await announced.next()).toMatchObject({ prefix: Path.from("room"), kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room"), kind: "announced" }); // Same publisher over a new route: transparent, and emphatically not an error. await send((w) => encodeAnnounceBroadcast(w, active, Version.DRAFT_05)); @@ -685,8 +695,81 @@ test("a draft-05 duplicate start is still a restart", async () => { // A different publisher does surface, which is what proves the reroute above passed. const replaced: AnnounceBroadcast = { status: "active", suffix: Path.from("room"), hops: [PUBLISHER_B] }; await send((w) => encodeAnnounceBroadcast(w, replaced, Version.DRAFT_05)); - expect(await announced.next()).toMatchObject({ prefix: Path.from("room"), kind: "retracted" }); - expect(await announced.next()).toMatchObject({ prefix: Path.from("room"), kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room"), kind: "retracted" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room"), kind: "announced" }); + + announced.close(); + subscriber.close(); +}); + +// ANNOUNCE_OK's count lands the initial set at exactly that many ANNOUNCE_STARTs: an unknown +// message does not count toward it, and a live update right behind the set comes after the +// marker. Mirrors the Rust `the_count_lands_at_the_last_initial_start`. +test("the count lands at the last initial start", async () => { + const { subscriber, send, settle } = announceHarness(Version.DRAFT_06); + const announced = subscriber.announced(); + await settle(); + + const start = (suffix: string) => (w: Writer) => + encodeAnnounceBroadcast(w, { status: "active", suffix: Path.from(suffix), hops: [] }, Version.DRAFT_06); + await send((w) => new AnnounceOk(PEER, 1).encode(w, Version.DRAFT_06)); + // An unknown announce type with an empty body, which decodes as skipped. + await send((w) => w.write(new Uint8Array([0x3f, 0x00]))); + await send(start("a")); + await send(start("b")); + + expect(await announced.next()).toMatchObject({ prefix: Path.from("a"), kind: "announced" }); + expect(await announced.next()).toEqual({ kind: "live" }); + expect(await announced.next()).toMatchObject({ prefix: Path.from("b"), kind: "announced" }); + + announced.close(); + subscriber.close(); +}); + +test("a count of zero is live before any announcement", async () => { + const { subscriber, send, settle } = announceHarness(Version.DRAFT_05); + const announced = subscriber.announced(); + await settle(); + + await send((w) => new AnnounceOk(PEER, 0).encode(w, Version.DRAFT_05)); + expect(await announced.next()).toEqual({ kind: "live" }); + + await send((w) => + encodeAnnounceBroadcast(w, { status: "active", suffix: Path.from("a"), hops: [] }, Version.DRAFT_05), + ); + expect(await announced.next()).toMatchObject({ prefix: Path.from("a"), kind: "announced" }); + + announced.close(); + subscriber.close(); +}); + +test("a draft-02 initial set is live after its ANNOUNCE_INIT", async () => { + const { subscriber, send, settle } = announceHarness(Version.DRAFT_02); + const announced = subscriber.announced(); + await settle(); + + await send((w) => new AnnounceInit([Path.from("a"), Path.from("b")]).encode(w, Version.DRAFT_02)); + expect(await announced.next()).toMatchObject({ prefix: Path.from("a"), kind: "announced" }); + expect(await announced.next()).toMatchObject({ prefix: Path.from("b"), kind: "announced" }); + expect(await announced.next()).toEqual({ kind: "live" }); + + announced.close(); + subscriber.close(); +}); + +// Lite-03/04 say nothing about where the initial set ends, so it lands once the stream goes quiet. +test("a draft-04 initial set is live once the stream goes quiet", async () => { + const { subscriber, send, settle } = announceHarness(Version.DRAFT_04); + const announced = subscriber.announced(); + await settle(); + + const start = (suffix: string) => (w: Writer) => + encodeAnnounceBroadcast(w, { status: "active", suffix: Path.from(suffix), hops: [] }, Version.DRAFT_04); + await send(start("a")); + await send(start("b")); + expect(await announced.next()).toMatchObject({ prefix: Path.from("a"), kind: "announced" }); + expect(await announced.next()).toMatchObject({ prefix: Path.from("b"), kind: "announced" }); + expect(await announced.next()).toEqual({ kind: "live" }); announced.close(); subscriber.close(); diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index 59d578c377..cb806a2910 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -203,6 +203,8 @@ export class Subscriber { // Opened outside the try so the catch can reach it: a protocol violation below has // to reset the stream, not just close our side of it. let stream: Stream; + // Lite03/04: lands the initial set once the stream goes quiet. + let quiet: announce.Quiet | undefined; try { stream = await Stream.open(this.#quic); } catch (err: unknown) { @@ -219,12 +221,15 @@ 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; + // Lite05+: the initial set ends after this many more ANNOUNCE_STARTs. + let remaining: number | undefined; if (hasAnnounceOk(this.version)) { const ok = await AnnounceOk.decode(stream.reader, this.version); // Keep a withheld 0: it names nobody for loop detection, but it is the // anonymous mark and must travel the reconstructed chain. Assigned identities // stay off this hop and are never forwarded. responderOrigin = ok.hop; + remaining = ok.active; } // Every advertisement the peer currently has live, keyed by suffix (at most one @@ -271,10 +276,18 @@ export class Subscriber { console.debug(`announced: broadcast=${path} active=true`); announced.append({ prefix: path, captures, kind: "announced", route }); } + announced.append({ kind: "live" }); break; } + case Version.DRAFT_03: + case Version.DRAFT_04: + // No AnnounceInit and no count: the initial set has landed once the stream goes quiet. + quiet = new announce.Quiet(() => { + if (announced.closed.peek() === undefined) announced.append({ kind: "live" }); + }); + break; default: - // Draft03+: no AnnounceInit, initial state comes via Announce messages. + // Lite05+: initial state comes via Announce messages, counted by AnnounceOk. break; } @@ -284,8 +297,14 @@ export class Subscriber { // doesn't know we skipped. const history = new AnnounceHistory(); - // Receive announce updates (for Draft03, this includes initial state) + // Receive announce updates (for Draft03+, this includes initial state) for (;;) { + // Land before decoding past the boundary, so no live update arrives ahead of the marker. + if (remaining === 0) { + announced.append({ kind: "live" }); + remaining = undefined; + } + const announce = await race([ decodeAnnounceBroadcastMaybe(stream.reader, this.version), announced.closed, @@ -294,6 +313,10 @@ export class Subscriber { if (!announce) break; if (announce instanceof Error) throw announce; + quiet?.heard(); + // The count is of ANNOUNCE_STARTs, not every message. + if (remaining !== undefined && announce.status === "active") remaining -= 1; + let path: Path.Valid; let active: boolean; // Present on active/restart; ended messages never carry hops worth checking. @@ -472,6 +495,8 @@ export class Subscriber { if (e instanceof ProtocolViolation) { this.#quic.close({ closeCode: PROTOCOL_VIOLATION_CODE, reason: closeReason(reason(e)) }); } + } finally { + quiet?.close(); } } diff --git a/js/net/src/origin.test.ts b/js/net/src/origin.test.ts index 405871d753..7eec372e87 100644 --- a/js/net/src/origin.test.ts +++ b/js/net/src/origin.test.ts @@ -8,6 +8,16 @@ import { Producer } from "./origin.ts"; import * as Path from "./path.ts"; import { wireOf } from "./wire.ts"; +/** The next route event, skipping the live marker: these tests pin routes, and the marker has its own. */ +async function nextRoute(announced: { + next(): Promise; +}): Promise | undefined> { + for (;;) { + const event = await announced.next(); + if (event?.kind !== "live") return event as Exclude | undefined; + } +} + function publish(origin: Producer, path: Path.Valid) { const broadcast = origin.createBroadcast(path); broadcast.announce(); @@ -297,6 +307,75 @@ test("the table is reactive", async () => { origin.close(); }); +/** Whether `next` is still waiting once everything queued has run. */ +async function pending(next: Promise): Promise { + const waiting = Symbol("waiting"); + const result = await Promise.race([next, settle().then(() => waiting)]); + return result === waiting; +} + +test("an empty origin is live at once, and the marker never repeats", async () => { + const origin = new Producer(); + const announced = origin.announced(); + expect(await announced.next()).toEqual({ kind: "live" }); + + // Later routes are live changes, even past a session that replays and lands. + const landed = wireOf(origin).replaying(); + landed(); + const alice = publish(origin, Path.from("alice")); + expect(await announced.next()).toMatchObject({ prefix: Path.from("alice"), kind: "announced" }); + expect(await pending(announced.next())).toBe(true); + + alice.close(); + origin.close(); +}); + +test("live follows the replayed routes", async () => { + const origin = new Producer(); + const b = publish(origin, Path.from("b")); + const c = publish(origin, Path.from("c")); + + const announced = origin.announced(); + const seen: string[] = []; + for (;;) { + const event = await announced.next(); + if (event?.kind === "live") break; + expect(event?.kind).toBe("announced"); + if (event?.kind === "announced") seen.push(event.prefix); + } + expect(seen.sort()).toEqual(["b", "c"]); + + b.close(); + c.close(); + origin.close(); +}); + +test("a replaying session withholds live until it lands", async () => { + const origin = new Producer(); + const first = wireOf(origin).replaying(); + const second = wireOf(origin).replaying(); + const announced = origin.announced(); + + // Routes a session lands arrive before the marker. + serve(origin, Path.from("room/alice"), () => new BroadcastProducer().consume()); + expect(await announced.next()).toMatchObject({ prefix: Path.from("room/alice"), kind: "announced" }); + + // Every session replaying when the stream opened has to land, and a release is idempotent. + first(); + first(); + const next = announced.next(); + expect(await pending(next)).toBe(true); + + // A session that starts replaying after the stream opened is not waited on. + const late = wireOf(origin).replaying(); + second(); + expect(await next).toEqual({ kind: "live" }); + + late(); + announced.close(); + origin.close(); +}); + test("announced streams the table under a scope with origin-relative paths", async () => { const origin = new Producer(); const consumer = origin.consume(); @@ -306,20 +385,20 @@ test("announced streams the table under a scope with origin-relative paths", asy const announced = consumer.announced(Path.Pattern.subtree(Path.from("room"))); // The initial state arrives first, named from the origin rather than the scope. - expect(await announced.next()).toMatchObject({ prefix: Path.from("room/a"), kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room/a"), kind: "announced" }); // Additions under the scope stream in; paths outside it are invisible. const b = publish(origin, Path.from("room/b")); publish(origin, Path.from("lobby/c")); - expect(await announced.next()).toMatchObject({ prefix: Path.from("room/b"), kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room/b"), kind: "announced" }); // Removals retract. b.close(); - expect(await announced.next()).toMatchObject({ prefix: Path.from("room/b"), kind: "retracted" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room/b"), kind: "retracted" }); // The stream ends when the origin closes. origin.close(); - expect(await announced.next()).toBeUndefined(); + expect(await nextRoute(announced)).toBeUndefined(); a.close(); }); @@ -336,9 +415,9 @@ test("hidden paths need an opt-in or a scope naming the dot segment", async () = expect([...consumer.broadcasts(Path.Pattern.parse(".stats/**")).peek().keys()]).toEqual([Path.from(".stats/node")]); const plain = consumer.announced(); - expect(await plain.next()).toMatchObject({ prefix: Path.from("room/catalog.pro") }); + expect(await nextRoute(plain)).toMatchObject({ prefix: Path.from("room/catalog.pro") }); const opted = consumer.announced(Path.Pattern.parse("room/**"), { hidden: true }); - const seen = [(await opted.next())?.prefix, (await opted.next())?.prefix].sort(); + const seen = [(await nextRoute(opted))?.prefix, (await nextRoute(opted))?.prefix].sort(); expect(seen).toEqual([Path.from("room/.internal"), Path.from("room/catalog.pro")]); plain.close(); @@ -364,10 +443,10 @@ test("a remote entry resolves by path and retracts on dispose", async () => { // Announced streams include remote entries. const announced = consumer.announced(); - expect(await announced.next()).toMatchObject({ prefix: path, kind: "announced", route: Route.default }); + expect(await nextRoute(announced)).toMatchObject({ prefix: path, kind: "announced", route: Route.default }); dispose(); - expect(await announced.next()).toMatchObject({ prefix: path, kind: "retracted" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: path, kind: "retracted" }); expect(wireOf(consumer).routes(path)).toBe(false); announced.close(); @@ -385,10 +464,10 @@ test("announced keeps a broader covering route at its prefix", async () => { // The scope filters the route without changing the prefix it claims. const scope = Path.Pattern.subtree(Path.from("room/alice")); const announced = consumer.announced(scope); - expect(await announced.next()).toMatchObject({ prefix: Path.from("room"), kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room"), kind: "announced" }); dispose(); - expect(await announced.next()).toMatchObject({ prefix: Path.from("room"), kind: "retracted" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("room"), kind: "retracted" }); announced.close(); upstream.close(); @@ -416,7 +495,7 @@ test("a local publish shadows a remote entry", async () => { // One path, one announcement, even though both tables route it. const announced = consumer.announced(); - expect(await announced.next()).toMatchObject({ prefix: path, kind: "announced", route: Route.default }); + expect(await nextRoute(announced)).toMatchObject({ prefix: path, kind: "announced", route: Route.default }); // Dropping the local publish falls back to the remote entry without a retraction. local.close(); @@ -540,12 +619,12 @@ test("disposing the newest remote route promotes the fallback", async () => { const disposeNewer = serve(origin, path, provider(newer)); const announced = consumer.announced(); - expect(await announced.next()).toMatchObject({ prefix: path, kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: path, kind: "announced" }); // The newer session dies: consumers see a retract then the promoted fallback. disposeNewer(); - expect(await announced.next()).toMatchObject({ prefix: path, kind: "retracted" }); - expect(await announced.next()).toMatchObject({ prefix: path, kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: path, kind: "retracted" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: path, kind: "announced" }); const handle = await routed(consumer, path); const track = handle?.track("chat").subscribe(); @@ -613,7 +692,7 @@ test("requests never appear in announced or the table", async () => { const announced = consumer.announced(); publish(origin, Path.from("real")); - expect(await announced.next()).toMatchObject({ prefix: Path.from("real"), kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("real"), kind: "announced" }); announced.close(); request.close(); @@ -627,12 +706,12 @@ test("a republish retracts then re-announces the path", async () => { publish(origin, path); const announced = consumer.announced(); - expect(await announced.next()).toMatchObject({ prefix: path, kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: path, kind: "announced" }); // A new broadcast takes the path: consumers must let go of the superseded one. publish(origin, path); - expect(await announced.next()).toMatchObject({ prefix: path, kind: "retracted" }); - expect(await announced.next()).toMatchObject({ prefix: path, kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: path, kind: "retracted" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: path, kind: "announced" }); announced.close(); origin.close(); @@ -994,7 +1073,7 @@ test("createBroadcast is invisible to everyone until announce", async () => { broadcast.announce({ cost: 4n }); expect(wireOf(consumer).routes(path)).toBe(true); expect(wireOf(consumer).advertised.peek()?.has(path)).toBe(true); - expect(await announced.next()).toMatchObject({ + expect(await nextRoute(announced)).toMatchObject({ prefix: path, kind: "announced", route: { hops: [], cost: { warm: 4n, cold: 4n } }, @@ -1006,18 +1085,18 @@ test("createBroadcast is invisible to everyone until announce", async () => { broadcast.unannounce(); expect(wireOf(consumer).routes(path)).toBe(false); expect(wireOf(consumer).advertised.peek()?.has(path)).toBe(false); - expect(await announced.next()).toMatchObject({ prefix: path, kind: "retracted" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: path, kind: "retracted" }); expect(request.active.peek()).toBeUndefined(); // Back on the air through a fresh handle. broadcast.announce(); - expect(await announced.next()).toMatchObject({ prefix: path, kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: path, kind: "announced" }); const again = request.active.peek(); expect(again).toBeDefined(); expect(again).not.toBe(first); broadcast.close(); - expect(await announced.next()).toMatchObject({ prefix: path, kind: "retracted" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: path, kind: "retracted" }); request.close(); announced.close(); origin.close(); @@ -1039,12 +1118,12 @@ test("a request prefers a cheaper received route over an announced local broadca const remote = request.active.peek(); expect(remote).toBeDefined(); const announced = consumer.announced(); - expect(await announced.next()).toMatchObject({ prefix: path, route: { hops: [PEER] } }); + expect(await nextRoute(announced)).toMatchObject({ prefix: path, route: { hops: [PEER] } }); // Re-priced below it, the local broadcast wins at once, and the remote front it replaced closes. local.announce({ cost: 0n }); - expect(await announced.next()).toMatchObject({ prefix: path, kind: "retracted" }); - expect(await announced.next()).toMatchObject({ prefix: path, kind: "announced", route: Route.default }); + expect(await nextRoute(announced)).toMatchObject({ prefix: path, kind: "retracted" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: path, kind: "announced", route: Route.default }); expect(request.active.peek()).not.toBe(remote); expect(remote?.closed.peek()).not.toBeUndefined(); @@ -1090,7 +1169,7 @@ test("a handle serves a request under live", async () => { const consumer = origin.consume(); const handle = origin.dynamic(Path.from("live")); - expect(await consumer.announced().next()).toMatchObject({ prefix: Path.from("live"), kind: "announced" }); + expect(await nextRoute(consumer.announced())).toMatchObject({ prefix: Path.from("live"), kind: "announced" }); const waiting = handle.requested().next(); const request = consumer.request(Path.from("live/cam")); @@ -1117,11 +1196,11 @@ test("a re-priced route is delivered as an update", async () => { const origin = new Producer(); const handle = origin.dynamic(Path.from("live")); const announced = origin.consume().announced(); - expect(await announced.next()).toMatchObject({ prefix: Path.from("live"), kind: "announced" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("live"), kind: "announced" }); handle.update({ cost: 5n }); - expect(await announced.next()).toMatchObject({ prefix: Path.from("live"), kind: "updated" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("live"), kind: "updated" }); handle.close(); - expect(await announced.next()).toMatchObject({ prefix: Path.from("live"), kind: "retracted" }); + expect(await nextRoute(announced)).toMatchObject({ prefix: Path.from("live"), kind: "retracted" }); announced.close(); origin.close(); }); @@ -1414,7 +1493,7 @@ test("announced filters by arbitrary patterns and reports captures", async () => const broadcast = publish(origin, Path.from("room/alice")); const announced = consumer.announced(Path.Pattern.parse("room/*")); - const update = await announced.next(); + const update = await nextRoute(announced); expect(update?.prefix).toBe(Path.from("room/alice")); expect(update?.captures?.map((capture) => capture.text)).toEqual(["alice"]); diff --git a/js/net/src/origin.ts b/js/net/src/origin.ts index b93d44e62f..0c27403e07 100644 --- a/js/net/src/origin.ts +++ b/js/net/src/origin.ts @@ -291,6 +291,10 @@ class OriginState { // this is coming" and "nothing here can ever serve you". answerers = new Signal(0); + // Sessions still replaying the peer's initial announce set into the table. An + // announcement stream opened while one replays withholds its live marker until it lands. + replaying = new Signal>(new Set()); + closed = new Once(); /** @@ -521,6 +525,7 @@ export class Producer implements Table { receive: (prefix, route) => this.#receive(prefix, route), attach: (discovery) => this.#attach(discovery), expect: () => this.#expect(), + replaying: () => this.#replaying(), get requests() { return thisProducer.#state.requests; }, @@ -733,6 +738,25 @@ export class Producer implements Table { }; } + /** + * Register a session replaying the peer's initial announce set into the table. Returns + * the release; call it once the set has landed, or the session dies. Idempotent. + * + * @internal + */ + #replaying(): Dispose { + const source = {}; + this.#state.replaying.mutate((sources) => { + sources.add(source); + }); + return () => { + if (!this.#state.replaying.peek().has(source)) return; + this.#state.replaying.mutate((sources) => { + sources.delete(source); + }); + }; + } + /** * Resolves once anything a serving session scans changes: the open requests, or either * side of the routing table. @@ -1168,7 +1192,9 @@ export class Consumer { /** * The announced routes matching `scope`, as a live stream: every currently advertised - * route arrives first as active, then additions and retractions as they happen. + * route arrives first as `announced`, then the `live` marker, then changes as they + * happen. The marker also waits for every session still replaying its peer's initial + * set when the stream opened, so a caller listing what is live stops there. * Any pattern is accepted. A local broadcast appears once it announces, exactly as a * peer sees it. A dynamic or received route announces the prefix it covers when its * subtree overlaps the scope. The stream ends when the origin closes or the consumer is @@ -1205,6 +1231,10 @@ export class Consumer { // update. let active = new Map(); + // The sessions replaying into the table when the stream opened; the live marker + // follows the diff after the last of them lands. Undefined once it was delivered. + let waiting: Set | undefined = new Set(this.#state.replaying.peek()); + try { for (;;) { const local = this.#state.local.peek(); @@ -1239,7 +1269,24 @@ export class Consumer { } active = next; - await Signal.race(this.#state.local, this.#state.advertisedLocal, this.#state.routes, producer.closed); + if (waiting) { + const replaying = this.#state.replaying.peek(); + for (const source of waiting) { + if (!replaying.has(source)) waiting.delete(source); + } + if (waiting.size === 0) { + producer.append({ kind: "live" }); + waiting = undefined; + } + } + + await Signal.race( + this.#state.local, + this.#state.advertisedLocal, + this.#state.routes, + this.#state.replaying, + producer.closed, + ); if (producer.closed.peek() !== undefined) return; } } catch { diff --git a/js/net/src/wire.ts b/js/net/src/wire.ts index 81dfe0f9c7..055b5b13ec 100644 --- a/js/net/src/wire.ts +++ b/js/net/src/wire.ts @@ -31,6 +31,8 @@ export interface OriginProducer { ): origin.Dynamic; attach(discovery: boolean): Dispose; expect(): Dispose; + /** Hold the live marker of announcement streams opened now, until the peer's initial set lands. */ + replaying(): Dispose; readonly requests: Getter | undefined>; changed(): GetPromise; answer(path: Path.Valid, front: broadcast.Consumer): Dispose | undefined; diff --git a/js/room/src/room.test.ts b/js/room/src/room.test.ts index 7ddc3140c0..a0f6c51f4c 100644 --- a/js/room/src/room.test.ts +++ b/js/room/src/room.test.ts @@ -17,7 +17,7 @@ test("room restores the announce prefix and reconciles local identity changes", origin: new Signal({ announced(scope: Net.Path.Pattern) { expect(scope.equals(Net.Path.Pattern.subtree(Net.Path.from("room-a")))).toBe(true); - let update: Net.Announce.Update | undefined = { + let update: Net.Announce.Event | undefined = { prefix: Net.Path.from("room-a/bob/camera.hang"), captures: [Net.Path.Pattern.literal(Net.Path.from("bob/camera.hang"))], kind: "announced", diff --git a/js/room/src/room.ts b/js/room/src/room.ts index 0cf2a48ba0..0ab7b92229 100644 --- a/js/room/src/room.ts +++ b/js/room/src/room.ts @@ -82,6 +82,7 @@ export class Room { for (;;) { const update = await effect.race(announced.next()); if (!update) break; + if (update.kind === "live") continue; // The scope's `**` captures what lies beneath the prefix. A broad route // that cannot pin that suffix names no participant to open. @@ -95,7 +96,7 @@ export class Room { const local = this.identity.peek(); if (local && parsed.identity === local) continue; - if (Moq.Announce.isActive(update.kind)) { + if (update.kind !== "retracted") { this.#add(parsed.identity, parsed.kind, Moq.Path.from(covered)); } else { this.#remove(parsed.identity, parsed.kind); diff --git a/js/watch/src/broadcast.ts b/js/watch/src/broadcast.ts index 1064a0708c..84e5ca8254 100644 --- a/js/watch/src/broadcast.ts +++ b/js/watch/src/broadcast.ts @@ -2,7 +2,7 @@ import * as Catalog from "@moq/hang/catalog"; import * as Json from "@moq/json"; import * as Msf from "@moq/msf"; import type * as Moq from "@moq/net"; -import { Announce, Error as NetError, Path } from "@moq/net"; +import { Error as NetError, Path } from "@moq/net"; import { Effect, type Getter, getter, type Inputs, type Readonlys, readonlys, Signal } from "@moq/signals"; import { toHang } from "./msf"; @@ -202,10 +202,11 @@ export class Broadcast { for (;;) { const entry = await effect.race(announced.next()); if (!entry) break; + if (entry.kind === "live") continue; this.#announced.mutate((active) => { if (!active) return; - if (Announce.isActive(entry.kind)) active.add(entry.prefix); - else active.delete(entry.prefix); + if (entry.kind === "retracted") active.delete(entry.prefix); + else active.add(entry.prefix); }); } }); diff --git a/quest/m1/README.md b/quest/m1/README.md index f7c6309748..50e49410fa 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -31,7 +31,6 @@ transport, benchmark tooling); worktrees isolate commits, not semantics. - [Relay peer set](/quest/m1/relay-peer-set.md) - a wire consumer tells a client hop from a peer hop, and every mesh credential can mark a peer - [JS origin scope](/quest/m1/js-origin-scope.md) - a scoped `@moq/net` origin subscribes only to its allowed prefixes - [Session close](/quest/m1/session-close.md) - a graceful session end withdraws announces and waits one second for the ack -- [JS caught up](/quest/m1/js-announce-caught-up.md) - @moq/net's announce consumer says when the initial set has landed, like Rust - [Bindings caught up](/quest/m1/announce-live-bindings.md) - moq-ffi, libmoq, and every wrapper yield the same flat announce event, `Live` included - [IETF announce count](/quest/m1/ietf-announce-count.md) - an opt-in moq-transport extension carries the replay count, so IETF announce consumers go live without a timer diff --git a/quest/m1/announce-live-bindings.md b/quest/m1/announce-live-bindings.md index a4823ceee8..39a1d88e9a 100644 --- a/quest/m1/announce-live-bindings.md +++ b/quest/m1/announce-live-bindings.md @@ -22,7 +22,3 @@ Mirror moq-net's `announce::Event` one to one: Public API: breaking in moq-ffi, libmoq's C ABI, and every wrapper, so it retargets to `dev` with the Rust break. Wire: none. - -## Related - -- [JS caught up](/quest/m1/js-announce-caught-up.md) - the same event in `@moq/net` diff --git a/quest/m1/js-announce-caught-up.md b/quest/m1/js-announce-caught-up.md deleted file mode 100644 index b537acba01..0000000000 --- a/quest/m1/js-announce-caught-up.md +++ /dev/null @@ -1,16 +0,0 @@ -# [S] @moq/net announce consumers know when they have caught up - -## Goal - -`@moq/net`'s announce consumer yields the same `Live` marker as the Rust -`announce::Consumer`, with the same ordering and per-source semantics, so a browser app can render "no -broadcasts" instead of a spinner that never resolves. - -## Plan - -Mirror the Rust shape (one flat `announce::Event`: `Announced`, `Updated`, -`Retracted`, each holding an `Announce`, or `Live`), its -per-source guards (a session holds one per announce stream until the count, -ANNOUNCE_INIT, or a quiet stream lands it), and the marker-less fallback. Test the same cases -in JS. Public API: the announce consumer's yield type changes, so it lands -with the Rust break. Wire: none. diff --git a/quest/m1/js-origin-scope.md b/quest/m1/js-origin-scope.md index 9bc8530c60..674ade77db 100644 --- a/quest/m1/js-origin-scope.md +++ b/quest/m1/js-origin-scope.md @@ -23,5 +23,4 @@ origin's scope. No new page. ## Related -- [JS caught up](/quest/m1/js-announce-caught-up.md) - when the replay of those prefixes has landed - [IETF announce count](/quest/m1/ietf-announce-count.md) - how an IETF consumer knows that replay is finished From aa7a4e7bd917aee035cb4ece63eef79b951cad86 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 26 Sep 2026 19:47:58 -0700 Subject: [PATCH 3/6] fix(js/net): scope each replay hold to its interest prefix A scoped origin opens one announce stream per interest, and each took a global replay hold, so a `room/**` listing waited on an unrelated `other/**` replay. Like Rust's `Producer::replaying(prefix)`, a hold now carries the absolute patterns it may announce under, and an announcement stream only waits on holds that overlap its scope. Co-Authored-By: Claude Opus 5.5 --- js/net/src/connection/forward.ts | 5 +++-- js/net/src/origin.test.ts | 36 ++++++++++++++++++++++++++++---- js/net/src/origin.ts | 31 ++++++++++++++++++--------- js/net/src/wire.ts | 7 +++++-- 4 files changed, 61 insertions(+), 18 deletions(-) diff --git a/js/net/src/connection/forward.ts b/js/net/src/connection/forward.ts index a9ce93f455..efe11a69e6 100644 --- a/js/net/src/connection/forward.ts +++ b/js/net/src/connection/forward.ts @@ -53,8 +53,9 @@ export function forwardAnnounced(conn: Established, origin: OriginProducer): voi const inserted = new Map(); // Taken before the session is handed out, so no announcement stream on the origin misses - // it: each one opened now withholds its live marker until this interest's initial set lands. - const landed = originWire.replaying(); + // it: each overlapping one opened now withholds its live marker until this interest's + // initial set lands. + const landed = originWire.replaying(prefix); // End the stream the moment the session closes rather than waiting for the wire to // error it, so the retractions below land promptly. diff --git a/js/net/src/origin.test.ts b/js/net/src/origin.test.ts index c4aab03844..ba6bc5204d 100644 --- a/js/net/src/origin.test.ts +++ b/js/net/src/origin.test.ts @@ -321,7 +321,7 @@ test("an empty origin is live at once, and the marker never repeats", async () = expect(await announced.next()).toEqual({ kind: "live" }); // Later routes are live changes, even past a session that replays and lands. - const landed = wireOf(origin).replaying(); + const landed = wireOf(origin).replaying(Path.empty()); landed(); const alice = publish(origin, Path.from("alice")); expect(await announced.next()).toMatchObject({ prefix: Path.from("alice"), kind: "announced" }); @@ -353,8 +353,8 @@ test("live follows the replayed routes", async () => { test("a replaying session withholds live until it lands", async () => { const origin = new Producer(); - const first = wireOf(origin).replaying(); - const second = wireOf(origin).replaying(); + const first = wireOf(origin).replaying(Path.empty()); + const second = wireOf(origin).replaying(Path.empty()); const announced = origin.announced(); // Routes a session lands arrive before the marker. @@ -368,7 +368,7 @@ test("a replaying session withholds live until it lands", async () => { expect(await pending(next)).toBe(true); // A session that starts replaying after the stream opened is not waited on. - const late = wireOf(origin).replaying(); + const late = wireOf(origin).replaying(Path.empty()); second(); expect(await next).toEqual({ kind: "live" }); @@ -377,6 +377,34 @@ test("a replaying session withholds live until it lands", async () => { origin.close(); }); +test("a stream waits only on sessions replaying into its scope", async () => { + const origin = new Producer(); + const room = wireOf(origin).replaying(Path.from("room")); + const other = wireOf(origin).replaying(Path.from("other")); + + // Nothing replays under `lobby`, and `other` cannot announce into `room/**`. + const lobby = origin.announced(Path.Pattern.parse("lobby/**")); + expect(await lobby.next()).toEqual({ kind: "live" }); + const announced = origin.announced(Path.Pattern.parse("room/**")); + const next = announced.next(); + expect(await pending(next)).toBe(true); + room(); + expect(await next).toEqual({ kind: "live" }); + + // A scoped handle rebases the prefix under its root before comparing. + const scoped = origin.scope(Path.from("tenant"), new Path.Patterns([Path.Pattern.all()])); + const inner = wireOf(scoped).replaying(Path.from("room")); + const outer = origin.announced(Path.Pattern.parse("room/**")); + expect(await outer.next()).toEqual({ kind: "live" }); + const tenant = origin.announced(Path.Pattern.parse("tenant/room/**")); + expect(await pending(tenant.next())).toBe(true); + + inner(); + other(); + for (const stream of [lobby, announced, outer, tenant]) stream.close(); + origin.close(); +}); + test("announced streams the table under a scope with origin-relative paths", async () => { const origin = new Producer(); const consumer = origin.consume(); diff --git a/js/net/src/origin.ts b/js/net/src/origin.ts index e7a49efab5..ca3b6b5a43 100644 --- a/js/net/src/origin.ts +++ b/js/net/src/origin.ts @@ -412,9 +412,10 @@ class OriginState { // this is coming" and "nothing here can ever serve you". answerers = new Signal(0); - // Sessions still replaying the peer's initial announce set into the table. An - // announcement stream opened while one replays withholds its live marker until it lands. - replaying = new Signal>(new Set()); + // Sessions still replaying the peer's initial announce set into the table, each with the + // absolute paths it may announce under. An announcement stream opened while an + // overlapping one replays withholds its live marker until it lands. + replaying = new Signal>(new Set()); closed = new Once(); @@ -656,7 +657,7 @@ export class Producer implements Table { this.#scope.allowed.overlaps(Path.Pattern.subtree(Path.join(this.#scope.root, prefix))), attach: (discovery) => this.#attach(discovery), expect: () => this.#expect(), - replaying: () => this.#replaying(), + replaying: (prefix) => this.#replaying(prefix), get requests() { if (thisProducer.#scope === Scope.all) return thisProducer.#state.requests; thisProducer.#requests ??= new Derived([thisProducer.#state.requests], (requests) => @@ -887,13 +888,21 @@ export class Producer implements Table { } /** - * Register a session replaying the peer's initial announce set into the table. Returns - * the release; call it once the set has landed, or the session dies. Idempotent. + * Register a session replaying the peer's initial announce set under `prefix`, relative + * to this handle's root. Only announcement streams overlapping it wait. Returns the + * release; call it once the set has landed, or the session dies. Idempotent. * * @internal */ - #replaying(): Dispose { - const source = {}; + #replaying(prefix: Path.Valid): Dispose { + let scope: Path.Patterns; + try { + scope = this.#scope.patterns(Path.Pattern.subtree(prefix)); + } catch { + // A subtree too complex to intersect waits on the whole scope instead, like Rust. + scope = this.#scope.allowed ?? new Path.Patterns([Path.Pattern.all()]); + } + const source = { scope }; this.#state.replaying.mutate((sources) => { sources.add(source); }); @@ -1397,9 +1406,11 @@ export class Consumer { // update. let active = new Map(); - // The sessions replaying into the table when the stream opened; the live marker + // The sessions replaying into this stream's scope when it opened; the live marker // follows the diff after the last of them lands. Undefined once it was delivered. - let waiting: Set | undefined = new Set(this.#state.replaying.peek()); + let waiting: Set<{ scope: Path.Patterns }> | undefined = new Set( + [...this.#state.replaying.peek()].filter(({ scope }) => [...scope].some((p) => patterns.overlaps(p))), + ); try { for (;;) { diff --git a/js/net/src/wire.ts b/js/net/src/wire.ts index 1249767fa9..38b5a87143 100644 --- a/js/net/src/wire.ts +++ b/js/net/src/wire.ts @@ -35,8 +35,11 @@ export interface OriginProducer { ): origin.Dynamic; attach(discovery: boolean): Dispose; expect(): Dispose; - /** Hold the live marker of announcement streams opened now, until the peer's initial set lands. */ - replaying(): Dispose; + /** + * Hold the live marker of announcement streams opened now that overlap `prefix`, until the + * peer's initial set under it lands. + */ + replaying(prefix: Path.Valid): Dispose; readonly requests: Getter | undefined>; changed(): GetPromise; answer(path: Path.Valid, front: broadcast.Consumer): Dispose | undefined; From 90a42931cb930779c9e93e98b9a7ddb1e10f51c6 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 26 Sep 2026 20:32:07 -0700 Subject: [PATCH 4/6] perf(js/net): stop watching replay holds once a stream is live Past the live marker a stream no longer reads the replay holds, but it still raced on them, so every reconnect's hold rescanned the table once for every open stream. Co-Authored-By: Claude Opus 5.5 --- js/net/src/origin.ts | 11 ++++------- 1 file changed, 4 insertions(+), 7 deletions(-) diff --git a/js/net/src/origin.ts b/js/net/src/origin.ts index ca3b6b5a43..87758ec648 100644 --- a/js/net/src/origin.ts +++ b/js/net/src/origin.ts @@ -1457,13 +1457,10 @@ export class Consumer { } } - await Signal.race( - this.#state.local, - this.#state.advertisedLocal, - this.#state.routes, - this.#state.replaying, - producer.closed, - ); + // Replay holds only matter until the marker is out; past it, a reconnect's holds + // must not rescan the table for every stream already live. + const table = [this.#state.local, this.#state.advertisedLocal, this.#state.routes, producer.closed]; + await (waiting ? Signal.race(...table, this.#state.replaying) : Signal.race(...table)); if (producer.closed.peek() !== undefined) return; } } catch { From 0367dc98150f5b17740386bcacf3d00b2280d593 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 26 Sep 2026 21:22:37 -0700 Subject: [PATCH 5/6] docs(js/net): the discovery summary mentions the live marker Co-Authored-By: Claude Opus 5.5 --- doc/lib/js/net.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/doc/lib/js/net.md b/doc/lib/js/net.md index c8b550dff8..350314dee4 100644 --- a/doc/lib/js/net.md +++ b/doc/lib/js/net.md @@ -50,7 +50,7 @@ for (;;) { - **Origins** hold the broadcasts, not the connection: closing a session unannounces them but leaves them created for the next one. `origin.request(path)` resolves an announced local broadcast with no round trip, so a page that watches what it publishes reads its own copy, unless a cheaper route announces the same path. A refusal from the most specific route ends the request: `request.closed` settles with the handler's error and no broader route is asked. Create, populate, then `announce()` for an exact path; use `dynamic(prefix, route)` when the set of paths is not known: an exact-path subscribe before the tracks exist is refused, and nobody, local or remote, can see or reach a broadcast until it announces. - **Connections** race WebTransport against WebSocket. `new Connection({ url })` pools one connection per relay URL and reconnects with backoff, which the elements use. Supplying WebTransport/WebSocket options, discovery, delay, or a caller-owned origin selects a private loop; explicit `share: true` refuses those options. `closed` settles when the handle is released (`null` on a clean close); the failure that stopped retrying the current URL is `error`, and a new URL recovers the same handle. A connection owns one send-rate sampler and one `Bandwidth.Allocator`; publishers reserve against it so their encoder targets sum to the estimate instead of each matching it. - **Bandwidth** (`Bandwidth.Allocator`) divides the connection's send-rate estimate by track priority, max-min fair within a tier. An idle track claims nothing. The receive side is untouched. -- **Discovery** by any pattern scope (`origin.announced(scope)`, such as `room/*/chat`; default everything). Each event's `prefix` is the covered prefix relative to the origin, `captures` reports what the scope's wildcards matched when the prefix pins them, and `kind` says whether it was announced, updated, or retracted. The consumer is an async iterable. `origin.broadcasts(scope)` is a live `Getter>` of the same covered prefixes for UIs that need the current set. A borrowed `Connection.origin` also exposes `dynamic(prefix, route)` for serving paths on demand. +- **Discovery** by any pattern scope (`origin.announced(scope)`, such as `room/*/chat`; default everything). A route event's `kind` says whether it was announced, updated, or retracted; its `prefix` is the covered prefix relative to the origin, and `captures` reports what the scope's wildcards matched when the prefix pins them. One `{ kind: "live" }` marker, with no prefix, follows the routes live at subscribe time. The consumer is an async iterable. `origin.broadcasts(scope)` is a live `Getter>` of the same covered prefixes for UIs that need the current set. A borrowed `Connection.origin` also exposes `dynamic(prefix, route)` for serving paths on demand. - **Publisher retention** is optional `Track.Info.maxAge`: omission sets no limit, while zero keeps the live edge. No implicit five-second window is applied. - **Subscriptions** carry a priority, a `Time.Milli` max age, and optional `groups` bounds. Groups arrive out of order and are read frame by frame, with `Error.TooFarBehind` when a reader asks for a frame the group never held and `Error.GroupTooLarge` when a write exceeds the cache budget and aborts the group. - **Track ends**: `close()` ends a track at its live edge, while `finishAt(n)` declares the exclusive end ahead of it and still accepts the groups below. A subscriber reads the end with `final()` or awaits `finished()`. A remote track ends only once every group below its end has arrived or was dropped; one reset before its header arrived is skipped after the subscription's max age on moq-lite (one second without one), or after one second on IETF. From 9d1ccb66c79cceb980c4bd9810fcccb29deb9b12 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 27 Sep 2026 16:07:54 -0700 Subject: [PATCH 6/6] quest: plan the live marker's page-load and empty-state follow-ups Co-Authored-By: Claude Opus 5.5 --- quest/m1/README.md | 2 ++ quest/m1/announce-empty-state.md | 16 ++++++++++++++++ quest/m1/announce-page-load.md | 19 +++++++++++++++++++ 3 files changed, 37 insertions(+) create mode 100644 quest/m1/announce-empty-state.md create mode 100644 quest/m1/announce-page-load.md diff --git a/quest/m1/README.md b/quest/m1/README.md index cb05aacf18..8a836abbc0 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -31,6 +31,8 @@ transport, benchmark tooling); worktrees isolate commits, not semantics. - [Close codes](/quest/m1/close-codes.md) - a client sees the peer's application close code over WebSocket and raw QUIC, like WebTransport - [Bindings caught up](/quest/m1/announce-live-bindings.md) - moq-ffi, moq-c, and every wrapper yield the same flat announce event, `Live` included - [JS active count](/quest/m1/js-active-count.md) - @moq/net speaks MoQ Active Count, so its IETF announce consumers go live without a timer +- [Page-load marker](/quest/m1/announce-page-load.md) - an announcement stream opened before the first connection waits for its replay before `live` +- [Empty state](/quest/m1/announce-empty-state.md) - watch, room, and the demo show "no broadcasts" once `live` arrives with nothing announced - [Publish delay](/quest/m1/publish-delay.md) - js/publish encoders advertise `delay` behind the earliest rendition, like moq-mux - [Data jitter](/quest/m1/data-jitter.md) - JSON and binary tracks with a capture time advertise a detected `delay` and `jitter` diff --git a/quest/m1/announce-empty-state.md b/quest/m1/announce-empty-state.md new file mode 100644 index 0000000000..54e4adbba8 --- /dev/null +++ b/quest/m1/announce-empty-state.md @@ -0,0 +1,16 @@ +# [S] Watch, room, and the demo show when nothing is live + +## Goal + +Once an announcement stream reports `live` with nothing announced, `js/watch`, +`js/room`, and `demo/web` show a "no broadcasts" state instead of a spinner +that never resolves. + +## Plan + +- Today each consumer skips the `live` marker. Track it next to the + announced set and render the empty state only after it arrives. + +## Required + +- [Page-load marker](/quest/m1/announce-page-load.md) - otherwise the empty state flashes on page load before the first connection replays diff --git a/quest/m1/announce-page-load.md b/quest/m1/announce-page-load.md new file mode 100644 index 0000000000..8cf4b4cfb5 --- /dev/null +++ b/quest/m1/announce-page-load.md @@ -0,0 +1,19 @@ +# [S] The live marker waits for the first connection on page load + +## Goal + +An announcement stream opened before the first session connects does not +report `live` until that session's replay lands, or the connection attempt +gives up. Today it has no session to wait on, so it goes `live` at once and +the broadcasts arrive after it, in both `@moq/net` and `rs/moq-net`. + +## Plan + +- Open question for the maintainer: what counts as giving up (the first failed + attempt, the backoff ceiling, or never). +- The JS reconnect loop (`js/net/src/connection/reload.ts`) already counts as + an answerer for requests through `expect()`; it can also take a replay hold + on the origin until its first session lands or it gives up. +- Rust has no reconnect loop of its own; decide whether an app that opens a + stream before connecting needs an equivalent hold there. +- Add a caught-up test that opens the stream before the first connection.