fix(net): keep a replaced route copy while a group it may carry is needed

js/net/src/delivered.test.ts

@@ -14,6 +14,8 @@ test("add joins neighbouring runs", () => {
 	expect(delivered.runs).toBe(1);
 	expect(delivered.has(4) && delivered.has(8)).toBe(true);
 	expect(delivered.has(3) || delivered.has(9)).toBe(false);
+	expect(delivered.covers(4, 9) && delivered.covers(9, 9)).toBe(true);
+	expect(delivered.covers(3, 5) || delivered.covers(8, 10)).toBe(false);
 });
 
 test("the extremes are sequences too", () => {

js/net/src/delivered.ts

@@ -34,6 +34,13 @@ export class Delivered {
 		return index >= 0 && sequence <= this.#runs[index][1];
 	}
 
+	/** Whether every sequence in `[start, end)` was handed out, or is below the floor. */
+	covers(start: number, end: number): boolean {
+		if (start >= end) return true;
+		const index = this.#below(start);
+		return index >= 0 && end - 1 <= this.#runs[index][1];
+	}
+
 	/** Record `sequence` as handed out; `false` when it already was. */
 	add(sequence: number): boolean {
 		const index = this.#below(sequence);

js/net/src/internal.ts

@@ -278,6 +278,24 @@ export type RecvDatagram =
 	| { kind: "done" }
 	| { kind: "error"; error: Error };
 
+/**
+ * Which groups of a track's live feed to ask whether one may still land: one the feed has not
+ * accounted for, the publisher did not drop, and the feed reaches, in a gap a later arrival
+ * opened less than `grace` ago or that nothing later opened, while a producer is left.
+ */
+export interface Land {
+	/** The first group asked about. */
+	start: number;
+	/** The first group past those asked about. */
+	end: number;
+	/** Whether the feed was asked to start at `start`: otherwise only from its first arrival. */
+	floor: boolean;
+	/** How long a gap a later arrival opened is waited for. */
+	grace: Milli;
+	/** When given, only a group it lacks counts. */
+	delivered?: Delivered;
+}
+
 /**
  * The reads a resumed {@link Subscriber} delegates: it reads its routes' copies rather than a
  * buffer of its own. Implemented by `resume.ts`, the counterpart of the Rust `Inner::Resume`.
@@ -288,10 +306,10 @@ export interface ResumeSource {
 	/** The next group not in `delivered`, in arrival or sequence order, without waiting. */
 	tryGroup(ordered: boolean, delivered: Delivered): Recv;
 	/**
-	 * The next datagram in arrival order, without waiting. `delivered` is the group record once
-	 * groups are read too, so a copy that still holds one for them is kept.
+	 * The next datagram in arrival order, without waiting. `delivered` is the group record, so a
+	 * copy that still holds a group for a later group read is kept.
 	 */
-	tryDatagram(delivered: Delivered | undefined): RecvDatagram;
+	tryDatagram(delivered: Delivered): RecvDatagram;
 	/** Wake once a read may go differently; fires at most once. */
 	changed(fn: () => void): Dispose;
 	/** Forget the gaps in `delivered` below each group past the budget, judged on the newest copy. */
@@ -423,6 +441,8 @@ export const hooks: {
 	cachedGroup: (subscriber: Subscriber, sequence: number) => GroupConsumer | undefined;
 	/** Whether {@link cachedGroup} would find the group, without making a read handle. */
 	holdsGroup: (subscriber: Subscriber, sequence: number) => boolean;
+	/** Whether a group may still land on the live feed of the subscriber's track; see {@link Land}. */
+	mayLand: (subscriber: Subscriber, land: Land) => boolean;
 	// private: capability gap
 	/** A subscriber's largest location now, and whether it can still advance. */
 	peekLargest: (subscriber: Subscriber) => { largest?: Location; ended: boolean };
@@ -575,6 +595,9 @@ export const hooks: {
 	holdsGroup: () => {
 		throw new Error("track.ts not loaded");
 	},
+	mayLand: () => {
+		throw new Error("track.ts not loaded");
+	},
 	peekLargest: () => {
 		throw new Error("track.ts not loaded");
 	},

js/net/src/ledger.ts

@@ -102,6 +102,23 @@ export class Ledger {
 		return this.#runs.some((run) => run.start <= start && end <= run.end);
 	}
 
+	/**
+	 * The gaps within `[start, end)`, each with when a later arrival opened it: none for the one
+	 * past the last run, which nothing has opened yet.
+	 */
+	gaps(start: number, end: number): { start: number; end: number; since?: Instant }[] {
+		const gaps: { start: number; end: number; since?: Instant }[] = [];
+		let below = start;
+		for (const run of this.#runs) {
+			if (below >= end) return gaps;
+			if (run.end <= below) continue;
+			if (below < run.start) gaps.push({ start: below, end: Math.min(run.start, end), since: run.since });
+			below = run.end;
+		}
+		if (below < end) gaps.push({ start: below, end });
+		return gaps;
+	}
+
 	/** Restart the age of every gap reaching into `[start, end)`, which the demand newly asks for. */
 	demand(start: number, end: number, at: Instant): void {
 		if (start >= end) return;

js/net/src/resume.test.ts

@@ -1,5 +1,5 @@
 import { expect, test } from "bun:test";
-import { Milli } from "@moq/clock";
+import { Manual, Milli, Nano } from "@moq/clock";
 import { Producer as BroadcastProducer } from "./broadcast.ts";
 import * as Epoch from "./epoch.ts";
 import { NotFound, StreamCode, StreamError } from "./error.ts";
@@ -1354,3 +1354,259 @@ test("the largest follows the serving copy across a swap without moving back, an
 	a.close();
 	b.close();
 });
+
+/** A manual clock, installed until `restore`. */
+function manual() {
+	const clock = new Manual();
+	return { clock, restore: clock.install() };
+}
+
+const SECOND = Milli(1000);
+
+test.each(["lands", "dropped", "expired"])(
+	"an arrival reader gets a group still in flight on the replaced route (%s)",
+	async (outcome) => {
+		const { clock, restore } = manual();
+		const front = new Front();
+		const a = new BroadcastProducer();
+		const b = new BroadcastProducer();
+		const ta = copy(a);
+		const tb = copy(b);
+		front.serve(a.consume());
+		const sub = front.consumer.track("seq").subscribe({ maxDelay: SECOND });
+		try {
+			ta.write(0);
+			expect(drain(sub)).toEqual([0]);
+			front.serve(b.consume());
+			await settle();
+			tb.write(2);
+			expect(drain(sub)).toEqual([2]);
+			expect(drain(sub)).toEqual([]);
+			expect(ta.producer.demand().wanted.peek()).toBeDefined();
+
+			if (outcome === "dropped") hooks.dropTrackGroups(ta.producer, 1, 2);
+			if (outcome === "expired") await clock.advanceAsync(Nano.fromMilli(SECOND));
+			if (outcome !== "lands") {
+				// The groups drained above are released once their close settles.
+				await settle();
+				expect(drain(sub)).toEqual([]);
+				await settle();
+				expect(ta.producer.demand().wanted.peek()).toBeUndefined();
+			}
+			ta.write(1);
+			expect(drain(sub)).toEqual(outcome === "lands" ? [1] : []);
+		} finally {
+			sub.close();
+			front.consumer.close();
+			a.close();
+			b.close();
+			restore();
+		}
+	},
+);
+
+test("an ordered reader keeps no copy for a group it passed", async () => {
+	const front = new Front();
+	const a = new BroadcastProducer();
+	const b = new BroadcastProducer();
+	const ta = copy(a);
+	const tb = copy(b);
+	front.serve(a.consume());
+	const sub = front.consumer.track("seq").subscribe({ maxDelay: SECOND });
+	const next = () => {
+		const recv = hooks.tryNextGroup(sub);
+		if (recv.kind !== "group") return undefined;
+		recv.group.close();
+		return recv.group.sequence;
+	};
+	ta.write(0);
+	expect(next()).toBe(0);
+	front.serve(b.consume());
+	await settle();
+	tb.write(2);
+	expect(next()).toBe(2);
+	// Group 2 is released once its close settles.
+	await settle();
+	expect(next()).toBeUndefined();
+	await settle();
+	expect(ta.producer.demand().wanted.peek()).toBeUndefined();
+
+	sub.close();
+	front.consumer.close();
+	a.close();
+	b.close();
+});
+
+test.each([true, false])(
+	"a group the replaced copy gets late backs the serving copy's (ordered: %p)",
+	async (ordered) => {
+		const front = new Front();
+		const a = new BroadcastProducer();
+		const b = new BroadcastProducer();
+		const ta = copy(a);
+		const tb = copy(b);
+		front.serve(a.consume());
+		const sub = front.consumer.track("seq").subscribe({ maxDelay: Milli(3_600_000) });
+		const view = ordered ? sub.ordered() : undefined;
+		const next = () => (view ? view.nextGroup() : sub.recvGroup());
+		ta.write(0);
+		(await next())?.close();
+		front.serve(b.consume());
+		await settle();
+
+		const failing = tb.write(1, 1, false);
+		const group = await next();
+		if (group?.sequence !== 1) throw new Error("expected group 1");
+		expect(dec.decode((await group.readFrame())?.payload)).toBe("1.0");
+		// The reader reads on before the old route has group 1 at all.
+		const later = next();
+		await settle();
+
+		// The old route finishes group 1 only now, and the new route's copy of it fails.
+		ta.write(1, 2);
+		failing.close(new StreamError(StreamCode.SessionClosed));
+		expect(dec.decode((await group.readFrame())?.payload)).toBe("1.1");
+		expect(await group.readFrame()).toBeUndefined();
+
+		group.close();
+		tb.write(2);
+		expect((await later)?.sequence).toBe(2);
+		view?.close();
+		sub.close();
+		front.consumer.close();
+		a.close();
+		b.close();
+	},
+);
+
+test.each([true, false])("a replaced copy backs a group before its first arrival (ordered: %p)", async (ordered) => {
+	const front = new Front();
+	const a = new BroadcastProducer();
+	const b = new BroadcastProducer();
+	const c = new BroadcastProducer();
+	const ta = copy(a);
+	const tb = copy(b);
+	const tc = copy(c);
+	front.serve(a.consume());
+	const sub = front.consumer.track("seq").subscribe({ maxDelay: Milli(3_600_000) });
+	const view = ordered ? sub.ordered() : undefined;
+	const next = () => (view ? view.nextGroup() : sub.recvGroup());
+	ta.write(0);
+	(await next())?.close();
+	ta.producer.close();
+	front.serve(b.consume());
+	await settle();
+	front.serve(c.consume());
+	await settle();
+
+	const failing = tc.write(1, 1, false);
+	const group = await next();
+	if (group?.sequence !== 1) throw new Error("expected group 1");
+	expect(dec.decode((await group.readFrame())?.payload)).toBe("1.0");
+	const later = next();
+	await settle();
+
+	// b's first group lands only now, and c's copy of it fails.
+	tb.write(1, 2);
+	failing.close(new StreamError(StreamCode.SessionClosed));
+	expect(dec.decode((await group.readFrame())?.payload)).toBe("1.1");
+	expect(await group.readFrame()).toBeUndefined();
+
+	group.close();
+	tc.write(2);
+	expect((await later)?.sequence).toBe(2);
+	view?.close();
+	sub.close();
+	front.consumer.close();
+	a.close();
+	b.close();
+	c.close();
+});
+
+test("a clean end waits for a group the replaced copy still carries", async () => {
+	const front = new Front();
+	const a = new BroadcastProducer();
+	const b = new BroadcastProducer();
+	const ta = copy(a);
+	const tb = copy(b);
+	front.serve(a.consume());
+	const sub = front.consumer.track("seq").subscribe({ maxDelay: SECOND });
+	try {
+		ta.write(0);
+		ta.write(2);
+		expect(drain(sub)).toEqual([0, 2]);
+
+		// The new route has only the group already handed out, and ends there.
+		tb.write(2);
+		tb.producer.close();
+		front.serve(b.consume());
+		await settle();
+		const next = sub.recvGroup();
+		expect(await isPending(next)).toBe(true);
+
+		ta.write(1);
+		expect((await next)?.sequence).toBe(1);
+		expect(await sub.recvGroup()).toBeUndefined();
+	} finally {
+		sub.close();
+		front.consumer.close();
+		a.close();
+		b.close();
+	}
+});
+
+test.each([false, true])(
+	"a reader that asked for groups keeps a replaced copy's unread group across a datagram read (datagrams first: %p)",
+	async (datagramsFirst) => {
+		const front = new Front();
+		const a = new BroadcastProducer();
+		const b = new BroadcastProducer();
+		const ta = copy(a);
+		copy(b);
+		front.serve(a.consume());
+		const sub = front.consumer.track("seq").subscribe({ maxDelay: Milli(10_000) });
+		expect(sub.tryRecvGroup()).toBeUndefined();
+		expect(hooks.tryRecvDatagram(sub).kind).toBe("idle");
+		ta.write(0);
+		front.serve(b.consume());
+		await settle();
+		if (datagramsFirst) expect(hooks.tryRecvDatagram(sub).kind).toBe("idle");
+		expect(drain(sub)).toEqual([0]);
+
+		sub.close();
+		front.consumer.close();
+		a.close();
+		b.close();
+	},
+);
+
+test.each([Milli(100), Milli(3_600_000)])(
+	"a datagram reader keeps no replaced copy, whatever its budget (%p ms)",
+	async (maxDelay) => {
+		const front = new Front();
+		const sources: BroadcastProducer[] = [];
+		const copies: ReturnType<typeof copy>[] = [];
+		const sub = front.consumer.track("seq").subscribe({ maxDelay });
+		try {
+			for (let sequence = 0; sequence < 20; sequence++) {
+				const source = new BroadcastProducer();
+				const held = copy(source);
+				sources.push(source);
+				copies.push(held);
+				front.serve(source.consume());
+				await settle();
+				held.write(sequence);
+				expect(hooks.tryRecvDatagram(sub).kind).toBe("idle");
+				await settle();
+				const subscribed = copies.filter((held) => held.producer.demand().wanted.peek() !== undefined);
+				expect(subscribed).toEqual([held]);
+			}
+			// Only the serving copy's group is left.
+			expect(drain(sub)).toEqual([19]);
+		} finally {
+			sub.close();
+			front.consumer.close();
+			for (const source of sources) source.close();
+		}
+	},
+);