From 956ecf5f4aeaf586f59f7917dc17dc931214c935 Mon Sep 17 00:00:00 2001 From: Richard R Date: Wed, 27 May 2026 01:32:14 -0600 Subject: [PATCH] refactor(api): add structured logging for worker op and SSE error states 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. --- src/app/(app)/pdf/[id]/usePdfDocument.ts | 6 +++- .../api/documents/[id]/parsed/events/route.ts | 25 ++++++++++++++++ src/app/api/documents/[id]/parsed/route.ts | 14 +++++++++ src/lib/server/compute/worker-op-state.ts | 30 +++++++++++++++---- 4 files changed, 69 insertions(+), 6 deletions(-) diff --git a/src/app/(app)/pdf/[id]/usePdfDocument.ts b/src/app/(app)/pdf/[id]/usePdfDocument.ts index 4451333..4966e17 100644 --- a/src/app/(app)/pdf/[id]/usePdfDocument.ts +++ b/src/app/(app)/pdf/[id]/usePdfDocument.ts @@ -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; diff --git a/src/app/api/documents/[id]/parsed/events/route.ts b/src/app/api/documents/[id]/parsed/events/route.ts index 315b3fc..5859d4a 100644 --- a/src/app/api/documents/[id]/parsed/events/route.ts +++ b/src/app/api/documents/[id]/parsed/events/route.ts @@ -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({ 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`)); } diff --git a/src/app/api/documents/[id]/parsed/route.ts b/src/app/api/documents/[id]/parsed/route.ts index 492f1db..8713c95 100644 --- a/src/app/api/documents/[id]/parsed/route.ts +++ b/src/app/api/documents/[id]/parsed/route.ts @@ -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, + }); } } diff --git a/src/lib/server/compute/worker-op-state.ts b/src/lib/server/compute/worker-op-state.ts index fcf4ce8..e1c76b6 100644 --- a/src/lib/server/compute/worker-op-state.ts +++ b/src/lib/server/compute/worker-op-state.ts @@ -12,7 +12,11 @@ export async function fetchWorkerOperationState( 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( 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; - 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); } } -