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

refactor(api): add structured logging for worker op and SSE error states

Browse source 
Introduce detailed console logging across document parsing API routes and worker
operation state fetches. Logs now capture stream openings, error conditions,
worker state unavailability, and invalid responses, improving observability and
debuggability for PDF parsing and ONNX layout processing flows.
This commit is contained in:
Richard R 2026-05-27 01:32:14 -06:00
parent 5c24ea5763
commit 956ecf5f4a
4 changed files with 69 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
src/app/(app)/pdf/[id]/usePdfDocument.ts
View file

@ -244,8 +244,12 @@ export function usePdfDocument(): PdfDocumentState {
}
},
onError: () => {
// EventSource reconnects automatically. Keep stream open.
// EventSource reconnects automatically. Keep stream open, but log so
// production debugging can correlate UI stalls with SSE churn.
if (controller.signal.aborted) return;
console.warn('[pdf] parsed/events stream error; waiting for auto-reconnect', {
documentId,
});
},
});
parseSseCloseRef.current = closeSse;

25
src/app/api/documents/[id]/parsed/events/route.ts
View file

@ -149,6 +149,12 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
const initialState = await toSnapshotState(row);
const workerCfg = getWorkerClientConfigFromEnv();
const encoder = new TextEncoder();
console.info('[parsed/events] stream open', {
documentId: id,
storageUserId,
initialParseStatus: initialState.snapshot.parseStatus,
initialOpId: initialState.opId,
});
const stream = new ReadableStream<Uint8Array>({
start(controller) {
@ -230,6 +236,11 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
}
}).catch((error) => {
if (closed) return;
console.warn('[parsed/events] db resync failed', {
documentId: id,
storageUserId,
error: error instanceof Error ? error.message : String(error),
});
controller.enqueue(encoder.encode(`event: error\ndata: ${JSON.stringify({ error: String(error) })}\n\n`));
});
}, SSE_RESYNC_INTERVAL_MS);
@ -251,6 +262,12 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
current = next.snapshot;
signature = next.signature;
currentOpId = next.opId;
if (!currentOpId) {
console.warn('[parsed/events] waiting for worker opId', {
documentId: id,
parseStatus: current.parseStatus,
});
}
if (current.parseStatus === 'ready' || current.parseStatus === 'failed') {
closeStream();
return;
@ -290,6 +307,10 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
continue;
}
if (!response.body) {
console.warn('[parsed/events] worker stream response missing body', {
documentId: id,
opId: currentOpId,
});
await sleep(500);
continue;
}
@ -380,6 +401,10 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
void runWorkerProxy()
.catch((error) => {
console.error('[parsed/events] worker proxy crashed', {
documentId: id,
error: error instanceof Error ? error.message : String(error),
});
if (!closed) {
controller.enqueue(encoder.encode(`event: error\ndata: ${JSON.stringify({ error: String(error) })}\n\n`));
}

14
src/app/api/documents/[id]/parsed/route.ts
View file

@ -96,6 +96,13 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
});
effectiveStatus = merged.parseStatus;
effectiveProgress = merged.parseProgress;
} else {
console.warn('[documents/parsed:get] worker state unavailable for op', {
documentId: id,
userId: row.userId,
opId,
parseStatus: effectiveStatus,
});
}
}
@ -220,6 +227,13 @@ export async function POST(req: NextRequest, ctx: { params: Promise<{ id: string
});
effectiveStatus = merged.parseStatus;
effectiveProgress = merged.parseProgress;
} else {
console.warn('[documents/parsed:post] worker state unavailable for op', {
documentId: id,
userId: row.userId,
opId,
parseStatus: effectiveStatus,
});
}
}

30
src/lib/server/compute/worker-op-state.ts
View file

@ -12,7 +12,11 @@ export async function fetchWorkerOperationState<Result>(
let cfg: { baseUrl: string; token: string };
try {
cfg = getWorkerClientConfigFromEnv();
} catch {
} catch (error) {
console.warn('[worker-op-state] worker client env missing/invalid', {
opId: normalized,
error: error instanceof Error ? error.message : String(error),
});
return null;
}
@ -30,14 +34,30 @@ export async function fetchWorkerOperationState<Result>(
signal: controller.signal,
});
if (!res.ok) return null;
if (!res.ok) {
const detail = await res.text().catch(() => '');
console.warn('[worker-op-state] worker op request failed', {
opId: normalized,
status: res.status,
detail,
});
return null;
}
const parsed = await res.json() as WorkerOperationState<Result>;
if (!parsed || typeof parsed !== 'object' || parsed.opId !== normalized) return null;
if (!parsed || typeof parsed !== 'object' || parsed.opId !== normalized) {
console.warn('[worker-op-state] worker op response invalid', {
opId: normalized,
});
return null;
}
return parsed;
} catch {
} catch (error) {
console.warn('[worker-op-state] worker op request threw', {
opId: normalized,
error: error instanceof Error ? error.message : String(error),
});
return null;
} finally {
clearTimeout(timeout);
}
}