danvics/openreader
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(worker): distinguish staleness thresholds for running and queued pdf ops

Browse source 
Update orphan recovery logic to apply separate timeouts for 'running' and
'queued' pdf_layout operations. Adjust tests to verify that only stale
'running' operations are failed, while stale 'queued' operations remain
untouched.
This commit is contained in:
Richard R 2026-06-03 01:48:41 -06:00
parent 80596da773
commit a10955280c
2 changed files with 31 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
compute/worker/src/runtime.ts
View file

@ -753,18 +753,22 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti
orphanRecoveryPromise = (async () => { orphanRecoveryPromise = (async () => {
const now = Date.now(); const now = Date.now();
const states = await operationStateStore.listOpStates!(); const states = await operationStateStore.listOpStates!();
const stalePdfStates = states.filter((state) => ( const stalePdfStates = states.filter((state) => {
state.kind === 'pdf_layout' if (state.kind !== 'pdf_layout' || !isInflightStatus(state.status)) return false;
&& isInflightStatus(state.status) const ageMs = now - state.updatedAt;
&& (now - state.updatedAt) > opStaleMs if (state.status === 'running') {
)); return ageMs > pdfTimeoutMs;
}
return ageMs > opStaleMs;
});
for (const state of stalePdfStates) { for (const state of stalePdfStates) {
const staleAfterMs = state.status === 'running' ? pdfTimeoutMs : opStaleMs;
await orchestrator.markFailed({ await orchestrator.markFailed({
opId: state.opId, opId: state.opId,
error: { error: {
code: 'WORKER_ORPHANED_OP', 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, updatedAt: now,
}); });

29
compute/worker/tests/api/routes.test.ts
View file

@ -120,36 +120,49 @@ describe('compute worker API routes', () => {
expect(stream.body).toContain('"status":"succeeded"'); 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({ fake.seedState({
opId: 'op-stale', opId: 'op-stale-running',
opKey: 'k-stale', opKey: 'k-stale-running',
kind: 'pdf_layout', kind: 'pdf_layout',
jobId: 'job-op-stale', jobId: 'job-op-stale-running',
status: 'running', status: 'running',
queuedAt: 1, 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({ const fetch = await runtime.app.inject({
method: 'GET', method: 'GET',
url: '/ops/op-stale', url: '/ops/op-stale-running',
headers: AUTH, headers: AUTH,
}); });
expect(fetch.statusCode).toBe(200); expect(fetch.statusCode).toBe(200);
expect(fetch.json()).toMatchObject({ expect(fetch.json()).toMatchObject({
opId: 'op-stale', opId: 'op-stale-running',
status: 'failed', status: 'failed',
error: { error: {
code: 'WORKER_ORPHANED_OP', code: 'WORKER_ORPHANED_OP',
}, },
}); });
expect(fake.getState('op-stale')).toMatchObject({ expect(fake.getState('op-stale-running')).toMatchObject({
status: 'failed', status: 'failed',
error: { error: {
code: 'WORKER_ORPHANED_OP', code: 'WORKER_ORPHANED_OP',
}, },
}); });
expect(fake.getState('op-stale-queued')).toMatchObject({
status: 'queued',
});
}); });
}); });