feat(net): announce consumers report when they have synced

doc/concept/moq-lite.md

@@ -37,6 +37,16 @@ how a conference room learns who joined, how a player learns a stream came
 online without polling, and how [relay clusters](/bin/relay/cluster) discover
 each other.
 
+Once that first reply is read, the reader has **synced**: a broadcast not
+announced by then is absent rather than late, so a player can say "offline"
+instead of waiting. A relay replies from what it holds now and never waits on
+its own upstreams. moq-lite 05 and later, and moq-transport with the
+[active-count](/draft/moq-active-count) extension, mark exactly where the first
+reply ends. Older versions sync on the peer's answer, or as soon as the
+request is sent, as a best effort. Reading an origin fed by sessions, a reader
+has synced once every current session's first reply is in, so a reconnect
+takes it back until the new reply lands.
+
 An announcement is a **route**: a claim that broadcasts at a path prefix, and
 every path beneath it, can be served. By convention a publisher announces each
 broadcast's exact path, so subscribers enumerate broadcasts by listing routes.

doc/concept/standard.md

@@ -42,7 +42,7 @@ ignores unknown parameters: [cluster](/draft/moq-cluster) routing hop lists,
 [hidden](/draft/moq-hidden) to keep `.`-named namespaces out of discovery,
 [auth](/draft/moq-auth) to tell each peer what it may publish and subscribe to,
 [active-count](/draft/moq-active-count) to count the `NAMESPACE` messages
-before a `SUBSCRIBE_NAMESPACE` is caught up, and
+before a `SUBSCRIBE_NAMESPACE` is synced, and
 [probe](/draft/moq-probe) for bandwidth estimation.
 Both libraries offer solicit and auth (draft 17 and later) unless a side's
 `extensions` setting declines one, which connects it as a peer that does not

drafts/draft-lcurley-moq-active-count.md

@@ -24,7 +24,7 @@ informative:
 --- abstract
 
 This document defines an extension for MoQ Transport {{moqt}} that tells a subscriber how many NAMESPACE messages answering its SUBSCRIBE_NAMESPACE come before the first change.
-The subscriber then knows when it has caught up, instead of guessing from a quiet stream.
+The subscriber then knows when it has synced, instead of guessing from a quiet stream.
 
 --- note_Note_to_Readers
 
@@ -37,7 +37,7 @@ Submit an [issue](https://github.com/moq-dev/moq/issues) or [PR](https://github.
 {::boilerplate bcp14-tagged}
 
 A suffix is **active** on a SUBSCRIBE_NAMESPACE stream from a NAMESPACE message carrying it until a NAMESPACE_DONE message carrying it.
-A subscriber is **caught up** on a SUBSCRIBE_NAMESPACE once it has read Count ({{count}}) NAMESPACE messages after REQUEST_OK, immediately on a Count of 0.
+A subscriber is **synced** on a SUBSCRIBE_NAMESPACE once it has read Count ({{count}}) NAMESPACE messages after REQUEST_OK, immediately on a Count of 0.
 
 
 # Introduction
@@ -87,7 +87,7 @@ An endpoint MUST close the session with a PROTOCOL_VIOLATION if a REQUEST_OK ans
 # Security Considerations
 
 The count reveals no more than the NAMESPACE messages that follow it.
-A publisher that promises more than it sends can only delay the subscriber's notion of having caught up, which it could equally do by withholding NAMESPACE messages.
+A publisher that promises more than it sends can only delay the subscriber's notion of having synced, which it could equally do by withholding NAMESPACE messages.
 
 
 # IANA Considerations

js/net/src/announced.test.ts

@@ -128,3 +128,131 @@ test("the local route wins over a blind request on a no-discovery origin", async
 	broadcast.close();
 	origin.close();
 });
