feat(net)!: one lifecycle vocabulary for finish, abort, closed, and close

bench/browser/page/transport.ts

@@ -45,6 +45,7 @@ async function publish(config: Transport): Promise<void> {
 	});
 	if (connection.transport !== "webtransport") throw new Error(`publisher negotiated ${connection.transport}`);
 	cleanup.push(() => {
+		track.finish();
 		track.close();
 		producer.close();
 		connection.abort();
@@ -58,7 +59,7 @@ async function publish(config: Transport): Promise<void> {
 			const group = track.appendGroup();
 			for (let frame = start; frame < Math.min(start + config.group, config.frames); frame++)
 				group.writeFrame({ payload: fixtures[frame], timestamp: Moq.Time.Timestamp.fromMillis(frame) });
-			group.close();
+			group.finish();
 		}
 	})();
 }

demo/web/src/publish.ts

@@ -408,7 +408,10 @@ meta.run((effect) => {
 		timescale: Net.Time.Timescale.MILLI,
 		maxAge: Milli(86_400_000),
 	});
-	effect.cleanup(() => track.close());
+	effect.cleanup(() => {
+		track.finish();
+		track.close();
+	});
 
 	const producer = new Json.Snapshot.Producer<unknown>({ track });
 	producer.update({ value: currentMeta, at: Net.Time.Timestamp.now() });

doc/lib/js/net.md

@@ -33,13 +33,13 @@ const broadcast = publication.broadcast;
 const track = broadcast.createTrack("messages", { timescale: Moq.Time.Timescale.MILLI });
 const group = track.appendGroup();
 group.writeFrame({ payload: new TextEncoder().encode("hello"), timestamp: Moq.Time.Timestamp.now() });
