fix(net,signals): js/net join, refusal, lister, and stale-effect phases; Reload status and Following hold one state

js/net/src/auth_session.ts

@@ -471,9 +471,11 @@ class IssuedGrant implements Issued {
 	#stream: Stream;
 	#wire: AuthWire;
 	#writes = Promise.resolve();
-	/** The latest grant, until its write starts: a newer one replaces it rather than queueing. */
-	#queued: Grant | undefined;
-	#done = false;
+	/**
+	 * Open, holding the latest grant until its write starts, so a newer one replaces it rather
+	 * than queueing; or done: closed, revoked, or a reply that could not be sent.
+	 */
+	#state: { kind: "open"; queued?: Grant } | { kind: "done" } = { kind: "open" };
 
 	constructor(stream: Stream, wire: AuthWire) {
 		this.#stream = stream;
@@ -495,41 +497,51 @@ class IssuedGrant implements Issued {
 			// The peer already closed the stream: nothing left to tell it.
 			if (err instanceof StreamError) return;
 			// A grant the wire cannot carry is withheld, never widened, and so is any other
-			// reply that fails to encode.
+			// reply that fails to encode. An update since then is dropped with it.
 			if (!(err instanceof Unsupported)) console.warn("auth reply not sent", err);
-			this.#done = true;
-			this.#queued = undefined;
+			this.#state = { kind: "done" };
 			return this.#wire.unsupported(this.#stream, error(err).message).catch(() => void 0);
 		});
 	}
 
 	update(grant: Grant): void {
-		if (this.#done) return;
+		const state = this.#state;
+		if (state.kind === "done") return;
 		// A grant still waiting to be written is replaced, so a peer withholding stream credit
 		// holds at most one.
-		const queued = this.#queued !== undefined;
-		this.#queued = grant;
+		const queued = state.queued !== undefined;
+		state.queued = grant;
 		if (queued) return;
 		this.#write(async () => {
-			const grant = this.#queued;
-			this.#queued = undefined;
-			if (grant === undefined) return;
-			const expires = grant.expires === undefined ? undefined : grant.expires - wall();
-			await this.#wire.grant(this.#stream, { publish: grant.publish, subscribe: grant.subscribe, expires });
+			// A close since then sends the grant itself.
+			const state = this.#state;
+			if (state.kind === "done" || state.queued === undefined) return;
+			const grant = state.queued;
+			state.queued = undefined;
+			await this.#send(grant);
 		});
 	}
 
