diff --git a/platform/packages/api/src/assets.ts b/platform/packages/api/src/assets.ts index 39168a8b0..038826ed4 100644 --- a/platform/packages/api/src/assets.ts +++ b/platform/packages/api/src/assets.ts @@ -70,6 +70,8 @@ function tryParseJson(text: string): T | null { } const ASSET_SESSION_MAX_RETRIES = 3; +// Cloudflare sets no subrequest time limit; 60 s matches the platform's other Cloudflare API calls. +const ASSET_SESSION_ATTEMPT_TIMEOUT_MS = 60_000; const ASSET_SESSION_RETRY_DELAY_MS = 1000; const ASSET_BUCKET_UPLOAD_RETRY_DELAY_MS = 1000; @@ -109,6 +111,10 @@ export function normalizeForCfAssetSession( * been uploaded") instead of JSON when all asset hashes already exist. This is * a CF bug — the response varies by edge node. Retrying typically hits a * different node that returns proper JSON. + * + * A 5xx or a network error (a dropped response body or a timed-out attempt + * included) is retried. A 4xx is final, even when its body drops. That + * includes a 429: the install shares one API token. */ export async function createAssetUploadSession( accountId: string, @@ -126,16 +132,49 @@ export async function createAssetUploadSession( const body = JSON.stringify({ manifest }); return retryWithBackoff( - async () => { - const response = await fetch(url, { method: 'POST', headers, body }); + async (attempt) => { + const lastAttempt = attempt === ASSET_SESSION_MAX_RETRIES; + let response: Response | undefined; + let text: string; + try { + response = await fetch(url, { + method: 'POST', + headers, + body, + signal: AbortSignal.timeout(ASSET_SESSION_ATTEMPT_TIMEOUT_MS), + }); + text = await response.text(); + } catch (error) { + if (response && !response.ok && response.status < 500) { + throw new Error(`asset upload session failed (HTTP ${response.status})`, { + cause: error, + }); + } + if (lastAttempt) { + throw (error as { name?: unknown } | null)?.name === 'TimeoutError' + ? new Error( + `asset upload session did not finish within ${ASSET_SESSION_ATTEMPT_TIMEOUT_MS / 1000} seconds`, + { cause: error }, + ) + : error; + } + console.warn( + `[assets] upload session request failed, retrying: ${error instanceof Error ? error.message : String(error)}`, + ); + return undefined; + } if (!response.ok) { - const error = await response.text(); - console.error(`[assets] upload session failed (${response.status}): ${error}`); - throw new Error(`asset upload session failed: ${error}`); + if (response.status >= 500 && !lastAttempt) { + console.warn( + `[assets] upload session failed (${response.status}), retrying: ${text.slice(0, 200)}`, + ); + return undefined; + } + console.error(`[assets] upload session failed (${response.status}): ${text}`); + throw new Error(`asset upload session failed: ${text}`); } - const text = await response.text(); const data = tryParseJson(text); if (data?.result?.jwt && Array.isArray(data.result.buckets)) { const bucketCount = data.result.buckets.length; diff --git a/platform/packages/api/src/deploy/shared.ts b/platform/packages/api/src/deploy/shared.ts index 1b2185931..3aa81c47b 100644 --- a/platform/packages/api/src/deploy/shared.ts +++ b/platform/packages/api/src/deploy/shared.ts @@ -99,9 +99,10 @@ export const R2_HEAD_VERIFY_CONCURRENCY = 20; // stores. A fresh retained/in-flight hash snapshot is taken for every batch. export const ASSET_GC_LOCK_BATCH = 100; -// Retention is post-cutover and best-effort. Bound all eviction + orphan work -// across one request so cleanup can never consume the Worker's 1,000 external -// operation budget before the already-live deployment emits `done`. +// Retention is post-cutover and best effort. Cap the asset hashes one deploy +// evicts or sweeps (each costs up to a KV delete and an R2 head) so eviction +// only briefly delays the `done` event, and the orphan sweep that runs in +// `waitUntil` after the response stays short. export const ASSET_GC_MAX_HASHES_PER_DEPLOY = 200; // Per-blake3 / per-md5 validation: 32 lowercase hex characters. diff --git a/platform/packages/api/src/deploy/upload-worker.ts b/platform/packages/api/src/deploy/upload-worker.ts index ec027cb8d..a04aae1da 100644 --- a/platform/packages/api/src/deploy/upload-worker.ts +++ b/platform/packages/api/src/deploy/upload-worker.ts @@ -34,7 +34,12 @@ import { issueProjectProxyToken } from '../proxy-token-rotation.ts'; import { RESERVED_BINDING_NAMES } from '../reserved-bindings.ts'; import { deployments } from '../schema.ts'; import { type DeployManifest, type DeployEvent, type Deployment, type Project } from '../types.ts'; -import { withKnownLength, workerUploadBody, type WorkerModule } from '../worker-upload.ts'; +import { + WorkerModuleIntegrityError, + withKnownLength, + workerUploadBody, + type WorkerModule, +} from '../worker-upload.ts'; import { buildWebSocketUploadBindings, buildSandboxUploadBindings, @@ -52,6 +57,52 @@ import { } from './shared.ts'; import { platformAppUploadUserAgent } from './user-agent.ts'; +// Cloudflare sets no subrequest time limit; a timed-out attempt is retried like a network error. +const WORKER_UPLOAD_ATTEMPT_TIMEOUT_MS = 120_000; + +/** The Cloudflare error code of a rejected replay of a saved upload, if the body has one. */ +function savedUploadReplayCode(body: string): number | undefined { + let errors: unknown; + try { + errors = (JSON.parse(body) as { errors?: unknown } | null)?.errors; + } catch { + return undefined; + } + if (!Array.isArray(errors)) { + return undefined; + } + for (const error of errors) { + const code = (error as { code?: unknown } | null)?.code; + // 100312: the assets completion token was used. 10079: the migration tag moved. + if (code === 100312 || code === 10079) { + return code; + } + } + return undefined; +} + +/** Whether the dispatch script exists and carries the uploaded migration tag. */ +async function workerScriptSaved( + url: string, + headers: Readonly>, + migrationTag: string | undefined, +): Promise { + try { + const response = await fetch(url, { headers, signal: AbortSignal.timeout(30_000) }); + if (!response.ok) { + void response.body?.cancel().catch(() => {}); + return false; + } + const body = (await response.json()) as { + result?: { script?: { migration_tag?: unknown } | null } | null; + } | null; + const script = body?.result?.script; + return !!script && (migrationTag === undefined || script.migration_tag === migrationTag); + } catch { + return false; + } +} + export async function uploadDeploymentWorker(input: { currentStep: string; readonly markPhase: (next: string) => void; @@ -418,46 +469,127 @@ export async function uploadDeploymentWorker(input: { } input.workerUploadUrl = `https://api.cloudflare.com/client/v4/accounts/${context.env.CF_ACCOUNT_ID}/workers/dispatch/namespaces/${dispatchNamespace}/scripts/${workerName}`; + const maxAttempts = CF_RETRY.maxRetries + 1; + const cfHeaders = { + Authorization: `Bearer ${context.env.CF_API_TOKEN}`, + 'User-Agent': platformAppUploadUserAgent(context.env), + }; + // Set once an attempt fails after Cloudflare may already have saved it. + let maybeSaved = false; + // A replay only rewrites this deployment's own script: the name is unique + // per deployment. Every attempt sends the same body, so a saved script with + // the uploaded migration tag is this upload. + const savedDespiteFailure = async (fields: Record) => { + const saved = await workerScriptSaved( + input.workerUploadUrl!, + cfHeaders, + durableObjectMigrations?.new_tag, + ); + if (saved) { + logger.warn('worker_upload', 'saved_despite_failure', fields); + } + return saved; + }; await retryWithBackoff( async (attempt) => { const attemptStart = Date.now(); + const lastAttempt = attempt === CF_RETRY.maxRetries; // A streamed body is sent once, so each attempt reopens the modules. const upload = workerUploadBody(metadata, workerFiles); - let cfResponse: Response; + let cfResponse: Response | undefined; + let cfBody: string; + // The request and its response body are retried together: a connection + // that drops while the body is read fails like one that drops earlier. try { cfResponse = await fetch(input.workerUploadUrl!, { method: 'PUT', - headers: { - Authorization: `Bearer ${context.env.CF_API_TOKEN}`, - 'User-Agent': platformAppUploadUserAgent(context.env), - 'Content-Type': upload.contentType, - }, + headers: { ...cfHeaders, 'Content-Type': upload.contentType }, body: withKnownLength(upload.stream, upload.length), + signal: AbortSignal.timeout(WORKER_UPLOAD_ATTEMPT_TIMEOUT_MS), }); + cfBody = await cfResponse.text(); } catch (error) { // A module that fails its size or digest check stops the body before - // its closing boundary. Report that, not the transport error it causes; - // thrown errors are not retried. - throw upload.failure() ?? error; + // its closing boundary. Report that, not the transport error it causes, + // and never retry it: the same bytes fail every attempt. + const failure = upload.failure(); + if (failure instanceof WorkerModuleIntegrityError) { + throw failure; + } + const timedOut = (error as { name?: unknown } | null)?.name === 'TimeoutError'; + const cause = failure ?? error; + const fields = { + attempt, + maxAttempts, + latencyMs: Date.now() - attemptStart, + code: timedOut ? 'timeout' : 'transport', + status: cfResponse?.status, + err: cause instanceof Error ? cause.message : String(cause), + }; + // A 4xx stays final even when its body drops, unless it replays an + // upload that was saved. Any other failure is retried. + if (cfResponse && !cfResponse.ok && cfResponse.status < 500) { + if (maybeSaved && (await savedDespiteFailure(fields))) { + return true; + } + throw new Error(`worker upload failed (HTTP ${cfResponse.status})`, { cause: error }); + } + if (lastAttempt) { + // This attempt or an earlier one may have been saved. + if (await savedDespiteFailure(fields)) { + return true; + } + // The timer also covers the module reads that feed the body. + throw ( + failure ?? + (timedOut + ? new Error( + `The Worker upload did not finish within ${WORKER_UPLOAD_ATTEMPT_TIMEOUT_MS / 1000} seconds. Deploy again.`, + { cause: error }, + ) + : error) + ); + } + maybeSaved = true; + logger.warn('worker_upload', 'attempt_failed', fields); + return undefined; } if (!cfResponse.ok) { - const error = await cfResponse.text(); - logger.error('worker_upload', 'attempt_failed', new Error(error), { + const fields = { attempt, - maxAttempts: CF_RETRY.maxRetries + 1, + maxAttempts, latencyMs: Date.now() - attemptStart, status: cfResponse.status, - bodyPreview: error.slice(0, 500), - }); - throw new Error(`worker upload failed: ${error}`); + bodyPreview: cfBody.slice(0, 500), + }; + // Only 5xx is retried. A 429 stays final: the install shares one API + // token, and Cloudflare blocks it for minutes once it hits the limit. + if (cfResponse.status >= 500 && !lastAttempt) { + maybeSaved = true; + logger.warn('worker_upload', 'attempt_failed', fields); + return undefined; + } + // A replay of a saved upload fails: the assets completion token is + // single-use and the migration tag has already moved. A 5xx here ends + // the last attempt, which may itself have been saved. + const code = maybeSaved ? savedUploadReplayCode(cfBody) : undefined; + if ( + (cfResponse.status >= 500 || code !== undefined) && + (await savedDespiteFailure({ ...fields, cfCode: code })) + ) { + return true; + } + logger.error('worker_upload', 'attempt_failed', new Error(cfBody), fields); + throw new Error(`worker upload failed: ${cfBody}`); } - const cfBody = await cfResponse.text(); if (isCfAssetPlainText(cfBody)) { + // Cloudflare accepted the request, so it may have saved the script. + maybeSaved = true; logger.warn('worker_upload', 'attempt_failed', { attempt, - maxAttempts: CF_RETRY.maxRetries + 1, + maxAttempts, latencyMs: Date.now() - attemptStart, code: 'cf_plain_text', bodyPreview: cfBody.slice(0, 100), diff --git a/platform/packages/api/test/integration/api-deploy.test.ts b/platform/packages/api/test/integration/api-deploy.test.ts index 110248d78..67f3dcaac 100644 --- a/platform/packages/api/test/integration/api-deploy.test.ts +++ b/platform/packages/api/test/integration/api-deploy.test.ts @@ -54,9 +54,11 @@ import { createTestProjectSecret, mockCfApi, cfSuccess, + cfError, readAllEvents, CF_ACCOUNT_ID, JWT_SECRET, + type MockCfApiHandle, type TestEnv, } from './helpers.ts'; import { setTestDb } from './setup.ts'; @@ -1865,29 +1867,6 @@ describe('POST /projects/:id/deploy', () => { ); }); - it('marks deployment failed when worker upload fails', async () => { - const user = await createTestUser(env.db); - const project = await createTestProject(env.db, user.id, 'fail-app'); - mockCfApi({ - uploadWorker: () => new Response('internal error', { status: 500 }), - }); - - const res = await app.fetch(await buildDeployRequest(env, project.id, user.jwt), env); - expect(res.status).toBe(200); - const events = await readAllEvents(res); - const error = requireEvent(events, 'error'); - expect(error.message).toContain('worker upload failed'); - expect(error.step).toBe('worker_upload'); - - // Verify deployment is marked failed with error message - const [dpl] = await env.db - .select({ status: deployments.status, error_message: deployments.error_message }) - .from(deployments) - .where(eq(deployments.project_id, project.id)); - expect(dpl?.status).toBe('failed'); - expect(dpl?.error_message).toContain('worker upload failed:'); - }); - it('provisions queue and adds consumer on deploy with queues manifest', async () => { const user = await createTestUser(env.db); const project = await createTestProject(env.db, user.id, 'queue-app'); @@ -2309,6 +2288,7 @@ describe('POST /projects/:id/deploy', () => { }); it('retries plain-text asset upload response and succeeds on next attempt', async () => { + mockSetTimeout(); const user = await createTestUser(env.db); const project = await createTestProject(env.db, user.id, 'asset-text-test'); let uploadCalls = 0; @@ -2443,7 +2423,7 @@ describe('POST /projects/:id/deploy', () => { async () => { const user = await createTestUser(env.db); const project = await createTestProject(env.db, user.id, 'worker-upload-text-error-test'); - mockCfApi({ + const cf = mockCfApi({ uploadWorker: () => new Response('Assets have already been uploaded.'), }); @@ -2463,9 +2443,671 @@ describe('POST /projects/:id/deploy', () => { step: 'worker_upload', message: 'worker upload returned an unexpected Cloudflare response', }); + // A plain-text reply never checks for a saved script. + expect(dispatchScriptReads(cf)).toBe(0); + }, + ); + + it('retries worker upload after a Cloudflare 5xx', async () => { + const user = await createTestUser(env.db); + const project = await createTestProject(env.db, user.id, 'worker-upload-5xx-retry'); + let uploadCalls = 0; + mockCfApi({ + uploadWorker: () => { + uploadCalls++; + return uploadCalls === 1 + ? new Response('service unavailable', { status: 503 }) + : cfSuccess({}); + }, + }); + + const res = await app.fetch(await buildDeployRequest(env, project.id, user.jwt), env); + + expect(res.status).toBe(200); + const events = parseDeployStream(await res.text()); + expect(events.at(-1)).toMatchObject({ event: 'done' }); + expect(uploadCalls).toBe(2); + }); + + const dispatchScriptReads = (cf: MockCfApiHandle) => + cf.mock.calls.filter((args: Array) => { + const init = args[1] as RequestInit | undefined; + return ( + (init?.method ?? 'GET') === 'GET' && + /\/workers\/dispatch\/namespaces\/[^/]+\/scripts\/[^/]+$/.test(String(args[0])) + ); + }).length; + + // Collects the deploy log records written for a saved upload. + const captureSavedUploadLogs = () => { + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}); + return () => + warn.mock.calls.flatMap(([line]) => { + try { + const record = JSON.parse(String(line)) as Record; + return record.msg === 'saved_despite_failure' ? [record] : []; + } catch { + return []; + } + }); + }; + + it('marks deployment failed when worker upload fails', async () => { + const user = await createTestUser(env.db); + const project = await createTestProject(env.db, user.id, 'fail-app'); + let uploadCalls = 0; + const cf = mockCfApi({ + uploadWorker: () => { + uploadCalls++; + return new Response('internal error', { status: 500 }); + }, + }); + + const res = await app.fetch(await buildDeployRequest(env, project.id, user.jwt), env); + expect(res.status).toBe(200); + const events = await readAllEvents(res); + const error = requireEvent(events, 'error'); + expect(error.message).toContain('worker upload failed: internal error'); + expect(error.step).toBe('worker_upload'); + expect(uploadCalls).toBe(4); + expect(dispatchScriptReads(cf)).toBe(1); + + // Verify deployment is marked failed with error message + const [dpl] = await env.db + .select({ status: deployments.status, error_message: deployments.error_message }) + .from(deployments) + .where(eq(deployments.project_id, project.id)); + expect(dpl?.status).toBe('failed'); + expect(dpl?.error_message).toContain('worker upload failed:'); + }); + + it('retries worker upload after a transport error', async () => { + const user = await createTestUser(env.db); + const project = await createTestProject(env.db, user.id, 'worker-upload-transport-retry'); + const worker = "export default { fetch() { return new Response('retried'); } }"; + const received: Array = []; + mockCfApi({ + uploadWorker: async (_url, init) => { + received.push(await ((init!.body as FormData).get('index.js') as Blob).text()); + if (received.length === 1) { + throw new TypeError('fetch failed'); + } + return cfSuccess({}); + }, + }); + + const res = await app.fetch( + await buildDeployRequest(env, project.id, user.jwt, { workers: { 'index.js': worker } }), + env, + ); + + expect(res.status).toBe(200); + const events = parseDeployStream(await res.text()); + expect(events.at(-1)).toMatchObject({ event: 'done' }); + // The second attempt sent the whole module again. + expect(received).toEqual([worker, worker]); + }); + + it('fails with the transport error after the last attempt', async () => { + const user = await createTestUser(env.db); + const project = await createTestProject(env.db, user.id, 'worker-upload-transport-fail'); + let uploadCalls = 0; + const cf = mockCfApi({ + uploadWorker: () => { + uploadCalls++; + throw new TypeError('fetch failed'); + }, + }); + + const res = await app.fetch(await buildDeployRequest(env, project.id, user.jwt), env); + + expect(res.status).toBe(200); + const events = parseDeployStream(await res.text()); + expect(events.at(-1)).toMatchObject({ + event: 'error', + step: 'worker_upload', + message: 'fetch failed', + }); + expect(uploadCalls).toBe(4); + expect(dispatchScriptReads(cf)).toBe(1); + }); + + // The headers arrive, then the connection drops while the body is read. + const brokenBodyResponse = (status: number) => + new Response( + new ReadableStream({ + pull(controller) { + controller.error(new TypeError('terminated')); + }, + }), + { status }, + ); + + it('retries worker upload when the response body fails after a 5xx', async () => { + const user = await createTestUser(env.db); + const project = await createTestProject(env.db, user.id, 'worker-upload-5xx-body-retry'); + let uploadCalls = 0; + mockCfApi({ + uploadWorker: () => { + uploadCalls++; + return uploadCalls === 1 ? brokenBodyResponse(503) : cfSuccess({}); + }, + }); + + const res = await app.fetch(await buildDeployRequest(env, project.id, user.jwt), env); + + expect(res.status).toBe(200); + const events = parseDeployStream(await res.text()); + expect(events.at(-1)).toMatchObject({ event: 'done' }); + expect(uploadCalls).toBe(2); + }); + + it('retries worker upload when a 2xx response body fails', async () => { + const user = await createTestUser(env.db); + const project = await createTestProject(env.db, user.id, 'worker-upload-2xx-body-retry'); + let uploadCalls = 0; + mockCfApi({ + uploadWorker: () => { + uploadCalls++; + return uploadCalls === 1 ? brokenBodyResponse(200) : cfSuccess({}); + }, + }); + + const res = await app.fetch(await buildDeployRequest(env, project.id, user.jwt), env); + + expect(res.status).toBe(200); + const events = parseDeployStream(await res.text()); + expect(events.at(-1)).toMatchObject({ event: 'done' }); + expect(uploadCalls).toBe(2); + }); + + it.each([400, 429])('does not retry a worker upload rejected with %i', async (status) => { + const user = await createTestUser(env.db); + const project = await createTestProject(env.db, user.id, `worker-upload-rejected-${status}`); + let uploadCalls = 0; + const cf = mockCfApi({ + uploadWorker: () => { + uploadCalls++; + return new Response('rejected', { + status, + headers: status === 429 ? { 'Retry-After': '1' } : {}, + }); + }, + }); + + const res = await app.fetch(await buildDeployRequest(env, project.id, user.jwt), env); + + expect(res.status).toBe(200); + const events = parseDeployStream(await res.text()); + const error = requireEvent(events, 'error'); + expect(error.step).toBe('worker_upload'); + expect(error.message).toContain('worker upload failed:'); + expect(uploadCalls).toBe(1); + expect(dispatchScriptReads(cf)).toBe(0); + }); + + it.each([400, 429])( + 'does not retry a worker upload rejected with %i when its body drops', + async (status) => { + const user = await createTestUser(env.db); + const project = await createTestProject( + env.db, + user.id, + `worker-upload-rejected-body-${status}`, + ); + let uploadCalls = 0; + const cf = mockCfApi({ + uploadWorker: () => { + uploadCalls++; + return brokenBodyResponse(status); + }, + }); + + const res = await app.fetch(await buildDeployRequest(env, project.id, user.jwt), env); + + expect(res.status).toBe(200); + const events = parseDeployStream(await res.text()); + const error = requireEvent(events, 'error'); + expect(error.step).toBe('worker_upload'); + expect(error.message).toContain(`worker upload failed (HTTP ${status})`); + expect(uploadCalls).toBe(1); + expect(dispatchScriptReads(cf)).toBe(0); }, ); + const timeoutError = () => + new DOMException('The operation was aborted due to timeout', 'TimeoutError'); + + it('retries a worker upload attempt that times out', async () => { + const user = await createTestUser(env.db); + const project = await createTestProject(env.db, user.id, 'worker-upload-timeout-retry'); + const signals: Array = []; + mockCfApi({ + uploadWorker: (_url, init) => { + signals.push(init?.signal); + if (signals.length === 1) { + throw timeoutError(); + } + return cfSuccess({}); + }, + }); + + const res = await app.fetch(await buildDeployRequest(env, project.id, user.jwt), env); + + expect(res.status).toBe(200); + const events = parseDeployStream(await res.text()); + expect(events.at(-1)).toMatchObject({ event: 'done' }); + expect(signals).toHaveLength(2); + expect(signals[0]).toBeInstanceOf(AbortSignal); + expect(signals[1]).toBeInstanceOf(AbortSignal); + // Each attempt has its own timer. + expect(signals[1]).not.toBe(signals[0]); + }); + + it('reports a timeout after the last worker upload attempt', async () => { + const user = await createTestUser(env.db); + const project = await createTestProject(env.db, user.id, 'worker-upload-timeout-fail'); + let uploadCalls = 0; + const cf = mockCfApi({ + uploadWorker: () => { + uploadCalls++; + throw timeoutError(); + }, + }); + + const res = await app.fetch(await buildDeployRequest(env, project.id, user.jwt), env); + + expect(res.status).toBe(200); + const events = parseDeployStream(await res.text()); + const error = requireEvent(events, 'error'); + expect(error.step).toBe('worker_upload'); + expect(error.message).toContain('did not finish within 120 seconds'); + expect(uploadCalls).toBe(4); + expect(dispatchScriptReads(cf)).toBe(1); + }); + + // Cloudflare stores the script, but the platform never sees its answer. + async function saveScript( + cf: MockCfApiHandle, + url: string, + init: RequestInit | undefined, + migrationTag?: string, + ): Promise { + const metadata = JSON.parse(await ((init!.body as FormData).get('metadata') as Blob).text()); + return cf.seedHostScript({ + scriptName: url.slice(url.lastIndexOf('/') + 1), + migrationTag: migrationTag ?? metadata.migrations?.new_tag, + bindings: metadata.bindings, + metadata, + }).migration_tag; + } + + const tokenConsumed = () => cfError('Completion token has already been consumed', 400, 100312); + const tagMoved = () => + cfError("Actor migration tag precondition failed, got tag ''", 412, 10079); + + const websocketManifest = { + version: 3, + bindings: {}, + websocket: { + routes: [ + { + pattern: '/ws/chat/:room', + filePath: 'routes/ws/chat/[room].ws.ts', + className: 'WsChatRoomWs', + bindingName: 'CHAT_ROOM', + mode: 'room', + hasAuthGate: false, + }, + ], + classes: ['WsChatRoomWs'], + migrations: [{ tag: 'ws-v1', newClasses: ['WsChatRoomWs'] }], + }, + }; + + it('treats an upload saved before a 5xx as done', async () => { + const user = await createTestUser(env.db); + const project = await createTestProject(env.db, user.id, 'worker-upload-saved-5xx'); + const savedLogs = captureSavedUploadLogs(); + let uploadCalls = 0; + const cf: MockCfApiHandle = mockCfApi({ + uploadWorker: async (url, init) => { + uploadCalls++; + if (uploadCalls === 1) { + await saveScript(cf, url, init); + return new Response('service unavailable', { status: 503 }); + } + return tokenConsumed(); + }, + }); + + const res = await app.fetch( + await buildDeployRequest(env, project.id, user.jwt, { + assets: { 'index.html': '

