diff --git a/src/app/api/tts/segments/ensure/route.ts b/src/app/api/tts/segments/ensure/route.ts index 610984e..cb0dc5d 100644 --- a/src/app/api/tts/segments/ensure/route.ts +++ b/src/app/api/tts/segments/ensure/route.ts @@ -35,6 +35,7 @@ import { getCompute } from '@/lib/server/compute'; import { getResolvedRuntimeConfig } from '@/lib/server/runtime-config'; import { resolveTtsModelForProvider } from '@/lib/shared/tts-provider-policy'; import { resolveSegmentAudioUrls } from '@/lib/server/tts/segment-audio-urls'; +import { getComputeTimeoutConfig } from '@openreader/compute-core/runtime/timeout-config'; import type { TTSSegmentInput, TTSSegmentManifestItem, @@ -45,8 +46,7 @@ import type { export const runtime = 'nodejs'; export const dynamic = 'force-dynamic'; const GENERATING_STALE_MS = 360_000; -const LOCAL_ALIGNMENT_TIMEOUT_MS = 30_000; -const WORKER_ALIGNMENT_TIMEOUT_MS = 60_000; +const SHARED_ALIGNMENT_TIMEOUT_MS = getComputeTimeoutConfig().whisperTimeoutMs; function attachDeviceIdCookie(response: NextResponse, deviceId: string | null, didCreate: boolean) { if (didCreate && deviceId) { @@ -145,10 +145,6 @@ async function withTimeout(promise: Promise, timeoutMs: number, label: str } } -function getAlignmentTimeoutMs(mode: 'local' | 'worker'): number { - return mode === 'worker' ? WORKER_ALIGNMENT_TIMEOUT_MS : LOCAL_ALIGNMENT_TIMEOUT_MS; -} - async function deleteEntryIfUnused(userId: string, segmentEntryId: string): Promise { const stillReferenced = await db .select({ segmentId: ttsSegmentVariants.segmentId }) @@ -411,7 +407,7 @@ export async function POST(request: NextRequest) { audioObjectKey: existing.audioKey, text: segment.text, }), - getAlignmentTimeoutMs(computeBackend.mode), + SHARED_ALIGNMENT_TIMEOUT_MS, `Whisper alignment (${computeBackend.mode})`, )).alignments; stageTimings.selfHealAlignMs = Date.now() - alignStartedAt; @@ -665,7 +661,7 @@ export async function POST(request: NextRequest) { audioObjectKey: audioKey, text: segment.text, }), - getAlignmentTimeoutMs(computeBackend.mode), + SHARED_ALIGNMENT_TIMEOUT_MS, `Whisper alignment (${computeBackend.mode})`, )).alignments; stageTimings.whisperAlignMs = Date.now() - alignStartedAt; diff --git a/src/lib/server/compute/local.ts b/src/lib/server/compute/local.ts index c5f88e1..392c9f7 100644 --- a/src/lib/server/compute/local.ts +++ b/src/lib/server/compute/local.ts @@ -2,13 +2,83 @@ import type { ComputeBackend, PdfLayoutInput, WhisperAlignInput, WhisperAlignRes import { LOCAL_COMPUTE_LIMITER } from '@/lib/server/compute/concurrency-limiter'; import { getDocumentBlob } from '@/lib/server/documents/blobstore'; import { getTtsSegmentAudioObject } from '@/lib/server/tts/segments-blobstore'; +import { getComputeTimeoutConfig } from '@openreader/compute-core/runtime/timeout-config'; import { runPdfLayoutFromPdfBuffer, runWhisperAlignmentFromAudioBuffer, } from '@openreader/compute-core/local-runtime'; +import type { PdfLayoutProgress } from '@openreader/compute-core/contracts'; + +async function withTimeout(promise: Promise, timeoutMs: number, label: string): Promise { + let timer: NodeJS.Timeout | null = null; + try { + return await Promise.race([ + promise, + new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error(`${label} timed out after ${timeoutMs}ms`)), timeoutMs); + }), + ]); + } finally { + if (timer) clearTimeout(timer); + } +} + +async function withIdleTimeoutAndHardCap(input: { + run: (touchProgress: () => void) => Promise; + idleTimeoutMs: number; + hardCapMs: number; + label: string; +}): Promise { + let idleTimer: NodeJS.Timeout | null = null; + let hardCapTimer: NodeJS.Timeout | null = null; + let settled = false; + let rejectTimeout!: (reason: unknown) => void; + + const clearTimers = () => { + if (idleTimer) { + clearTimeout(idleTimer); + idleTimer = null; + } + if (hardCapTimer) { + clearTimeout(hardCapTimer); + hardCapTimer = null; + } + }; + + const failTimeout = (kind: 'idle' | 'hard cap', timeoutMs: number) => { + if (settled) return; + settled = true; + clearTimers(); + rejectTimeout(new Error(`${input.label} ${kind} timed out after ${timeoutMs}ms`)); + }; + + const touchProgress = () => { + if (settled) return; + if (idleTimer) clearTimeout(idleTimer); + idleTimer = setTimeout(() => failTimeout('idle', input.idleTimeoutMs), input.idleTimeoutMs); + }; + + const timeoutPromise = new Promise((_, reject) => { + rejectTimeout = reject; + hardCapTimer = setTimeout(() => failTimeout('hard cap', input.hardCapMs), input.hardCapMs); + touchProgress(); + }); + + try { + const result = await Promise.race([input.run(touchProgress), timeoutPromise]); + settled = true; + clearTimers(); + return result as T; + } catch (error) { + settled = true; + clearTimers(); + throw error; + } +} export class LocalComputeBackend implements ComputeBackend { readonly mode = 'local' as const; + private readonly timeoutConfig = getComputeTimeoutConfig(); async alignWords(input: WhisperAlignInput): Promise { return LOCAL_COMPUTE_LIMITER.run(async () => { @@ -20,12 +90,16 @@ export class LocalComputeBackend implements ComputeBackend { if (!audioBuffer) { throw new Error('Local compute alignment requires audioBuffer or audioObjectKey'); } - return runWhisperAlignmentFromAudioBuffer({ - audioBuffer, - text: input.text, - cacheKey: input.cacheKey, - lang: input.lang, - }); + return withTimeout( + runWhisperAlignmentFromAudioBuffer({ + audioBuffer, + text: input.text, + cacheKey: input.cacheKey, + lang: input.lang, + }), + this.timeoutConfig.whisperTimeoutMs, + 'local whisper alignment job', + ); }); } @@ -39,15 +113,25 @@ export class LocalComputeBackend implements ComputeBackend { if (!pdfBytes) { throw new Error('Local compute PDF layout requires pdfBytes or (documentId + namespace)'); } + return { - parsed: (await runPdfLayoutFromPdfBuffer({ - documentId: input.documentId, - pdfBytes, - onPageParsed: (page) => input.onProgress?.({ - totalPages: page.totalPages, - pagesParsed: page.pageNumber, - currentPage: page.pageNumber, - phase: 'infer', + parsed: (await withIdleTimeoutAndHardCap({ + idleTimeoutMs: Math.max(this.timeoutConfig.pdfTimeoutMs, 1_000), + hardCapMs: this.timeoutConfig.pdfHardCapMs, + label: 'local pdf layout job', + run: async (touchProgress) => runPdfLayoutFromPdfBuffer({ + documentId: input.documentId, + pdfBytes, + onPageParsed: (page) => { + touchProgress(); + const progress: PdfLayoutProgress = { + totalPages: page.totalPages, + pagesParsed: page.pageNumber, + currentPage: page.pageNumber, + phase: 'infer', + }; + return input.onProgress?.(progress); + }, }), })).parsed, }; diff --git a/src/lib/server/compute/worker.ts b/src/lib/server/compute/worker.ts index 4ec9c79..19356d3 100644 --- a/src/lib/server/compute/worker.ts +++ b/src/lib/server/compute/worker.ts @@ -1,5 +1,6 @@ import { createHash, randomUUID } from 'node:crypto'; import type { ComputeBackend, PdfLayoutInput, WhisperAlignInput, WhisperAlignResult } from '@/lib/server/compute/types'; +import { getWorkerClientWaitTimeoutMs } from '@openreader/compute-core/runtime/timeout-config'; import type { PdfLayoutJobRequest, PdfLayoutJobResult, @@ -20,7 +21,6 @@ class WorkerHttpError extends Error { } } -const DEFAULT_WAIT_TIMEOUT_MS = 120_000; const DEFAULT_RETRIES = 2; const LOG_PREFIX = '[compute-worker-client]'; const MAX_LOG_DETAIL_CHARS = 600; @@ -236,13 +236,16 @@ export class WorkerComputeBackend implements ComputeBackend { readonly mode = 'worker' as const; private readonly baseUrl: string; private readonly token: string; - private readonly waitTimeoutMs: number; + private readonly waitTimeoutMsByKind: Record<'whisper_align' | 'pdf_layout', number>; private readonly retries: number; constructor() { this.baseUrl = normalizeWorkerBaseUrl(readRequiredEnv('COMPUTE_WORKER_URL')); this.token = readRequiredEnv('COMPUTE_WORKER_TOKEN'); - this.waitTimeoutMs = DEFAULT_WAIT_TIMEOUT_MS; + this.waitTimeoutMsByKind = { + whisper_align: getWorkerClientWaitTimeoutMs('whisper_align'), + pdf_layout: getWorkerClientWaitTimeoutMs('pdf_layout'), + }; this.retries = DEFAULT_RETRIES; } @@ -268,7 +271,7 @@ export class WorkerComputeBackend implements ComputeBackend { cacheKey: input.cacheKey ?? null, lang: input.lang ?? null, textLength: input.text.length, - waitTimeoutMs: this.waitTimeoutMs, + waitTimeoutMs: this.waitTimeoutMsByKind.whisper_align, maxRetries: this.retries, }); @@ -292,6 +295,7 @@ export class WorkerComputeBackend implements ComputeBackend { kind: 'whisper_align', opKeyHash, attempt, + waitTimeoutMs: this.waitTimeoutMsByKind.whisper_align, }); if (final.status !== 'succeeded' || !final.result) { @@ -350,7 +354,7 @@ export class WorkerComputeBackend implements ComputeBackend { documentId: input.documentId, namespace: input.namespace ?? null, documentObjectKey: input.documentObjectKey, - waitTimeoutMs: this.waitTimeoutMs, + waitTimeoutMs: this.waitTimeoutMsByKind.pdf_layout, maxRetries: this.retries, }); @@ -376,6 +380,7 @@ export class WorkerComputeBackend implements ComputeBackend { opKeyHash, documentId: input.documentId, attempt, + waitTimeoutMs: this.waitTimeoutMsByKind.pdf_layout, onSnapshot: (snapshot) => { if (snapshot.progress) { void input.onProgress?.(snapshot.progress); @@ -489,18 +494,22 @@ export class WorkerComputeBackend implements ComputeBackend { private async waitForOperation( opId: string, context: Record & { + waitTimeoutMs?: number; onSnapshot?: (snapshot: WorkerOperationState) => void } = {}, ): Promise> { + const waitTimeoutMs = typeof context.waitTimeoutMs === 'number' + ? context.waitTimeoutMs + : this.waitTimeoutMsByKind.whisper_align; const controller = new AbortController(); - const timeout = setTimeout(() => controller.abort(), this.waitTimeoutMs); + const timeout = setTimeout(() => controller.abort(), waitTimeoutMs); const startedAt = Date.now(); const traceId = typeof context.traceId === 'string' ? context.traceId : randomUUID(); logWorker('info', 'sse.wait.start', { ...context, traceId, opId, - waitTimeoutMs: this.waitTimeoutMs, + waitTimeoutMs, }); try { diff --git a/tsconfig.json b/tsconfig.json index a06fb0d..d2923e6 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -22,7 +22,8 @@ "@/*": ["./src/*"], "@openreader/compute-core": ["./compute/core/src/index.ts"], "@openreader/compute-core/contracts": ["./compute/core/src/contracts.ts"], - "@openreader/compute-core/local-runtime": ["./compute/core/src/local-runtime.ts"] + "@openreader/compute-core/local-runtime": ["./compute/core/src/local-runtime.ts"], + "@openreader/compute-core/runtime/timeout-config": ["./compute/core/src/runtime/timeout-config.ts"] } }, "include": ["next-env.d.ts", "**/*.ts", "**/*.tsx", ".next/types/**/*.ts"],