From 46c4be614d0281affabca11ead0dfdc8edcd68b7 Mon Sep 17 00:00:00 2001 From: Richard R Date: Thu, 4 Jun 2026 20:46:02 -0600 Subject: [PATCH] refactor(pdf-parse): extract encodeParserVersion utility and strengthen artifact readiness Move encodeParserVersion to a dedicated module for unified access and remove local duplicates. Update all consumers to import from the new entrypoint. Tighten artifact readiness checks in both client and API route to ensure 'ready' is only reported when the artifact is accessible, with retries and stricter validation. Add tests for edge cases and operation state validation. --- compute/core/src/api-contracts/index.ts | 1 + compute/core/src/index.ts | 1 + compute/core/src/pdf/parser-version-key.ts | 9 ++++ compute/worker/src/runtime.ts | 6 +-- src/app/(app)/pdf/[id]/usePdfDocument.ts | 25 ++++++--- src/app/api/documents/[id]/parsed/route.ts | 13 ++++- src/lib/server/compute/worker-op-state.ts | 13 +++++ src/lib/server/documents/blobstore.ts | 7 +-- src/lib/server/pdf-parse/snapshot.ts | 5 +- ...d-events-route-worker-proxy.vitest.spec.ts | 14 +++++ ...df-parsed-route-worker-flow.vitest.spec.ts | 51 +++++++++++++++++++ 11 files changed, 124 insertions(+), 21 deletions(-) create mode 100644 compute/core/src/pdf/parser-version-key.ts diff --git a/compute/core/src/api-contracts/index.ts b/compute/core/src/api-contracts/index.ts index 0884d9b..487a618 100644 --- a/compute/core/src/api-contracts/index.ts +++ b/compute/core/src/api-contracts/index.ts @@ -18,6 +18,7 @@ export type { export const ALIGN_QUEUE_NAME = 'whisper-align'; export const PDF_LAYOUT_QUEUE_NAME = 'pdf-layout'; export { PDF_PARSER_VERSION } from '../pdf/parser-version'; +export { encodeParserVersion } from '../pdf/parser-version-key'; export interface WhisperAlignJobBase { text: string; diff --git a/compute/core/src/index.ts b/compute/core/src/index.ts index 4360a58..ea7b118 100644 --- a/compute/core/src/index.ts +++ b/compute/core/src/index.ts @@ -17,6 +17,7 @@ export { export { renderPage } from './pdf/render'; export { mergeTextWithRegions } from './pdf/merge'; export { PDF_PARSER_VERSION } from './pdf/parser-version'; +export { encodeParserVersion } from './pdf/parser-version-key'; export { stitchCrossPageBlocks } from './pdf/stitch'; export { normalizeTextItemsForLayout } from './pdf/normalize-text'; export { mapWordsToSentenceOffsets, type WhisperWord } from './whisper/alignment-map'; diff --git a/compute/core/src/pdf/parser-version-key.ts b/compute/core/src/pdf/parser-version-key.ts new file mode 100644 index 0000000..f9f82b2 --- /dev/null +++ b/compute/core/src/pdf/parser-version-key.ts @@ -0,0 +1,9 @@ +import { PDF_PARSER_VERSION } from './parser-version'; + +export function encodeParserVersion( + parserVersion: string, + defaultVersion = PDF_PARSER_VERSION, +): string { + const normalized = parserVersion.trim() || defaultVersion; + return encodeURIComponent(normalized); +} diff --git a/compute/worker/src/runtime.ts b/compute/worker/src/runtime.ts index 7b14f0f..a00121d 100644 --- a/compute/worker/src/runtime.ts +++ b/compute/worker/src/runtime.ts @@ -31,6 +31,7 @@ import { getAvailableCpuCores, getOnnxThreadsPerJob, PDF_PARSER_VERSION, + encodeParserVersion, withIdleTimeoutAndHardCap, withTimeout, } from '@openreader/compute-core'; @@ -245,11 +246,6 @@ function sanitizeNamespace(namespace: string | null): string | null { return SAFE_NAMESPACE_REGEX.test(namespace) ? namespace : null; } -function encodeParserVersion(parserVersion: string): string { - const normalized = parserVersion.trim() || 'unknown-parser'; - return encodeURIComponent(normalized); -} - function documentParsedKey(id: string, namespace: string | null, prefix: string): string { if (!DOCUMENT_ID_REGEX.test(id)) { throw new Error(`Invalid document id: ${id}`); diff --git a/src/app/(app)/pdf/[id]/usePdfDocument.ts b/src/app/(app)/pdf/[id]/usePdfDocument.ts index 9cb6d45..8ec76c4 100644 --- a/src/app/(app)/pdf/[id]/usePdfDocument.ts +++ b/src/app/(app)/pdf/[id]/usePdfDocument.ts @@ -119,6 +119,10 @@ export interface PdfDocumentState { isAudioCombining: boolean; } +function delay(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + /** * Main PDF route hook. */ @@ -231,13 +235,20 @@ export function usePdfDocument(): PdfDocumentState { if (snapshot.parseStatus === 'ready') { isResolvingTerminalState = true; void (async () => { - try { - await loadParsedDocumentOnce(documentId, controller.signal); - } catch (error) { - if (error instanceof DOMException && error.name === 'AbortError') return; - console.error('Failed to load parsed PDF after ready status:', error); - resetParsedDocumentState(); - } finally { + let loaded = false; + let retryMs = 500; + while (!controller.signal.aborted && !loaded) { + try { + await loadParsedDocumentOnce(documentId, controller.signal); + loaded = true; + } catch (error) { + if (error instanceof DOMException && error.name === 'AbortError') return; + console.warn('Parsed PDF reported ready before artifact was readable; retrying:', error); + await delay(retryMs); + retryMs = Math.min(retryMs * 2, 2_000); + } + } + if (loaded) { if (parseSseCloseRef.current === closeSse) { closeSse(); parseSseCloseRef.current = null; diff --git a/src/app/api/documents/[id]/parsed/route.ts b/src/app/api/documents/[id]/parsed/route.ts index e95d538..0234ade 100644 --- a/src/app/api/documents/[id]/parsed/route.ts +++ b/src/app/api/documents/[id]/parsed/route.ts @@ -192,7 +192,18 @@ export async function POST(req: NextRequest, ctx: { params: Promise<{ id: string if (snapshot.parseStatus === 'failed') { return jsonSnapshot(snapshot); } - return jsonSnapshot(snapshot, snapshot.parseStatus === 'ready' ? 200 : 202); + if (snapshot.parseStatus === 'ready') { + const artifactKey = parsedObjectKeyFromWorkerState(currentOp); + if (artifactKey && await readParsedPdfArtifactByKey(artifactKey)) { + return jsonSnapshot(snapshot, 200); + } + return jsonSnapshot({ + parseStatus: 'running', + parseProgress: null, + opId: snapshot.opId, + }, 202); + } + return jsonSnapshot(snapshot, 202); } } diff --git a/src/lib/server/compute/worker-op-state.ts b/src/lib/server/compute/worker-op-state.ts index feddab5..a6debf0 100644 --- a/src/lib/server/compute/worker-op-state.ts +++ b/src/lib/server/compute/worker-op-state.ts @@ -150,6 +150,19 @@ export async function fetchWorkerOperationStateByKey( }); return null; } + if (typeof parsed.opKey !== 'string' || parsed.opKey.trim() !== normalized) { + logDegraded(serverLogger, { + event: 'compute.worker_op_lookup.response.mismatch', + msg: 'Worker op lookup response did not match requested op key', + step: 'validate_worker_op_lookup_response', + context: { + opKey: normalized, + responseOpId: parsed.opId, + responseOpKey: typeof parsed.opKey === 'string' ? parsed.opKey : null, + }, + }); + return null; + } return parsed; } catch (error) { logDegraded(serverLogger, { diff --git a/src/lib/server/documents/blobstore.ts b/src/lib/server/documents/blobstore.ts index a1a862f..732ca0b 100644 --- a/src/lib/server/documents/blobstore.ts +++ b/src/lib/server/documents/blobstore.ts @@ -8,7 +8,7 @@ import { PutObjectCommand, } from '@aws-sdk/client-s3'; import { getSignedUrl } from '@aws-sdk/s3-request-presigner'; -import { PDF_PARSER_VERSION } from '@openreader/compute-core/api-contracts'; +import { PDF_PARSER_VERSION, encodeParserVersion } from '@openreader/compute-core/api-contracts'; import { getS3Client, getS3Config, getS3ProxyClient } from '@/lib/server/storage/s3'; const DOCUMENT_ID_REGEX = /^[a-f0-9]{64}$/i; @@ -106,11 +106,6 @@ export function documentParsedKey(id: string, namespace: string | null): string return documentParsedKeyForVersion(id, namespace, PDF_PARSER_VERSION); } -function encodeParserVersion(parserVersion: string): string { - const normalized = parserVersion.trim() || PDF_PARSER_VERSION; - return encodeURIComponent(normalized); -} - export function documentParsedKeyForVersion( id: string, namespace: string | null, diff --git a/src/lib/server/pdf-parse/snapshot.ts b/src/lib/server/pdf-parse/snapshot.ts index a393c29..49c6f61 100644 --- a/src/lib/server/pdf-parse/snapshot.ts +++ b/src/lib/server/pdf-parse/snapshot.ts @@ -12,9 +12,10 @@ function mapWorkerStatusToParseStatus(status: WorkerOperationState['status']): P return 'ready'; case 'failed': return 'failed'; - default: - return 'pending'; } + + const exhaustive: never = status; + return exhaustive; } export function parsedObjectKeyFromWorkerState( diff --git a/tests/unit/pdf-parsed-events-route-worker-proxy.vitest.spec.ts b/tests/unit/pdf-parsed-events-route-worker-proxy.vitest.spec.ts index 87fd852..98ceb4d 100644 --- a/tests/unit/pdf-parsed-events-route-worker-proxy.vitest.spec.ts +++ b/tests/unit/pdf-parsed-events-route-worker-proxy.vitest.spec.ts @@ -129,6 +129,20 @@ describe('GET /api/documents/[id]/parsed/events worker event proxy', () => { ); }); + test('denies proxying when the op does not belong to the document', async () => { + hoisted.isPdfParseOperationForDocument.mockReturnValue(false); + + const { GET } = await import('../../src/app/api/documents/[id]/parsed/events/route'); + const response = await GET(new NextRequest('http://localhost/api/documents/doc-1/parsed/events?opId=op-1'), { + params: Promise.resolve({ id: 'doc-1' }), + }); + + expect(response.status).toBe(404); + await expect(response.json()).resolves.toMatchObject({ error: 'Operation not found' }); + expect(hoisted.fetchPdfParseOperation).toHaveBeenCalledWith('op-1'); + expect(global.fetch).not.toHaveBeenCalled(); + }); + afterEach(() => { global.fetch = originalFetch; }); diff --git a/tests/unit/pdf-parsed-route-worker-flow.vitest.spec.ts b/tests/unit/pdf-parsed-route-worker-flow.vitest.spec.ts index 13fa85b..7160c92 100644 --- a/tests/unit/pdf-parsed-route-worker-flow.vitest.spec.ts +++ b/tests/unit/pdf-parsed-route-worker-flow.vitest.spec.ts @@ -174,4 +174,55 @@ describe('GET/POST /api/documents/[id]/parsed worker flow', () => { }); expect(hoisted.createOrReuseCurrentPdfParseOperation).toHaveBeenCalled(); }); + + test('POST does not report ready for a succeeded op until its artifact is readable', async () => { + hoisted.lookupCurrentPdfParseOperation.mockResolvedValue({ + opId: 'op-ready-1', + opKey: 'pdf_layout|v1|parser|doc-1||doc-key|', + kind: 'pdf_layout', + jobId: 'job-ready-1', + status: 'succeeded', + queuedAt: Date.now() - 1000, + updatedAt: Date.now(), + result: { parsedObjectKey: 'missing-parsed-key.json' }, + }); + hoisted.readParsedPdfArtifactByKey.mockResolvedValue(null); + + const { POST } = await import('../../src/app/api/documents/[id]/parsed/route'); + const request = new NextRequest('http://localhost/api/documents/doc-1/parsed', { + method: 'POST', + body: JSON.stringify({ replace: false }), + headers: { 'Content-Type': 'application/json' }, + }); + const response = await POST(request, { + params: Promise.resolve({ id: 'doc-1' }), + }); + + expect(response.status).toBe(202); + await expect(response.json()).resolves.toMatchObject({ + parseStatus: 'running', + opId: 'op-ready-1', + }); + expect(hoisted.readParsedPdfArtifactByKey).toHaveBeenCalledWith('missing-parsed-key.json'); + expect(hoisted.createOrReuseCurrentPdfParseOperation).not.toHaveBeenCalled(); + }); + + test('POST returns the rate-limited response without creating a worker op', async () => { + const sentinel = new Response('rate limited', { status: 429 }); + hoisted.checkJobRate.mockResolvedValue({ allowed: false }); + hoisted.buildComputeRateLimitedResponse.mockReturnValue(sentinel); + + const { POST } = await import('../../src/app/api/documents/[id]/parsed/route'); + const request = new NextRequest('http://localhost/api/documents/doc-1/parsed', { + method: 'POST', + body: JSON.stringify({ replace: true }), + headers: { 'Content-Type': 'application/json' }, + }); + const response = await POST(request, { + params: Promise.resolve({ id: 'doc-1' }), + }); + + expect(response).toBe(sentinel); + expect(hoisted.createOrReuseCurrentPdfParseOperation).not.toHaveBeenCalled(); + }); });