diff --git a/.env.example b/.env.example index 8173d6d..02c3e57 100644 --- a/.env.example +++ b/.env.example @@ -65,6 +65,7 @@ S3_BUCKET= # COMPUTE_JOB_CONCURRENCY=1 # COMPUTE_WHISPER_TIMEOUT_MS=30000 # COMPUTE_PDF_TIMEOUT_MS=300000 +# COMPUTE_PDF_JOB_ATTEMPTS=1 # COMPUTE_OP_STALE_MS=1800000 # WHISPER_MODEL_BASE_URL=https://huggingface.co/onnx-community/whisper-base_timestamped/resolve/main # PDF_LAYOUT_MODEL_BASE_URL=https://huggingface.co/Bei0001/PP-DocLayoutV3-ONNX/resolve/main diff --git a/compute/core/src/control-plane/index.ts b/compute/core/src/control-plane/index.ts index 8e6b18c..5b8255d 100644 --- a/compute/core/src/control-plane/index.ts +++ b/compute/core/src/control-plane/index.ts @@ -1,5 +1,4 @@ export * from './types'; export * from './state-machine'; export * from './orchestrator'; -export * from './in-memory'; export * from './sse'; diff --git a/compute/core/src/control-plane/orchestrator.ts b/compute/core/src/control-plane/orchestrator.ts index 950fe71..7f621a4 100644 --- a/compute/core/src/control-plane/orchestrator.ts +++ b/compute/core/src/control-plane/orchestrator.ts @@ -263,6 +263,36 @@ export class OperationOrchestrator { return next; } + async markFailedIfUnchanged(input: { + current: WorkerOperationState; + expectedRevision: number; + error: WorkerJobErrorShape | string; + updatedAt?: number; + timing?: WorkerJobTiming; + }): Promise { + const now = input.updatedAt ?? this.clock.now(); + const shape = typeof input.error === 'string' ? { message: input.error } : input.error; + + const next: WorkerOperationState = { + ...input.current, + status: 'failed', + startedAt: input.current.startedAt ?? now, + updatedAt: now, + error: shape, + ...(input.timing ? { timing: input.timing } : {}), + }; + + const updated = await this.stateStore.compareAndSetOpState({ + opId: input.current.opId, + expectedRevision: input.expectedRevision, + newState: next, + }); + if (!updated) return null; + + await this.eventStream.append(next.opId, next); + return next; + } + async explainReuseDecision(input: { current: WorkerOperationState; requestKind: WorkerOperationKind; diff --git a/compute/core/src/control-plane/types.ts b/compute/core/src/control-plane/types.ts index 79fd16c..9fc66ff 100644 --- a/compute/core/src/control-plane/types.ts +++ b/compute/core/src/control-plane/types.ts @@ -26,9 +26,20 @@ export interface OperationIndexEntry { opId: string; } +export interface OperationStateRecord { + state: OperationState; + revision: number; +} + export interface OperationStateStore { getOpState(opId: string): Promise | null>; + getOpStateRecord(opId: string): Promise | null>; putOpState(state: OperationState): Promise; + compareAndSetOpState(input: { + opId: string; + expectedRevision: number; + newState: OperationState; + }): Promise; getOpIndex(opKey: string): Promise; compareAndSetOpIndex(input: { opKey: string; diff --git a/compute/core/src/pdf/merge.ts b/compute/core/src/pdf/merge.ts index aeff9ab..e1f71b2 100644 --- a/compute/core/src/pdf/merge.ts +++ b/compute/core/src/pdf/merge.ts @@ -53,7 +53,11 @@ function joinText(items: PdfTextItem[]): string { const prevEndX = prev.x + prev.width; const gap = item.x - prevEndX; const lineJump = item.y - prev.y; - const lineBreak = lineJump > Math.max(2, Math.min(prev.height, item.height) * 0.6); + const prevBottom = prev.y + prev.height; + const itemBottom = item.y + item.height; + const verticalOverlap = Math.max(0, Math.min(prevBottom, itemBottom) - Math.max(prev.y, item.y)); + const sharesLineBand = verticalOverlap >= Math.max(1, Math.min(prev.height, item.height) * 0.5); + const lineBreak = !sharesLineBand && lineJump > Math.max(2, Math.min(prev.height, item.height) * 0.6); const avgCharWidth = item.width / Math.max(1, item.text.length); const needsSpace = lineBreak || gap > Math.max(avgCharWidth * 0.3, 2); out += needsSpace ? ` ${item.text}` : item.text; diff --git a/compute/core/src/pdf/normalize-text.ts b/compute/core/src/pdf/normalize-text.ts index 2c2eca7..1e91101 100644 --- a/compute/core/src/pdf/normalize-text.ts +++ b/compute/core/src/pdf/normalize-text.ts @@ -1,7 +1,52 @@ import type { TextItem } from 'pdfjs-dist/types/src/display/api'; import type { PdfTextItem } from './types'; -export function normalizeTextItemsForLayout(items: TextItem[], pageHeight: number): PdfTextItem[] { +interface ViewportLike { + height: number; + transform: readonly number[]; +} + +interface TextStyleLike { + ascent?: number; + descent?: number; +} + +function applyViewportTransform( + x: number, + y: number, + transform: readonly number[], +): { x: number; y: number } { + const a = Number(transform[0] ?? 1); + const b = Number(transform[1] ?? 0); + const c = Number(transform[2] ?? 0); + const d = Number(transform[3] ?? 1); + const e = Number(transform[4] ?? 0); + const f = Number(transform[5] ?? 0); + return { + x: (a * x) + (c * y) + e, + y: (b * x) + (d * y) + f, + }; +} + +function resolveTopOffset(height: number, style: TextStyleLike | undefined): number { + const ascent = Number(style?.ascent); + if (Number.isFinite(ascent) && ascent > 0) { + return height * ascent; + } + + const descent = Number(style?.descent); + if (Number.isFinite(descent) && descent < 0) { + return height * (1 + descent); + } + + return height; +} + +export function normalizeTextItemsForLayout( + items: TextItem[], + viewport: ViewportLike, + styles: Record = {}, +): PdfTextItem[] { return items .filter((item) => { if (!(typeof item.str === 'string' && item.str.trim().length > 0)) return false; @@ -17,13 +62,20 @@ export function normalizeTextItemsForLayout(items: TextItem[], pageHeight: numbe return true; }) .map((item) => { - const x = Number(item.transform[4] ?? 0); + const origin = applyViewportTransform( + Number(item.transform[4] ?? 0), + Number(item.transform[5] ?? 0), + viewport.transform, + ); const width = Math.max(0, Number(item.width ?? 0)); const height = Math.max(1, Math.abs(Number(item.transform[3] ?? 1))); - const baselineY = Number(item.transform[5] ?? 0); - // pdf.js text transforms are in PDF user-space (origin bottom-left). - // Normalize into top-left page coordinates to match rendered image/model boxes. - const y = Math.max(0, pageHeight - baselineY - height); + const topOffset = resolveTopOffset(height, styles[item.fontName ?? '']); + // pdf.js text transforms are in PDF user-space and may include non-zero + // page origins via the viewport transform. Map the text baseline into + // viewport coordinates first, then adjust upward using font ascent, + // which matches how pdf.js positions glyph runs in its text layer. + const x = Math.max(0, origin.x); + const y = Math.max(0, origin.y - topOffset); return { text: item.str, x, diff --git a/compute/core/src/pdf/parse.ts b/compute/core/src/pdf/parse.ts index a6dbe22..2b3a6f7 100644 --- a/compute/core/src/pdf/parse.ts +++ b/compute/core/src/pdf/parse.ts @@ -72,7 +72,8 @@ export async function parsePdf(input: ParsePdfInput): Promise const textContent = await page.getTextContent(); const textItems = normalizeTextItemsForLayout( textContent.items.filter((item): item is TextItem => 'str' in item && 'transform' in item), - viewport.height, + viewport, + textContent.styles, ); if (textItems.length > 0) sawText = true; diff --git a/compute/core/src/pdf/runLayoutModel.ts b/compute/core/src/pdf/runLayoutModel.ts index 5b8fb43..5c726e0 100644 --- a/compute/core/src/pdf/runLayoutModel.ts +++ b/compute/core/src/pdf/runLayoutModel.ts @@ -196,8 +196,10 @@ function preprocessResized( const ctx = canvas.getContext('2d'); ctx.fillStyle = '#ffffff'; ctx.fillRect(0, 0, preprocessor.inputWidth, preprocessor.inputHeight); - ctx.imageSmoothingEnabled = true; - ctx.imageSmoothingQuality = 'high'; + // Match the upstream image processor more closely. The official + // implementation explicitly disables antialiasing during resize to stay + // close to OpenCV semantics for PP-DocLayoutV3 inputs. + ctx.imageSmoothingEnabled = false; ctx.drawImage(image, 0, 0, preprocessor.inputWidth, preprocessor.inputHeight); const imageData = ctx.getImageData(0, 0, preprocessor.inputWidth, preprocessor.inputHeight); @@ -284,12 +286,29 @@ export async function runLayoutModel(input: RunLayoutInput): Promise { let firstAttempt = true; const conflictStore = { getOpState: store.getOpState.bind(store), + getOpStateRecord: store.getOpStateRecord.bind(store), putOpState: store.putOpState.bind(store), + compareAndSetOpState: store.compareAndSetOpState.bind(store), getOpIndex: store.getOpIndex.bind(store), compareAndSetOpIndex: async (input: { opKey: string; newOpId: string; expectedOpId: string | null }) => { if (firstAttempt && input.expectedOpId === null) { @@ -92,4 +94,45 @@ describe('operation orchestrator', () => { expect(created.opId).toMatch(/^op-/); expect(await store.getOpIndex('cas-key')).toEqual({ opId: created.opId }); }); + + test('markFailedIfUnchanged only writes once for the expected revision', async () => { + const queue = new InMemoryOperationQueue(); + const stateStore = new InMemoryOperationStateStore(); + const eventStream = new InMemoryOperationEventStream(); + const orchestrator = new OperationOrchestrator({ + queue, + stateStore, + eventStream, + config: { opStaleMs: 2_000, maxCasRetries: 5 }, + }); + + const created = await orchestrator.enqueueOrReuse(buildRequest('stale-op')); + await orchestrator.markRunning({ opId: created.opId, updatedAt: 2_000 }); + + const record = await stateStore.getOpStateRecord(created.opId); + expect(record).not.toBeNull(); + + const first = await orchestrator.markFailedIfUnchanged({ + current: record!.state, + expectedRevision: record!.revision, + error: { code: 'WORKER_ORPHANED_OP', message: 'stale op' }, + updatedAt: 3_000, + }); + const second = await orchestrator.markFailedIfUnchanged({ + current: record!.state, + expectedRevision: record!.revision, + error: { code: 'WORKER_ORPHANED_OP', message: 'stale op' }, + updatedAt: 3_000, + }); + + expect(first).toMatchObject({ + opId: created.opId, + status: 'failed', + error: { code: 'WORKER_ORPHANED_OP' }, + }); + expect(second).toBeNull(); + + const events = await eventStream.listSince(created.opId, 0); + expect(events.filter((event) => event.snapshot.status === 'failed')).toHaveLength(1); + }); }); diff --git a/compute/core/tests/control-plane/run-layout-model.test.ts b/compute/core/tests/control-plane/run-layout-model.test.ts new file mode 100644 index 0000000..e951bf3 --- /dev/null +++ b/compute/core/tests/control-plane/run-layout-model.test.ts @@ -0,0 +1,175 @@ +import { beforeEach, describe, expect, test, vi } from 'vitest'; + +const mockState = vi.hoisted(() => ({ + runOutput: { + logits: { data: new Float32Array() }, + pred_boxes: { data: new Float32Array() }, + }, +})); + +vi.mock('onnxruntime-node', () => ({ + InferenceSession: { + create: vi.fn(async () => ({ + run: vi.fn(async () => mockState.runOutput), + })), + }, + Tensor: class Tensor { + constructor( + public type: string, + public data: Float32Array, + public dims: number[], + ) {} + }, +})); + +vi.mock('fs/promises', () => ({ + readFile: vi.fn(async (path: string) => { + if (path === '/tmp/model-config.json') { + return JSON.stringify({ + id2label: { + 0: 'text', + 1: 'table', + }, + }); + } + + if (path === '/tmp/model-preprocessor.json') { + return JSON.stringify({ + size: { width: 2, height: 2 }, + rescale_factor: 1 / 255, + image_mean: [0, 0, 0], + image_std: [1, 1, 1], + }); + } + + throw new Error(`unexpected readFile path: ${path}`); + }), +})); + +vi.mock('@napi-rs/canvas', () => { + const createCanvas = (width: number, height: number) => ({ + getContext: () => ({ + fillStyle: '#ffffff', + fillRect: () => {}, + drawImage: () => {}, + imageSmoothingEnabled: true, + getImageData: () => ({ + data: new Uint8ClampedArray(width * height * 4).fill(255), + }), + }), + }); + const loadImage = vi.fn(async () => ({ width: 2, height: 2 })); + + return { + createCanvas, + loadImage, + default: { + createCanvas, + loadImage, + }, + }; +}); + +vi.mock('../../src/pdf/model', () => ({ + ensureModel: vi.fn(async () => '/tmp/model.onnx'), + MODEL_CONFIG_PATH: '/tmp/model-config.json', + MODEL_PREPROCESSOR_PATH: '/tmp/model-preprocessor.json', +})); + +vi.mock('../../src/config/cpu-budget', () => ({ + getOnnxThreadsPerJob: vi.fn(() => 1), +})); + +describe('runLayoutModel', () => { + beforeEach(() => { + vi.resetModules(); + mockState.runOutput = { + logits: { data: new Float32Array() }, + pred_boxes: { data: new Float32Array() }, + }; + }); + + test('keeps one winner per query instead of dropping later queries behind duplicate class rows', async () => { + mockState.runOutput = { + logits: { + data: new Float32Array([ + 3, + 4, + 2.5, + 0.1, + ]), + }, + pred_boxes: { + data: new Float32Array([ + 0.25, 0.25, 0.3, 0.3, + 0.75, 0.75, 0.3, 0.3, + ]), + }, + }; + + const { runLayoutModel } = await import('../../src/pdf/runLayoutModel'); + const regions = await runLayoutModel({ + pageWidth: 100, + pageHeight: 100, + textItems: [{} as never], + pageImage: Buffer.from([1]), + }); + + expect(regions).toHaveLength(2); + expect(regions[0]?.label).toBe('text'); + expect(regions[0]?.confidence).toBeCloseTo(0.9168273, 6); + expect(regions[0]?.bbox).toEqual([ + expect.closeTo(60, 5), + expect.closeTo(60, 5), + expect.closeTo(90, 5), + expect.closeTo(90, 5), + ]); + expect(regions[1]?.label).toBe('table'); + expect(regions[1]?.confidence).toBeCloseTo(0.73105858, 6); + expect(regions[1]?.bbox).toEqual([ + expect.closeTo(10, 5), + expect.closeTo(10, 5), + expect.closeTo(40, 5), + expect.closeTo(40, 5), + ]); + }); + + test('drops unlabeled query winners and keeps only labeled regions', async () => { + mockState.runOutput = { + logits: { + data: new Float32Array([ + 0.1, + 0.2, + 5, + 4, + 0.1, + 0.1, + ]), + }, + pred_boxes: { + data: new Float32Array([ + 0.25, 0.25, 0.3, 0.3, + 0.75, 0.75, 0.3, 0.3, + ]), + }, + }; + + const { runLayoutModel } = await import('../../src/pdf/runLayoutModel'); + const regions = await runLayoutModel({ + pageWidth: 100, + pageHeight: 100, + textItems: [{} as never], + pageImage: Buffer.from([1]), + }); + + expect(regions).toHaveLength(1); + expect(regions[0]?.label).toBe('text'); + expect(regions[0]?.confidence).toBeCloseTo(0.96109135, 5); + expect(regions[0]?.bbox).toEqual([ + expect.closeTo(60, 5), + expect.closeTo(60, 5), + expect.closeTo(90, 5), + expect.closeTo(90, 5), + ]); + }); +}); diff --git a/compute/core/src/control-plane/in-memory.ts b/compute/core/tests/helpers/in-memory-control-plane.ts similarity index 80% rename from compute/core/src/control-plane/in-memory.ts rename to compute/core/tests/helpers/in-memory-control-plane.ts index fc77c0e..cc9b9ce 100644 --- a/compute/core/src/control-plane/in-memory.ts +++ b/compute/core/tests/helpers/in-memory-control-plane.ts @@ -1,4 +1,5 @@ import { EventEmitter } from 'node:events'; +import type { WorkerOperationKind } from '../../src/api-contracts'; import type { OperationEvent, OperationEventStream, @@ -7,8 +8,7 @@ import type { OperationState, OperationStateStore, QueuedOperation, -} from './types'; -import type { WorkerOperationKind } from '../api-contracts'; +} from '../../src/control-plane/types'; function topicFor(opId: string): string { return `op.${opId}`; @@ -49,14 +49,39 @@ export class InMemoryOperationQueue implements OperationQueue { export class InMemoryOperationStateStore implements OperationStateStore { private readonly stateByOpId = new Map(); + private readonly revisionByOpId = new Map(); private readonly opIndexByKey = new Map(); async getOpState(opId: string): Promise { return this.stateByOpId.get(opId) ?? null; } + async getOpStateRecord(opId: string): Promise<{ state: OperationState; revision: number } | null> { + const state = this.stateByOpId.get(opId); + if (!state) return null; + return { + state, + revision: this.revisionByOpId.get(opId) ?? 0, + }; + } + async putOpState(state: OperationState): Promise { this.stateByOpId.set(state.opId, state); + this.revisionByOpId.set(state.opId, (this.revisionByOpId.get(state.opId) ?? 0) + 1); + } + + async compareAndSetOpState(input: { + opId: string; + expectedRevision: number; + newState: OperationState; + }): Promise { + const currentState = this.stateByOpId.get(input.opId); + if (!currentState) return false; + const currentRevision = this.revisionByOpId.get(input.opId) ?? 0; + if (currentRevision !== input.expectedRevision) return false; + this.stateByOpId.set(input.opId, input.newState); + this.revisionByOpId.set(input.opId, currentRevision + 1); + return true; } async getOpIndex(opKey: string): Promise { diff --git a/compute/worker/src/control-plane/jetstream.ts b/compute/worker/src/control-plane/jetstream.ts index 642906d..9225d96 100644 --- a/compute/worker/src/control-plane/jetstream.ts +++ b/compute/worker/src/control-plane/jetstream.ts @@ -27,6 +27,7 @@ export interface KvStoreLike { put(key: string, data: Uint8Array): Promise; create(key: string, data: Uint8Array): Promise; update(key: string, data: Uint8Array, version: number): Promise; + keys(filter?: string | string[]): Promise>; } function toErrorMessage(error: unknown): string { @@ -80,10 +81,18 @@ export class JetStreamOperationStateStore implements Operation } async getOpState(opId: string): Promise | null> { + const record = await this.getOpStateRecord(opId); + return record?.state ?? null; + } + + async getOpStateRecord(opId: string): Promise<{ state: OperationState; revision: number } | null> { const kv = await this.getKv(); const entry = await kv.get(opStateKvKey(opId)); if (!isPut(entry)) return null; - return this.opStateCodec.decode(entry.value); + return { + state: this.opStateCodec.decode(entry.value), + revision: entry.revision, + }; } async putOpState(state: OperationState): Promise { @@ -91,6 +100,37 @@ export class JetStreamOperationStateStore implements Operation await kv.put(opStateKvKey(state.opId), this.opStateCodec.encode(state)); } + async compareAndSetOpState(input: { + opId: string; + expectedRevision: number; + newState: OperationState; + }): Promise { + const kv = await this.getKv(); + try { + await kv.update( + opStateKvKey(input.opId), + this.opStateCodec.encode(input.newState), + input.expectedRevision, + ); + return true; + } catch (error) { + if (isCasConflictError(error)) return false; + throw error; + } + } + + async listOpStates(): Promise[]> { + const kv = await this.getKv(); + const keys = await kv.keys('op_state.*'); + const states: OperationState[] = []; + for await (const key of keys) { + const entry = await kv.get(key); + if (!isPut(entry)) continue; + states.push(this.opStateCodec.decode(entry.value)); + } + return states; + } + async getOpIndex(opKey: string): Promise<{ opId: string } | null> { const kv = await this.getKv(); const entry = await kv.get(opIndexKvKey(opKey)); diff --git a/compute/worker/src/orphan-recovery.ts b/compute/worker/src/orphan-recovery.ts new file mode 100644 index 0000000..882ea94 --- /dev/null +++ b/compute/worker/src/orphan-recovery.ts @@ -0,0 +1,100 @@ +import type { + PdfLayoutJobResult, + WhisperAlignJobResult, + WorkerJobTiming, + WorkerJobState, + WorkerOperationState, +} from '@openreader/compute-core/api-contracts'; + +export type StreamedOperationState = WorkerOperationState; + +export interface OrphanRecoveryStateStore { + getOpStateRecord(opId: string): Promise<{ state: StreamedOperationState; revision: number } | null>; + listOpStates(): Promise; +} + +export interface OrphanRecoveryOrchestrator { + markFailedIfUnchanged(input: { + current: StreamedOperationState; + expectedRevision: number; + error: { message: string; code?: string } | string; + updatedAt?: number; + timing?: WorkerJobTiming; + }): Promise; +} + +export interface RecoverOrphanedOperationsInput { + operationStateStore: OrphanRecoveryStateStore; + orchestrator: OrphanRecoveryOrchestrator; + whisperTimeoutMs: number; + pdfTimeoutMs: number; + opStaleMs: number; + nowMs?: number; +} + +function isInflightStatus(status: WorkerJobState): boolean { + return status === 'queued' || status === 'running'; +} + +export function getOrphanRecoveryThresholdMs(input: { + state: StreamedOperationState; + whisperTimeoutMs: number; + pdfTimeoutMs: number; + opStaleMs: number; +}): number | null { + if (!isInflightStatus(input.state.status)) return null; + if (input.state.status === 'running') { + return input.state.kind === 'whisper_align' ? input.whisperTimeoutMs : input.pdfTimeoutMs; + } + if (input.state.kind !== 'pdf_layout') return null; + return input.opStaleMs; +} + +export async function recoverOrphanedOperations( + input: RecoverOrphanedOperationsInput, +): Promise>> { + const nowMs = input.nowMs ?? Date.now(); + const states = await input.operationStateStore.listOpStates(); + const candidateStates = states.filter((state) => ( + getOrphanRecoveryThresholdMs({ + state, + whisperTimeoutMs: input.whisperTimeoutMs, + pdfTimeoutMs: input.pdfTimeoutMs, + opStaleMs: input.opStaleMs, + }) !== null + )); + const recoveredStates: Array> = []; + + for (const candidate of candidateStates) { + const record = await input.operationStateStore.getOpStateRecord(candidate.opId); + if (!record) continue; + const staleAfterMs = getOrphanRecoveryThresholdMs({ + state: record.state, + whisperTimeoutMs: input.whisperTimeoutMs, + pdfTimeoutMs: input.pdfTimeoutMs, + opStaleMs: input.opStaleMs, + }); + if (staleAfterMs === null) continue; + const ageMs = nowMs - record.state.updatedAt; + if (ageMs <= staleAfterMs) continue; + + const recovered = await input.orchestrator.markFailedIfUnchanged({ + current: record.state, + expectedRevision: record.revision, + error: { + code: 'WORKER_ORPHANED_OP', + message: `Worker stopped before completion; stale operation recovered during reconciliation after ${staleAfterMs}ms`, + }, + updatedAt: nowMs, + }); + if (!recovered) continue; + + recoveredStates.push({ + opId: recovered.opId, + kind: recovered.kind, + status: record.state.status, + }); + } + + return recoveredStates; +} diff --git a/compute/worker/src/runtime.ts b/compute/worker/src/runtime.ts index 2a43774..e208e63 100644 --- a/compute/worker/src/runtime.ts +++ b/compute/worker/src/runtime.ts @@ -56,6 +56,10 @@ import { hashOpKey, } from './control-plane/jetstream'; import { type JsonCodec, createJsonCodec } from './control-plane/json-codec'; +import { + recoverOrphanedOperations, + type StreamedOperationState, +} from './orphan-recovery'; import { buildInferProgressForPageParsed, buildInferProgressForPageStart } from './pdf-progress'; import { buildQueueWaitTiming, decideRetryAction } from './worker-loop-policy'; @@ -69,6 +73,7 @@ const COMPUTE_STATE_BUCKET = 'compute_state'; const COMPUTE_STATE_TTL_MS = 24 * 60 * 60 * 1000; const LOOP_ERROR_BACKOFF_MS = 500; const RUNNING_HEARTBEAT_MS = 5000; +const OP_EVENTS_KEEPALIVE_MS = 15_000; const DOCUMENT_ID_REGEX = /^[a-f0-9]{64}$/i; const SAFE_NAMESPACE_REGEX = /^[a-zA-Z0-9._-]{1,128}$/; const WHISPER_MAX_DELIVER = 1; @@ -78,6 +83,7 @@ const NATS_API_TIMEOUT_MS = 60_000; // put it to sleep. Reconnect happens lazily on the next inbound request. const IDLE_DISCONNECT_MS = 120_000; const IDLE_CHECK_INTERVAL_MS = 5_000; +const ORPHAN_SWEEP_INTERVAL_MS = 15_000; // Bounded pull window so consumer loops yield periodically and can be stopped // cleanly when going idle, instead of blocking on a long-lived pull. const PULL_EXPIRES_MS = 5_000; @@ -106,8 +112,6 @@ interface NatsSession { layoutConsumer: Consumer; } -type StreamedOperationState = WorkerOperationState; - interface OperationEventStreamLike { subscribe(input: { opId: string; @@ -119,6 +123,8 @@ interface OperationEventStreamLike { interface OperationStateStoreLike { getOpState(opId: string): Promise; + getOpStateRecord?(opId: string): Promise<{ state: StreamedOperationState; revision: number } | null>; + listOpStates?(): Promise; } interface OrchestratorLike { @@ -147,6 +153,13 @@ interface OrchestratorLike { updatedAt?: number; timing?: WorkerJobTiming; }): Promise; + markFailedIfUnchanged?(input: { + current: StreamedOperationState; + expectedRevision: number; + error: { message: string; code?: string } | string; + updatedAt?: number; + timing?: WorkerJobTiming; + }): Promise; } export interface ComputeWorkerRouteDeps { @@ -540,6 +553,7 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti let connecting: Promise | null = null; let workerLoops: Promise[] = []; let idleTimer: NodeJS.Timeout | null = null; + let orphanSweepTimer: NodeJS.Timeout | null = null; let stopping = false; let loopStopRequested = false; @@ -568,6 +582,19 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti idleTimer.unref?.(); } + function startOrphanSweepTimer(): void { + if (orphanSweepTimer) return; + orphanSweepTimer = setInterval(() => { + if (!session || stopping) return; + void runOrphanedOpRecovery({ force: true }).catch((error) => { + app.log.error({ + error: toErrorMessage(error), + }, 'periodic orphaned operation recovery failed'); + }); + }, ORPHAN_SWEEP_INTERVAL_MS); + orphanSweepTimer.unref?.(); + } + async function disconnect(reason: string): Promise { const current = session; if (!current) return; @@ -579,6 +606,10 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti clearInterval(idleTimer); idleTimer = null; } + if (orphanSweepTimer) { + clearInterval(orphanSweepTimer); + orphanSweepTimer = null; + } try { await current.nc.close(); } catch { @@ -616,9 +647,12 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti const layoutConsumer = await js.consumers.get(JOBS_STREAM_NAME, LAYOUT_CONSUMER_NAME); const next: NatsSession = { nc, js, jsm, kv, whisperConsumer, layoutConsumer }; session = next; + sessionGeneration += 1; + orphanRecoveryDoneForGeneration = -1; markActivity(); startWorkerLoops(next); startIdleTimer(); + startOrphanSweepTimer(); // Safety net: if the connection closes for any reason (network drop after // exhausting reconnects, or our own disconnect), drop the stale session so // the next request reconnects cleanly. @@ -730,8 +764,65 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti const operationStateStore = options.routeDeps?.operationStateStore ?? defaultOperationStateStore; const operationEventStream = options.routeDeps?.operationEventStream ?? defaultOperationEventStream; const orchestrator = options.routeDeps?.orchestrator ?? defaultOrchestrator; + let orphanRecoveryPromise: Promise | null = null; + let orphanRecoveryDoneForGeneration = -1; + let sessionGeneration = options.routeDeps ? 0 : -1; + + const runOrphanedOpRecovery = async (options?: { force?: boolean }): Promise => { + if (typeof operationStateStore.listOpStates !== 'function') return; + if (typeof operationStateStore.getOpStateRecord !== 'function') return; + if (typeof orchestrator.markFailedIfUnchanged !== 'function') return; + if (!options?.force && orphanRecoveryDoneForGeneration === sessionGeneration) return; + if (orphanRecoveryPromise) { + await orphanRecoveryPromise; + return; + } + + orphanRecoveryPromise = (async () => { + const recoveredStates = await recoverOrphanedOperations({ + operationStateStore: operationStateStore as typeof operationStateStore & { + getOpStateRecord(opId: string): Promise<{ state: StreamedOperationState; revision: number } | null>; + listOpStates(): Promise; + }, + orchestrator: orchestrator as typeof orchestrator & { + markFailedIfUnchanged(input: { + current: StreamedOperationState; + expectedRevision: number; + error: { message: string; code?: string } | string; + updatedAt?: number; + timing?: WorkerJobTiming; + }): Promise; + }, + whisperTimeoutMs, + pdfTimeoutMs, + opStaleMs, + }); + + if (recoveredStates.length > 0) { + app.log.warn({ + recoveredCount: recoveredStates.length, + ops: recoveredStates.map((state) => ({ + opId: state.opId, + kind: state.kind, + status: state.status, + })), + }, 'recovered stale in-flight operations during reconciliation'); + } + + orphanRecoveryDoneForGeneration = sessionGeneration; + })().finally(() => { + orphanRecoveryPromise = null; + }); + + await orphanRecoveryPromise; + }; + + const ensureOrphanedOpRecovery = async (): Promise => { + await runOrphanedOpRecovery(); + }; const getOpState = async (opId: string): Promise => { + await ensureOrphanedOpRecovery(); return await operationStateStore.getOpState(opId); }; @@ -797,6 +888,7 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti } const requestOp = parsed.data as WorkerOperationRequest; + await ensureOrphanedOpRecovery(); const op = await orchestrator.enqueueOrReuse(requestOp); app.log.info({ kind: requestOp.kind, @@ -863,6 +955,7 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti let closed = false; let unsubscribe: (() => void) | null = null; + let keepalive: NodeJS.Timeout | null = null; const writeSnapshot = (snapshot: StreamedOperationState, eventId: number): void => { if (closed || reply.raw.writableEnded) return; @@ -884,6 +977,10 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti unsubscribe(); unsubscribe = null; } + if (keepalive) { + clearInterval(keepalive); + keepalive = null; + } activeSse = Math.max(0, activeSse - 1); markActivity(); if (!reply.raw.writableEnded) { @@ -903,6 +1000,11 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti return reply; } + keepalive = setInterval(() => { + if (closed || reply.raw.writableEnded) return; + reply.raw.write(': keepalive\n\n'); + }, OP_EVENTS_KEEPALIVE_MS); + unsubscribe = await operationEventStream.subscribe({ opId: params.data.opId, sinceEventId, @@ -1247,6 +1349,17 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti const result = await input.run(decoded.payload, context.queueWaitTiming?.queueWaitMs ?? 0, { onProgress: async (progress) => { + try { + input.msg.working(); + } catch (ackError) { + app.log.warn({ + worker: input.workerLabel, + kind: context?.decoded.kind, + opId: context?.decoded.opId, + jobId: context?.decoded.jobId, + error: toErrorMessage(ackError), + }, 'failed to extend JetStream ack wait on progress'); + } await markProgress(context!, progress, Date.now()); }, }); @@ -1380,6 +1493,10 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti clearInterval(idleTimer); idleTimer = null; } + if (orphanSweepTimer) { + clearInterval(orphanSweepTimer); + orphanSweepTimer = null; + } await app.close(); await Promise.allSettled(workerLoops); const current = session; diff --git a/compute/worker/tests/api/routes.test.ts b/compute/worker/tests/api/routes.test.ts index 77c4d05..dad89d9 100644 --- a/compute/worker/tests/api/routes.test.ts +++ b/compute/worker/tests/api/routes.test.ts @@ -119,4 +119,79 @@ describe('compute worker API routes', () => { expect(stream.body).toContain('id: 7'); expect(stream.body).toContain('"status":"succeeded"'); }); + + test('marks stale running whisper and pdf ops failed during request-time orphan recovery but leaves queued ops on the conservative path', async () => { + const now = Date.now(); + fake.seedState({ + opId: 'op-stale-whisper-running', + opKey: 'k-stale-whisper-running', + kind: 'whisper_align', + jobId: 'job-op-stale-whisper-running', + status: 'running', + queuedAt: 1, + updatedAt: now - 40_000, + }); + fake.seedState({ + opId: 'op-stale-whisper-queued', + opKey: 'k-stale-whisper-queued', + kind: 'whisper_align', + jobId: 'job-op-stale-whisper-queued', + status: 'queued', + queuedAt: 1, + updatedAt: now - 40_000, + }); + fake.seedState({ + opId: 'op-stale-pdf-running', + opKey: 'k-stale-pdf-running', + kind: 'pdf_layout', + jobId: 'job-op-stale-pdf-running', + status: 'running', + queuedAt: 1, + updatedAt: now - 310_000, + }); + fake.seedState({ + opId: 'op-stale-pdf-queued', + opKey: 'k-stale-pdf-queued', + kind: 'pdf_layout', + jobId: 'job-op-stale-pdf-queued', + status: 'queued', + queuedAt: 1, + updatedAt: now - 310_000, + }); + + // GET /ops/:opId resolves via getOpState(), which first awaits the shared + // orphanRecoveryPromise path through ensureOrphanedOpRecovery(). + const fetch = await runtime.app.inject({ + method: 'GET', + url: '/ops/op-stale-whisper-running', + headers: AUTH, + }); + + expect(fetch.statusCode).toBe(200); + expect(fetch.json()).toMatchObject({ + opId: 'op-stale-whisper-running', + status: 'failed', + error: { + code: 'WORKER_ORPHANED_OP', + }, + }); + expect(fake.getState('op-stale-whisper-running')).toMatchObject({ + status: 'failed', + error: { + code: 'WORKER_ORPHANED_OP', + }, + }); + expect(fake.getState('op-stale-whisper-queued')).toMatchObject({ + status: 'queued', + }); + expect(fake.getState('op-stale-pdf-running')).toMatchObject({ + status: 'failed', + error: { + code: 'WORKER_ORPHANED_OP', + }, + }); + expect(fake.getState('op-stale-pdf-queued')).toMatchObject({ + status: 'queued', + }); + }); }); diff --git a/compute/worker/tests/fixtures/fake-control-plane.ts b/compute/worker/tests/fixtures/fake-control-plane.ts index 2d18f2b..9822523 100644 --- a/compute/worker/tests/fixtures/fake-control-plane.ts +++ b/compute/worker/tests/fixtures/fake-control-plane.ts @@ -13,6 +13,7 @@ type ComputeEvent = WorkerOperationEvent; export class FakeControlPlane { private readonly stateByOpId = new Map(); + private readonly revisionByOpId = new Map(); private readonly opIdByOpKey = new Map(); private readonly eventsByOpId = new Map(); private nextOpId = 1; @@ -44,9 +45,19 @@ export class FakeControlPlane { updatedAt: input.updatedAt, timing: input.timing, }), + markFailedIfUnchanged: async (input) => this.compareAndSetFailed(input), }, operationStateStore: { getOpState: async (opId) => this.stateByOpId.get(opId) ?? null, + getOpStateRecord: async (opId) => { + const state = this.stateByOpId.get(opId); + if (!state) return null; + return { + state, + revision: this.revisionByOpId.get(opId) ?? 0, + }; + }, + listOpStates: async () => Array.from(this.stateByOpId.values()), }, operationEventStream: { subscribe: async ({ opId, sinceEventId, onEvent }) => { @@ -62,6 +73,7 @@ export class FakeControlPlane { seedState(state: ComputeState): void { this.stateByOpId.set(state.opId, state); + this.revisionByOpId.set(state.opId, (this.revisionByOpId.get(state.opId) ?? 0) + 1); this.opIdByOpKey.set(state.opKey, state.opId); } @@ -71,6 +83,10 @@ export class FakeControlPlane { this.eventsByOpId.set(opId, list); } + getState(opId: string): ComputeState | null { + return this.stateByOpId.get(opId) ?? null; + } + private async enqueueOrReuse(request: WorkerOperationRequest): Promise { const existingId = this.opIdByOpKey.get(request.opKey); if (existingId) { @@ -91,11 +107,30 @@ export class FakeControlPlane { }; this.stateByOpId.set(opId, state); + this.revisionByOpId.set(opId, 1); this.opIdByOpKey.set(request.opKey, opId); this.seedEvent(opId, { eventId: 1, snapshot: state }); return state; } + private async compareAndSetFailed(input: { + current: ComputeState; + expectedRevision: number; + error: { message: string; code?: string } | string; + updatedAt?: number; + timing?: ComputeState['timing']; + }): Promise { + const currentRevision = this.revisionByOpId.get(input.current.opId) ?? 0; + if (currentRevision !== input.expectedRevision) return null; + return this.updateState(input.current.opId, { + ...input.current, + status: 'failed', + error: typeof input.error === 'string' ? { message: input.error } : input.error, + updatedAt: input.updatedAt, + timing: input.timing, + }); + } + private async updateState( opId: string, patch: Partial, @@ -110,6 +145,7 @@ export class FakeControlPlane { updatedAt: patch.updatedAt ?? Date.now(), }; this.stateByOpId.set(opId, next); + this.revisionByOpId.set(opId, (this.revisionByOpId.get(opId) ?? 0) + 1); const currentEvents = this.eventsByOpId.get(opId) ?? []; const nextEventId = (currentEvents.at(-1)?.eventId ?? 0) + 1; currentEvents.push({ eventId: nextEventId, snapshot: next }); diff --git a/compute/worker/tests/unit/orphan-recovery.test.ts b/compute/worker/tests/unit/orphan-recovery.test.ts new file mode 100644 index 0000000..712fdf6 --- /dev/null +++ b/compute/worker/tests/unit/orphan-recovery.test.ts @@ -0,0 +1,59 @@ +import { afterEach, describe, expect, test, vi } from 'vitest'; +import { recoverOrphanedOperations } from '../../src/orphan-recovery'; +import { FakeControlPlane } from '../fixtures/fake-control-plane'; + +describe('orphan recovery', () => { + afterEach(() => { + vi.useRealTimers(); + }); + + test('recovers a running whisper op when a later sweep crosses the timeout in the same session', async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date('2026-06-03T08:00:00.000Z')); + + const fake = new FakeControlPlane(); + const startedAt = Date.now(); + fake.seedState({ + opId: 'op-whisper-running', + opKey: 'k-whisper-running', + kind: 'whisper_align', + jobId: 'job-op-whisper-running', + status: 'running', + queuedAt: startedAt, + updatedAt: startedAt, + }); + + const firstSweep = await recoverOrphanedOperations({ + operationStateStore: fake.deps.operationStateStore, + orchestrator: fake.deps.orchestrator, + whisperTimeoutMs: 30_000, + pdfTimeoutMs: 300_000, + opStaleMs: 1_800_000, + }); + expect(firstSweep).toEqual([]); + expect(fake.getState('op-whisper-running')).toMatchObject({ + status: 'running', + }); + + vi.advanceTimersByTime(31_000); + + const secondSweep = await recoverOrphanedOperations({ + operationStateStore: fake.deps.operationStateStore, + orchestrator: fake.deps.orchestrator, + whisperTimeoutMs: 30_000, + pdfTimeoutMs: 300_000, + opStaleMs: 1_800_000, + }); + expect(secondSweep).toEqual([{ + opId: 'op-whisper-running', + kind: 'whisper_align', + status: 'running', + }]); + expect(fake.getState('op-whisper-running')).toMatchObject({ + status: 'failed', + error: { + code: 'WORKER_ORPHANED_OP', + }, + }); + }); +}); diff --git a/docs-site/docs/deploy/compute-worker.md b/docs-site/docs/deploy/compute-worker.md index b5ec152..cd279db 100644 --- a/docs-site/docs/deploy/compute-worker.md +++ b/docs-site/docs/deploy/compute-worker.md @@ -41,7 +41,7 @@ S3_SECRET_ACCESS_KEY=... - Embedded/local mode: configure the root `.env` only. - External worker mode: set `COMPUTE_WORKER_URL` and `COMPUTE_WORKER_TOKEN` on the app, and worker runtime values on the worker service. -- Keep shared values aligned across app and worker: `COMPUTE_WORKER_TOKEN`, `S3_*`, `COMPUTE_WHISPER_TIMEOUT_MS`, `COMPUTE_PDF_TIMEOUT_MS`, and `COMPUTE_OP_STALE_MS`. +- Keep shared values aligned across app and worker: `COMPUTE_WORKER_TOKEN`, `S3_*`, `COMPUTE_WHISPER_TIMEOUT_MS`, `COMPUTE_PDF_TIMEOUT_MS`, `COMPUTE_PDF_JOB_ATTEMPTS`, and `COMPUTE_OP_STALE_MS`. ::: Common optional variables: @@ -76,6 +76,7 @@ COMPUTE_WORKER_TOKEN= # Optional shared overrides: # COMPUTE_WHISPER_TIMEOUT_MS=30000 # COMPUTE_PDF_TIMEOUT_MS=300000 +# COMPUTE_PDF_JOB_ATTEMPTS=1 # COMPUTE_OP_STALE_MS=1800000 ``` diff --git a/docs-site/docs/deploy/local-development.md b/docs-site/docs/deploy/local-development.md index 338aeb0..7454b7b 100644 --- a/docs-site/docs/deploy/local-development.md +++ b/docs-site/docs/deploy/local-development.md @@ -189,9 +189,11 @@ COMPUTE_WORKER_TOKEN= ``` Ownership in external worker mode: -- root `.env`: app routing/auth (`COMPUTE_WORKER_URL`, `COMPUTE_WORKER_TOKEN`) plus optional shared timeout/stale overrides +- root `.env`: app routing/auth (`COMPUTE_WORKER_URL`, `COMPUTE_WORKER_TOKEN`) plus optional shared timeout/stale/retry overrides such as `COMPUTE_PDF_JOB_ATTEMPTS` - `compute/worker/.env*` (or worker platform env): worker runtime variables (`NATS_*`, `S3_*`, model base URLs, worker tuning) +For embedded worker startup (`COMPUTE_WORKER_URL` unset), worker tuning values such as `COMPUTE_PDF_JOB_ATTEMPTS` must be set in the root `.env` because `compute/worker/.env*` is ignored in that mode. + Worker mode requires worker-reachable shared object storage (S3-compatible endpoint). For external worker mode, object storage must be shared/reachable by both app and worker services. diff --git a/docs-site/docs/reference/environment-variables.md b/docs-site/docs/reference/environment-variables.md index 45786fc..d80bd8e 100644 --- a/docs-site/docs/reference/environment-variables.md +++ b/docs-site/docs/reference/environment-variables.md @@ -48,6 +48,7 @@ Runtime site features are seeded with `RUNTIME_SEED_JSON` / `RUNTIME_SEED_JSON_P | `COMPUTE_JOB_CONCURRENCY` | Compute | `1` | Shared compute concurrency cap | | `COMPUTE_WHISPER_TIMEOUT_MS` | Compute | `30000` | Whisper alignment timeout budget | | `COMPUTE_PDF_TIMEOUT_MS` | Compute | `300000` | PDF parse timeout budget | +| `COMPUTE_PDF_JOB_ATTEMPTS` | Compute | `1` | Max JetStream deliveries for PDF layout jobs | | `COMPUTE_OP_STALE_MS` | Compute | `max(30m, 4x max compute timeout)` | Shared stale window for compute op replacement | | `WHISPER_MODEL_BASE_URL` | Compute model source | onnx-community default | Override Whisper ONNX model base URL | | `PDF_LAYOUT_MODEL_BASE_URL` | Compute model source | PP-DocLayoutV3 default | Override PDF layout ONNX model base URL | @@ -307,6 +308,13 @@ PDF parse timeout budget. - Default: `300000` +### COMPUTE_PDF_JOB_ATTEMPTS + +Max JetStream deliveries for PDF layout jobs. + +- Default: `1` +- In embedded worker mode, set this in the root `.env` + ### COMPUTE_OP_STALE_MS Stale operation window before worker/app cleanup logic can replace an op. diff --git a/src/app/(app)/pdf/[id]/page.tsx b/src/app/(app)/pdf/[id]/page.tsx index 0d1b3b9..79d4060 100644 --- a/src/app/(app)/pdf/[id]/page.tsx +++ b/src/app/(app)/pdf/[id]/page.tsx @@ -19,6 +19,7 @@ import { RateLimitBanner } from '@/components/auth/RateLimitBanner'; import { useAuthRateLimit } from '@/contexts/AuthRateLimitContext'; import { useFeatureFlag } from '@/contexts/RuntimeConfigContext'; import { LoadingSpinner } from '@/components/Spinner'; +import { PdfLayoutScan } from '@/components/reader/PdfLayoutScan'; import { Button, ButtonLink } from '@/components/ui'; import { FORCE_REPARSE_CONFIRM_MESSAGE, @@ -279,21 +280,13 @@ export default function PDFViewerPage() { const isMerging = parseProgress?.phase === 'merge'; let statusText = 'Loading PDF...'; - let statusSubText = 'Initializing document renderer'; if (!isLoading) { if (parseUiState === 'pending') { statusText = 'Preparing PDF layout...'; - statusSubText = parseProgress?.phase === 'merge' - ? 'Finalizing stitched block structure' - : 'Queueing parser and preparing page extraction'; } else if (parseUiState === 'running') { statusText = 'Parsing PDF layout blocks...'; - statusSubText = parseProgress?.phase === 'merge' - ? 'Merging cross-page sections' - : 'Inferring reading order and text regions'; } else if (parseUiState === 'failed') { statusText = 'PDF parsing failed. Retry to continue.'; - statusSubText = 'The parser could not build a usable layout map'; } } @@ -305,34 +298,89 @@ export default function PDFViewerPage() { return (
-
+
{showDetailedParseLoader ? ( -
-
-
-
+
+ {/* prism top edge + corner glow for depth */} +
+
+ {/* dotted texture spanning the whole loader */} +
+ +
+ {/* header: status badge + model attribution */} +
- + {parseUiState === 'failed' ? ( + + + + ) : ( + + )} PDF Layout Parse
-

{statusText}

+
+ + PP-DocLayout-V3 +
-
-
-

- {hasMeasuredProgress ? `Page ${pagesParsed} / ${totalPages}` : 'Awaiting first page'} -

-

{stageLabel}

+
+ {/* animated layout scanner — static "halted" view when failed */} +
+
-
-
-
-

- {hasMeasuredProgress ? `${Math.round(progressPercent)}% complete` : statusSubText} + + {/* live status + progress */} + {parseUiState === 'failed' ? ( +

+

{statusText}

+

{stageLabel}

+
+ ) : ( +
+
+

+ {hasMeasuredProgress ? `Page ${pagesParsed} / ${totalPages}` : 'Awaiting first page'} +

+

{stageLabel}

+
+
+
+
+

+ {hasMeasuredProgress ? `${Math.round(progressPercent)}% complete` : 'Calibrating layout pass'} +

+
+ )} +
+ + {/* attribution footer */} +
+ + + +

+ Classifying footnotes, titles, tables, figures, formulas, & more with PP-DocLayout-V3.

diff --git a/src/app/(app)/pdf/[id]/usePdfDocument.ts b/src/app/(app)/pdf/[id]/usePdfDocument.ts index 1720f63..b252618 100644 --- a/src/app/(app)/pdf/[id]/usePdfDocument.ts +++ b/src/app/(app)/pdf/[id]/usePdfDocument.ts @@ -291,6 +291,10 @@ export function usePdfDocument(): PdfDocumentState { pdfDocumentRef.current = pdfDocument; }, [pdfDocument]); + useEffect(() => { + pageTextCacheRef.current.clear(); + }, [parsedDocument, documentSettings.pdf?.skipBlockKinds]); + useEffect(() => { setCurrDocPage(currDocPageNumber); }, [currDocPageNumber]); @@ -538,7 +542,10 @@ export function usePdfDocument(): PdfDocumentState { if (!currDocId) return; try { const forced = await forceReparsePdfDocument(currDocId); + loadSeqRef.current += 1; + pageTextCacheRef.current.clear(); setParsedDocument(null); + setCurrDocText(undefined); setParseStatus(forced.status); setParseProgress(null); setActiveParseOpId(forced.opId ?? null); diff --git a/src/app/api/documents/[id]/parsed/events/route.ts b/src/app/api/documents/[id]/parsed/events/route.ts index 31af7fe..2649ae5 100644 --- a/src/app/api/documents/[id]/parsed/events/route.ts +++ b/src/app/api/documents/[id]/parsed/events/route.ts @@ -5,7 +5,7 @@ import { documents } from '@/db/schema'; import { requireAuthContext } from '@/lib/server/auth/auth'; import { getWorkerClientConfigFromEnv } from '@/lib/server/compute/worker'; import { isAbortLikeError } from '@/lib/server/compute/abort-like-error'; -import { snapshotFromWorkerState } from '@/lib/server/compute/worker-parse-state'; +import { isWorkerOperationStateStale, snapshotFromWorkerState } from '@/lib/server/compute/worker-parse-state'; import { fetchWorkerOperationState } from '@/lib/server/compute/worker-op-state'; import { isValidDocumentId } from '@/lib/server/documents/blobstore'; import { normalizeParseStatus, parseDocumentParseState } from '@/lib/server/documents/parse-state'; @@ -16,6 +16,7 @@ import { errorResponse } from '@/lib/server/errors/next-response'; import { logDegraded, logServerError } from '@/lib/server/errors/logging'; import type { PdfParseProgress, PdfParseStatus } from '@/types/parsed-pdf'; import { parseSseEventId, parseSsePayload } from '@openreader/compute-core'; +import { getComputeOpStaleMs } from '@openreader/compute-core'; import type { PdfLayoutJobResult, WorkerOperationEvent, WorkerOperationState } from '@openreader/compute-core/api-contracts'; export const dynamic = 'force-dynamic'; @@ -55,6 +56,7 @@ function sleep(ms: number): Promise { } async function toSnapshotState(row: ParseRow, preferredOpId?: string | null): Promise { + const opStaleMs = getComputeOpStaleMs(); const state = await healStaleDocumentParseState({ documentId: row.id, userId: row.userId, @@ -68,7 +70,11 @@ async function toSnapshotState(row: ParseRow, preferredOpId?: string | null): Pr // the per-user document row currently says "ready" or has a different opId. if (opId && (requestedOpId !== null || parseStatus !== 'ready')) { const workerState = await fetchWorkerOperationState(opId); - if (workerState && workerState.opId === opId) { + if ( + workerState + && workerState.opId === opId + && !isWorkerOperationStateStale(workerState, opStaleMs) + ) { return { snapshot: { ...snapshotFromWorkerState(workerState), @@ -325,119 +331,136 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string continue; } - workerAbort = new AbortController(); - const query = lastEventId && lastEventId > 0 - ? `?sinceEventId=${encodeURIComponent(String(lastEventId))}` - : ''; - const response = await fetch( - `${workerCfg.baseUrl}/ops/${encodeURIComponent(currentOpId)}/events${query}`, - { - method: 'GET', - headers: { - Authorization: `Bearer ${workerCfg.token}`, - Accept: 'text/event-stream', - ...(lastEventId && lastEventId > 0 ? { 'Last-Event-ID': String(lastEventId) } : {}), + try { + workerAbort = new AbortController(); + const query = lastEventId && lastEventId > 0 + ? `?sinceEventId=${encodeURIComponent(String(lastEventId))}` + : ''; + const response = await fetch( + `${workerCfg.baseUrl}/ops/${encodeURIComponent(currentOpId)}/events${query}`, + { + method: 'GET', + headers: { + Authorization: `Bearer ${workerCfg.token}`, + Accept: 'text/event-stream', + ...(lastEventId && lastEventId > 0 ? { 'Last-Event-ID': String(lastEventId) } : {}), + }, + cache: 'no-store', + signal: workerAbort.signal, }, - cache: 'no-store', - signal: workerAbort.signal, - }, - ); + ); - if (closed) return; + if (closed) return; - if (!response.ok) { - const upstreamResponseBody = await response.text().catch(() => ''); - logger.warn({ - event: 'documents.parsed.events.worker_stream_request_failed', - degraded: true, - step: 'worker_stream_request', - documentId: id, - opId: currentOpId, - status: response.status, - upstreamResponseBody, - error: { - name: 'WorkerStreamRequestFailed', - message: `Worker stream request failed with status ${response.status}`, - }, - }, 'Worker stream request failed'); - await sleep(500); - continue; - } - if (!response.body) { - logger.warn({ - event: 'documents.parsed.events.worker_stream_missing_body', - degraded: true, - step: 'worker_stream_body', - documentId: id, - opId: currentOpId, - error: { - name: 'WorkerStreamMissingBody', - message: 'Worker stream response missing body', - }, - }, 'Worker stream response missing body'); - await sleep(500); - continue; - } - - const reader = response.body.getReader(); - const decoder = new TextDecoder(); - let buffer = ''; - let streamEnded = false; - - while (!closed && !streamEnded) { - const read = await reader.read(); - if (read.done) { - streamEnded = true; - break; + if (!response.ok) { + const upstreamResponseBody = await response.text().catch(() => ''); + logger.warn({ + event: 'documents.parsed.events.worker_stream_request_failed', + degraded: true, + step: 'worker_stream_request', + documentId: id, + opId: currentOpId, + status: response.status, + upstreamResponseBody, + error: { + name: 'WorkerStreamRequestFailed', + message: `Worker stream request failed with status ${response.status}`, + }, + }, 'Worker stream request failed'); + await sleep(500); + continue; + } + if (!response.body) { + logger.warn({ + event: 'documents.parsed.events.worker_stream_missing_body', + degraded: true, + step: 'worker_stream_body', + documentId: id, + opId: currentOpId, + error: { + name: 'WorkerStreamMissingBody', + message: 'Worker stream response missing body', + }, + }, 'Worker stream response missing body'); + await sleep(500); + continue; } - buffer += decoder.decode(read.value, { stream: true }); + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ''; + let streamEnded = false; - while (true) { - const frameEnd = buffer.indexOf('\n\n'); - if (frameEnd < 0) break; - const frame = buffer.slice(0, frameEnd); - buffer = buffer.slice(frameEnd + 2); - - const eventId = parseSseEventId(frame); - if (eventId && eventId > 0) { - lastEventId = eventId; + while (!closed && !streamEnded) { + const read = await reader.read(); + if (read.done) { + streamEnded = true; + break; } - const payload = parseSsePayload(frame); - if (!payload) continue; + buffer += decoder.decode(read.value, { stream: true }); - let parsed: WorkerOperationEvent | WorkerOperationState; - try { - parsed = JSON.parse(payload) as WorkerOperationEvent | WorkerOperationState; - } catch { - continue; - } + while (true) { + const frameEnd = buffer.indexOf('\n\n'); + if (frameEnd < 0) break; + const frame = buffer.slice(0, frameEnd); + buffer = buffer.slice(frameEnd + 2); - const workerSnapshot: WorkerOperationState = ( - parsed && typeof parsed === 'object' && 'snapshot' in parsed - ? parsed.snapshot - : parsed as WorkerOperationState - ); - if (!workerSnapshot || workerSnapshot.opId !== currentOpId) continue; + const eventId = parseSseEventId(frame); + if (eventId && eventId > 0) { + lastEventId = eventId; + } - const nextSnapshot: ParsedSnapshot = { - ...snapshotFromWorkerState(workerSnapshot), - opId: workerSnapshot.opId, - }; - const nextSignature = JSON.stringify(nextSnapshot); - if (nextSignature !== signature) { - current = nextSnapshot; - signature = nextSignature; - currentFromWorker = true; - writeSnapshot(current); - } + const payload = parseSsePayload(frame); + if (!payload) continue; - if (shouldCloseForTerminalSnapshot(current, currentFromWorker)) { - closeStream(); - return; + let parsed: WorkerOperationEvent | WorkerOperationState; + try { + parsed = JSON.parse(payload) as WorkerOperationEvent | WorkerOperationState; + } catch { + continue; + } + + const workerSnapshot: WorkerOperationState = ( + parsed && typeof parsed === 'object' && 'snapshot' in parsed + ? parsed.snapshot + : parsed as WorkerOperationState + ); + if (!workerSnapshot || workerSnapshot.opId !== currentOpId) continue; + + const nextSnapshot: ParsedSnapshot = { + ...snapshotFromWorkerState(workerSnapshot), + opId: workerSnapshot.opId, + }; + const nextSignature = JSON.stringify(nextSnapshot); + if (nextSignature !== signature) { + current = nextSnapshot; + signature = nextSignature; + currentFromWorker = true; + writeSnapshot(current); + } + + if (shouldCloseForTerminalSnapshot(current, currentFromWorker)) { + closeStream(); + return; + } } } + } catch (error) { + if (closed || isAbortLikeError(error)) return; + logDegraded(logger, { + event: 'documents.parsed.events.worker_stream_read_failed', + msg: 'Worker stream read failed; reconnecting', + step: 'worker_stream_read', + context: { + documentId: id, + opId: currentOpId, + requestId, + }, + error, + }); + await sleep(500); + continue; } if (closed) return; diff --git a/src/app/api/documents/[id]/parsed/route.ts b/src/app/api/documents/[id]/parsed/route.ts index 9738b90..0a783f2 100644 --- a/src/app/api/documents/[id]/parsed/route.ts +++ b/src/app/api/documents/[id]/parsed/route.ts @@ -7,6 +7,7 @@ import { requireAuthContext } from '@/lib/server/auth/auth'; import { createOrReusePdfWorkerOperation } from '@/lib/server/compute/worker-op-create'; import { documentParseStateFromWorkerState, + isWorkerOperationStateStale, snapshotFromWorkerState, } from '@/lib/server/compute/worker-parse-state'; import { fetchWorkerOperationState } from '@/lib/server/compute/worker-op-state'; @@ -33,6 +34,7 @@ import { createRequestLogger, hashForLog, type ServerLogger } from '@/lib/server import { errorResponse } from '@/lib/server/errors/next-response'; import { logDegraded } from '@/lib/server/errors/logging'; import type { ParsedPdfDocument } from '@/types/parsed-pdf'; +import { getComputeOpStaleMs } from '@openreader/compute-core'; import type { PdfLayoutJobResult, WorkerOperationState } from '@openreader/compute-core/api-contracts'; export const dynamic = 'force-dynamic'; @@ -192,6 +194,7 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string request: req, }); try { + const opStaleMs = getComputeOpStaleMs(); if (!isS3Configured()) return s3NotConfiguredResponse(); const authCtxOrRes = await requireAuthContext(req); @@ -255,7 +258,11 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string if (effectiveOpId && effectiveStatus !== 'ready') { const workerState = await fetchWorkerOperationState(effectiveOpId); - if (workerState && workerState.opId === effectiveOpId) { + if ( + workerState + && workerState.opId === effectiveOpId + && !isWorkerOperationStateStale(workerState, opStaleMs) + ) { return finalizeFromWorkerState({ workerState, row, @@ -348,6 +355,7 @@ export async function POST(req: NextRequest, ctx: { params: Promise<{ id: string request: req, }); try { + const opStaleMs = getComputeOpStaleMs(); if (!isS3Configured()) return s3NotConfiguredResponse(); const authCtxOrRes = await requireAuthContext(req); @@ -388,7 +396,12 @@ export async function POST(req: NextRequest, ctx: { params: Promise<{ id: string const existingOpId = normalizeOpId(state.opId); if (existingOpId) { const existing = await fetchWorkerOperationState(existingOpId); - if (existing && (existing.status === 'queued' || existing.status === 'running') && !replace) { + if ( + existing + && !isWorkerOperationStateStale(existing, opStaleMs) + && (existing.status === 'queued' || existing.status === 'running') + && !replace + ) { const snapshot = snapshotFromWorkerState(existing); return NextResponse.json({ error: 'Parse operation already in progress', diff --git a/src/app/globals.css b/src/app/globals.css index 284d34c..1d69c28 100644 --- a/src/app/globals.css +++ b/src/app/globals.css @@ -424,3 +424,31 @@ h1, h2, h3, h4, h5, h6 { -webkit-mask: linear-gradient(90deg, transparent 0%, black 8%, black 92%, transparent 100%); border-radius: 999px; /* subtle rounding to reinforce taper */ } + +/* Generic determinate progress fill with a moving sheen. Apply to the colored + fill element (the one whose width tracks progress); its track should clip. */ +.progress-fill { + position: relative; + overflow: hidden; +} +.progress-fill::after { + content: ''; + position: absolute; + inset: 0; + background: linear-gradient( + 90deg, + transparent, + color-mix(in srgb, var(--background) 55%, transparent), + transparent + ); + transform: translateX(-100%); + animation: progress-sheen 2.2s linear infinite; +} +@keyframes progress-sheen { + 0% { transform: translateX(-100%); } + /* sweep across at constant speed, then rest off-screen so the loop reset + (from the right edge back to the left) is never visible */ + 65% { transform: translateX(220%); } + 100% { transform: translateX(220%); } +} + diff --git a/src/components/ProgressCard.tsx b/src/components/ProgressCard.tsx index f8341d0..2933c9d 100644 --- a/src/components/ProgressCard.tsx +++ b/src/components/ProgressCard.tsx @@ -66,7 +66,7 @@ export function ProgressCard({ {/* Progress bar */}
diff --git a/src/components/auth/UserMenu.tsx b/src/components/auth/UserMenu.tsx index d9e44ea..6f31aa5 100644 --- a/src/components/auth/UserMenu.tsx +++ b/src/components/auth/UserMenu.tsx @@ -1,6 +1,5 @@ 'use client'; -import Link from 'next/link'; import { useAuthConfig } from '@/contexts/AuthRateLimitContext'; import { useFeatureFlag } from '@/contexts/RuntimeConfigContext'; import { useAuthSession } from '@/hooks/useAuthSession'; @@ -35,21 +34,19 @@ export function UserMenu({ if (variant === 'sidebar') { return (
- + } + label="Connect" + /> + {enableUserSignups && ( } - label="Connect" + label="Create account" /> - - {enableUserSignups && ( - - } - label="Create account" - /> - )}
); diff --git a/src/components/documents/DexieMigrationModal.tsx b/src/components/documents/DexieMigrationModal.tsx index df469d1..2efa278 100644 --- a/src/components/documents/DexieMigrationModal.tsx +++ b/src/components/documents/DexieMigrationModal.tsx @@ -155,7 +155,7 @@ export function DexieMigrationModal({

{status}

-
+
)} diff --git a/src/components/reader/PdfLayoutScan.module.css b/src/components/reader/PdfLayoutScan.module.css new file mode 100644 index 0000000..5ee5a31 --- /dev/null +++ b/src/components/reader/PdfLayoutScan.module.css @@ -0,0 +1,211 @@ +/* Scanner visualization for the PDF layout-parse loader. Scoped via CSS Module + so it stays out of the global stylesheet and works across the helper that + renders region content. Used by PdfLayoutScan.tsx. */ + +.stage { + position: relative; + width: 100%; + aspect-ratio: 1 / 1; + display: grid; + place-items: center; + background: transparent; +} + +.page { + position: relative; + width: 68%; + aspect-ratio: 3 / 4; + border-radius: 6px; + background: linear-gradient(180deg, color-mix(in srgb, var(--background) 92%, var(--foreground)), var(--background)); + border: 1px solid var(--line); + box-shadow: + var(--elev-2), + 0 0 0 1px color-mix(in srgb, var(--background) 60%, transparent), + inset 0 1px 0 color-mix(in srgb, var(--foreground) 6%, transparent); + overflow: hidden; +} + +/* ── readout chip (in the stage band, above the page) ─────────────────── */ +.tagRow { + position: absolute; + top: 4%; + left: 0; + right: 0; + display: flex; + justify-content: center; + z-index: 5; + padding: 0 4%; +} +.tag { + display: inline-flex; + align-items: center; + max-width: 100%; + padding: 3px 9px; + border-radius: 6px; + font-family: var(--font-display), system-ui, sans-serif; + font-size: 10px; + font-weight: 700; + line-height: 1.25; + letter-spacing: 0.01em; + white-space: nowrap; + color: var(--background); + background: var(--accent); + box-shadow: 0 2px 10px color-mix(in srgb, var(--accent) 50%, transparent); + animation: tag-in 0.34s var(--ease); +} +@keyframes tag-in { + from { opacity: 0; transform: translateY(-4px) scale(0.92); } + to { opacity: 1; transform: translateY(0) scale(1); } +} + +/* ── failed / halted state ────────────────────────────────────────────── */ +.tagFailed { + color: var(--foreground); + background: color-mix(in srgb, var(--foreground) 12%, transparent); + box-shadow: none; + animation: none; +} + +.pageFailed { + display: grid; + place-items: center; + opacity: 0.55; + filter: grayscale(0.4); +} +.alert { + width: 38%; + height: 38%; + stroke: var(--accent); + stroke-width: 1.6; + color: var(--accent); +} + +/* ── region boxes (one shown at a time, centered) ─────────────────────── */ +.solos { + position: absolute; + inset: 8%; + display: grid; + place-items: center; +} +.solo { + grid-area: 1 / 1; /* stack so regions cross-fade in place */ + display: flex; + align-items: center; + justify-content: center; + border-radius: 6px; + border: 1.5px solid var(--accent); + background: color-mix(in srgb, var(--foreground) 8%, transparent); + box-shadow: + 0 0 0 1px var(--accent-line), + 0 0 22px color-mix(in srgb, var(--accent) 30%, transparent); + padding: 4px 7px; + opacity: 0; + transform: scale(0.9); + transition: opacity 0.32s var(--ease), transform 0.32s var(--ease); + will-change: opacity, transform; +} +.solo.active { + opacity: 1; + transform: scale(1); +} + +/* inner content so each region reads as real document material */ +.lines { + display: flex; + flex-direction: column; + justify-content: center; + gap: 11%; + width: 100%; + height: 100%; +} +.linesHeading { gap: 22%; } +.lines > span { + display: block; + height: 3px; + width: 100%; + border-radius: 999px; + background: color-mix(in srgb, var(--foreground) 26%, transparent); +} +.linesHeading > span { + height: 6px; + background: color-mix(in srgb, var(--foreground) 46%, transparent); +} + +.glyph { + width: 100%; + height: 100%; + stroke: color-mix(in srgb, var(--foreground) 36%, transparent); + stroke-width: 2; + stroke-linejoin: round; + fill: none; +} +.glyph circle { + fill: color-mix(in srgb, var(--foreground) 30%, transparent); + stroke: none; +} + +.table { + display: grid; + grid-template-columns: repeat(4, 1fr); + grid-template-rows: repeat(4, 1fr); + gap: 2px; + width: 100%; + height: 100%; + background: color-mix(in srgb, var(--foreground) 16%, transparent); + border: 1px solid color-mix(in srgb, var(--foreground) 16%, transparent); + border-radius: 3px; + overflow: hidden; +} +.table > span { + background: color-mix(in srgb, var(--background) 82%, var(--foreground)); +} +.table > span:nth-child(-n + 4) { + background: color-mix(in srgb, var(--foreground) 22%, transparent); +} + +/* ── scan beam ────────────────────────────────────────────────────────── */ +.beam { + position: absolute; + left: -4%; + right: -4%; + height: 2px; + border-radius: 999px; + background: linear-gradient(90deg, transparent, var(--accent), transparent); + box-shadow: 0 0 14px 2px color-mix(in srgb, var(--accent) 55%, transparent); + animation: beam 1.05s var(--ease) infinite; + will-change: top, opacity; +} +.beam::after { + content: ''; + position: absolute; + left: 0; + right: 0; + bottom: 1px; + height: 30px; + background: linear-gradient(to top, color-mix(in srgb, var(--accent) 18%, transparent), transparent); +} +@keyframes beam { + 0% { top: -4%; opacity: 0; } + 8% { opacity: 1; } + 90% { opacity: 1; } + 100% { top: 104%; opacity: 0; } +} + +/* camera-style corner reticles */ +.corner { + position: absolute; + width: 10px; + height: 10px; + border: 1.5px solid var(--accent-line); + pointer-events: none; +} +.cornerTl { top: 5px; left: 5px; border-right: 0; border-bottom: 0; border-top-left-radius: 3px; } +.cornerTr { top: 5px; right: 5px; border-left: 0; border-bottom: 0; border-top-right-radius: 3px; } +.cornerBl { bottom: 5px; left: 5px; border-right: 0; border-top: 0; border-bottom-left-radius: 3px; } +.cornerBr { bottom: 5px; right: 5px; border-left: 0; border-top: 0; border-bottom-right-radius: 3px; } + +@media (prefers-reduced-motion: reduce) { + .beam { animation: none; opacity: 0; } + .tag { animation: none; } + .solo { transition: none; } +} diff --git a/src/components/reader/PdfLayoutScan.tsx b/src/components/reader/PdfLayoutScan.tsx new file mode 100644 index 0000000..7f46d00 --- /dev/null +++ b/src/components/reader/PdfLayoutScan.tsx @@ -0,0 +1,172 @@ +'use client'; + +import { useEffect, useState } from 'react'; +import styles from './PdfLayoutScan.module.css'; + +/** + * PdfLayoutScan — ambient visualization of PP-DocLayout-V3 at work. + * + * A miniature document page sits directly on the loader's dotted background + * while a scan beam sweeps across it. One layout region is "detected" at a time + * — a box rendered with one + * of a few simple shapes — while a readout chip above the page names its class. + * It cycles through the complete set of region classes PP-DocLayout-V3 emits + * (see PARSED_PDF_BLOCK_KINDS in types/parsed-pdf). Purely decorative; honours + * prefers-reduced-motion by freezing on a single region. Styles live in the + * adjacent CSS module so they stay out of the global stylesheet. + * + * When `failed` is set the animation is replaced by a static "halted" view — + * the beam stops, the page dims, and an alert glyph sits on it — so the loader + * never implies active work after a parse failure. + */ + +// Only a handful of real shapes — most regions are just text lines. +type BlockShape = 'heading' | 'text' | 'small' | 'image' | 'table'; + +interface ScanBlock { + label: string; + shape: BlockShape; +} + +// One consistent box size per shape, so every region renders the same way and +// nothing gets clipped — only the label changes. +const SHAPE_SIZE: Record = { + heading: { width: 82, height: 22 }, + text: { width: 88, height: 66 }, + small: { width: 72, height: 24 }, + image: { width: 80, height: 68 }, + table: { width: 88, height: 64 }, +}; + +// Every PP-DocLayout-V3 class, in document-flow order, mapped to a simple shape. +const SCAN_BLOCKS: ScanBlock[] = [ + { label: 'Header', shape: 'small' }, + { label: 'Doc title', shape: 'heading' }, + { label: 'Abstract', shape: 'text' }, + { label: 'Paragraph title', shape: 'heading' }, + { label: 'Text', shape: 'text' }, + { label: 'Formula', shape: 'text' }, + { label: 'Formula number', shape: 'small' }, + { label: 'Figure title', shape: 'small' }, + { label: 'Image', shape: 'image' }, + { label: 'Chart', shape: 'image' }, + { label: 'Table', shape: 'table' }, + { label: 'Algorithm', shape: 'text' }, + { label: 'Content', shape: 'text' }, + { label: 'Aside text', shape: 'text' }, + { label: 'Number', shape: 'small' }, + { label: 'Reference', shape: 'small' }, + { label: 'Reference content', shape: 'text' }, + { label: 'Footnote', shape: 'small' }, + { label: 'Vision footnote', shape: 'small' }, + { label: 'Seal', shape: 'image' }, + { label: 'Footer', shape: 'small' }, +]; + +const STEP_MS = 1050; + +const cx = (...parts: Array) => parts.filter(Boolean).join(' '); + +function BlockContent({ shape }: { shape: BlockShape }) { + if (shape === 'image') { + return ( + + + + + ); + } + if (shape === 'table') { + return ( +
+ {Array.from({ length: 16 }).map((_, i) => ( + + ))} +
+ ); + } + const isHeading = shape === 'heading'; + const lines = isHeading ? 2 : shape === 'small' ? 2 : 5; + return ( +
+ {Array.from({ length: lines }).map((_, i) => ( + + ))} +
+ ); +} + +export function PdfLayoutScan({ failed = false }: { failed?: boolean }) { + const [active, setActive] = useState(0); + + useEffect(() => { + if (typeof window === 'undefined' || failed) return; // no cycling once halted + const reduced = window.matchMedia?.('(prefers-reduced-motion: reduce)').matches; + if (reduced) return; // freeze on the first region for reduced-motion users + const id = window.setInterval(() => { + setActive((i) => (i + 1) % SCAN_BLOCKS.length); + }, STEP_MS); + return () => window.clearInterval(id); + }, [failed]); + + if (failed) { + return ( +
+
+ Parse halted +
+ +
+ + + + + + + + + + +
+
+ ); + } + + return ( +
+ {/* readout chip: floats in the stage band above the page so long class + names are never clipped by the page's rounded overflow */} +
+ + {SCAN_BLOCKS[active].label} + +
+ +
+
+ {SCAN_BLOCKS.map((block, i) => { + const size = SHAPE_SIZE[block.shape]; + return ( +
+ +
+ ); + })} +
+ +
+ + + + +
+
+ ); +} diff --git a/src/components/ui/range.module.css b/src/components/ui/range.module.css new file mode 100644 index 0000000..4191c45 --- /dev/null +++ b/src/components/ui/range.module.css @@ -0,0 +1,115 @@ +/* Range slider — "precision gauge": a hairline rail with ruler notches per step + and a slim accent needle instead of a ball. Scoped via CSS Module so it stays + out of the global stylesheet. Per-instance values (--range-progress, + --range-tick-size, --range-tick-color) come from inline styles set in + range.tsx. Used by RangeInput. */ + +.range { + --range-track-h: 2px; + --range-needle-h: 1rem; + --range-needle-w: 3px; + --range-tick-size: 25%; /* segment width; only used when ticks are enabled */ + --range-tick-color: transparent; /* set to a real color inline for discrete sliders */ + --range-fill: var(--muted); + --range-empty: color-mix(in srgb, var(--offbase) 82%, var(--background)); + -webkit-appearance: none; + appearance: none; + width: 100%; + height: 1.25rem; + background: transparent; + cursor: pointer; +} +.range:disabled { + cursor: not-allowed; + opacity: 0.6; +} +.range:focus-visible { + outline: none; +} + +/* Ticks ride above the muted fill / empty rail (first listed background = top). */ +.range::-webkit-slider-runnable-track { + height: var(--range-track-h); + border-radius: 999px; + background: + repeating-linear-gradient( + to right, + var(--range-tick-color) 0, + var(--range-tick-color) 1.5px, + transparent 1.5px, + transparent var(--range-tick-size) + ), + linear-gradient( + to right, + var(--range-fill) 0, + var(--range-fill) var(--range-progress), + var(--range-empty) var(--range-progress), + var(--range-empty) 100% + ); +} +.range::-webkit-slider-thumb { + -webkit-appearance: none; + appearance: none; + width: var(--range-needle-w); + height: var(--range-needle-h); + margin-top: calc((var(--range-track-h) - var(--range-needle-h)) / 2); + border-radius: 999px; + background: var(--accent); + box-shadow: + 0 0 0 1px color-mix(in srgb, var(--background) 70%, transparent), + 0 1px 6px color-mix(in srgb, var(--accent) 55%, transparent); + transition: transform 150ms ease, box-shadow 150ms ease; +} +.range:hover::-webkit-slider-thumb { + transform: scaleY(1.3); +} +.range:active::-webkit-slider-thumb { + transform: scaleY(1.55); +} +.range:focus-visible::-webkit-slider-thumb { + box-shadow: + 0 0 0 4px color-mix(in srgb, var(--accent) 28%, transparent), + 0 1px 6px color-mix(in srgb, var(--accent) 55%, transparent); +} + +/* Firefox: track + native progress fill + matching needle. */ +.range::-moz-range-track { + height: var(--range-track-h); + border-radius: 999px; + background: + repeating-linear-gradient( + to right, + var(--range-tick-color) 0, + var(--range-tick-color) 1.5px, + transparent 1.5px, + transparent var(--range-tick-size) + ), + var(--range-empty); +} +.range::-moz-range-progress { + height: var(--range-track-h); + border-radius: 999px; + background: var(--range-fill); +} +.range::-moz-range-thumb { + width: var(--range-needle-w); + height: var(--range-needle-h); + border: 0; + border-radius: 999px; + background: var(--accent); + box-shadow: + 0 0 0 1px color-mix(in srgb, var(--background) 70%, transparent), + 0 1px 6px color-mix(in srgb, var(--accent) 55%, transparent); + transition: transform 150ms ease, box-shadow 150ms ease; +} +.range:hover::-moz-range-thumb { + transform: scaleY(1.3); +} +.range:active::-moz-range-thumb { + transform: scaleY(1.55); +} +.range:focus-visible::-moz-range-thumb { + box-shadow: + 0 0 0 4px color-mix(in srgb, var(--accent) 28%, transparent), + 0 1px 6px color-mix(in srgb, var(--accent) 55%, transparent); +} diff --git a/src/components/ui/range.tsx b/src/components/ui/range.tsx index 033fadb..e0f1ae0 100644 --- a/src/components/ui/range.tsx +++ b/src/components/ui/range.tsx @@ -1,8 +1,11 @@ import type { CSSProperties, InputHTMLAttributes } from 'react'; import { cn } from './cn'; +import styles from './range.module.css'; type RangeStyle = CSSProperties & { '--range-progress'?: string; + '--range-tick-size'?: string; + '--range-tick-color'?: string; }; function toNumber(value: string | number | readonly string[] | undefined, fallback: number): number { @@ -24,30 +27,16 @@ function resolveRangeProgress(props: InputHTMLAttributes): num return Math.min(100, Math.max(0, progress)); } -const rangeInputClass = cn( - 'h-6 w-full cursor-pointer appearance-none bg-transparent [--range-track-h:0.5rem] [--range-thumb-h:1.25rem]', - 'focus-visible:outline-none focus-visible:[&::-webkit-slider-thumb]:ring-4 focus-visible:[&::-webkit-slider-thumb]:ring-accent/25', - 'focus-visible:[&::-moz-range-thumb]:ring-4 focus-visible:[&::-moz-range-thumb]:ring-accent/25', - 'disabled:cursor-not-allowed disabled:opacity-60', - '[&::-webkit-slider-runnable-track]:h-[var(--range-track-h)] [&::-webkit-slider-runnable-track]:rounded-full', - '[&::-webkit-slider-runnable-track]:bg-[linear-gradient(to_right,var(--secondary-accent)_0%,var(--secondary-accent)_var(--range-progress),color-mix(in_srgb,var(--offbase)_82%,var(--background))_var(--range-progress),color-mix(in_srgb,var(--offbase)_82%,var(--background))_100%)]', - '[&::-webkit-slider-runnable-track]:shadow-[inset_0_1px_0_color-mix(in_srgb,var(--foreground)_10%,transparent)]', - '[&::-webkit-slider-thumb]:mt-[calc((var(--range-track-h)-var(--range-thumb-h))/2)] [&::-webkit-slider-thumb]:h-[var(--range-thumb-h)] [&::-webkit-slider-thumb]:w-[var(--range-thumb-h)] [&::-webkit-slider-thumb]:appearance-none [&::-webkit-slider-thumb]:rounded-full', - '[&::-webkit-slider-thumb]:border-2 [&::-webkit-slider-thumb]:border-[color-mix(in_srgb,var(--background)_78%,white)]', - '[&::-webkit-slider-thumb]:bg-accent', - '[&::-webkit-slider-thumb]:transition-transform [&::-webkit-slider-thumb]:duration-150 [&::-webkit-slider-thumb]:ease-out', - 'active:[&::-webkit-slider-thumb]:scale-[1.07]', - '[&::-moz-range-track]:h-[var(--range-track-h)] [&::-moz-range-track]:rounded-full [&::-moz-range-track]:border-0', - '[&::-moz-range-track]:bg-[color-mix(in_srgb,var(--offbase)_82%,var(--background))]', - '[&::-moz-range-progress]:h-[var(--range-track-h)] [&::-moz-range-progress]:rounded-full [&::-moz-range-progress]:border-0 [&::-moz-range-progress]:bg-secondary-accent', - '[&::-moz-range-thumb]:h-[var(--range-thumb-h)] [&::-moz-range-thumb]:w-[var(--range-thumb-h)] [&::-moz-range-thumb]:appearance-none [&::-moz-range-thumb]:rounded-full [&::-moz-range-thumb]:border-2', - '[&::-moz-range-thumb]:border-[color-mix(in_srgb,var(--background)_78%,white)] [&::-moz-range-thumb]:bg-accent', - '[&::-moz-range-thumb]:transition-transform [&::-moz-range-thumb]:duration-150 [&::-moz-range-thumb]:ease-out', - 'active:[&::-moz-range-thumb]:scale-[1.07]', -); - -function rangeInputClassName(className?: string) { - return cn(rangeInputClass, className); +// Discrete sliders get a ruler: one notch per step. Returns the segment width +// (as a %) for the repeating tick gradient, or null for continuous/large ranges. +function resolveTickSize(props: InputHTMLAttributes): string | null { + const min = toNumber(props.min, 0); + const max = toNumber(props.max, 100); + const step = toNumber(props.step, 1); + if (step <= 0) return null; + const segments = Math.round((max - min) / step); + if (!Number.isFinite(segments) || segments < 2 || segments > 24) return null; + return `${100 / segments}%`; } export function RangeInput({ @@ -55,10 +44,17 @@ export function RangeInput({ style, ...props }: Omit, 'type'>) { + const tickSize = resolveTickSize(props); const rangeStyle: RangeStyle = { ...style, '--range-progress': `${resolveRangeProgress(props)}%`, + ...(tickSize + ? { + '--range-tick-size': tickSize, + '--range-tick-color': 'color-mix(in srgb, var(--foreground) 22%, transparent)', + } + : {}), }; - return ; + return ; } diff --git a/src/components/ui/sidebar-nav.tsx b/src/components/ui/sidebar-nav.tsx index b556d96..9ff0da7 100644 --- a/src/components/ui/sidebar-nav.tsx +++ b/src/components/ui/sidebar-nav.tsx @@ -1,4 +1,5 @@ -import { forwardRef, type AnchorHTMLAttributes, type ButtonHTMLAttributes, type HTMLAttributes, type ReactNode } from 'react'; +import Link from 'next/link'; +import { forwardRef, type ButtonHTMLAttributes, type ComponentPropsWithoutRef, type HTMLAttributes, type ReactNode } from 'react'; import { cn } from './cn'; import { focusRing, motionColors } from './tokens'; @@ -170,7 +171,7 @@ export function SidebarNavItem({ ); } -export const SidebarNavLink = forwardRef & { +type SidebarNavLinkProps = ComponentPropsWithoutRef & { active?: boolean; icon?: ReactNode; label?: ReactNode; @@ -179,7 +180,9 @@ export const SidebarNavLink = forwardRef(function SidebarNavLink({ +}; + +export const SidebarNavLink = forwardRef(function SidebarNavLink({ active = false, icon, label, @@ -193,7 +196,7 @@ export const SidebarNavLink = forwardRef {children} - + ); }); diff --git a/src/lib/server/compute/worker-parse-state.ts b/src/lib/server/compute/worker-parse-state.ts index 0e2b861..cce75a4 100644 --- a/src/lib/server/compute/worker-parse-state.ts +++ b/src/lib/server/compute/worker-parse-state.ts @@ -2,6 +2,10 @@ import type { PdfLayoutJobResult, WorkerOperationState } from '@openreader/compu import type { PdfParseProgress, PdfParseStatus } from '@/types/parsed-pdf'; import type { DocumentParseState } from '@/lib/server/documents/parse-state'; +function isInflightWorkerStatus(status: WorkerOperationState['status']): boolean { + return status === 'queued' || status === 'running'; +} + export function mapWorkerStatusToParseStatus(status: WorkerOperationState['status']): PdfParseStatus { switch (status) { case 'queued': @@ -44,6 +48,18 @@ export function documentParseStateFromWorkerState( }; } +export function isWorkerOperationStateStale( + state: WorkerOperationState, + staleMs: number, + nowMs = Date.now(), +): boolean { + if (!isInflightWorkerStatus(state.status)) return false; + if (!Number.isFinite(staleMs) || staleMs <= 0) return false; + const updatedAt = Number(state.updatedAt ?? 0); + if (!Number.isFinite(updatedAt) || updatedAt <= 0) return false; + return (nowMs - updatedAt) > staleMs; +} + export function mergeNonReadyParseSnapshot(input: { parseStatus: PdfParseStatus; parseProgress: PdfParseProgress | null; diff --git a/tests/unit/pdf-merge-text-with-regions.vitest.spec.ts b/tests/unit/pdf-merge-text-with-regions.vitest.spec.ts index 135e9c3..449824e 100644 --- a/tests/unit/pdf-merge-text-with-regions.vitest.spec.ts +++ b/tests/unit/pdf-merge-text-with-regions.vitest.spec.ts @@ -34,4 +34,19 @@ describe('mergeTextWithRegions', () => { expect(merged).toHaveLength(1); expect(merged[0].text).toBe('inside'); }); + + test('keeps decorative drop caps attached to the same line when boxes overlap vertically', () => { + const regions = [ + { bbox: [0, 0, 200, 120] as [number, number, number, number], label: 'text' as const }, + ]; + + const textItems = [ + { text: 'I', x: 0, y: 10, width: 12, height: 60 }, + { text: 't’s funny,', x: 12, y: 40, width: 50, height: 12 }, + ]; + + const merged = mergeTextWithRegions(regions, textItems); + expect(merged).toHaveLength(1); + expect(merged[0].text).toBe('It’s funny,'); + }); }); diff --git a/tests/unit/pdf-parse-normalize-text-items.vitest.spec.ts b/tests/unit/pdf-parse-normalize-text-items.vitest.spec.ts index 3ceccd0..b61bd26 100644 --- a/tests/unit/pdf-parse-normalize-text-items.vitest.spec.ts +++ b/tests/unit/pdf-parse-normalize-text-items.vitest.spec.ts @@ -32,7 +32,12 @@ describe('normalizeTextItemsForLayout', () => { [0, 10, -10, 0, 30, 400], ); - const normalized = normalizeTextItemsForLayout([horizontal, rotated], 800); + const normalized = normalizeTextItemsForLayout([horizontal, rotated], { + height: 800, + transform: [1, 0, 0, -1, 0, 800], + }, { + test: { ascent: 0.8, descent: -0.2 }, + }); expect(normalized).toHaveLength(1); expect(normalized[0]?.text).toBe('Powered by large language models'); }); @@ -41,7 +46,49 @@ describe('normalizeTextItemsForLayout', () => { const vertical = makeTextItem('Side label', [0, 10, -10, 0, 30, 200]); const skewed = makeTextItem('Watermark', [10, 5, 2, 10, 200, 500]); - const normalized = normalizeTextItemsForLayout([vertical, skewed], 800); + const normalized = normalizeTextItemsForLayout([vertical, skewed], { + height: 800, + transform: [1, 0, 0, -1, 0, 800], + }, { + test: { ascent: 0.8, descent: -0.2 }, + }); expect(normalized).toEqual([]); }); + + test('accounts for non-zero page origins in the viewport transform', () => { + const croppedPageLine = makeTextItem( + 'Vasher turned away.', + [11.2, 0, 0, 11.2, 127.5, 644.4128], + 100, + ); + + const normalized = normalizeTextItemsForLayout([croppedPageLine], { + height: 666.0074, + transform: [1, 0, 0, -1, -53.4352, 720.565], + }, { + test: { ascent: 0.716, descent: -0.269 }, + }); + + expect(normalized).toHaveLength(1); + expect(normalized[0]?.x).toBeCloseTo(74.0648, 4); + expect(normalized[0]?.y).toBeCloseTo(68.1356, 2); + }); + + test('uses font ascent to place decorative initials closer to the visible glyph top', () => { + const dropCap = makeTextItem( + 'I', + [60, 0, 0, 60, 111.1326, 434.46], + 9.18, + ); + + const normalized = normalizeTextItemsForLayout([dropCap], { + height: 666.0074, + transform: [1, 0, 0, -1, -53.4352, 720.565], + }, { + test: { ascent: 0.638, descent: -0.134 }, + }); + + expect(normalized).toHaveLength(1); + expect(normalized[0]?.y).toBeCloseTo(247.825, 3); + }); }); diff --git a/tests/unit/worker-parse-state.vitest.spec.ts b/tests/unit/worker-parse-state.vitest.spec.ts index 062cf87..dfdd37e 100644 --- a/tests/unit/worker-parse-state.vitest.spec.ts +++ b/tests/unit/worker-parse-state.vitest.spec.ts @@ -2,6 +2,7 @@ import { describe, expect, test } from 'vitest'; import type { PdfLayoutJobResult, WorkerOperationState } from '@openreader/compute-core/api-contracts'; import { documentParseStateFromWorkerState, + isWorkerOperationStateStale, snapshotFromWorkerState, } from '../../src/lib/server/compute/worker-parse-state'; @@ -88,4 +89,30 @@ describe('worker parse state mapping', () => { error: 'layout model crashed', }); }); + + test('treats old inflight worker states as stale', () => { + const workerState = makeWorkerState({ + status: 'running', + updatedAt: 1_000, + progress: { + totalPages: 500, + pagesParsed: 250, + currentPage: 251, + phase: 'infer', + }, + }); + + expect(isWorkerOperationStateStale(workerState, 5_000, 6_001)).toBe(true); + expect(isWorkerOperationStateStale(workerState, 5_000, 5_999)).toBe(false); + }); + + test('never treats terminal worker states as stale', () => { + const failedState = makeWorkerState({ + status: 'failed', + updatedAt: 1_000, + error: { code: 'PDF_PARSE_FAILED', message: 'crashed' }, + }); + + expect(isWorkerOperationStateStale(failedState, 5_000, 99_999)).toBe(false); + }); });