[Fix] Retry the worker's job-claim callbacks while the public edge settles

apps/dev/src/services/pm2.ts

@@ -167,6 +167,52 @@ export class PM2Service {
     } else {
       validateServices.succeed();
     }
+
+    await this.validatePublicEdge(options);
+  }
+
+  /**
+   * Docker workers call back into the API through the public URL (ngrok →
+   * Caddy edge), which can lag behind the local services after a (re)start.
+   * Poll the API health route through the edge so jobs launched right after
+   * startup don't race a proxy that isn't serving yet.
+   */
+  private static async validatePublicEdge(
+    options: ScriptOptions,
+  ): Promise<void> {
+    if (!options.publicUrl) {
+      return;
+    }
+
+    const healthUrl = `${options.publicUrl.replace(/\/+$/, '')}/_roomote-api/`;
+    const checkEdge = ora(`Checking public edge at ${healthUrl}`).start();
+    const deadline = Date.now() + 45_000;
+    let lastFailure = 'no response';
+
+    while (Date.now() < deadline) {
+      try {
+        const response = await fetch(healthUrl, {
+          signal: AbortSignal.timeout(5_000),
+        });
+        const contentType = response.headers.get('content-type') ?? '';
+
+        if (response.ok && contentType.includes('application/json')) {
+          checkEdge.succeed();
+          return;
+        }
+
+        lastFailure = `status ${response.status}, content-type ${contentType || 'missing'}`;
+      } catch (error) {
+        lastFailure = error instanceof Error ? error.message : String(error);
+      }
+
+      await new Promise((resolve) => setTimeout(resolve, 2_000));
+    }
+
+    checkEdge.warn();
+    console.warn(
+      `  ⚠️  Public edge did not serve the API health check within 45s (last: ${lastFailure}). Worker callbacks may fail until it settles.`,
+    );
   }
 
   private static getExpectedServices(options: ScriptOptions): string[] {

packages/sdk/src/client/index.ts

@@ -70,9 +70,6 @@ export interface CreateClientOptions {
 const WORKER_QUERY_RETRYABLE_STATUS_CODES = new Set([408, 429, 502, 503, 504]);
 const WORKER_QUERY_RETRY_MAX_ATTEMPTS = 4;
 const WORKER_QUERY_RETRY_BASE_DELAY_MS = 250;
-const RETRYABLE_WORKER_TRPC_MUTATION_PATHS = new Set([
-  'cloudJobs.recordMessageEnvelope',
-]);
 
 type FetchLike = typeof fetch;
 
@@ -81,6 +78,28 @@ export interface WorkerQueryRetryOptions {
   baseDelayMs?: number;
 }
 
+// The worker's first callbacks claim its job through the public edge, and a
+// freshly (re)started proxy can briefly answer 5xx or without a JSON
+// content-type before its upstreams settle. Claiming is idempotent server-side
+// (FOR UPDATE SKIP LOCKED returns nothing on a second attempt), so these paths
+// get a longer budget (~31s of backoff) instead of failing the whole job on
+// one bad response.
+const WORKER_STARTUP_MUTATION_RETRY_OPTIONS: WorkerQueryRetryOptions = {
+  maxAttempts: 6,
+  baseDelayMs: 1_000,
+};
+
+// Worker tRPC mutations that are safe to replay at the transport layer, with
+// optional per-path overrides of the default retry budget.
+const RETRYABLE_WORKER_TRPC_MUTATION_PATHS = new Map<
+  string,
+  WorkerQueryRetryOptions
+>([
+  ['cloudJobs.recordMessageEnvelope', {}],
+  ['cloudJobs.dequeue', WORKER_STARTUP_MUTATION_RETRY_OPTIONS],
+  ['cloudJobs.resume', WORKER_STARTUP_MUTATION_RETRY_OPTIONS],
+]);
+
 function isQueryRequest(method?: string): boolean {
   return (method ?? 'GET').toUpperCase() === 'GET';
 }
@@ -171,26 +190,55 @@ function shouldRetryWorkerQueryResponse(status: number): boolean {
   return WORKER_QUERY_RETRYABLE_STATUS_CODES.has(status);
 }
 
-function shouldRetryWorkerTrpcRequest(
+type WorkerTrpcRetryPlan =
+  | { retryable: false }
+  | { retryable: true; maxAttempts: number; baseDelayMs: number };
+
+function resolveWorkerTrpcRetryPlan(
   method: string,
   requestUrl: string,
-): boolean {
+): WorkerTrpcRetryPlan {
   if (isQueryRequest(method)) {
-    return true;
+    return {
+      retryable: true,
+      maxAttempts: WORKER_QUERY_RETRY_MAX_ATTEMPTS,
+      baseDelayMs: WORKER_QUERY_RETRY_BASE_DELAY_MS,
+    };
   }
 
   if (!isPostRequest(method)) {
-    return false;
+    return { retryable: false };
   }
 
   const procedurePaths = getWorkerTrpcProcedurePaths(requestUrl);
 
-  return (
-    procedurePaths.length > 0 &&
-    procedurePaths.every((path) =>
-      RETRYABLE_WORKER_TRPC_MUTATION_PATHS.has(path),
-    )
-  );
+  if (procedurePaths.length === 0) {
+    return { retryable: false };
+  }
+
+  // A batched request is only retryable when every procedure in it is, and the
+  // smallest configured budget wins so batching never extends a path's budget.
+  let maxAttempts = Infinity;
+  let baseDelayMs = Infinity;
+
+  for (const path of procedurePaths) {
+    const overrides = RETRYABLE_WORKER_TRPC_MUTATION_PATHS.get(path);
+
+    if (!overrides) {
+      return { retryable: false };
+    }
+
+    maxAttempts = Math.min(
+      maxAttempts,
+      overrides.maxAttempts ?? WORKER_QUERY_RETRY_MAX_ATTEMPTS,
+    );
+    baseDelayMs = Math.min(
+      baseDelayMs,
+      overrides.baseDelayMs ?? WORKER_QUERY_RETRY_BASE_DELAY_MS,
+    );
+  }
+
+  return { retryable: true, maxAttempts, baseDelayMs };
 }
 
 function getWorkerQueryRetryDelayMs(
@@ -204,18 +252,25 @@ export function createWorkerFetchWithRetry(
   baseFetch: FetchLike = globalThis.fetch.bind(globalThis),
   options: WorkerQueryRetryOptions = {},
 ): FetchLike {
-  const maxAttempts = options.maxAttempts ?? WORKER_QUERY_RETRY_MAX_ATTEMPTS;
-  const baseDelayMs = options.baseDelayMs ?? WORKER_QUERY_RETRY_BASE_DELAY_MS;
-
   return async (input, init) => {
     const requestMethod = getRequestMethod(input, init);
     const requestUrl = getRequestUrl(input);
-    const shouldRetryRequest = shouldRetryWorkerTrpcRequest(
-      requestMethod,
-      requestUrl,
-    );
-
-    if (!shouldRetryRequest || maxAttempts <= 1) {
+    const retryPlan = resolveWorkerTrpcRetryPlan(requestMethod, requestUrl);
+
+    // Explicit wrapper options take precedence over per-path budgets so
+    // callers (and tests) can pin the retry behavior.
+    const maxAttempts =
+      options.maxAttempts ??
+      (retryPlan.retryable
+        ? retryPlan.maxAttempts
+        : WORKER_QUERY_RETRY_MAX_ATTEMPTS);
+    const baseDelayMs =
+      options.baseDelayMs ??
+      (retryPlan.retryable
+        ? retryPlan.baseDelayMs
+        : WORKER_QUERY_RETRY_BASE_DELAY_MS);
+
+    if (!retryPlan.retryable || maxAttempts <= 1) {
       const response = await baseFetch(input, init);
 
       await assertValidWorkerTrpcResponse(response, requestUrl);