fix(auth): present a lost AUTH stream's token again on the same session

js/net/src/auth.test.ts

@@ -134,6 +134,99 @@ describe.each([Lite.ALPN_07_WIP, Ietf.ALPN.DRAFT_17, Ietf.ALPN.DRAFT_22])("%s",
 		second.server.abort();
 	});
 
+	test("a shared handle presents a token whose stream is lost again on the same session", async () => {
+		const clock = new Manual();
+		const installed = clock.install();
+		const warned = spyOn(console, "warn").mockImplementation(() => {});
+		// The server's side of each AUTH stream carrying our token, in arrival order.
+		const Wire = protocol === Lite.ALPN_07_WIP ? LiteAuthWire : IetfAuthWire;
+		const original = Wire.prototype.accept;
+		const streams: Stream[] = [];
+		const accepted = spyOn(Wire.prototype, "accept").mockImplementation(async function (
+			this: InstanceType<typeof Wire>,
+			stream: Stream,
+		) {
+			const token = await original.call(this, stream);
+			if (new TextDecoder().decode(token) === "token") streams.push(stream);
+			return token;
+		});
+		const handle = new Handle();
+		const { client, server } = await connect({ publish: new OriginProducer(), protocol, auth: handle });
+		answer(server.auth.requests(), (token) => (token ? grant(["a"], []) : grant([], [])));
+		try {
+			const token = await handle.add("token");
+			await waitFor(token.grant, (g) => g?.publish.equals(patterns("a")) === true);
+			let ended = false;
+			void token.closed.then(() => {
+				ended = true;
+			});
+
+			streams[0]?.abort(new StreamError(StreamCode.Internal, { message: "lost" }));
+			await waitFor(token.grant, (g) => g === undefined);
+			await clock.advance(Nano.fromMilli(Milli(999)));
+			for (let i = 0; i < 10; i++) await nextTask();
+			expect(streams.length).toBe(1);
+			await clock.advance(Nano.fromMilli(Milli(1)));
+			await waitFor(token.grant, (g) => g?.publish.equals(patterns("a")) === true);
+			expect(streams.length).toBe(2);
+			expect(ended).toBe(false);
+		} finally {
+			accepted.mockRestore();
+			warned.mockRestore();
+			client.abort();
+			server.abort();
+			installed.close();
+		}
+	});
+
+	test("a token whose peer resets only its sending half ends that stream before presenting again", async () => {
+		const clock = new Manual();
+		const installed = clock.install();
+		const warned = spyOn(console, "warn").mockImplementation(() => {});
+		const Wire = protocol === Lite.ALPN_07_WIP ? LiteAuthWire : IetfAuthWire;
+		const original = Wire.prototype.accept;
+		const streams: Stream[] = [];
+		const accepted = spyOn(Wire.prototype, "accept").mockImplementation(async function (
+			this: InstanceType<typeof Wire>,
+			stream: Stream,
+		) {
+			const token = await original.call(this, stream);
+			if (new TextDecoder().decode(token) === "token") streams.push(stream);
+			return token;
+		});
+		const handle = new Handle();
+		const { client, server } = await connect({ publish: new OriginProducer(), protocol, auth: handle });
+		answer(server.auth.requests(), (token) => (token ? grant(["a"], []) : grant([], [])));
+		try {
+			const token = await handle.add("token");
+			await waitFor(token.grant, (g) => g?.publish.equals(patterns("a")) === true);
+			const first = streams[0];
+			if (!first) throw new Error("the token was never presented");
+			let abandoned: unknown;
+			first.reader.closed.then(
+				() => {
+					abandoned = "finished";
+				},
+				(err: unknown) => {
+					abandoned = err;
+				},
+			);
+
+			first.writer.reset(new StreamError(StreamCode.Internal, { message: "lost" }));
+			await waitFor(token.grant, (g) => g === undefined);
+			await clock.advance(Nano.fromMilli(Milli(1000)));
+			await waitFor(token.grant, (g) => g?.publish.equals(patterns("a")) === true);
+			expect(streams.length).toBe(2);
+			expect(abandoned).toBeInstanceOf(Error);
+		} finally {
+			accepted.mockRestore();
+			warned.mockRestore();
+			client.abort();
+			server.abort();
+			installed.close();
+		}
+	});
+
 	test("a handle backs one active session", async () => {
 		const handle = new Handle();
 		const first = await connect({ publish: new OriginProducer(), protocol, auth: handle });
@@ -693,8 +786,9 @@ test("lite-07 presents a transiently refused token again after a backoff", async
 	}
 });
 
