From a24e4eefd3cdfee41716a5b42896583ae81e6106 Mon Sep 17 00:00:00 2001 From: daniel-lxs Date: Thu, 9 Jul 2026 00:09:39 -0500 Subject: [PATCH] [Fix] Retry the worker's job-claim callbacks while the public edge settles The first callbacks a docker worker makes (cloudJobs.dequeue / resume) go through the public URL, and a freshly (re)started ngrok/Caddy edge can briefly answer 5xx or without a JSON content-type. The worker failed the whole job on the first bad response, orphaning it in 'dequeued'. Claiming a job is idempotent server-side (FOR UPDATE SKIP LOCKED), so add the claim mutations to the retryable-path allowlist with a longer budget (6 attempts, ~31s of backoff). Also have the dev CLI poll the API health route through the public edge after starting services, so local stacks don't hand jobs to workers before the edge is serving. Co-Authored-By: Claude Fable 5 --- apps/dev/src/services/pm2.ts | 46 ++++++++++++ packages/sdk/src/client/index.test.ts | 103 ++++++++++++++++++++++++++ packages/sdk/src/client/index.ts | 99 +++++++++++++++++++------ 3 files changed, 226 insertions(+), 22 deletions(-) diff --git a/apps/dev/src/services/pm2.ts b/apps/dev/src/services/pm2.ts index d0ac41ce7..bb2a732da 100644 --- a/apps/dev/src/services/pm2.ts +++ b/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 { + 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[] { diff --git a/packages/sdk/src/client/index.test.ts b/packages/sdk/src/client/index.test.ts index 37f65b449..b63898968 100644 --- a/packages/sdk/src/client/index.test.ts +++ b/packages/sdk/src/client/index.test.ts @@ -316,6 +316,109 @@ describe('createWorkerFetchWithRetry', () => { expect(fetchMock).toHaveBeenCalledTimes(2); }); + it('retries the dequeue claim when the public edge answers without a JSON content-type', async () => { + const fetchMock = vi + .fn() + .mockResolvedValueOnce(new Response(null, { status: 502 })) + .mockResolvedValueOnce( + new Response('[]', { + status: 200, + headers: { 'content-type': 'application/json' }, + }), + ); + + const workerFetch = createWorkerFetchWithRetry(fetchMock, { + baseDelayMs: 0, + }); + + const response = await workerFetch( + 'https://web-dan.ngrok.dev/_roomote-api/trpc/cloudJobs.dequeue?batch=1', + { + method: 'POST', + body: '{"0":{"json":{"cloudJobId":68}}}', + }, + ); + + expect(response.status).toBe(200); + expect(fetchMock).toHaveBeenCalledTimes(2); + }); + + it('gives the dequeue claim a longer default retry budget than other callbacks', async () => { + const fetchMock = vi + .fn() + .mockRejectedValueOnce(new TypeError('fetch failed')) + .mockRejectedValueOnce(new TypeError('fetch failed')) + .mockRejectedValueOnce(new TypeError('fetch failed')) + .mockRejectedValueOnce(new TypeError('fetch failed')) + .mockRejectedValueOnce(new TypeError('fetch failed')) + .mockResolvedValueOnce( + new Response('[]', { + status: 200, + headers: { 'content-type': 'application/json' }, + }), + ); + + const workerFetch = createWorkerFetchWithRetry(fetchMock, { + baseDelayMs: 0, + }); + + const response = await workerFetch( + 'https://api.roomote.dev/trpc/cloudJobs.dequeue?batch=1', + { + method: 'POST', + body: '{"0":{"json":{"cloudJobId":68}}}', + }, + ); + + expect(response.status).toBe(200); + expect(fetchMock).toHaveBeenCalledTimes(6); + }); + + it('retries the resume claim on transport failure', async () => { + const fetchMock = vi + .fn() + .mockRejectedValueOnce(new TypeError('fetch failed')) + .mockResolvedValueOnce( + new Response('[]', { + status: 200, + headers: { 'content-type': 'application/json' }, + }), + ); + + const workerFetch = createWorkerFetchWithRetry(fetchMock, { + baseDelayMs: 0, + }); + + const response = await workerFetch( + 'https://api.roomote.dev/trpc/cloudJobs.resume?batch=1', + { + method: 'POST', + body: '{"0":{"json":{"cloudJobId":68}}}', + }, + ); + + expect(response.status).toBe(200); + expect(fetchMock).toHaveBeenCalledTimes(2); + }); + + it('lets explicit wrapper options pin the dequeue retry budget', async () => { + const fetchError = new TypeError('fetch failed'); + const fetchMock = vi.fn().mockRejectedValue(fetchError); + + const workerFetch = createWorkerFetchWithRetry(fetchMock, { + maxAttempts: 2, + baseDelayMs: 0, + }); + + await expect( + workerFetch('https://api.roomote.dev/trpc/cloudJobs.dequeue?batch=1', { + method: 'POST', + body: '{"0":{"json":{"cloudJobId":68}}}', + }), + ).rejects.toBe(fetchError); + expect(fetchMock).toHaveBeenCalledTimes(2); + }); + it('does not retry non-idempotent worker mutations', async () => { const fetchError = new TypeError('fetch failed'); const fetchMock = vi.fn().mockRejectedValue(fetchError); diff --git a/packages/sdk/src/client/index.ts b/packages/sdk/src/client/index.ts index 9146db35b..26aad9883 100644 --- a/packages/sdk/src/client/index.ts +++ b/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);