diff --git a/.changeset/retry-after-rate-limits.md b/.changeset/retry-after-rate-limits.md new file mode 100644 index 0000000000..db91c1c6bb --- /dev/null +++ b/.changeset/retry-after-rate-limits.md @@ -0,0 +1,5 @@ +--- +"e2b": patch +--- + +Retry control-plane HTTP requests up to three times after `429` responses using the server's delta-seconds `Retry-After` delay. Retries can be configured or disabled with `retries`, and stop when waiting would exhaust the request timeout. Envd requests, including filesystem operations, and volume-content requests are not retried. diff --git a/packages/js-sdk/src/api/index.ts b/packages/js-sdk/src/api/index.ts index a89ed09510..de6289f587 100644 --- a/packages/js-sdk/src/api/index.ts +++ b/packages/js-sdk/src/api/index.ts @@ -11,6 +11,7 @@ import { SandboxError, } from '../errors' import { createApiLogger } from '../logs' +import { withRateLimitRetry } from '../retry' /** * Map an API error code and message to the matching error class — the same @@ -105,7 +106,11 @@ class ApiClient { this.api = createClient({ baseUrl: config.apiUrl, - fetch: createApiFetch(config.proxy), + fetch: withRateLimitRetry( + createApiFetch(config.proxy), + config.retries, + config.requestTimeoutMs + ), // In HTTP 1.1, all connections are considered persistent unless declared otherwise // keepalive: true, headers: { diff --git a/packages/js-sdk/src/connectionConfig.ts b/packages/js-sdk/src/connectionConfig.ts index c243337bf8..a9ef7fdf84 100644 --- a/packages/js-sdk/src/connectionConfig.ts +++ b/packages/js-sdk/src/connectionConfig.ts @@ -1,11 +1,13 @@ import { Logger } from './logs' import { getEnvVar, version } from './api/metadata' import { runtime } from './utils' +import { resolveRetries } from './retry' // Remove once all deployments support sandbox subdomains const supportedDomains = ['e2b.app', 'e2b.dev', 'e2b.pro', 'e2b-staging.dev'] export const REQUEST_TIMEOUT_MS = 60_000 // 60 seconds +export const DEFAULT_RETRIES = 3 export const DEFAULT_SANDBOX_TIMEOUT_MS = 300_000 // 300 seconds export const KEEPALIVE_PING_INTERVAL_SEC = 50 // 50 seconds @@ -58,6 +60,15 @@ export interface ConnectionOpts { * @default 60_000 // 60 seconds */ requestTimeoutMs?: number + /** + * Number of control-plane API retries after a 429 response with a valid, + * non-negative integer delta-seconds `Retry-After` header. HTTP-date and + * malformed values are not retried. + * Retry waits use a 60-second total limit when request timeouts are disabled. + * + * @default 3 + */ + retries?: number /** * Logger to use for logging messages. It can accept any object that implements `Logger` interface—for example, {@link console}. */ @@ -408,6 +419,7 @@ export class ConnectionConfig { readonly logger?: Logger readonly requestTimeoutMs: number + readonly retries: number readonly apiKey?: string /** @@ -435,6 +447,7 @@ export class ConnectionConfig { this.debug = opts?.debug ?? ConnectionConfig.debug this.domain = opts?.domain || ConnectionConfig.domain this.requestTimeoutMs = opts?.requestTimeoutMs ?? REQUEST_TIMEOUT_MS + this.retries = resolveRetries(opts?.retries ?? DEFAULT_RETRIES) this.logger = opts?.logger this.requestSource = ConnectionConfig.getRequestSource() this.headers = { ...(opts?.headers ?? {}), ...(opts?.apiHeaders ?? {}) } diff --git a/packages/js-sdk/src/retry.ts b/packages/js-sdk/src/retry.ts new file mode 100644 index 0000000000..7ad186ba9c --- /dev/null +++ b/packages/js-sdk/src/retry.ts @@ -0,0 +1,99 @@ +import { InvalidArgumentError } from './errors' +import { isReadableStreamLike } from './is' + +const MAX_RETRY_AFTER_SECONDS = 2_147_483 +const MAX_RETRY_WAIT_WITHOUT_TIMEOUT_MS = 60_000 + +export function resolveRetries(retries: number): number { + if (!Number.isInteger(retries) || retries < 0) { + throw new InvalidArgumentError( + `Invalid retries=${retries}: expected a non-negative integer.` + ) + } + return retries +} + +export function parseRetryAfter( + value: string | null | undefined +): number | undefined { + if (!value) return undefined + + const trimmed = value.trim() + if (!/^\d+$/.test(trimmed)) return undefined + + const delay = Number(trimmed) + return Number.isSafeInteger(delay) && delay <= MAX_RETRY_AFTER_SECONDS + ? delay + : undefined +} + +type RetryDependencies = { + monotonic?: () => number + sleep?: (delayMs: number, signal: AbortSignal) => Promise +} + +function wait(delayMs: number, signal: AbortSignal): Promise { + if (signal.aborted) return Promise.reject(signal.reason) + + return new Promise((resolve, reject) => { + const onAbort = () => { + clearTimeout(timer) + reject(signal.reason) + } + const timer = setTimeout(() => { + signal.removeEventListener('abort', onAbort) + resolve() + }, delayMs) + signal.addEventListener('abort', onAbort, { once: true }) + }) +} + +/** Retry replayable requests after a 429 carrying `Retry-After`. */ +export function withRateLimitRetry( + fetchImpl: typeof fetch, + retries: number, + requestTimeoutMs: number, + dependencies: RetryDependencies = {} +): typeof fetch { + const monotonic = dependencies.monotonic ?? (() => performance.now()) + const sleep = dependencies.sleep ?? wait + + return (async (input, init) => { + // Streaming bodies would be consumed by the first attempt and cannot be + // replayed without buffering them, so they get a single attempt. + if (retries === 0 || isReadableStreamLike(init?.body)) { + return fetchImpl(input, init) + } + + // Replaying a Request-form input via `clone()` is safe because the only + // producer of those is openapi-fetch, which serializes every body to a + // string before constructing the Request — cloning never tees a live + // stream. + const request = + input instanceof Request && init === undefined + ? input + : new Request(input, init) + const deadline = + monotonic() + (requestTimeoutMs || MAX_RETRY_WAIT_WITHOUT_TIMEOUT_MS) + + for (let attempt = 0; ; attempt++) { + const response = await fetchImpl( + attempt === retries ? request : request.clone() + ) + const retryAfter = parseRetryAfter(response.headers.get('Retry-After')) + const delayMs = retryAfter === undefined ? undefined : retryAfter * 1000 + + if ( + response.status !== 429 || + delayMs === undefined || + attempt === retries || + monotonic() + delayMs >= deadline + ) { + return response + } + + await response.body?.cancel().catch(() => {}) + await sleep(delayMs, request.signal) + } + }) as typeof fetch +} diff --git a/packages/js-sdk/src/sandbox/sandboxApi.ts b/packages/js-sdk/src/sandbox/sandboxApi.ts index 0b6f215a5a..af7b1b15ae 100644 --- a/packages/js-sdk/src/sandbox/sandboxApi.ts +++ b/packages/js-sdk/src/sandbox/sandboxApi.ts @@ -514,6 +514,7 @@ export interface SandboxApiOpts extends Partial< | 'debug' | 'domain' | 'requestTimeoutMs' + | 'retries' | 'signal' > > {} diff --git a/packages/js-sdk/tests/client.test.ts b/packages/js-sdk/tests/client.test.ts index bb7f47280c..b277fb8969 100644 --- a/packages/js-sdk/tests/client.test.ts +++ b/packages/js-sdk/tests/client.test.ts @@ -167,6 +167,53 @@ test('per-call options take precedence over the client config', async () => { assert.equal(lastRequest().apiKey, API_KEY_B) }) +test('client retries rate-limited control-plane requests', async () => { + let attempts = 0 + server.use( + http.get(/\/v2\/sandboxes/, () => { + attempts++ + if (attempts === 1) { + return new HttpResponse(null, { + status: 429, + headers: { 'Retry-After': '0' }, + }) + } + return HttpResponse.json([]) + }) + ) + const client = new E2B({ + apiKey: API_KEY_A, + domain: DOMAIN_A, + }) + + await client.Sandbox.list().nextItems() + + assert.equal(attempts, 2) +}) + +test('client replays serialized control-plane JSON after a rate limit', async () => { + const bodies: unknown[] = [] + server.use( + http.post(/\/sandboxes$/, async ({ request }) => { + bodies.push(await request.json()) + if (bodies.length === 1) { + return new HttpResponse(null, { + status: 429, + headers: { 'Retry-After': '0' }, + }) + } + return HttpResponse.json(sandboxResponse) + }) + ) + const client = new E2B({ apiKey: API_KEY_A, domain: DOMAIN_A }) + + await client.Sandbox.create() + + expect(bodies).toHaveLength(2) + expect(bodies[0]).toMatchObject({ templateID: 'base' }) + expect(bodies[1]).toEqual(bodies[0]) +}) + test('client.Sandbox can be rebound to a variable', async () => { const client = new E2B({ apiKey: API_KEY_A, domain: DOMAIN_A }) const S = client.Sandbox diff --git a/packages/js-sdk/tests/connectionConfig.test.ts b/packages/js-sdk/tests/connectionConfig.test.ts index a81de70ace..2a9a08152f 100644 --- a/packages/js-sdk/tests/connectionConfig.test.ts +++ b/packages/js-sdk/tests/connectionConfig.test.ts @@ -1,6 +1,7 @@ import { assert, test, beforeEach, afterEach } from 'vitest' import { ConnectionConfig, + DEFAULT_RETRIES, setupRequestController, wrapStreamWithConnectionCleanup, } from '../src/connectionConfig' @@ -42,6 +43,12 @@ test('api_url defaults correctly', () => { assert.equal(config.apiUrl, 'https://api.e2b.app') }) +test('retries default to three and accept a non-negative integer', () => { + assert.equal(new ConnectionConfig().retries, DEFAULT_RETRIES) + assert.equal(new ConnectionConfig({ retries: 2 }).retries, 2) + assert.throws(() => new ConnectionConfig({ retries: -1 })) +}) + test('api_url in args', () => { const config = new ConnectionConfig({ apiUrl: 'http://localhost:8080' }) assert.equal(config.apiUrl, 'http://localhost:8080') diff --git a/packages/js-sdk/tests/retry.test.ts b/packages/js-sdk/tests/retry.test.ts new file mode 100644 index 0000000000..dfb9347821 --- /dev/null +++ b/packages/js-sdk/tests/retry.test.ts @@ -0,0 +1,250 @@ +import { describe, expect, test, vi } from 'vitest' + +import { + parseRetryAfter, + resolveRetries, + withRateLimitRetry, +} from '../src/retry' +import { EnvdApiClient } from '../src/envd/api' +import { InvalidArgumentError } from '../src/errors' + +describe('resolveRetries', () => { + test('accepts non-negative integers', () => { + expect(resolveRetries(0)).toBe(0) + expect(resolveRetries(3)).toBe(3) + }) + + test.each([-1, 1.5, Number.NaN])('rejects %s', (retries) => { + expect(() => resolveRetries(retries)).toThrow(InvalidArgumentError) + expect(() => resolveRetries(retries)).toThrow( + 'expected a non-negative integer' + ) + }) +}) + +describe('parseRetryAfter', () => { + test.each([ + ['0', 0], + [' 12 ', 12], + ['2147483', 2_147_483], + [null, undefined], + ['', undefined], + ['-1', undefined], + ['1.5', undefined], + ['Wed, 21 Oct 2015 07:28:00 GMT', undefined], + ['2147484', undefined], + ])('parses %j as %s', (value, expected) => { + expect(parseRetryAfter(value)).toBe(expected) + }) +}) + +test('retries a buffered request and cancels the intermediate response', async () => { + const bodies: string[] = [] + const rateLimited = new Response('rate limited', { + status: 429, + headers: { 'Retry-After': '2' }, + }) + const fetchImpl = vi.fn(async (input: RequestInfo | URL) => { + bodies.push(await (input as Request).text()) + return bodies.length === 1 ? rateLimited : new Response('ok') + }) as typeof fetch + const sleep = vi.fn(async () => {}) + const fetchWithRetry = withRateLimitRetry(fetchImpl, 1, 10_000, { + monotonic: () => 0, + sleep, + }) + + const response = await fetchWithRetry('https://api.e2b.test/resource', { + method: 'POST', + body: 'request body', + }) + + expect(await response.text()).toBe('ok') + expect(bodies).toEqual(['request body', 'request body']) + expect(sleep).toHaveBeenCalledWith(2_000, expect.any(AbortSignal)) + expect(rateLimited.bodyUsed).toBe(true) +}) + +test.each(['', '{"templateID":"base"}'])( + 'replays a serialized API body %j across attempts', + async (body) => { + const controller = new AbortController() + const request = new Request('https://api.e2b.test/resource', { + method: 'POST', + body, + headers: { 'Content-Type': 'application/json', 'X-API-KEY': 'test-key' }, + signal: controller.signal, + redirect: 'manual', + credentials: 'include', + }) + const attempts: Request[] = [] + const bodies: string[] = [] + const fetchImpl = vi.fn(async (input: RequestInfo | URL) => { + const attempt = input as Request + attempts.push(attempt) + bodies.push(await attempt.text()) + return new Response(null, { + status: attempts.length < 3 ? 429 : 200, + headers: { 'Retry-After': '0' }, + }) + }) as typeof fetch + + const response = await withRateLimitRetry(fetchImpl, 2, 10_000)(request) + + expect(response.status).toBe(200) + expect(bodies).toEqual([body, body, body]) + expect(new Set(attempts).size).toBe(3) + controller.abort() + for (const attempt of attempts) { + expect(attempt.url).toBe(request.url) + expect(attempt.method).toBe('POST') + expect([...attempt.headers]).toEqual([...request.headers]) + expect(attempt.redirect).toBe('manual') + // Deno and Cloudflare do not expose Request.credentials. + expect(attempt.credentials).toBe(request.credentials) + expect(attempt.signal.aborted).toBe(true) + } + } +) + +test('returns the final 429 after exhausting retries', async () => { + const fetchImpl = vi.fn(async () => { + return new Response(null, { + status: 429, + headers: { 'Retry-After': '0' }, + }) + }) as typeof fetch + const fetchWithRetry = withRateLimitRetry(fetchImpl, 2, 10_000, { + monotonic: () => 0, + sleep: async () => {}, + }) + + const response = await fetchWithRetry('https://api.e2b.test/resource') + + expect(response.status).toBe(429) + expect(fetchImpl).toHaveBeenCalledTimes(3) +}) + +test('propagates 429 when Retry-After exceeds the request timeout', async () => { + const fetchImpl = vi.fn(async () => { + return new Response(null, { + status: 429, + headers: { 'Retry-After': '2' }, + }) + }) as typeof fetch + const sleep = vi.fn(async () => {}) + const fetchWithRetry = withRateLimitRetry(fetchImpl, 1, 1_000, { + monotonic: () => 0, + sleep, + }) + + const response = await fetchWithRetry('https://api.e2b.test/resource') + + expect(response.status).toBe(429) + expect(fetchImpl).toHaveBeenCalledOnce() + expect(sleep).not.toHaveBeenCalled() +}) + +test('bounds retry waits when the request timeout is disabled', async () => { + const fetchImpl = vi.fn( + async () => + new Response(null, { + status: 429, + headers: { 'Retry-After': '60' }, + }) + ) as typeof fetch + const sleep = vi.fn(async () => {}) + const fetchWithRetry = withRateLimitRetry(fetchImpl, 3, 0, { + monotonic: () => 0, + sleep, + }) + + const response = await fetchWithRetry('https://api.e2b.test/resource') + + expect(response.status).toBe(429) + expect(fetchImpl).toHaveBeenCalledOnce() + expect(sleep).not.toHaveBeenCalled() +}) + +test('propagates 429 without Retry-After as is', async () => { + const rateLimited = new Response('rate limited', { status: 429 }) + const fetchImpl = vi.fn(async () => rateLimited) as typeof fetch + const sleep = vi.fn(async () => {}) + const fetchWithRetry = withRateLimitRetry(fetchImpl, 2, 10_000, { + monotonic: () => 0, + sleep, + }) + + const response = await fetchWithRetry('https://api.e2b.test/resource') + + expect(response).toBe(rateLimited) + expect(response.bodyUsed).toBe(false) + expect(fetchImpl).toHaveBeenCalledOnce() + expect(sleep).not.toHaveBeenCalled() +}) + +test('aborting during Retry-After sleep rejects promptly', async () => { + const rateLimited = new Response('rate limited', { + status: 429, + headers: { 'Retry-After': '1' }, + }) + const fetchImpl = vi.fn(async () => rateLimited) as typeof fetch + const fetchWithRetry = withRateLimitRetry(fetchImpl, 1, 10_000) + const controller = new AbortController() + const reason = new Error('cancelled') + + const response = fetchWithRetry('https://api.e2b.test/resource', { + signal: controller.signal, + }) + await new Promise((resolve) => setTimeout(resolve, 0)) + controller.abort(reason) + + await expect(response).rejects.toBe(reason) + expect(rateLimited.bodyUsed).toBe(true) + expect(fetchImpl).toHaveBeenCalledOnce() +}) + +test.each([0, 3])( + 'retries=%s passes a streaming request through unchanged', + async (retries) => { + const fetchImpl = vi.fn(async () => new Response('ok')) as typeof fetch + const fetchWithRetry = withRateLimitRetry(fetchImpl, retries, 10_000) + const body = new ReadableStream() + const init = { + method: 'POST', + body, + duplex: 'half' as const, + } + + await fetchWithRetry('https://api.e2b.test/resource', init) + + expect(fetchImpl).toHaveBeenCalledWith( + 'https://api.e2b.test/resource', + init + ) + expect(body.locked).toBe(false) + } +) + +test('envd clients do not retry rate-limited requests', async () => { + const fetchImpl = vi.fn( + async () => + new Response(null, { + status: 429, + headers: { 'Retry-After': '0' }, + }) + ) as typeof fetch + const client = new EnvdApiClient( + { + apiUrl: 'https://envd.e2b.test', + logger: undefined, + fetch: fetchImpl, + }, + { version: '0.0.0' } + ) + + const response = await client.api.GET('/health') + + expect(response.response.status).toBe(429) + expect(fetchImpl).toHaveBeenCalledOnce() +}) diff --git a/packages/js-sdk/tests/sandbox/configPropagation.test.ts b/packages/js-sdk/tests/sandbox/configPropagation.test.ts index eef1314bb9..98ec5017f0 100644 --- a/packages/js-sdk/tests/sandbox/configPropagation.test.ts +++ b/packages/js-sdk/tests/sandbox/configPropagation.test.ts @@ -10,6 +10,7 @@ const baseConfig = { apiKey: TEST_API_KEY, domain: 'base.e2b.dev', requestTimeoutMs: 1111, + retries: 2, debug: false, apiHeaders: { 'X-Test': 'base' }, } @@ -49,6 +50,7 @@ describe('Sandbox API config propagation', () => { assert.equal(opts?.apiKey, baseConfig.apiKey) assert.equal(opts?.domain, baseConfig.domain) assert.equal(opts?.requestTimeoutMs, baseConfig.requestTimeoutMs) + assert.equal(opts?.retries, baseConfig.retries) assert.equal(opts?.debug, baseConfig.debug) assert.equal(opts?.headers?.['X-Test'], baseConfig.apiHeaders['X-Test']) }) @@ -68,12 +70,14 @@ describe('Sandbox API config propagation', () => { await sandbox.pause({ domain: 'override.e2b.dev', requestTimeoutMs: 9999, + retries: 0, }) const opts = pauseSpy.mock.calls[0][1] assert.equal(opts?.apiKey, baseConfig.apiKey) assert.equal(opts?.domain, 'override.e2b.dev') assert.equal(opts?.requestTimeoutMs, 9999) + assert.equal(opts?.retries, 0) assert.equal(opts?.debug, baseConfig.debug) }) diff --git a/packages/js-sdk/tests/volume/file.test.ts b/packages/js-sdk/tests/volume/file.test.ts index 3b350f5f7e..b35efa8675 100644 --- a/packages/js-sdk/tests/volume/file.test.ts +++ b/packages/js-sdk/tests/volume/file.test.ts @@ -1,8 +1,14 @@ import { afterAll, beforeAll, beforeEach, describe, expect } from 'vitest' import { setupServer } from 'msw/node' - -import { VolumeError, VolumeFileType, VolumePathNotFoundError } from '../../src' -import { volumeTest } from '../setup' +import { http, HttpResponse } from 'msw' + +import { + RateLimitError, + VolumeError, + VolumeFileType, + VolumePathNotFoundError, +} from '../../src' +import { apiUrl, volumeTest } from '../setup' import { createMockVolumeApi } from './mockVolumeContent' const server = setupServer() @@ -15,6 +21,30 @@ beforeEach(() => server.resetHandlers(...createMockVolumeApi())) describe('Volume File Operations', () => { describe('writeFile and readFile', () => { + volumeTest( + 'does not retry a buffered upload after rate limiting', + async ({ volume }) => { + let attempts = 0 + server.use( + http.put(apiUrl('/volumecontent/:volumeID/file'), () => { + attempts++ + return HttpResponse.json( + { message: 'rate limited' }, + { + status: 429, + headers: { 'Retry-After': '0' }, + } + ) + }) + ) + + await expect( + volume.writeFile('/retry.txt', 'retry') + ).rejects.toBeInstanceOf(RateLimitError) + expect(attempts).toBe(1) + } + ) + volumeTest('should write and read a text file', async ({ volume }) => { const path = '/test.txt' const content = 'Hello, World!'