hi

' }, + }), + env, + ); + + expect(res.status).toBe(200); + const events = parseDeployStream(await res.text()); + expect(events.at(-1)).toMatchObject({ event: 'done' }); + expect(uploadCalls).toBe(2); + expect(dispatchScriptReads(cf)).toBe(1); + const logs = savedLogs(); + expect(logs).toEqual([ + expect.objectContaining({ + attempt: 1, + maxAttempts: 4, + latencyMs: expect.any(Number), + status: 400, + cfCode: 100312, + bodyPreview: expect.stringContaining('100312'), + }), + ]); + // `code` names the failure kind; Cloudflare's number has its own field. + expect(logs[0]).not.toHaveProperty('code'); + }); + + it('treats a saved migration tag as done', async () => { + const user = await createTestUser(env.db); + const project = await createTestProject(env.db, user.id, 'worker-upload-saved-tag'); + let uploadCalls = 0; + let savedTag: string | undefined; + const cf: MockCfApiHandle = mockCfApi({ + uploadWorker: async (url, init) => { + uploadCalls++; + if (uploadCalls === 1) { + savedTag = await saveScript(cf, url, init); + throw new TypeError('fetch failed'); + } + return tagMoved(); + }, + }); + + const res = await app.fetch( + await buildDeployRequest(env, project.id, user.jwt, { manifest: websocketManifest }), + env, + ); + + expect(res.status).toBe(200); + const events = parseDeployStream(await res.text()); + expect(events.at(-1)).toMatchObject({ event: 'done' }); + expect(savedTag).toBe('ws-v1'); + expect(uploadCalls).toBe(2); + expect(dispatchScriptReads(cf)).toBe(1); + }); + + it('treats an upload saved before a plain-text reply as done', async () => { + const user = await createTestUser(env.db); + const project = await createTestProject(env.db, user.id, 'worker-upload-saved-text'); + let uploadCalls = 0; + const cf: MockCfApiHandle = mockCfApi({ + uploadWorker: async (url, init) => { + uploadCalls++; + if (uploadCalls === 1) { + await saveScript(cf, url, init); + return new Response('Assets have already been uploaded.'); + } + return tokenConsumed(); + }, + }); + + const res = await app.fetch( + await buildDeployRequest(env, project.id, user.jwt, { + assets: { 'index.html': '

hi

' }, + }), + env, + ); + + expect(res.status).toBe(200); + const events = parseDeployStream(await res.text()); + expect(events.at(-1)).toMatchObject({ event: 'done' }); + expect(uploadCalls).toBe(2); + expect(dispatchScriptReads(cf)).toBe(1); + }); + + it('treats an upload saved before a replay whose 4xx body drops as done', async () => { + const user = await createTestUser(env.db); + const project = await createTestProject(env.db, user.id, 'worker-upload-saved-4xx-body'); + const savedLogs = captureSavedUploadLogs(); + let uploadCalls = 0; + const cf: MockCfApiHandle = mockCfApi({ + uploadWorker: async (url, init) => { + uploadCalls++; + if (uploadCalls === 1) { + await saveScript(cf, url, init); + throw new TypeError('fetch failed'); + } + return brokenBodyResponse(400); + }, + }); + + const res = await app.fetch(await buildDeployRequest(env, project.id, user.jwt), env); + + expect(res.status).toBe(200); + const events = parseDeployStream(await res.text()); + expect(events.at(-1)).toMatchObject({ event: 'done' }); + expect(uploadCalls).toBe(2); + expect(dispatchScriptReads(cf)).toBe(1); + expect(savedLogs()).toEqual([ + expect.objectContaining({ + attempt: 1, + maxAttempts: 4, + latencyMs: expect.any(Number), + code: 'transport', + status: 400, + err: 'terminated', + }), + ]); + }); + + it('treats an upload saved before the last attempt times out as done', async () => { + const user = await createTestUser(env.db); + const project = await createTestProject(env.db, user.id, 'worker-upload-saved-timeout'); + const savedLogs = captureSavedUploadLogs(); + let uploadCalls = 0; + const cf: MockCfApiHandle = mockCfApi({ + uploadWorker: async (url, init) => { + uploadCalls++; + if (uploadCalls === 1) { + await saveScript(cf, url, init); + throw new TypeError('fetch failed'); + } + throw timeoutError(); + }, + }); + + const res = await app.fetch(await buildDeployRequest(env, project.id, user.jwt), env); + + expect(res.status).toBe(200); + const events = parseDeployStream(await res.text()); + expect(events.at(-1)).toMatchObject({ event: 'done' }); + expect(uploadCalls).toBe(4); + expect(dispatchScriptReads(cf)).toBe(1); + expect(savedLogs()).toEqual([ + expect.objectContaining({ + attempt: 3, + maxAttempts: 4, + latencyMs: expect.any(Number), + code: 'timeout', + err: expect.stringContaining('timeout'), + }), + ]); + }); + + it('reports the timeout when the saved migration tag differs', async () => { + const user = await createTestUser(env.db); + const project = await createTestProject(env.db, user.id, 'worker-upload-timeout-other-tag'); + let uploadCalls = 0; + const cf: MockCfApiHandle = mockCfApi({ + uploadWorker: async (url, init) => { + uploadCalls++; + if (uploadCalls === 1) { + await saveScript(cf, url, init, 'ws-v0'); + } + throw timeoutError(); + }, + }); + + const res = await app.fetch( + await buildDeployRequest(env, project.id, user.jwt, { manifest: websocketManifest }), + env, + ); + + expect(res.status).toBe(200); + const events = parseDeployStream(await res.text()); + const error = requireEvent(events, 'error'); + expect(error.step).toBe('worker_upload'); + expect(error.message).toContain('did not finish within 120 seconds'); + expect(uploadCalls).toBe(4); + expect(dispatchScriptReads(cf)).toBe(1); + }); + + it('treats an upload saved before a 5xx on the last attempt as done', async () => { + const user = await createTestUser(env.db); + const project = await createTestProject(env.db, user.id, 'worker-upload-saved-last-5xx'); + const savedLogs = captureSavedUploadLogs(); + let uploadCalls = 0; + const cf: MockCfApiHandle = mockCfApi({ + uploadWorker: async (url, init) => { + uploadCalls++; + if (uploadCalls === 1) { + await saveScript(cf, url, init); + } + return new Response('service unavailable', { status: 503 }); + }, + }); + + const res = await app.fetch(await buildDeployRequest(env, project.id, user.jwt), env); + + expect(res.status).toBe(200); + const events = parseDeployStream(await res.text()); + expect(events.at(-1)).toMatchObject({ event: 'done' }); + expect(uploadCalls).toBe(4); + expect(dispatchScriptReads(cf)).toBe(1); + expect(savedLogs()).toEqual([ + expect.objectContaining({ + attempt: 3, + maxAttempts: 4, + latencyMs: expect.any(Number), + status: 503, + bodyPreview: 'service unavailable', + }), + ]); + }); + + it('does not check for a saved upload after an unrelated rejection', async () => { + const user = await createTestUser(env.db); + const project = await createTestProject(env.db, user.id, 'worker-upload-other-rejection'); + let uploadCalls = 0; + const cf = mockCfApi({ + uploadWorker: () => { + uploadCalls++; + return uploadCalls === 1 + ? new Response('service unavailable', { status: 503 }) + : cfError('Uncaught SyntaxError', 400, 10021); + }, + }); + + const res = await app.fetch(await buildDeployRequest(env, project.id, user.jwt), env); + + expect(res.status).toBe(200); + const events = parseDeployStream(await res.text()); + const error = requireEvent(events, 'error'); + expect(error.step).toBe('worker_upload'); + expect(error.message).toContain('worker upload failed:'); + expect(error.message).toContain('10021'); + expect(uploadCalls).toBe(2); + expect(dispatchScriptReads(cf)).toBe(0); + }); + + it('fails when the saved migration tag differs', async () => { + const user = await createTestUser(env.db); + const project = await createTestProject(env.db, user.id, 'worker-upload-other-tag'); + let uploadCalls = 0; + const cf: MockCfApiHandle = mockCfApi({ + uploadWorker: async (url, init) => { + uploadCalls++; + if (uploadCalls === 1) { + await saveScript(cf, url, init, 'ws-v0'); + return new Response('service unavailable', { status: 503 }); + } + return tagMoved(); + }, + }); + + const res = await app.fetch( + await buildDeployRequest(env, project.id, user.jwt, { manifest: websocketManifest }), + env, + ); + + expect(res.status).toBe(200); + const events = parseDeployStream(await res.text()); + const error = requireEvent(events, 'error'); + expect(error.step).toBe('worker_upload'); + expect(error.message).toContain('worker upload failed:'); + expect(error.message).toContain('10079'); + expect(uploadCalls).toBe(2); + expect(dispatchScriptReads(cf)).toBe(1); + }); + + it('fails when the script was not saved', async () => { + const user = await createTestUser(env.db); + const project = await createTestProject(env.db, user.id, 'worker-upload-not-saved'); + let uploadCalls = 0; + const cf = mockCfApi({ + uploadWorker: () => { + uploadCalls++; + return uploadCalls === 1 + ? new Response('service unavailable', { status: 503 }) + : tokenConsumed(); + }, + }); + + const res = await app.fetch( + await buildDeployRequest(env, project.id, user.jwt, { + assets: { 'index.html': '

hi

' }, + }), + env, + ); + + expect(res.status).toBe(200); + const events = parseDeployStream(await res.text()); + const error = requireEvent(events, 'error'); + expect(error.step).toBe('worker_upload'); + expect(error.message).toContain('worker upload failed:'); + expect(error.message).toContain('100312'); + expect(uploadCalls).toBe(2); + expect(dispatchScriptReads(cf)).toBe(1); + }); + + it('does not check for a saved upload without an earlier unclear failure', async () => { + const user = await createTestUser(env.db); + const project = await createTestProject(env.db, user.id, 'worker-upload-first-rejected'); + let uploadCalls = 0; + const cf = mockCfApi({ + uploadWorker: () => { + uploadCalls++; + return tokenConsumed(); + }, + }); + + const res = await app.fetch( + await buildDeployRequest(env, project.id, user.jwt, { + assets: { 'index.html': '

hi

' }, + }), + env, + ); + + expect(res.status).toBe(200); + const events = parseDeployStream(await res.text()); + const error = requireEvent(events, 'error'); + expect(error.step).toBe('worker_upload'); + expect(error.message).toContain('100312'); + expect(uploadCalls).toBe(1); + expect(dispatchScriptReads(cf)).toBe(0); + }); + it('polls readiness probe before dispatching migrations', async () => { const user = await createTestUser(env.db); const project = await createTestProject(env.db, user.id, 'migrate-dispatch-retry-test'); diff --git a/platform/packages/api/test/integration/deploy-direct-r2.test.ts b/platform/packages/api/test/integration/deploy-direct-r2.test.ts index 324a3f0ff..2d8c230e2 100644 --- a/platform/packages/api/test/integration/deploy-direct-r2.test.ts +++ b/platform/packages/api/test/integration/deploy-direct-r2.test.ts @@ -55,6 +55,15 @@ afterEach(() => { vi.restoreAllMocks(); }); +const nativeSetTimeout = globalThis.setTimeout.bind(globalThis); + +function mockSetTimeout() { + vi.spyOn(globalThis, 'setTimeout').mockImplementation((( + fn: TimerHandler, + ...args: Array + ) => nativeSetTimeout(fn, 0, ...args)) as typeof setTimeout); +} + // Synthetic 32-hex blake3 / md5 strings. Real hashes round-trip through the // presigner unchanged; the deploy pipeline only validates shape (32 hex) // and looks up bytes by `assets/${projectId}/${blake3}`. Tests put the @@ -2146,7 +2155,7 @@ describe('direct Worker module upload', () => { const code = new TextEncoder().encode('export default {};'); const { id, modules } = await upload(project.id, { 'index.js': code }); const uploaded = vi.fn(async () => cfSuccess({})); - mockCfApi({ uploadWorker: uploaded }); + const cf = mockCfApi({ uploadWorker: uploaded }); // Same size, so the finalize listing accepts it; only the digest differs. await env.STATIC_ASSETS_R2.put( `deployment-inputs/${project.id}/${id}/0`, @@ -2162,6 +2171,8 @@ describe('direct Worker module upload', () => { }), ); expect(uploaded).not.toHaveBeenCalled(); + // Changed bytes fail every attempt, so the upload is not retried. + expect(workerUploadCalls(cf, project.slug)).toHaveLength(1); const [deployment] = await env.db .select({ status: deployments.status }) .from(deployments) @@ -2174,7 +2185,34 @@ describe('direct Worker module upload', () => { expect(left.objects).toEqual([]); }); + it('retries the upload when reading a stored module fails once', async () => { + const project = await createTestProject(env.db, user.id, 'wu-read-retry'); + const code = new TextEncoder().encode('export default {};'); + const { id, modules } = await upload(project.id, { 'index.js': code }); + const cf = mockCfApi(); + const key = `deployment-inputs/${project.id}/${id}/0`; + const originalGet = env.STATIC_ASSETS_R2.get.bind(env.STATIC_ASSETS_R2); + let failed = false; + vi.spyOn(env.STATIC_ASSETS_R2, 'get').mockImplementation(async (getKey, options) => { + if (getKey === key && !failed) { + failed = true; + throw new Error('R2 unavailable'); + } + return originalGet(getKey, options); + }); + // Skip the retry backoff. + mockSetTimeout(); + + const events = await readAllEvents( + await finalize(project.id, { workerInput: { id, modules } }), + ); + + expect(events.find((e) => isEvent(e, 'done'))).toBeDefined(); + expect(workerUploadCalls(cf, project.slug)).toHaveLength(2); + }); + it('reopens every module when Cloudflare asks for the upload again', async () => { + mockSetTimeout(); const project = await createTestProject(env.db, user.id, 'wu-retry'); const files = { 'index.js': new TextEncoder().encode('export default {};'), diff --git a/platform/packages/api/test/unit/assets.test.ts b/platform/packages/api/test/unit/assets.test.ts index b360efe6f..81a59f120 100644 --- a/platform/packages/api/test/unit/assets.test.ts +++ b/platform/packages/api/test/unit/assets.test.ts @@ -738,13 +738,143 @@ describe('createAssetUploadSession', () => { expect(fetchMock).toHaveBeenCalledTimes(4); }); - it('does not retry on HTTP error responses', async () => { - const fetchMock = vi.fn(async () => new Response('Unauthorized', { status: 401 })); + it('retries a 5xx response and succeeds', async () => { + const fetchMock = vi + .fn() + .mockResolvedValueOnce(new Response('Service Unavailable', { status: 503 })) + .mockResolvedValueOnce(Response.json({ result: { jwt: 'retry-jwt', buckets: [] } })); + vi.stubGlobal('fetch', fetchMock); + + const result = await createAssetUploadSession('acct', 'token', 'ns', 'worker', manifest, { + retryDelayMs: 0, + }); + + expect(result).toEqual({ jwt: 'retry-jwt', buckets: [] }); + expect(fetchMock).toHaveBeenCalledTimes(2); + }); + + it('retries a network error and succeeds', async () => { + const fetchMock = vi + .fn() + .mockRejectedValueOnce(new TypeError('fetch failed')) + .mockResolvedValueOnce(Response.json({ result: { jwt: 'retry-jwt', buckets: [] } })); + vi.stubGlobal('fetch', fetchMock); + + const result = await createAssetUploadSession('acct', 'token', 'ns', 'worker', manifest, { + retryDelayMs: 0, + }); + + expect(result).toEqual({ jwt: 'retry-jwt', buckets: [] }); + expect(fetchMock).toHaveBeenCalledTimes(2); + }); + + // The headers arrive, then the connection drops while the body is read. + const droppedResponse = (status = 200) => + new Response( + new ReadableStream({ + start(controller) { + controller.error(new TypeError('terminated')); + }, + }), + { status }, + ); + + it('retries when the response body drops and succeeds', async () => { + const fetchMock = vi + .fn() + .mockResolvedValueOnce(droppedResponse()) + .mockResolvedValueOnce(Response.json({ result: { jwt: 'retry-jwt', buckets: [] } })); + vi.stubGlobal('fetch', fetchMock); + + const result = await createAssetUploadSession('acct', 'token', 'ns', 'worker', manifest, { + retryDelayMs: 0, + }); + + expect(result).toEqual({ jwt: 'retry-jwt', buckets: [] }); + expect(fetchMock).toHaveBeenCalledTimes(2); + }); + + it('throws the network error after exhausting retries', async () => { + const fetchMock = vi.fn(async () => { + throw new TypeError('fetch failed'); + }); + vi.stubGlobal('fetch', fetchMock); + + await expect( + createAssetUploadSession('acct', 'token', 'ns', 'worker', manifest, { retryDelayMs: 0 }), + ).rejects.toThrow('fetch failed'); + + expect(fetchMock).toHaveBeenCalledTimes(4); + }); + + const timeoutError = () => + new DOMException('The operation was aborted due to timeout', 'TimeoutError'); + + it('gives each attempt a fresh timeout signal', async () => { + const signals: Array = []; + const fetchMock = vi.fn(async (_input: RequestInfo | URL, init?: RequestInit) => { + signals.push(init?.signal); + if (signals.length === 1) { + throw timeoutError(); + } + return Response.json({ result: { jwt: 'retry-jwt', buckets: [] } }); + }); + vi.stubGlobal('fetch', fetchMock); + + const result = await createAssetUploadSession('acct', 'token', 'ns', 'worker', manifest, { + retryDelayMs: 0, + }); + + expect(result).toEqual({ jwt: 'retry-jwt', buckets: [] }); + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(signals[0]).toBeInstanceOf(AbortSignal); + expect(signals[1]).toBeInstanceOf(AbortSignal); + expect(signals[1]).not.toBe(signals[0]); + }); + + it('reports a timeout after the last attempt', async () => { + const fetchMock = vi.fn(async () => { + throw timeoutError(); + }); + vi.stubGlobal('fetch', fetchMock); + + await expect( + createAssetUploadSession('acct', 'token', 'ns', 'worker', manifest, { retryDelayMs: 0 }), + ).rejects.toThrow('did not finish within 60 seconds'); + + expect(fetchMock).toHaveBeenCalledTimes(4); + }); + + it('throws the last 5xx response after exhausting retries', async () => { + const fetchMock = vi.fn(async () => new Response('Bad Gateway', { status: 502 })); + vi.stubGlobal('fetch', fetchMock); + + await expect( + createAssetUploadSession('acct', 'token', 'ns', 'worker', manifest, { retryDelayMs: 0 }), + ).rejects.toThrow('asset upload session failed: Bad Gateway'); + + // 1 initial + 3 retries = 4 total + expect(fetchMock).toHaveBeenCalledTimes(4); + }); + + it.each([401, 429])('does not retry a 4xx response (%i)', async (status) => { + const fetchMock = vi.fn(async () => new Response('Denied', { status })); + vi.stubGlobal('fetch', fetchMock); + + await expect( + createAssetUploadSession('acct', 'token', 'ns', 'worker', manifest, { retryDelayMs: 0 }), + ).rejects.toThrow('asset upload session failed: Denied'); + + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + it.each([401, 429])('does not retry a 4xx response whose body drops (%i)', async (status) => { + const fetchMock = vi.fn(async () => droppedResponse(status)); vi.stubGlobal('fetch', fetchMock); await expect( createAssetUploadSession('acct', 'token', 'ns', 'worker', manifest, { retryDelayMs: 0 }), - ).rejects.toThrow('asset upload session failed: Unauthorized'); + ).rejects.toThrow(`asset upload session failed (HTTP ${status})`); expect(fetchMock).toHaveBeenCalledTimes(1); });