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

hard-cut batch B: worker-proxy parsed SSE and remove app progress bus

Browse source
This commit is contained in:
Richard R 2026-05-26 15:58:14 -06:00
parent f5eb593554
commit f77ee8cfd4
8 changed files with 22 additions and 353 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

97
src/app/api/documents/[id]/parsed/events/route.ts
View file

@ -3,10 +3,7 @@ import { and, eq, inArray } from 'drizzle-orm';
import { db } from '@/db'; import { db } from '@/db';
import { documents } from '@/db/schema'; import { documents } from '@/db/schema';
import { requireAuthContext } from '@/lib/server/auth/auth'; import { requireAuthContext } from '@/lib/server/auth/auth';
import { readComputeMode } from '@/lib/server/compute/mode';
import { getWorkerClientConfigFromEnv } from '@/lib/server/compute/worker'; import { getWorkerClientConfigFromEnv } from '@/lib/server/compute/worker';
import { getParseProgressBus } from '@/lib/server/documents/parse-progress-bus';
import { hashUserId } from '@/lib/server/documents/parse-progress-events';
import { isValidDocumentId } from '@/lib/server/documents/blobstore'; import { isValidDocumentId } from '@/lib/server/documents/blobstore';
import { normalizeParseStatus, parseDocumentParseState } from '@/lib/server/documents/parse-state'; import { normalizeParseStatus, parseDocumentParseState } from '@/lib/server/documents/parse-state';
import { healStaleDocumentParseState } from '@/lib/server/documents/parse-state-healing'; import { healStaleDocumentParseState } from '@/lib/server/documents/parse-state-healing';
@ -161,22 +158,16 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
} }
const initialState = await toSnapshotState(row); const initialState = await toSnapshotState(row);
const computeMode = readComputeMode(); const workerCfg = getWorkerClientConfigFromEnv();
const bus = computeMode === 'local' ? await getParseProgressBus() : null;
const workerCfg = computeMode === 'worker' ? getWorkerClientConfigFromEnv() : null;
const encoder = new TextEncoder(); const encoder = new TextEncoder();
const stream = new ReadableStream<Uint8Array>({ const stream = new ReadableStream<Uint8Array>({
start(controller) { start(controller) {
let closed = false; let closed = false;
let unsubscribe: (() => void) | null = null;
let keepaliveTimer: ReturnType<typeof setInterval> | null = null; let keepaliveTimer: ReturnType<typeof setInterval> | null = null;
let resyncTimer: ReturnType<typeof setInterval> | null = null; let resyncTimer: ReturnType<typeof setInterval> | null = null;
let workerAbort: AbortController | null = null; let workerAbort: AbortController | null = null;
const allowedUserHashes = new Set(allowedUserIds.map((candidate) => hashUserId(candidate)));
const writeSnapshot = (snapshot: ParsedSnapshot): void => { const writeSnapshot = (snapshot: ParsedSnapshot): void => {
controller.enqueue(encoder.encode(`event: snapshot\ndata: ${JSON.stringify(snapshot)}\n\n`)); controller.enqueue(encoder.encode(`event: snapshot\ndata: ${JSON.stringify(snapshot)}\n\n`));
}; };
@ -192,10 +183,6 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
clearInterval(resyncTimer); clearInterval(resyncTimer);
resyncTimer = null; resyncTimer = null;
} }
if (unsubscribe) {
unsubscribe();
unsubscribe = null;
}
if (workerAbort) { if (workerAbort) {
workerAbort.abort(); workerAbort.abort();
workerAbort = null; workerAbort = null;
@ -207,85 +194,7 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
} }
}; };
const runLocalRealtime = async () => {
let current = initialState.snapshot;
let signature = JSON.stringify(current);
writeSnapshot(current);
if (current.parseStatus === 'ready' || current.parseStatus === 'failed') {
closeStream();
return;
}
keepaliveTimer = setInterval(() => {
if (closed) return;
controller.enqueue(encoder.encode(': keepalive\n\n'));
}, SSE_KEEPALIVE_MS);
resyncTimer = setInterval(() => {
if (closed) return;
void syncFromDb({
documentId: id,
storageUserId,
allowedUserIds,
signature,
writeSnapshot,
}).then((next) => {
if (closed) return;
if (!next) {
closeStream();
return;
}
current = next.snapshot;
signature = next.signature;
if (current.parseStatus === 'ready' || current.parseStatus === 'failed') {
closeStream();
}
}).catch((error) => {
if (closed) return;
controller.enqueue(encoder.encode(`event: error\ndata: ${JSON.stringify({ error: String(error) })}\n\n`));
closeStream();
});
}, SSE_RESYNC_INTERVAL_MS);
if (!bus) throw new Error('Local parse progress bus unavailable');
const nextUnsubscribe = await bus.subscribe({
documentId: id,
userIdHashes: allowedUserHashes,
onEvent: async () => {
const next = await syncFromDb({
documentId: id,
storageUserId,
allowedUserIds,
signature,
writeSnapshot,
});
if (!next) {
closeStream();
return;
}
current = next.snapshot;
signature = next.signature;
if (current.parseStatus === 'ready' || current.parseStatus === 'failed') {
closeStream();
}
},
onError: (error) => {
if (closed) return;
controller.enqueue(encoder.encode(`event: error\ndata: ${JSON.stringify({ error: String(error) })}\n\n`));
closeStream();
},
});
if (closed) {
nextUnsubscribe();
return;
}
unsubscribe = nextUnsubscribe;
};
const runWorkerProxy = async () => { const runWorkerProxy = async () => {
if (!workerCfg) throw new Error('Worker mode selected but worker config is unavailable');
let current = initialState.snapshot; let current = initialState.snapshot;
let signature = JSON.stringify(current); let signature = JSON.stringify(current);
let currentOpId = initialState.opId; let currentOpId = initialState.opId;
@ -480,9 +389,7 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
} }
}; };
const run = computeMode === 'local' ? runLocalRealtime : runWorkerProxy; void runWorkerProxy()
void run()
.catch((error) => { .catch((error) => {
if (!closed) { if (!closed) {
controller.enqueue(encoder.encode(`event: error\ndata: ${JSON.stringify({ error: String(error) })}\n\n`)); controller.enqueue(encoder.encode(`event: error\ndata: ${JSON.stringify({ error: String(error) })}\n\n`));

36
src/lib/server/documents/parse-progress-bus.memory.ts
View file

@ -1,36 +0,0 @@
import { EventEmitter } from 'node:events';
import type { ParseProgressBus, ParseProgressBusSubscribeInput } from '@/lib/server/documents/parse-progress-bus.types';
import type { ParseProgressEvent } from '@/lib/server/documents/parse-progress-events';
type Listener = (event: ParseProgressEvent) => void;
function topicForDocument(documentId: string): string {
return `parse-progress.${documentId}`;
}
export class InMemoryParseProgressBus implements ParseProgressBus {
readonly kind = 'memory' as const;
private readonly emitter = new EventEmitter();
constructor() {
this.emitter.setMaxListeners(0);
}
async publish(event: ParseProgressEvent): Promise<void> {
this.emitter.emit(topicForDocument(event.documentId), event);
}
async subscribe(input: ParseProgressBusSubscribeInput): Promise<() => void> {
const listener: Listener = (event) => {
if (!input.userIdHashes.has(event.userIdHash)) return;
Promise.resolve(input.onEvent(event)).catch((error) => {
input.onError?.(error);
});
};
const topic = topicForDocument(input.documentId);
this.emitter.on(topic, listener);
return () => {
this.emitter.off(topic, listener);
};
}
}

16
src/lib/server/documents/parse-progress-bus.ts
View file

@ -1,16 +0,0 @@
import { InMemoryParseProgressBus } from '@/lib/server/documents/parse-progress-bus.memory';
import type { ParseProgressBus } from '@/lib/server/documents/parse-progress-bus.types';
let busPromise: Promise<ParseProgressBus> | null = null;
export async function getParseProgressBus(): Promise<ParseProgressBus> {
if (busPromise) return busPromise;
busPromise = Promise.resolve(new InMemoryParseProgressBus());
return busPromise;
}
export function __resetParseProgressBusForTests(): void {
busPromise = null;
}
export type { ParseProgressBus, ParseProgressBusSubscribeInput } from '@/lib/server/documents/parse-progress-bus.types';

14
src/lib/server/documents/parse-progress-bus.types.ts
View file

@ -1,14 +0,0 @@
import type { ParseProgressEvent } from '@/lib/server/documents/parse-progress-events';
export interface ParseProgressBusSubscribeInput {
documentId: string;
userIdHashes: Set<string>;
onEvent: (event: ParseProgressEvent) => void | Promise<void>;
onError?: (error: unknown) => void;
}
export interface ParseProgressBus {
readonly kind: 'memory';
publish(event: ParseProgressEvent): Promise<void>;
subscribe(input: ParseProgressBusSubscribeInput): Promise<() => void>;
}

18
src/lib/server/documents/parse-progress-events.ts
View file

@ -1,18 +0,0 @@
import { createHash } from 'node:crypto';
import type { PdfParseProgress, PdfParseStatus } from '@/types/parsed-pdf';
export interface ParseProgressEvent {
version: 1;
documentId: string;
userIdHash: string;
parseStatus: PdfParseStatus;
parseProgress: PdfParseProgress | null;
updatedAt: number;
opId?: string;
jobId?: string;
error?: string | null;
}
export function hashUserId(userId: string): string {
return createHash('sha256').update(userId).digest('hex').slice(0, 24);
}

23
src/lib/server/documents/parse-state-healing.ts
View file

@ -8,8 +8,6 @@ import {
stringifyDocumentParseState, stringifyDocumentParseState,
type DocumentParseState, type DocumentParseState,
} from '@/lib/server/documents/parse-state'; } from '@/lib/server/documents/parse-state';
import { getParseProgressBus } from '@/lib/server/documents/parse-progress-bus';
import { hashUserId } from '@/lib/server/documents/parse-progress-events';
export async function healStaleDocumentParseState(input: { export async function healStaleDocumentParseState(input: {
documentId: string; documentId: string;
@ -31,26 +29,5 @@ export async function healStaleDocumentParseState(input: {
.set({ parseState: stringifyDocumentParseState(nextState) }) .set({ parseState: stringifyDocumentParseState(nextState) })
.where(and(eq(documents.id, input.documentId), eq(documents.userId, input.userId))); .where(and(eq(documents.id, input.documentId), eq(documents.userId, input.userId)));
try {
const bus = await getParseProgressBus();
await bus.publish({
version: 1,
documentId: input.documentId,
userIdHash: hashUserId(input.userId),
parseStatus: nextState.status,
parseProgress: nextState.progress ?? null,
updatedAt: nextState.updatedAt ?? Date.now(),
...(nextState.opId ? { opId: nextState.opId } : {}),
...(nextState.jobId ? { jobId: nextState.jobId } : {}),
...(nextState.error ? { error: nextState.error } : {}),
});
} catch (error) {
console.warn('[parse-state-healing] failed to publish stale state event', {
documentId: input.documentId,
userId: input.userId,
error: error instanceof Error ? error.message : String(error),
});
}
return nextState; return nextState;
} }

114
src/lib/server/jobs/user-pdf-layout-job.ts
View file

@ -8,8 +8,6 @@ import {
stringifyDocumentParseState, stringifyDocumentParseState,
type DocumentParseState, type DocumentParseState,
} from '@/lib/server/documents/parse-state'; } from '@/lib/server/documents/parse-state';
import { getParseProgressBus } from '@/lib/server/documents/parse-progress-bus';
import { hashUserId } from '@/lib/server/documents/parse-progress-events';
import { getCompute } from '@/lib/server/compute'; import { getCompute } from '@/lib/server/compute';
import { clearTtsSegmentCache } from '@/lib/server/tts/segments-cache'; import { clearTtsSegmentCache } from '@/lib/server/tts/segments-cache';
import { UNCLAIMED_USER_ID } from '@/lib/server/storage/docstore-legacy'; import { UNCLAIMED_USER_ID } from '@/lib/server/storage/docstore-legacy';
@ -29,7 +27,7 @@ const running = new Set<string>();
const FOLLOWER_WAIT_TIMEOUT_MS = 180_000; const FOLLOWER_WAIT_TIMEOUT_MS = 180_000;
const FOLLOWER_POLL_MS = 1_200; const FOLLOWER_POLL_MS = 1_200;
const WORKER_PROGRESS_DB_THROTTLE_MS = 10_000; const PROGRESS_DB_THROTTLE_MS = 10_000;
type ParseRow = { type ParseRow = {
userId: string; userId: string;
@ -122,41 +120,6 @@ async function updateParseStateForUsers(input: {
); );
} }
function normalizeEventState(state: DocumentParseState): DocumentParseState {
return parseDocumentParseState(stringifyDocumentParseState(state));
}
async function emitParseStateEvents(input: {
documentId: string;
userIds: string[];
state: DocumentParseState;
}): Promise<void> {
if (input.userIds.length === 0) return;
const normalized = normalizeEventState(input.state);
const bus = await getParseProgressBus();
const publishResults = await Promise.allSettled(input.userIds.map(async (userId) => bus.publish({
version: 1,
documentId: input.documentId,
userIdHash: hashUserId(userId),
parseStatus: normalized.status,
parseProgress: normalized.progress ?? null,
updatedAt: normalized.updatedAt ?? Date.now(),
...(normalized.opId ? { opId: normalized.opId } : {}),
...(normalized.jobId ? { jobId: normalized.jobId } : {}),
...(normalized.error ? { error: normalized.error } : {}),
})));
publishResults.forEach((result, index) => {
if (result.status === 'fulfilled') return;
console.warn('[parsePdfJob] failed to publish parse progress event', {
documentId: input.documentId,
userId: input.userIds[index],
error: result.reason instanceof Error ? result.reason.message : String(result.reason),
});
});
}
async function syncCallerToSharedResult(input: UserPdfLayoutJobRequest): Promise<void> { async function syncCallerToSharedResult(input: UserPdfLayoutJobRequest): Promise<void> {
const deadline = Date.now() + FOLLOWER_WAIT_TIMEOUT_MS; const deadline = Date.now() + FOLLOWER_WAIT_TIMEOUT_MS;
@ -175,11 +138,6 @@ async function syncCallerToSharedResult(input: UserPdfLayoutJobRequest): Promise
parseState: stringifyDocumentParseState(readyState), parseState: stringifyDocumentParseState(readyState),
parsedJsonKey: ready.parsedJsonKey, parsedJsonKey: ready.parsedJsonKey,
}); });
await emitParseStateEvents({
documentId: input.documentId,
userIds: [input.userId],
state: readyState,
});
return; return;
} }
@ -198,11 +156,6 @@ async function syncCallerToSharedResult(input: UserPdfLayoutJobRequest): Promise
userIds: [input.userId], userIds: [input.userId],
parseState: stringifyDocumentParseState(nextFailedState), parseState: stringifyDocumentParseState(nextFailedState),
}); });
await emitParseStateEvents({
documentId: input.documentId,
userIds: [input.userId],
state: nextFailedState,
});
return; return;
} }
@ -240,11 +193,6 @@ export async function parsePdfJob(input: UserPdfLayoutJobRequest): Promise<void>
parseState: stringifyDocumentParseState(readyState), parseState: stringifyDocumentParseState(readyState),
parsedJsonKey: ready.parsedJsonKey, parsedJsonKey: ready.parsedJsonKey,
}); });
await emitParseStateEvents({
documentId: input.documentId,
userIds: [input.userId],
state: readyState,
});
return; return;
} }
} }
@ -289,21 +237,15 @@ export async function parsePdfJob(input: UserPdfLayoutJobRequest): Promise<void>
userIds: cohortUserIds, userIds: cohortUserIds,
parseState: runningState, parseState: runningState,
}); });
await emitParseStateEvents({
documentId: input.documentId,
userIds: cohortUserIds,
state: runningStateData,
});
const compute = await getCompute(); const compute = await getCompute();
const isWorkerCompute = compute.mode === 'worker';
let activeOpId: string = claimOpId; let activeOpId: string = claimOpId;
let activeJobId: string | undefined; let activeJobId: string | undefined;
let lastWorkerProgressWriteAt = 0; let lastProgressWriteAt = 0;
let lastWorkerSnapshotWriteAt = 0; let lastSnapshotWriteAt = 0;
let lastWorkerSnapshotStatus: 'pending' | 'running' | null = null; let lastSnapshotStatus: 'pending' | 'running' | null = null;
let lastWorkerSnapshotOpId: string = claimOpId; let lastSnapshotOpId: string = claimOpId;
let lastWorkerSnapshotJobId: string | undefined; let lastSnapshotJobId: string | undefined;
const persistRunningState = async (state: DocumentParseState): Promise<void> => { const persistRunningState = async (state: DocumentParseState): Promise<void> => {
await updateParseStateForUsers({ await updateParseStateForUsers({
@ -311,15 +253,9 @@ export async function parsePdfJob(input: UserPdfLayoutJobRequest): Promise<void>
userIds: cohortUserIds, userIds: cohortUserIds,
parseState: stringifyDocumentParseState(state), parseState: stringifyDocumentParseState(state),
}); });
await emitParseStateEvents({
documentId: input.documentId,
userIds: cohortUserIds,
state,
});
}; };
const onWorkerSnapshot = async (snapshot: WorkerOperationState<PdfLayoutJobResult>): Promise<void> => { const onWorkerSnapshot = async (snapshot: WorkerOperationState<PdfLayoutJobResult>): Promise<void> => {
if (!isWorkerCompute) return;
if (snapshot.opId) activeOpId = snapshot.opId; if (snapshot.opId) activeOpId = snapshot.opId;
if (snapshot.jobId) activeJobId = snapshot.jobId; if (snapshot.jobId) activeJobId = snapshot.jobId;
@ -332,11 +268,11 @@ export async function parsePdfJob(input: UserPdfLayoutJobRequest): Promise<void>
const now = Date.now(); const now = Date.now();
const forceWrite = ( const forceWrite = (
mappedStatus !== lastWorkerSnapshotStatus mappedStatus !== lastSnapshotStatus
|| activeOpId !== lastWorkerSnapshotOpId || activeOpId !== lastSnapshotOpId
|| activeJobId !== lastWorkerSnapshotJobId || activeJobId !== lastSnapshotJobId
); );
if (!forceWrite && (now - lastWorkerSnapshotWriteAt) < WORKER_PROGRESS_DB_THROTTLE_MS) { if (!forceWrite && (now - lastSnapshotWriteAt) < PROGRESS_DB_THROTTLE_MS) {
return; return;
} }
@ -348,20 +284,19 @@ export async function parsePdfJob(input: UserPdfLayoutJobRequest): Promise<void>
...(activeJobId ? { jobId: activeJobId } : {}), ...(activeJobId ? { jobId: activeJobId } : {}),
}; };
await persistRunningState(nextState); await persistRunningState(nextState);
lastWorkerSnapshotWriteAt = now; lastSnapshotWriteAt = now;
lastWorkerSnapshotStatus = mappedStatus; lastSnapshotStatus = mappedStatus;
lastWorkerSnapshotOpId = activeOpId; lastSnapshotOpId = activeOpId;
lastWorkerSnapshotJobId = activeJobId; lastSnapshotJobId = activeJobId;
}; };
const writeProgress = async (progress: PdfLayoutProgress): Promise<void> => { const writeProgress = async (progress: PdfLayoutProgress): Promise<void> => {
if (isWorkerCompute) { const now = Date.now();
const now = Date.now(); if ((now - lastProgressWriteAt) < PROGRESS_DB_THROTTLE_MS && progress.pagesParsed < progress.totalPages) {
if ((now - lastWorkerProgressWriteAt) < WORKER_PROGRESS_DB_THROTTLE_MS && progress.pagesParsed < progress.totalPages) { return;
return;
}
lastWorkerProgressWriteAt = now;
} }
lastProgressWriteAt = now;
const runningProgressState: DocumentParseState = { const runningProgressState: DocumentParseState = {
status: 'running', status: 'running',
progress: { progress: {
@ -376,6 +311,7 @@ export async function parsePdfJob(input: UserPdfLayoutJobRequest): Promise<void>
}; };
await persistRunningState(runningProgressState); await persistRunningState(runningProgressState);
}; };
const layout = await compute.parsePdfLayout({ const layout = await compute.parsePdfLayout({
documentId: input.documentId, documentId: input.documentId,
namespace: input.namespace, namespace: input.namespace,
@ -406,11 +342,6 @@ export async function parsePdfJob(input: UserPdfLayoutJobRequest): Promise<void>
parseState: stringifyDocumentParseState(readyState), parseState: stringifyDocumentParseState(readyState),
parsedJsonKey, parsedJsonKey,
}); });
await emitParseStateEvents({
documentId: input.documentId,
userIds: readyUserIds,
state: readyState,
});
// Best-effort cache invalidation should not block parse readiness. // Best-effort cache invalidation should not block parse readiness.
for (const userId of readyUserIds) { for (const userId of readyUserIds) {
@ -454,11 +385,6 @@ export async function parsePdfJob(input: UserPdfLayoutJobRequest): Promise<void>
userIds: failedUserIds, userIds: failedUserIds,
parseState: stringifyDocumentParseState(failedState), parseState: stringifyDocumentParseState(failedState),
}); });
await emitParseStateEvents({
documentId: input.documentId,
userIds: failedUserIds,
state: failedState,
});
} catch (statusError) { } catch (statusError) {
console.error('[parsePdfJob] failed to write parse status', { console.error('[parsePdfJob] failed to write parse status', {
documentId: input.documentId, documentId: input.documentId,

57
tests/unit/parse-progress-bus.spec.ts
View file

@ -1,57 +0,0 @@
import { expect, test } from '@playwright/test';
import { __resetParseProgressBusForTests, getParseProgressBus } from '../../src/lib/server/documents/parse-progress-bus';
import { InMemoryParseProgressBus } from '../../src/lib/server/documents/parse-progress-bus.memory';
import { hashUserId, type ParseProgressEvent } from '../../src/lib/server/documents/parse-progress-events';
test.describe('parse progress bus', () => {
test.afterEach(() => {
__resetParseProgressBusForTests();
});
test('uses in-memory singleton bus', async () => {
const busA = await getParseProgressBus();
const busB = await getParseProgressBus();
expect(busA.kind).toBe('memory');
expect(busA).toBe(busB);
});
test('in-memory bus publishes only to subscribed user hashes', async () => {
const bus = new InMemoryParseProgressBus();
const documentId = 'd'.repeat(64);
const wantedHash = hashUserId('user-a');
const otherHash = hashUserId('user-b');
const received: ParseProgressEvent[] = [];
const close = await bus.subscribe({
documentId,
userIdHashes: new Set([wantedHash]),
onEvent: (event) => {
received.push(event);
},
});
await bus.publish({
version: 1,
documentId,
userIdHash: otherHash,
parseStatus: 'running',
parseProgress: { totalPages: 4, pagesParsed: 1, currentPage: 2, phase: 'infer' },
updatedAt: Date.now(),
});
await bus.publish({
version: 1,
documentId,
userIdHash: wantedHash,
parseStatus: 'running',
parseProgress: { totalPages: 4, pagesParsed: 2, currentPage: 3, phase: 'infer' },
updatedAt: Date.now(),
opId: 'op-1',
});
close();
expect(received).toHaveLength(1);
expect(received[0].userIdHash).toBe(wantedHash);
expect(received[0].opId).toBe('op-1');
});
});