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

feat(pdf,worker): enhance PDF layout parsing UX and add robust timeout controls

Browse source 
Introduce idle and hard cap timeouts for PDF layout parsing in the compute worker,
ensuring long-running jobs are capped and progress is tracked with explicit
timeouts. Refactor NATS JetStream client initialization to use a configurable
API timeout. Update PDF viewer UI to treat parse state as 'pending' by default,
improving clarity during document preparation and parsing. Refine parse loader
labels and progress logic for more accurate user feedback.
This commit is contained in:
Richard R 2026-05-21 13:50:40 -06:00
parent a2c714c2a5
commit 671d5c6adb
3 changed files with 93 additions and 31 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

105
compute/worker/src/server.ts
View file

@ -55,6 +55,8 @@ const RUNNING_HEARTBEAT_MS = 5000;
const DOCUMENT_ID_REGEX = /^[a-f0-9]{64}$/i;
const SAFE_NAMESPACE_REGEX = /^[a-zA-Z0-9._-]{1,128}$/;
const WHISPER_MAX_DELIVER = 1;
const PDF_LAYOUT_HARD_CAP_MS = 24 * 60 * 60 * 1000;
const NATS_API_TIMEOUT_MS = 60_000;
interface QueuedJob<TPayload> {
jobId: string;
@ -212,6 +214,59 @@ async function withTimeout<T>(promise: Promise<T>, timeoutMs: number, label: str
}
}
async function withIdleTimeoutAndHardCap<T>(input: {
run: (touchProgress: () => void) => Promise<T>;
idleTimeoutMs: number;
hardCapMs: number;
label: string;
}): Promise<T> {
let idleTimer: NodeJS.Timeout | null = null;
let hardCapTimer: NodeJS.Timeout | null = null;
let settled = false;
let rejectTimeout!: (reason: unknown) => void;
const clearTimers = () => {
if (idleTimer) {
clearTimeout(idleTimer);
idleTimer = null;
}
if (hardCapTimer) {
clearTimeout(hardCapTimer);
hardCapTimer = null;
}
};
const failTimeout = (kind: 'idle' | 'hard cap', timeoutMs: number) => {
if (settled) return;
settled = true;
clearTimers();
rejectTimeout(new Error(`${input.label} ${kind} timed out after ${timeoutMs}ms`));
};
const touchProgress = () => {
if (settled) return;
if (idleTimer) clearTimeout(idleTimer);
idleTimer = setTimeout(() => failTimeout('idle', input.idleTimeoutMs), input.idleTimeoutMs);
};
const timeoutPromise = new Promise<never>((_, reject) => {
rejectTimeout = reject;
hardCapTimer = setTimeout(() => failTimeout('hard cap', input.hardCapMs), input.hardCapMs);
touchProgress();
});
try {
const result = await Promise.race([input.run(touchProgress), timeoutPromise]);
settled = true;
clearTimers();
return result as T;
} catch (error) {
settled = true;
clearTimers();
throw error;
}
}
function isAuthed(request: FastifyRequest, expectedToken: string): boolean {
const auth = request.headers.authorization;
if (!auth?.startsWith('Bearer ')) return false;
@ -401,8 +456,8 @@ async function main(): Promise<void> {
}
const nc: NatsConnection = await connect(connectOpts);
const js: JetStreamClient = jetstream(nc);
const jsm: JetStreamManager = await jetstreamManager(nc);
const js: JetStreamClient = jetstream(nc, { timeout: NATS_API_TIMEOUT_MS });
const jsm: JetStreamManager = await jetstreamManager(nc, { timeout: NATS_API_TIMEOUT_MS });
await ensureJetStreamResources(jsm, whisperTimeoutMs, pdfTimeoutMs, pdfAttempts, jobsStreamMaxBytes);
@ -710,6 +765,7 @@ async function main(): Promise<void> {
reply.raw.setHeader('X-Accel-Buffering', 'no');
const writeSnapshot = (snapshot: WorkerOperationState<WhisperAlignJobResult | PdfLayoutJobResult>): void => {
if (closed || reply.raw.writableEnded) return;
reply.raw.write(`event: snapshot\ndata: ${JSON.stringify(snapshot)}\n\n`);
};
@ -718,25 +774,34 @@ async function main(): Promise<void> {
closed = true;
});
let current = initial;
writeSnapshot(current);
let signature = JSON.stringify(current);
try {
let current = initial;
writeSnapshot(current);
let signature = JSON.stringify(current);
while (!closed && !isTerminalStatus(current.status)) {
await sleep(SSE_POLL_INTERVAL_MS);
const next = await getOpState(params.data.opId);
if (!next) break;
const nextSignature = JSON.stringify(next);
if (nextSignature !== signature) {
current = next;
signature = nextSignature;
writeSnapshot(current);
while (!closed && !isTerminalStatus(current.status)) {
await sleep(SSE_POLL_INTERVAL_MS);
const next = await getOpState(params.data.opId);
if (!next) break;
const nextSignature = JSON.stringify(next);
if (nextSignature !== signature) {
current = next;
signature = nextSignature;
writeSnapshot(current);
}
}
} catch (error) {
app.log.warn({
opId: params.data.opId,
error: toErrorMessage(error),
}, 'op events stream loop error');
}
if (!reply.raw.writableEnded) {
reply.raw.end();
}
return reply;
});
const whisperConsumer = await js.consumers.get(JOBS_STREAM_NAME, WHISPER_CONSUMER_NAME);
@ -791,11 +856,15 @@ async function main(): Promise<void> {
let lastTotalPages = 0;
let lastPagesParsed = 0;
const computeStartedAt = Date.now();
const result = await withTimeout(
runPdfLayoutFromPdfBuffer({
const result = await withIdleTimeoutAndHardCap({
idleTimeoutMs: Math.max(pdfTimeoutMs, 1_000),
hardCapMs: PDF_LAYOUT_HARD_CAP_MS,
label: 'pdf layout job',
run: async (touchProgress) => runPdfLayoutFromPdfBuffer({
documentId: parsed.documentId,
pdfBytes,
onPageParsed: async ({ pageNumber, totalPages }) => {
touchProgress();
lastTotalPages = totalPages;
lastPagesParsed = pageNumber;
if (!hooks?.onProgress) return;
@ -807,9 +876,7 @@ async function main(): Promise<void> {
});
},
}),
pdfTimeoutMs,
'pdf layout job',
);
});
const computeMs = Date.now() - computeStartedAt;
if (hooks?.onProgress && lastTotalPages > 0) {

17
src/app/(app)/pdf/[id]/page.tsx
View file

@ -75,14 +75,14 @@ export default function PDFViewerPage() {
const backNavTimeoutRef = useRef<ReturnType<typeof setTimeout> | null>(null);
const clearCurrDocRef = useRef(clearCurrDoc);
const [isNavigatingBack, setIsNavigatingBack] = useState(false);
const parseUiState: 'unknown' | NonNullable<typeof parseStatus> = parseStatus ?? 'unknown';
const parseUiState: NonNullable<typeof parseStatus> = parseStatus ?? 'pending';
const isParseReady = parseUiState === 'ready';
const forceReparseDisabled = isForceReparseDisabled(parseStatus);
const hasRealParseProgress = !!parseProgress
&& parseProgress.totalPages > 0
&& parseProgress.pagesParsed >= 0;
const shouldShowExpandedParseLoader = !isLoading
&& (parseUiState === 'failed' || hasRealParseProgress);
&& (parseUiState === 'pending' || parseUiState === 'running' || parseUiState === 'failed' || hasRealParseProgress);
useEffect(() => {
setIsLoading(true);
@ -286,10 +286,10 @@ export default function PDFViewerPage() {
const renderPdfStatusLoader = () => {
const compactLabel = isLoading
? 'Opening PDF...'
: (parseUiState === 'ready' ? 'Rendering pages...' : 'Checking parse state...');
: (parseUiState === 'ready' ? 'Rendering pages...' : 'Preparing PDF layout...');
const compactSubLabel = isLoading
? 'Loading document data'
: (parseUiState === 'ready' ? 'Preparing first frame' : 'Waiting for parse status');
: (parseUiState === 'ready' ? 'Preparing first frame' : 'Queueing parser and preparing page extraction');
const totalPages = parseProgress?.totalPages ?? 0;
const pagesParsed = parseProgress?.pagesParsed ?? 0;
@ -302,10 +302,7 @@ export default function PDFViewerPage() {
let statusText = 'Loading PDF...';
let statusSubText = 'Initializing document renderer';
if (!isLoading) {
if (parseUiState === 'unknown') {
statusText = 'Checking parse state...';
statusSubText = 'Waiting for server status snapshot';
} else if (parseUiState === 'pending') {
if (parseUiState === 'pending') {
statusText = 'Preparing PDF layout...';
statusSubText = parseProgress?.phase === 'merge'
? 'Finalizing stitched block structure'
@ -321,9 +318,7 @@ export default function PDFViewerPage() {
}
}
const stageLabel = parseUiState === 'unknown'
? 'Stage: checking'
: parseUiState === 'failed'
const stageLabel = parseUiState === 'failed'
? 'Stage: blocked'
: (parseUiState === 'pending'
? 'Stage: prepare'

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

@ -450,7 +450,7 @@ export function usePdfDocument(): PdfDocumentState {
setCurrDocName(undefined);
setCurrDocData(undefined);
setParsedDocument(null);
setParseStatus(null);
setParseStatus('pending');
setParseProgress(null);
setDocumentSettings(DEFAULT_DOCUMENT_SETTINGS);