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) {