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

@@ -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<void>;
   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 };

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<void>;
   onPageParsed?: (input: {
     pageNumber: number;
     totalPages: number;
@@ -57,6 +61,12 @@ export async function parsePdf(input: ParsePdfInput): Promise<ParsedPdfDocument>
 
     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<ParsedPdfDocument>
         pageImage: rendered.image,
       });
       const merged = mergeTextWithRegions(regions, layoutTextItems);
-      if (textItems.length > 0 && merged.length === 0) {
+      // Do not fail the full document parse when a page has text but model
-        throw new Error(`layout-merge-empty: page=${pageNumber} regions=${regions.length}`);
+      // 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) => ({

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',
+  };
+}

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<void> {
     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<void> {
     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<void> {
       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',
-          });
         },
       }),
     });

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,

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;
+}

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);
+  });
+});

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',
+    });
+  });
+});