danvics/openreader
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(pdf): prevent redundant parse jobs and improve docx-to-pdf upload reuse

Browse source 
Avoid enqueuing duplicate parse jobs for PDFs that are already parsed or in terminal states. Update docx-to-pdf upload to reuse existing parsed PDF results when possible, skipping unnecessary parse operations and job enqueuing. Adjust parse-state-backfill to only enqueue jobs for pending or running states. Refine client upload logic to better infer document type from MIME when file names are missing. Enhance usePdfDocument to handle terminal parse states more robustly.

This change ensures idempotent parse job creation and reduces unnecessary worker load, while improving document upload and parse state handling.
This commit is contained in:
Richard R 2026-06-04 12:47:37 -06:00
parent 62c913e96d
commit 7344fcbc51
6 changed files with 90 additions and 47 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

31
src/app/(app)/pdf/[id]/usePdfDocument.ts
View file

@ -213,31 +213,42 @@ export function usePdfDocument(): PdfDocumentState {
setActiveParseOpId(initialOpId?.trim() || null);
const controller = new AbortController();
parseStreamAbortRef.current = controller;
let isResolvingTerminalState = false;
const closeSse = subscribeParsedPdfDocumentEvents(documentId, {
opId: initialOpId?.trim() || null,
}, {
onSnapshot: (snapshot) => {
if (controller.signal.aborted) return;
if (isResolvingTerminalState) return;
if (typeof snapshot.opId === 'string' && snapshot.opId.trim()) {
setActiveParseOpId(snapshot.opId.trim());
}
setParseStatus(snapshot.parseStatus);
setParseProgress(snapshot.parseProgress);
if (snapshot.parseStatus === 'ready') {
closeSse();
parseSseCloseRef.current = null;
if (parseStreamAbortRef.current === controller) {
parseStreamAbortRef.current = null;
}
void 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();
});
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 {
if (parseSseCloseRef.current === closeSse) {
closeSse();
parseSseCloseRef.current = null;
}
if (parseStreamAbortRef.current === controller) {
parseStreamAbortRef.current = null;
}
}
})();
return;
}
if (snapshot.parseStatus === 'failed') {
isResolvingTerminalState = true;
closeSse();
parseSseCloseRef.current = null;
if (parseStreamAbortRef.current === controller) {

46
src/app/api/documents/docx-to-pdf/upload/route.ts
View file

@ -8,6 +8,7 @@ import { pathToFileURL } from 'url';
import { requireAuthContext } from '@/lib/server/auth/auth';
import { db } from '@/db';
import { documents } from '@/db/schema';
import { findReusableParsedPdfResult } from '@/lib/server/documents/parsed-pdf-reuse';
import { safeDocumentName } from '@/lib/server/documents/utils';
import { enqueueDocumentPreview } from '@/lib/server/documents/previews';
import { startPdfParseOperation } from '@/lib/server/documents/pdf-parse-operation';
@ -18,6 +19,7 @@ import { isS3Configured } from '@/lib/server/storage/s3';
import { putDocumentBlob } from '@/lib/server/documents/blobstore';
import { errorToLog, serverLogger } from '@/lib/server/logger';
import { errorResponse } from '@/lib/server/errors/next-response';
import { PDF_PARSER_VERSION } from '@openreader/compute-core';
const DOCSTORE_DIR = path.join(process.cwd(), 'docstore');
const TEMP_DIR = path.join(DOCSTORE_DIR, 'tmp');
@ -130,11 +132,19 @@ export async function POST(req: NextRequest) {
const derivedName = safeDocumentName(`${path.parse(file.name).name}.pdf`, `${id}.pdf`);
const lastModified = Number.isFinite(file.lastModified) ? file.lastModified : Date.now();
const startedParse = await startPdfParseOperation({
documentId: id,
userId: storageUserId,
namespace: testNamespace,
});
const reusableParsedPdf = await findReusableParsedPdfResult(id);
const startedParse = reusableParsedPdf
? null
: await startPdfParseOperation({
documentId: id,
userId: storageUserId,
namespace: testNamespace,
});
const parseState = stringifyDocumentParseState(
reusableParsedPdf
? { status: 'ready', progress: null, updatedAt: Date.now(), parserVersion: PDF_PARSER_VERSION }
: startedParse!.parseState,
);
await db
.insert(documents)
@ -146,8 +156,8 @@ export async function POST(req: NextRequest) {
size: pdfContent.length,
lastModified,
filePath: id,
parseState: stringifyDocumentParseState(startedParse.parseState),
parsedJsonKey: null,
parseState,
parsedJsonKey: reusableParsedPdf?.parsedJsonKey ?? null,
})
.onConflictDoUpdate({
target: [documents.id, documents.userId],
@ -157,8 +167,8 @@ export async function POST(req: NextRequest) {
size: pdfContent.length,
lastModified,
filePath: id,
parseState: stringifyDocumentParseState(startedParse.parseState),
parsedJsonKey: null,
parseState,
parsedJsonKey: reusableParsedPdf?.parsedJsonKey ?? null,
},
});
@ -179,14 +189,16 @@ export async function POST(req: NextRequest) {
}, 'Failed to enqueue preview for converted DOCX');
});
enqueueParsePdfJob({
documentId: id,
userId: storageUserId,
namespace: testNamespace,
initialOpId: startedParse.workerState.opId,
initialJobId: startedParse.workerState.jobId,
initialStatus: startedParse.parseState.status === 'running' ? 'running' : 'pending',
});
if (startedParse) {
enqueueParsePdfJob({
documentId: id,
userId: storageUserId,
namespace: testNamespace,
initialOpId: startedParse.workerState.opId,
initialJobId: startedParse.workerState.jobId,
initialStatus: startedParse.parseState.status === 'running' ? 'running' : 'pending',
});
}
return NextResponse.json({
stored: {

19
src/lib/client/api/documents.ts
View file

@ -47,6 +47,14 @@ function documentTypeForName(name: string): DocumentType {
return 'html';
}
function documentTypeForMime(contentType: string): DocumentType | null {
const normalized = contentType.trim().toLowerCase();
if (normalized === 'application/pdf') return 'pdf';
if (normalized === 'application/epub+zip') return 'epub';
if (normalized === 'application/vnd.openxmlformats-officedocument.wordprocessingml.document') return 'docx';
return null;
}
export function mimeTypeForDoc(doc: Pick<BaseDocument, 'type' | 'name'>): string {
if (doc.type === 'pdf') return 'application/pdf';
if (doc.type === 'epub') return 'application/epub+zip';
@ -314,11 +322,14 @@ export async function uploadDocuments(files: File[], options?: UploadOptions): P
const sources: UploadSource[] = [];
for (const file of files) {
const type = documentTypeForName(file.name);
const name = file.name || `upload.${type}`;
const contentType = file.type || mimeTypeForDoc({ name, type });
const name = file.name || '';
const type = name
? documentTypeForName(name)
: (documentTypeForMime(file.type) ?? 'html');
const resolvedName = name || `upload.${type}`;
const contentType = file.type || mimeTypeForDoc({ name: resolvedName, type });
sources.push({
name,
name: resolvedName,
type,
size: file.size,
lastModified: Number.isFinite(file.lastModified) ? file.lastModified : Date.now(),

2
src/lib/server/compute/worker-parse-state.ts
View file

@ -68,7 +68,7 @@ export function mergeNonReadyParseSnapshot(input: {
workerState: WorkerOperationState<PdfLayoutJobResult>;
}): { parseStatus: PdfParseStatus; parseProgress: PdfParseProgress | null } {
const workerSnapshot = snapshotFromWorkerState(input.workerState);
if (workerSnapshot.parseStatus === 'ready' || workerSnapshot.parseStatus === 'failed') {
if (workerSnapshot.parseStatus === 'ready') {
return {
parseStatus: input.parseStatus,
parseProgress: input.parseProgress,

18
src/lib/server/documents/parse-state-backfill.ts
View file

@ -23,13 +23,15 @@ export async function backfillPendingPdfParseOperation(input: {
userId: input.userId,
namespace: input.namespace,
});
enqueueParsePdfJob({
documentId: input.documentId,
userId: input.userId,
namespace: input.namespace,
initialOpId: startedParse.workerState.opId,
initialJobId: startedParse.workerState.jobId,
initialStatus: startedParse.parseState.status === 'running' ? 'running' : 'pending',
});
if (startedParse.parseState.status === 'pending' || startedParse.parseState.status === 'running') {
enqueueParsePdfJob({
documentId: input.documentId,
userId: input.userId,
namespace: input.namespace,
initialOpId: startedParse.workerState.opId,
initialJobId: startedParse.workerState.jobId,
initialStatus: startedParse.parseState.status,
});
}
return startedParse.workerState;
}

21
tests/unit/pdf-parsed-route-legacy-backfill.vitest.spec.ts
View file

@ -14,8 +14,9 @@ const hoisted = vi.hoisted(() => ({
},
requireAuthContext: vi.fn(),
fetchWorkerOperationState: vi.fn(),
backfillPendingPdfParseOperation: vi.fn(),
healStaleDocumentParseState: vi.fn(async ({ state }) => state),
startPdfParseOperation: vi.fn(),
enqueueParsePdfJob: vi.fn(),
}));
vi.mock('@/db', () => ({
@ -32,14 +33,18 @@ vi.mock('@/lib/server/compute/worker-op-state', () => ({
fetchWorkerOperationState: hoisted.fetchWorkerOperationState,
}));
vi.mock('@/lib/server/documents/parse-state-backfill', () => ({
backfillPendingPdfParseOperation: hoisted.backfillPendingPdfParseOperation,
}));
vi.mock('@/lib/server/documents/parse-state-healing', () => ({
healStaleDocumentParseState: hoisted.healStaleDocumentParseState,
}));
vi.mock('@/lib/server/documents/pdf-parse-operation', () => ({
startPdfParseOperation: hoisted.startPdfParseOperation,
}));
vi.mock('@/lib/server/jobs/user-pdf-layout-job', () => ({
enqueueParsePdfJob: hoisted.enqueueParsePdfJob,
}));
vi.mock('@/lib/server/documents/blobstore', () => ({
documentKey: vi.fn(),
getParsedDocumentBlob: vi.fn(),
@ -104,8 +109,9 @@ describe('GET /api/documents/[id]/parsed pure data fetch', () => {
hoisted.requireAuthContext.mockResolvedValue({ userId: 'user-1' });
hoisted.fetchWorkerOperationState.mockReset();
hoisted.fetchWorkerOperationState.mockResolvedValue(null);
hoisted.backfillPendingPdfParseOperation.mockReset();
hoisted.healStaleDocumentParseState.mockClear();
hoisted.startPdfParseOperation.mockReset();
hoisted.enqueueParsePdfJob.mockReset();
});
test('returns non-ready status without creating a worker op for legacy pending PDFs without opId', async () => {
@ -120,7 +126,8 @@ describe('GET /api/documents/[id]/parsed pure data fetch', () => {
parseStatus: 'pending',
opId: null,
});
expect(hoisted.backfillPendingPdfParseOperation).not.toHaveBeenCalled();
expect(hoisted.startPdfParseOperation).not.toHaveBeenCalled();
expect(hoisted.enqueueParsePdfJob).not.toHaveBeenCalled();
expect(hoisted.row.parseState).toBeNull();
});
});