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.
This commit is contained in:
Richard R 2026-06-04 20:46:02 -06:00
parent c05a60a228
commit 46c4be614d
11 changed files with 124 additions and 21 deletions

View file

@ -18,6 +18,7 @@ export type {
export const ALIGN_QUEUE_NAME = 'whisper-align'; export const ALIGN_QUEUE_NAME = 'whisper-align';
export const PDF_LAYOUT_QUEUE_NAME = 'pdf-layout'; export const PDF_LAYOUT_QUEUE_NAME = 'pdf-layout';
export { PDF_PARSER_VERSION } from '../pdf/parser-version'; export { PDF_PARSER_VERSION } from '../pdf/parser-version';
export { encodeParserVersion } from '../pdf/parser-version-key';
export interface WhisperAlignJobBase { export interface WhisperAlignJobBase {
text: string; text: string;

View file

@ -17,6 +17,7 @@ export {
export { renderPage } from './pdf/render'; export { renderPage } from './pdf/render';
export { mergeTextWithRegions } from './pdf/merge'; export { mergeTextWithRegions } from './pdf/merge';
export { PDF_PARSER_VERSION } from './pdf/parser-version'; export { PDF_PARSER_VERSION } from './pdf/parser-version';
export { encodeParserVersion } from './pdf/parser-version-key';
export { stitchCrossPageBlocks } from './pdf/stitch'; export { stitchCrossPageBlocks } from './pdf/stitch';
export { normalizeTextItemsForLayout } from './pdf/normalize-text'; export { normalizeTextItemsForLayout } from './pdf/normalize-text';
export { mapWordsToSentenceOffsets, type WhisperWord } from './whisper/alignment-map'; export { mapWordsToSentenceOffsets, type WhisperWord } from './whisper/alignment-map';

View file

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

View file

@ -31,6 +31,7 @@ import {
getAvailableCpuCores, getAvailableCpuCores,
getOnnxThreadsPerJob, getOnnxThreadsPerJob,
PDF_PARSER_VERSION, PDF_PARSER_VERSION,
encodeParserVersion,
withIdleTimeoutAndHardCap, withIdleTimeoutAndHardCap,
withTimeout, withTimeout,
} from '@openreader/compute-core'; } from '@openreader/compute-core';
@ -245,11 +246,6 @@ function sanitizeNamespace(namespace: string | null): string | null {
return SAFE_NAMESPACE_REGEX.test(namespace) ? namespace : 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 { function documentParsedKey(id: string, namespace: string | null, prefix: string): string {
if (!DOCUMENT_ID_REGEX.test(id)) { if (!DOCUMENT_ID_REGEX.test(id)) {
throw new Error(`Invalid document id: ${id}`); throw new Error(`Invalid document id: ${id}`);

View file

@ -119,6 +119,10 @@ export interface PdfDocumentState {
isAudioCombining: boolean; isAudioCombining: boolean;
} }
function delay(ms: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms));
}
/** /**
* Main PDF route hook. * Main PDF route hook.
*/ */
@ -231,13 +235,20 @@ export function usePdfDocument(): PdfDocumentState {
if (snapshot.parseStatus === 'ready') { if (snapshot.parseStatus === 'ready') {
isResolvingTerminalState = true; isResolvingTerminalState = true;
void (async () => { void (async () => {
try { let loaded = false;
await loadParsedDocumentOnce(documentId, controller.signal); let retryMs = 500;
} catch (error) { while (!controller.signal.aborted && !loaded) {
if (error instanceof DOMException && error.name === 'AbortError') return; try {
console.error('Failed to load parsed PDF after ready status:', error); await loadParsedDocumentOnce(documentId, controller.signal);
resetParsedDocumentState(); loaded = true;
} finally { } 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) { if (parseSseCloseRef.current === closeSse) {
closeSse(); closeSse();
parseSseCloseRef.current = null; parseSseCloseRef.current = null;

View file

@ -192,7 +192,18 @@ export async function POST(req: NextRequest, ctx: { params: Promise<{ id: string
if (snapshot.parseStatus === 'failed') { if (snapshot.parseStatus === 'failed') {
return jsonSnapshot(snapshot); 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);
} }
} }

View file

@ -150,6 +150,19 @@ export async function fetchWorkerOperationStateByKey<Result>(
}); });
return null; 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; return parsed;
} catch (error) { } catch (error) {
logDegraded(serverLogger, { logDegraded(serverLogger, {

View file

@ -8,7 +8,7 @@ import {
PutObjectCommand, PutObjectCommand,
} from '@aws-sdk/client-s3'; } from '@aws-sdk/client-s3';
import { getSignedUrl } from '@aws-sdk/s3-request-presigner'; 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'; import { getS3Client, getS3Config, getS3ProxyClient } from '@/lib/server/storage/s3';
const DOCUMENT_ID_REGEX = /^[a-f0-9]{64}$/i; 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); 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( export function documentParsedKeyForVersion(
id: string, id: string,
namespace: string | null, namespace: string | null,

View file

@ -12,9 +12,10 @@ function mapWorkerStatusToParseStatus(status: WorkerOperationState['status']): P
return 'ready'; return 'ready';
case 'failed': case 'failed':
return 'failed'; return 'failed';
default:
return 'pending';
} }
const exhaustive: never = status;
return exhaustive;
} }
export function parsedObjectKeyFromWorkerState( export function parsedObjectKeyFromWorkerState(

View file

@ -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(() => { afterEach(() => {
global.fetch = originalFetch; global.fetch = originalFetch;
}); });

View file

@ -174,4 +174,55 @@ describe('GET/POST /api/documents/[id]/parsed worker flow', () => {
}); });
expect(hoisted.createOrReuseCurrentPdfParseOperation).toHaveBeenCalled(); 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();
});
}); });