fix(platform): retry Cloudflare uploads after a 5xx, network error or timeout

platform/packages/api/src/assets.ts

@@ -70,6 +70,8 @@ function tryParseJson<T>(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<UploadSessionResponse>(text);
       if (data?.result?.jwt && Array.isArray(data.result.buckets)) {
         const bucketCount = data.result.buckets.length;

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.

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<Record<string, string>>,
+  migrationTag: string | undefined,
+): Promise<boolean> {
+  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<string, unknown>) => {
+    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),