From 94fc9742b51bf9bb60380355b127c1cc91fff4a6 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 3 Aug 2026 17:49:46 -0700 Subject: [PATCH 1/9] feat(net): add announcedBroadcast, a reactive handle to one broadcast Consuming a path nobody publishes gets the subscription reset, so a consumer that starts before the publisher stays silent forever unless it retries. Every caller that cared has hand-rolled the same announce-stream drain to avoid it; add the primitive instead. Co-Authored-By: Claude Opus 5 --- js/net/README.md | 1 + js/net/examples/wait.ts | 41 ++++++++++ js/net/src/announced.ts | 117 ++++++++++++++++++++++++++- js/net/src/connection/established.ts | 16 +++- js/net/src/connection/reload.test.ts | 59 ++++++++++++++ js/net/src/connection/reload.ts | 15 ++++ js/net/src/ietf/connection.ts | 12 ++- js/net/src/integration.test.ts | 94 +++++++++++++++++++++ js/net/src/lite/connection.ts | 6 +- js/watch/src/broadcast.ts | 50 +++--------- 10 files changed, 370 insertions(+), 41 deletions(-) create mode 100644 js/net/examples/wait.ts diff --git a/js/net/README.md b/js/net/README.md index 67860a63e4..091ab62a3d 100644 --- a/js/net/README.md +++ b/js/net/README.md @@ -60,6 +60,7 @@ await quicheLoaded; //This is a promise, connect after it resolves - **[Publishing](examples/publish.ts)** - Publish data to a broadcast - **[Subscribing](examples/subscribe.ts)** - Subscribe to and receive broadcast data - **[Discovery](examples/discovery.ts)** - Discover broadcasts announced by the server +- **[Waiting](examples/wait.ts)** - Wait for one known broadcast to come online, and follow it - **[Server side usage](https://github.com/sb2702/webcodecs-examples/tree/main/src/moq-server)** - Publish from browser to a server ## License diff --git a/js/net/examples/wait.ts b/js/net/examples/wait.ts new file mode 100644 index 0000000000..7bf32c21c5 --- /dev/null +++ b/js/net/examples/wait.ts @@ -0,0 +1,41 @@ +import * as Moq from "@moq/net"; +import { Effect } from "@moq/signals"; + +async function main() { + const url = new URL("https://cdn.moq.dev/anon"); + const connection = new Moq.Connection.Reload({ url, enabled: true }); + + // Wait for a broadcast that may not exist yet. `consume` would subscribe blind and get reset + // if nobody is publishing the path; this waits for the announcement instead. + const broadcast = connection.announcedBroadcast(Moq.Path.from("my-broadcast")); + + const effect = new Effect(); + effect.run((effect) => { + // Re-runs every time the broadcast comes online or goes away, including across reconnects + // and same-name republishes. + const active = effect.get(broadcast.active); + if (!active) { + console.log("broadcast is offline"); + return; + } + + console.log("broadcast is live"); + const track = active.track("chat").subscribe({ priority: 0 }); + effect.cleanup(() => track.close()); + + effect.spawn(async () => { + for (;;) { + const group = await Promise.race([effect.cancel, track.recvGroup()]); + if (!group) break; + console.log("received:", await group.readString()); + } + }); + }); + + // Run until interrupted, then release the handle and the connection. + await connection.closed; + effect.close(); + broadcast.close(); +} + +main().catch(console.error); diff --git a/js/net/src/announced.ts b/js/net/src/announced.ts index f2fdfc1c7f..0c9e9d3784 100644 --- a/js/net/src/announced.ts +++ b/js/net/src/announced.ts @@ -3,7 +3,9 @@ * * @module */ -import { type GetPromise, Once, Signal } from "@moq/signals"; +import { Effect, type GetPromise, type Getter, type GetterInit, getter, Once, Signal } from "@moq/signals"; +import type * as broadcast from "./broadcast.js"; +import type { Established } from "./connection/established.js"; import * as Path from "./path.js"; /** @@ -126,3 +128,116 @@ export class Consumer { closeState(this.#state, abort); } } + +// Connections already warned about missing broadcast discovery, so the fallback logs at most +// once per connection instead of once per watched path. +const warnedNoDiscovery = new WeakSet(); + +/** + * A reactive handle to a single broadcast: {@link Broadcast.active} holds a live + * {@link broadcast.Consumer} while the path is announced and `undefined` while nobody + * publishes it. + * + * Use this instead of {@link Established.consume} whenever the broadcast may not exist yet. + * Subscribing to a path nobody publishes gets the stream reset, so a consumer that races the + * publisher stays silent forever unless it retries; this waits for the announcement instead. + * + * The handle re-consumes on every (re-)announce, so a same-name republish (a new publisher, or + * a relay-failover RESTART) re-attaches to the new instance rather than clinging to the dead + * one. Built from a reconnecting `Connection.Reload`, it also spans reconnects: the + * broadcast drops to `undefined` while disconnected and resolves again once the new connection + * announces it. + * + * Falls back to consuming blind (and warns once) on a relay without + * {@link Established.discovery}, where there is no announcement to wait for. + * + * Close it to release the announcement stream and the current broadcast. + * + * @public + */ +export class Broadcast { + /** The broadcast path this handle watches. */ + readonly path: Path.Valid; + + /** The live broadcast, or `undefined` while it is offline. */ + readonly active: Getter; + + #active = new Signal(undefined); + #signals = new Effect(); + + /** + * Watch `path` on a connection. Accepts a live {@link Established} session or a reactive + * one (a `Connection.Reload`'s `established`), which is how the handle survives reconnects. + * + * Obtain one from `announcedBroadcast(path)` on either connection type rather than + * constructing it directly. + */ + constructor(connection: GetterInit, path: Path.Valid) { + this.path = path; + this.active = this.#active; + + const source = getter(connection); + this.#signals.run((effect) => { + const conn = effect.get(source); + if (!conn) return; + + // Without discovery no announcement ever arrives, so waiting would hang forever. + if (!conn.discovery) { + if (!warnedNoDiscovery.has(conn)) { + warnedNoDiscovery.add(conn); + console.warn("relay does not support broadcast discovery; consuming without waiting."); + } + + const blind = conn.consume(path); + effect.cleanup(() => blind.close()); + effect.set(this.#active, blind, undefined); + return; + } + + const announced = conn.announced(path); + effect.cleanup(() => announced.close()); + + let current: broadcast.Consumer | undefined; + const offline = () => { + current?.close(); + current = undefined; + this.#active.set(undefined); + }; + effect.cleanup(offline); + + effect.spawn(async () => { + try { + for (;;) { + const event = await Promise.race([effect.cancel, announced.next()]); + if (!event) break; + + // Scoped to `path`, so the exact broadcast arrives with an empty suffix; ignore children. + if (event.path !== Path.empty()) continue; + + if (event.active) { + // A live subscription survives a redundant (re-)announce; only replace a dead one. + if (current && current.closed.peek() === undefined) continue; + current?.close(); + current = conn.consume(path); + this.#active.set(current); + } else { + offline(); + } + } + } catch (err) { + // The stream was reset, which means the session died under it. + console.debug("announcement stream reset", err); + } + + // The stream ended, or this run was torn down (its cleanup already ran). Either + // way there is nothing left announcing the path, so don't hold a dead broadcast. + offline(); + }); + }); + } + + /** Closes the handle and the broadcast it currently holds. Idempotent. */ + close() { + this.#signals.close(); + } +} diff --git a/js/net/src/connection/established.ts b/js/net/src/connection/established.ts index bc1ac61736..89bda93169 100644 --- a/js/net/src/connection/established.ts +++ b/js/net/src/connection/established.ts @@ -35,9 +35,23 @@ export interface Established { /** Publish a broadcast at the given path. */ publish(path: Path.Valid, broadcast: broadcast.Producer): void; - /** Consume the broadcast at the given path. */ + /** + * Consume the broadcast at the given path, immediately. + * + * The subscription is reset if nobody publishes the path, so use + * {@link announcedBroadcast} instead when the broadcast may not be online yet. + */ consume(path: Path.Valid): broadcast.Consumer; + /** + * A reactive handle to the broadcast at the given path, live only while it is announced. + * + * The announcement-gated counterpart to {@link consume}: it waits for the broadcast to come + * online instead of resetting, and drops back to `undefined` when it goes away. See + * {@link announce.Broadcast}. Close the handle when done. + */ + announcedBroadcast(path: Path.Valid): announce.Broadcast; + /** * Snapshot the transport's counters, querying it fresh on each call. * diff --git a/js/net/src/connection/reload.test.ts b/js/net/src/connection/reload.test.ts index 1cf639fd99..18d7ae7d18 100644 --- a/js/net/src/connection/reload.test.ts +++ b/js/net/src/connection/reload.test.ts @@ -1,6 +1,8 @@ import { expect, test } from "bun:test"; +import { Producer as BroadcastProducer } from "../broadcast.ts"; import * as Lite from "../lite/index.ts"; import { createMockTransportPair } from "../mock.ts"; +import * as Path from "../path.ts"; import { accept } from "./index.ts"; import { Reload, type ReloadProps } from "./reload.ts"; @@ -115,3 +117,60 @@ test("a peer that severs immediately keeps escalating the backoff", async () => globalThis.WebTransport = original; } }); + +// Polls until `pred` holds, so a regression fails the test instead of hanging it. +async function waitUntil(pred: () => boolean): Promise { + for (let i = 0; i < 500; i++) { + if (pred()) return; + await settle(); + } + throw new Error("timed out waiting for condition"); +} + +test("announcedBroadcast follows the reconnect loop", async () => { + const original = globalThis.WebTransport; + const url = new URL("https://example.com/"); + + // Every connect attempt gets a fresh session, whose server publishes the path once the + // handshake finishes. The client therefore always asks before the broadcast exists. + const sessions: { close: () => void }[] = []; + const published: BroadcastProducer[] = []; + const stub = function StubWebTransport() { + const pair = createMockTransportPair(Lite.ALPN_06_WIP); + void accept(pair.server, url).then((server) => { + sessions.push(server); + const broadcast = new BroadcastProducer(); + published.push(broadcast); + server.publish(Path.from("late"), broadcast); + }); + return pair.client; + }; + globalThis.WebTransport = stub as unknown as typeof WebTransport; + + const reload = new Reload({ + enabled: true, + url, + websocket: { enabled: false }, + delay: { initial: 10, multiplier: 1, max: 10 }, + }); + const watched = reload.announcedBroadcast(Path.from("late")); + + try { + await waitUntil(() => watched.active.peek() !== undefined); + const first = watched.active.peek(); + + // The session dies: the handle drops the broadcast rather than clinging to a dead one. + sessions[0]?.close(); + await waitUntil(() => watched.active.peek() === undefined); + + // The reconnect re-announces it, and the handle re-consumes on the new session. + await waitUntil(() => watched.active.peek() !== undefined); + expect(watched.active.peek()).not.toBe(first); + } finally { + watched.close(); + reload.close(); + for (const broadcast of published) broadcast.close(); + for (const session of sessions) session.close(); + globalThis.WebTransport = original; + } +}); diff --git a/js/net/src/connection/reload.ts b/js/net/src/connection/reload.ts index 93a92bb6f3..b2ecc72818 100644 --- a/js/net/src/connection/reload.ts +++ b/js/net/src/connection/reload.ts @@ -293,6 +293,21 @@ export class Reload { return consumer; } + /** + * A reactive handle to one broadcast, spanning reconnects. + * + * The same {@link Announce.Broadcast} as {@link Established.announcedBroadcast}, but it + * follows the reconnect loop: the broadcast drops to `undefined` when the connection dies + * and resolves again once the new connection announces the path. Use it instead of + * consuming off {@link Reload.established} whenever the broadcast may come online after you + * do, which is exactly the case a blind `consume` loses. + * + * Close the handle when done; {@link Reload.close} only drops it to `undefined`. + */ + announcedBroadcast(path: Path.Valid): Announce.Broadcast { + return new Announce.Broadcast(this.established, path); + } + /** * Snapshot the live connection's transport counters, or undefined while disconnected. * See {@link Established.stats}. diff --git a/js/net/src/ietf/connection.ts b/js/net/src/ietf/connection.ts index 7063dc4e62..27868cca5d 100644 --- a/js/net/src/ietf/connection.ts +++ b/js/net/src/ietf/connection.ts @@ -1,5 +1,5 @@ import { type Getter, Signal } from "@moq/signals"; -import type * as announce from "../announced.ts"; +import * as announce from "../announced.ts"; import type * as broadcast from "../broadcast.ts"; import type { Established } from "../connection/established.ts"; import { type Probe, type Stats, transportStats } from "../connection/stats.ts"; @@ -174,6 +174,16 @@ export class Connection implements Established { return this.#subscriber.consume(path); } + /** + * Watches a broadcast, live only while it is announced. + * + * @param path - The path of the broadcast to watch + * @returns A reactive handle to the broadcast + */ + announcedBroadcast(path: Path.Valid): announce.Broadcast { + return new announce.Broadcast(this, path); + } + /** * Accepts bidi streams (virtual for v14-v16, real for v17) and dispatches. */ diff --git a/js/net/src/integration.test.ts b/js/net/src/integration.test.ts index b3fcf04133..98f318754c 100644 --- a/js/net/src/integration.test.ts +++ b/js/net/src/integration.test.ts @@ -1,4 +1,5 @@ import { expect, test } from "bun:test"; +import type { Getter } from "@moq/signals"; import { Producer as BroadcastProducer } from "./broadcast.ts"; import { accept, connect } from "./connection/index.ts"; import { RemoteError } from "./error.ts"; @@ -880,3 +881,96 @@ test("integration: subscribe to non-existent broadcast", async () => { client.close(); server.close(); }); + +// Resolves once `signal` satisfies `pred`, returning the matching value. +async function waitFor(signal: Getter, pred: (value: T) => boolean): Promise { + for (;;) { + const value = signal.peek(); + if (pred(value)) return value; + await signal.changed(); + } +} + +test("integration: announcedBroadcast waits for a late publisher", async () => { + const pair = createMockTransportPair(Lite.ALPN_06_WIP); + const [client, server] = await Promise.all([connect(url, { transport: pair.client }), accept(pair.server, url)]); + + // Serves every requested track with `payload`, until the broadcast closes. + const serve = async (broadcast: BroadcastProducer, payload: string) => { + for (;;) { + const req = await broadcast.requested(); + if (!req) break; + req.accept().writeString(payload); + } + }; + + // Nobody publishes this path yet. A blind consume would be reset (see the + // "subscribe to non-existent broadcast" test); the handle just stays offline. + const watched = client.announcedBroadcast(Path.from("late")); + await sleep(50); + expect(watched.active.peek()).toBeUndefined(); + + // The publisher arrives afterwards. + const first = new BroadcastProducer(); + const servingFirst = serve(first, "hello"); + server.publish(Path.from("late"), first); + + const active = await waitFor(watched.active, (b) => b !== undefined); + if (!active) throw new Error("expected an active broadcast"); + expect(await active.subscribe("video").readString()).toBe("hello"); + + // It goes away. + first.close(); + await servingFirst; + await waitFor(watched.active, (b) => b === undefined); + + // And comes back under the same name: a fresh consumer, not the dead one. + const second = new BroadcastProducer(); + const servingSecond = serve(second, "world"); + server.publish(Path.from("late"), second); + + const republished = await waitFor(watched.active, (b) => b !== undefined); + if (!republished) throw new Error("expected a republished broadcast"); + expect(republished).not.toBe(active); + expect(await republished.subscribe("video").readString()).toBe("world"); + + // Closing the handle releases the broadcast it held. + watched.close(); + expect(watched.active.peek()).toBeUndefined(); + expect(republished.closed.peek()).not.toBeUndefined(); + + second.close(); + await servingSecond; + client.close(); + server.close(); +}); + +test("integration: announcedBroadcast consumes blind without discovery", async () => { + const pair = createMockTransportPair(Lite.ALPN_06_WIP); + const [client, server] = await Promise.all([ + connect(url, { transport: pair.client, discovery: false }), + accept(pair.server, url), + ]); + + const broadcast = new BroadcastProducer(); + const serving = (async () => { + for (;;) { + const req = await broadcast.requested(); + if (!req) break; + req.accept().writeString("blind"); + } + })(); + server.publish(Path.from("test"), broadcast); + + // No announcement ever arrives, so waiting for one would hang. Subscribe anyway. + const watched = client.announcedBroadcast(Path.from("test")); + const active = await waitFor(watched.active, (b) => b !== undefined); + if (!active) throw new Error("expected an active broadcast"); + expect(await active.subscribe("video").readString()).toBe("blind"); + + watched.close(); + broadcast.close(); + await serving; + client.close(); + server.close(); +}); diff --git a/js/net/src/lite/connection.ts b/js/net/src/lite/connection.ts index 45854a0633..a6c1c23eb0 100644 --- a/js/net/src/lite/connection.ts +++ b/js/net/src/lite/connection.ts @@ -1,5 +1,5 @@ import { type Getter, Signal } from "@moq/signals"; -import type * as announce from "../announced.ts"; +import * as announce from "../announced.ts"; import type * as broadcast from "../broadcast.ts"; import type { Established } from "../connection/established.ts"; import { type Probe, type Stats, transportStats } from "../connection/stats.ts"; @@ -177,6 +177,10 @@ export class Connection implements Established { return this.#subscriber.consume(path); } + announcedBroadcast(path: Path.Valid): announce.Broadcast { + return new announce.Broadcast(this, path); + } + async #runSession() { if (!this.#session) { return; diff --git a/js/watch/src/broadcast.ts b/js/watch/src/broadcast.ts index 3596458a39..658d41050b 100644 --- a/js/watch/src/broadcast.ts +++ b/js/watch/src/broadcast.ts @@ -11,13 +11,15 @@ import { toHang } from "./msf"; // announcement check logs at most once per connection. const warnedNoDiscovery = new WeakSet(); -// Whether to skip the announcement gate for this connection: without discovery, waiting on an -// announcement would hang forever, so subscribe immediately and warn once per connection. +// Whether to skip the announcement gate for a cross-broadcast reference: without discovery, +// waiting on an announcement would hang forever, so subscribe immediately and warn once per +// connection. The main broadcast doesn't need this; @moq/net's `announcedBroadcast` falls back +// on its own. function skipDiscovery(conn: Moq.Connection.Established): boolean { if (conn.discovery) return false; if (!warnedNoDiscovery.has(conn)) { warnedNoDiscovery.add(conn); - console.warn("relay does not support broadcast discovery; ignoring reload signal."); + console.warn("relay does not support broadcast discovery; subscribing to siblings blind."); } return true; } @@ -155,10 +157,9 @@ export class Broadcast { return active.has(path); } - // Subscribe to the broadcast, re-consuming on every (re-)announce so a same-name republish (a new - // publisher, or a relay-failover RESTART) re-attaches to the new instance instead of clinging to - // the dead one. Driven off the announcement stream's updates rather than a membership flag, since - // a coalesced republish leaves the active set unchanged yet still emits a fresh update. + // Subscribe to the broadcast, waiting for its announcement so we never race a publisher that + // comes online after us. @moq/net drives the re-consume on a same-name republish and the blind + // fallback on a relay without discovery; mirror its handle into `active`. #runBroadcast(effect: Effect): void { const enabled = effect.get(this.in.enabled); if (!enabled) return; @@ -168,44 +169,19 @@ export class Broadcast { const name = effect.get(this.in.name); - // No announcement gate: subscribe immediately (reload off, or the relay lacks discovery). - if (!effect.get(this.in.reload) || skipDiscovery(conn)) { + // No announcement gate: subscribe immediately. + if (!effect.get(this.in.reload)) { const broadcast = conn.consume(name); effect.cleanup(() => broadcast.close()); effect.set(this.#out.active, broadcast, undefined); return; } - const announced = conn.announced(name); + const announced = conn.announcedBroadcast(name); effect.cleanup(() => announced.close()); - let current: Moq.Broadcast.Consumer | undefined; - effect.cleanup(() => { - current?.close(); - current = undefined; - this.#out.active.set(undefined); - }); - - effect.spawn(async () => { - for (;;) { - const event = await Promise.race([effect.cancel, announced.next()]); - if (!event) break; - - // Scoped to `name`, so the exact broadcast arrives with an empty suffix; ignore children. - if (event.path !== Path.empty()) continue; - - if (event.active) { - // A live subscription survives a redundant (re-)announce; only replace a dead one. - if (current && current.closed.peek() === undefined) continue; - current?.close(); - current = conn.consume(name); - this.#out.active.set(current); - } else { - current?.close(); - current = undefined; - this.#out.active.set(undefined); - } - } + effect.run((nested) => { + nested.set(this.#out.active, nested.get(announced.active), undefined); }); } From 4de3f289aa729270557f9b06bae2531bf939e671 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 3 Aug 2026 19:30:17 -0700 Subject: [PATCH 2/9] fix(net): resolve a lite restart by publisher identity, not blindly ANNOUNCE_UPDATE means two different things. Same first hop is the same content on a new route, where in-flight subscriptions resume and a consumer should observe nothing. A different first hop is a new generation taking the path, where nothing carries over. The JS subscriber flattened both into a bare active:true, so a consumer could either ignore genuine republishes or tear down working subscriptions on every reroute, with no way to tell which. Apply the first-hop rule the draft specifies and rs/moq-net's restart_announce already implements: swallow a reroute, and surface a replacement as an end before the start. Also stop the IETF subscriber from turning a rejected SUBSCRIBE_NAMESPACE into a clean close, which made discovery failure indistinguishable from an unpublished prefix. Co-Authored-By: Claude Opus 5 --- js/net/src/announced.test.ts | 6 +- js/net/src/announced.ts | 10 +++- js/net/src/ietf/subscriber.ts | 6 ++ js/net/src/lite/subscriber.test.ts | 93 ++++++++++++++++++++++++++++++ js/net/src/lite/subscriber.ts | 41 ++++++++++++- 5 files changed, 150 insertions(+), 6 deletions(-) diff --git a/js/net/src/announced.test.ts b/js/net/src/announced.test.ts index a6a02315ca..1cd3dc31fc 100644 --- a/js/net/src/announced.test.ts +++ b/js/net/src/announced.test.ts @@ -19,9 +19,9 @@ test("a same-name re-announce is a distinct update", async () => { const producer = new Announce.Producer(); const consumer = producer.consume(); - // A republish (a lite-06 RESTART, or an unannounce+announce that coalesces) arrives as a - // redundant active:true. The stream carries it as its own update, which is what lets a watcher - // notice the new instance even though membership never observably flipped. + // The stream is a log, not a set: it carries a redundant active:true as its own update rather + // than collapsing it. Deciding what a repeat means belongs to the session layer, which resolves + // a restart into either nothing (a route change) or an end + start (a new publisher). producer.append({ path: p("a"), active: true }); producer.append({ path: p("a"), active: true }); diff --git a/js/net/src/announced.ts b/js/net/src/announced.ts index 0c9e9d3784..bb15e53c5f 100644 --- a/js/net/src/announced.ts +++ b/js/net/src/announced.ts @@ -151,6 +151,11 @@ const warnedNoDiscovery = new WeakSet(); * Falls back to consuming blind (and warns once) on a relay without * {@link Established.discovery}, where there is no announcement to wait for. * + * If discovery fails on a live session (the announcement stream is reset, or the relay + * refuses it) the handle goes offline and stays there: nothing reopens the stream on that + * connection. Build it from a `Connection.Reload` if you need it to recover, since a new + * connection starts a new stream. + * * Close it to release the announcement stream and the current broadcast. * * @public @@ -225,8 +230,9 @@ export class Broadcast { } } } catch (err) { - // The stream was reset, which means the session died under it. - console.debug("announcement stream reset", err); + // Discovery failed: the session died under the stream, or the relay refused + // to answer. Nothing reopens it on this connection, so say so out loud. + console.warn("broadcast discovery failed", err); } // The stream ended, or this run was torn down (its cleanup already ran). Either diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index 524a01fc09..957d53ecb2 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -202,6 +202,12 @@ export class Subscriber { } catch (err: unknown) { const e = error(err); console.warn(`subscribe_namespace error: ${reason(e)}`); + + // Abort the stream rather than letting the caller's `finally` close it cleanly. + // A rejected namespace subscription is a failure, and a consumer that can't tell + // it from "nothing is published under this prefix" waits forever on a broadcast + // that will never be announced. Matches the lite subscriber. + announced.close(e); } } diff --git a/js/net/src/lite/subscriber.test.ts b/js/net/src/lite/subscriber.test.ts index a80c579011..420db6d13b 100644 --- a/js/net/src/lite/subscriber.test.ts +++ b/js/net/src/lite/subscriber.test.ts @@ -1,6 +1,9 @@ import { expect, spyOn, test } from "bun:test"; import { Signal } from "@moq/signals"; import type { Probe } from "../connection/stats.ts"; +import * as Path from "../path.ts"; +import { Writer } from "../stream.ts"; +import { AnnounceOk, encodeAnnounceBroadcast } from "./announce.ts"; import { OriginSchema } from "./origin.ts"; import { Subscriber } from "./subscriber.ts"; import { Version } from "./version.ts"; @@ -30,3 +33,93 @@ test("closing the subscriber suppresses probe stream warnings", async () => { warn.mockRestore(); } }); + +// Drives a Subscriber's announce stream directly: the harness plays the peer, writing +// forged announce messages into the stream the subscriber opens. +function announceHarness(version: Version, origin = 1n) { + let inbound!: ReadableStreamDefaultController; + const quic = { + createBidirectionalStream: async () => ({ + readable: new ReadableStream({ start: (controller) => (inbound = controller) }), + writable: new WritableStream(), + }), + } as unknown as WebTransport; + + const subscriber = new Subscriber(quic, version, OriginSchema.parse(origin)); + + const send = async (f: (w: Writer) => Promise) => { + const written: Uint8Array[] = []; + const writer = new Writer( + new WritableStream({ write: (chunk) => void written.push(new Uint8Array(chunk)) }), + ); + await f(writer); + writer.close(); + await writer.closed; + + const total = written.reduce((sum, c) => sum + c.byteLength, 0); + const out = new Uint8Array(total); + let offset = 0; + for (const chunk of written) { + out.set(chunk, offset); + offset += chunk.byteLength; + } + inbound.enqueue(out); + }; + + return { subscriber, send, settle: () => new Promise((resolve) => setTimeout(resolve, 0)) }; +} + +const PUBLISHER_A = OriginSchema.parse(7n); +const PUBLISHER_B = OriginSchema.parse(8n); +const PEER = OriginSchema.parse(2n); + +test("a restart from the same publisher is a route change, not a republish", async () => { + const { subscriber, send, settle } = announceHarness(Version.DRAFT_06); + const announced = subscriber.announced(Path.empty()); + await settle(); + + await send((w) => new AnnounceOk(PEER, 0).encode(w, Version.DRAFT_06)); + await send((w) => + encodeAnnounceBroadcast( + w, + { status: "active", suffix: Path.from("room"), hops: [PUBLISHER_A] }, + Version.DRAFT_06, + ), + ); + expect(await announced.next()).toEqual({ path: Path.from("room"), active: true }); + + // Same publisher over a new route. In-flight subscriptions resume across it, so the + // subscriber must not surface anything that would make a consumer re-subscribe. + await send((w) => encodeAnnounceBroadcast(w, { status: "restart", id: 0n, hops: [PUBLISHER_A] }, Version.DRAFT_06)); + + // A different publisher took the path: nothing carries over, so this one does surface, + // as an end before the start. Reaching it proves the reroute above emitted nothing. + await send((w) => encodeAnnounceBroadcast(w, { status: "restart", id: 0n, hops: [PUBLISHER_B] }, Version.DRAFT_06)); + expect(await announced.next()).toEqual({ path: Path.from("room"), active: false }); + expect(await announced.next()).toEqual({ path: Path.from("room"), active: true }); + + announced.close(); + subscriber.close(); +}); + +test("a lite-05 duplicate announce follows the same restart rule", async () => { + const { subscriber, send, settle } = announceHarness(Version.DRAFT_05); + const announced = subscriber.announced(Path.empty()); + await settle(); + + await send((w) => new AnnounceOk(PEER, 0).encode(w, Version.DRAFT_05)); + const active = (hops: ReturnType[]) => (w: Writer) => + encodeAnnounceBroadcast(w, { status: "active", suffix: Path.from("room"), hops }, Version.DRAFT_05); + + await send(active([PUBLISHER_A])); + expect(await announced.next()).toEqual({ path: Path.from("room"), active: true }); + + // 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()).toEqual({ path: Path.from("room"), active: false }); + expect(await announced.next()).toEqual({ path: Path.from("room"), active: true }); + + announced.close(); + subscriber.close(); +}); diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index e0d8915df5..fd01e52497 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -206,6 +206,12 @@ export class Subscriber { let nextAnnounceId = 0n; const announcedById = new Map(); + // The publisher behind each path we currently advertise, so a restart can tell a + // route change (same publisher, subscriptions resume) from a replacement (a new + // generation took the path, nothing carries over). At most one advertisement per + // path is current, so the path is the key. + const advertised = new Map(); + // Receive announce updates (for Draft03, this includes initial state) for (;;) { const announce = await Promise.race([ @@ -254,16 +260,49 @@ export class Subscriber { } } + const path = Path.join(prefix, suffix); + // In Lite05+ the sender's origin arrives via AnnounceOk, not in each hop // list, so fold it back in before checking. if (hops !== undefined && dropReflected) { const full = responderOrigin !== undefined ? [...hops, responderOrigin] : hops; if (full.includes(this.origin)) { + // A reflected restart means the peer's remaining route loops back through + // us, so the advertisement is gone even though the message says active. + if (advertised.delete(suffix)) { + console.debug(`announced: broadcast=${path} active=false`); + announced.append({ path: suffix, active: false }); + } continue; } } - const path = Path.join(prefix, suffix); + if (active) { + // The first hop identifies the original publisher; an empty chain means the + // peer itself originated it. See `restart_announce` in the Rust subscriber. + const publisher = hops?.[0] ?? responderOrigin; + const restart = advertised.has(suffix); + const previous = advertised.get(suffix); + advertised.set(suffix, publisher); + + // A second advertisement for a path we already carry is a restart: either an + // explicit ANNOUNCE_UPDATE, or (lite-05) a duplicate ANNOUNCE. + if (restart) { + if (previous === publisher) { + // Same publisher, new route. In-flight subscriptions resume across it, + // so there is nothing for a consumer to react to. + console.debug(`announced: broadcast=${path} rerouted`); + continue; + } + + // A different publisher took the path, so cached track info and existing + // subscriptions must not carry over. Surface a real end before the start. + console.debug(`announced: broadcast=${path} active=false`); + announced.append({ path: suffix, active: false }); + } + } else { + advertised.delete(suffix); + } console.debug(`announced: broadcast=${path} active=${active}`); announced.append({ path: suffix, active }); From c3a373914a4ab5c4fb345fac520a615ae5f8a33a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 3 Aug 2026 19:36:22 -0700 Subject: [PATCH 3/9] refactor(net): insulate Broadcast construction behind the connection factories The exported class published a second, positional creation API alongside the documented `announcedBroadcast(path)` entry point, so a later option would have been a breaking signature change. Make the constructor private and reach it through an internal factory, matching Announce.Consumer. Also drop the example's direct @moq/signals import: an app installing only @moq/net can't resolve it under pnpm's strict layout or Yarn PnP, and @moq/net already re-exports the primitives. Co-Authored-By: Claude Opus 5 --- js/net/examples/wait.ts | 4 +++- js/net/src/announced.ts | 29 +++++++++++++++++++++-------- js/net/src/connection/reload.ts | 2 +- js/net/src/ietf/connection.ts | 2 +- js/net/src/lite/connection.ts | 2 +- 5 files changed, 27 insertions(+), 12 deletions(-) diff --git a/js/net/examples/wait.ts b/js/net/examples/wait.ts index 7bf32c21c5..4b4f8a068b 100644 --- a/js/net/examples/wait.ts +++ b/js/net/examples/wait.ts @@ -1,5 +1,7 @@ import * as Moq from "@moq/net"; -import { Effect } from "@moq/signals"; + +// @moq/net re-exports the reactive primitives, so an app only needs the one dependency. +const { Effect } = Moq.Signals; async function main() { const url = new URL("https://cdn.moq.dev/anon"); diff --git a/js/net/src/announced.ts b/js/net/src/announced.ts index bb15e53c5f..ce5286071e 100644 --- a/js/net/src/announced.ts +++ b/js/net/src/announced.ts @@ -133,6 +133,20 @@ export class Consumer { // once per connection instead of once per watched path. const warnedNoDiscovery = new WeakSet(); +// Constructs a Broadcast without exposing a public constructor. The connection types are the +// documented entry point, and they live in other modules, so this is re-exported as +// {@link watchBroadcast} rather than assigned from within this one. +let makeBroadcast: (connection: GetterInit, path: Path.Valid) => Broadcast; + +/** + * Construct a {@link Broadcast}. Call `announcedBroadcast(path)` on a connection instead. + * + * @internal + */ +export function watchBroadcast(connection: GetterInit, path: Path.Valid): Broadcast { + return makeBroadcast(connection, path); +} + /** * A reactive handle to a single broadcast: {@link Broadcast.active} holds a live * {@link broadcast.Consumer} while the path is announced and `undefined` while nobody @@ -170,14 +184,13 @@ export class Broadcast { #active = new Signal(undefined); #signals = new Effect(); - /** - * Watch `path` on a connection. Accepts a live {@link Established} session or a reactive - * one (a `Connection.Reload`'s `established`), which is how the handle survives reconnects. - * - * Obtain one from `announcedBroadcast(path)` on either connection type rather than - * constructing it directly. - */ - constructor(connection: GetterInit, path: Path.Valid) { + static { + makeBroadcast = (connection, path) => new Broadcast(connection, path); + } + + // Accepts a live Established session or a reactive one (a Reload's `established`), which is + // how the handle survives reconnects. + private constructor(connection: GetterInit, path: Path.Valid) { this.path = path; this.active = this.#active; diff --git a/js/net/src/connection/reload.ts b/js/net/src/connection/reload.ts index b2ecc72818..c42f5f155d 100644 --- a/js/net/src/connection/reload.ts +++ b/js/net/src/connection/reload.ts @@ -305,7 +305,7 @@ export class Reload { * Close the handle when done; {@link Reload.close} only drops it to `undefined`. */ announcedBroadcast(path: Path.Valid): Announce.Broadcast { - return new Announce.Broadcast(this.established, path); + return Announce.watchBroadcast(this.established, path); } /** diff --git a/js/net/src/ietf/connection.ts b/js/net/src/ietf/connection.ts index 27868cca5d..ddfc04b149 100644 --- a/js/net/src/ietf/connection.ts +++ b/js/net/src/ietf/connection.ts @@ -181,7 +181,7 @@ export class Connection implements Established { * @returns A reactive handle to the broadcast */ announcedBroadcast(path: Path.Valid): announce.Broadcast { - return new announce.Broadcast(this, path); + return announce.watchBroadcast(this, path); } /** diff --git a/js/net/src/lite/connection.ts b/js/net/src/lite/connection.ts index a6c1c23eb0..359f44c129 100644 --- a/js/net/src/lite/connection.ts +++ b/js/net/src/lite/connection.ts @@ -178,7 +178,7 @@ export class Connection implements Established { } announcedBroadcast(path: Path.Valid): announce.Broadcast { - return new announce.Broadcast(this, path); + return announce.watchBroadcast(this, path); } async #runSession() { From b7ef2c28c7abcaac05e2d3ed6e482c185b547685 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 3 Aug 2026 19:44:25 -0700 Subject: [PATCH 4/9] fix(net): drop the shared broadcast when its advertisement goes away Consumed broadcasts are reference-counted and shared per path, so closing one handle does not release the cache entry. A second holder (another watcher, or a caller consuming the path directly) therefore kept the departed publisher's entry live, and the next consume() cloned its already-reset tracks instead of subscribing to the replacement. Evict the path from the consume cache wherever a retraction is surfaced, in both the lite and IETF subscribers, so a later announce always subscribes fresh. Existing handles are left to their holders. Also guard the announce handle against clearing a newer run's consumer, clear it when a blind (no-discovery) consume is reset, and document the lite connection method. Co-Authored-By: Claude Opus 5 --- js/net/examples/wait.ts | 13 ++++++--- js/net/src/announced.ts | 11 +++++++- js/net/src/consume.ts | 14 ++++++++++ js/net/src/ietf/subscriber.ts | 4 +++ js/net/src/integration.test.ts | 48 ++++++++++++++++++++++++++++++++++ js/net/src/lite/connection.ts | 6 +++++ js/net/src/lite/subscriber.ts | 25 +++++++++++------- 7 files changed, 107 insertions(+), 14 deletions(-) diff --git a/js/net/examples/wait.ts b/js/net/examples/wait.ts index 4b4f8a068b..fa1eb06c6e 100644 --- a/js/net/examples/wait.ts +++ b/js/net/examples/wait.ts @@ -34,10 +34,15 @@ async function main() { }); }); - // Run until interrupted, then release the handle and the connection. - await connection.closed; - effect.close(); - broadcast.close(); + // Run until interrupted. `closed` rejects if the reconnect loop gives up, so release + // everything on the way out either way. + try { + await connection.closed; + } finally { + effect.close(); + broadcast.close(); + connection.close(); + } } main().catch(console.error); diff --git a/js/net/src/announced.ts b/js/net/src/announced.ts index ce5286071e..90bb28d7f2 100644 --- a/js/net/src/announced.ts +++ b/js/net/src/announced.ts @@ -209,6 +209,12 @@ export class Broadcast { const blind = conn.consume(path); effect.cleanup(() => blind.close()); effect.set(this.#active, blind, undefined); + + // Nothing will announce it back, but `active` still promises a *live* broadcast, + // so stop advertising this one once the wire resets it. + void blind.closed.then(() => { + if (this.#active.peek() === blind) this.#active.set(undefined); + }); return; } @@ -217,9 +223,12 @@ export class Broadcast { let current: broadcast.Consumer | undefined; const offline = () => { + const mine = current; current?.close(); current = undefined; - this.#active.set(undefined); + // Only clear what this run put there. A spawn task that resumes after its run was + // torn down would otherwise wipe the consumer a newer run already installed. + if (this.#active.peek() === mine) this.#active.set(undefined); }; effect.cleanup(offline); diff --git a/js/net/src/consume.ts b/js/net/src/consume.ts index f736c249bc..f14047434e 100644 --- a/js/net/src/consume.ts +++ b/js/net/src/consume.ts @@ -40,4 +40,18 @@ export class BroadcastCache { return consumer; } + + /** + * Stop sharing the broadcast cached for `path`, so the next request subscribes fresh. + * + * Call when the path's advertisement goes away. A handle only leaves the cache on its own + * once *every* holder has closed it, so one holder outliving the publisher (a second + * watcher, or a caller consuming the path directly) would otherwise keep the dead + * generation's cached tracks alive and hand them to whoever consumes the path next. + * Existing handles are left alone: they belong to their holders, and the wire resets + * whatever they still have open. + */ + evict(path: Path.Valid): void { + this.#cache.delete(path); + } } diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index 957d53ecb2..9f5742b20e 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -164,6 +164,7 @@ export class Subscriber { console.debug(`announced: broadcast=${path} active=false`); this.#announced.delete(path); + this.#consumes.evict(path); for (const consumer of this.#announcedConsumers) { const suffix = Path.stripPrefix(consumer.prefix, path); if (suffix === null) continue; @@ -454,6 +455,9 @@ export class Subscriber { console.debug(`runPublishNamespace: stream.reader.closed resolved for ${path}`); } finally { this.#announced.delete(path); + // The path is gone, so stop sharing its broadcast: a holder outliving the publisher + // would otherwise hand the dead generation to whoever consumes the path next. + this.#consumes.evict(path); console.debug(`announced: broadcast=${path} active=false`); for (const consumer of this.#announcedConsumers) { diff --git a/js/net/src/integration.test.ts b/js/net/src/integration.test.ts index 98f318754c..dfae4c1cad 100644 --- a/js/net/src/integration.test.ts +++ b/js/net/src/integration.test.ts @@ -974,3 +974,51 @@ test("integration: announcedBroadcast consumes blind without discovery", async ( client.close(); server.close(); }); + +test("integration: a republish is not served from the previous generation's cache", async () => { + const pair = createMockTransportPair(Lite.ALPN_06_WIP); + const [client, server] = await Promise.all([connect(url, { transport: pair.client }), accept(pair.server, url)]); + + const serve = async (broadcast: BroadcastProducer, payload: string) => { + for (;;) { + const req = await broadcast.requested(); + if (!req) break; + req.accept().writeString(payload); + } + }; + + const first = new BroadcastProducer(); + const servingFirst = serve(first, "old"); + server.publish(Path.from("shared"), first); + + const watched = client.announcedBroadcast(Path.from("shared")); + const active = await waitFor(watched.active, (b) => b !== undefined); + if (!active) throw new Error("expected an active broadcast"); + expect(await active.subscribe("video").readString()).toBe("old"); + + // A second holder of the same path, which is what makes the cache reachable: consumed + // broadcasts are reference-counted, so the handle closing its own copy below does not + // release the shared one. + const bystander = client.consume(Path.from("shared")); + + first.close(); + await servingFirst; + await waitFor(watched.active, (b) => b === undefined); + + // The republish must subscribe fresh. Cloning the cached entry would resolve the previous + // generation's tracks, which the wire has already reset. + const second = new BroadcastProducer(); + const servingSecond = serve(second, "new"); + server.publish(Path.from("shared"), second); + + const republished = await waitFor(watched.active, (b) => b !== undefined); + if (!republished) throw new Error("expected a republished broadcast"); + expect(await republished.subscribe("video").readString()).toBe("new"); + + bystander.close(); + watched.close(); + second.close(); + await servingSecond; + client.close(); + server.close(); +}); diff --git a/js/net/src/lite/connection.ts b/js/net/src/lite/connection.ts index 359f44c129..9286911bd2 100644 --- a/js/net/src/lite/connection.ts +++ b/js/net/src/lite/connection.ts @@ -177,6 +177,12 @@ export class Connection implements Established { return this.#subscriber.consume(path); } + /** + * Watches a broadcast, live only while it is announced. + * + * @param path - The path of the broadcast to watch + * @returns A reactive handle to the broadcast + */ announcedBroadcast(path: Path.Valid): announce.Broadcast { return announce.watchBroadcast(this, path); } diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index fd01e52497..34b37105b6 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -262,6 +262,16 @@ export class Subscriber { const path = Path.join(prefix, suffix); + // Retract the path: forget the advertisement, drop the shared consume entry so a + // later announce subscribes fresh rather than cloning the dead generation's tracks, + // and tell the consumer. + const retract = () => { + advertised.delete(suffix); + this.#consumes.evict(path); + console.debug(`announced: broadcast=${path} active=false`); + announced.append({ path: suffix, active: false }); + }; + // In Lite05+ the sender's origin arrives via AnnounceOk, not in each hop // list, so fold it back in before checking. if (hops !== undefined && dropReflected) { @@ -269,10 +279,7 @@ export class Subscriber { if (full.includes(this.origin)) { // A reflected restart means the peer's remaining route loops back through // us, so the advertisement is gone even though the message says active. - if (advertised.delete(suffix)) { - console.debug(`announced: broadcast=${path} active=false`); - announced.append({ path: suffix, active: false }); - } + if (advertised.has(suffix)) retract(); continue; } } @@ -297,15 +304,15 @@ export class Subscriber { // A different publisher took the path, so cached track info and existing // subscriptions must not carry over. Surface a real end before the start. - console.debug(`announced: broadcast=${path} active=false`); - announced.append({ path: suffix, active: false }); + retract(); } } else { - advertised.delete(suffix); + retract(); + continue; } - console.debug(`announced: broadcast=${path} active=${active}`); - announced.append({ path: suffix, active }); + console.debug(`announced: broadcast=${path} active=true`); + announced.append({ path: suffix, active: true }); } announced.close(); From d8b494d624979bd60a1f1bd824ca3fa220eebb15 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 3 Aug 2026 19:49:21 -0700 Subject: [PATCH 5/9] refactor(net): give Broadcast an options-object constructor The private constructor plus an exported factory was insulation in name only: index.ts re-exports the module wholesale, and no tsconfig sets stripInternal, so `Announce.watchBroadcast` stayed callable and typed. Drop the factory and make the constructor public taking a props object, so there is one creation path and a later option stays additive. Direct construction earns its place: it accepts any `Getter`, which neither connection method covers. Co-Authored-By: Claude Opus 5 --- js/net/src/announced.ts | 35 ++++++++++++++++++--------------- js/net/src/connection/reload.ts | 2 +- js/net/src/ietf/connection.ts | 2 +- js/net/src/lite/connection.ts | 2 +- 4 files changed, 22 insertions(+), 19 deletions(-) diff --git a/js/net/src/announced.ts b/js/net/src/announced.ts index 90bb28d7f2..a3b2070740 100644 --- a/js/net/src/announced.ts +++ b/js/net/src/announced.ts @@ -133,18 +133,20 @@ export class Consumer { // once per connection instead of once per watched path. const warnedNoDiscovery = new WeakSet(); -// Constructs a Broadcast without exposing a public constructor. The connection types are the -// documented entry point, and they live in other modules, so this is re-exported as -// {@link watchBroadcast} rather than assigned from within this one. -let makeBroadcast: (connection: GetterInit, path: Path.Valid) => Broadcast; - /** - * Construct a {@link Broadcast}. Call `announcedBroadcast(path)` on a connection instead. + * What to watch, for {@link Broadcast}. * - * @internal + * @public */ -export function watchBroadcast(connection: GetterInit, path: Path.Valid): Broadcast { - return makeBroadcast(connection, path); +export interface BroadcastProps { + /** + * The connection to watch on. Accepts a live {@link Established} session, or a reactive one + * (a `Connection.Reload`'s `established`), which is how the handle survives reconnects. + */ + connection: GetterInit; + + /** The broadcast path to watch. */ + path: Path.Valid; } /** @@ -184,13 +186,14 @@ export class Broadcast { #active = new Signal(undefined); #signals = new Effect(); - static { - makeBroadcast = (connection, path) => new Broadcast(connection, path); - } - - // Accepts a live Established session or a reactive one (a Reload's `established`), which is - // how the handle survives reconnects. - private constructor(connection: GetterInit, path: Path.Valid) { + /** + * Watch a path on a connection. + * + * Prefer `announcedBroadcast(path)` on the connection itself. Reach for this when the + * session you want to follow isn't either connection type, e.g. your own + * `Getter`. + */ + constructor({ connection, path }: BroadcastProps) { this.path = path; this.active = this.#active; diff --git a/js/net/src/connection/reload.ts b/js/net/src/connection/reload.ts index c42f5f155d..bf2360b29f 100644 --- a/js/net/src/connection/reload.ts +++ b/js/net/src/connection/reload.ts @@ -305,7 +305,7 @@ export class Reload { * Close the handle when done; {@link Reload.close} only drops it to `undefined`. */ announcedBroadcast(path: Path.Valid): Announce.Broadcast { - return Announce.watchBroadcast(this.established, path); + return new Announce.Broadcast({ connection: this.established, path }); } /** diff --git a/js/net/src/ietf/connection.ts b/js/net/src/ietf/connection.ts index ddfc04b149..55db200534 100644 --- a/js/net/src/ietf/connection.ts +++ b/js/net/src/ietf/connection.ts @@ -181,7 +181,7 @@ export class Connection implements Established { * @returns A reactive handle to the broadcast */ announcedBroadcast(path: Path.Valid): announce.Broadcast { - return announce.watchBroadcast(this, path); + return new announce.Broadcast({ connection: this, path }); } /** diff --git a/js/net/src/lite/connection.ts b/js/net/src/lite/connection.ts index 9286911bd2..4611e56175 100644 --- a/js/net/src/lite/connection.ts +++ b/js/net/src/lite/connection.ts @@ -184,7 +184,7 @@ export class Connection implements Established { * @returns A reactive handle to the broadcast */ announcedBroadcast(path: Path.Valid): announce.Broadcast { - return announce.watchBroadcast(this, path); + return new announce.Broadcast({ connection: this, path }); } async #runSession() { From 4f5128f92a9c0c5c9b08d6a47f98b1c260abe992 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 3 Aug 2026 19:53:33 -0700 Subject: [PATCH 6/9] docs(net): note the eviction dedup window Two announcement streams on one path can retract out of order, so a stale retraction may drop the replacement's cache entry and cost a duplicate subscription. Reconciling them needs a generation id on the wire, so record why eviction stays unconditional in the meantime. Co-Authored-By: Claude Opus 5 --- js/net/src/consume.ts | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/js/net/src/consume.ts b/js/net/src/consume.ts index f14047434e..197309ee82 100644 --- a/js/net/src/consume.ts +++ b/js/net/src/consume.ts @@ -50,6 +50,13 @@ export class BroadcastCache { * generation's cached tracks alive and hand them to whoever consumes the path next. * Existing handles are left alone: they belong to their holders, and the wire resets * whatever they still have open. + * + * Eviction is unconditional, which costs a dedup miss when two announcement streams watch + * one path: the second stream's retraction can arrive after the first has already seen the + * replacement, dropping the fresh entry so the next request subscribes again instead of + * sharing. Telling that stale retraction from a live one needs a generation id on the + * advertisement (moq-lite's `Epoch`, not yet on the wire), so until then this errs toward a + * duplicate subscription rather than risk handing out a dead one. */ evict(path: Path.Valid): void { this.#cache.delete(path); From 050f6bc0efbf446686e158cdb88110d850c02de1 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 3 Aug 2026 20:10:34 -0700 Subject: [PATCH 7/9] fix(net): keep the replacement publisher on record after a takeover The retraction helper clears the path's advertised entry, and it ran after the new publisher was recorded, so the entry was left empty. A second takeover then read as a first announcement: no end event, no cache eviction, and a bare active:true that a watcher with a live consumer ignores, stranding it on the previous publisher. Record the publisher after retracting, and cover A -> B -> C. Also correct the handle's docs: a failover that keeps the publisher resumes the subscription rather than re-consuming, so only a publisher replacement produces an offline/online transition. Co-Authored-By: Claude Opus 5 --- js/net/src/announced.ts | 12 +++++++----- js/net/src/lite/subscriber.test.ts | 12 ++++++++++++ js/net/src/lite/subscriber.ts | 12 +++++++----- 3 files changed, 26 insertions(+), 10 deletions(-) diff --git a/js/net/src/announced.ts b/js/net/src/announced.ts index a3b2070740..b4e972ca4e 100644 --- a/js/net/src/announced.ts +++ b/js/net/src/announced.ts @@ -158,11 +158,13 @@ export interface BroadcastProps { * Subscribing to a path nobody publishes gets the stream reset, so a consumer that races the * publisher stays silent forever unless it retries; this waits for the announcement instead. * - * The handle re-consumes on every (re-)announce, so a same-name republish (a new publisher, or - * a relay-failover RESTART) re-attaches to the new instance rather than clinging to the dead - * one. Built from a reconnecting `Connection.Reload`, it also spans reconnects: the - * broadcast drops to `undefined` while disconnected and resolves again once the new connection - * announces it. + * A same-name republish re-consumes, so the handle attaches to the new instance rather than + * clinging to the dead one. A relay failover that keeps the same publisher does *not*: the + * subscription resumes across the new route, so `active` holds the same consumer throughout and + * never goes offline. Only a change of publisher produces an offline/online transition. + * + * Built from a reconnecting `Connection.Reload`, the handle also spans reconnects: the broadcast + * drops to `undefined` while disconnected and resolves again once the new connection announces it. * * Falls back to consuming blind (and warns once) on a relay without * {@link Established.discovery}, where there is no announcement to wait for. diff --git a/js/net/src/lite/subscriber.test.ts b/js/net/src/lite/subscriber.test.ts index 420db6d13b..8d6cd99094 100644 --- a/js/net/src/lite/subscriber.test.ts +++ b/js/net/src/lite/subscriber.test.ts @@ -71,6 +71,7 @@ function announceHarness(version: Version, origin = 1n) { const PUBLISHER_A = OriginSchema.parse(7n); const PUBLISHER_B = OriginSchema.parse(8n); +const PUBLISHER_C = OriginSchema.parse(9n); const PEER = OriginSchema.parse(2n); test("a restart from the same publisher is a route change, not a republish", async () => { @@ -98,6 +99,17 @@ test("a restart from the same publisher is a route change, not a republish", asy expect(await announced.next()).toEqual({ path: Path.from("room"), active: false }); expect(await announced.next()).toEqual({ path: Path.from("room"), active: true }); + // 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()).toEqual({ path: Path.from("room"), active: false }); + expect(await announced.next()).toEqual({ path: Path.from("room"), active: true }); + + // 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()).toEqual({ path: Path.from("room"), active: false }); + announced.close(); subscriber.close(); }); diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index 34b37105b6..24c80c595d 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -288,14 +288,11 @@ export class Subscriber { // The first hop identifies the original publisher; an empty chain means the // peer itself originated it. See `restart_announce` in the Rust subscriber. const publisher = hops?.[0] ?? responderOrigin; - const restart = advertised.has(suffix); - const previous = advertised.get(suffix); - advertised.set(suffix, publisher); // A second advertisement for a path we already carry is a restart: either an // explicit ANNOUNCE_UPDATE, or (lite-05) a duplicate ANNOUNCE. - if (restart) { - if (previous === publisher) { + if (advertised.has(suffix)) { + if (advertised.get(suffix) === publisher) { // Same publisher, new route. In-flight subscriptions resume across it, // so there is nothing for a consumer to react to. console.debug(`announced: broadcast=${path} rerouted`); @@ -306,6 +303,11 @@ export class Subscriber { // subscriptions must not carry over. Surface a real end before the start. retract(); } + + // After `retract()`, which clears the entry: the path is advertised again, by + // whoever just took it over. Recording it before would leave nothing behind, so + // the *next* takeover would read as a first announcement and skip its own end. + advertised.set(suffix, publisher); } else { retract(); continue; From 0c149e42fac396d3d4765bd40cd86f727a6f693c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 4 Aug 2026 10:40:49 -0700 Subject: [PATCH 8/9] fix(net): clear a blind handle on session death, not on the consumer The no-discovery fallback watched the blind consumer's own `closed` to decide when to stop advertising it. That never fires: a consumed broadcast is a path-scoped handle, so a rejected subscribe kills the track and a dead session leaves the handle untouched. The guard was dead code, and `active` outlived its session. Watch the session instead, which mirrors what the announcement-gated path already gets from its stream ending. Also document what `active` means without discovery: assumed present rather than known live, since nothing reports whether the path exists. The handle is scoped to the path, not to a publisher, so a subscribe made after one finally appears succeeds; test both. Co-Authored-By: Claude Opus 5 --- js/net/src/announced.ts | 14 ++++++--- js/net/src/integration.test.ts | 56 ++++++++++++++++++++++++++++++++++ 2 files changed, 66 insertions(+), 4 deletions(-) diff --git a/js/net/src/announced.ts b/js/net/src/announced.ts index b4e972ca4e..88f6685bd0 100644 --- a/js/net/src/announced.ts +++ b/js/net/src/announced.ts @@ -167,7 +167,11 @@ export interface BroadcastProps { * drops to `undefined` while disconnected and resolves again once the new connection announces it. * * Falls back to consuming blind (and warns once) on a relay without - * {@link Established.discovery}, where there is no announcement to wait for. + * {@link Established.discovery}, where there is no announcement to wait for. `active` then + * means *assumed present* rather than known live: nothing reports whether the path exists, so + * a subscribe to a missing broadcast is how a caller finds out. The handle stays usable either + * way, and because it is scoped to the path rather than to one publisher, a subscribe made + * after a publisher finally appears succeeds. * * If discovery fails on a live session (the announcement stream is reset, or the relay * refuses it) the handle goes offline and stays there: nothing reopens the stream on that @@ -215,9 +219,11 @@ export class Broadcast { effect.cleanup(() => blind.close()); effect.set(this.#active, blind, undefined); - // Nothing will announce it back, but `active` still promises a *live* broadcast, - // so stop advertising this one once the wire resets it. - void blind.closed.then(() => { + // The announcement-gated path below goes offline when the stream ends with the + // session; without discovery there is no stream, so watch the session itself. + // A consumed broadcast is a path-scoped handle, not a subscription, so its own + // `closed` says nothing about whether the path exists or the session is alive. + void conn.closed.then(() => { if (this.#active.peek() === blind) this.#active.set(undefined); }); return; diff --git a/js/net/src/integration.test.ts b/js/net/src/integration.test.ts index dfae4c1cad..abfddbfdb1 100644 --- a/js/net/src/integration.test.ts +++ b/js/net/src/integration.test.ts @@ -1022,3 +1022,59 @@ test("integration: a republish is not served from the previous generation's cach client.close(); server.close(); }); + +test("integration: a blind handle picks up a publisher that arrives late", async () => { + const pair = createMockTransportPair(Lite.ALPN_06_WIP); + const [client, server] = await Promise.all([ + connect(url, { transport: pair.client, discovery: false }), + accept(pair.server, url), + ]); + + // Without discovery there is no announcement to wait for, so the handle consumes blind. + const watched = client.announcedBroadcast(Path.from("later")); + const blind = await waitFor(watched.active, (b) => b !== undefined); + if (!blind) throw new Error("expected a blind consumer"); + + // Nobody publishes the path yet, so a subscribe is how the caller finds out. That kills the + // track, not the handle: a consumed broadcast is scoped to the path, not to one publisher. + await expect(blind.subscribe("video").readString()).rejects.toThrow(); + expect(watched.active.peek()).toBe(blind); + + // So a subscribe made after the publisher finally shows up still works, on the same handle. + const producer = new BroadcastProducer(); + const serving = (async () => { + for (;;) { + const req = await producer.requested(); + if (!req) break; + req.accept().writeString("late"); + } + })(); + server.publish(Path.from("later"), producer); + + expect(await blind.subscribe("video").readString()).toBe("late"); + + watched.close(); + producer.close(); + await serving; + client.close(); + server.close(); +}); + +test("integration: a blind handle goes offline when the session dies", async () => { + const pair = createMockTransportPair(Lite.ALPN_06_WIP); + const [client, server] = await Promise.all([ + connect(url, { transport: pair.client, discovery: false }), + accept(pair.server, url), + ]); + + const watched = client.announcedBroadcast(Path.from("whatever")); + await waitFor(watched.active, (b) => b !== undefined); + + // The gated path goes offline when the announcement stream ends with the session. There is + // no stream here, so the session itself is what has to clear it. + server.close(); + client.close(); + await waitFor(watched.active, (b) => b === undefined); + + watched.close(); +}); From 64b137b5719e67f32e8cea2ceff2cfbbeeeb14c5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 4 Aug 2026 11:17:38 -0700 Subject: [PATCH 9/9] test(net): cover the ietf handle, and release the blind session watcher Every announcedBroadcast test used lite, leaving the IETF method with no coverage at all. Add the blind late-publisher case against draft-14. An IETF equivalent of the republish/eviction test is blocked on a pre-existing bug: unpublishing over draft-14 throws "unknown namespace" out of the control-stream adapter and tears the session down instead of retracting. That is in code this PR does not touch, so it stays out of scope and the IETF eviction is uncovered for now. Also race the blind path's session watcher against the run's teardown, so a closed handle isn't retained until the session ends, and fix the `advertised` comment: the map is keyed by suffix, not path. Co-Authored-By: Claude Opus 5 --- js/net/src/announced.ts | 5 ++++- js/net/src/integration.test.ts | 36 ++++++++++++++++++++++++++++++++++ js/net/src/lite/subscriber.ts | 3 ++- 3 files changed, 42 insertions(+), 2 deletions(-) diff --git a/js/net/src/announced.ts b/js/net/src/announced.ts index 88f6685bd0..ae34010938 100644 --- a/js/net/src/announced.ts +++ b/js/net/src/announced.ts @@ -223,7 +223,10 @@ export class Broadcast { // session; without discovery there is no stream, so watch the session itself. // A consumed broadcast is a path-scoped handle, not a subscription, so its own // `closed` says nothing about whether the path exists or the session is alive. - void conn.closed.then(() => { + // Raced against the run's teardown so a closed handle isn't retained until the + // session ends; the cleanup above has already cleared `active` in that case. + effect.spawn(async () => { + await Promise.race([effect.cancel, conn.closed]); if (this.#active.peek() === blind) this.#active.set(undefined); }); return; diff --git a/js/net/src/integration.test.ts b/js/net/src/integration.test.ts index abfddbfdb1..72fabfc4a6 100644 --- a/js/net/src/integration.test.ts +++ b/js/net/src/integration.test.ts @@ -1078,3 +1078,39 @@ test("integration: a blind handle goes offline when the session dies", async () watched.close(); }); + +// The handle and the consume-cache eviction are protocol-agnostic, but their implementations +// are not: each subscriber resolves announcements its own way. These mirror the lite cases. +test("integration: ietf blind handle picks up a publisher that arrives late", async () => { + const pair = createMockTransportPair(""); + const [client, server] = await Promise.all([ + connect(url, { transport: pair.client, discovery: false }), + accept(pair.server, url, { version: Ietf.Version.DRAFT_14 }), + ]); + + const watched = client.announcedBroadcast(Path.from("later")); + const blind = await waitFor(watched.active, (b) => b !== undefined); + if (!blind) throw new Error("expected a blind consumer"); + + // Rejects with 404 rather than resetting the whole handle. + await expect(blind.subscribe("video").readString()).rejects.toThrow(); + expect(watched.active.peek()).toBe(blind); + + const producer = new BroadcastProducer(); + const serving = (async () => { + for (;;) { + const req = await producer.requested(); + if (!req) break; + req.accept().writeString("ietf-late"); + } + })(); + server.publish(Path.from("later"), producer); + + expect(await blind.subscribe("video").readString()).toBe("ietf-late"); + + watched.close(); + producer.close(); + await serving; + client.close(); + server.close(); +}); diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index 24c80c595d..dae9151b0a 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -209,7 +209,8 @@ export class Subscriber { // The publisher behind each path we currently advertise, so a restart can tell a // route change (same publisher, subscriptions resume) from a replacement (a new // generation took the path, nothing carries over). At most one advertisement per - // path is current, so the path is the key. + // path is current, and every announce on this stream shares `prefix`, so the + // suffix is the key. const advertised = new Map(); // Receive announce updates (for Draft03, this includes initial state)