-// The peer replied to the first presentation, so it knows AUTH: a reset is a lost stream.
-test("lite-07 keeps a shared token whose retry stream is reset", async () => {
+// The peer replied to the first presentation, so it knows AUTH: a reset is a lost stream,
+// presented again after the doubled backoff.
+test("lite-07 presents a shared token again when its retry stream is reset", async () => {
 	const clock = new Manual();
 	const installed = clock.install();
 	const warned = spyOn(console, "warn").mockImplementation(() => {});
@@ -726,6 +820,10 @@ test("lite-07 keeps a shared token whose retry stream is reset", async () => {
 		await settles(() => logged("auth token ended") > 0);
 		for (let i = 0; i < 10; i++) await nextTask();
 		expect(rejected).toBeUndefined();
+
+		await clock.advance(Nano.fromMilli(Milli(2000)));
+		await settles(() => logged("auth token ended") > 1);
+		expect(rejected).toBeUndefined();
 	} finally {
 		read?.mockRestore();
 		client.abort();

js/net/src/auth_session.ts

@@ -67,6 +67,12 @@ function transient(err: Error | null): boolean {
 	);
 }
 
+// Whether a token's stream ending with `result` is the peer's verdict on it: a refusal that will
+// not pass, Unsupported, or a clean close after a grant. Anything else left the token undecided.
+function isVerdict(refused: boolean, granted: boolean, result: Error | null): boolean {
+	return (refused && !transient(result)) || result instanceof Unsupported || (result === null && granted);
+}
+
 /** A grant as the wire carries it: the expiry is milliseconds from now. @internal */
 export interface WireGrant {
 	publish: Path.Patterns;
@@ -341,7 +347,8 @@ export class Handle {
 	 * Present another token on every session of this handle, resolving once a peer answers it.
 	 *
 	 * With no session live it waits for the next one. A refusal that may pass, such as an
-	 * unreachable auth server, is presented again after a backoff on moq-lite. Rejects with
+	 * unreachable auth server, is presented again after a backoff on moq-lite, and so is a token
+	 * whose stream the peer drops without a verdict. Rejects with
 	 * {@link Unsupported} when a session neither carries AUTH nor lets a request carry a token
 	 * (moq-lite without AUTH), with the peer's refusal as an `Error.Session`, or with the
 	 * handle's end. On moq-transport without AUTH it resolves at once with no grant: only a
@@ -858,7 +865,7 @@ export class AuthSession {
 					if ("grant" in reply) {
 						const expires = reply.grant.expires === undefined ? undefined : wall() + reply.grant.expires;
 						presented.grant = { publish: reply.grant.publish, subscribe: reply.grant.subscribe, expires };
-						answered = granted = true;
+						answered = granted = replied = true;
 						delay = undefined;
 						this.#replied = true;
 						this.#answered(presented);
@@ -879,6 +886,9 @@ export class AuthSession {
 					break;
 				}
 			} catch (err: unknown) {
+				// End both halves: a peer that reset only its own leaves ours open, and a retry that
+				// abandoned it would hold the stream's credit until the session ends.
+				presented.stream?.abort(cancel());
 				closed = transportClosed(err);
 				if (presented.withdrawn || presented.ended) {
 					result = null;
@@ -895,10 +905,21 @@ export class AuthSession {
 			}
 			presented.stream = undefined;
 
-			// A refusal that may pass, such as an unreachable auth server: present the token
-			// again after a backoff while it is still wanted here. Only moq-lite's AUTH_ERROR
-			// carries such a code.
-			if (!refused || !transient(result) || this.#closed || presented.withdrawn) break;
+			// Present the token again after a backoff while it is still wanted on an active
+			// session: after a refusal that may pass, such as an unreachable auth server (only
+			// moq-lite's AUTH_ERROR carries such a code), or, on a shared handle, after its stream
+			// was lost without a verdict, so the session never runs on without it. A handle of the
+			// session's own ends a lost token instead, which tells the application.
+			if (
+				isVerdict(refused, granted, result) ||
+				closed ||
+				!this.active ||
+				presented.withdrawn ||
+				presented.ended ||
+				result instanceof ProtocolViolation ||
+				(!refused && this.#state.own)
+			)
+				break;
 			presented.grant = undefined;
 			granted = false;
 			// Unanswered again, as on its first presentation: a request or publication the token
@@ -934,20 +955,21 @@ export class AuthSession {
 		if (owner && this.#byToken.get(owner) === presented) this.#byToken.delete(owner);
 		this.#ended.update((n) => n + 1);
 		// Only the peer's verdict ends one of the handle's tokens on every session: a refusal,
-		// Unsupported, or a clean close after a grant. Anything else leaves it for the next.
+		// Unsupported, or a clean close after a grant. Anything else leaves it for the next
+		// session, since an active one presents it again instead.
 		// A draining session's verdict is stale while an active session still presents the
 		// token: that session's own answer decides, even mid-retry with no grant right now.
 		const superseded =
 			!this.active &&
 			owner !== undefined &&
 			this.#state.sessions.some((session) => session !== this && session.active && session.presents(owner));
 		// A handle of the session's own has no next session to present it, so any end is final.
-		const verdict =
+		const settled =
 			!this.#closed &&
 			!presented.withdrawn &&
 			!superseded &&
-			(this.#state.own || refused || result instanceof Unsupported || (result === null && granted));
-		if (owner && verdict) this.#state.verdict(owner, result);
+			(this.#state.own || isVerdict(refused, granted, result));
+		if (owner && settled) this.#state.verdict(owner, result);
 		// Closed before the union shrinks, so the scope never reads it as a revocation.
 		if (closed) this.close();
 		this.#recompute();

quest/m1/auth/README.md

@@ -148,7 +148,6 @@ because `main` assigned 0x3A to NOT_FETCHABLE first.
   AUTH streams: a session starts on the token its SETUP or URL carried and its
   AUTH streams add the rest
 - [Binding auth polish](/quest/m1/auth/bindings-polish.md) - OBS reconnects with the presented token, Swift grant sequences
-- [Re-present a lost token](/quest/m1/auth/token-represent.md) - a token whose AUTH stream ended without a verdict is presented again on the live session
 
 ## Related