+	async #send(grant: Grant): Promise<void> {
+		const expires = grant.expires === undefined ? undefined : grant.expires - wall();
+		await this.#wire.grant(this.#stream, { publish: grant.publish, subscribe: grant.subscribe, expires });
+	}
+
 	revoke(code: SessionCode, reason: string): void {
-		if (this.#done) return;
+		if (this.#state.kind === "done") return;
 		// The peer never saw a grant still queued, so the refusal replaces it.
-		this.#queued = undefined;
+		this.#state = { kind: "done" };
 		this.#write(() => this.#wire.refuse(this.#stream, code, reason));
-		this.close();
+		this.#writes = this.#writes.then(() => this.#stream.writer.close());
 	}
 
 	close(): void {
-		if (this.#done) return;
-		this.#done = true;
+		const state = this.#state;
+		if (state.kind === "done") return;
+		this.#state = { kind: "done" };
+		// The grant still queued is the last thing the peer hears before the end.
+		const last = state.queued;
+		if (last !== undefined) this.#write(() => this.#send(last));
 		this.#writes = this.#writes.then(() => this.#stream.writer.close());
 	}
 }
@@ -621,6 +633,8 @@ export class AuthSession {
 	#presented = new Set<Presented>();
 	// The handle's tokens as this session presents them, so a request carries only those.
 	#byToken = new Map<Token, Presented>();
+	// How many presented tokens await a reply: at setup, and with bytes. Recounted from
+	// #presented by #recount after every change to it or to an `isAnswered`.
 	#setupPending = new Signal(0);
 	#pending = new Signal(0);
 	#closed = false;
@@ -843,8 +857,6 @@ export class AuthSession {
 	close(err: Error = new Error("session closed")) {
 		if (this.#closed) return;
 		this.#closed = true;
-		this.#pending.set(0);
-		this.#setupPending.set(0);
 		for (const presented of this.#presented) {
 			presented.ended = true;
 			presented.isAnswered = true;
@@ -859,8 +871,7 @@ export class AuthSession {
 
 	#present(wire: AuthWire, presented: Presented) {
 		this.#presented.add(presented);
-		if (presented.setup) this.#setupPending.update((n) => n + 1);
-		if (presented.token.byteLength > 0) this.#pending.update((n) => n + 1);
+		this.#recount();
 		void this.#run(wire, presented);
 	}
 
@@ -967,7 +978,7 @@ export class AuthSession {
 			granted = false;
 			// Unanswered again, as on its first presentation: a request or publication the token
 			// might cover waits for the retry's verdict rather than being judged without it.
-			this.#unsettle(presented);
+			presented.isAnswered = false;
 			this.#recompute();
 			delay = delay === undefined ? RETRY_INITIAL : Milli(Math.min(delay * 2, RETRY_MAX));
 			presented.retry = new AbortController();
@@ -992,7 +1003,7 @@ export class AuthSession {
 		presented.grant = undefined;
 		// A token that ends unanswered counts as answered for enforcement: its refusal is
 		// the reply.
-		this.#settle(presented);
+		presented.isAnswered = true;
 		this.#presented.delete(presented);
 		const owner = presented.owner;
 		if (owner && this.#byToken.get(owner) === presented) this.#byToken.delete(owner);
@@ -1020,27 +1031,25 @@ export class AuthSession {
 
 	/** The peer granted `presented`: it counts as answered, and so does the handle's token. */
 	#answered(presented: Presented) {
-		this.#settle(presented);
-		presented.owner?.answer();
-	}
-
-	/** Stop counting `presented` as unanswered, for enforcement and the setup wait. */
-	#settle(presented: Presented) {
-		if (presented.isAnswered) return;
 		presented.isAnswered = true;
-		if (presented.setup) this.#setupPending.update((n) => n - 1);
-		if (presented.token.byteLength > 0 && !this.#closed) this.#pending.update((n) => n - 1);
+		presented.owner?.answer();
 	}
 
-	/** Count `presented` as unanswered again, as a retry presents it anew. */
-	#unsettle(presented: Presented) {
-		if (!presented.isAnswered) return;
-		presented.isAnswered = false;
-		if (presented.setup) this.#setupPending.update((n) => n + 1);
-		if (presented.token.byteLength > 0) this.#pending.update((n) => n + 1);
+	/** Count the presented tokens still awaiting a reply, for enforcement and the setup wait. */
+	#recount() {
+		let setup = 0;
+		let pending = 0;
+		for (const presented of this.#presented) {
+			if (presented.isAnswered) continue;
+			if (presented.setup) setup += 1;
+			if (presented.token.byteLength > 0) pending += 1;
+		}
+		this.#setupPending.set(setup);
+		this.#pending.set(pending);
 	}
 
 	#recompute() {
+		this.#recount();
 		const granted: Grant[] = [];
 		for (const presented of this.#presented) {
 			if (presented.grant) granted.push(presented.grant);

js/net/src/connection/reload.ts

@@ -329,7 +329,7 @@ export class Reload {
 	// outlives the capped delay, or a new sequence starts.
 	#redirects = new Redirects();
 
-	// Whether a connect attempt is in flight, so a retiring predecessor reports the right status.
+	// Whether a run is dialing: from its start until each connect attempt settles.
 	#dialing = false;
 
 	// The current wait between attempts, doubling per failure, and when the retry window expires.
@@ -524,8 +524,7 @@ export class Reload {
 			this.#givenUpHref = undefined;
 		}
 
-		// A drained predecessor still serves while its replacement dials.
-		if (!this.#draining) this.status.set("connecting");
+		this.#setDialing(true);
 
 		// This run's teardown, handed to connect() so a rerun cancels the attempt in flight.
 		const signal = effect.abort;
@@ -536,10 +535,10 @@ export class Reload {
 		effect.cleanup(() => {
 			if (current) {
 				this.#end(current);
-				if (this.established.peek() === current) this.established.set(undefined);
+				if (this.established.peek() === current) this.#serve(undefined);
 				current = undefined;
 			}
-			if (this.established.peek() === undefined) this.status.set("disconnected");
+			this.#setDialing(false);
 		});
 
 		effect.spawn(async () => {
@@ -556,7 +555,7 @@ export class Reload {
 					// spends it too.
 					this.#deadline ??= this.#window(now());
 
-					this.#dialing = true;
+					this.#setDialing(true);
 					this.#dialedTokens = this.tokens.peek();
 					let connection: Established;
 					try {
@@ -574,8 +573,9 @@ export class Reload {
 							viewers: this.viewers,
 							signal,
 						});
-					} finally {
-						this.#dialing = false;
+					} catch (err) {
+						this.#setDialing(false);
+						throw err;
 					}
 
 					// Hand the connection to the effect. One that lands after this run is over never
@@ -587,8 +587,8 @@ export class Reload {
 					current = connection;
 
 					// The replacement serves now; a predecessor keeps draining its groups in flight.
-					this.established.set(connection);
-					this.status.set("connected");
+					this.#serve(connection);
+					this.#setDialing(false);
 					connected = now();
 
 					// A cancelled effect resolves undefined, so the sentinel tells the session
@@ -600,7 +600,7 @@ export class Reload {
 						console.warn(`connection closed, reconnecting: ${ended ? reason(ended) : "no error"}`);
 						this.#redirects.reset();
 						current = undefined;
-						if (this.established.peek() === connection) this.established.set(undefined);
+						if (this.established.peek() === connection) this.#serve(undefined);
 						this.#retry(effect, connected, ended ?? undefined);
 						return;
 					}
@@ -682,8 +682,7 @@ export class Reload {
 			// would ignore it, so stop here.
 			console.warn("GOAWAY redirect refused:", err);
 			this.#end(connection);
-			this.established.set(undefined);
-			this.status.set("disconnected");
+			this.#serve(undefined);
 			this.#giveUp(error(err));
 			return false;
 		}
@@ -701,9 +700,7 @@ export class Reload {
 		const draining = new Draining(connection, cap, () => {
 			if (this.#draining === draining) this.#draining = undefined;
 			// If nothing replaced it yet, nothing is serving.
-			if (this.established.peek() !== connection) return;
-			this.established.set(undefined);
-			this.status.set(this.#dialing ? "connecting" : "disconnected");
+			if (this.established.peek() === connection) this.#serve(undefined);
 		});
 		this.#draining = draining;
 		return true;
@@ -725,6 +722,23 @@ export class Reload {
 		void ending.then(() => this.#ending.delete(connection));
 	}
 
+	/** Serve `connection`, or nothing. */
+	#serve(connection: Established | undefined): void {
+		this.established.set(connection);
+		this.#report();
+	}
+
+	#setDialing(dialing: boolean): void {
+		this.#dialing = dialing;
+		this.#report();
+	}
+
+	// The one derivation of status: serving, else dialing, else neither. A drained predecessor
+	// serves until it retires, so its replacement's dial reads as connected.
+	#report(): void {
+		this.status.set(this.established.peek() ? "connected" : this.#dialing ? "connecting" : "disconnected");
+	}
+
 	/** Stop serving from a drained predecessor, ending it if still open. */
 	#retire(): void {
 		const draining = this.#draining?.retire();
@@ -749,7 +763,6 @@ export class Reload {
 	 * window: the window restarts once the wait is over.
 	 */
 	#restart(effect: Effect): void {
-		if (this.established.peek() === undefined) this.status.set("disconnected");
 		const wait = this.#backoff();
 		const tick = this.#tick.peek() + 1;
 		effect.timer(() => {
@@ -774,10 +787,6 @@ export class Reload {
 	 * `cause` the error that killed it, if it died with one.
 	 */
 	#retry(effect: Effect, connected: Instant | undefined, cause?: unknown): void {
-		// Report disconnected during the backoff rather than when the retry reruns the
-		// effect, unless a drained predecessor still serves until it retires.
-		if (this.established.peek() === undefined) this.status.set("disconnected");
-
 		// A healthy session's end clears the backoff and starts a fresh retry window: a one-off
 		// drop should reconnect promptly. Anything shorter is a peer that accepts and soon
 		// severs, which has to keep escalating or we hammer it forever at the initial delay.

js/net/src/ietf/join.test.ts

@@ -564,7 +564,7 @@ test("a clean end finishes a head nothing claimed", () => {
 	expect(j.open(() => producer)).toBe(producer);
 	producer.writeFrame({ payload: text("c") });
 	j.wrote();
-	j.finish();
+	j.finish(producer);
 	expect(j.outstanding).toBe(true);
 	j.end();
 	expect(producer.closed.peek()).toBeNull();
@@ -671,3 +671,29 @@ test("an evicted joined group is fetched again as standalone history", async ()
 	group.close();
 	consumer.close();
 });
+
+test("a second fetch stream after the track ended leaves the first head to its own stream", () => {
+	const j = new Join(0n, new Tail());
+	j.end();
+	const first = new track.Producer("catalog").createGroup(0n);
+	expect(j.open(() => first)).toBe(first);
+	const second = new track.Producer("catalog").createGroup(0n);
+	expect(j.open(() => second)).toBeUndefined();
+	// The second stream, discarding, ends first: the head is still the first stream's.
+	j.finish(undefined);
+	expect(first.closed.peek()).toBeUndefined();
+	first.writeFrame({ payload: text("c") });
+	j.finish(first);
+	expect(first.closed.peek()).toBeNull();
+});
+
+test("a discarding second fetch stream leaves a head being written alone", () => {
+	const j = new Join(0n, new Tail());
+	const first = new track.Producer("catalog").createGroup(0n);
+	expect(j.open(() => first)).toBe(first);
+	expect(j.open(() => new track.Producer("catalog").createGroup(0n))).toBeUndefined();
+	j.fail(new Error("reset"), undefined);
+	expect(first.closed.peek()).toBeUndefined();
+	expect(j.pending).toBe(false);
+	expect(j.outstanding).toBe(true);
+});