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

feat(worker): extend orphan recovery to handle whisper_align ops and improve logging

Browse source 
Update orphan recovery to detect and fail stale 'running' whisper_align operations
in addition to pdf_layout. Refactor recovery logic to generalize staleness checks
across operation kinds and enhance log output with detailed operation info.
Expand tests to verify correct handling of both whisper_align and pdf_layout
operations in running and queued states.
This commit is contained in:
Richard R 2026-06-03 01:56:54 -06:00
parent a10955280c
commit 99dbef667e
2 changed files with 55 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -753,17 +753,21 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti
orphanRecoveryPromise = (async () => {
const now = Date.now();
const states = await operationStateStore.listOpStates!();
const stalePdfStates = states.filter((state) => {
if (state.kind !== 'pdf_layout' || !isInflightStatus(state.status)) return false;
const staleStates = states.filter((state) => {
if (!isInflightStatus(state.status)) return false;
const ageMs = now - state.updatedAt;
if (state.status === 'running') {
return ageMs > pdfTimeoutMs;
const runningTimeoutMs = state.kind === 'whisper_align' ? whisperTimeoutMs : pdfTimeoutMs;
return ageMs > runningTimeoutMs;
}
if (state.kind !== 'pdf_layout') return false;
return ageMs > opStaleMs;
});
for (const state of stalePdfStates) {
const staleAfterMs = state.status === 'running' ? pdfTimeoutMs : opStaleMs;
for (const state of staleStates) {
const staleAfterMs = state.status === 'running'
? (state.kind === 'whisper_align' ? whisperTimeoutMs : pdfTimeoutMs)
: opStaleMs;
await orchestrator.markFailed({
opId: state.opId,
error: {
@ -774,11 +778,15 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti
});
}
if (stalePdfStates.length > 0) {
if (staleStates.length > 0) {
app.log.warn({
recoveredCount: stalePdfStates.length,
opIds: stalePdfStates.map((state) => state.opId),
}, 'recovered stale in-flight pdf operations on startup');
recoveredCount: staleStates.length,
ops: staleStates.map((state) => ({
opId: state.opId,
kind: state.kind,
status: state.status,
})),
}, 'recovered stale in-flight operations on startup');
}
orphanRecoveryDoneForGeneration = sessionGeneration;

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

@ -120,22 +120,40 @@ describe('compute worker API routes', () => {
expect(stream.body).toContain('"status":"succeeded"');
});
test('marks stale running pdf ops failed during startup reconciliation but leaves queued ops on the conservative path', async () => {
test('marks stale running whisper and pdf ops failed during startup reconciliation but leaves queued ops on the conservative path', async () => {
const now = Date.now();
fake.seedState({
opId: 'op-stale-running',
opKey: 'k-stale-running',
opId: 'op-stale-whisper-running',
opKey: 'k-stale-whisper-running',
kind: 'whisper_align',
jobId: 'job-op-stale-whisper-running',
status: 'running',
queuedAt: 1,
updatedAt: now - 40_000,
});
fake.seedState({
opId: 'op-stale-whisper-queued',
opKey: 'k-stale-whisper-queued',
kind: 'whisper_align',
jobId: 'job-op-stale-whisper-queued',
status: 'queued',
queuedAt: 1,
updatedAt: now - 40_000,
});
fake.seedState({
opId: 'op-stale-pdf-running',
opKey: 'k-stale-pdf-running',
kind: 'pdf_layout',
jobId: 'job-op-stale-running',
jobId: 'job-op-stale-pdf-running',
status: 'running',
queuedAt: 1,
updatedAt: now - 310_000,
});
fake.seedState({
opId: 'op-stale-queued',
opKey: 'k-stale-queued',
opId: 'op-stale-pdf-queued',
opKey: 'k-stale-pdf-queued',
kind: 'pdf_layout',
jobId: 'job-op-stale-queued',
jobId: 'job-op-stale-pdf-queued',
status: 'queued',
queuedAt: 1,
updatedAt: now - 310_000,
@ -143,25 +161,34 @@ describe('compute worker API routes', () => {
const fetch = await runtime.app.inject({
method: 'GET',
url: '/ops/op-stale-running',
url: '/ops/op-stale-whisper-running',
headers: AUTH,
});
expect(fetch.statusCode).toBe(200);
expect(fetch.json()).toMatchObject({
opId: 'op-stale-running',
opId: 'op-stale-whisper-running',
status: 'failed',
error: {
code: 'WORKER_ORPHANED_OP',
},
});
expect(fake.getState('op-stale-running')).toMatchObject({
expect(fake.getState('op-stale-whisper-running')).toMatchObject({
status: 'failed',
error: {
code: 'WORKER_ORPHANED_OP',
},
});
expect(fake.getState('op-stale-queued')).toMatchObject({
expect(fake.getState('op-stale-whisper-queued')).toMatchObject({
status: 'queued',
});
expect(fake.getState('op-stale-pdf-running')).toMatchObject({
status: 'failed',
error: {
code: 'WORKER_ORPHANED_OP',
},
});
expect(fake.getState('op-stale-pdf-queued')).toMatchObject({
status: 'queued',
});
});