-group.close();
+group.finish();
 publication.announce();
 ```
 
 - **The origin holds the broadcasts, not the connection.** Closing a session unannounces them and leaves them created for the next one. A publication controls announcements; its `broadcast` fills tracks. The broadcast is invisible, locally and remotely, until the publication calls `announce()`.
 - **Answer claimed paths with `using`.** `dynamic(prefix)` claims a prefix and yields each requested path. Use `using prepared = request.prepare(epoch)`, fill its broadcast, then `prepared.publish()` to answer with a new publisher instance and keep the `Publication` it returns. Leaving the scope without publishing, including on a throw, refuses the request as unroutable. Explicit `prepared.close()` or closing its unpublished broadcast does the same; disposal after publishing leaves the returned publication and its fill broadcast live. Reject before preparing with `request.reject(err)`. A retained, unclosed prepared answer keeps the request pending; JavaScript garbage collection provides no prompt cleanup guarantee.
-- **End content by closing tracks.** A broadcast has no finished state. When the input ends, close each track cleanly; the broadcast stays announced and keeps serving them until you `close()` it. How long to keep it up for late readers is yours to decide.
+- **End content by finishing tracks.** A broadcast has no finished state. When the input ends, `finish()` each track, then `close()` it; closing a track without finishing aborts it. The broadcast stays announced and keeps serving them until you `close()` it. How long to keep it up for late readers is yours to decide.
 - **Requests can pin an epoch.** `consumer.request(path, { epoch })` resolves only through a route announcing that [publisher epoch](/concept/moq-lite#publisher-epochs). The resolved broadcast's `epoch` names the route it came through, pinned or not.
 - **Follow restarts.** `origin.request(path, { announced: true })` follows whichever publisher instance serves the path: its `active` changes to a fresh broadcast on a restart, clears on an end, and resolves again on the next start, so read it reactively and close the request when done. Instances are never stitched, so subscribe again to each new broadcast. Another route of the same epoch taking over (epochs travel only on `moq-lite-07-wip`) keeps `active` as is, and each track subscription resumes there from the first group it lacks. A refusal ends the request. Add `epoch` to pin one instance; a plain `request(path)` keeps its accepted instance, can resolve blindly, and ends once that instance stops serving. To list what is live, iterate `announced(scope)`.
 - **Discovery follows what you watch.** A `consume` origin asks the relay for announcements only under the prefixes its readers watch with `announced(scope)` or `request(path, { announced: true })`, stops asking about a prefix 2s after nothing at or beneath it is watched, and asks again on reconnect. A plain `request(path)` asks nothing and subscribes blind, but once it resolves through a discovered route it watches its path until closed, so it outlives the listing that found it. A relay that refuses an announce request turns that session's announced requests blind too.

doc/lib/js/publish.md

@@ -95,7 +95,10 @@ signals.run((effect) => {
         timescale: Moq.Time.Timescale.MILLI,
         maxAge: Milli(86_400_000),
     });
-    effect.cleanup(() => track.close());
+    effect.cleanup(() => {
+        track.finish();
+        track.close();
+    });
 
     const meta = new Json.Snapshot.Producer<Meta>({ track });
     meta.update({ value: current, at: Moq.Time.Timestamp.now() });

js/e2ee/src/e2ee.test.ts

@@ -246,15 +246,32 @@ test("a broadcast speaks semantic names and puts physical names on the wire", as
 	net.close();
 });
 
+// A broadcast as a session hands it out: no producer, so it ends once no handle holds it.
+class SessionBroadcast extends Broadcast.Consumer {
+	// biome-ignore lint/complexity/noUselessConstructor: widens the protected base constructor to public
+	constructor() {
+		super();
+	}
+}
+
 test("closing a protected consumer releases its broadcast handle", () => {
 	const generation = credential().produce();
-	const net = new Broadcast.Producer();
-	const consumer = new Protected.Consumer(net.consume(), generation.credential.consume(generation.epoch));
+	const net = new SessionBroadcast();
+	const consumer = new Protected.Consumer(net, generation.credential.consume(generation.epoch));
 	expect(consumer.closed.peek()).toBeUndefined();
 	consumer.close();
 	consumer.close();
 	expect(consumer.closed.peek()).toBeNull();
+});
+
+test("closing a protected consumer leaves a produced broadcast to its producer", () => {
+	const generation = credential().produce();
+	const net = new Broadcast.Producer();
+	const consumer = new Protected.Consumer(net.consume(), generation.credential.consume(generation.epoch));
+	consumer.close();
+	expect(net.closed.peek()).toBeUndefined();
 	net.close();
+	expect(consumer.closed.peek()).toBeNull();
 });
 
 test("closing a protected producer ends its broadcast", () => {
@@ -388,7 +405,7 @@ test("grouped authentication failure ends the track", async () => {
 	const bad = new NetGroup.Producer(1);
 	p.net.writeGroup(bad);
 	bad.writeFrame({ timestamp: ms(2), payload: await sealed(p, GROUP, 1, 0, "ok", true) });
-	bad.close();
+	bad.finish();
 
 	expect(await readText(await recvGroup(p.consumer))).toBe("ok");
 	const forged = await recvGroup(p.consumer);
@@ -751,7 +768,7 @@ test("produce waits for an unaccepted track's timescale", async () => {
 	const video = await generation.name("video");
 	const closed = new NetTrack.Producer(video);
 	const refused = generation.produce(closed, "video");
-	closed.close(new Error("gone"));
+	closed.abort(new Error("gone"));
 	await expect(refused).rejects.toThrow("gone");
 	await generation.produce(netTrack(video), "video");
 	expect(await failure(generation.produce(netTrack(video), "video"))).toBe("reuse");

js/e2ee/src/generation.ts

@@ -132,7 +132,7 @@ export class Producer {
 			if (this.#claims.has(semantic)) throw new Failure("reuse");
 		} catch (err) {
 			// Both refusals are Failures.
-			track.close(err as Failure);
+			track.abort(err as Failure);
 			throw err;
 		}
 		this.#claims.add(semantic);

js/e2ee/src/group.ts

@@ -96,14 +96,14 @@ export class Producer {
 		}
 		this.#closed = true;
 		void this.#writer.pump.run(() => {
-			if (!this.#aborted) this.#inner.close();
+			if (!this.#aborted) this.#inner.finish();
 		});
 	}
 
 	#abort(error: Error): void {
 		this.#closed = true;
 		this.#aborted ??= error;
-		this.#inner.close(error);
+		this.#inner.abort(error);
 	}
 }
 
@@ -171,7 +171,8 @@ export class Consumer {
 			return { timestamp: frame.timestamp, plaintext };
 		} catch (err) {
 			if (err instanceof Failure && (err.code === "authentication" || err.code === "identity")) {
-				this.#inner.close(this.#reader.terminal.fail());
+				this.#reader.terminal.fail();
+				this.#inner.close();
 			}
 			throw err;
 		}
@@ -181,7 +182,7 @@ export class Consumer {
 	#check(): void {
 		const failed = this.#reader.terminal.failed;
 		if (!failed) return;
-		this.#inner.close(failed);
+		this.#inner.close();
 		throw failed;
 	}
 

js/e2ee/src/track.ts

@@ -107,11 +107,14 @@ export class Producer {
 		this.#writer.closed = true;
 		if (abort) {
 			this.#writer.aborted = abort;
-			this.#inner.close(abort);
+			this.#inner.abort(abort);
 			return;
 		}
 		void this.#writer.pump.run(() => {
-			if (!this.#writer.aborted) this.#inner.close();
+			if (!this.#writer.aborted) {
+				this.#inner.finish();
+				this.#inner.close();
+			}
 		});
 	}
 

js/flate/bench/decoder.ts

@@ -40,7 +40,8 @@ for (const excess of [64 * 1024, 1024 * 1024, 8 * 1024 * 1024]) {
 	const track = new Track.Producer("overflow-bench").accept({});
 	const group = track.appendGroup();
 	group.writeFrame({ payload: slice });
-	group.close();
+	group.finish();
+	track.finish();
 	track.close();
 	const samples: number[] = [];
 	for (let sample = 0; sample < 4; sample++) {

js/flate/src/snapshot/producer.ts

@@ -61,12 +61,13 @@ export class Producer {
 		} finally {
 			// The group is already visible on the track, so leaving it open on a failed write would
 			// strand a subscriber that advanced into it with nothing to read and no end.
-			group.close();
+			group.finish();
 		}
 	}
 
 	/** Finish the track. */
 	finish(): void {
+		this.#track.finish();
 		this.#track.close();
 	}
 }

js/flate/src/snapshot/snapshot.test.ts

@@ -151,11 +151,12 @@ test("latest lets a newer group preempt an open one", async () => {
 	// A complete newer value arrives while the previous group is still open.
 	const fresh = track.appendGroup();
 	fresh.writeFrame({ payload: bytes(2), timestamp: Time.Timestamp.now() });
-	fresh.close();
+	fresh.finish();
 
 	expect((await consumer.latest())?.value).toEqual(bytes(2));
 
-	stale.close();
+	stale.finish();
+	track.finish();
 	track.close();
 });
 
@@ -181,7 +182,7 @@ test("an aborted track surfaces its error instead of spinning", async () => {
 	const consumer = new Consumer({ track: track.subscribe() });
 
 	const boom = new Error("subscription aborted");
-	track.close(boom);
+	track.abort(boom);
 
 	await expect(consumer.next()).rejects.toThrow("subscription aborted");
 });

js/flate/src/stream/producer.ts

@@ -88,9 +88,9 @@ export class Producer {
 			// end. The group reaches a reader already inside it, which keeps its own handle and would
 			// otherwise get a generic error when ours is dropped.
 			const abort = err instanceof Error ? err : new Error(String(err));
-			this.#group?.close(abort);
+			this.#group?.abort(abort);
 			this.#group = undefined;
-			this.#track.close(abort);
+			this.#track.abort(abort);
 			throw err;
 		}
 
@@ -100,8 +100,9 @@ export class Producer {
 
 	/** Finish the track, closing the group. */
 	finish(): void {
-		this.#group?.close();
+		this.#group?.finish();
 		this.#group = undefined;
+		this.#track.finish();
 		this.#track.close();
 	}
 }

js/flate/src/stream/stream.test.ts

@@ -75,8 +75,9 @@ test("a second group is a rolled log, not a continuation", async () => {
 	for (const pair of [payloads(2), payloads(2)]) {
 		const group = track.appendGroup();
 		for (const payload of pair) group.writeFrame({ payload, timestamp: Time.Timestamp.now() });
-		group.close();
+		group.finish();
 	}
+	track.finish();
 	track.close();
 
 	const consumer = new Consumer({ track: track.subscribe({ maxDelay: REPLAY_LATENCY }) });

js/hang/src/container/consumer.holes.test.ts

@@ -26,13 +26,13 @@ test.each([0, 100])("a sequence hole before a paused data marker expires at %sms
 		const first = new Group.Producer(0);
 		track.writeGroup(first);
 		first.writeFrame({ payload: encodeFrame(new Uint8Array([1]), Micro(10_000)), timestamp: Time.Timestamp.now() });
-		first.close();
+		first.finish();
 		await flush();
 		expect(results).toHaveLength(1);
 		const marker = new Group.Producer(2);
 		track.writeGroup(marker);
 		marker.writeFrame({ payload: new Uint8Array(), timestamp: Time.Timestamp.now() });
-		marker.close();
+		marker.finish();
 		await flush();
 		if (budget > 0) {
 			expect(results).toHaveLength(1);
@@ -46,6 +46,7 @@ test.each([0, 100])("a sequence hole before a paused data marker expires at %sms
 		]);
 	} finally {
 		consumer.close();
+		track.finish();
 		track.close();
 		await reading;
 		installed.close();
@@ -75,7 +76,7 @@ test("a filled hole cancels its timer and the next hole gets a fresh budget", as
 			payload: timestamp === undefined ? new Uint8Array() : encodeFrame(new Uint8Array([1]), Micro(timestamp)),
 			timestamp: Time.Timestamp.fromMicros(timestamp ?? 20_000),
 		});
-		group.close();
+		group.finish();
 	};
 	try {
 		write(0, 0);
@@ -97,6 +98,7 @@ test("a filled hole cancels its timer and the next hole gets a fresh budget", as
 		]);
 	} finally {
 		consumer.close();
+		track.finish();
 		track.close();
 		await reading;
 		installed.close();
@@ -126,7 +128,7 @@ test("the holes a stalled source left behind share one budget", async () => {
 			payload: encodeFrame(new Uint8Array([1]), Micro(sequence * 10_000)),
 			timestamp: Time.Timestamp.fromMicros(sequence * 10_000),
 		});
-		group.close();
+		group.finish();
 	};
 	try {
 		write(0);
@@ -138,6 +140,7 @@ test("the holes a stalled source left behind share one budget", async () => {
 		expect(results.map((read) => read.kind)).toEqual(["frame", "gap", "frame", "gap", "frame", "gap", "frame"]);
 	} finally {
 		consumer.close();
+		track.finish();
 		track.close();
 		await reading;
 		installed.close();
@@ -167,7 +170,7 @@ test("a hole is late from the earliest arrival past it", async () => {
 			payload: encodeFrame(new Uint8Array([1]), Micro(sequence * 10_000)),
 			timestamp: Time.Timestamp.fromMicros(sequence * 10_000),
 		});
-		group.close();
+		group.finish();
 	};
 	try {
 		write(0);
@@ -182,6 +185,7 @@ test("a hole is late from the earliest arrival past it", async () => {
 		expect(results.map((read) => read.kind)).toEqual(["frame", "gap", "frame", "gap", "frame"]);
 	} finally {
 		consumer.close();
+		track.finish();
 		track.close();
 		await reading;
 		installed.close();

js/hang/src/container/consumer.nonsequential.test.ts

@@ -46,7 +46,7 @@ test("Consumer waits on a non-sequential hang group because it has no per-frame
 		payload: encodeLegacyFrame(0 as Micro, new Uint8Array([0x01])),
 		timestamp: Time.Timestamp.now(),
 	});
-	a.close();
+	a.finish();
 	track.writeGroup(a);
 
 	// Drain A's frame.

js/hang/src/container/consumer.outoforder.test.ts

@@ -35,7 +35,7 @@ function publish(track: Track.Producer, sequence: number, millis: number) {
 	const group = new Group.Producer(sequence);
 	track.writeGroup(group);
 	publishFrame(group, millis, sequence);
-	group.close();
+	group.finish();
 }
 
 /** Every frame the consumer will hand over right now, as [group, payload] pairs. */
@@ -111,13 +111,14 @@ test("a below-cursor group still downloading is not truncated", async () => {
 	// rather than shift the group out, so the second frame still arrives.
 	const pending = nextFrame(consumer);
 	publishFrame(backlog, 1500, 11);
-	backlog.close();
+	backlog.finish();
 
 	const second = await pending;
 	expect(second?.group).toBe(1);
 	expect(second?.frame.payload[0]).toBe(11);
 
-	live.close();
+	live.finish();
+	track.finish();
 	track.close();
 	consumer.close();
 });
@@ -150,6 +151,7 @@ test("delayed older groups survive a floor established by two newer groups", asy
 		await expect(nextFrame(consumer)).rejects.toThrow("below the previous group start");
 	} finally {
 		consumer.close();
+		track.finish();
 		track.close();
 		clock.mockRestore();
 	}