From 671d5c6adb0468a1dbac2e4656d350e02ee9fbaf Mon Sep 17 00:00:00 2001 From: Richard R Date: Thu, 21 May 2026 13:50:40 -0600 Subject: [PATCH] feat(pdf,worker): enhance PDF layout parsing UX and add robust timeout controls Introduce idle and hard cap timeouts for PDF layout parsing in the compute worker, ensuring long-running jobs are capped and progress is tracked with explicit timeouts. Refactor NATS JetStream client initialization to use a configurable API timeout. Update PDF viewer UI to treat parse state as 'pending' by default, improving clarity during document preparation and parsing. Refine parse loader labels and progress logic for more accurate user feedback. --- compute/worker/src/server.ts | 105 +++++++++++++++++++---- src/app/(app)/pdf/[id]/page.tsx | 17 ++-- src/app/(app)/pdf/[id]/usePdfDocument.ts | 2 +- 3 files changed, 93 insertions(+), 31 deletions(-) diff --git a/compute/worker/src/server.ts b/compute/worker/src/server.ts index 10e6298..6800068 100644 --- a/compute/worker/src/server.ts +++ b/compute/worker/src/server.ts @@ -55,6 +55,8 @@ const RUNNING_HEARTBEAT_MS = 5000; const DOCUMENT_ID_REGEX = /^[a-f0-9]{64}$/i; const SAFE_NAMESPACE_REGEX = /^[a-zA-Z0-9._-]{1,128}$/; const WHISPER_MAX_DELIVER = 1; +const PDF_LAYOUT_HARD_CAP_MS = 24 * 60 * 60 * 1000; +const NATS_API_TIMEOUT_MS = 60_000; interface QueuedJob { jobId: string; @@ -212,6 +214,59 @@ async function withTimeout(promise: Promise, timeoutMs: number, label: str } } +async function withIdleTimeoutAndHardCap(input: { + run: (touchProgress: () => void) => Promise; + idleTimeoutMs: number; + hardCapMs: number; + label: string; +}): Promise { + let idleTimer: NodeJS.Timeout | null = null; + let hardCapTimer: NodeJS.Timeout | null = null; + let settled = false; + let rejectTimeout!: (reason: unknown) => void; + + const clearTimers = () => { + if (idleTimer) { + clearTimeout(idleTimer); + idleTimer = null; + } + if (hardCapTimer) { + clearTimeout(hardCapTimer); + hardCapTimer = null; + } + }; + + const failTimeout = (kind: 'idle' | 'hard cap', timeoutMs: number) => { + if (settled) return; + settled = true; + clearTimers(); + rejectTimeout(new Error(`${input.label} ${kind} timed out after ${timeoutMs}ms`)); + }; + + const touchProgress = () => { + if (settled) return; + if (idleTimer) clearTimeout(idleTimer); + idleTimer = setTimeout(() => failTimeout('idle', input.idleTimeoutMs), input.idleTimeoutMs); + }; + + const timeoutPromise = new Promise((_, reject) => { + rejectTimeout = reject; + hardCapTimer = setTimeout(() => failTimeout('hard cap', input.hardCapMs), input.hardCapMs); + touchProgress(); + }); + + try { + const result = await Promise.race([input.run(touchProgress), timeoutPromise]); + settled = true; + clearTimers(); + return result as T; + } catch (error) { + settled = true; + clearTimers(); + throw error; + } +} + function isAuthed(request: FastifyRequest, expectedToken: string): boolean { const auth = request.headers.authorization; if (!auth?.startsWith('Bearer ')) return false; @@ -401,8 +456,8 @@ async function main(): Promise { } const nc: NatsConnection = await connect(connectOpts); - const js: JetStreamClient = jetstream(nc); - const jsm: JetStreamManager = await jetstreamManager(nc); + const js: JetStreamClient = jetstream(nc, { timeout: NATS_API_TIMEOUT_MS }); + const jsm: JetStreamManager = await jetstreamManager(nc, { timeout: NATS_API_TIMEOUT_MS }); await ensureJetStreamResources(jsm, whisperTimeoutMs, pdfTimeoutMs, pdfAttempts, jobsStreamMaxBytes); @@ -710,6 +765,7 @@ async function main(): Promise { reply.raw.setHeader('X-Accel-Buffering', 'no'); const writeSnapshot = (snapshot: WorkerOperationState): void => { + if (closed || reply.raw.writableEnded) return; reply.raw.write(`event: snapshot\ndata: ${JSON.stringify(snapshot)}\n\n`); }; @@ -718,25 +774,34 @@ async function main(): Promise { closed = true; }); - let current = initial; - writeSnapshot(current); - let signature = JSON.stringify(current); + try { + let current = initial; + writeSnapshot(current); + let signature = JSON.stringify(current); - while (!closed && !isTerminalStatus(current.status)) { - await sleep(SSE_POLL_INTERVAL_MS); - const next = await getOpState(params.data.opId); - if (!next) break; - const nextSignature = JSON.stringify(next); - if (nextSignature !== signature) { - current = next; - signature = nextSignature; - writeSnapshot(current); + while (!closed && !isTerminalStatus(current.status)) { + await sleep(SSE_POLL_INTERVAL_MS); + const next = await getOpState(params.data.opId); + if (!next) break; + const nextSignature = JSON.stringify(next); + if (nextSignature !== signature) { + current = next; + signature = nextSignature; + writeSnapshot(current); + } } + } catch (error) { + app.log.warn({ + opId: params.data.opId, + error: toErrorMessage(error), + }, 'op events stream loop error'); } if (!reply.raw.writableEnded) { reply.raw.end(); } + + return reply; }); const whisperConsumer = await js.consumers.get(JOBS_STREAM_NAME, WHISPER_CONSUMER_NAME); @@ -791,11 +856,15 @@ async function main(): Promise { let lastTotalPages = 0; let lastPagesParsed = 0; const computeStartedAt = Date.now(); - const result = await withTimeout( - runPdfLayoutFromPdfBuffer({ + const result = await withIdleTimeoutAndHardCap({ + idleTimeoutMs: Math.max(pdfTimeoutMs, 1_000), + hardCapMs: PDF_LAYOUT_HARD_CAP_MS, + label: 'pdf layout job', + run: async (touchProgress) => runPdfLayoutFromPdfBuffer({ documentId: parsed.documentId, pdfBytes, onPageParsed: async ({ pageNumber, totalPages }) => { + touchProgress(); lastTotalPages = totalPages; lastPagesParsed = pageNumber; if (!hooks?.onProgress) return; @@ -807,9 +876,7 @@ async function main(): Promise { }); }, }), - pdfTimeoutMs, - 'pdf layout job', - ); + }); const computeMs = Date.now() - computeStartedAt; if (hooks?.onProgress && lastTotalPages > 0) { diff --git a/src/app/(app)/pdf/[id]/page.tsx b/src/app/(app)/pdf/[id]/page.tsx index caaccae..5441aea 100644 --- a/src/app/(app)/pdf/[id]/page.tsx +++ b/src/app/(app)/pdf/[id]/page.tsx @@ -75,14 +75,14 @@ export default function PDFViewerPage() { const backNavTimeoutRef = useRef | null>(null); const clearCurrDocRef = useRef(clearCurrDoc); const [isNavigatingBack, setIsNavigatingBack] = useState(false); - const parseUiState: 'unknown' | NonNullable = parseStatus ?? 'unknown'; + const parseUiState: NonNullable = parseStatus ?? 'pending'; const isParseReady = parseUiState === 'ready'; const forceReparseDisabled = isForceReparseDisabled(parseStatus); const hasRealParseProgress = !!parseProgress && parseProgress.totalPages > 0 && parseProgress.pagesParsed >= 0; const shouldShowExpandedParseLoader = !isLoading - && (parseUiState === 'failed' || hasRealParseProgress); + && (parseUiState === 'pending' || parseUiState === 'running' || parseUiState === 'failed' || hasRealParseProgress); useEffect(() => { setIsLoading(true); @@ -286,10 +286,10 @@ export default function PDFViewerPage() { const renderPdfStatusLoader = () => { const compactLabel = isLoading ? 'Opening PDF...' - : (parseUiState === 'ready' ? 'Rendering pages...' : 'Checking parse state...'); + : (parseUiState === 'ready' ? 'Rendering pages...' : 'Preparing PDF layout...'); const compactSubLabel = isLoading ? 'Loading document data' - : (parseUiState === 'ready' ? 'Preparing first frame' : 'Waiting for parse status'); + : (parseUiState === 'ready' ? 'Preparing first frame' : 'Queueing parser and preparing page extraction'); const totalPages = parseProgress?.totalPages ?? 0; const pagesParsed = parseProgress?.pagesParsed ?? 0; @@ -302,10 +302,7 @@ export default function PDFViewerPage() { let statusText = 'Loading PDF...'; let statusSubText = 'Initializing document renderer'; if (!isLoading) { - if (parseUiState === 'unknown') { - statusText = 'Checking parse state...'; - statusSubText = 'Waiting for server status snapshot'; - } else if (parseUiState === 'pending') { + if (parseUiState === 'pending') { statusText = 'Preparing PDF layout...'; statusSubText = parseProgress?.phase === 'merge' ? 'Finalizing stitched block structure' @@ -321,9 +318,7 @@ export default function PDFViewerPage() { } } - const stageLabel = parseUiState === 'unknown' - ? 'Stage: checking' - : parseUiState === 'failed' + const stageLabel = parseUiState === 'failed' ? 'Stage: blocked' : (parseUiState === 'pending' ? 'Stage: prepare' diff --git a/src/app/(app)/pdf/[id]/usePdfDocument.ts b/src/app/(app)/pdf/[id]/usePdfDocument.ts index 4c50459..76f9b22 100644 --- a/src/app/(app)/pdf/[id]/usePdfDocument.ts +++ b/src/app/(app)/pdf/[id]/usePdfDocument.ts @@ -450,7 +450,7 @@ export function usePdfDocument(): PdfDocumentState { setCurrDocName(undefined); setCurrDocData(undefined); setParsedDocument(null); - setParseStatus(null); + setParseStatus('pending'); setParseProgress(null); setDocumentSettings(DEFAULT_DOCUMENT_SETTINGS);