From bfe3c8c099c76ff1b231409dad8dc0b5bd029699 Mon Sep 17 00:00:00 2001 From: LongYinan Date: Fri, 16 Oct 2026 12:14:38 +0800 Subject: [PATCH 1/6] fix(platform): retry the Worker upload after a Cloudflare 5xx or network error The Worker upload retried only when Cloudflare answered 2xx with a plain-text body. A 5xx or a connection reset failed the deploy at once, and nothing above it retries: compensation deletes the script and marks the deployment failed. 5xx responses and network errors now use the same four attempts. A module that fails its size or digest check still fails at once, and 4xx (429 included) stays final. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../packages/api/src/deploy/upload-worker.ts | 45 +++++- .../api/test/integration/api-deploy.test.ts | 146 +++++++++++++++--- .../test/integration/deploy-direct-r2.test.ts | 34 +++- 3 files changed, 193 insertions(+), 32 deletions(-) diff --git a/platform/packages/api/src/deploy/upload-worker.ts b/platform/packages/api/src/deploy/upload-worker.ts index ec027cb8d..2589a49a7 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, @@ -418,9 +423,13 @@ 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; + // A replay only rewrites this deployment's own script: the name is unique + // per deployment. 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; @@ -436,20 +445,40 @@ export async function uploadDeploymentWorker(input: { }); } 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. Any other + // failure is retried. + const failure = upload.failure(); + if (failure instanceof WorkerModuleIntegrityError || lastAttempt) { + throw failure ?? error; + } + const cause = failure ?? error; + logger.warn('worker_upload', 'attempt_failed', { + attempt, + maxAttempts, + latencyMs: Date.now() - attemptStart, + code: 'transport', + err: cause instanceof Error ? cause.message : String(cause), + }); + 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), - }); + }; + // 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) { + logger.warn('worker_upload', 'attempt_failed', fields); + return undefined; + } + logger.error('worker_upload', 'attempt_failed', new Error(error), fields); throw new Error(`worker upload failed: ${error}`); } @@ -457,7 +486,7 @@ export async function uploadDeploymentWorker(input: { if (isCfAssetPlainText(cfBody)) { 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..1019cf40f 100644 --- a/platform/packages/api/test/integration/api-deploy.test.ts +++ b/platform/packages/api/test/integration/api-deploy.test.ts @@ -1865,29 +1865,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'); @@ -2466,6 +2443,129 @@ describe('POST /projects/:id/deploy', () => { }, ); + 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); + }); + + 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; + 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); + + // 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; + 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); + }); + + 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; + 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); + }); + 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..ba335d6c3 100644 --- a/platform/packages/api/test/integration/deploy-direct-r2.test.ts +++ b/platform/packages/api/test/integration/deploy-direct-r2.test.ts @@ -2146,7 +2146,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 +2162,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,6 +2176,36 @@ 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. + const nativeSetTimeout = globalThis.setTimeout; + vi.spyOn(globalThis, 'setTimeout').mockImplementation((( + fn: TimerHandler, + ...args: Array + ) => nativeSetTimeout(fn, 0, ...args)) as typeof setTimeout); + + 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 () => { const project = await createTestProject(env.db, user.id, 'wu-retry'); const files = { From 4be7ede61283474be1878d07c3ca944476f384a1 Mon Sep 17 00:00:00 2001 From: LongYinan Date: Fri, 16 Oct 2026 12:43:49 +0800 Subject: [PATCH 2/6] fix(platform): retry the Worker upload when the response body drops A connection that drops while the platform reads Cloudflare's answer failed the deploy at once, even when the same drop before the headers was retried. The request and its response body now retry together. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../packages/api/src/deploy/upload-worker.ts | 12 +++-- .../api/test/integration/api-deploy.test.ts | 49 +++++++++++++++++++ 2 files changed, 56 insertions(+), 5 deletions(-) diff --git a/platform/packages/api/src/deploy/upload-worker.ts b/platform/packages/api/src/deploy/upload-worker.ts index 2589a49a7..e61b4dba2 100644 --- a/platform/packages/api/src/deploy/upload-worker.ts +++ b/platform/packages/api/src/deploy/upload-worker.ts @@ -433,6 +433,9 @@ export async function uploadDeploymentWorker(input: { // A streamed body is sent once, so each attempt reopens the modules. const upload = workerUploadBody(metadata, workerFiles); let cfResponse: Response; + 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', @@ -443,6 +446,7 @@ export async function uploadDeploymentWorker(input: { }, body: withKnownLength(upload.stream, upload.length), }); + 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, @@ -464,13 +468,12 @@ export async function uploadDeploymentWorker(input: { } if (!cfResponse.ok) { - const error = await cfResponse.text(); const fields = { attempt, maxAttempts, latencyMs: Date.now() - attemptStart, status: cfResponse.status, - bodyPreview: error.slice(0, 500), + 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. @@ -478,11 +481,10 @@ export async function uploadDeploymentWorker(input: { logger.warn('worker_upload', 'attempt_failed', fields); return undefined; } - logger.error('worker_upload', 'attempt_failed', new Error(error), fields); - throw new Error(`worker upload failed: ${error}`); + 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)) { logger.warn('worker_upload', 'attempt_failed', { attempt, diff --git a/platform/packages/api/test/integration/api-deploy.test.ts b/platform/packages/api/test/integration/api-deploy.test.ts index 1019cf40f..5e3c7e09e 100644 --- a/platform/packages/api/test/integration/api-deploy.test.ts +++ b/platform/packages/api/test/integration/api-deploy.test.ts @@ -2542,6 +2542,55 @@ describe('POST /projects/:id/deploy', () => { expect(uploadCalls).toBe(4); }); + // 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}`); From 95789cd4480e7d602179025572a409e1a00da3c9 Mon Sep 17 00:00:00 2001 From: LongYinan Date: Fri, 16 Oct 2026 13:10:45 +0800 Subject: [PATCH 3/6] fix(platform): detect a Worker upload saved before an unclear failure When Cloudflare saves the upload but the platform sees a 5xx, a network error or a dropped body, the retry is a replay. Cloudflare rejects every replay: the assets completion token is single-use (400, code 100312) and the migration tag has already moved (412, code 10079). The deploy then failed with a misleading message. After an unclear failure, a replay rejected with one of those codes now reads the dispatch script. If it exists with the expected migration tag, the upload counts as done. Each attempt also gets a 120-second timeout. Cloudflare sets no time limit on a subrequest, so a stalled connection held the deployment until the Queue's 15-minute limit. A 4xx stays final when its body drops. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../packages/api/src/deploy/upload-worker.ts | 111 +++++- .../api/test/integration/api-deploy.test.ts | 328 ++++++++++++++++++ 2 files changed, 428 insertions(+), 11 deletions(-) diff --git a/platform/packages/api/src/deploy/upload-worker.ts b/platform/packages/api/src/deploy/upload-worker.ts index e61b4dba2..25fee9ddf 100644 --- a/platform/packages/api/src/deploy/upload-worker.ts +++ b/platform/packages/api/src/deploy/upload-worker.ts @@ -57,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; @@ -424,6 +470,12 @@ 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. await retryWithBackoff( @@ -432,36 +484,52 @@ export async function uploadDeploymentWorker(input: { 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, - // and never retry it: the same bytes fail every attempt. Any other - // failure is retried. + // and never retry it: the same bytes fail every attempt. const failure = upload.failure(); - if (failure instanceof WorkerModuleIntegrityError || lastAttempt) { - throw failure ?? error; + if (failure instanceof WorkerModuleIntegrityError) { + throw failure; } + // A 4xx stays final even when its body drops. Any other failure is + // retried. + if (cfResponse && !cfResponse.ok && cfResponse.status < 500) { + throw new Error(`worker upload failed (HTTP ${cfResponse.status})`, { cause: error }); + } + const timedOut = (error as { name?: unknown } | null)?.name === 'TimeoutError'; + if (lastAttempt) { + // 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; const cause = failure ?? error; logger.warn('worker_upload', 'attempt_failed', { attempt, maxAttempts, latencyMs: Date.now() - attemptStart, - code: 'transport', + code: timedOut ? 'timeout' : 'transport', + status: cfResponse?.status, err: cause instanceof Error ? cause.message : String(cause), }); return undefined; @@ -478,14 +546,35 @@ export async function uploadDeploymentWorker(input: { // 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. + const code = maybeSaved ? savedUploadReplayCode(cfBody) : undefined; + if ( + code !== undefined && + (await workerScriptSaved( + input.workerUploadUrl!, + cfHeaders, + durableObjectMigrations?.new_tag, + )) + ) { + logger.warn('worker_upload', 'saved_by_earlier_attempt', { + attempt, + status: cfResponse.status, + code, + }); + return true; + } logger.error('worker_upload', 'attempt_failed', new Error(cfBody), fields); throw new Error(`worker upload failed: ${cfBody}`); } if (isCfAssetPlainText(cfBody)) { + // Cloudflare accepted the request, so it may have saved the script. + maybeSaved = true; logger.warn('worker_upload', 'attempt_failed', { attempt, maxAttempts, diff --git a/platform/packages/api/test/integration/api-deploy.test.ts b/platform/packages/api/test/integration/api-deploy.test.ts index 5e3c7e09e..0afcaeab4 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'; @@ -2286,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; @@ -2615,6 +2618,331 @@ describe('POST /projects/:id/deploy', () => { expect(uploadCalls).toBe(1); }); + 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; + 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); + }, + ); + + 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; + 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); + }); + + // 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 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; + + 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'); + 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); + }); + + 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('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'); From b2275d5603d2d7b3121f5561682e693c591c1b6c Mon Sep 17 00:00:00 2001 From: LongYinan Date: Fri, 16 Oct 2026 13:10:46 +0800 Subject: [PATCH 4/6] fix(platform): retry the asset upload session after a 5xx or network error The asset upload session request failed the deploy on the first Cloudflare 5xx or network error. It now uses the same rule as the Worker upload. A repeated session request is safe: only the token from the attempt that succeeds is used. Also correct the asset cleanup cap comment, which cited an old subrequest limit, and stop two deploy tests from sleeping for real on retry backoff. Co-Authored-By: Claude Opus 5.5 (1M context) --- platform/packages/api/src/assets.ts | 39 ++++++-- platform/packages/api/src/deploy/shared.ts | 7 +- .../test/integration/deploy-direct-r2.test.ts | 16 ++- .../packages/api/test/unit/assets.test.ts | 98 ++++++++++++++++++- 4 files changed, 143 insertions(+), 17 deletions(-) diff --git a/platform/packages/api/src/assets.ts b/platform/packages/api/src/assets.ts index 39168a8b0..0a2af6a29 100644 --- a/platform/packages/api/src/assets.ts +++ b/platform/packages/api/src/assets.ts @@ -109,6 +109,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 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 +130,39 @@ 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 }); + 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; + } + 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/test/integration/deploy-direct-r2.test.ts b/platform/packages/api/test/integration/deploy-direct-r2.test.ts index ba335d6c3..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 @@ -2192,11 +2201,7 @@ describe('direct Worker module upload', () => { return originalGet(getKey, options); }); // Skip the retry backoff. - const nativeSetTimeout = globalThis.setTimeout; - vi.spyOn(globalThis, 'setTimeout').mockImplementation((( - fn: TimerHandler, - ...args: Array - ) => nativeSetTimeout(fn, 0, ...args)) as typeof setTimeout); + mockSetTimeout(); const events = await readAllEvents( await finalize(project.id, { workerInput: { id, modules } }), @@ -2207,6 +2212,7 @@ describe('direct Worker module upload', () => { }); 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..a0bf94659 100644 --- a/platform/packages/api/test/unit/assets.test.ts +++ b/platform/packages/api/test/unit/assets.test.ts @@ -738,13 +738,105 @@ 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); + }); + + 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); }); From 204b6daf6de531fc842c8ddeceb3e26e7e37d5f2 Mon Sep 17 00:00:00 2001 From: LongYinan Date: Fri, 16 Oct 2026 13:33:59 +0800 Subject: [PATCH 5/6] fix(platform): check for a saved Worker upload before the final failure A timeout, network error or 5xx on the last attempt failed the deploy without the saved-upload check, which ran only when a replay got a 100312 or 10079 answer. The last attempt now reads the dispatch script once before it fails, because that attempt may itself have been saved. The asset upload session also gets a 60-second timeout for each attempt, like the platform's other Cloudflare API calls. Co-Authored-By: Claude Opus 5.5 (1M context) --- platform/packages/api/src/assets.ts | 22 ++- .../packages/api/src/deploy/upload-worker.ts | 53 +++--- .../api/test/integration/api-deploy.test.ts | 160 ++++++++++++++++-- .../packages/api/test/unit/assets.test.ts | 38 +++++ 4 files changed, 231 insertions(+), 42 deletions(-) diff --git a/platform/packages/api/src/assets.ts b/platform/packages/api/src/assets.ts index 0a2af6a29..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; @@ -110,9 +112,9 @@ export function normalizeForCfAssetSession( * 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 included) is retried. A - * 4xx is final, even when its body drops. That includes a 429: the install - * shares one API token. + * 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, @@ -135,7 +137,12 @@ export async function createAssetUploadSession( let response: Response | undefined; let text: string; try { - response = await fetch(url, { method: 'POST', headers, body }); + 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) { @@ -144,7 +151,12 @@ export async function createAssetUploadSession( }); } if (lastAttempt) { - throw error; + 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)}`, diff --git a/platform/packages/api/src/deploy/upload-worker.ts b/platform/packages/api/src/deploy/upload-worker.ts index 25fee9ddf..8282ee796 100644 --- a/platform/packages/api/src/deploy/upload-worker.ts +++ b/platform/packages/api/src/deploy/upload-worker.ts @@ -477,7 +477,19 @@ export async function uploadDeploymentWorker(input: { // 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. + // 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(); @@ -510,7 +522,20 @@ export async function uploadDeploymentWorker(input: { throw new Error(`worker upload failed (HTTP ${cfResponse.status})`, { cause: error }); } 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), + }; 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 ?? @@ -523,15 +548,7 @@ export async function uploadDeploymentWorker(input: { ); } maybeSaved = true; - const cause = failure ?? error; - logger.warn('worker_upload', 'attempt_failed', { - attempt, - maxAttempts, - latencyMs: Date.now() - attemptStart, - code: timedOut ? 'timeout' : 'transport', - status: cfResponse?.status, - err: cause instanceof Error ? cause.message : String(cause), - }); + logger.warn('worker_upload', 'attempt_failed', fields); return undefined; } @@ -551,21 +568,13 @@ export async function uploadDeploymentWorker(input: { return undefined; } // A replay of a saved upload fails: the assets completion token is - // single-use and the migration tag has already moved. + // 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 ( - code !== undefined && - (await workerScriptSaved( - input.workerUploadUrl!, - cfHeaders, - durableObjectMigrations?.new_tag, - )) + (cfResponse.status >= 500 || code !== undefined) && + (await savedDespiteFailure({ ...fields, cfCode: code })) ) { - logger.warn('worker_upload', 'saved_by_earlier_attempt', { - attempt, - status: cfResponse.status, - code, - }); return true; } logger.error('worker_upload', 'attempt_failed', new Error(cfBody), fields); diff --git a/platform/packages/api/test/integration/api-deploy.test.ts b/platform/packages/api/test/integration/api-deploy.test.ts index 0afcaeab4..e92d6e725 100644 --- a/platform/packages/api/test/integration/api-deploy.test.ts +++ b/platform/packages/api/test/integration/api-deploy.test.ts @@ -2423,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.'), }); @@ -2443,6 +2443,8 @@ 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); }, ); @@ -2467,11 +2469,34 @@ describe('POST /projects/:id/deploy', () => { 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; - mockCfApi({ + const cf = mockCfApi({ uploadWorker: () => { uploadCalls++; return new Response('internal error', { status: 500 }); @@ -2485,6 +2510,7 @@ describe('POST /projects/:id/deploy', () => { 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 @@ -2526,7 +2552,7 @@ describe('POST /projects/:id/deploy', () => { const user = await createTestUser(env.db); const project = await createTestProject(env.db, user.id, 'worker-upload-transport-fail'); let uploadCalls = 0; - mockCfApi({ + const cf = mockCfApi({ uploadWorker: () => { uploadCalls++; throw new TypeError('fetch failed'); @@ -2543,6 +2569,7 @@ describe('POST /projects/:id/deploy', () => { message: 'fetch failed', }); expect(uploadCalls).toBe(4); + expect(dispatchScriptReads(cf)).toBe(1); }); // The headers arrive, then the connection drops while the body is read. @@ -2598,7 +2625,7 @@ describe('POST /projects/:id/deploy', () => { const user = await createTestUser(env.db); const project = await createTestProject(env.db, user.id, `worker-upload-rejected-${status}`); let uploadCalls = 0; - mockCfApi({ + const cf = mockCfApi({ uploadWorker: () => { uploadCalls++; return new Response('rejected', { @@ -2616,6 +2643,7 @@ describe('POST /projects/:id/deploy', () => { 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])( @@ -2628,7 +2656,7 @@ describe('POST /projects/:id/deploy', () => { `worker-upload-rejected-body-${status}`, ); let uploadCalls = 0; - mockCfApi({ + const cf = mockCfApi({ uploadWorker: () => { uploadCalls++; return brokenBodyResponse(status); @@ -2643,6 +2671,7 @@ describe('POST /projects/:id/deploy', () => { expect(error.step).toBe('worker_upload'); expect(error.message).toContain(`worker upload failed (HTTP ${status})`); expect(uploadCalls).toBe(1); + expect(dispatchScriptReads(cf)).toBe(0); }, ); @@ -2679,7 +2708,7 @@ describe('POST /projects/:id/deploy', () => { const user = await createTestUser(env.db); const project = await createTestProject(env.db, user.id, 'worker-upload-timeout-fail'); let uploadCalls = 0; - mockCfApi({ + const cf = mockCfApi({ uploadWorker: () => { uploadCalls++; throw timeoutError(); @@ -2694,6 +2723,7 @@ describe('POST /projects/:id/deploy', () => { 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. @@ -2716,15 +2746,6 @@ describe('POST /projects/:id/deploy', () => { const tagMoved = () => cfError("Actor migration tag precondition failed, got tag ''", 412, 10079); - 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; - const websocketManifest = { version: 3, bindings: {}, @@ -2747,6 +2768,7 @@ describe('POST /projects/:id/deploy', () => { 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) => { @@ -2771,6 +2793,19 @@ describe('POST /projects/:id/deploy', () => { 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 () => { @@ -2831,6 +2866,101 @@ describe('POST /projects/:id/deploy', () => { expect(dispatchScriptReads(cf)).toBe(1); }); + 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'); diff --git a/platform/packages/api/test/unit/assets.test.ts b/platform/packages/api/test/unit/assets.test.ts index a0bf94659..81a59f120 100644 --- a/platform/packages/api/test/unit/assets.test.ts +++ b/platform/packages/api/test/unit/assets.test.ts @@ -807,6 +807,44 @@ describe('createAssetUploadSession', () => { 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); From e5a160a57da5b2dd4cb3fe43e2981b3a96fe5504 Mon Sep 17 00:00:00 2001 From: LongYinan Date: Fri, 16 Oct 2026 13:43:08 +0800 Subject: [PATCH 6/6] fix(platform): check for a saved upload when a replay's 4xx body drops After an unclear failure, the replay is rejected with a 4xx. If that response body dropped, the deploy failed before the saved-upload check ran. It now runs the same check first. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../packages/api/src/deploy/upload-worker.ts | 13 ++++--- .../api/test/integration/api-deploy.test.ts | 35 +++++++++++++++++++ 2 files changed, 43 insertions(+), 5 deletions(-) diff --git a/platform/packages/api/src/deploy/upload-worker.ts b/platform/packages/api/src/deploy/upload-worker.ts index 8282ee796..a04aae1da 100644 --- a/platform/packages/api/src/deploy/upload-worker.ts +++ b/platform/packages/api/src/deploy/upload-worker.ts @@ -516,11 +516,6 @@ export async function uploadDeploymentWorker(input: { if (failure instanceof WorkerModuleIntegrityError) { throw failure; } - // A 4xx stays final even when its body drops. Any other failure is - // retried. - if (cfResponse && !cfResponse.ok && cfResponse.status < 500) { - throw new Error(`worker upload failed (HTTP ${cfResponse.status})`, { cause: error }); - } const timedOut = (error as { name?: unknown } | null)?.name === 'TimeoutError'; const cause = failure ?? error; const fields = { @@ -531,6 +526,14 @@ export async function uploadDeploymentWorker(input: { 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)) { diff --git a/platform/packages/api/test/integration/api-deploy.test.ts b/platform/packages/api/test/integration/api-deploy.test.ts index e92d6e725..67f3dcaac 100644 --- a/platform/packages/api/test/integration/api-deploy.test.ts +++ b/platform/packages/api/test/integration/api-deploy.test.ts @@ -2866,6 +2866,41 @@ describe('POST /projects/:id/deploy', () => { 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');