fix(net): track and group end-of-life are one outcome value each

js/net/src/group.ts

@@ -770,13 +770,18 @@ class Expired implements GetPromise<Error> {
 	}
 }
 
-// A consumer can finish from its own expiry verdict or from the shared group state. Keep
-// one stable public handle while letting peek() observe either source synchronously, before
-// signal subscribers run in the next microtask. The source is re-read on every check, since a
+// A consumer can finish from its own verdict or from the shared group state. Keep one stable
+// public handle while letting peek() observe either source synchronously, before signal
+// subscribers run in the next microtask. The source is re-read on every check, since a
 // recovered group moves to another copy's state, and `final` says whether its value ends the
-// consumer: a failure recovery can still continue does not.
+// consumer: a failure recovery can still continue does not, so it never reaches a caller. Built
+// on first read, which only a caller holding the group makes: a reader attaches its recovery
+// (`hooks.recoverGroup`) before handing the group out, and the track's own scans read
+// `hooks.groupFailed` instead. A front serving another front is the exception: the inner front
+// holds the handle before the outer attaches its recovery, so attaching rebuilds a handle that
+// latched a failure, and the inner front alone keeps the failure it saw.
 class CombinedClosed implements GetPromise<Error | null> {
-	#preferred: Once<Error | null>;
+	#preferred: Once<Error>;
 	#source: () => Once<Error | null>;
 	#final: (value: Error | null) => boolean;
 	#value = new Once<Error | null>();
@@ -785,11 +790,7 @@ class CombinedClosed implements GetPromise<Error | null> {
 	#unwatch?: Dispose;
 	#dispose: Dispose[] = [];
 
-	constructor(
-		preferred: Once<Error | null>,
-		source: () => Once<Error | null>,
-		final: (value: Error | null) => boolean,
-	) {
+	constructor(preferred: Once<Error>, source: () => Once<Error | null>, final: (value: Error | null) => boolean) {
 		this.#preferred = preferred;
 		this.#source = source;
 		this.#final = final;
@@ -875,9 +876,9 @@ export class Consumer {
 	#attempt?: Attempt;
 	// The frame a replacement copy resumes at: frames below it were read from an earlier copy.
 	#skip = 0;
-	// Sticky failure, set once, including a budget that gave up on the group.
-	#terminal?: Error;
-	#verdict = new Once<Error | null>();
+	// This reader's own failure, set once, including a budget that gave up on the group: it
+	// ends the consumer whatever its copy does. See `#fail`.
+	#verdict = new Once<Error>();
 	#closed?: CombinedClosed;
 	#expired?: Expired;
 	// Closed by its reader, which ends it whatever is left unread.
@@ -935,18 +936,27 @@ export class Consumer {
 		this.#expired ??= new Expired(
 			() => {
 				this.#expire();
-				return this.#terminal instanceof StreamError && this.#terminal.code === StreamCode.Old
-					? this.#terminal
-					: undefined;
+				const verdict = this.#verdict.peek();
+				return verdict instanceof StreamError && verdict.code === StreamCode.Old ? verdict : undefined;
 			},
 			() => this.#expiry?.changed ?? [],
 		);
 		return this.#expired;
 	}
 
+	// The failure that ends this reader, as `closed` would settle with it, without building that
+	// handle: a track's scans read this, so none latches a failure before a reader attaches the
+	// recovery that continues past it.
+	#failed(): Error | undefined {
+		const verdict = this.#verdict.peek();
+		if (verdict) return verdict;
+		const closed = this.#state.closed.peek();
+		return closed instanceof Error && !this.#recovers(closed) ? closed : undefined;
+	}
+
 	// Whether a failure of the copy being read may still be continued from another route.
 	#recovers(failed: Error): boolean {
-		return !this.#terminal && !this.#torn() && this.#recover?.wants(failed) === true;
+		return !this.#verdict.peek() && !this.#torn() && this.#recover?.wants(failed) === true;
 	}
 
 	// Whether a frame handed out to stream was cut short.
@@ -984,10 +994,11 @@ export class Consumer {
 		hooks.expireGroup = (group, expiry) => {
 			group.#expiry = expiry;
 		};
+		hooks.groupFailed = (group) => group.#failed();
 		hooks.recoverGroup = (group, recover) => {
 			group.#recover = recover;
-			// A track's drift scan may have read `closed` already, latching a failure recovery can
-			// now continue: derive it again with the recovery in view.
+			// A front serving another front attaches its recovery after the inner one handed the
+			// group out, whose reads may have latched a failure this recovery continues past.
 			if (group.#closed?.peek() instanceof Error && group.#verdict.peek() === undefined)
 				group.#closed = undefined;
 		};
@@ -1007,9 +1018,10 @@ export class Consumer {
 	// reader operation: its verdict and replacement copy, not a raw producer end, decide it.
 	async #end(until?: number): Promise<{ end: number; complete: boolean }> {
 		for (;;) {
-			if (this.#terminal) {
+			const verdict = this.#verdict.peek();
+			if (verdict) {
 				this.#abandon();
-				throw this.#terminal;
+				throw verdict;
 			}
 			const closed = this.#state.closed.peek();
 			const count = this.frameCount;
@@ -1046,17 +1058,25 @@ export class Consumer {
 	// A reader that took every frame still fails with Old: the producer never finished
 	// the group, so a clean end would pass the prefix off as the whole group.
 	#expire(): boolean {
-		if (this.#terminal) return false;
+		if (this.#verdict.peek()) return false;
 		if (this.#state.closed.peek() instanceof Error) return false;
 		if (!this.#expiry?.expired()) return false;
 
-		this.#terminal = new StreamError(StreamCode.Old, {
-			message: "group exceeded the subscription max delay budget",
-		});
-		if (this.#verdict.peek() === undefined) this.#verdict.set(this.#terminal);
+		this.#fail(
+			new StreamError(StreamCode.Old, {
+				message: "group exceeded the subscription max delay budget",
+			}),
+		);
 		return true;
 	}
 
+	// Fail this reader with `error`, unless it already failed: the first verdict stands.
+	#fail(error: unknown): Error {
+		if (this.#verdict.peek() === undefined)
+			this.#verdict.set(error instanceof Error ? error : new Error(String(error)));
+		return this.#verdict.peek() as Error;
+	}
+
 	#evict(lost: Error): boolean {
 		// A group read on from another copy is that copy's to reclaim.
 		if (this.#state !== this.#mirrored) return false;
@@ -1066,10 +1086,7 @@ export class Consumer {
 			// on this group that is the more specific reason, so ask it first; otherwise
 			// the cache reclaimed it before the reader got to it.
 			this.#expire();
-			if (!this.#terminal) {
-				this.#terminal = lost;
-				if (this.#verdict.peek() === undefined) this.#verdict.set(this.#terminal);
-			}
+			this.#fail(lost);
 		}
 		if (frames.length > 0) {
 			this.#state.start += frames.length;
@@ -1112,8 +1129,7 @@ export class Consumer {
 				(err: unknown) => {
 					started.settled = true;
 					if (abort.signal.aborted) return;
-					this.#terminal ??= err instanceof Error ? err : new Error(String(err));
-					if (this.#verdict.peek() === undefined) this.#verdict.set(this.#terminal);
+					this.#fail(err);
 				},
 			);
 			attempt = started;
@@ -1161,8 +1177,7 @@ export class Consumer {
 		const index = this.#index();
 		if (next.start > index) {
 			replacement.discard();
-			this.#terminal ??= new Error(`group ${this.sequence} cannot resume at frame ${index}`);
-			if (this.#verdict.peek() === undefined) this.#verdict.set(this.#terminal);
+			this.#fail(new Error(`group ${this.sequence} cannot resume at frame ${index}`));
 			return;
 		}
 		replacement.adopt();
@@ -1229,10 +1244,9 @@ export class Consumer {
 				arrived: buffered.arrived,
 			};
 		} catch (error) {
-			this.#terminal = error instanceof Error ? error : new Error(String(error));
-			if (this.#verdict.peek() === undefined) this.#verdict.set(this.#terminal);
+			const verdict = this.#fail(error);
 			this.#closed?.refresh();
-			throw this.#terminal;
+			throw verdict;
 		}
 		if (frames.length === 0 && this.#state.closed.peek() !== undefined) this.#state.flate = undefined;
 		const sequence = this.#state.start++;
@@ -1244,13 +1258,14 @@ export class Consumer {
 	/** True once no further frames can be read: this handle or the group has closed, and every buffered frame is read. */
 	get done(): boolean {
 		return (
-			this.#terminal !== undefined || (this.#bufferedFrames().length === 0 && (this.#released || this.#settled()))
+			this.#verdict.peek() !== undefined ||
+			(this.#bufferedFrames().length === 0 && (this.#released || this.#settled()))
 		);
 	}
 
 	/** True once this handle or the group has been closed, regardless of whether buffered frames remain unread. Synchronous complement to the {@link closed} promise. */
 	get isClosed(): boolean {
-		return this.#terminal !== undefined || this.#released || this.#settled();
+		return this.#verdict.peek() !== undefined || this.#released || this.#settled();
 	}
 
 	// Whether the copy being read ended, for good: a failure recovery may continue is not an end.
@@ -1277,13 +1292,13 @@ export class Consumer {
 	 * end-of-group: check {@link done} to tell "no frame buffered yet" from "finished".
 	 */
 	tryReadFrame(): Frame | undefined {
-		if (this.#terminal) return undefined;
+		if (this.#verdict.peek()) return undefined;
 		return this.#readBufferedFrame()?.frame;
 	}
 
 	/** Like {@link tryReadFrame} but also reports the frame's sequence number within the group. */
 	tryReadFrameSequence(): ({ sequence: number } & Frame) | undefined {
-		if (this.#terminal) return undefined;
+		if (this.#verdict.peek()) return undefined;
 		const read = this.#readBufferedFrame();
 		if (!read) return undefined;
 		return { sequence: read.sequence, ...read.frame };
@@ -1292,7 +1307,7 @@ export class Consumer {
 	/** Resolves once {@link readFrame} would not block. */
 	async readable(): Promise<void> {
 		for (;;) {
-			if (this.#terminal) {
+			if (this.#verdict.peek()) {
 				this.#abandon();
 				return;
 			}
@@ -1343,9 +1358,10 @@ export class Consumer {
 
 	async #readFrame(from: number, compressed = false, streaming = false): Promise<ReadGroupFrame | undefined> {
 		for (;;) {
-			if (this.#terminal) {
+			const verdict = this.#verdict.peek();
+			if (verdict) {
 				this.#abandon();
-				throw this.#terminal;
+				throw verdict;
 			}
 
 			// A whole read streams a frame still being written into its own buffer, rather
@@ -1391,7 +1407,8 @@ export class Consumer {
 		let size = 0;
 		try {
 			for (;;) {
-				if (this.#terminal) throw this.#terminal;
+				const verdict = this.#verdict.peek();
+				if (verdict) throw verdict;
 				const chunk = pollChunk(chunks);
 				if (chunk === undefined) break;
 				if (chunk !== WAIT) {
@@ -1412,10 +1429,9 @@ export class Consumer {
 			const payload = parts.length === 0 ? EMPTY : joined(parts, size);
 			return { sequence, frame: { payload, timestamp: chunks.timestamp, arrived: chunks.arrived } };
 		} catch (error) {
-			this.#terminal ??= error instanceof Error ? error : new Error(String(error));
-			if (this.#verdict.peek() === undefined) this.#verdict.set(this.#terminal);
+			const verdict = this.#fail(error);
 			this.#closed?.refresh();
-			throw this.#terminal;
+			throw verdict;
 		}
 	}
 
@@ -1453,8 +1469,7 @@ export class Consumer {
 	#release(abort: Error | undefined): void {
 		// A copy failure recovery was continuing past ends with the abort instead.
 		if (abort && this.#recover && this.#state.closed.peek() instanceof Error) {
-			this.#terminal ??= abort;
-			if (this.#verdict.peek() === undefined) this.#verdict.set(this.#terminal);
+			this.#fail(abort);
 		}
 		this.#abandon();
 		this.#recover?.close?.();

js/net/src/internal.ts

@@ -476,7 +476,7 @@ export interface StagedTrack {
 
 /**
  * How a received copy's start bounds the fills its readers ask below it. Mirrors the Rust
- * `TrackState::start_resolved` and `fills_from_start`.
+ * `Start::Resolved` and `Producer::fills_from_start`.
  *
  * Below a start the publisher resolved from the budget (`resolvesStart`), a reader fills the
  * history its own budget reaches: an update can cross the response, so nobody knows which
@@ -560,8 +560,9 @@ export const hooks: {
 	 */
 	supersedeStale: (producer: Producer, judged: readonly Run[] | undefined) => Run[];
 	/**
-	 * Mark a track as fed by a wire subscription, which owes missing groups. `starts` says how
-	 * that wire's start bounds the fills its readers ask below it. Assigned by `track.ts`.
+	 * Mark a track as fed by a wire subscription, which owes missing groups, until
+	 * {@link cacheTrack} ends it. `starts` says how that wire's start bounds the fills its readers
+	 * ask below it. Assigned by `track.ts`.
 	 */
 	receiveTrack: (producer: Producer, starts?: FillStarts) => void;
 	/** Create a received group at its wire offset before publishing it to readers. Assigned by `track.ts`. */
@@ -588,7 +589,8 @@ export const hooks: {
 	 * IETF SUBSCRIBE_OK's Largest) for the floor its SUBSCRIBE requested: while the demand keeps
 	 * that floor, a pending tail owes nothing below it. Mirrors the Rust `start_at`; assigned by
 	 * `track.ts`. A null `start` resolves an empty range; without `start`, the SUBSCRIBE still
-	 * waits on its START, as the Rust `request_start`.
+	 * waits on its START, as the Rust `request_start`. Only after {@link receiveTrack}: the start
+	 * is that subscription's, and ends with it.
 	 */
 	startTrack: (producer: Producer, served: { start?: bigint | null; requested?: bigint }) => void;
 	/**
@@ -826,7 +828,12 @@ export const hooks: {
 	arriveGroup: (group: GroupConsumer, at: Instant) => void;
 	/** Keep applying a subscription's drift policy after it hands a group out. */
 	expireGroup: (group: GroupConsumer, expiry: Expiry) => void;
-	/** Carry a handed-out group across its logical track's route changes. */
+	/** The failure that ends a group's reader, as its `closed` would settle, without building that handle. */
+	groupFailed: (group: GroupConsumer) => Error | undefined;
+	/**
+	 * Carry a group across its logical track's route changes. Attached before the group is handed
+	 * out, so its `closed` never settles on a failure the recovery continues past.
+	 */
 	recoverGroup: (group: GroupConsumer, recover: GroupRecover) => void;
 	/** Read a payload chunk, releasing its wait when the output stream closes. */
 	readFrameChunk: (frame: FrameConsumer, closed: Promise<void>) => Promise<Uint8Array | undefined>;
@@ -1162,6 +1169,9 @@ export const hooks: {
 	expireGroup: () => {
 		throw new Error("group.ts not loaded");
 	},
+	groupFailed: () => {
+		throw new Error("group.ts not loaded");
+	},
 	recoverGroup: () => {
 		throw new Error("group.ts not loaded");
 	},

js/net/src/lite/publisher.test.ts

@@ -1704,6 +1704,7 @@ for (const version of [Version.DRAFT_05, Version.DRAFT_06, Version.DRAFT_07]) {
 		test(`lite ${version}: descending delivery resolves source start ${source} at ${expected}`, async () => {
 			const sub = await servedSubscription({ version, startGroup: 2n });
 			try {
+				hooks.receiveTrack(sub.track);
 				hooks.startTrack(sub.track, { start: source, requested: 2n });
 				sub.serve(6n);
 				expect(await sub.servedSequence()).toBe(6n);
@@ -1727,6 +1728,7 @@ for (const version of [Version.DRAFT_06, Version.DRAFT_07]) {
 	test(`lite ${version}: START waits for the upstream START before resolving`, async () => {
 		const sub = await servedSubscription({ version, startGroup: 2n });
 		try {
+			hooks.receiveTrack(sub.track);
 			hooks.startTrack(sub.track, { requested: 2n });
 			sub.serve(6n);
 			await flush();
@@ -1744,6 +1746,7 @@ for (const version of [Version.DRAFT_06, Version.DRAFT_07]) {
 	test(`lite ${version}: a closed source releases the START it was waiting for`, async () => {
 		const sub = await servedSubscription({ version, startGroup: 2n });
 		try {
+			hooks.receiveTrack(sub.track);
 			hooks.startTrack(sub.track, { requested: 2n });
 			sub.serve(6n);
 			await flush();
@@ -1766,6 +1769,7 @@ for (const skipped of ["failed", "stale"] as const) {
 			maxAge: skipped === "stale" ? Milli(0) : undefined,
 		});
 		try {
+			hooks.receiveTrack(sub.track);
 			hooks.startTrack(sub.track, { start: 5n, requested: 5n });
 			const group = sub.track.createGroup(5n);
 			group.writeFrame(textFrame("old", Timestamp.fromMillis(5)));