From a2c714c2a55ba6fd364f4ce854c1d577ffc6f5fb Mon Sep 17 00:00:00 2001 From: Richard R Date: Thu, 21 May 2026 12:37:21 -0600 Subject: [PATCH] refactor(worker,tts): unify PDF and Whisper retry config, add alignment timeouts Update compute worker to use separate environment variable for PDF job attempts (`COMPUTE_PDF_JOB_ATTEMPTS`) and set Whisper job max deliveries to 1, clarifying and separating retry logic between job types. Adjust `.env.example` and deployment docs to comment out defaults and document new/renamed variables for advanced tuning. In TTS segment ensure route, introduce explicit timeouts for Whisper alignment operations, with configurable durations for local and worker modes, improving robustness and error handling for long-running alignments. --- compute/worker/.env.example | 24 ++++++------- compute/worker/src/server.ts | 31 ++++++++++------ docs-site/docs/deploy/compute-worker.md | 20 +++++++++-- src/app/api/tts/segments/ensure/route.ts | 46 +++++++++++++++++++----- 4 files changed, 87 insertions(+), 34 deletions(-) diff --git a/compute/worker/.env.example b/compute/worker/.env.example index eef2843..c8b889a 100644 --- a/compute/worker/.env.example +++ b/compute/worker/.env.example @@ -2,9 +2,9 @@ # Platform note: # - Local/manual: keep PORT=8081 # - Railway/Render/Fly/etc: platform injects PORT -COMPUTE_WORKER_HOST=0.0.0.0 -PORT=8081 -COMPUTE_LOG_FORMAT=pretty +# COMPUTE_WORKER_HOST=0.0.0.0 +# PORT=8081 +# COMPUTE_LOG_FORMAT=pretty # COMPUTE_LOG_LEVEL=info # App <-> worker auth @@ -23,20 +23,20 @@ S3_BUCKET=openreader-documents S3_REGION=us-east-1 S3_ACCESS_KEY_ID=devkey S3_SECRET_ACCESS_KEY=devsecret -S3_PREFIX=openreader +# S3_PREFIX=openreader # Optional for non-AWS S3-compatible endpoints: S3_ENDPOINT=http://host.docker.internal:8333 S3_FORCE_PATH_STYLE=true # Queue + execution tuning -COMPUTE_PREWARM_MODELS=true -COMPUTE_WHISPER_CONCURRENCY=1 -COMPUTE_PDF_CONCURRENCY=2 -COMPUTE_WHISPER_TIMEOUT_MS=30000 -COMPUTE_PDF_TIMEOUT_MS=90000 -COMPUTE_JOB_ATTEMPTS=2 -COMPUTE_JOBS_STREAM_MAX_BYTES=268435456 -COMPUTE_JOB_STATES_MAX_BYTES=67108864 +# COMPUTE_PREWARM_MODELS=true +# COMPUTE_WHISPER_CONCURRENCY=1 +# COMPUTE_PDF_CONCURRENCY=2 +# COMPUTE_WHISPER_TIMEOUT_MS=30000 +# COMPUTE_PDF_TIMEOUT_MS=90000 +# COMPUTE_PDF_JOB_ATTEMPTS=2 +# COMPUTE_JOBS_STREAM_MAX_BYTES=268435456 +# COMPUTE_JOB_STATES_MAX_BYTES=67108864 # Optional stale window for reusing in-flight opKey entries before forcing a new attempt # Default is max(30m, 4x max compute timeout); running jobs also refresh heartbeat every 5s # COMPUTE_OP_STALE_MS=1800000 diff --git a/compute/worker/src/server.ts b/compute/worker/src/server.ts index 36c818b..10e6298 100644 --- a/compute/worker/src/server.ts +++ b/compute/worker/src/server.ts @@ -54,6 +54,7 @@ const SSE_POLL_INTERVAL_MS = 400; const RUNNING_HEARTBEAT_MS = 5000; const DOCUMENT_ID_REGEX = /^[a-f0-9]{64}$/i; const SAFE_NAMESPACE_REGEX = /^[a-zA-Z0-9._-]{1,128}$/; +const WHISPER_MAX_DELIVER = 1; interface QueuedJob { jobId: string; @@ -311,7 +312,7 @@ async function ensureJetStreamResources( jsm: JetStreamManager, whisperTimeoutMs: number, pdfTimeoutMs: number, - attempts: number, + pdfAttempts: number, maxBytes: number, ): Promise { const streamConfig = { @@ -332,7 +333,12 @@ async function ensureJetStreamResources( }); } - const ensureConsumer = async (name: string, subject: string, ackWaitMs: number): Promise => { + const ensureConsumer = async ( + name: string, + subject: string, + ackWaitMs: number, + maxDeliver: number, + ): Promise => { const config = { durable_name: name, ack_policy: AckPolicy.Explicit, @@ -340,7 +346,7 @@ async function ensureJetStreamResources( replay_policy: ReplayPolicy.Instant, filter_subject: subject, ack_wait: nanos(Math.max(ackWaitMs, 1_000)), - max_deliver: attempts, + max_deliver: maxDeliver, }; try { @@ -350,14 +356,14 @@ async function ensureJetStreamResources( await jsm.consumers.update(JOBS_STREAM_NAME, name, { filter_subject: subject, ack_wait: nanos(Math.max(ackWaitMs, 1_000)), - max_deliver: attempts, + max_deliver: maxDeliver, }); } }; await Promise.all([ - ensureConsumer(WHISPER_CONSUMER_NAME, WHISPER_JOBS_SUBJECT, whisperTimeoutMs + 15_000), - ensureConsumer(LAYOUT_CONSUMER_NAME, LAYOUT_JOBS_SUBJECT, pdfTimeoutMs + 15_000), + ensureConsumer(WHISPER_CONSUMER_NAME, WHISPER_JOBS_SUBJECT, whisperTimeoutMs + 15_000, WHISPER_MAX_DELIVER), + ensureConsumer(LAYOUT_CONSUMER_NAME, LAYOUT_JOBS_SUBJECT, pdfTimeoutMs + 15_000, pdfAttempts), ]); } @@ -371,7 +377,7 @@ async function main(): Promise { const pdfConcurrency = readIntEnv('COMPUTE_PDF_CONCURRENCY', 2); const whisperTimeoutMs = readIntEnv('COMPUTE_WHISPER_TIMEOUT_MS', 30_000); const pdfTimeoutMs = readIntEnv('COMPUTE_PDF_TIMEOUT_MS', 90_000); - const attempts = readIntEnv('COMPUTE_JOB_ATTEMPTS', 2); + const pdfAttempts = readIntEnv('COMPUTE_PDF_JOB_ATTEMPTS', 2); 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); @@ -398,7 +404,7 @@ async function main(): Promise { const js: JetStreamClient = jetstream(nc); const jsm: JetStreamManager = await jetstreamManager(nc); - await ensureJetStreamResources(jsm, whisperTimeoutMs, pdfTimeoutMs, attempts, jobsStreamMaxBytes); + await ensureJetStreamResources(jsm, whisperTimeoutMs, pdfTimeoutMs, pdfAttempts, jobsStreamMaxBytes); const kv = await new Kvm(js).create(COMPUTE_STATE_BUCKET, { history: 1, @@ -962,7 +968,9 @@ async function main(): Promise { } catch (error) { const message = toErrorMessage(error); const deliveryCount = input.msg.info.deliveryCount; - const hasRetriesLeft = deliveryCount < attempts; + const isWhisperAlign = decoded?.kind === 'whisper_align'; + const maxAttempts = isWhisperAlign ? WHISPER_MAX_DELIVER : pdfAttempts; + const hasRetriesLeft = !isWhisperAlign && deliveryCount < maxAttempts; if (decoded) { const now = Date.now(); @@ -1020,7 +1028,7 @@ async function main(): Promise { jobId: decoded?.jobId, error: message, deliveryCount, - maxAttempts: attempts, + maxAttempts, }, 'job failed, nacked for retry'); } else { input.msg.term(message); @@ -1030,7 +1038,8 @@ async function main(): Promise { jobId: decoded?.jobId, error: message, deliveryCount, - maxAttempts: attempts, + maxAttempts, + retrySuppressed: isWhisperAlign ? 'whisper_align' : undefined, }, 'job failed, max attempts reached'); } } finally { diff --git a/docs-site/docs/deploy/compute-worker.md b/docs-site/docs/deploy/compute-worker.md index 454f704..fef4995 100644 --- a/docs-site/docs/deploy/compute-worker.md +++ b/docs-site/docs/deploy/compute-worker.md @@ -44,9 +44,18 @@ Common optional: - `COMPUTE_WORKER_HOST=0.0.0.0` - `PORT=8081` (local/manual; on Railway platform injects this) - `COMPUTE_LOG_FORMAT=pretty` (default) or `json` + +Advanced tuning (usually leave unset unless you need overrides): + - `COMPUTE_PREWARM_MODELS=true` +- `COMPUTE_WHISPER_CONCURRENCY=1` +- `COMPUTE_PDF_CONCURRENCY=2` +- `COMPUTE_WHISPER_TIMEOUT_MS=30000` +- `COMPUTE_PDF_TIMEOUT_MS=90000` +- `COMPUTE_PDF_JOB_ATTEMPTS=2` (PDF layout retry attempts) - `COMPUTE_JOBS_STREAM_MAX_BYTES=268435456` (256MB JetStream jobs stream cap) - `COMPUTE_JOB_STATES_MAX_BYTES=67108864` (64MB JetStream KV bucket cap) +- `COMPUTE_OP_STALE_MS=1800000` (stale op replacement window) ## App server environment variables (worker mode) @@ -110,8 +119,15 @@ COMPUTE_WORKER_HOST=0.0.0.0 # PORT=8081 # Railway: rely on injected PORT COMPUTE_WORKER_TOKEN= -COMPUTE_JOBS_STREAM_MAX_BYTES=268435456 -COMPUTE_JOB_STATES_MAX_BYTES=67108864 +# Optional advanced tuning overrides (defaults shown): +# COMPUTE_PREWARM_MODELS=true +# COMPUTE_WHISPER_CONCURRENCY=1 +# COMPUTE_PDF_CONCURRENCY=2 +# COMPUTE_WHISPER_TIMEOUT_MS=30000 +# COMPUTE_PDF_TIMEOUT_MS=90000 +# COMPUTE_PDF_JOB_ATTEMPTS=2 +# COMPUTE_JOBS_STREAM_MAX_BYTES=268435456 +# COMPUTE_JOB_STATES_MAX_BYTES=67108864 NATS_URL=tls://connect.ngs.global:4222 NATS_CREDS="-----BEGIN NATS USER JWT----- diff --git a/src/app/api/tts/segments/ensure/route.ts b/src/app/api/tts/segments/ensure/route.ts index e084a05..610984e 100644 --- a/src/app/api/tts/segments/ensure/route.ts +++ b/src/app/api/tts/segments/ensure/route.ts @@ -44,7 +44,9 @@ import type { export const runtime = 'nodejs'; export const dynamic = 'force-dynamic'; -const GENERATING_STALE_MS = 45_000; +const GENERATING_STALE_MS = 360_000; +const LOCAL_ALIGNMENT_TIMEOUT_MS = 30_000; +const WORKER_ALIGNMENT_TIMEOUT_MS = 60_000; function attachDeviceIdCookie(response: NextResponse, deviceId: string | null, didCreate: boolean) { if (didCreate && deviceId) { @@ -129,6 +131,24 @@ function isAbortLikeError(error: unknown): boolean { return /abort/i.test(message); } +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); + } +} + +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 }) @@ -386,10 +406,14 @@ export async function POST(request: NextRequest) { try { const alignStartedAt = Date.now(); const computeBackend = await getComputeBackend(); - const aligned = (await computeBackend.alignWords({ - audioObjectKey: existing.audioKey, - text: segment.text, - })).alignments; + const aligned = (await withTimeout( + computeBackend.alignWords({ + audioObjectKey: existing.audioKey, + text: segment.text, + }), + getAlignmentTimeoutMs(computeBackend.mode), + `Whisper alignment (${computeBackend.mode})`, + )).alignments; stageTimings.selfHealAlignMs = Date.now() - alignStartedAt; alignment = aligned[0] ? { ...aligned[0], sentenceIndex: segment.original.segmentIndex } : null; @@ -636,10 +660,14 @@ export async function POST(request: NextRequest) { failedStage = 'whisper.align'; const alignStartedAt = Date.now(); const computeBackend = await getComputeBackend(); - const aligned = (await computeBackend.alignWords({ - audioObjectKey: audioKey, - text: segment.text, - })).alignments; + const aligned = (await withTimeout( + computeBackend.alignWords({ + audioObjectKey: audioKey, + text: segment.text, + }), + getAlignmentTimeoutMs(computeBackend.mode), + `Whisper alignment (${computeBackend.mode})`, + )).alignments; stageTimings.whisperAlignMs = Date.now() - alignStartedAt; alignment = aligned[0] ? { ...aligned[0], sentenceIndex: segment.original.segmentIndex } : null; } catch (alignError) {