feat(net): invite a retry on a REQUEST_ERROR that may pass

js/net/src/auth_session.ts

@@ -50,9 +50,14 @@ export class Duplicate extends Error {
 	}
 }
 
-// How long a token refused for a reason that may pass waits before its session presents it
-// again, doubling on each refusal up to RETRY_MAX.
-const RETRY_INITIAL = Milli(1000);
+/**
+ * How long a token refused for a reason that may pass waits before its session presents it
+ * again, doubling on each refusal up to RETRY_MAX. A moq-transport refusal that may pass asks
+ * its peer to wait as long.
+ *
+ * @internal
+ */
+export const RETRY_INITIAL = Milli(1000);
 const RETRY_MAX = Milli(30_000);
 
 // Whether a refusal is the acceptor failing to decide rather than its verdict on the token,

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

@@ -420,6 +420,49 @@ test("TRACK_STATUS for a missing broadcast or track is refused DOES_NOT_EXIST",
 	}
 });
 
+// A refused track invites a retry when asking again may pass, as in Rust, and asks the peer never
+// to retry a verdict.
+test("a TRACK_STATUS refusal that may pass invites a retry", async () => {
+	for (const version of IETF_VERSIONS) {
+		for (const [err, errorCode, retryInterval] of [
+			[new StreamError(StreamCode.GoingAway, { message: "draining" }), 0x6, 1000n],
+			[new SessionError(SessionCode.Internal), 0x0, 1000n],
+			[new StreamError(StreamCode.NotFound, { message: "no video" }), 0x10, 0n],
+			[new StreamError(StreamCode.Unauthorized), 0x1, 0n],
+			[new StreamError(StreamCode.DeliveryTimeout), 0x2, 0n],
+			[new StreamError(StreamCode.MalformedTrack), 0x12, 0n],
+			[new SessionError(SessionCode.Unauthorized), 0x1, 0n],
+			[new SessionError(SessionCode.Expired), 0x5, 0n],
+			[new SessionError(SessionCode.Timeout), 0x2, 0n],
+			[new SessionError(SessionCode.Version), 0x3, 0n],
+			[new SessionError(SessionCode.Unsupported), 0x3, 0n],
+		] as const) {
+			const pair = createMockTransportPair(ALPNS[version]);
+			const session = new NativeSession(pair.server, version, true);
+			const { pub, origin } = publisher(pair.server, { session });
+			const requests = origin.routePrefix(Path.from("room")).requested();
+
+			const status = requestStream(version);
+			const answering = pub.runTrackStatusRequest(trackStatus("room"), status.stream);
+			const request = (await requests.next()).value;
+			if (!request) throw new Error("no request for room");
+			const served = new BroadcastProducer();
+			const trackRequest = wireOf(served).requested();
+			request.serve(served);
+			(await trackRequest)?.reject(err);
+			await answering;
+
+			const wire = status.wire();
+			expect(await wire.u53()).toBe(RequestError.id);
+			const refused = await RequestError.decode(wire, version);
+			expect([refused.errorCode, refused.retryInterval]).toEqual([errorCode, retryInterval]);
+			await requests.return?.();
+			served.close();
+			origin.close();
+		}
+	}
+});
+
 // The requester leaving is owed nothing, and gives back the origin demand or the track
 // lookup it held, whichever was still resolving.
 test("a cancelled TRACK_STATUS releases its demand on every draft", async () => {

js/net/src/ietf/publisher.ts

@@ -27,6 +27,7 @@ import type { Consumer as OriginConsumer } from "../origin.ts";
 import * as Path from "../path.ts";
 import { sendPayload } from "../payload.ts";
 import { Priority, sendOrder } from "../priority.ts";
+import { routeFailed } from "../resume.ts";
 import { type Stream, Writer } from "../stream.ts";
 import { Timescale } from "../time.ts";
 import {
@@ -48,7 +49,7 @@ import * as Varint from "../varint.ts";
 import { type Advertised, type Advertisements, sameInstance, wireOf } from "../wire.ts";
 import * as Cluster from "./cluster.ts";
 import { encodeDatagram } from "./datagram.ts";
-import { requestReason } from "./error.ts";
+import { type RequestCondition, requestReason } from "./error.ts";
 import { type Fetch, FetchHeader, FetchOk, type FetchType } from "./fetch.ts";
 import * as Filter from "./filter.ts";
 import { MAX_AGE_PARAMETER } from "./max_age.ts";
@@ -311,6 +312,39 @@ interface RunFill {
 	streams: StreamCount;
 }
 
+// The condition a request that failed for `err` is refused with, mirroring Rust's `to_code`: a
+// verdict stays one, so the peer is never invited to retry it.
+function refusedAs(err: Error): RequestCondition {
+	if (err instanceof SessionError) {
+		switch (err.code) {
+			case SessionCode.Unauthorized:
+				return "unauthorized";
+			case SessionCode.Expired:
+				return "expired";
+			case SessionCode.Timeout:
+				return "timeout";
+			case SessionCode.Version:
+			case SessionCode.Unsupported:
+				return "not_supported";
+		}
+	} else if (err instanceof StreamError) {
+		switch (err.code) {
+			case StreamCode.Unauthorized:
+				return "unauthorized";
+			case StreamCode.DeliveryTimeout:
+				return "timeout";
+			case StreamCode.MalformedTrack:
+				return "malformed_track";
+			case StreamCode.GoingAway:
+				return "going_away";
+			case StreamCode.NotFound:
+			case StreamCode.Unroutable:
+				return "does_not_exist";
+		}
+	}
+	return "internal";
+}
+
 /**
  * Handles publishing broadcasts using moq-transport protocol.
  * Uses the stream-per-request pattern (real bidi streams for v17, virtual for v14-v16).
@@ -2708,16 +2742,18 @@ export class Publisher {
 		}
 	}
 
-	/** What a request that failed to resolve is refused with. */
+	/**
+	 * What a request that failed to resolve is refused with. A missing publisher a rendezvous
+	 * waited for is a timeout, which, like Rust's `rendezvous_error`, invites no retry.
+	 */
 	#failed(err: unknown, wait = false, resolved = false): Refusal {
 		const e = error(err);
-		const condition =
-			e instanceof StreamError && (e.code === StreamCode.NotFound || e.code === StreamCode.Unroutable)
-				? wait && (!resolved || e.code === StreamCode.Unroutable)
-					? "timeout"
-					: "does_not_exist"
-				: "internal";
-		return { condition, reasonPhrase: reason(e) };
+		const reasonPhrase = reason(e);
+		const missing =
+			e instanceof StreamError &&
+			(e.code === StreamCode.Unroutable || (e.code === StreamCode.NotFound && !resolved));
+		if (wait && missing) return { condition: "timeout", reasonPhrase };
+		return { condition: refusedAs(e), reasonPhrase, retryable: routeFailed(e) };
 	}
 
 	/**

js/net/src/ietf/request_stream.ts

@@ -1,3 +1,4 @@
+import { RETRY_INITIAL } from "../auth_session.ts";
 import { error, ProtocolViolation, StreamCode, StreamError } from "../error.ts";
 import type * as Path from "../path.ts";
 import { Reader, type Stream, type Writer } from "../stream.ts";
@@ -30,12 +31,19 @@ function finCancels(version: IetfVersion): boolean {
 export interface Refusal {
 	condition: RequestCondition;
 	reasonPhrase: string;
+	/** Whether asking again may pass, which invites the peer to retry. Absent is a verdict. */
+	retryable?: boolean;
 }
 
 async function writeRefusal(w: Writer, version: IetfVersion, kind: RequestKind, refusal: Refusal): Promise<void> {
+	const errorCode = toRequestCode(refusal.condition, kind, version);
+	// 0 asks the peer never to retry. As in Rust's `retry_interval`, the code must leave the
+	// request open too: UNINTERESTED tells an offer to stop, and the rest are verdicts.
+	const open = errorCode === 0x0 || errorCode === 0x6 || errorCode === 0x10;
 	const err = new RequestError({
-		errorCode: toRequestCode(refusal.condition, kind, version),
+		errorCode,
 		reasonPhrase: refusal.reasonPhrase,
+		retryInterval: refusal.retryable && open ? BigInt(RETRY_INITIAL) : 0n,
 	});
 	await Message.write(w, RequestError.id, err, version);
 }

js/net/src/resume.ts

@@ -74,7 +74,7 @@ import { clamped, type Resolved, type Window } from "./window.ts";
 import { type Broadcast as BroadcastWire, registerWire, wireOf } from "./wire.ts";
 
 // Stream codes that report the route failing to deliver rather than judging what it carried:
-// another copy may still deliver what it failed. Mirrors the Rust `route_failed`.
+// another copy may still deliver what it failed. Mirrors the Rust `Error::retryable`.
 const ROUTE_FAILURES: ReadonlySet<number> = new Set([
 	StreamCode.Internal,
 	StreamCode.Cancel,
@@ -113,7 +113,7 @@ function closing(err: Error): boolean {
 
 // A source losing its route to the content: only another route continues it, since asking the
 // same one again could only replace the withdrawal's cause with a refusal. Mirrors the Rust
-// `unroutable`.
+// `Error::unroutable`.
 function unroutable(err: Error): boolean {
 	return err instanceof StreamError && err.code === StreamCode.Unroutable;
 }

rs/moq-net/src/auth.rs

@@ -150,8 +150,9 @@ fn earliest(a: Option<Instant>, b: Option<Instant>) -> Option<Instant> {
 }
 
 /// How long a token refused for a reason that may pass waits before its session presents it
-/// again, doubling on each refusal up to [`RETRY_MAX`].
-const RETRY_INITIAL: Duration = Duration::from_secs(1);
+/// again, doubling on each refusal up to [`RETRY_MAX`]. A moq-transport refusal that may pass
+/// asks its peer to wait as long.
+pub(crate) const RETRY_INITIAL: Duration = Duration::from_secs(1);
 const RETRY_MAX: Duration = Duration::from_secs(30);
 
 /// Whether a refusal is the acceptor failing to decide rather than its verdict on the

rs/moq-net/src/error.rs

@@ -594,6 +594,41 @@ impl Error {
 			_ => Self::Transport(err.to_string()),
 		}
 	}
+
+	/// Whether asking again may succeed: the route failed to deliver rather than judging what
+	/// it carried, which would repeat on every attempt. Mirrors `routeFailed` in
+	/// `js/net/src/resume.ts`.
+	pub(crate) fn retryable(&self) -> bool {
+		self.unroutable() || self.transient()
+	}
+
+	/// Whether a source lost its route to the content: only another route continues it.
+	pub(crate) fn unroutable(&self) -> bool {
+		matches!(self, Self::Unroutable | Self::Stream(StreamError::Unroutable))
+	}
+
+	/// Whether the same source may deliver a fresh copy. An unroutable source needs a new
+	/// route instead: asking it again can replace the withdrawal's cause with a refusal.
+	pub(crate) fn transient(&self) -> bool {
+		matches!(
+			self,
+			Self::Transport(_)
+				| Self::IdleTimeout
+				| Self::Session(_)
+				| Self::GoingAway
+				| Self::GoawayTimeout
+				| Self::Dropped
+				| Self::Closed
+				| Self::Cancel
+				| Self::Stream(
+					StreamError::Session(_)
+						| StreamError::Internal
+						| StreamError::Cancel
+						| StreamError::GoingAway
+						| StreamError::ControlTimeout
+				)
+		)
+	}
 }
 
 /// Preserve the session registry when carrying a protocol error.

rs/moq-net/src/ietf/error.rs

@@ -248,6 +248,19 @@ pub(crate) mod request {
 		registered.unwrap_or(INTERNAL_ERROR)
 	}
 
+	/// The Retry Interval, in milliseconds, for a refusal with `err` that reads `code` on the wire.
+	///
+	/// A refusal asking again may pass invites the backoff a refused token starts from. 0 asks the
+	/// peer never to retry a verdict, which would repeat. The code must leave the request open
+	/// too: a failed session can carry a verdict (UNAUTHORIZED, NOT_SUPPORTED), and UNINTERESTED
+	/// tells an offer to stop.
+	pub(crate) fn retry_interval(err: &Error, code: u64) -> u64 {
+		match err.retryable() && matches!(code, INTERNAL_ERROR | DOES_NOT_EXIST | GOING_AWAY) {
+			true => crate::auth::RETRY_INITIAL.as_millis() as u64,
+			false => 0,
+		}
+	}
+
 	/// Read a rejection code received on the negotiated draft.
 	///
 	/// A code the draft does not assign for this request, INTERNAL_ERROR included, stays
@@ -304,6 +317,29 @@ pub(crate) mod request {
 			Error::Remote(0x30),
 		];
 
+		/// Only a refusal that asking again may pass invites a retry. An unroutable offer reads
+		/// UNINTERESTED, which tells the peer to stop offering, and a session failing with a
+		/// verdict (`Session(Unsupported)`) keeps it, so neither does.
+		#[test]
+		fn only_a_refusal_that_may_pass_invites_a_retry() {
+			let retry = crate::auth::RETRY_INITIAL.as_millis() as u64;
+			for version in ALL {
+				for kind in KINDS {
+					for err in EVERY_ERROR {
+						let code = to_code(&err, kind, version);
+						let expected = match err {
+							Error::GoingAway => retry,
+							Error::Unroutable if matches!(kind, Kind::Subscribe | Kind::Fetch | Kind::TrackStatus) => {
+								retry
+							}
+							_ => 0,
+						};
+						assert_eq!(retry_interval(&err, code), expected, "{version:?} {kind:?} {err:?}");
+					}
+				}
+			}
+		}
+
 		/// A relay bridging a moq-lite revocation onto a moq-transport request refuses it as
 		/// UNAUTHORIZED, the same as a local refusal.
 		#[test]

rs/moq-net/src/ietf/namespace.rs

@@ -175,15 +175,12 @@ pub(super) fn refusal(namespace: &Path, version: super::Version) -> Option<super
 	if !matches!(namespace.as_str().split('/').next(), Some("." | ".session")) {
 		return None;
 	}
-	Some(super::RequestError {
-		error_code: super::error::request::to_code(
-			&crate::Error::NotFound,
-			super::error::request::Kind::Subscribe,
-			version,
-		),
-		reason_phrase: "reserved namespace does not exist".into(),
-		retry_interval: 0,
-	})
+	Some(super::RequestError::refuse(
+		&crate::Error::NotFound,
+		super::error::request::Kind::Subscribe,
+		"reserved namespace does not exist",
+		version,
+	))
 }
 
 #[cfg(test)]