diff --git a/.changeset/cloud-api-failover.md b/.changeset/cloud-api-failover.md new file mode 100644 index 00000000..b6254514 --- /dev/null +++ b/.changeset/cloud-api-failover.md @@ -0,0 +1,5 @@ +--- +'livekit-server-sdk': patch +--- + +Retry LiveKit Cloud API (`cloud-api.livekit.io`) requests on transport errors instead of failing after a single attempt, add full jitter to the failover backoff, and retry a 503 only when it carries Retry-After, waiting the indicated delay. diff --git a/packages/livekit-server-sdk/src/TwirpRPC.test.ts b/packages/livekit-server-sdk/src/TwirpRPC.test.ts index f964465d..e4735428 100644 --- a/packages/livekit-server-sdk/src/TwirpRPC.test.ts +++ b/packages/livekit-server-sdk/src/TwirpRPC.test.ts @@ -95,7 +95,7 @@ describe('request id', () => { } as unknown as Response; } attempt += 1; - return attempt < 3 ? errorResponse(503) : okResponse(); + return attempt < 3 ? errorResponse(502) : okResponse(); }); const rpc = new TwirpRpc('https://primary.retryid.livekit.cloud', 'livekit', { @@ -112,3 +112,111 @@ describe('request id', () => { expect(new Set(ids).size).toBe(1); }); }); + +describe('failover without a fallback origin', () => { + afterEach(() => { + vi.restoreAllMocks(); + }); + + // No /settings/regions, so no fallback origin ever exists. + const host = 'https://cloud-api.example.com'; + + const okResponse = () => + ({ ok: true, status: 200, json: async () => ({}) }) as unknown as Response; + + const errorResponse = (status: number, retryAfter: string | null = null) => + ({ + ok: false, + status, + statusText: 'Bad Gateway', + headers: { get: (name: string) => (name === 'retry-after' ? retryAfter : null) }, + text: async () => 'bad gateway', + }) as unknown as Response; + + const isDiscovery = (input: unknown) => `${input}`.endsWith('/settings/regions'); + + it('without a fallback origin, a transport error retries the same host', async () => { + let attempt = 0; + const fetchSpy = vi.spyOn(globalThis, 'fetch').mockImplementation(async (input) => { + if (isDiscovery(input)) { + return errorResponse(404); + } + attempt += 1; + if (attempt === 1) { + throw new Error('read: connection reset by peer'); + } + return okResponse(); + }); + + const rpc = new TwirpRpc(host, 'livekit', { failoverForce: true, failoverBackoffMs: 0 }); + await expect(rpc.request('RoomService', 'CreateRoom', {}, {})).resolves.toEqual({}); + + const attempts = fetchSpy.mock.calls.filter(([input]) => !isDiscovery(input)); + expect(attempts).toHaveLength(2); + expect(attempts.map(([input]) => new URL(`${input}`).host)).toEqual([ + 'cloud-api.example.com', + 'cloud-api.example.com', + ]); + }); + + it('a Cloud API host retries without consulting region discovery', async () => { + let attempt = 0; + const fetchSpy = vi.spyOn(globalThis, 'fetch').mockImplementation(async (input) => { + if (isDiscovery(input)) { + return errorResponse(404); + } + attempt += 1; + if (attempt === 1) { + throw new Error('read: connection reset by peer'); + } + return okResponse(); + }); + + const rpc = new TwirpRpc('https://cloud-api.livekit.io', 'livekit', { failoverBackoffMs: 0 }); + await expect(rpc.request('RoomService', 'CreateRoom', {}, {})).resolves.toEqual({}); + + expect(fetchSpy.mock.calls.filter(([input]) => isDiscovery(input))).toHaveLength(0); + expect(fetchSpy.mock.calls.filter(([input]) => !isDiscovery(input))).toHaveLength(2); + }); + + it('without a fallback origin, a 5xx retries the same host', async () => { + let attempt = 0; + const fetchSpy = vi.spyOn(globalThis, 'fetch').mockImplementation(async (input) => { + if (isDiscovery(input)) { + return errorResponse(404); + } + attempt += 1; + return attempt === 1 ? errorResponse(502) : okResponse(); + }); + + const rpc = new TwirpRpc(host, 'livekit', { failoverForce: true, failoverBackoffMs: 0 }); + await expect(rpc.request('RoomService', 'CreateRoom', {}, {})).resolves.toEqual({}); + + expect(fetchSpy.mock.calls.filter(([input]) => !isDiscovery(input))).toHaveLength(2); + }); + + it('a 503 without Retry-After is not retried', async () => { + const fetchSpy = vi.spyOn(globalThis, 'fetch').mockResolvedValue(errorResponse(503)); + + const rpc = new TwirpRpc(host, 'livekit', { failoverForce: true, failoverBackoffMs: 0 }); + await expect(rpc.request('RoomService', 'CreateRoom', {}, {})).rejects.toThrow(); + + expect(fetchSpy.mock.calls.filter(([input]) => !isDiscovery(input))).toHaveLength(1); + }); + + it('a 503 with Retry-After is retried', async () => { + let attempt = 0; + const fetchSpy = vi.spyOn(globalThis, 'fetch').mockImplementation(async (input) => { + if (isDiscovery(input)) { + return errorResponse(404); + } + attempt += 1; + return attempt === 1 ? errorResponse(503, '0') : okResponse(); + }); + + const rpc = new TwirpRpc(host, 'livekit', { failoverForce: true, failoverBackoffMs: 0 }); + await expect(rpc.request('RoomService', 'CreateRoom', {}, {})).resolves.toEqual({}); + + expect(fetchSpy.mock.calls.filter(([input]) => !isDiscovery(input))).toHaveLength(2); + }); +}); diff --git a/packages/livekit-server-sdk/src/TwirpRPC.ts b/packages/livekit-server-sdk/src/TwirpRPC.ts index f93a9d04..41422104 100644 --- a/packages/livekit-server-sdk/src/TwirpRPC.ts +++ b/packages/livekit-server-sdk/src/TwirpRPC.ts @@ -5,8 +5,11 @@ import type { JsonValue } from '@bufbuild/protobuf'; import { randomUUID } from './crypto/uuid.js'; import { FAILOVER_BACKOFF_BASE_MS, + backoffMs, failoverAttempts, hostKey, + isCloudApi, + parseRetryAfter, pickNext, regionOrigins, sleep, @@ -166,9 +169,10 @@ export class TwirpRpc { /** * Issues a Twirp request, failing over to alternative regions on retryable - * errors. On any transport error or HTTP 5xx it discovers regions via + * errors. On any transport error or HTTP 5xx (a 503 only with Retry-After, + * which then sets the delay) it discovers regions via * /settings/regions and replays the request — body and headers intact — - * against the next untried region, with exponential backoff. A 4xx is + * against the next untried region, with jittered exponential backoff. A 4xx is * returned immediately. */ async request( @@ -196,7 +200,8 @@ export class TwirpRpc { timeout, ); const attempted = new Set([hostKey(origin)]); - let regions: string[] | undefined; + // A Cloud API host has a single origin; region discovery is never consulted. + let regions: string[] | undefined = isCloudApi(origin.hostname) ? [] : undefined; let current = this.host; for (let attempt = 0; attempt < maxAttempts; attempt += 1) { @@ -223,15 +228,24 @@ export class TwirpRpc { return (await response.json()) as Record; } - // Only retryable failures (a transport error or HTTP 5xx) continue; - // a 4xx is terminal. - const retryable = transportError !== undefined || (!!response && response.status >= 500); + const retryAfter = response + ? parseRetryAfter(response.headers.get('retry-after')) + : undefined; + // Only retryable failures (a transport error or HTTP 5xx) continue; a 4xx + // is terminal. A 503 is the server shedding load, so it is retried only + // when Retry-After says when. + const retryable = + transportError !== undefined || + (!!response && + response.status >= 500 && + (response.status !== 503 || retryAfter !== undefined)); let next: string | undefined; if (retryable && !isLast) { if (!regions) { regions = await regionOrigins(origin, headers); } - next = pickNext(regions, attempted); + // With no fallback origin, a retryable failure is retried against the same host. + next = pickNext(regions, attempted) ?? current; } if (!retryable || next === undefined) { @@ -245,7 +259,7 @@ export class TwirpRpc { console.warn( `livekit API request to ${new URL(current).host} failed (${reason}), retrying with fallback url ${next}`, ); - await sleep(this.failoverBackoffMs * 2 ** attempt); + await sleep(retryAfter ?? backoffMs(attempt, this.failoverBackoffMs)); attempted.add(hostKey(new URL(next))); current = next; } diff --git a/packages/livekit-server-sdk/src/failover.test.ts b/packages/livekit-server-sdk/src/failover.test.ts index c528ab63..2dd06374 100644 --- a/packages/livekit-server-sdk/src/failover.test.ts +++ b/packages/livekit-server-sdk/src/failover.test.ts @@ -5,9 +5,11 @@ import { afterEach, describe, expect, it, vi } from 'vitest'; import { FAILOVER_MAX_ATTEMPTS, MIN_FAILOVER_TIMEOUT_SECONDS, + backoffMs, failoverAttempts, hostKey, parseMaxAge, + parseRetryAfter, pickNext, regionOrigins, sleep, @@ -19,8 +21,15 @@ describe('failoverAttempts', () => { expect(failoverAttempts(true, 'myproject.region.livekit.cloud')).toBe(FAILOVER_MAX_ATTEMPTS); }); + it('fails over for the LiveKit Cloud API hosts', () => { + expect(failoverAttempts(true, 'cloud-api.livekit.io')).toBe(FAILOVER_MAX_ATTEMPTS); + expect(failoverAttempts(true, 'cloud-api.staging.livekit.io')).toBe(FAILOVER_MAX_ATTEMPTS); + expect(failoverAttempts(true, 'CLOUD-API.LIVEKIT.IO')).toBe(FAILOVER_MAX_ATTEMPTS); + }); + it('does not fail over for non-cloud hosts', () => { expect(failoverAttempts(true, 'myproject.livekit.io')).toBe(1); + expect(failoverAttempts(true, 'cloud-api.example.com')).toBe(1); expect(failoverAttempts(true, 'example.com')).toBe(1); expect(failoverAttempts(true, '127.0.0.1')).toBe(1); expect(failoverAttempts(true, 'notlivekit.cloud')).toBe(1); @@ -72,6 +81,44 @@ describe('pickNext', () => { }); }); +describe('backoffMs', () => { + afterEach(() => { + vi.restoreAllMocks(); + }); + + it('draws uniformly from [0, base * 2^attempt)', () => { + vi.spyOn(Math, 'random').mockReturnValue(0); + expect(backoffMs(2, 200)).toBe(0); + vi.spyOn(Math, 'random').mockReturnValue(0.5); + expect(backoffMs(0, 200)).toBe(100); + expect(backoffMs(1, 200)).toBe(200); + expect(backoffMs(2, 200)).toBe(400); + }); +}); + +describe('parseRetryAfter', () => { + it('returns undefined when the header is absent or invalid', () => { + expect(parseRetryAfter(null)).toBeUndefined(); + expect(parseRetryAfter('')).toBeUndefined(); + expect(parseRetryAfter('soon')).toBeUndefined(); + }); + + it('parses delay-seconds to milliseconds', () => { + expect(parseRetryAfter('0')).toBe(0); + expect(parseRetryAfter(' 2 ')).toBe(2000); + }); + + it('parses an HTTP-date relative to now, clamped at zero', () => { + vi.useFakeTimers({ now: Date.parse('Wed, 21 Oct 2026 07:28:00 GMT') }); + try { + expect(parseRetryAfter('Wed, 21 Oct 2026 07:28:03 GMT')).toBe(3000); + expect(parseRetryAfter('Wed, 21 Oct 2026 07:27:00 GMT')).toBe(0); + } finally { + vi.useRealTimers(); + } + }); +}); + describe('sleep', () => { it('resolves immediately for non-positive durations', async () => { await expect(sleep(0)).resolves.toBeUndefined(); diff --git a/packages/livekit-server-sdk/src/failover.ts b/packages/livekit-server-sdk/src/failover.ts index a8dbe827..c829b244 100644 --- a/packages/livekit-server-sdk/src/failover.ts +++ b/packages/livekit-server-sdk/src/failover.ts @@ -4,9 +4,10 @@ // Region failover for the Twirp API clients. // -// On a retryable failure (any transport error or HTTP 5xx) the client discovers +// On a retryable failure (any transport error, or an HTTP 5xx other than a 503 +// without Retry-After) the client discovers // alternative LiveKit Cloud regions via /settings/regions and replays the -// request against the next region, with exponential backoff. 4xx responses are +// request against the next region, with jittered exponential backoff. 4xx responses are // returned immediately. // Total attempts (the original request + fallback regions) and the base retry @@ -21,8 +22,8 @@ export const MIN_FAILOVER_TIMEOUT_SECONDS = 5; /** * Total request attempts for a host; 1 means no failover. Failover only engages - * when enabled, the host is a LiveKit Cloud domain, and the request timeout is - * long enough to retry. `force` bypasses the cloud-host check (test-only). + * when enabled, the host is a LiveKit Cloud project or Cloud API domain, and the + * request timeout is long enough to retry. `force` bypasses the cloud-host check (test-only). */ export function failoverAttempts( enabled: boolean, @@ -30,7 +31,7 @@ export function failoverAttempts( force = false, timeoutSeconds = 0, ): number { - if (!enabled || !(force || isCloud(hostname))) { + if (!enabled || !(force || isCloud(hostname) || isCloudApi(hostname))) { return 1; } if (timeoutSeconds > 0 && timeoutSeconds < MIN_FAILOVER_TIMEOUT_SECONDS) { @@ -44,6 +45,12 @@ function isCloud(hostname: string): boolean { return hostname.endsWith('.livekit.cloud'); } +// The LiveKit Cloud API hosts: cloud-api.livekit.io and cloud-api..livekit.io. +export function isCloudApi(hostname: string): boolean { + const host = hostname.toLowerCase(); + return host.startsWith('cloud-api.') && host.endsWith('.livekit.io'); +} + /** Normalizes a region URL to an http(s) scheme (ws -> http, wss -> https). */ function toHttp(url: string): string { return url.startsWith('ws') ? `http${url.slice(2)}` : url; @@ -68,6 +75,25 @@ export function pickNext(regionOrigins: string[], attempted: Set): strin return undefined; } +// Full jitter: a uniform delay in [0, base * 2^attempt) so clients that failed +// together don't retry in lockstep. +export function backoffMs(attempt: number, baseMs: number): number { + return Math.random() * baseMs * 2 ** attempt; +} + +/** Parses a Retry-After header (delay-seconds or an HTTP-date) to milliseconds. */ +export function parseRetryAfter(header: string | null): number | undefined { + const value = header?.trim(); + if (!value) { + return undefined; + } + if (/^\d+$/.test(value)) { + return Number(value) * 1000; + } + const date = Date.parse(value); + return Number.isNaN(date) ? undefined : Math.max(0, date - Date.now()); +} + export function sleep(ms: number): Promise { return ms > 0 ? new Promise((resolve) => setTimeout(resolve, ms)) : Promise.resolve(); } diff --git a/packages/livekit-server-sdk/test/api/failover.test.ts b/packages/livekit-server-sdk/test/api/failover.test.ts index 4edc1971..0cea52a9 100644 --- a/packages/livekit-server-sdk/test/api/failover.test.ts +++ b/packages/livekit-server-sdk/test/api/failover.test.ts @@ -27,7 +27,8 @@ const call = ( { authorization: 'Bearer test-token', // These tests exercise failover, not authz; skip the mock's permission check. - 'X-Lk-Mock': JSON.stringify({ skipAuth: true, ...mock }), + // The mock fails with a 503 by default, which is retried only with Retry-After. + 'X-Lk-Mock': JSON.stringify({ skipAuth: true, failStatus: 502, ...mock }), }, ); };