diff --git a/CLAUDE.md b/CLAUDE.md index 2678695774..842cc25a40 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -97,6 +97,10 @@ Don't document deprecated flags, options, or APIs. User-facing docs (`/doc`), `- The rename/removal rationale lives in the commit message and PR description, not in docs that users read. Warning someone who *uses* the deprecated path is not just fine but encouraged -- at compile time (Rust's `#[deprecated(note = "...")]`) or at runtime (a log line). Those fire on use, so they reach the one person who needs them and nobody else; they aren't documentation. A standing note in the docs that advertises the dead name is what's banned. +## Retries + +Fail fast; retry only what a few seconds can fix. Every retry loop uses capped exponential backoff with jitter, inlined at the call site (no shared `Backoff` type), and is bounded by *time*, not by error type: a short budget (~10s), then surface the last real error. Don't classify errors as retryable (`is_retryable()`); an ephemeral failure is one that clears within the budget, so the budget is the classifier. The one exception is an answer a peer actually sent, where the protocol defines the meaning: an HTTP status short-circuits (a `404` fails immediately; `408`/`429`/gateway statuses ride the backoff) via the `status()` accessors on `moq_native::Error` / `moq_hls::Error`. The only unbounded loops are process-lifetime supervisors with nobody to return an error to (cluster peers, device reopen, accept loops); they retry forever but cap the delay at seconds and warn per attempt, loudly broken rather than silently parked. Exactly one layer owns a retry: an outer loop that rebuilds an inner one resets its escalation, so watch the inner loop's terminal signal instead. + ## Root Cause First - Before fixing a bug, reproduce it and explain the mechanism. A fix that adds a retry, sleep, widened timeout, defensive check, or call-site special case without a stated mechanism is a symptom patch, not a fix. diff --git a/Cargo.lock b/Cargo.lock index 7efcd7ad4d..cee367d1c6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4353,6 +4353,7 @@ dependencies = [ "objc2-core-media", "objc2-foundation", "objc2-screen-capture-kit", + "rand 0.10.2", "rubato", "sonora", "thiserror 2.0.19", @@ -4491,6 +4492,7 @@ dependencies = [ "moq-mux", "moq-net", "percent-encoding", + "rand 0.10.2", "reqwest", "thiserror 2.0.19", "tokio", @@ -4668,6 +4670,7 @@ dependencies = [ "moq-stats", "moq-token", "qmux", + "rand 0.10.2", "rcgen", "reqwest", "reqwest-middleware", diff --git a/doc/bin/gstreamer.md b/doc/bin/gstreamer.md index fe13843891..028ab17971 100644 --- a/doc/bin/gstreamer.md +++ b/doc/bin/gstreamer.md @@ -41,9 +41,13 @@ signal when it changes, so you can poll it via `g_object_get` or connect to `not | `estimated-send-bitrate` | uint64 | Estimated send bitrate in bits per second (congestion controller); 0 when unavailable | | `estimated-recv-bitrate` | uint64 | Estimated receive bitrate in bits per second; 0 when unavailable | -`status` distinguishes a transient drop (`disconnected`, the reconnect loop is still retrying) from a -permanent give-up (`failed`, a non-retryable error such as an auth rejection), which a bare -`connected` bool cannot. +`status` distinguishes a drop the reconnect loop is still retrying (`disconnected`) from a permanent +give-up (`failed`), which a bare `connected` bool cannot. The sink retries for as long as the +pipeline runs, so a relay outage of any length is ridden out; it goes `failed` only on an answer the +relay actually gave that redialing cannot change: a rejected token, or a CONNECT answered with a +status that isn't an invitation to retry. Everything else keeps retrying, so watch the logs when a +sink stays `disconnected` from the very first attempt: a pipeline that has never connected once is +far more likely misconfigured than waiting out an outage. ## Prerequisites diff --git a/doc/bin/relay/cluster.md b/doc/bin/relay/cluster.md index f1c545b058..7c5916c828 100644 --- a/doc/bin/relay/cluster.md +++ b/doc/bin/relay/cluster.md @@ -117,6 +117,12 @@ Cluster peers must authenticate to each other: See [Authentication](/bin/relay/auth) for the full setup. +Peers are redialed indefinitely, with exponential backoff and jitter so a restarting cluster doesn't +reconnect in lockstep. That includes a peer that rejects us: a bad token logs `cluster peer error; +will retry` on every attempt rather than giving up, so watch for a peer that never reaches +`cluster peer session closed`. The delay escalates to ten seconds at most, so a dead or rejecting +peer stays loudly visible in the logs and a returning one is picked up within seconds. + ## Migration from older configs `cluster.root` was removed. To dial cluster peers use `cluster.connect`; to advertise this relay's own address set `cluster.node` and enable `cluster.mesh`. `cluster.mesh` is now a boolean gossip toggle (it used to take this relay's URL); the URL moved to `cluster.node`. The old `mesh = ""` form still works for backwards compatibility: it enables gossip and is treated as `cluster.node`, with a deprecation warning (or an error if it conflicts with an explicit `cluster.node`). diff --git a/js/CLAUDE.md b/js/CLAUDE.md index 4fcf7aae71..37f6898f33 100644 --- a/js/CLAUDE.md +++ b/js/CLAUDE.md @@ -80,6 +80,7 @@ Plain custom elements built directly on `@moq/signals`, no framework (except moq ## Conventions +- **Retry loops inline their backoff** (root Retries has the policy). Escalate a local delay toward a `max`, jitter each wait (`delay * (0.5 + Math.random() / 2)`), and hand it to `effect.timer`; a short time or attempt budget stops the loop, never a classification of the thrown value. - **Avoid callback parameters.** A function taking a `fn`/`create`/`onXxx` to invoke later reads poorly and hides control flow. Prefer returning a value the caller acts on, exposing a method or getter, or splitting into a couple of small calls the caller sequences itself (e.g. a cache `get()` then `insert(value)`, not `getOrCreate(key, () => value)`). Reserve callbacks for genuine event/subscription sinks where there is no alternative (`effect.subscribe`, DOM listeners, `Signal` subscriptions). - ESM only (`"type": "module"`). Relative imports include the `.ts`/`.tsx` extension in the lower-level packages (`net`, `signals`, `hang`); `rewriteRelativeImportExtensions` in `tsconfig.json` rewrites them to `.js` on build. Some higher-level packages (watch/publish) still omit extensions, so match the file you are editing. - Document every exported symbol and add a top-of-file `@module` doc block to each entrypoint (root convention; the published JSR/`.d.ts` docs render these). Use `@public` on the load-bearing classes. diff --git a/js/net/src/connection/reload.ts b/js/net/src/connection/reload.ts index bf2360b29f..e8b0bd9969 100644 --- a/js/net/src/connection/reload.ts +++ b/js/net/src/connection/reload.ts @@ -1,12 +1,18 @@ import { Effect, type Getter, Signal } from "@moq/signals"; import * as Announce from "../announced.ts"; +import { error } from "../error.ts"; import type * as Path from "../path.ts"; import { empty as emptyPath } from "../path.ts"; import { type ConnectProps, connect, type WebSocketOptions, type WebTransportProps } from "./connect.ts"; import type { Established } from "./established.ts"; import type { Probe, Stats } from "./stats.ts"; -/** Exponential backoff settings for {@link Reload}'s reconnect loop. */ +/** + * Exponential backoff settings for {@link Reload}'s reconnect loop. + * + * The delays carry jitter, so a fleet of tabs knocked offline together doesn't reconnect in + * lockstep. Every failure is retried; {@link ReloadDelay.timeout} is what stops the loop. + */ export type ReloadDelay = { /** The delay in milliseconds before reconnecting (default: 1000). */ initial: DOMHighResTimeStamp; @@ -14,13 +20,12 @@ export type ReloadDelay = { /** The multiplier for the delay (default: 2). */ multiplier: number; - /** The maximum delay in milliseconds (default: 30000). */ + /** The maximum delay in milliseconds (default: 5000). */ max: DOMHighResTimeStamp; /** * Maximum total time in milliseconds to spend retrying before giving up (default: - * 300000, 5 minutes). Resets after each successful connection. Set to 0 for - * unlimited retries. + * 10000). Resets after each successful connection. Set to 0 for unlimited retries. */ timeout?: DOMHighResTimeStamp; }; @@ -37,6 +42,14 @@ export type ReloadProps = Omit & { delay?: ReloadDelay; }; +/** + * How long to keep retrying before giving up, when {@link ReloadDelay.timeout} is unset. + * + * Short on purpose: a failure that clears within it was transient, and one that doesn't should + * surface as an error rather than leave the page silently reconnecting for minutes. + */ +const DEFAULT_TIMEOUT = 10000; + /** Current state of a {@link Reload} connection. */ export type ReloadStatus = "connecting" | "connected" | "disconnected"; @@ -80,15 +93,20 @@ export class Reload { /** The reactive effect scope driving the connect loop; closed by {@link Reload.close}. */ #signals = new Effect(); - /** Resolves when the reconnect loop stops via {@link Reload.close} or the retry timeout. */ + /** + * Resolves when the reconnect loop stops via {@link Reload.close}. + * + * Rejects when the loop gives up instead, carrying the failure that was in flight when the + * retry window expired. + */ closed: Promise; #closedResolve!: () => void; #closedReject!: (err: Error) => void; - #delay: DOMHighResTimeStamp; - - // Timestamp when the current retry sequence started (for timeout). - #retryStart: DOMHighResTimeStamp | undefined; + // The current wait between attempts, doubling per failure, and when the retry window expires. + // Both are undefined between sequences, so a later edit to `delay` applies to the next one. + #delay: DOMHighResTimeStamp | undefined; + #deadline: DOMHighResTimeStamp | undefined; // Increased by 1 each time to trigger a reload. #tick = new Signal(0); @@ -102,18 +120,21 @@ export class Reload { constructor(props?: ReloadProps) { this.url = Signal.from(props?.url); this.enabled = Signal.from(props?.enabled ?? false); - this.delay = props?.delay ?? { initial: 1000, multiplier: 2, max: 30000 }; + this.delay = props?.delay ?? { initial: 1000, multiplier: 2, max: 5000 }; this.webtransport = props?.webtransport; this.websocket = props?.websocket; this.discovery = props?.discovery; - this.#delay = this.delay.initial; - this.closed = new Promise((resolve, reject) => { this.#closedResolve = resolve; this.#closedReject = reject; }); + // A caller is free to never await `closed`, and giving up rejects it unprompted. Marking the + // rejection handled here keeps that from surfacing as an `unhandledrejection`; a consumer + // awaiting the same promise still receives it. + this.closed.catch(() => {}); + if (typeof window !== "undefined" && typeof document !== "undefined") { this.#signals.event(window, "pagehide", () => this.#suspended.set(true)); this.#signals.event(window, "pageshow", () => this.#suspended.set(false)); @@ -190,9 +211,9 @@ export class Reload { } /** - * Schedule the next connect attempt after the current backoff, or give up when the - * retry window has expired. `connected` is when the dead session was established, if - * it ever was, and `cause` the error that killed it, if it died with one. + * Schedule the next connect attempt after the current backoff, or stop once the retry window + * has expired. `connected` is when the dead session was established, if it ever was, and + * `cause` the error that killed it, if it died with one. */ #retry(effect: Effect, connected: DOMHighResTimeStamp | undefined, cause?: unknown): void { // Any session is dead now: report disconnected during the backoff rather than @@ -205,29 +226,29 @@ export class Reload { // shorter is a peer that accepts and immediately severs, which has to keep // escalating or we hammer it forever at the initial delay. if (connected !== undefined && performance.now() - connected >= this.delay.initial) { - this.#delay = this.delay.initial; - this.#retryStart = undefined; + this.#delay = undefined; + this.#deadline = undefined; } - // Track retry start for timeout. - this.#retryStart ??= performance.now(); - - const timeout = this.delay.timeout ?? 300000; - if (timeout > 0) { - const elapsed = performance.now() - this.#retryStart; - if (elapsed >= timeout) { - console.warn("reconnect timed out"); - // A graceful close has no error, so report the timeout itself. - if (cause === undefined) this.#closedReject(new Error("reconnect timed out")); - else this.#closedReject(cause instanceof Error ? cause : new Error(String(cause))); - return; - } - } + const now = performance.now(); + const timeout = this.delay.timeout ?? DEFAULT_TIMEOUT; + this.#delay ??= this.delay.initial; + this.#deadline ??= timeout > 0 ? now + timeout : Number.POSITIVE_INFINITY; - const tick = this.#tick.peek() + 1; - effect.timer(() => this.#tick.update((prev) => Math.max(prev, tick)), this.#delay); + if (now >= this.#deadline) { + console.warn("reconnect timed out"); + // A graceful close has no error, so report the timeout itself. + this.#closedReject(cause === undefined ? new Error("reconnect timed out") : error(cause)); + return; + } + // Equal jitter, so a fleet of tabs knocked offline together doesn't reconnect on the same + // tick, and never past the deadline the retry window promised. + const wait = Math.min(this.#delay * (0.5 + Math.random() / 2), this.#deadline - now); this.#delay = Math.min(this.#delay * this.delay.multiplier, this.delay.max); + + const tick = this.#tick.peek() + 1; + effect.timer(() => this.#tick.update((prev) => Math.max(prev, tick)), wait); } /** diff --git a/js/publish/src/source/camera.ts b/js/publish/src/source/camera.ts index b6c4028de6..2c33a6b6b8 100644 --- a/js/publish/src/source/camera.ts +++ b/js/publish/src/source/camera.ts @@ -76,9 +76,10 @@ export class Camera { const constraints = effect.get(this.constraints); if (!this.#retry.begin(effect, [device, constraints])) { - // Out of budget with the same settings. Only a change to what is plugged in is new - // information worth another attempt, so watch the device list here and not while - // healthy, where a rerun would restart a working capture for unrelated device churn. + // Waiting out a backoff, or out of budget entirely, with the same settings. Either way + // a change to what is plugged in is new information worth acting on now, so watch the + // device list here and not while healthy, where a rerun would restart a working capture + // for unrelated device churn. const spent = this.device.out.available.peek(); effect.subscribe(this.device.out.available, (available) => { if (available !== spent) this.#retry.refund(); diff --git a/js/publish/src/source/microphone.ts b/js/publish/src/source/microphone.ts index 250084ff28..54506beabd 100644 --- a/js/publish/src/source/microphone.ts +++ b/js/publish/src/source/microphone.ts @@ -64,9 +64,10 @@ export class Microphone { const constraints = effect.get(this.constraints); if (!this.#retry.begin(effect, [device, constraints])) { - // Out of budget with the same settings. Only a change to what is plugged in is new - // information worth another attempt, so watch the device list here and not while - // healthy, where a rerun would restart a working capture for unrelated device churn. + // Waiting out a backoff, or out of budget entirely, with the same settings. Either way + // a change to what is plugged in is new information worth acting on now, so watch the + // device list here and not while healthy, where a rerun would restart a working capture + // for unrelated device churn. const spent = this.device.out.available.peek(); effect.subscribe(this.device.out.available, (available) => { if (available !== spent) this.#retry.refund(); diff --git a/js/publish/src/source/retry.test.ts b/js/publish/src/source/retry.test.ts index 2f7c6ae0ac..4a604d98b9 100644 --- a/js/publish/src/source/retry.test.ts +++ b/js/publish/src/source/retry.test.ts @@ -43,7 +43,12 @@ class FakeMediaDevices extends EventTarget { return this.devices; } + // Every capture attempt, including the ones that never produce a track. `tracks` can't stand in: + // a missing device rejects without pushing one. + attempts = 0; + async getUserMedia(): Promise { + this.attempts += 1; if (this.missing) throw new Error("NotFoundError"); const track = new FakeTrack(); @@ -109,6 +114,48 @@ async function settle(times = 20): Promise { for (let i = 0; i < times; i++) await flush(); } +/** + * Poll until `pred` holds, so a regression fails the test instead of hanging it. + * + * A reopen waits out {@link Retry.DELAY} on a real timer, so microtask flushing alone never + * reaches it. Polling rather than sleeping for the exact delay keeps a loaded runner from + * deciding the outcome. + */ +async function waitUntil(pred: () => boolean): Promise { + const deadline = Date.now() + WAIT_TIMEOUT; + while (!pred()) { + if (Date.now() > deadline) throw new Error("timed out waiting for condition"); + await new Promise((resolve) => setTimeout(resolve, POLL_INTERVAL)); + } +} + +/** + * Poll until the capture stops reopening, meaning the budget is spent. + * + * Watches for quiet rather than an exact attempt count: a device-list change can refund the budget + * mid-cascade and buy another round. The window outlasts the largest backoff, so quiet here means + * stopped rather than mid-wait. + */ +async function waitSpent(media: FakeMediaDevices): Promise { + const quiet = (Retry.DELAY.max ?? 0) + QUIET_MARGIN; + + let seen = -1; + while (seen !== media.attempts) { + seen = media.attempts; + await new Promise((resolve) => setTimeout(resolve, quiet)); + } + await settle(); +} + +/** How long a `waitUntil` polls before calling the condition unreachable. */ +const WAIT_TIMEOUT = 10_000; +/** How often `waitUntil` re-checks; short, since everything it watches is in-process. */ +const POLL_INTERVAL = 5; +/** Quiet time past the largest backoff that proves the capture really stopped reopening. */ +const QUIET_MARGIN = 100; +/** Burning the whole budget waits out every backoff, which outlasts the default per-test timeout. */ +const SPENT_TIMEOUT = 30_000; + /** The track a source published, or undefined. */ function published(source: { track: MediaStreamTrack } | MediaStreamTrack | undefined): unknown { if (!source) return undefined; @@ -123,10 +170,10 @@ test("a microphone re-opens when its track dies", async () => { expect(media.tracks).toHaveLength(1); media.latest().end(); + await waitUntil(() => media.tracks.length === 2); await settle(); // A second capture happened, and the live replacement is what got published. - expect(media.tracks).toHaveLength(2); expect(published(mic.out.source.peek())).toBe(media.latest()); mic.close(); @@ -140,9 +187,9 @@ test("a camera re-opens when its track dies", async () => { expect(media.tracks).toHaveLength(1); media.latest().end(); + await waitUntil(() => media.tracks.length === 2); await settle(); - expect(media.tracks).toHaveLength(2); expect(published(camera.out.source.peek())).toBe(media.latest()); camera.close(); @@ -154,12 +201,18 @@ test("running out of retries clears the source instead of publishing a corpse", const mic = new Microphone({ enabled: true }); await settle(); - // Kill every replacement the moment it arrives. - for (let i = 0; i < Retry.LIMIT + 5; i++) { + // Kill every replacement the moment it arrives; the budget allows one reopen per failure. + for (let i = 0; i < Retry.LIMIT; i++) { + const before = media.tracks.length; media.latest().end(); - await settle(); + await waitUntil(() => media.tracks.length > before); } + // The budget is spent, so this death schedules nothing at all: a settle is enough to catch a + // reopen that shouldn't happen. + media.latest().end(); + await settle(); + // The first capture plus one retry per allowed failure, and nothing after. expect(media.tracks).toHaveLength(Retry.LIMIT + 1); @@ -182,85 +235,102 @@ test("a track that arrives dead is not published", async () => { mic.close(); }); -test("a replug recovers a capture whose retries all failed", async () => { - const media = install(new FakeMediaDevices()); +test( + "a replug recovers a capture whose retries all failed", + async () => { + const media = install(new FakeMediaDevices()); - const camera = new Camera({ enabled: true }); - await settle(); - expect(media.tracks).toHaveLength(1); + const camera = new Camera({ enabled: true }); + await settle(); + expect(media.tracks).toHaveLength(1); - // Unplug the only camera: the track dies and every reopen finds nothing. - media.missing = true; - media.replug([]); - media.latest().end(); - await settle(); + // Unplug the only camera: the track dies and every reopen finds nothing. + media.missing = true; + media.replug([]); + media.latest().end(); + await waitSpent(media); - const spent = media.tracks.length; - expect(camera.out.source.peek()).toBeUndefined(); + const spent = media.tracks.length; + expect(camera.out.source.peek()).toBeUndefined(); - // Plug it back in. Without the device-list refund this stays dead forever, because an unpinned - // `requested` is undefined both before and after, so nothing else reruns the capture. - media.missing = false; - media.replug([device("cam")]); - await settle(40); + // Plug it back in. Without the device-list refund this stays dead forever, because an unpinned + // `requested` is undefined both before and after, so nothing else reruns the capture. + media.missing = false; + media.replug([device("cam")]); + await waitUntil(() => media.tracks.length > spent); + await settle(); - expect(media.tracks.length).toBeGreaterThan(spent); - expect(published(camera.out.source.peek())).toBe(media.latest()); + expect(media.tracks.length).toBeGreaterThan(spent); + expect(published(camera.out.source.peek())).toBe(media.latest()); - camera.close(); -}); + camera.close(); + }, + SPENT_TIMEOUT, +); -test("picking a different device revives a capture whose retries all failed", async () => { - const media = install(new FakeMediaDevices()); - media.devices = [device("cam"), device("cam2")]; +test( + "picking a different device revives a capture whose retries all failed", + async () => { + const media = install(new FakeMediaDevices()); + media.devices = [device("cam"), device("cam2")]; - const camera = new Camera({ enabled: true }); - await settle(); + const camera = new Camera({ enabled: true }); + await settle(); - // Burn the budget: every attempt hands back a dead track. - media.bornDead = true; - camera.device.preferred.set("cam"); - await settle(40); + // Burn the budget: every attempt hands back a dead track. + media.bornDead = true; + camera.device.preferred.set("cam"); + await waitSpent(media); - const spent = media.tracks.length; - expect(camera.out.source.peek()).toBeUndefined(); + const spent = media.tracks.length; + expect(camera.out.source.peek()).toBeUndefined(); - // Selecting another device is the user's obvious recovery, and the device list has not changed, - // so nothing else would rerun the capture. - media.bornDead = false; - camera.device.preferred.set("cam2"); - await settle(40); + // Selecting another device is the user's obvious recovery, and the device list has not changed, + // so nothing else would rerun the capture. + media.bornDead = false; + camera.device.preferred.set("cam2"); + await waitUntil(() => media.tracks.length > spent); + await settle(); - expect(media.tracks.length).toBeGreaterThan(spent); - expect(published(camera.out.source.peek())).toBe(media.latest()); + expect(media.tracks.length).toBeGreaterThan(spent); + expect(published(camera.out.source.peek())).toBe(media.latest()); - camera.close(); -}); + camera.close(); + }, + SPENT_TIMEOUT, +); -test("fixing a constraint revives a capture whose retries all failed", async () => { - const media = install(new FakeMediaDevices()); +test( + "changing the constraints revives a capture whose retries all failed", + async () => { + const media = install(new FakeMediaDevices()); - const mic = new Microphone({ enabled: true, constraints: { channelCount: 99 } }); - await settle(); + const mic = new Microphone({ enabled: true, constraints: { channelCount: 99 } }); + await settle(); - // An impossible constraint fails instantly on every attempt, which is the quickest way to spend - // the whole budget. - media.missing = true; - mic.constraints.set({ channelCount: 98 }); - await settle(40); + // Spend the whole budget. The fake ignores constraints, so `missing` is what makes every + // attempt fail; the constraint edit only moves the settings the budget is keyed to. + media.missing = true; + mic.constraints.set({ channelCount: 98 }); + await waitSpent(media); - const spent = media.tracks.length; - expect(mic.out.source.peek()).toBeUndefined(); + const spent = media.tracks.length; + expect(mic.out.source.peek()).toBeUndefined(); - media.missing = false; - mic.constraints.set({ channelCount: 1 }); - await settle(40); + // Editing a constraint is new intent rather than another go at the same thing, so it buys a + // fresh budget. Nothing else here would rerun the capture: the device list never changed. + media.missing = false; + mic.constraints.set({ channelCount: 1 }); + await waitUntil(() => media.tracks.length > spent); + await settle(); - expect(media.tracks.length).toBeGreaterThan(spent); - expect(published(mic.out.source.peek())).toBe(media.latest()); + expect(media.tracks.length).toBeGreaterThan(spent); + expect(published(mic.out.source.peek())).toBe(media.latest()); - mic.close(); -}); + mic.close(); + }, + SPENT_TIMEOUT, +); test("unrelated device churn does not disturb a healthy capture", async () => { const media = install(new FakeMediaDevices()); diff --git a/js/publish/src/source/retry.ts b/js/publish/src/source/retry.ts index fba3a5df67..9a5a57e6df 100644 --- a/js/publish/src/source/retry.ts +++ b/js/publish/src/source/retry.ts @@ -18,17 +18,32 @@ export class Retry { /** How long a track has to survive before earlier failures stop counting against it. */ static readonly SETTLED = 5000; + /** + * Escalating pause between attempts, so a device mid-re-enumerate gets a moment rather than + * being asked three more times in the same millisecond. + * + * No give-up deadline: {@link LIMIT} is this budget, counted in attempts. Counting time instead + * would make the outcome depend on how long the OS takes to say no. + */ + static readonly DELAY = { initial: 250, multiplier: 2, max: 1000 }; + readonly #rerun = new Signal(0); // Deliberately plain fields: effect reruns must not unwind them, or the budget never runs out. #failures = 0; #settings: unknown[] | undefined; + #delay = Retry.DELAY.initial; + + // How long the next attempt still owes the backoff, set by `failed` and paid by `begin`. + #wait: DOMHighResTimeStamp | undefined; /** - * Subscribe the capture effect and report whether an attempt is still worth making. + * Subscribe the capture effect and report whether an attempt is worth making right now. * - * False means the budget is spent: return from the run without capturing, and the previous run's - * cleanup clears whatever it published. + * False means don't capture on this run: either the budget is spent, or the backoff from the + * last failure hasn't elapsed and a rerun is already scheduled for when it has. Return from the + * run either way, and the previous run's cleanup clears whatever it published. Splitting it like + * this is what lets a dead track be dropped immediately while the *reopen* still waits. * * The budget belongs to `settings`, the caller's live capture settings. Changing any of them is * new intent rather than another go at the same thing, so it starts a fresh budget: picking a @@ -39,22 +54,35 @@ export class Retry { if (settings.some((setting, i) => setting !== this.#settings?.[i])) { this.#settings = settings; - this.#failures = 0; + this.#clear(); + } + + if (this.#failures > Retry.LIMIT) return false; + + const wait = this.#wait; + if (wait !== undefined) { + this.#wait = undefined; + effect.timer(() => this.#rerun.update((rerun) => rerun + 1), wait); + return false; } - return this.#failures <= Retry.LIMIT; + return true; } /** The attempt produced no usable track. Spends budget and reruns the effect. */ failed(): void { this.#failures += 1; + // Unlimited budget, so there is always a next delay. + // Equal jitter, so a page with several captures doesn't reopen them all on the same tick. + this.#wait = this.#delay * (0.5 + Math.random() / 2); + this.#delay = Math.min(this.#delay * Retry.DELAY.multiplier, Retry.DELAY.max); this.#rerun.update((rerun) => rerun + 1); } /** The attempt produced a live track. Reruns the effect if it dies. */ succeeded(effect: Effect, track: MediaStreamTrack): void { effect.timer(() => { - this.#failures = 0; + this.#clear(); }, Retry.SETTLED); effect.event(track, "ended", () => this.failed()); @@ -62,9 +90,16 @@ export class Retry { /** Refund the budget, because something changed that makes another attempt worth trying. */ refund(): void { - if (this.#failures === 0) return; + if (this.#failures === 0 && this.#wait === undefined) return; - this.#failures = 0; + this.#clear(); this.#rerun.update((rerun) => rerun + 1); } + + /** Forget the failures so far, along with the pause they earned. */ + #clear(): void { + this.#failures = 0; + this.#wait = undefined; + this.#delay = Retry.DELAY.initial; + } } diff --git a/nix/modules/moq-relay.nix b/nix/modules/moq-relay.nix index 4abe8351a2..341a9eb9dd 100644 --- a/nix/modules/moq-relay.nix +++ b/nix/modules/moq-relay.nix @@ -201,8 +201,15 @@ in ExecStart = "${cfg.package}/bin/moq-relay"; + # Escalating restart backoff: 5s after the first failure, stepping up to a minute. A relay + # that dies on a bad config or an unbindable port would otherwise restart every 5s forever, + # which buries the real error under a restart loop and hammers whatever it dials. + # RestartSteps/RestartMaxDelaySec need systemd 254+; older versions log an unknown-directive + # warning and keep the fixed RestartSec. Restart = "on-failure"; RestartSec = "5s"; + RestartSteps = 5; + RestartMaxDelaySec = "1min"; # Security hardening NoNewPrivileges = true; diff --git a/packaging/moq-relay/moq-relay.service b/packaging/moq-relay/moq-relay.service index cb383948cd..9ea9d34621 100644 --- a/packaging/moq-relay/moq-relay.service +++ b/packaging/moq-relay/moq-relay.service @@ -7,8 +7,15 @@ Wants=network-online.target [Service] Type=notify ExecStart=/usr/bin/moq-relay --file /etc/moq-relay/relay.toml +# Escalating restart backoff: 5s after the first failure, stepping up to a minute. A relay that +# dies on a bad config or an unbindable port would otherwise restart every 5s forever, which buries +# the real error under a restart loop and hammers whatever it dials. RestartSteps and +# RestartMaxDelaySec need systemd 254+; older versions log an unknown-directive warning and keep the +# fixed RestartSec. Restart=on-failure RestartSec=5s +RestartSteps=5 +RestartMaxDelaySec=1min LimitNOFILE=1048576 # Sandboxing. DynamicUser allocates a transient UID, StateDirectory creates diff --git a/rs/CLAUDE.md b/rs/CLAUDE.md index 865c5a045c..44ff36d3f0 100644 --- a/rs/CLAUDE.md +++ b/rs/CLAUDE.md @@ -109,6 +109,7 @@ Negotiation: `version::NEGOTIATED` lists SETUP-negotiated versions in preference ## Rust conventions +- **Retry loops inline their backoff** (root Retries has the policy). Escalate a local `Duration` toward a `const MAX` (seconds, not minutes), jitter each wait (`delay.mul_f64(0.5 + rand::rng().random::() / 2.0)`), and track the give-up deadline as a `tokio::time::Instant` next to the delay. No shared backoff helper, and no `is_retryable()` on error types; the one thing worth reading off a failure is a status a peer actually sent, via the `status()` accessors on `moq_native::Error` / `moq_hls::Error`. - **Prefer `kio` over tokio sync primitives**: reach for `kio::Producer`/`Consumer` (and the `poll_*` plumbing) instead of `tokio::sync` channels or `watch`. A `tokio::sync::watch` (or a channel) carrying a single value is a code smell. `kio` ties into the runtime-free `poll_*` model and avoids a hard runtime dependency. - **Errors**: `thiserror` with `#[from]` for libraries, `anyhow` (with `.context("...")`, not `.map_err(|_| anyhow!())`) for binaries. Always `#[non_exhaustive]` on public error enums (e.g. `moq-net/src/error.rs`, `moq-ffi/src/error.rs`, `moq-loc/src/lib.rs`). Use `#[error(transparent)]` + `#[from]` for wrapped foreign errors (see `moq-token/src/error.rs`). - **Config + TOML merge**: any `#[arg]` field on a TOML-loadable config must be `Option`, never a bare `bool`/`String`/etc. The TOML->CLI merge re-applies clap defaults and silently clobbers TOML values for bare fields. See `moq-relay/src/config.rs` and its regression tests (`cli_does_not_clobber_toml_*`); add such a test for any new flag. diff --git a/rs/moq-audio/Cargo.toml b/rs/moq-audio/Cargo.toml index 688210210b..9cd2011e6e 100644 --- a/rs/moq-audio/Cargo.toml +++ b/rs/moq-audio/Cargo.toml @@ -60,6 +60,7 @@ moq-mux = { workspace = true } moq-net = { workspace = true } # We only use the sinc resampler (Async::new_sinc), so skip the default # fft_resampler feature and its RustFFT/realfft dependencies. +rand = "0.10" rubato = { version = "4.0", default-features = false } # Pure-Rust port of WebRTC's audio processing module (AEC3, noise suppression, # AGC2). The reason the `aec` feature can exist at all: every other option is a diff --git a/rs/moq-audio/src/playback/driver.rs b/rs/moq-audio/src/playback/driver.rs index 410597e1b7..bca51f9d36 100644 --- a/rs/moq-audio/src/playback/driver.rs +++ b/rs/moq-audio/src/playback/driver.rs @@ -13,15 +13,18 @@ use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; use cpal::traits::{DeviceTrait, StreamTrait}; +use rand::RngExt; use super::mixer::{self, Mixer}; use super::sink::{Registration, Sink}; use crate::Error; -/// Backoff bounds for reopening a device that failed. The first retry is quick -/// because the common case is a device that came right back (a USB re-enumerate, -/// a sample-rate change); the ceiling keeps a permanently gone device from -/// spinning. +/// Backoff bounds for reopening a device that failed. The first retry is quick because the common +/// case is a device that came right back (a USB re-enumerate, a sample-rate change); the ceiling +/// keeps a permanently gone device from spinning. +/// +/// No give-up budget: the engine outlives any one device, and the user plugging a headset back in +/// is exactly the external change a retry is waiting for. const RETRY_MIN: Duration = Duration::from_millis(500); const RETRY_MAX: Duration = Duration::from_secs(4); @@ -417,6 +420,7 @@ struct Driver { /// Bumped on every stream, so an error from a retired one can be told apart /// from one the live stream raised. generation: u64, + /// Delay before reopening a device that would not start, doubling per failure. retry: Duration, /// When a failed start may be retried, and what the command wait times out /// against. `None` while the stream is healthy. @@ -593,10 +597,13 @@ impl Driver { } /// When the next restart may be attempted, doubling the backoff. + /// + /// Jittered so a host running many streams doesn't reopen the device in lockstep after a + /// suspend or a driver reload. fn schedule(&mut self) -> Instant { - let at = Instant::now() + self.retry; + let wait = self.retry.mul_f64(0.5 + rand::rng().random::() / 2.0); self.retry = (self.retry * 2).min(RETRY_MAX); - at + Instant::now() + wait } /// Whether a failure reported by stream `generation` should rebuild the diff --git a/rs/moq-gst/src/sink/session.rs b/rs/moq-gst/src/sink/session.rs index fc2170bb20..89289bfec6 100644 --- a/rs/moq-gst/src/sink/session.rs +++ b/rs/moq-gst/src/sink/session.rs @@ -29,7 +29,7 @@ pub(crate) static CAT: LazyLock = /// The publish connection's lifecycle, surfaced as the `status` property. /// /// Bundles what a bare `connected` bool can't: `Failed` (a terminal give-up) is distinct from -/// `Disconnected` (a transient drop the reconnect loop is still retrying), so a consumer watching +/// `Disconnected` (a drop the reconnect loop is still retrying), so a consumer watching /// `notify::status` learns when a connection is newly established or permanently rejected. #[derive(Debug, Clone, Copy, PartialEq, Eq, Default, glib::Enum)] #[enum_type(name = "GstMoqSinkConnectionStatus")] @@ -42,7 +42,8 @@ pub enum ConnectionStatus { /// A session is connected and publishing. #[enum_value(name = "Connected: session established", nick = "connected")] Connected, - /// The reconnect loop gave up permanently (a non-retryable error, e.g. auth rejection). Terminal. + /// The reconnect loop gave up permanently (an auth rejection, or a CONNECT status that isn't an + /// invitation to retry). Terminal. #[enum_value(name = "Failed: connection rejected, gave up", nick = "failed")] Failed, } @@ -136,10 +137,13 @@ impl Session { let errored = Arc::new(AtomicBool::new(false)); // Publish through a background reconnect loop: connect, wait for close, reconnect with backoff. - // `timeout = 0` retries transport/connection failures indefinitely so an unattended publisher - // outlives relay/QUIC outages; non-retryable errors (e.g. auth) stay terminal. During an outage - // the pad threads keep writing (bounded by moq-net's per-group eviction) and the relay catches up - // from a group boundary on reconnect. A bounded policy is available via `ClientConfig::backoff`. + // `timeout = 0` drops the give-up deadline so an unattended publisher outlives relay/QUIC + // outages of any length, which is the trade this element wants: a pipeline nobody is watching + // should still be publishing when the relay comes back. The loop still ends on the two answers + // a server states outright (an auth rejection, or a CONNECT status that isn't an invitation to + // retry), posting the bus error below. During an outage the pad threads keep writing (bounded + // by moq-net's per-group eviction) and the relay catches up from a group boundary on + // reconnect. A bounded policy is available via `ClientConfig::backoff`. let mut config = moq_native::ClientConfig::default(); config.tls.disable_verify = Some(settings.tls_disable_verify); config.backoff.timeout = std::time::Duration::ZERO; diff --git a/rs/moq-hls/Cargo.toml b/rs/moq-hls/Cargo.toml index 836eebf7b1..5ec7e329de 100644 --- a/rs/moq-hls/Cargo.toml +++ b/rs/moq-hls/Cargo.toml @@ -30,6 +30,7 @@ m3u8-rs = "6" moq-mux = { workspace = true } moq-net = { workspace = true } percent-encoding = { workspace = true } +rand = "0.10" reqwest = { version = "0.13", default-features = false, features = ["rustls", "gzip"] } thiserror = "2" tokio = { workspace = true, features = ["full"] } diff --git a/rs/moq-hls/src/error.rs b/rs/moq-hls/src/error.rs index efa4b09532..50055295e3 100644 --- a/rs/moq-hls/src/error.rs +++ b/rs/moq-hls/src/error.rs @@ -18,6 +18,15 @@ impl std::fmt::Display for SequenceKind { } } +/// Whether an HTTP response status means "ask again later". +/// +/// A response that arrived is the server's answer, and only this narrow set invites another +/// attempt: request timeout, rate limit, and the gateway/overload statuses. Every other status, +/// `404` and `403` included, is settled. +pub(crate) fn status_retryable(status: u16) -> bool { + matches!(status, 408 | 429 | 502 | 503 | 504) +} + /// Errors produced by the HLS <-> MoQ gateway (import and export). #[derive(Debug, Clone, thiserror::Error)] #[non_exhaustive] @@ -124,6 +133,20 @@ pub enum Error { Other(std::sync::Arc), } +impl Error { + /// The HTTP status the origin answered with, if it answered with one at all. + /// + /// The import loop consults it: a `503` on a playlist fetch is worth another pass, a `404` is + /// the origin's settled answer. Nothing else here is classified; a failure with no status falls + /// through to the backoff budget. + pub fn status(&self) -> Option { + match self { + Self::Reqwest(err) => err.status().map(|status| status.as_u16()), + _ => None, + } + } +} + impl From for Error { fn from(err: reqwest::Error) -> Self { Error::Reqwest(std::sync::Arc::new(err)) @@ -144,3 +167,17 @@ impl From for Error { /// Convenience alias for results from the HLS gateway. pub type Result = std::result::Result; + +#[cfg(test)] +mod tests { + use super::*; + + /// The import loop consults this, so an origin's settled answer has to reach it intact. + #[test] + fn an_http_failure_reports_its_status() { + // A failure the origin never answered carries no status, so the budget decides instead. + assert_eq!(Error::NoVariants.status(), None); + assert_eq!(Error::ParsePlaylist("not a playlist".to_string()).status(), None); + assert_eq!(Error::Moq(moq_net::Error::Transport("lost".to_string())).status(), None); + } +} diff --git a/rs/moq-hls/src/export/mod.rs b/rs/moq-hls/src/export/mod.rs index e6178e8b4a..cac5652225 100644 --- a/rs/moq-hls/src/export/mod.rs +++ b/rs/moq-hls/src/export/mod.rs @@ -27,12 +27,24 @@ use std::sync::{Arc, Mutex}; use std::time::Duration; use moq_mux::catalog::{self, CatalogFormat, Stream}; +use rand::RngExt; pub(crate) use playlist::render_media; pub use rendition::{Kind, Rendition}; -/// How long to wait before retrying the initial catalog subscription. -const CATALOG_RETRY: Duration = Duration::from_millis(250); +/// Backoff bounds for the initial catalog subscription. +/// +/// The usual failure is a publisher that has announced its broadcast but not yet written its +/// catalog track, so this waits for external state rather than repeating a failed request. +/// Escalating for that reason: a source that stays silent for an hour must not be polled four times +/// a second for an hour. +/// +/// Deliberately no give-up budget, and no attempt to judge which failures are worth waiting on. The +/// broadcast closing is what ends the wait, and a relay-side broadcast outlives its publisher's +/// session, so any deadline here is a window in which a publisher outage leaves the broadcaster +/// permanently empty with nothing to recover it. +const CATALOG_RETRY_MIN: Duration = Duration::from_millis(250); +const CATALOG_RETRY_MAX: Duration = Duration::from_secs(5); /// Export tuning shared across renditions. /// @@ -173,13 +185,17 @@ async fn watch_catalog( config: Config, renditions: renditions::Producer, ) { + let mut delay = CATALOG_RETRY_MIN; + let mut consumer = loop { match catalog::Consumer::<()>::new(&broadcast, CatalogFormat::Hang).await { Ok(consumer) => break consumer, Err(err) => { tracing::warn!(%err, "failed to subscribe to broadcast catalog, retrying"); + let wait = delay.mul_f64(0.5 + rand::rng().random::() / 2.0); + delay = (delay * 2).min(CATALOG_RETRY_MAX); tokio::select! { - _ = tokio::time::sleep(CATALOG_RETRY) => {} + _ = tokio::time::sleep(wait) => {} _ = kio::wait(|waiter| broadcast.poll_closed(waiter)) => { renditions.close(); return; diff --git a/rs/moq-hls/src/import.rs b/rs/moq-hls/src/import.rs index 8aef546c66..abdb9376eb 100644 --- a/rs/moq-hls/src/import.rs +++ b/rs/moq-hls/src/import.rs @@ -19,6 +19,7 @@ use m3u8_rs::{ use moq_mux::catalog::Producer as CatalogProducer; use moq_mux::container::fmp4::Import as Fmp4; use moq_mux::select; +use rand::RngExt; use reqwest::Client; use tokio::io::{AsyncReadExt, AsyncSeekExt}; use tracing::{debug, info, warn}; @@ -29,9 +30,15 @@ use crate::{Error, Result, SequenceKind}; /// Per-request timeout for the default HTTP client (playlist + segment fetches). const REQUEST_TIMEOUT: Duration = Duration::from_secs(30); -/// Backoff before retrying after a failed import step, so a transient upstream -/// error (a 5xx, a truncated segment) doesn't tear down the whole import. -const ERROR_BACKOFF: Duration = Duration::from_secs(1); +/// Backoff bounds for retrying a failed import step, so a transient upstream error (a 503, a +/// dropped connection) doesn't tear down the whole import. +/// +/// The budget is short on purpose: a failure that clears within it was transient, and one that +/// doesn't is an outage the caller should hear about rather than one the import papers over while +/// publishing nothing. No error is classified; the budget is the filter. +const STEP_RETRY_MIN: Duration = Duration::from_secs(1); +const STEP_RETRY_MAX: Duration = Duration::from_secs(5); +const STEP_RETRY_BUDGET: Duration = Duration::from_secs(10); /// How far back from the live edge to start when (re-)anchoring to a playlist window. /// @@ -86,6 +93,21 @@ struct StepOutcome { wrote_segments: usize, /// Target segment duration (in seconds) from the playlist, if known. target_duration: Option, + /// Renditions that ingested without error, however many segments each wrote. + /// + /// A live playlist with no new segments yet still counts: that's a healthy rendition with + /// nothing to add this pass, not a failure. Counting segments instead would read a quiet + /// playlist as a dead one. + ok: usize, + /// A rendition failure from this step, if any. + /// + /// [`OnError::Warn`] keeps the other renditions going after one fails, so a step can report + /// `Ok` having imported nothing at all. The loop needs to tell that apart from a quiet playlist + /// with no new segments, or it treats a permanently broken source as steady progress. + /// + /// When several renditions fail, this is the one whose failure another pass could still clear, + /// so a single permanently-dead variant doesn't end an import the others could still serve. + failed: Option, } /// What a step does when a rendition fails. @@ -589,15 +611,58 @@ impl Import { /// Run the import loop until cancelled. /// - /// A failed step (e.g. a transient playlist fetch error) is logged and - /// retried after a short backoff rather than ending the import. + /// A failed step is logged and retried with escalating backoff, and the import ends with the + /// last error once the backoff budget is spent, so a broken source surfaces within seconds + /// instead of looping forever. The one shortcut is an HTTP status the origin actually sent: a + /// `404` playlist ends the import immediately, since no amount of waiting turns it into a `200`. pub async fn run(&mut self) -> Result<()> { + let mut delay = STEP_RETRY_MIN; + let mut deadline = tokio::time::Instant::now() + STEP_RETRY_BUDGET; + loop { - let outcome = match self.step(OnError::Warn).await { - Ok(outcome) => outcome, + // A step where *nothing* ingested while a rendition was failing is a failed pass wearing + // an `Ok`: `step` swallows per-rendition errors so one bad variant doesn't drop the rest. + // Letting it through would reset the backoff every pass, so a source that returns 404 + // forever would spin at the refresh cadence while publishing nothing. + // + // Keyed on renditions that ingested, not on segments written: a healthy live playlist + // with nothing new this pass writes no segments, and reading that as a failure would end + // a multi-rendition import over one dead variant the others were covering for. + let stepped = match self.step(OnError::Warn).await { + Ok(StepOutcome { + ok: 0, + failed: Some(err), + .. + }) => Err(err), + stepped => stepped, + }; + + let outcome = match stepped { + Ok(outcome) => { + // The source is alive, so the earlier failures no longer describe it. + delay = STEP_RETRY_MIN; + deadline = tokio::time::Instant::now() + STEP_RETRY_BUDGET; + outcome + } + // A status the origin actually sent is its answer: a 404 playlist is not going to + // become a 200 on the next pass. Everything else rides the backoff budget. + Err(err) if err.status().is_some_and(|status| !crate::status_retryable(status)) => { + return Err(err); + } Err(err) => { + let now = tokio::time::Instant::now(); + if now >= deadline { + return Err(err); + } + warn!(%err, "HLS import step failed, retrying"); - tokio::time::sleep(ERROR_BACKOFF).await; + // Jittered so a fleet of importers pointed at one origin doesn't retry in + // lockstep, and never past the deadline the budget promised. + let wait = delay + .mul_f64(0.5 + rand::rng().random::() / 2.0) + .min(deadline - now); + delay = (delay * 2).min(STEP_RETRY_MAX); + tokio::time::sleep(wait).await; continue; } }; @@ -644,15 +709,30 @@ impl Import { let mut wrote_segments = 0; let mut target_duration = None; + let mut failed = None; + let mut ok = 0; for track in self.video.iter_mut().chain(self.audio.iter_mut()) { match track.ingest(&self.fetcher, &mut target_duration).await { - Ok(count) => wrote_segments += count, + Ok(count) => { + wrote_segments += count; + ok += 1; + } Err(err) => match on_error { OnError::Fail => return Err(err), // Keep the other renditions going: one bad variant or segment shouldn't // drop the rest or abort the whole step. - OnError::Warn => warn!(label = %track.label, %err, "rendition import step failed, will retry"), + OnError::Warn => { + warn!(label = %track.label, %err, "rendition import step failed, will retry"); + // Prefer a failure another pass could still clear, so a single + // permanently-dead variant doesn't end an import the others could still + // serve. Keeping whichever error came last instead would make the outcome + // depend on rendition order. + let recoverable = err.status().is_none_or(crate::status_retryable); + if recoverable || failed.is_none() { + failed = Some(err); + } + } }, } } @@ -660,6 +740,8 @@ impl Import { Ok(StepOutcome { wrote_segments, target_duration, + failed, + ok, }) } diff --git a/rs/moq-hls/src/lib.rs b/rs/moq-hls/src/lib.rs index 5b97dd6cda..1708142543 100644 --- a/rs/moq-hls/src/lib.rs +++ b/rs/moq-hls/src/lib.rs @@ -25,6 +25,7 @@ pub mod import; #[cfg(feature = "server")] pub mod server; +pub(crate) use error::status_retryable; pub use error::*; #[cfg(feature = "server")] pub use server::Server; diff --git a/rs/moq-native/src/error.rs b/rs/moq-native/src/error.rs index 434a85ae7b..da4251168c 100644 --- a/rs/moq-native/src/error.rs +++ b/rs/moq-native/src/error.rs @@ -1,5 +1,14 @@ use std::sync::Arc; +/// Whether an HTTP response status means "ask again later". +/// +/// A response that arrived is the server's answer, and only this narrow set invites another +/// attempt: request timeout, rate limit, and the gateway/overload statuses. Every other status, +/// `404` and `403` included, is settled. +pub(crate) fn status_retryable(status: u16) -> bool { + matches!(status, 408 | 429 | 502 | 503 | 504) +} + /// Errors produced while configuring or establishing native MoQ connections. /// /// Backend-specific failures live in per-backend error types ([`crate::tls::Error`], @@ -145,6 +154,37 @@ impl Error { pub fn is_auth(&self) -> bool { self.connect_error().is_some_and(|err| err.is_auth()) } + + /// The HTTP status a server answered a connection attempt with, if it answered with one at all. + /// + /// `None` covers everything else: a dial that never got a response, a QUIC handshake that + /// failed, a URL we couldn't parse. Only a status the peer actually sent shows up here, and + /// whether it invites another attempt is the caller's call (`408`, `429`, `502`, `503`, and + /// `504` are the ones worth repeating). This deliberately does not try to say whether some + /// *other* kind of failure is worth retrying; that's a guess, and a backoff budget bounds it + /// instead. + pub fn status(&self) -> Option { + match self { + // A race is only settled when both halves were answered, and answered with something not + // worth repeating: one transport being refused says nothing about the other, so a `404` + // over QUIC alongside a dead WebSocket is still just a failed dial. + #[cfg(feature = "websocket")] + Self::TransportRace { quic, websocket } => match (quic.status(), websocket.status()) { + (Some(quic), Some(websocket)) if !status_retryable(quic) && !status_retryable(websocket) => Some(quic), + _ => None, + }, + + #[cfg(feature = "quinn")] + Self::Quinn(err) => err.status(), + #[cfg(feature = "noq")] + Self::Noq(err) => err.status(), + #[cfg(feature = "quiche")] + Self::Quiche(err) => err.status(), + #[cfg(feature = "websocket")] + Self::WebSocket(err) => err.status(), + _ => None, + } + } } // The wrapped sources aren't `Clone`, so `#[from]` can't store them behind `Arc` diff --git a/rs/moq-native/src/noq.rs b/rs/moq-native/src/noq.rs index 8ca8e26786..01f4e44e0e 100644 --- a/rs/moq-native/src/noq.rs +++ b/rs/moq-native/src/noq.rs @@ -397,6 +397,33 @@ impl Error { _ => None, } } + + /// The HTTP status a server answered with, if it answered with one at all. + /// + /// Two places see a real status: the insecure `http://` fingerprint bootstrap, and the + /// WebTransport CONNECT response. See [`crate::Error::status`]. + pub(crate) fn status(&self) -> Option { + match self { + Self::FetchFingerprint(err) | Self::FingerprintStatus(err) | Self::ReadFingerprint(err) => { + err.status().map(|status| status.as_u16()) + } + Self::Client(err) => client_status(err), + // Every raced address has to have answered, and answered with something not worth + // repeating, before the set counts as settled: one address refusing says nothing about + // the others, which may simply have been unroutable. + Self::Failover(failures) => { + let mut settled = None; + for failure in failures { + match failure.error.status() { + Some(status) if !crate::error::status_retryable(status) => settled = Some(status), + _ => return None, + } + } + settled + } + _ => None, + } + } } fn map_client_error(err: web_transport_noq::ClientError) -> Error { @@ -408,26 +435,35 @@ fn map_client_error(err: web_transport_noq::ClientError) -> Error { } fn classify_client_error(err: &web_transport_noq::ClientError) -> Option { + client_status(err).and_then(crate::ConnectError::from_status_u16) +} + +/// The HTTP status the server answered the WebTransport CONNECT with, when it answered with one at +/// all (as opposed to the connection failing underneath the request). +/// +/// Both classifications read this: [`classify_client_error`] turns an auth status into a +/// [`crate::ConnectError`], and [`Error::status`] hands it to the caller, whose backoff consults +/// the status. A `404` or `405` is the server's settled answer, so retrying +/// it just burns the reconnect budget on a URL that will never work. +fn client_status(err: &web_transport_noq::ClientError) -> Option { match err { - web_transport_noq::ClientError::HttpError(err) => classify_connect_error(err), + web_transport_noq::ClientError::HttpError(err) => connect_status(err), _ => None, } } -fn classify_connect_error(err: &web_transport_noq::ConnectError) -> Option { +fn connect_status(err: &web_transport_noq::ConnectError) -> Option { match err { - web_transport_noq::ConnectError::ErrorStatus(status) => crate::ConnectError::from_status_u16(status.as_u16()), - web_transport_noq::ConnectError::ProtoError(err) => classify_proto_error(err), + web_transport_noq::ConnectError::ErrorStatus(status) => Some(status.as_u16()), + web_transport_noq::ConnectError::ProtoError(err) => proto_status(err), _ => None, } } -fn classify_proto_error(err: &web_transport_noq::proto::ConnectError) -> Option { +fn proto_status(err: &web_transport_noq::proto::ConnectError) -> Option { match err { web_transport_noq::proto::ConnectError::ErrorStatus(status) - | web_transport_noq::proto::ConnectError::WrongStatus(Some(status)) => { - crate::ConnectError::from_status_u16(status.as_u16()) - } + | web_transport_noq::proto::ConnectError::WrongStatus(Some(status)) => Some(status.as_u16()), _ => None, } } diff --git a/rs/moq-native/src/quiche.rs b/rs/moq-native/src/quiche.rs index f93e65f45f..6cc28a5793 100644 --- a/rs/moq-native/src/quiche.rs +++ b/rs/moq-native/src/quiche.rs @@ -473,6 +473,33 @@ impl Error { _ => None, } } + + /// The HTTP status a server answered with, if it answered with one at all. + /// + /// Two places see a real status: the insecure `http://` fingerprint bootstrap, and the + /// WebTransport CONNECT response. See [`crate::Error::status`]. + pub(crate) fn status(&self) -> Option { + match self { + Self::FetchFingerprint(err) | Self::FingerprintStatus(err) | Self::ReadFingerprint(err) => { + err.status().map(|status| status.as_u16()) + } + Self::ClientConnect(err) => client_status(err), + // Every raced address has to have answered, and answered with something not worth + // repeating, before the set counts as settled: one address refusing says nothing about + // the others, which may simply have been unroutable. + Self::Failover(failures) => { + let mut settled = None; + for failure in failures { + match failure.error.status() { + Some(status) if !crate::error::status_retryable(status) => settled = Some(status), + _ => return None, + } + } + settled + } + _ => None, + } + } } fn map_client_error(err: web_transport_quiche::ClientError) -> Error { @@ -484,26 +511,35 @@ fn map_client_error(err: web_transport_quiche::ClientError) -> Error { } fn classify_client_error(err: &web_transport_quiche::ClientError) -> Option { + client_status(err).and_then(crate::ConnectError::from_status_u16) +} + +/// The HTTP status the server answered the WebTransport CONNECT with, when it answered with one at +/// all (as opposed to the connection failing underneath the request). +/// +/// Both classifications read this: [`classify_client_error`] turns an auth status into a +/// [`crate::ConnectError`], and [`Error::status`] hands it to the caller, whose backoff consults +/// the status. A `404` or `405` is the server's settled answer, so retrying +/// it just burns the reconnect budget on a URL that will never work. +fn client_status(err: &web_transport_quiche::ClientError) -> Option { match err { - web_transport_quiche::ClientError::Connect(err) => classify_connect_error(err), + web_transport_quiche::ClientError::Connect(err) => connect_status(err), _ => None, } } -fn classify_connect_error(err: &web_transport_quiche::h3::ConnectError) -> Option { +fn connect_status(err: &web_transport_quiche::h3::ConnectError) -> Option { match err { - web_transport_quiche::h3::ConnectError::Status(status) => crate::ConnectError::from_status_u16(status.as_u16()), - web_transport_quiche::h3::ConnectError::Proto(err) => classify_proto_error(err), + web_transport_quiche::h3::ConnectError::Status(status) => Some(status.as_u16()), + web_transport_quiche::h3::ConnectError::Proto(err) => proto_status(err), _ => None, } } -fn classify_proto_error(err: &web_transport_quiche::proto::ConnectError) -> Option { +fn proto_status(err: &web_transport_quiche::proto::ConnectError) -> Option { match err { web_transport_quiche::proto::ConnectError::ErrorStatus(status) - | web_transport_quiche::proto::ConnectError::WrongStatus(Some(status)) => { - crate::ConnectError::from_status_u16(status.as_u16()) - } + | web_transport_quiche::proto::ConnectError::WrongStatus(Some(status)) => Some(status.as_u16()), _ => None, } } diff --git a/rs/moq-native/src/quinn.rs b/rs/moq-native/src/quinn.rs index 1ba24ba6e3..f0a536c3c4 100644 --- a/rs/moq-native/src/quinn.rs +++ b/rs/moq-native/src/quinn.rs @@ -410,6 +410,33 @@ impl Error { _ => None, } } + + /// The HTTP status a server answered with, if it answered with one at all. + /// + /// Two places see a real status: the insecure `http://` fingerprint bootstrap, and the + /// WebTransport CONNECT response. See [`crate::Error::status`]. + pub(crate) fn status(&self) -> Option { + match self { + Self::FetchFingerprint(err) | Self::FingerprintStatus(err) | Self::ReadFingerprint(err) => { + err.status().map(|status| status.as_u16()) + } + Self::Client(err) => client_status(err), + // Every raced address has to have answered, and answered with something not worth + // repeating, before the set counts as settled: one address refusing says nothing about + // the others, which may simply have been unroutable. + Self::Failover(failures) => { + let mut settled = None; + for failure in failures { + match failure.error.status() { + Some(status) if !crate::error::status_retryable(status) => settled = Some(status), + _ => return None, + } + } + settled + } + _ => None, + } + } } fn map_client_error(err: web_transport_quinn::ClientError) -> Error { @@ -421,26 +448,35 @@ fn map_client_error(err: web_transport_quinn::ClientError) -> Error { } fn classify_client_error(err: &web_transport_quinn::ClientError) -> Option { + client_status(err).and_then(crate::ConnectError::from_status_u16) +} + +/// The HTTP status the server answered the WebTransport CONNECT with, when it answered with one at +/// all (as opposed to the connection failing underneath the request). +/// +/// Both classifications read this: [`classify_client_error`] turns an auth status into a +/// [`crate::ConnectError`], and [`Error::status`] hands it to the caller, whose backoff consults +/// the status. A `404` or `405` is the server's settled answer, so retrying +/// it just burns the reconnect budget on a URL that will never work. +fn client_status(err: &web_transport_quinn::ClientError) -> Option { match err { - web_transport_quinn::ClientError::HttpError(err) => classify_connect_error(err), + web_transport_quinn::ClientError::HttpError(err) => connect_status(err), _ => None, } } -fn classify_connect_error(err: &web_transport_quinn::ConnectError) -> Option { +fn connect_status(err: &web_transport_quinn::ConnectError) -> Option { match err { - web_transport_quinn::ConnectError::ErrorStatus(status) => crate::ConnectError::from_status_u16(status.as_u16()), - web_transport_quinn::ConnectError::ProtoError(err) => classify_proto_error(err), + web_transport_quinn::ConnectError::ErrorStatus(status) => Some(status.as_u16()), + web_transport_quinn::ConnectError::ProtoError(err) => proto_status(err), _ => None, } } -fn classify_proto_error(err: &web_transport_quinn::proto::ConnectError) -> Option { +fn proto_status(err: &web_transport_quinn::proto::ConnectError) -> Option { match err { web_transport_quinn::proto::ConnectError::ErrorStatus(status) - | web_transport_quinn::proto::ConnectError::WrongStatus(Some(status)) => { - crate::ConnectError::from_status_u16(status.as_u16()) - } + | web_transport_quinn::proto::ConnectError::WrongStatus(Some(status)) => Some(status.as_u16()), _ => None, } } @@ -686,6 +722,43 @@ impl quinn::ConnectionIdGenerator for ServerIdGenerator { mod tests { use super::*; + fn connect_rejected(status: u16) -> Error { + Error::Client(web_transport_quinn::ClientError::HttpError( + web_transport_quinn::ConnectError::ErrorStatus( + web_transport_quinn::http::StatusCode::from_u16(status).unwrap(), + ), + )) + } + + /// A CONNECT the relay answered carries its status through to the caller, so a wrong path or an + /// endpoint that doesn't speak WebTransport can surface immediately rather than after the whole + /// reconnect budget. + #[test] + fn a_rejected_connect_reports_its_status() { + for status in [400, 404, 405, 410, 501] { + assert_eq!(connect_rejected(status).status(), Some(status)); + assert!( + !crate::error::status_retryable(status), + "{status} should stop the reconnect loop" + ); + } + + for status in [408, 429, 502, 503, 504] { + assert_eq!(connect_rejected(status).status(), Some(status)); + assert!(crate::error::status_retryable(status), "{status} should be retried"); + } + + // Auth is peeled off into its own variant before reaching the generic client arm. + assert_eq!( + connect_rejected(401).connect_error(), + Some(crate::ConnectError::Unauthorized) + ); + assert_eq!( + connect_rejected(403).connect_error(), + Some(crate::ConnectError::Forbidden) + ); + } + /// Build a controller from each family's factory and downcast it to the /// concrete quinn implementation it must map to. #[test] diff --git a/rs/moq-native/src/reconnect.rs b/rs/moq-native/src/reconnect.rs index a409cac753..7041e66495 100644 --- a/rs/moq-native/src/reconnect.rs +++ b/rs/moq-native/src/reconnect.rs @@ -4,11 +4,21 @@ use std::time::Duration; use moq_net::Version; use moq_net::bandwidth::{Consumer as BandwidthConsumer, Producer as BandwidthProducer}; use moq_net::kio; +use rand::RngExt; use url::Url; use crate::{Client, Error}; /// Exponential backoff configuration for reconnection attempts. +/// +/// This decides how long to wait between reconnect attempts and when to give up. The delays carry +/// jitter, so a fleet knocked offline together doesn't reconnect in lockstep. +/// +/// [`timeout`](Self::timeout) is what ends a hopeless loop: every failure rides the same backoff, +/// and the short default budget is what surfaces a broken target instead of hiding it. The only +/// failures short-circuited are answers a server actually gave (an auth rejection, or a CONNECT +/// status that isn't an invitation to retry). A zero timeout removes the backstop, so it belongs +/// only where an unattended process must outlive an outage of any length. #[derive(Clone, Debug, clap::Args, serde::Serialize, serde::Deserialize)] #[serde(default, deny_unknown_fields)] #[non_exhaustive] @@ -32,7 +42,7 @@ pub struct Backoff { #[arg( id = "backoff-max", long, - default_value = "30s", + default_value = "5s", env = "MOQ_BACKOFF_MAX", value_parser = humantime::parse_duration, )] @@ -46,7 +56,7 @@ pub struct Backoff { #[arg( id = "backoff-timeout", long, - default_value = "5m", + default_value = "10s", env = "MOQ_BACKOFF_TIMEOUT", value_parser = humantime::parse_duration, )] @@ -59,8 +69,8 @@ impl Default for Backoff { Self { initial: Duration::from_secs(1), multiplier: 2, - max: Duration::from_secs(30), - timeout: Duration::from_secs(300), + max: Duration::from_secs(5), + timeout: Duration::from_secs(10), } } } @@ -79,6 +89,15 @@ impl Backoff { } } +/// When a reconnect sequence gives up, or `None` when [`Backoff::timeout`] is zero and it never +/// does. Measured from now, so it covers the connect attempts as well as the waits between them. +fn deadline_from(backoff: &Backoff) -> Option { + match backoff.timeout.is_zero() { + true => None, + false => Some(tokio::time::Instant::now() + backoff.timeout), + } +} + /// A connection lifecycle transition reported by [`Reconnect::status`]. #[derive(Clone, Copy, Debug, PartialEq, Eq)] #[non_exhaustive] @@ -99,7 +118,8 @@ struct State { status: Option, /// The negotiated MoQ version of the live session, or `None` when disconnected. version: Option, - /// Set when the reconnect loop permanently gives up (reconnect timeout exceeded). + /// Set when the reconnect loop permanently gives up: the backoff timeout expiring, or a server + /// answer that redialing cannot change. error: Option, /// The currently-connected session, or `None` while reconnecting. Read by /// [`ConnectionStatsReader`] to snapshot live connection stats. @@ -125,7 +145,11 @@ impl ConnectionStatsReader { /// Handle to a background reconnect loop. /// /// Spawns a tokio task that connects, waits for session close, then reconnects with exponential -/// backoff. The read surface mirrors [`moq_net::Session`] so a caller can treat it like a session +/// backoff until [`Backoff::timeout`] runs out. This loop is the only retry owner for the connection: +/// a caller that rebuilds it on failure restarts the backoff from its initial delay, which turns the +/// escalation back into a tight loop. Watch [`closed`](Self::closed) instead. +/// +/// The read surface mirrors [`moq_net::Session`] so a caller can treat it like a session /// that transparently reconnects: [`version`](Self::version), [`send_bandwidth`](Self::send_bandwidth), /// and [`recv_bandwidth`](Self::recv_bandwidth) track the live session and reset while disconnected. /// The extra toggle a plain session doesn't have is the connection lifecycle: [`connected`](Self::connected) @@ -180,20 +204,14 @@ impl Reconnect { url: Url, backoff: Backoff, ) -> crate::Result<()> { + // The escalating wait between attempts, and the instant the give-up budget expires. Both + // restart after a session that stayed healthy, so a one-off drop reconnects promptly. A zero + // timeout means no deadline at all: retry for as long as the process lives. let mut delay = backoff.initial; - let mut retry_start = tokio::time::Instant::now(); + let mut deadline = deadline_from(&backoff); let mut last_error: Option = None; loop { - if !backoff.timeout.is_zero() && retry_start.elapsed() > backoff.timeout { - let timeout = backoff.timeout; - let msg = match last_error { - Some(err) => format!("reconnect timed out after {timeout:?}: {err}"), - None => format!("reconnect timed out after {timeout:?}"), - }; - return Err(Error::Reconnect(msg)); - } - tracing::info!(%url, "connecting"); match client.connect(url.clone()).await { @@ -223,7 +241,7 @@ impl Reconnect { // window so a one-off drop reconnects promptly. tracing::warn!(%url, "session closed, reconnecting"); delay = backoff.initial; - retry_start = tokio::time::Instant::now(); + deadline = deadline_from(&backoff); last_error = None; } else { // Connected then dropped almost immediately (e.g. the server accepts then @@ -240,16 +258,42 @@ impl Reconnect { } } Err(err) => { + // The two answers a server can give that redialing cannot change: it rejected our + // credentials, or it answered the CONNECT with a status that isn't an invitation + // to come back. Everything else falls through to the backoff, whose budget is + // what stops the loop. if err.is_auth() { return Err(err); } + if let Some(status) = err.status() + && !crate::error::status_retryable(status) + { + return Err(err); + } last_error = Some(err); } } - tracing::warn!(%url, ?delay, "reconnecting after backoff"); - tokio::time::sleep(delay).await; - delay = std::cmp::min(delay * backoff.multiplier, backoff.max); + let now = tokio::time::Instant::now(); + if deadline.is_some_and(|deadline| now >= deadline) { + let timeout = backoff.timeout; + let msg = match last_error { + Some(err) => format!("reconnect timed out after {timeout:?}: {err}"), + None => format!("reconnect timed out after {timeout:?}"), + }; + return Err(Error::Reconnect(msg)); + } + + // Jittered so a fleet knocked offline together doesn't reconnect on the same tick, and + // never past the deadline the budget promised. + let mut wait = delay.mul_f64(0.5 + rand::rng().random::() / 2.0); + if let Some(deadline) = deadline { + wait = wait.min(deadline - now); + } + delay = (delay * backoff.multiplier.max(1)).min(backoff.max); + + tracing::warn!(%url, ?wait, "reconnecting after backoff"); + tokio::time::sleep(wait).await; } } @@ -315,8 +359,9 @@ impl Reconnect { /// Poll whether the reconnect loop has stopped. /// - /// `Ready(Err)` if it permanently gave up (reconnect timeout exceeded), `Ready(Ok(()))` if - /// stopped by dropping the handle, `Pending` while it's still running. + /// `Ready(Err)` if it permanently gave up (a failure no retry can clear, or the backoff timeout + /// expiring), `Ready(Ok(()))` if stopped by dropping the handle, `Pending` while it's still + /// running. pub fn poll_closed(&self, waiter: &kio::Waiter) -> Poll> { ready!(self.state.poll_closed(waiter)); Poll::Ready(match &self.state.read().error { @@ -415,8 +460,8 @@ mod tests { let backoff = Backoff::default(); assert_eq!(backoff.initial, Duration::from_secs(1)); assert_eq!(backoff.multiplier, 2); - assert_eq!(backoff.max, Duration::from_secs(30)); - assert_eq!(backoff.timeout, Duration::from_secs(300)); + assert_eq!(backoff.max, Duration::from_secs(5)); + assert_eq!(backoff.timeout, Duration::from_secs(10)); } /// The linger outlives the give-up timeout (so the reconnect error surfaces diff --git a/rs/moq-native/src/websocket.rs b/rs/moq-native/src/websocket.rs index c1fb1d1433..f99e0b66bc 100644 --- a/rs/moq-native/src/websocket.rs +++ b/rs/moq-native/src/websocket.rs @@ -223,6 +223,17 @@ impl Error { _ => None, } } + + /// The HTTP status the server answered the upgrade with, if it answered with one at all. + /// + /// qmux surfaces a non-101 WebSocket upgrade response as `Http(status)`. See + /// [`crate::Error::status`]. + pub(crate) fn status(&self) -> Option { + match self { + Self::Connect(qmux::Error::Http(status)) => Some(*status), + _ => None, + } + } } /// Listens for incoming WebSocket connections on a TCP port. @@ -246,16 +257,7 @@ impl Listener { // `qmux_versions_for` returns `&[]` (every QMux draft) for ALPNs the spec // doesn't restrict; qmux by default also accepts legacy clients that // only offer a bare wire-format ALPN (today's moq-net clients still do). - // - // Keep-alive matches the dial side (5s ping / 30s deadline, parity with QUIC's - // idle timeout) and is on by default because the accept side is where its - // absence costs something: a peer whose host crashed sends no FIN, and a - // WebSocket has no idle timeout of its own, so the session -- and every - // broadcast announced through it -- would survive until the OS probes the - // socket hours later. `qmux::Server` defaults it OFF, so this must be explicit. - let server = qmux::Server::new() - .with_protocols(alpns.iter().map(|&a| (a, qmux_versions_for(a)))) - .with_keep_alive(qmux::KeepAlive::default()); + let server = qmux::Server::new().with_protocols(alpns.iter().map(|&a| (a, qmux_versions_for(a)))); Ok(Self { listener, server }) } diff --git a/rs/moq-native/tests/reconnect.rs b/rs/moq-native/tests/reconnect.rs new file mode 100644 index 0000000000..2e45ff0774 --- /dev/null +++ b/rs/moq-native/tests/reconnect.rs @@ -0,0 +1,53 @@ +//! What ends the reconnect loop. +//! +//! Dials over plain TCP (`tcp://`), which fails fast and locally: no TLS material, no QUIC +//! handshake, no server. That keeps the assertion about the *budget* rather than about how long a +//! particular backend takes to give up. + +#![cfg(feature = "tcp")] + +use std::time::Duration; + +/// A client whose reconnect loop escalates fast enough to assert on inside a test. +fn client(backoff: moq_native::Backoff) -> moq_native::Client { + let mut config = moq_native::ClientConfig::default(); + config.backoff = backoff; + config.init().expect("failed to init client") +} + +/// A transient failure is retried, escalating, until the budget runs out. The give-up error names +/// the underlying cause so an operator sees why rather than just "timed out". +#[tokio::test] +async fn a_transient_failure_retries_until_the_budget_runs_out() { + let mut backoff = moq_native::Backoff::default(); + backoff.initial = Duration::from_millis(20); + backoff.max = Duration::from_millis(40); + backoff.timeout = Duration::from_millis(200); + + // Nothing listens on port 1, so every attempt is refused: transient as far as this layer knows. + let url = "tcp://127.0.0.1:1".parse().expect("failed to parse url"); + let started = tokio::time::Instant::now(); + let reconnect = client(backoff).reconnect(url); + + let err = tokio::time::timeout(Duration::from_secs(10), reconnect.closed()) + .await + .expect("reconnect loop never gave up") + .expect_err("reconnect loop stopped without an error"); + + assert!( + matches!(err, moq_native::Error::Reconnect(_)), + "stopped with {err} rather than exhausting the budget" + ); + assert_ne!( + err.to_string(), + "reconnect timed out after 200ms", + "give-up error lost the underlying cause" + ); + // The budget is spent on sleeping between attempts, so reaching it takes at least most of it. + // Jitter draws each delay from the top half of its window, hence half rather than the whole. + assert!( + started.elapsed() >= Duration::from_millis(100), + "gave up after {:?} without retrying", + started.elapsed() + ); +} diff --git a/rs/moq-relay/Cargo.toml b/rs/moq-relay/Cargo.toml index eefb8f7ea0..cad8ea69e6 100644 --- a/rs/moq-relay/Cargo.toml +++ b/rs/moq-relay/Cargo.toml @@ -56,6 +56,7 @@ moq-net = { workspace = true } moq-stats = { workspace = true } moq-token = { workspace = true, features = ["tokio"] } qmux = { workspace = true, features = ["ws"], optional = true } +rand = "0.10" reqwest = { version = "0.13", default-features = false, features = ["rustls"] } reqwest-middleware = "0.5" rustls = { version = "0.23", features = ["aws-lc-rs"], default-features = false } diff --git a/rs/moq-relay/src/cluster.rs b/rs/moq-relay/src/cluster.rs index d272312c02..355aaccb6e 100644 --- a/rs/moq-relay/src/cluster.rs +++ b/rs/moq-relay/src/cluster.rs @@ -17,6 +17,7 @@ use tracing::Instrument as _; use url::Url; use crate::{AuthToken, nodes::MESH_PREFIX}; +use rand::RngExt; /// How often the discovery loop scans for stale entries. const SWEEP_INTERVAL: Duration = Duration::from_secs(30); @@ -917,37 +918,46 @@ impl Cluster { url.query_pairs_mut().append_pair("jwt", &token); } - let base_backoff = tokio::time::Duration::from_secs(1); - let max_backoff = tokio::time::Duration::from_secs(300); + // A peer is supervised for the life of the relay, so there is no give-up deadline: one that is + // unreachable for an hour still has to be redialed when it comes back. The ceiling stays low + // so a returning peer is picked up within seconds, and the warn below fires at least that + // often, so a dead peer is loudly broken rather than silently parked. + let base_delay = tokio::time::Duration::from_secs(1); + let max_delay = tokio::time::Duration::from_secs(10); + let mut delay = base_delay; + // Sessions shorter than this are treated as churn: we keep backing off // instead of resetting, otherwise a peer that rejects us instantly would // turn into a tight reconnect loop. let stable_threshold = tokio::time::Duration::from_secs(10); - let mut backoff = base_backoff; - loop { - let started = tokio::time::Instant::now(); - let result = self.run_remote_once(&url, cost).await; - let elapsed = started.elapsed(); - - match result { - Ok(()) if elapsed >= stable_threshold => backoff = base_backoff, - Ok(()) => { - tracing::warn!(?elapsed, "cluster peer session closed cleanly but quickly; backing off"); - backoff = (backoff * 2).min(max_backoff); - } - Err(err) => { - tracing::warn!(%err, "cluster peer error; will retry"); - backoff = (backoff * 2).min(max_backoff); + let attempt = self.run_remote_once(&url, cost).await; + + // A session that came up and lasted is a healthy peer: clear the escalation so a one-off + // drop redials promptly. Keyed on the session's own lifetime, not on how long the call + // took, because a peer that blackholes until the connect timeout would otherwise look + // stable and reset the backoff on every attempt, so the escalation would never happen. + let stable = attempt.connected.is_some_and(|at| at.elapsed() >= stable_threshold); + if stable { + delay = base_delay; + } + + if let Err(err) = attempt.result { + match stable { + true => tracing::warn!(%err, "cluster peer session closed; reconnecting"), + false => tracing::warn!(%err, "cluster peer error; will retry"), } } - tokio::time::sleep(backoff).await; + // Jittered so a restarting cluster doesn't have every peer redial on the same tick. + let wait = delay.mul_f64(0.5 + rand::rng().random::() / 2.0); + delay = (delay * 2).min(max_delay); + tokio::time::sleep(wait).await; } } - async fn run_remote_once(&self, url: &Url, cost: Option) -> anyhow::Result<()> { + async fn run_remote_once(&self, url: &Url, cost: Option) -> Attempt { // Each attempt is its own session, so it gets its own id. Matches the span an // accepted connection runs under, so both directions log the same way. let id = self.next_connection_id(); @@ -956,16 +966,20 @@ impl Cluster { .await } - async fn run_remote_session(&self, id: u64, url: &Url, cost: Option) -> anyhow::Result<()> { + async fn run_remote_session(&self, id: u64, url: &Url, cost: Option) -> Attempt { let mut log_url = url.clone(); log_url.set_query(None); tracing::info!(url = %log_url, "dialing cluster peer"); // Checked at the start of `run`; per-peer tasks inherit that guarantee. - let client = self + let client = match self .client .clone() - .context("internal: cluster peer dial without an attached QUIC client")?; + .context("internal: cluster peer dial without an attached QUIC client") + { + Ok(client) => client, + Err(err) => return Attempt::failed(err), + }; // Cluster dials use their configured stats tier. Cluster peers carry no auth // root, so presence is keyed under the empty root within the cluster tier. @@ -976,13 +990,44 @@ impl Cluster { if let Some(cost) = cost { client = client.with_cost(cost); } - let cs = client + let cs = match client .connect(url.clone()) .await - .context("failed to connect to cluster peer")?; + .context("failed to connect to cluster peer") + { + Ok(cs) => cs, + Err(err) => return Attempt::failed(err), + }; + + let connected = tokio::time::Instant::now(); let _connection = self.nodes.connect_outbound(id, log_url.to_string()); - Err(cs.closed().await.into()) + Attempt { + connected: Some(connected), + result: Err(cs.closed().await.into()), + } + } +} + +/// One peer dial: when its session came up, and how it ended. +/// +/// The instant is what the backoff reset keys on, which is why the dial and the session are timed +/// separately. A session always ends in an `Err` (it hands back its own close reason), so the +/// outcome alone can't tell a healthy peer from a dead one. +struct Attempt { + /// When `connect` handed back a live session, or `None` if the dial never got that far. + connected: Option, + /// How the attempt ended. + result: anyhow::Result<()>, +} + +impl Attempt { + /// An attempt that never established a session. + fn failed(err: anyhow::Error) -> Self { + Self { + connected: None, + result: Err(err), + } } } diff --git a/rs/moq-rtmp/src/server.rs b/rs/moq-rtmp/src/server.rs index 412e77790a..60d5397e22 100644 --- a/rs/moq-rtmp/src/server.rs +++ b/rs/moq-rtmp/src/server.rs @@ -50,6 +50,7 @@ use tokio::net::{TcpListener, TcpStream}; use crate::Result; use crate::flv; +use rand::RngExt; /// Read buffer size for pulling RTMP chunk-stream bytes off the socket. const READ_BUFFER: usize = 16 * 1024; @@ -216,6 +217,12 @@ impl AsyncWrite for Conn { } } +/// Backoff bounds after a failed `accept`. The listener is supervised for the process's lifetime, +/// so there is no give-up budget: the descriptor pressure or firewall rule behind a failed accept +/// clears on its own, and the next connection resets the escalation. +const ACCEPT_RETRY_MIN: Duration = Duration::from_millis(100); +const ACCEPT_RETRY_MAX: Duration = Duration::from_secs(5); + /// An RTMP server that yields each connection's pending request as a [`Request`]. /// /// Build it with [`bind`](Self::bind), optionally enable RTMPS with @@ -233,17 +240,30 @@ pub struct Server { /// In-flight handshakes; each resolves to a ready [`Request`], or `None` if /// the connection closed or errored before issuing a publish or play. pending: FuturesUnordered>>>, + + /// Delay after a failed `accept`, doubling per consecutive failure. Lives on the server rather + /// than inside [`accept`](Self::accept) so the escalation survives across calls, and resets on + /// the next connection that does come in. + accept_delay: Duration, + + /// While set, `accept` stops asking the listener until this instant. In-flight handshakes keep + /// being polled meanwhile: a connection that already got through must not wait out a backoff + /// earned by a different one. + accept_retry: Option, } impl Server { /// Bind an RTMP listener on `addr` (RTMP's well-known port is 1935). pub async fn bind(addr: SocketAddr) -> Result { let listener = TcpListener::bind(addr).await?; + Ok(Self { listener, #[cfg(feature = "tls")] tls: None, pending: FuturesUnordered::new(), + accept_delay: ACCEPT_RETRY_MIN, + accept_retry: None, }) } @@ -270,6 +290,9 @@ impl Server { /// if the listener itself stops (it currently never does). pub async fn accept(&mut self) -> Option> { loop { + // Copied out so the timer arm below doesn't borrow `self` alongside the other two. + let retry = self.accept_retry; + tokio::select! { // A handshake finished: yield its request, or skip a dead connection. Some(maybe) = self.pending.next(), if !self.pending.is_empty() => { @@ -277,9 +300,15 @@ impl Server { return Some(request); } } - // A new TCP connection: start its (TLS +) handshake concurrently. - res = self.listener.accept(), if self.pending.len() < MAX_PENDING_REQUESTS => match res { + // A failed accept's backoff elapsed: start asking the listener again. + () = sleep_until(retry), if retry.is_some() => self.accept_retry = None, + + // A new TCP connection: start its (TLS +) handshake concurrently. Paused while a + // failed accept is backing off, so a persistent error doesn't busy-spin. + res = self.listener.accept(), if retry.is_none() && self.pending.len() < MAX_PENDING_REQUESTS => match res { Ok((stream, peer)) => { + // A connection got through, so whatever the last failure was has cleared. + self.accept_delay = ACCEPT_RETRY_MIN; configure_socket(&stream, peer); #[cfg(feature = "tls")] let tls = self.tls.clone(); @@ -319,10 +348,17 @@ impl Server { } } Err(err) => { - // A failed accept must not take the listener down; back off so a - // persistent error doesn't busy-spin. + // A failed accept must not take the listener down: the usual causes + // (descriptor exhaustion, a connection the firewall dropped mid-handshake) + // are per-connection or clear on their own, and none of them is a reason to + // stop serving. Escalate the wait so a persistent one stops busy-spinning + // instead of retrying ten times a second forever. Recorded as a deadline + // rather than slept on here, so the loop keeps serving in-flight handshakes + // through the pause. tracing::warn!(%err, "failed to accept RTMP connection; continuing"); - tokio::time::sleep(Duration::from_millis(100)).await; + let wait = self.accept_delay.mul_f64(0.5 + rand::rng().random::() / 2.0); + self.accept_delay = (self.accept_delay * 2).min(ACCEPT_RETRY_MAX); + self.accept_retry = Some(tokio::time::Instant::now() + wait); } }, } @@ -330,6 +366,17 @@ impl Server { } } +/// Sleep until `at`, or park forever when there is nothing to wait for. +/// +/// The `select!` arm that uses this is guarded on `at` being set; the pending branch keeps the arm +/// well-formed rather than leaving the macro with a `None` to unwrap. +async fn sleep_until(at: Option) { + match at { + Some(at) => tokio::time::sleep_until(at).await, + None => std::future::pending().await, + } +} + /// Tune an RTMP TCP socket (accepted by the server or dialed by the client): /// Nagle off for latency, keepalive on so a dead peer is reaped rather than /// pinning a broadcast forever.