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);