From fe59685454e2d9c4e39e063913ceccecb27ddef2 Mon Sep 17 00:00:00 2001 From: Richard R Date: Fri, 29 May 2026 20:13:59 -0600 Subject: [PATCH] feat(worker): add granular PDF progress reporting and abort error handling Introduce per-page progress events for PDF parsing by adding onPageStarted hooks throughout the compute worker and core PDF pipeline. Add compute/worker/src/pdf-progress.ts to encapsulate progress event builders. Update server logic to emit page start and page parsed events, improving client feedback during long-running PDF jobs. Add src/lib/server/compute/abort-like-error.ts and integrate isAbortLikeError to robustly detect and suppress expected aborts in API event streaming. Update API route to ignore abort-like errors on worker proxy crash. Add unit tests for abort-like error detection and PDF progress event generation. Refactor PDF parse logic to allow empty merged regions without throwing, ensuring downstream flows remain resilient to edge cases. These changes improve observability, error handling, and reliability for document parsing and event streaming. --- compute/core/src/local-runtime.ts | 5 +++ compute/core/src/pdf/parse.ts | 16 ++++++++-- compute/worker/src/pdf-progress.ts | 25 +++++++++++++++ compute/worker/src/server.ts | 30 +++++++++++++---- .../api/documents/[id]/parsed/events/route.ts | 2 ++ src/lib/server/compute/abort-like-error.ts | 9 ++++++ tests/unit/compute-abort-like-error.spec.ts | 18 +++++++++++ .../unit/compute-worker-pdf-progress.spec.ts | 32 +++++++++++++++++++ 8 files changed, 127 insertions(+), 10 deletions(-) create mode 100644 compute/worker/src/pdf-progress.ts create mode 100644 src/lib/server/compute/abort-like-error.ts create mode 100644 tests/unit/compute-abort-like-error.spec.ts create mode 100644 tests/unit/compute-worker-pdf-progress.spec.ts diff --git a/compute/core/src/local-runtime.ts b/compute/core/src/local-runtime.ts index ad57513..604d8d7 100644 --- a/compute/core/src/local-runtime.ts +++ b/compute/core/src/local-runtime.ts @@ -25,6 +25,10 @@ export async function runWhisperAlignmentFromAudioBuffer(input: { export async function runPdfLayoutFromPdfBuffer(input: { documentId: string; pdfBytes: ArrayBuffer; + onPageStarted?: (input: { + pageNumber: number; + totalPages: number; + }) => void | Promise; onPageParsed?: (input: { pageNumber: number; totalPages: number; @@ -34,6 +38,7 @@ export async function runPdfLayoutFromPdfBuffer(input: { const parsed = await parsePdf({ documentId: input.documentId, pdfBytes: input.pdfBytes, + onPageStarted: input.onPageStarted, onPageParsed: input.onPageParsed, }); return { parsed }; diff --git a/compute/core/src/pdf/parse.ts b/compute/core/src/pdf/parse.ts index 6aef27d..a6dbe22 100644 --- a/compute/core/src/pdf/parse.ts +++ b/compute/core/src/pdf/parse.ts @@ -11,6 +11,10 @@ import { normalizeTextItemsForLayout } from './normalize-text'; interface ParsePdfInput { documentId: string; pdfBytes: ArrayBuffer; + onPageStarted?: (input: { + pageNumber: number; + totalPages: number; + }) => void | Promise; onPageParsed?: (input: { pageNumber: number; totalPages: number; @@ -57,6 +61,12 @@ export async function parsePdf(input: ParsePdfInput): Promise for (let pageNumber = 1; pageNumber <= pdf.numPages; pageNumber += 1) { const pageStartedAt = Date.now(); + if (input.onPageStarted) { + await input.onPageStarted({ + pageNumber, + totalPages: pdf.numPages, + }); + } const page = await pdf.getPage(pageNumber); const viewport = page.getViewport({ scale: 1.0 }); const textContent = await page.getTextContent(); @@ -91,9 +101,9 @@ export async function parsePdf(input: ParsePdfInput): Promise pageImage: rendered.image, }); const merged = mergeTextWithRegions(regions, layoutTextItems); - if (textItems.length > 0 && merged.length === 0) { - throw new Error(`layout-merge-empty: page=${pageNumber} regions=${regions.length}`); - } + // Do not fail the full document parse when a page has text but model + // detection/assignment yields no merged regions. Emit an empty page so + // downstream playback/tts flows can naturally skip it. const blocks = merged .map((entry, readingOrder) => ({ diff --git a/compute/worker/src/pdf-progress.ts b/compute/worker/src/pdf-progress.ts new file mode 100644 index 0000000..756a256 --- /dev/null +++ b/compute/worker/src/pdf-progress.ts @@ -0,0 +1,25 @@ +import type { PdfLayoutProgress } from '@openreader/compute-core/api-contracts'; + +export function buildInferProgressForPageStart(input: { + pageNumber: number; + totalPages: number; +}): PdfLayoutProgress { + return { + totalPages: input.totalPages, + pagesParsed: Math.max(0, input.pageNumber - 1), + currentPage: input.pageNumber, + phase: 'infer', + }; +} + +export function buildInferProgressForPageParsed(input: { + pageNumber: number; + totalPages: number; +}): PdfLayoutProgress { + return { + totalPages: input.totalPages, + pagesParsed: input.pageNumber, + currentPage: input.pageNumber, + phase: 'infer', + }; +} diff --git a/compute/worker/src/server.ts b/compute/worker/src/server.ts index 2b9a04b..8c62d03 100644 --- a/compute/worker/src/server.ts +++ b/compute/worker/src/server.ts @@ -56,6 +56,7 @@ import { hashOpKey, } from './control-plane/jetstream'; import { type JsonCodec, createJsonCodec } from './control-plane/json-codec'; +import { buildInferProgressForPageParsed, buildInferProgressForPageStart } from './pdf-progress'; const JOBS_STREAM_NAME = 'compute_jobs'; const WHISPER_JOBS_SUBJECT = 'jobs.whisper'; @@ -874,7 +875,11 @@ async function main(): Promise { const parsed = alignSchema.parse(payload); const s3FetchStartedAt = Date.now(); - const audioBuffer = await readObjectByKey(parsed.audioObjectKey); + const audioBuffer = await withTimeout( + readObjectByKey(parsed.audioObjectKey), + whisperTimeoutMs, + 'whisper s3 fetch', + ); const s3FetchMs = Date.now() - s3FetchStartedAt; const computeStartedAt = Date.now(); @@ -908,7 +913,11 @@ async function main(): Promise { const parsed = layoutSchema.parse(payload); const s3FetchStartedAt = Date.now(); - const pdfBytes = await readObjectByKey(parsed.documentObjectKey); + const pdfBytes = await withTimeout( + readObjectByKey(parsed.documentObjectKey), + Math.max(pdfTimeoutMs, 1_000), + 'pdf s3 fetch', + ); const s3FetchMs = Date.now() - s3FetchStartedAt; let lastTotalPages = 0; @@ -921,17 +930,24 @@ async function main(): Promise { run: async (touchProgress) => runPdfLayoutFromPdfBuffer({ documentId: parsed.documentId, pdfBytes, + onPageStarted: async ({ pageNumber, totalPages }) => { + touchProgress(); + lastTotalPages = totalPages; + if (!hooks?.onProgress) return; + await hooks.onProgress(buildInferProgressForPageStart({ + pageNumber, + totalPages, + })); + }, onPageParsed: async ({ pageNumber, totalPages }) => { touchProgress(); lastTotalPages = totalPages; lastPagesParsed = pageNumber; if (!hooks?.onProgress) return; - await hooks.onProgress({ + await hooks.onProgress(buildInferProgressForPageParsed({ + pageNumber, totalPages, - pagesParsed: pageNumber, - currentPage: pageNumber, - phase: 'infer', - }); + })); }, }), }); diff --git a/src/app/api/documents/[id]/parsed/events/route.ts b/src/app/api/documents/[id]/parsed/events/route.ts index eb7e550..b9091f9 100644 --- a/src/app/api/documents/[id]/parsed/events/route.ts +++ b/src/app/api/documents/[id]/parsed/events/route.ts @@ -4,6 +4,7 @@ import { db } from '@/db'; import { documents } from '@/db/schema'; import { requireAuthContext } from '@/lib/server/auth/auth'; import { getWorkerClientConfigFromEnv } from '@/lib/server/compute/worker'; +import { isAbortLikeError } from '@/lib/server/compute/abort-like-error'; import { snapshotFromWorkerState } from '@/lib/server/compute/worker-parse-state'; import { fetchWorkerOperationState } from '@/lib/server/compute/worker-op-state'; import { isValidDocumentId } from '@/lib/server/documents/blobstore'; @@ -472,6 +473,7 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string void runWorkerProxy() .catch((error) => { + if (closed || isAbortLikeError(error)) return; logServerError(logger, { event: 'documents.parsed.events.worker_proxy_crashed', error, diff --git a/src/lib/server/compute/abort-like-error.ts b/src/lib/server/compute/abort-like-error.ts new file mode 100644 index 0000000..2ce848b --- /dev/null +++ b/src/lib/server/compute/abort-like-error.ts @@ -0,0 +1,9 @@ +export function isAbortLikeError(error: unknown): boolean { + if (!error) return false; + if (error instanceof DOMException) return error.name === 'AbortError'; + if (error instanceof Error) return error.name === 'AbortError' || error.message === 'This operation was aborted'; + if (typeof error === 'object' && error !== null && 'name' in error) { + return (error as { name?: unknown }).name === 'AbortError'; + } + return false; +} diff --git a/tests/unit/compute-abort-like-error.spec.ts b/tests/unit/compute-abort-like-error.spec.ts new file mode 100644 index 0000000..e31287e --- /dev/null +++ b/tests/unit/compute-abort-like-error.spec.ts @@ -0,0 +1,18 @@ +import { expect, test } from '@playwright/test'; +import { isAbortLikeError } from '../../src/lib/server/compute/abort-like-error'; + +test.describe('isAbortLikeError', () => { + test('matches abort-shaped errors', () => { + expect(isAbortLikeError(new DOMException('This operation was aborted', 'AbortError'))).toBe(true); + expect(isAbortLikeError(Object.assign(new Error('random'), { name: 'AbortError' }))).toBe(true); + expect(isAbortLikeError(new Error('This operation was aborted'))).toBe(true); + expect(isAbortLikeError({ name: 'AbortError' })).toBe(true); + }); + + test('does not match non-abort errors', () => { + expect(isAbortLikeError(new Error('boom'))).toBe(false); + expect(isAbortLikeError({ name: 'TypeError' })).toBe(false); + expect(isAbortLikeError(null)).toBe(false); + expect(isAbortLikeError(undefined)).toBe(false); + }); +}); diff --git a/tests/unit/compute-worker-pdf-progress.spec.ts b/tests/unit/compute-worker-pdf-progress.spec.ts new file mode 100644 index 0000000..284fd04 --- /dev/null +++ b/tests/unit/compute-worker-pdf-progress.spec.ts @@ -0,0 +1,32 @@ +import { expect, test } from '@playwright/test'; +import { + buildInferProgressForPageParsed, + buildInferProgressForPageStart, +} from '../../compute/worker/src/pdf-progress'; + +test.describe('compute worker pdf progress helpers', () => { + test('page-start progress keeps current page but does not count it as parsed yet', () => { + expect(buildInferProgressForPageStart({ pageNumber: 1, totalPages: 12 })).toEqual({ + totalPages: 12, + pagesParsed: 0, + currentPage: 1, + phase: 'infer', + }); + + expect(buildInferProgressForPageStart({ pageNumber: 5, totalPages: 12 })).toEqual({ + totalPages: 12, + pagesParsed: 4, + currentPage: 5, + phase: 'infer', + }); + }); + + test('page-parsed progress counts the current page as parsed', () => { + expect(buildInferProgressForPageParsed({ pageNumber: 5, totalPages: 12 })).toEqual({ + totalPages: 12, + pagesParsed: 5, + currentPage: 5, + phase: 'infer', + }); + }); +});