diff --git a/compute/worker/src/runtime.ts b/compute/worker/src/runtime.ts index efd9208..d535543 100644 --- a/compute/worker/src/runtime.ts +++ b/compute/worker/src/runtime.ts @@ -753,18 +753,22 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti orphanRecoveryPromise = (async () => { const now = Date.now(); const states = await operationStateStore.listOpStates!(); - const stalePdfStates = states.filter((state) => ( - state.kind === 'pdf_layout' - && isInflightStatus(state.status) - && (now - state.updatedAt) > opStaleMs - )); + const stalePdfStates = states.filter((state) => { + if (state.kind !== 'pdf_layout' || !isInflightStatus(state.status)) return false; + const ageMs = now - state.updatedAt; + if (state.status === 'running') { + return ageMs > pdfTimeoutMs; + } + return ageMs > opStaleMs; + }); for (const state of stalePdfStates) { + const staleAfterMs = state.status === 'running' ? pdfTimeoutMs : opStaleMs; await orchestrator.markFailed({ opId: state.opId, error: { code: 'WORKER_ORPHANED_OP', - message: `Worker stopped before completion; stale operation recovered on startup after ${opStaleMs}ms`, + message: `Worker stopped before completion; stale operation recovered on startup after ${staleAfterMs}ms`, }, updatedAt: now, }); diff --git a/compute/worker/tests/api/routes.test.ts b/compute/worker/tests/api/routes.test.ts index 5be0a59..d5d2745 100644 --- a/compute/worker/tests/api/routes.test.ts +++ b/compute/worker/tests/api/routes.test.ts @@ -120,36 +120,49 @@ describe('compute worker API routes', () => { expect(stream.body).toContain('"status":"succeeded"'); }); - test('marks stale in-flight pdf ops failed during startup reconciliation', async () => { + test('marks stale running pdf ops failed during startup reconciliation but leaves queued ops on the conservative path', async () => { + const now = Date.now(); fake.seedState({ - opId: 'op-stale', - opKey: 'k-stale', + opId: 'op-stale-running', + opKey: 'k-stale-running', kind: 'pdf_layout', - jobId: 'job-op-stale', + jobId: 'job-op-stale-running', status: 'running', queuedAt: 1, - updatedAt: 1, + updatedAt: now - 310_000, + }); + fake.seedState({ + opId: 'op-stale-queued', + opKey: 'k-stale-queued', + kind: 'pdf_layout', + jobId: 'job-op-stale-queued', + status: 'queued', + queuedAt: 1, + updatedAt: now - 310_000, }); const fetch = await runtime.app.inject({ method: 'GET', - url: '/ops/op-stale', + url: '/ops/op-stale-running', headers: AUTH, }); expect(fetch.statusCode).toBe(200); expect(fetch.json()).toMatchObject({ - opId: 'op-stale', + opId: 'op-stale-running', status: 'failed', error: { code: 'WORKER_ORPHANED_OP', }, }); - expect(fake.getState('op-stale')).toMatchObject({ + expect(fake.getState('op-stale-running')).toMatchObject({ status: 'failed', error: { code: 'WORKER_ORPHANED_OP', }, }); + expect(fake.getState('op-stale-queued')).toMatchObject({ + status: 'queued', + }); }); });