+
+const tick = () => new Promise((resolve) => setTimeout(resolve, 0));
+const start = (prefix: string): Announce.Event => ({
+	prefix: p(prefix),
+	captures: undefined,
+	kind: "start",
+	route: Route.default,
+});
+
+test("a stream syncs once the reader is past what was queued at the boundary", async () => {
+	const producer = new Announce.Producer();
+	const consumer = producer.consume();
+	producer.append(start("a"));
+	producer.append(start("b"));
+	expect(consumer.synced.peek()).toBe(false);
+
+	producer.synced();
+	// Appended after the boundary, so not owed.
+	producer.append(start("c"));
+	expect(consumer.synced.peek()).toBe(false);
+
+	await consumer.next();
+	await consumer.next();
+	// Still handling b.
+	expect(consumer.synced.peek()).toBe(false);
+	expect(await consumer.next()).toMatchObject({ prefix: p("c") });
+	expect(consumer.synced.peek()).toBe(true);
+});
+
+test("an event taken before the boundary is handled before syncing", async () => {
+	const producer = new Announce.Producer();
+	const consumer = producer.consume();
+	producer.append(start("a"));
+	await consumer.next();
+	producer.synced();
+	expect(consumer.synced.peek()).toBe(false);
+
+	void consumer.next();
+	expect(consumer.synced.peek()).toBe(true);
+	producer.close();
+});
+
+test("behind takes sync back, and closing hands it back for good", () => {
+	const producer = new Announce.Producer();
+	const consumer = producer.consume();
+	producer.synced();
+	expect(consumer.synced.peek()).toBe(true);
+	producer.behind();
+	expect(consumer.synced.peek()).toBe(false);
+	producer.close();
+	expect(consumer.synced.peek()).toBe(true);
+	producer.behind();
+	expect(consumer.synced.peek()).toBe(true);
+});
+
+test("an origin with no session delivering syncs on its own replay", async () => {
+	const origin = new OriginProducer();
+	const room = publish(origin, p("room"));
+	const announced = origin.consume().announced();
+	expect(await announced.next()).toMatchObject({ prefix: p("room"), kind: "start" });
+	void announced.next();
+	await tick();
+	expect(announced.synced.peek()).toBe(true);
+	room.close();
+	announced.close();
+});
+
+test("an origin stream waits on every overlapping initial set in flight", async () => {
+	const origin = new OriginProducer();
+	const wire = wireOf(origin);
+	const room = wire.snapshot(p("room"));
+	const other = wire.snapshot(p("other"));
+
+	const scoped = origin.consume().announced(Path.Pattern.subtree(p("room")));
+	const all = origin.consume().announced();
+	await tick();
+	expect(scoped.synced.peek()).toBe(false);
+	expect(all.synced.peek()).toBe(false);
+
+	other();
+	await tick();
+	// Still held by the room snapshot; the other one never held the scoped stream.
+	expect(scoped.synced.peek()).toBe(false);
+	expect(all.synced.peek()).toBe(false);
+
+	// The initial set lands, and so does its snapshot.
+	const route = wire.receive(p("room/cam"));
+	room();
+	expect(await scoped.next()).toMatchObject({ prefix: p("room/cam"), kind: "start" });
+	void scoped.next();
+	await tick();
+	expect(scoped.synced.peek()).toBe(true);
+
+	// A new session, as a reconnect makes, takes it back until its own set lands.
+	const second = wire.snapshot(p(""));
+	await tick();
+	expect(scoped.synced.peek()).toBe(false);
+	second();
+	await tick();
+	expect(scoped.synced.peek()).toBe(true);
+
+	route.close();
+	scoped.close();
+	all.close();
+});
+
+test("a snapshot elsewhere never holds a scoped stream", async () => {
+	const origin = new OriginProducer();
+	const other = wireOf(origin).snapshot(p("other"));
+	const scoped = origin.consume().announced(Path.Pattern.subtree(p("room")));
+	await tick();
+	expect(scoped.synced.peek()).toBe(true);
+	other();
+	scoped.close();
+});
+
+test("a snapshot under a hidden segment holds only a reader that opts in", async () => {
+	const origin = new OriginProducer();
+	const stats = wireOf(origin).snapshot(p(".stats"));
+	const all = origin.consume().announced();
+	const hidden = origin.consume().announced(Path.Pattern.all(), { hidden: true });
+	await tick();
+	expect(all.synced.peek()).toBe(true);
+	expect(hidden.synced.peek()).toBe(false);
+	stats();
+	all.close();
+	hidden.close();
+});

