From 09e8898683284536fb48a1f153830961c401dad3 Mon Sep 17 00:00:00 2001 From: Richard R Date: Fri, 22 May 2026 01:21:18 -0600 Subject: [PATCH] feat(pdf,config): implement stale parse-state healing and unify op stale window config Introduce automatic healing for stale PDF parse states in API routes. Add `isDocumentParseStateStale` utility and use a shared `COMPUTE_OP_STALE_MS` environment variable (with config getter) to control the stale window for both worker op replacement and app-side parse-state healing. Update documentation and environment variable references to reflect new config. Enables automatic marking of stuck parses as failed, improving reliability and retry behavior in distributed compute environments. --- .env.example | 1 + compute/core/src/config/timeout.ts | 12 ++++++ compute/core/src/index.ts | 1 + compute/worker/src/server.ts | 6 +-- docs-site/docs/deploy/compute-worker.md | 8 ++++ .../docs/reference/environment-variables.md | 12 ++++++ .../api/documents/[id]/parsed/events/route.ts | 38 ++++++++++++++++-- src/app/api/documents/[id]/parsed/route.ts | 39 ++++++++++++++++++- src/lib/server/documents/parse-state.ts | 16 ++++++++ 9 files changed, 123 insertions(+), 10 deletions(-) diff --git a/.env.example b/.env.example index cb40fba..34f84d4 100644 --- a/.env.example +++ b/.env.example @@ -93,6 +93,7 @@ COMPUTE_MODE=local # COMPUTE_JOB_CONCURRENCY=1 # COMPUTE_WHISPER_TIMEOUT_MS=30000 # COMPUTE_PDF_TIMEOUT_MS=300000 +# COMPUTE_OP_STALE_MS=1800000 # Worker mode requires worker-reachable shared object storage. # Non-exposed embedded weed mini is not supported in worker mode. diff --git a/compute/core/src/config/timeout.ts b/compute/core/src/config/timeout.ts index cbd133e..40fd406 100644 --- a/compute/core/src/config/timeout.ts +++ b/compute/core/src/config/timeout.ts @@ -1,6 +1,7 @@ const DEFAULT_COMPUTE_WHISPER_TIMEOUT_MS = 30_000; const DEFAULT_COMPUTE_PDF_TIMEOUT_MS = 300_000; const DEFAULT_COMPUTE_PDF_HARD_CAP_MS = 24 * 60 * 60 * 1000; +const DEFAULT_COMPUTE_OP_STALE_MIN_MS = 30 * 60_000; const DEFAULT_WORKER_WAIT_BUFFER_MS = 15_000; const DEFAULT_WORKER_WAIT_MIN_MS = 60_000; @@ -27,6 +28,7 @@ function readPositiveIntEnv(name: string, fallback: number): number { } let timeoutConfigCache: ComputeTimeoutConfig | null = null; +let opStaleMsCache: number | null = null; export function getComputeTimeoutConfig(): ComputeTimeoutConfig { if (timeoutConfigCache) return timeoutConfigCache; @@ -38,6 +40,16 @@ export function getComputeTimeoutConfig(): ComputeTimeoutConfig { return timeoutConfigCache; } +export function getComputeOpStaleMs(): number { + if (typeof opStaleMsCache === 'number') return opStaleMsCache; + const config = getComputeTimeoutConfig(); + opStaleMsCache = readPositiveIntEnv( + 'COMPUTE_OP_STALE_MS', + Math.max(DEFAULT_COMPUTE_OP_STALE_MIN_MS, Math.max(config.whisperTimeoutMs, config.pdfTimeoutMs) * 4), + ); + return opStaleMsCache; +} + export function getWorkerClientWaitTimeoutMs(kind: ComputeOperationKind): number { const config = getComputeTimeoutConfig(); const timeoutMs = kind === 'pdf_layout' ? config.pdfTimeoutMs : config.whisperTimeoutMs; diff --git a/compute/core/src/index.ts b/compute/core/src/index.ts index ebd947e..74f96a6 100644 --- a/compute/core/src/index.ts +++ b/compute/core/src/index.ts @@ -6,6 +6,7 @@ export { } from './config/cpu-budget'; export { getComputeTimeoutConfig, + getComputeOpStaleMs, getWorkerClientWaitTimeoutMs, withTimeout, withIdleTimeoutAndHardCap, diff --git a/compute/worker/src/server.ts b/compute/worker/src/server.ts index 176304f..0f1de24 100644 --- a/compute/worker/src/server.ts +++ b/compute/worker/src/server.ts @@ -28,6 +28,7 @@ import { } from '@openreader/compute-core/local-runtime'; import { getComputeTimeoutConfig, + getComputeOpStaleMs, getAvailableCpuCores, getOnnxThreadsPerJob, withIdleTimeoutAndHardCap, @@ -432,10 +433,7 @@ async function main(): Promise { const prewarmModels = parseBoolEnv('COMPUTE_PREWARM_MODELS', true); const jobsStreamMaxBytes = readIntEnv('COMPUTE_JOBS_STREAM_MAX_BYTES', 256 * 1024 * 1024); const jobStatesMaxBytes = readIntEnv('COMPUTE_JOB_STATES_MAX_BYTES', 64 * 1024 * 1024); - const opStaleMs = readIntEnv( - 'COMPUTE_OP_STALE_MS', - Math.max(30 * 60_000, Math.max(whisperTimeoutMs, pdfTimeoutMs) * 4), - ); + const opStaleMs = getComputeOpStaleMs(); const connectOpts: Parameters[0] = { servers: natsUrl }; const natsCreds = process.env.NATS_CREDS?.trim(); diff --git a/docs-site/docs/deploy/compute-worker.md b/docs-site/docs/deploy/compute-worker.md index f4fc432..26f3d44 100644 --- a/docs-site/docs/deploy/compute-worker.md +++ b/docs-site/docs/deploy/compute-worker.md @@ -72,10 +72,18 @@ COMPUTE_WORKER_TOKEN= # Optional shared timeout overrides (keep equal to worker service values): # COMPUTE_WHISPER_TIMEOUT_MS=30000 # COMPUTE_PDF_TIMEOUT_MS=300000 +# COMPUTE_OP_STALE_MS=1800000 ``` Model artifact overrides (`WHISPER_MODEL_BASE_URL`, `PDF_LAYOUT_MODEL_BASE_URL`) are worker runtime variables and should be set on the compute worker service environment. +`COMPUTE_OP_STALE_MS` is shared by both services in worker mode: + +- Worker: opKey stale replacement window in compute op state. +- App server: stale PDF parse-state healing window (`/api/documents/[id]/parsed*`). + +Set the same value on app + worker envs. + `COMPUTE_MODE=worker` has no local fallback. If worker is unavailable, affected requests fail. ## Production notes diff --git a/docs-site/docs/reference/environment-variables.md b/docs-site/docs/reference/environment-variables.md index bd6c6c4..07d4ec1 100644 --- a/docs-site/docs/reference/environment-variables.md +++ b/docs-site/docs/reference/environment-variables.md @@ -59,6 +59,7 @@ For auth-enabled deployments, use **Settings → Admin** as the primary source o | `COMPUTE_JOB_CONCURRENCY` | Heavy compute backend | `1` | Local in-process compute concurrency cap; set worker service concurrency in compute-worker env/docs | | `COMPUTE_WHISPER_TIMEOUT_MS` | Heavy compute backend | `30000` | Shared whisper alignment timeout budget (local + worker + worker client wait budget) | | `COMPUTE_PDF_TIMEOUT_MS` | Heavy compute backend | `300000` | Shared PDF idle-timeout budget (local + worker + worker client wait budget) | +| `COMPUTE_OP_STALE_MS` | Heavy compute backend | `max(30m, 4x max compute timeout)` | Shared stale window for worker op replacement and app-side stale PDF parse-state healing | | `PDF_LAYOUT_MODEL_BASE_URL` | PDF layout model | PP-DocLayoutV3 ONNX base URL | Optional base URL override for `ensureModel()` | | `WHISPER_MODEL_BASE_URL` | Whisper ONNX model | onnx-community defaults | Optional base URL override for ONNX whisper-base_timestamped int8 downloads | | `FFMPEG_BIN` | Audio runtime | auto-detected (`ffmpeg-static`) | Override ffmpeg binary path | @@ -408,6 +409,17 @@ Shared PDF idle-timeout budget in milliseconds. - Worker compute PDF runtime (idle timeout) - App server worker-client wait budget (SSE wait timeout) +### COMPUTE_OP_STALE_MS + +Shared stale window in milliseconds. + +- Default: `max(30m, 4x max(COMPUTE_WHISPER_TIMEOUT_MS, COMPUTE_PDF_TIMEOUT_MS))` +- Used by: + - Worker op reuse/replacement guard (`/ops` opKey stale detection) + - App-server PDF parse-state stale healing in `/api/documents/[id]/parsed*` +- If a parse row is stuck in `pending`/`running` past this window, app routes mark it failed so retries/reparse can proceed. +- In `COMPUTE_MODE=worker`, keep this value aligned on both app-server and worker service envs. + ### PDF_LAYOUT_MODEL_BASE_URL Optional base URL override for PP-DocLayoutV3 artifacts downloaded by `ensureModel()`. diff --git a/src/app/api/documents/[id]/parsed/events/route.ts b/src/app/api/documents/[id]/parsed/events/route.ts index 76c4955..6853160 100644 --- a/src/app/api/documents/[id]/parsed/events/route.ts +++ b/src/app/api/documents/[id]/parsed/events/route.ts @@ -8,7 +8,13 @@ import { getOpenReaderTestNamespace, getUnclaimedUserIdForNamespace } from '@/li import { isS3Configured } from '@/lib/server/storage/s3'; import type { PdfParseProgress, PdfParseStatus } from '@/types/parsed-pdf'; import { isValidDocumentId } from '@/lib/server/documents/blobstore'; -import { normalizeParseStatus, parseDocumentParseState } from '@/lib/server/documents/parse-state'; +import { + isDocumentParseStateStale, + normalizeParseStatus, + parseDocumentParseState, + stringifyDocumentParseState, +} from '@/lib/server/documents/parse-state'; +import { getComputeOpStaleMs } from '@openreader/compute-core'; export const dynamic = 'force-dynamic'; @@ -36,8 +42,32 @@ function sleep(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); } -function toSnapshot(row: ParseRow): ParsedSnapshot { +async function maybeHealStaleParseState(row: ParseRow): Promise { const state = parseDocumentParseState(row.parseState); + const staleMs = getComputeOpStaleMs(); + if (!isDocumentParseStateStale(state, staleMs)) return row; + + const nextState = stringifyDocumentParseState({ + status: 'failed', + progress: null, + updatedAt: Date.now(), + error: `Parse state stale for more than ${staleMs}ms; marked failed for retry`, + }); + + await db + .update(documents) + .set({ parseState: nextState }) + .where(and(eq(documents.id, row.id), eq(documents.userId, row.userId))); + + return { + ...row, + parseState: nextState, + }; +} + +async function toSnapshot(row: ParseRow): Promise { + const healed = await maybeHealStaleParseState(row); + const state = parseDocumentParseState(healed.parseState); const parseStatus = normalizeParseStatus(state.status); return { parseStatus, @@ -90,7 +120,7 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string return NextResponse.json({ error: 'Not found' }, { status: 404 }); } - const initial = toSnapshot(row); + const initial = await toSnapshot(row); if (initial.parseStatus === 'pending') { enqueueParsePdfJob({ documentId: id, @@ -126,7 +156,7 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string }); if (!nextRow) break; - const next = toSnapshot(nextRow); + const next = await toSnapshot(nextRow); if (next.parseStatus === 'pending') { enqueueParsePdfJob({ documentId: id, diff --git a/src/app/api/documents/[id]/parsed/route.ts b/src/app/api/documents/[id]/parsed/route.ts index 4c38631..c2e6c13 100644 --- a/src/app/api/documents/[id]/parsed/route.ts +++ b/src/app/api/documents/[id]/parsed/route.ts @@ -12,6 +12,7 @@ import { } from '@/lib/server/documents/blobstore'; import { enqueueParsePdfJob } from '@/lib/server/jobs/parsePdfJob'; import { + isDocumentParseStateStale, normalizeParseStatus, parseDocumentParseState, stringifyDocumentParseState, @@ -19,6 +20,7 @@ import { import { getOpenReaderTestNamespace, getUnclaimedUserIdForNamespace } from '@/lib/server/testing/test-namespace'; import { isS3Configured } from '@/lib/server/storage/s3'; import type { ParsedPdfDocument } from '@/types/parsed-pdf'; +import { getComputeOpStaleMs } from '@openreader/compute-core'; export const dynamic = 'force-dynamic'; @@ -34,6 +36,29 @@ function hasAnyParsedBlocks(doc: ParsedPdfDocument | null): boolean { return doc.pages.some((page) => Array.isArray(page.blocks) && page.blocks.length > 0); } +async function healStaleInProgressParseState(input: { + documentId: string; + userId: string; + state: ReturnType; +}): Promise> { + const staleMs = getComputeOpStaleMs(); + if (!isDocumentParseStateStale(input.state, staleMs)) return input.state; + + const nextState = parseDocumentParseState(stringifyDocumentParseState({ + status: 'failed', + progress: null, + updatedAt: Date.now(), + error: `Parse state stale for more than ${staleMs}ms; marked failed for retry`, + })); + + await db + .update(documents) + .set({ parseState: stringifyDocumentParseState(nextState) }) + .where(and(eq(documents.id, input.documentId), eq(documents.userId, input.userId))); + + return nextState; +} + export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string }> }) { try { if (!isS3Configured()) return s3NotConfiguredResponse(); @@ -73,7 +98,12 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string return NextResponse.json({ error: 'Not found' }, { status: 404 }); } - const state = parseDocumentParseState(row.parseState); + let state = parseDocumentParseState(row.parseState); + state = await healStaleInProgressParseState({ + documentId: id, + userId: row.userId, + state, + }); const effectiveStatus = normalizeParseStatus(state.status); if (effectiveStatus === 'failed' && retryFailed) { @@ -185,7 +215,12 @@ export async function POST(req: NextRequest, ctx: { params: Promise<{ id: string return NextResponse.json({ error: 'Not found' }, { status: 404 }); } - const state = parseDocumentParseState(row.parseState); + let state = parseDocumentParseState(row.parseState); + state = await healStaleInProgressParseState({ + documentId: id, + userId: row.userId, + state, + }); const effectiveStatus = normalizeParseStatus(state.status); if (effectiveStatus !== 'running') { diff --git a/src/lib/server/documents/parse-state.ts b/src/lib/server/documents/parse-state.ts index 133a6bf..431e20c 100644 --- a/src/lib/server/documents/parse-state.ts +++ b/src/lib/server/documents/parse-state.ts @@ -9,6 +9,22 @@ export interface DocumentParseState { jobId?: string; } +export function isInProgressParseStatus(status: PdfParseStatus): status is 'pending' | 'running' { + return status === 'pending' || status === 'running'; +} + +export function isDocumentParseStateStale( + state: DocumentParseState, + staleMs: number, + nowMs = Date.now(), +): boolean { + if (!isInProgressParseStatus(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 normalizeParseStatus(status: string | null | undefined): PdfParseStatus { if (status === 'pending' || status === 'running' || status === 'ready' || status === 'failed') { return status;