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