js/net/src/announced.ts

@@ -3,7 +3,7 @@
  *
  * @module
  */
-import { type GetPromise, Once, Signal } from "@moq/signals";
+import { type GetPromise, type Getter, Once, Signal } from "@moq/signals";
 import type { Route } from "./hop.js";
 import type * as Path from "./path.js";
 
@@ -62,6 +62,18 @@ export interface Options {
 class AnnounceState {
 	queue = new Signal<Event[]>([]);
 	closed = new Once<Error | null>();
+	synced = new Signal(false);
+	// The queued events still to read before the stream counts as synced, or undefined
+	// while the initial set is still arriving.
+	owed: number | undefined;
+	// A reader holds the last event it took before syncing until it asks for the next
+	// one, so whatever it does with that event lands before synced flips.
+	reading = false;
+}
+
+// Synced once nothing is owed and the reader has come back for more.
+function settle(state: AnnounceState) {
+	if (state.owed === 0 && !state.reading) state.synced.set(true);
 }
 
 // Once.set throws on a second settle, and both ends of a stream can close independently.
@@ -71,6 +83,10 @@ function closeState(state: AnnounceState, abort?: Error) {
 	state.queue.mutate((queue) => {
 		queue.length = 0;
 	});
+	// Nothing more is coming, so nothing waiting on the boundary should hang.
+	state.owed = 0;
+	state.reading = false;
+	state.synced.set(true);
 }
 
 // pending: /quest/m1/public-model/answer.md
@@ -103,6 +119,33 @@ export class Producer {
 		});
 	}
 
