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 d00b083553..1f3f163efd 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -139,16 +139,16 @@ 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 { prefix, 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, and every binding over moq-ffi or moq-c, 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`, -moq-lite-01/02 send them in `ANNOUNCE_INIT`, IETF draft-16+ sessions count them -in `REQUEST_OK` when both sides speak +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, and every binding over +moq-ffi or moq-c, 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`, Rust IETF draft-16+ sessions count +them in `REQUEST_OK` when both sides speak [active-count](/draft/moq-active-count), and anything else waits for the stream to go quiet. diff --git a/doc/lib/js/net.md b/doc/lib/js/net.md index 0ce30a89e2..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. @@ -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 f115c38ca4..3b436e7fef 100644 --- a/js/net/src/connection/forward.test.ts +++ b/js/net/src/connection/forward.test.ts @@ -189,6 +189,55 @@ test("a request replaced across one coalesced wakeup still gets answered", 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("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(); +}); + test("scoped origins subscribe to disjoint literal heads with hidden routes", () => { const origin = new OriginProducer().scope( Path.empty(), diff --git a/js/net/src/connection/forward.ts b/js/net/src/connection/forward.ts index f594d65fa5..efe11a69e6 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 * as Path from "../path.ts"; import { wireOf } from "../wire.ts"; @@ -53,6 +52,11 @@ export function forwardAnnounced(conn: Established, origin: OriginProducer): voi const announced = conn.announced(Path.Pattern.subtree(prefix), { hidden: true }); const inserted = new Map(); + // Taken before the session is handed out, so no announcement stream on the origin misses + // 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. void conn.closed.then(() => announced.close()); @@ -63,9 +67,13 @@ export function forwardAnnounced(conn: Established, origin: OriginProducer): voi for (;;) { const event = await announced.next(); if (!event) break; + if (event.kind === "live") { + landed(); + continue; + } if (!originWire.accepts(event.prefix)) continue; - if (isActive(event.kind)) { + if (event.kind !== "retracted") { const existing = inserted.get(event.prefix); if (existing) { existing.update(event.route); @@ -85,6 +93,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/connection/scope.test.ts b/js/net/src/connection/scope.test.ts index 01bf4ab52a..3fc2df36be 100644 --- a/js/net/src/connection/scope.test.ts +++ b/js/net/src/connection/scope.test.ts @@ -33,11 +33,12 @@ for (const protocol of [Lite.ALPN_07_WIP, Ietf.ALPN.DRAFT_19]) { const received: string[] = []; for (let index = 0; index < 3; index++) { const update = await withTimeout(announced.next(), 1000, "scoped announcement did not arrive"); - if (!update) throw new Error("announcement stream ended before replay"); - expect(update.kind).toBe("announced"); + if (update?.kind !== "announced") throw new Error(`expected an announcement, got ${update?.kind}`); received.push(update.prefix); } expect(received.sort()).toEqual(["client/other/live", "client/room/.hidden", "client/room/live"]); + // The marker waits for every interest's initial set, not just the first to land. + expect(await withTimeout(announced.next(), 1000, "live marker did not arrive")).toEqual({ kind: "live" }); expect([...consume.broadcasts().peek().keys()].sort()).toEqual([ Path.from("other/live"), Path.from("room/live"), 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 cf8418c0de..dd75dc752e 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 057f8cda98..e8bf9da7f0 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 a85937941b..ba6bc5204d 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(); @@ -298,6 +308,103 @@ 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(Path.empty()); + 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(Path.empty()); + const second = wireOf(origin).replaying(Path.empty()); + 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(Path.empty()); + second(); + expect(await next).toEqual({ kind: "live" }); + + late(); + announced.close(); + 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(); @@ -307,20 +414,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(); }); @@ -337,9 +444,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(); @@ -365,10 +472,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(); @@ -386,10 +493,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(); @@ -417,7 +524,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(); @@ -541,12 +648,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(); @@ -614,7 +721,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(); @@ -628,12 +735,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(); @@ -995,7 +1102,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 } }, @@ -1007,18 +1114,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(); @@ -1040,12 +1147,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(); @@ -1091,7 +1198,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")); @@ -1118,11 +1225,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(); }); @@ -1415,7 +1522,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"]); @@ -1479,7 +1586,7 @@ test("a rooted producer shares the table and enforces its pattern union", async expect([...nested.broadcasts().peek().keys()]).toEqual([Path.from("alice")]); expect(() => nested.createBroadcast(Path.from("alice/deep"))).toThrow("outside the origin scope"); const announced = nested.announced(); - expect((await announced.next())?.prefix).toBe(Path.from("alice")); + expect(await announced.next()).toMatchObject({ kind: "announced", prefix: Path.from("alice") }); announced.close(); request.close(); broadcast.close(); diff --git a/js/net/src/origin.ts b/js/net/src/origin.ts index 741e43e304..87758ec648 100644 --- a/js/net/src/origin.ts +++ b/js/net/src/origin.ts @@ -412,6 +412,11 @@ 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, 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(); /** @@ -652,6 +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: (prefix) => this.#replaying(prefix), get requests() { if (thisProducer.#scope === Scope.all) return thisProducer.#state.requests; thisProducer.#requests ??= new Derived([thisProducer.#state.requests], (requests) => @@ -881,6 +887,33 @@ export class Producer implements Table { }; } + /** + * 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(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); + }); + 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. @@ -1321,7 +1354,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 @@ -1371,6 +1406,12 @@ export class Consumer { // update. let active = new Map(); + // 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<{ scope: Path.Patterns }> | undefined = new Set( + [...this.#state.replaying.peek()].filter(({ scope }) => [...scope].some((p) => patterns.overlaps(p))), + ); + try { for (;;) { const local = this.#state.local.peek(); @@ -1405,7 +1446,21 @@ 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; + } + } + + // 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 { diff --git a/js/net/src/wire.ts b/js/net/src/wire.ts index af77e0cde7..38b5a87143 100644 --- a/js/net/src/wire.ts +++ b/js/net/src/wire.ts @@ -35,6 +35,11 @@ export interface OriginProducer { ): origin.Dynamic; attach(discovery: boolean): Dispose; expect(): 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; 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 1328d866d7..115ea483da 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -29,8 +29,9 @@ transport, benchmark tooling); worktrees isolate commits, not semantics. - [Track demand](/quest/m1/track-demand.md) - Rust and JS watch a track's subscribers through `demand()` alone - [Session close](/quest/m1/session-close.md) - a graceful session end withdraws announces and waits one second for the ack - [Close codes](/quest/m1/close-codes.md) - a client sees the peer's application close code over WebSocket and raw QUIC, like WebTransport -- [JS caught up](/quest/m1/js-announce-caught-up.md) - @moq/net's announce consumer says when the initial set has landed, like Rust - [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. diff --git a/quest/m1/js-active-count.md b/quest/m1/js-active-count.md index 9f9ae15ce7..1730c213c9 100644 --- a/quest/m1/js-active-count.md +++ b/quest/m1/js-active-count.md @@ -21,7 +21,3 @@ the count instead of the quiet timer. land without waiting out the quiet gap. Public API: none. Wire: the extension, already specified. - -## Required - -- [JS caught up](/quest/m1/js-announce-caught-up.md) - the per-source guard the count lands 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/wildcard/README.md b/quest/m1/wildcard/README.md index e7be970d69..05db0e70d0 100644 --- a/quest/m1/wildcard/README.md +++ b/quest/m1/wildcard/README.md @@ -54,7 +54,7 @@ same exact containment check. `Cost { warm, cold }` [moq#3225](https://github.com/moq-dev/moq/pull/3225) moved a long way toward this. An announcement carries a `Pattern` covering a set of paths. Rust -`announce::Update.pattern` and TypeScript `Announce.Update.pattern` use the +`announce::Announce` and TypeScript `Announce.Announce` use the matcher directly, so callers explicitly select prefix-shaped claims when they need a concrete broadcast path.