+	// pending: /quest/m1/public-model/answer.md
+	/**
+	 * Marks the initial set as delivered: the stream syncs once the reader has taken
+	 * everything queued so far. A no-op until {@link behind} once marked.
+	 *
+	 * @internal
+	 */
+	synced() {
+		const state = this.#state;
+		if (state.owed !== undefined) return;
+		state.owed = state.queue.peek().length;
+		settle(state);
+	}
+
+	// pending: /quest/m1/public-model/answer.md
+	/**
+	 * Takes sync back until the next {@link synced}, as when another initial set starts.
+	 *
+	 * @internal
+	 */
+	behind() {
+		const state = this.#state;
+		if (state.closed.peek() !== undefined) return;
+		state.owed = undefined;
+		state.synced.set(false);
+	}
+
 	/** Closes the writer. Idempotent. */
 	close(abort?: Error) {
 		closeState(this.#state, abort);
@@ -137,6 +180,19 @@ export class Consumer {
 		makeConsumer = (state) => new Consumer(state);
 	}
 
+	/**
+	 * Whether the stream has delivered its source's initial announcements and the reader has
+	 * taken them, so a prefix not announced by now is absent rather than not yet heard of.
+	 *
+	 * Flips once the reader asks for the event after the last of them, and is true once
+	 * closed. Through a session, the initial set is the peer's view when it answered: exact
+	 * on versions that count it, best-effort on the rest. Reading an origin, it is every
+	 * attached session's initial set, so a reconnect takes it back until the new one lands.
+	 */
+	get synced(): Getter<boolean> {
+		return this.#state.synced;
+	}
+
 	/** The events as they arrive, until the stream closes. */
 	async *[Symbol.asyncIterator](): AsyncGenerator<Event, void, undefined> {
 		for (;;) {
@@ -148,9 +204,16 @@ export class Consumer {
 
 	/** Returns the next event, or undefined once the stream closes. */
 	async next(): Promise<Event | undefined> {
+		const state = this.#state;
+		state.reading = false;
+		settle(state);
 		for (;;) {
-			const announce = this.#state.queue.peek().shift();
-			if (announce) return announce;
+			const announce = state.queue.peek().shift();
+			if (announce) {
+				if (state.owed) state.owed -= 1;
+				if (!state.synced.peek()) state.reading = true;
+				return announce;
+			}
 
 			const closed = this.#state.closed.peek();
 			if (closed instanceof Error) throw closed;

js/net/src/connection/accept.ts

@@ -118,7 +118,7 @@ async function acceptAlpn(
 	version: Ietf.IetfVersion,
 	wiring: SessionProps,
 ): Promise<Established> {
-	const { control, early, solicit, hidden, cluster, auth } = await exchangeSetup(
+	const { control, early, solicit, hidden, activeCount, cluster, auth } = await exchangeSetup(
 		transport,
 		version,
 		"moq-lite-js",
@@ -134,6 +134,7 @@ async function acceptAlpn(
 		early,
 		solicit,
 		hidden,
+		activeCount,
 		cluster,
 		auth,
 		// v17+ uses NativeSession which manages its own request IDs; maxRequestId is unused.
@@ -171,6 +172,7 @@ async function acceptSetup(
 	params.setBytes(Ietf.SetupOption.Implementation, encoder.encode("moq-lite-js"));
 	Extensions.intoSetup(params, wiring.extensions, version);
 	Ietf.hiddenIntoSetup(params);
+	Ietf.activeCountIntoSetup(params, version);
 
 	const server = new Ietf.ServerSetup({ version, parameters: params });
 	await server.encode(stream.writer, version);
@@ -187,6 +189,7 @@ async function acceptSetup(
 		version,
 		solicit: Ietf.solicitFromSetup(client.parameters),
 		hidden: Ietf.hiddenFromSetup(client.parameters),
+		activeCount: Ietf.activeCountFromSetup(client.parameters, version),
 	});
 }
 
@@ -231,6 +234,7 @@ async function acceptNegotiated(
 	params.setBytes(Ietf.SetupOption.Implementation, encoder.encode("moq-lite-js"));
 	Extensions.intoSetup(params, wiring.extensions, setupVersion);
 	Ietf.hiddenIntoSetup(params);
+	Ietf.activeCountIntoSetup(params, setupVersion);
 
 	const server = new Ietf.ServerSetup({ version: selectedVersion, parameters: params });
 	await server.encode(stream.writer, setupVersion);
@@ -255,6 +259,7 @@ async function acceptNegotiated(
 			version: selectedVersion as Ietf.IetfVersion,
 			solicit: Ietf.solicitFromSetup(client.parameters),
 			hidden: Ietf.hiddenFromSetup(client.parameters),
+			activeCount: Ietf.activeCountFromSetup(client.parameters, selectedVersion as Ietf.IetfVersion),
 		});
 	} else {
 		throw new Error(`unsupported version: ${selectedVersion.toString(16)}`);

js/net/src/connection/connect.ts

@@ -313,6 +313,7 @@ async function negotiate(url: URL, session: WebTransport, wiring: SessionProps):
 	params.setBytes(Ietf.SetupOption.Implementation, encoder.encode("moq-lite-js"));
 	Extensions.intoSetup(params, wiring.extensions, setupVersion);
 	Ietf.hiddenIntoSetup(params);
+	Ietf.activeCountIntoSetup(params, setupVersion);
 
 	const client = new Ietf.ClientSetup({
 		versions:
@@ -352,6 +353,7 @@ async function negotiate(url: URL, session: WebTransport, wiring: SessionProps):
 			version: server.version as Ietf.IetfVersion,
 			solicit: Ietf.solicitFromSetup(server.parameters),
 			hidden: Ietf.hiddenFromSetup(server.parameters),
+			activeCount: Ietf.activeCountFromSetup(server.parameters, server.version as Ietf.IetfVersion),
 		});
 	} else {
 		throw new Error(`unsupported server version: ${server.version.toString()}`);
@@ -368,7 +370,7 @@ async function handshakeAlpn(
 	version: Ietf.IetfVersion,
 	wiring: SessionProps,
 ): Promise<Established> {
-	const { control, early, solicit, hidden, cluster, auth } = await exchangeSetup(
+	const { control, early, solicit, hidden, activeCount, cluster, auth } = await exchangeSetup(
 		session,
 		version,
 		"moq-lite-js",
@@ -384,6 +386,7 @@ async function handshakeAlpn(
 		early,
 		solicit,
 		hidden,
+		activeCount,
 		cluster,
 		auth,
 		// v17+ uses NativeSession which manages its own request IDs; maxRequestId is unused.