chore(worker): migrate compute worker from Redis/BullMQ to NATS JetStream

Replace Redis and BullMQ with NATS JetStream and KV for job queuing and state management in the compute worker service. Update environment variables, Docker Compose, and documentation to reflect the new message queue backend. Adjust dependencies and code to use NATS-based work queue and key-value storage for durable job processing. This change improves scalability and reliability of distributed compute workloads by leveraging NATS JetStream's work queue and persistence features. Documentation and configuration examples are updated to guide deployments using the new backend.
2026-05-20 07:38:03 -06:00 · 2026-05-20 07:38:03 -06:00 · 4b26531172
commit 4b26531172
parent 502c98f11f
14 changed files with 1184 additions and 473 deletions
--- a/.env.example
+++ b/.env.example
@ -79,7 +79,7 @@ IMPORT_LIBRARY_DIRS=

 # Heavy compute backend mode for ONNX whisper alignment + PDF layout parsing.
 # local  = run compute in-process (default)
-# worker = external durable worker mode (requires Redis-backed compute-worker service)
+# worker = external durable worker mode (requires NATS JetStream-backed compute-worker service)
 COMPUTE_MODE=local
 # Required when COMPUTE_MODE=worker
 # COMPUTE_WORKER_URL=http://localhost:8081
--- a/README.md
+++ b/README.md
@ -19,7 +19,7 @@ OpenReader is an open source, self-host-friendly text-to-speech document reader
 ## ✨ Highlights

 - 🧱 **Layout-aware PDF parsing** with PP-DocLayoutV3 (ONNX) — structured block detection, cross-page stitching, and geometry-based highlighting for precise read-along sync.
- ⏱️ **Word-by-word highlighting** via built-in ONNX Whisper alignment — no external dependencies, runs in-process (`COMPUTE_MODE=local`) or offloaded to a scalable Redis-backed worker (`COMPUTE_MODE=worker`).
+- ⏱️ **Word-by-word highlighting** via built-in ONNX Whisper alignment — no external dependencies, runs in-process (`COMPUTE_MODE=local`) or offloaded to a scalable NATS JetStream-backed worker (`COMPUTE_MODE=worker`).
 - ⚡ **Segment-based read-along** for EPUB, PDF, TXT, MD, and DOCX — sentence-aware TTS with cached audio segments, background preloading, and resumable playback.
 - 🎯 **Multi-provider TTS** — self-hosted OpenAI-compatible servers (Kokoro-FastAPI, KittenTTS-FastAPI, Orpheus-FastAPI) or cloud APIs (OpenAI, Replicate, DeepInfra).
 - 🎧 **Audiobook export** in `m4b`/`mp3` with resumable chapter processing.
@ -32,7 +32,7 @@ OpenReader is an open source, self-host-friendly text-to-speech document reader
 | --- | --- |
 | Run with Docker | [Docker Quick Start](https://docs.openreader.richardr.dev/docker-quick-start) |
 | Deploy on Vercel | [Vercel Deployment](https://docs.openreader.richardr.dev/deploy/vercel-deployment) |
-| Deploy external compute worker | [Compute Worker (Redis + BullMQ)](https://docs.openreader.richardr.dev/deploy/compute-worker) |
+| Deploy external compute worker | [Compute Worker (NATS JetStream)](https://docs.openreader.richardr.dev/deploy/compute-worker) |
 | Develop locally | [Local Development](https://docs.openreader.richardr.dev/deploy/local-development) |
 | Configure auth | [Auth](https://docs.openreader.richardr.dev/configure/auth) |
 | Configure SQL database | [Database and Migrations](https://docs.openreader.richardr.dev/configure/database) |
--- a/compute/worker/.env.example
+++ b/compute/worker/.env.example
@ -7,8 +7,8 @@ COMPUTE_LOG_FORMAT=pretty
 # App <-> worker auth
 COMPUTE_WORKER_TOKEN=local-compute-token

-# Redis/BullMQ
-REDIS_URL=redis://redis:6379
+# NATS/JetStream
+NATS_URL=nats://nats:4222

 # Shared object storage (must be reachable from worker)
 S3_BUCKET=openreader-documents
@ -21,5 +21,4 @@ S3_ENDPOINT=http://host.docker.internal:8333
 S3_FORCE_PATH_STYLE=true

 # Queue + execution tuning
-COMPUTE_QUEUE_MAX_DEPTH=64
 COMPUTE_PREWARM_MODELS=true
--- a/compute/worker/docker-compose.yml
+++ b/compute/worker/docker-compose.yml
@ -1,9 +1,13 @@
 services:
-  redis:
-    image: redis:7-alpine
-    container_name: openreader-compute-redis
+  nats:
+    image: nats:2.14-alpine
+    container_name: openreader-compute-nats
+    command: ["-js", "-sd", "/data"]
    ports:
-      - "6379:6379"
+      - "4222:4222"
+      - "8222:8222"
+    volumes:
+      - nats-data:/data

  compute-worker:
    build:
@ -11,11 +15,11 @@ services:
      dockerfile: compute/worker/Dockerfile
    container_name: openreader-compute-worker
    depends_on:
-      - redis
+      - nats
    env_file:
      - ./.env
    environment:
-      REDIS_URL: ${REDIS_URL:-redis://redis:6379}
+      NATS_URL: ${NATS_URL:-nats://nats:4222}
      COMPUTE_WORKER_HOST: ${COMPUTE_WORKER_HOST:-0.0.0.0}
      COMPUTE_WORKER_PORT: ${COMPUTE_WORKER_PORT:-8081}
      COMPUTE_WORKER_TOKEN: ${COMPUTE_WORKER_TOKEN:-local-compute-token}
@ -39,3 +43,6 @@ services:
          path: ../../pnpm-lock.yaml
        - action: rebuild
          path: ../../pnpm-workspace.yaml
+
+volumes:
+  nats-data:
--- a/compute/worker/package.json
+++ b/compute/worker/package.json
@ -8,16 +8,17 @@
    "start": "tsx src/server.ts"
  },
  "dependencies": {
-    "@aws-sdk/client-s3": "^3.1045.0",
+    "@aws-sdk/client-s3": "^3.1050.0",
+    "@nats-io/jetstream": "^3.4.0",
+    "@nats-io/kv": "^3.4.0",
+    "@nats-io/transport-node": "^3.4.0",
    "@openreader/compute-core": "workspace:*",
-    "bullmq": "^5.61.2",
    "fastify": "^5.6.2",
-    "ioredis": "^5.8.2",
-    "pino": "^9.14.0",
+    "pino": "^10.3.1",
    "pino-pretty": "^13.1.2",
    "zod": "^4.1.12"
  },
  "devDependencies": {
-    "tsx": "^4.20.6"
+    "tsx": "^4.22.3"
  }
 }
--- a/compute/worker/src/server.ts
+++ b/compute/worker/src/server.ts
@ -1,10 +1,25 @@
-import Fastify, { type FastifyReply, type FastifyRequest } from 'fastify';
-import { Queue, Worker, type Job, type JobsOptions } from 'bullmq';
-import IORedis from 'ioredis';
+import Fastify, { type FastifyRequest } from 'fastify';
 import { z } from 'zod';
 import {
-  ALIGN_QUEUE_NAME,
-  PDF_LAYOUT_QUEUE_NAME,
+  connect,
+  nanos,
+  type NatsConnection,
+} from '@nats-io/transport-node';
+import {
+  AckPolicy,
+  DeliverPolicy,
+  ReplayPolicy,
+  RetentionPolicy,
+  StorageType,
+  jetstream,
+  jetstreamManager,
+  type Consumer,
+  type JetStreamClient,
+  type JetStreamManager,
+  type JsMsg,
+} from '@nats-io/jetstream';
+import { Kvm, type KV } from '@nats-io/kv';
+import {
  ensureComputeModels,
  runPdfLayoutFromPdfBuffer,
  runWhisperAlignmentFromAudioBuffer,
@ -17,6 +32,33 @@ import {
 } from '@openreader/compute-core';
 import { GetObjectCommand, S3Client } from '@aws-sdk/client-s3';

+const JOBS_STREAM_NAME = 'compute_jobs';
+const WHISPER_JOBS_SUBJECT = 'jobs.whisper';
+const LAYOUT_JOBS_SUBJECT = 'jobs.layout';
+const WHISPER_CONSUMER_NAME = 'compute_whisper';
+const LAYOUT_CONSUMER_NAME = 'compute_layout';
+const JOB_STATES_BUCKET = 'job_states';
+const JOB_STATES_TTL_MS = 24 * 60 * 60 * 1000;
+const PULL_EXPIRES_MS = 1000;
+const LOOP_ERROR_BACKOFF_MS = 500;
+
+interface QueuedJob<TPayload> {
+  jobId: string;
+  queuedAt: number;
+  payload: TPayload;
+}
+
+interface StoredJobState<Result> extends WorkerJobStatusResponse<Result> {
+  timestamp: number;
+  startedAt?: number;
+  updatedAt: number;
+}
+
+type JsonCodec<T> = {
+  encode(value: T): Uint8Array;
+  decode(data: Uint8Array): T;
+};
+
 function requireEnv(name: string): string {
  const value = process.env[name]?.trim();
  if (!value) throw new Error(`${name} is required`);
@ -126,12 +168,6 @@ async function withTimeout<T>(promise: Promise<T>, timeoutMs: number, label: str
  }
 }

-function parseRetryAfterSeconds(raw: string | undefined): number {
-  const parsed = Number(raw ?? '');
-  if (!Number.isFinite(parsed) || parsed <= 0) return 2;
-  return Math.max(1, Math.floor(parsed));
-}
-
 function isAuthed(request: FastifyRequest, expectedToken: string): boolean {
  const auth = request.headers.authorization;
  if (!auth?.startsWith('Bearer ')) return false;
@ -139,6 +175,34 @@ function isAuthed(request: FastifyRequest, expectedToken: string): boolean {
  return token === expectedToken;
 }

+function safeDurationMs(start: number | undefined, end: number | undefined): number | undefined {
+  if (!Number.isFinite(start) || !Number.isFinite(end)) return undefined;
+  return Math.max(0, Math.floor((end as number) - (start as number)));
+}
+
+function toErrorMessage(error: unknown): string {
+  if (error instanceof Error && error.message) return error.message;
+  return String(error);
+}
+
+function isAlreadyExistsError(error: unknown): boolean {
+  const message = toErrorMessage(error).toLowerCase();
+  return message.includes('already in use') || message.includes('already exists');
+}
+
+function createJsonCodec<T>(): JsonCodec<T> {
+  const encoder = new TextEncoder();
+  const decoder = new TextDecoder();
+  return {
+    encode(value: T): Uint8Array {
+      return encoder.encode(JSON.stringify(value));
+    },
+    decode(data: Uint8Array): T {
+      return JSON.parse(decoder.decode(data)) as T;
+    },
+  };
+}
+
 const alignSchema = z.object({
  text: z.string().trim().min(1),
  lang: z.string().trim().min(1).max(16).optional(),
@ -152,83 +216,194 @@ const layoutSchema = z.object({
  documentObjectKey: z.string().trim().min(1).max(2048),
 });

-function mapJobState<Result>(job: Job): WorkerJobStatusResponse<Result> {
-  const result = job.returnvalue as Result | undefined;
-  const resultTiming = readResultTiming(result);
-  const timing = buildJobTimingSnapshot(job, resultTiming);
-
-  if (job.failedReason) {
-    return {
-      status: 'failed',
-      error: {
-        message: job.failedReason || 'Worker job failed',
-      },
-      ...(timing ? { timing } : {}),
-    };
-  }
-  if (typeof job.returnvalue !== 'undefined' && job.finishedOn) {
-    return {
-      status: 'succeeded',
-      result: job.returnvalue as Result,
-      ...(timing ? { timing } : {}),
-    };
-  }
-
-  if (job.processedOn) {
-    return {
-      status: 'running',
-      ...(timing ? { timing } : {}),
-    };
-  }
-  return {
-    status: 'queued',
-    ...(timing ? { timing } : {}),
-  };
+async function putState<Result>(
+  kv: KV,
+  codec: JsonCodec<StoredJobState<Result>>,
+  jobId: string,
+  state: StoredJobState<Result>,
+): Promise<void> {
+  await kv.put(jobId, codec.encode(state));
 }

-function readResultTiming<Result>(result: Result | undefined): WorkerJobTiming | undefined {
-  if (!result || typeof result !== 'object') return undefined;
-  const maybe = result as { timing?: WorkerJobTiming };
-  if (!maybe.timing || typeof maybe.timing !== 'object') return undefined;
-  return maybe.timing;
+async function getState<Result>(
+  kv: KV,
+  codec: JsonCodec<StoredJobState<Result>>,
+  jobId: string,
+): Promise<StoredJobState<Result> | null> {
+  const entry = await kv.get(jobId);
+  if (!entry || entry.operation !== 'PUT') return null;
+  return codec.decode(entry.value);
 }

-function toSafeDurationMs(start: number | undefined, end: number | undefined): number | undefined {
-  if (!Number.isFinite(start) || !Number.isFinite(end)) return undefined;
-  return Math.max(0, Math.floor((end as number) - (start as number)));
-}
-
-function buildJobTimingSnapshot(job: Job, base?: WorkerJobTiming): WorkerJobTiming | undefined {
-  const now = Date.now();
-  const timestamp = typeof job.timestamp === 'number' ? job.timestamp : undefined;
-  const processedOn = typeof job.processedOn === 'number' ? job.processedOn : undefined;
-
-  const timing: WorkerJobTiming = {
-    ...(base ?? {}),
+async function ensureJetStreamResources(
+  jsm: JetStreamManager,
+  whisperTimeoutMs: number,
+  pdfTimeoutMs: number,
+  attempts: number,
+): Promise<void> {
+  const streamConfig = {
+    name: JOBS_STREAM_NAME,
+    subjects: [WHISPER_JOBS_SUBJECT, LAYOUT_JOBS_SUBJECT],
+    retention: RetentionPolicy.Workqueue,
+    storage: StorageType.File,
  };

-  if (typeof timing.queueWaitMs !== 'number') {
-    timing.queueWaitMs = processedOn
-      ? toSafeDurationMs(timestamp, processedOn)
-      : toSafeDurationMs(timestamp, now);
+  try {
+    await jsm.streams.add(streamConfig);
+  } catch (error) {
+    if (!isAlreadyExistsError(error)) throw error;
+    await jsm.streams.update(JOBS_STREAM_NAME, {
+      subjects: [WHISPER_JOBS_SUBJECT, LAYOUT_JOBS_SUBJECT],
+    });
  }

-  const hasAnyTiming = Object.values(timing).some((value) => typeof value === 'number' && Number.isFinite(value));
-  return hasAnyTiming ? timing : undefined;
+  const ensureConsumer = async (name: string, subject: string, ackWaitMs: number): Promise<void> => {
+    const config = {
+      durable_name: name,
+      ack_policy: AckPolicy.Explicit,
+      deliver_policy: DeliverPolicy.All,
+      replay_policy: ReplayPolicy.Instant,
+      filter_subject: subject,
+      ack_wait: nanos(Math.max(ackWaitMs, 1_000)),
+      max_deliver: attempts,
+    };
+
+    try {
+      await jsm.consumers.add(JOBS_STREAM_NAME, config);
+    } catch (error) {
+      if (!isAlreadyExistsError(error)) throw error;
+      await jsm.consumers.update(JOBS_STREAM_NAME, name, {
+        filter_subject: subject,
+        ack_wait: nanos(Math.max(ackWaitMs, 1_000)),
+        max_deliver: attempts,
+      });
+    }
+  };
+
+  await Promise.all([
+    ensureConsumer(WHISPER_CONSUMER_NAME, WHISPER_JOBS_SUBJECT, whisperTimeoutMs + 15_000),
+    ensureConsumer(LAYOUT_CONSUMER_NAME, LAYOUT_JOBS_SUBJECT, pdfTimeoutMs + 15_000),
+  ]);
 }

-async function getQueueDepth(queue: Queue): Promise<number> {
-  const counts = await queue.getJobCounts('waiting', 'active', 'prioritized', 'delayed');
-  return (counts.waiting ?? 0) + (counts.active ?? 0) + (counts.prioritized ?? 0) + (counts.delayed ?? 0);
+async function createWorkerLoop<TPayload, TResult>(input: {
+  consumer: Consumer;
+  kv: KV;
+  stateCodec: JsonCodec<StoredJobState<TResult>>;
+  jobCodec: JsonCodec<QueuedJob<TPayload>>;
+  run: (payload: TPayload, queueWaitMs: number) => Promise<TResult>;
+  maxAttempts: number;
+  logLabel: string;
+  shouldStop: () => boolean;
+  log: {
+    error: (obj: Record<string, unknown>, msg: string) => void;
+    info: (obj: Record<string, unknown>, msg: string) => void;
+  };
+}): Promise<void> {
+  while (!input.shouldStop()) {
+    let msg: JsMsg | null = null;
+    try {
+      msg = await input.consumer.next({ expires: PULL_EXPIRES_MS });
+    } catch (error) {
+      if (input.shouldStop()) return;
+      input.log.error({ error: toErrorMessage(error), worker: input.logLabel }, 'worker pull failed');
+      await new Promise((resolve) => setTimeout(resolve, LOOP_ERROR_BACKOFF_MS));
+      continue;
+    }
+
+    if (!msg) continue;
+
+    let decoded: QueuedJob<TPayload> | null = null;
+    try {
+      const job = input.jobCodec.decode(msg.data);
+      decoded = job;
+      const startedAt = Date.now();
+      const queueWaitMs = safeDurationMs(job.queuedAt, startedAt);
+
+      await putState(input.kv, input.stateCodec, job.jobId, {
+        status: 'running',
+        timestamp: job.queuedAt,
+        startedAt,
+        updatedAt: startedAt,
+        ...(typeof queueWaitMs === 'number' ? { timing: { queueWaitMs } } : {}),
+      });
+
+      const result = await input.run(job.payload, queueWaitMs ?? 0);
+      const resultTiming = result && typeof result === 'object' && 'timing' in result
+        ? (result as { timing?: WorkerJobTiming }).timing
+        : undefined;
+
+      await putState(input.kv, input.stateCodec, job.jobId, {
+        status: 'succeeded',
+        timestamp: job.queuedAt,
+        startedAt,
+        updatedAt: Date.now(),
+        result,
+        ...(resultTiming ? { timing: resultTiming } : {}),
+      });
+
+      msg.ack();
+      input.log.info({ worker: input.logLabel, jobId: job.jobId, timing: resultTiming }, 'job succeeded');
+    } catch (error) {
+      const message = toErrorMessage(error);
+      const deliveryCount = msg.info.deliveryCount;
+      const hasRetriesLeft = deliveryCount < input.maxAttempts;
+
+      if (decoded?.jobId) {
+        const now = Date.now();
+        const queueWaitMs = safeDurationMs(decoded.queuedAt, now);
+        const state: StoredJobState<TResult> = hasRetriesLeft
+          ? {
+            status: 'running',
+            timestamp: decoded.queuedAt,
+            updatedAt: now,
+            ...(typeof queueWaitMs === 'number' ? { timing: { queueWaitMs } } : {}),
+          }
+          : {
+            status: 'failed',
+            timestamp: decoded.queuedAt,
+            updatedAt: now,
+            error: { message },
+            ...(typeof queueWaitMs === 'number' ? { timing: { queueWaitMs } } : {}),
+          };
+
+        await putState(input.kv, input.stateCodec, decoded.jobId, state).catch((stateError) => {
+          input.log.error({
+            worker: input.logLabel,
+            jobId: decoded?.jobId,
+            error: toErrorMessage(stateError),
+          }, 'failed to persist failed state');
+        });
+      }
+
+      if (hasRetriesLeft) {
+        msg.nak();
+        input.log.error({
+          worker: input.logLabel,
+          jobId: decoded?.jobId,
+          error: message,
+          deliveryCount,
+          maxAttempts: input.maxAttempts,
+        }, 'job failed, nacked for retry');
+      } else {
+        msg.term(message);
+        input.log.error({
+          worker: input.logLabel,
+          jobId: decoded?.jobId,
+          error: message,
+          deliveryCount,
+          maxAttempts: input.maxAttempts,
+        }, 'job failed, max attempts reached');
+      }
+    }
+  }
 }

 async function main(): Promise<void> {
  const port = readIntEnv('COMPUTE_WORKER_PORT', 8081);
  const host = process.env.COMPUTE_WORKER_HOST?.trim() || '0.0.0.0';
  const workerToken = requireEnv('COMPUTE_WORKER_TOKEN');
-  const redisUrl = requireEnv('REDIS_URL');
-  const queueMaxDepth = readIntEnv('COMPUTE_QUEUE_MAX_DEPTH', 64);
-  const retryAfterSec = parseRetryAfterSeconds(process.env.COMPUTE_QUEUE_RETRY_AFTER_SEC);
+  const natsUrl = requireEnv('NATS_URL');

  const whisperConcurrency = readIntEnv('COMPUTE_WHISPER_CONCURRENCY', 1);
  const pdfConcurrency = readIntEnv('COMPUTE_PDF_CONCURRENCY', 2);
@ -237,34 +412,15 @@ async function main(): Promise<void> {
  const attempts = readIntEnv('COMPUTE_JOB_ATTEMPTS', 2);
  const prewarmModels = parseBoolEnv('COMPUTE_PREWARM_MODELS', true);

-  const redis = new IORedis(redisUrl, {
-    maxRetriesPerRequest: null,
-    enableReadyCheck: true,
-  });
+  const nc: NatsConnection = await connect({ servers: natsUrl });
+  const js: JetStreamClient = jetstream(nc);
+  const jsm: JetStreamManager = await jetstreamManager(nc);

-  const queueDefaults: JobsOptions = {
-    attempts,
-    backoff: {
-      type: 'exponential',
-      delay: 500,
-    },
-    removeOnComplete: {
-      age: 60 * 60,
-      count: 1000,
-    },
-    removeOnFail: {
-      age: 24 * 60 * 60,
-      count: 5000,
-    },
-  };
+  await ensureJetStreamResources(jsm, whisperTimeoutMs, pdfTimeoutMs, attempts);

-  const alignQueue = new Queue<WhisperAlignJobRequest, WhisperAlignJobResult>(ALIGN_QUEUE_NAME, {
-    connection: redis,
-    defaultJobOptions: queueDefaults,
-  });
-  const layoutQueue = new Queue<PdfLayoutJobRequest, PdfLayoutJobResult>(PDF_LAYOUT_QUEUE_NAME, {
-    connection: redis,
-    defaultJobOptions: queueDefaults,
+  const kv = await new Kvm(js).create(JOB_STATES_BUCKET, {
+    history: 1,
+    ttl: JOB_STATES_TTL_MS,
  });

  const s3 = buildS3Client();
@ -289,123 +445,16 @@ async function main(): Promise<void> {
    return toArrayBuffer(new Uint8Array(bytes));
  };

-  const alignWorker = new Worker<WhisperAlignJobRequest, WhisperAlignJobResult>(
-    ALIGN_QUEUE_NAME,
-    async (job) => {
-      const processingStartedAt = Date.now();
-      const queueWaitMs = typeof job.timestamp === 'number'
-        ? Math.max(0, processingStartedAt - job.timestamp)
-        : undefined;
-      const parsed = alignSchema.parse(job.data);
-      const s3FetchStartedAt = Date.now();
-      const audioBuffer = await readObjectByKey(parsed.audioObjectKey);
-      const s3FetchMs = Date.now() - s3FetchStartedAt;
-      const computeStartedAt = Date.now();
-      const result = await withTimeout(
-        runWhisperAlignmentFromAudioBuffer({
-          audioBuffer,
-          text: parsed.text,
-          cacheKey: parsed.cacheKey,
-          lang: parsed.lang,
-        }),
-        whisperTimeoutMs,
-        'whisper alignment job',
-      );
-      const computeMs = Date.now() - computeStartedAt;
-      return {
-        ...result,
-        timing: {
-          ...(result.timing ?? {}),
-          queueWaitMs,
-          s3FetchMs,
-          computeMs,
-        },
-      };
-    },
-    {
-      connection: redis,
-      concurrency: whisperConcurrency,
-    },
-  );
-
-  const layoutWorker = new Worker<PdfLayoutJobRequest, PdfLayoutJobResult>(
-    PDF_LAYOUT_QUEUE_NAME,
-    async (job) => {
-      const processingStartedAt = Date.now();
-      const queueWaitMs = typeof job.timestamp === 'number'
-        ? Math.max(0, processingStartedAt - job.timestamp)
-        : undefined;
-      const parsed = layoutSchema.parse(job.data);
-      const s3FetchStartedAt = Date.now();
-      const pdfBytes = await readObjectByKey(parsed.documentObjectKey);
-      const s3FetchMs = Date.now() - s3FetchStartedAt;
-      const computeStartedAt = Date.now();
-      const result = await withTimeout(
-        runPdfLayoutFromPdfBuffer({
-          documentId: parsed.documentId,
-          pdfBytes,
-        }),
-        pdfTimeoutMs,
-        'pdf layout job',
-      );
-      const computeMs = Date.now() - computeStartedAt;
-      return {
-        ...result,
-        timing: {
-          ...(result.timing ?? {}),
-          queueWaitMs,
-          s3FetchMs,
-          computeMs,
-        },
-      };
-    },
-    {
-      connection: redis,
-      concurrency: pdfConcurrency,
-    },
-  );
-
  if (prewarmModels) {
    await ensureComputeModels();
  }

-  const app = Fastify({
-    logger: buildLoggerConfig(),
-  });
+  const app = Fastify({ logger: buildLoggerConfig() });

-  alignWorker.on('completed', (job, result) => {
-    app.log.info({
-      queue: ALIGN_QUEUE_NAME,
-      jobId: job.id,
-      timing: buildJobTimingSnapshot(job, readResultTiming(result)),
-    }, 'whisper align job completed');
-  });
-
-  alignWorker.on('failed', (job, err) => {
-    app.log.error({
-      queue: ALIGN_QUEUE_NAME,
-      jobId: job?.id,
-      error: err.message,
-      timing: job ? buildJobTimingSnapshot(job) : undefined,
-    }, 'whisper align job failed');
-  });
-
-  layoutWorker.on('completed', (job, result) => {
-    app.log.info({
-      queue: PDF_LAYOUT_QUEUE_NAME,
-      jobId: job.id,
-      timing: buildJobTimingSnapshot(job, readResultTiming(result)),
-    }, 'pdf layout job completed');
-  });
-
-  layoutWorker.on('failed', (job, err) => {
-    app.log.error({
-      queue: PDF_LAYOUT_QUEUE_NAME,
-      jobId: job?.id,
-      error: err.message,
-      timing: job ? buildJobTimingSnapshot(job) : undefined,
-    }, 'pdf layout job failed');
-  });
+  const whisperStateCodec = createJsonCodec<StoredJobState<WhisperAlignJobResult>>();
+  const layoutStateCodec = createJsonCodec<StoredJobState<PdfLayoutJobResult>>();
+  const whisperJobCodec = createJsonCodec<QueuedJob<WhisperAlignJobRequest>>();
+  const layoutJobCodec = createJsonCodec<QueuedJob<PdfLayoutJobRequest>>();

  app.addHook('onRequest', async (request, reply) => {
    const path = request.url.split('?')[0] ?? request.url;
@ -420,30 +469,17 @@ async function main(): Promise<void> {

  app.get('/health/ready', async (_request, reply) => {
    try {
-      await redis.ping();
+      await nc.flush();
      return { ok: true };
    } catch (error) {
      reply.code(503);
      return {
        ok: false,
-        error: error instanceof Error ? error.message : String(error),
+        error: toErrorMessage(error),
      };
    }
  });

-  const rejectIfSaturated = async (queue: Queue, reply: FastifyReply): Promise<boolean> => {
-    const depth = await getQueueDepth(queue);
-    if (depth < queueMaxDepth) return false;
-    reply.header('Retry-After', String(retryAfterSec));
-    reply.code(429).send({
-      error: 'Queue is saturated',
-      retryAfterSeconds: retryAfterSec,
-      queueDepth: depth,
-      queueMaxDepth,
-    });
-    return true;
-  };
-
  app.post('/align/whisper/jobs', async (request, reply) => {
    const parsed = alignSchema.safeParse(request.body);
    if (!parsed.success) {
@ -453,11 +489,24 @@ async function main(): Promise<void> {
        issues: parsed.error.issues,
      };
    }
-    if (await rejectIfSaturated(alignQueue, reply)) return;

-    const job = await alignQueue.add('align', parsed.data);
+    const jobId = crypto.randomUUID();
+    const queuedAt = Date.now();
+
+    await putState(kv, whisperStateCodec, jobId, {
+      status: 'queued',
+      timestamp: queuedAt,
+      updatedAt: queuedAt,
+    });
+
+    await js.publish(WHISPER_JOBS_SUBJECT, whisperJobCodec.encode({
+      jobId,
+      queuedAt,
+      payload: parsed.data,
+    }));
+
    reply.code(202);
-    return { jobId: String(job.id) };
+    return { jobId };
  });

  app.get('/align/whisper/jobs/:jobId', async (request, reply) => {
@ -466,12 +515,21 @@ async function main(): Promise<void> {
      reply.code(400);
      return { error: 'Invalid job id' };
    }
-    const job = await alignQueue.getJob(params.data.jobId);
-    if (!job) {
+
+    const state = await getState(kv, whisperStateCodec, params.data.jobId);
+    if (!state) {
      reply.code(404);
      return { error: 'Job not found' };
    }
-    return mapJobState<WhisperAlignJobResult>(job);
+
+    const response: WorkerJobStatusResponse<WhisperAlignJobResult> = {
+      status: state.status,
+      ...(state.result ? { result: state.result } : {}),
+      ...(state.error ? { error: state.error } : {}),
+      ...(state.timing ? { timing: state.timing } : {}),
+    };
+
+    return response;
  });

  app.post('/layout/pdf/jobs', async (request, reply) => {
@ -483,11 +541,24 @@ async function main(): Promise<void> {
        issues: parsed.error.issues,
      };
    }
-    if (await rejectIfSaturated(layoutQueue, reply)) return;

-    const job = await layoutQueue.add('layout', parsed.data);
+    const jobId = crypto.randomUUID();
+    const queuedAt = Date.now();
+
+    await putState(kv, layoutStateCodec, jobId, {
+      status: 'queued',
+      timestamp: queuedAt,
+      updatedAt: queuedAt,
+    });
+
+    await js.publish(LAYOUT_JOBS_SUBJECT, layoutJobCodec.encode({
+      jobId,
+      queuedAt,
+      payload: parsed.data,
+    }));
+
    reply.code(202);
-    return { jobId: String(job.id) };
+    return { jobId };
  });

  app.get('/layout/pdf/jobs/:jobId', async (request, reply) => {
@ -496,28 +567,136 @@ async function main(): Promise<void> {
      reply.code(400);
      return { error: 'Invalid job id' };
    }
-    const job = await layoutQueue.getJob(params.data.jobId);
-    if (!job) {
+
+    const state = await getState(kv, layoutStateCodec, params.data.jobId);
+    if (!state) {
      reply.code(404);
      return { error: 'Job not found' };
    }
-    return mapJobState<PdfLayoutJobResult>(job);
+
+    const response: WorkerJobStatusResponse<PdfLayoutJobResult> = {
+      status: state.status,
+      ...(state.result ? { result: state.result } : {}),
+      ...(state.error ? { error: state.error } : {}),
+      ...(state.timing ? { timing: state.timing } : {}),
+    };
+
+    return response;
  });

+  const whisperConsumer = await js.consumers.get(JOBS_STREAM_NAME, WHISPER_CONSUMER_NAME);
+  const layoutConsumer = await js.consumers.get(JOBS_STREAM_NAME, LAYOUT_CONSUMER_NAME);
+
+  let stopping = false;
+
+  const runWhisper = async (
+    payload: WhisperAlignJobRequest,
+    queueWaitMs: number,
+  ): Promise<WhisperAlignJobResult> => {
+    const parsed = alignSchema.parse(payload);
+
+    const s3FetchStartedAt = Date.now();
+    const audioBuffer = await readObjectByKey(parsed.audioObjectKey);
+    const s3FetchMs = Date.now() - s3FetchStartedAt;
+
+    const computeStartedAt = Date.now();
+    const result = await withTimeout(
+      runWhisperAlignmentFromAudioBuffer({
+        audioBuffer,
+        text: parsed.text,
+        cacheKey: parsed.cacheKey,
+        lang: parsed.lang,
+      }),
+      whisperTimeoutMs,
+      'whisper alignment job',
+    );
+
+    const computeMs = Date.now() - computeStartedAt;
+    return {
+      ...result,
+      timing: {
+        ...(result.timing ?? {}),
+        queueWaitMs,
+        s3FetchMs,
+        computeMs,
+      },
+    };
+  };
+
+  const runLayout = async (
+    payload: PdfLayoutJobRequest,
+    queueWaitMs: number,
+  ): Promise<PdfLayoutJobResult> => {
+    const parsed = layoutSchema.parse(payload);
+
+    const s3FetchStartedAt = Date.now();
+    const pdfBytes = await readObjectByKey(parsed.documentObjectKey);
+    const s3FetchMs = Date.now() - s3FetchStartedAt;
+
+    const computeStartedAt = Date.now();
+    const result = await withTimeout(
+      runPdfLayoutFromPdfBuffer({
+        documentId: parsed.documentId,
+        pdfBytes,
+      }),
+      pdfTimeoutMs,
+      'pdf layout job',
+    );
+
+    const computeMs = Date.now() - computeStartedAt;
+    return {
+      ...result,
+      timing: {
+        ...(result.timing ?? {}),
+        queueWaitMs,
+        s3FetchMs,
+        computeMs,
+      },
+    };
+  };
+
+  const workerLoops: Promise<void>[] = [];
+
+  for (let i = 0; i < whisperConcurrency; i += 1) {
+    workerLoops.push(createWorkerLoop({
+      consumer: whisperConsumer,
+      kv,
+      stateCodec: whisperStateCodec,
+      jobCodec: whisperJobCodec,
+      run: runWhisper,
+      maxAttempts: attempts,
+      logLabel: `whisper-${i + 1}`,
+      shouldStop: () => stopping,
+      log: app.log,
+    }));
+  }
+
+  for (let i = 0; i < pdfConcurrency; i += 1) {
+    workerLoops.push(createWorkerLoop({
+      consumer: layoutConsumer,
+      kv,
+      stateCodec: layoutStateCodec,
+      jobCodec: layoutJobCodec,
+      run: runLayout,
+      maxAttempts: attempts,
+      logLabel: `layout-${i + 1}`,
+      shouldStop: () => stopping,
+      log: app.log,
+    }));
+  }
+
  const close = async (): Promise<void> => {
+    if (stopping) return;
+    stopping = true;
    await app.close();
-    await Promise.allSettled([
-      alignWorker.close(),
-      layoutWorker.close(),
-      alignQueue.close(),
-      layoutQueue.close(),
-      redis.quit(),
-    ]);
+    await Promise.allSettled(workerLoops);
+    await nc.drain();
  };

  process.once('SIGINT', () => {
    void close().finally(() => process.exit(0));
  });
+
  process.once('SIGTERM', () => {
    void close().finally(() => process.exit(0));
  });
--- a/docs-site/docs/deploy/compute-worker.md
+++ b/docs-site/docs/deploy/compute-worker.md
@ -1,5 +1,4 @@
---
-title: Compute Worker (Redis + BullMQ)
+title: Compute Worker (NATS JetStream)
 ---

 Use this guide for `COMPUTE_MODE=worker` deployments where heavy compute runs outside the Next.js app server.
@ -11,7 +10,7 @@ The compute worker handles:
 - Whisper word alignment (`/align/whisper/jobs`)
 - PDF layout parsing (`/layout/pdf/jobs`)

-The app server enqueues jobs and polls status. Queue durability and retries are backed by Redis + BullMQ.
+The app server enqueues jobs and polls status. Queue durability and retries are backed by NATS JetStream WorkQueue consumers and NATS KV.

 ## Published image

@ -23,7 +22,7 @@ The app server enqueues jobs and polls status. Queue durability and retries are
 Required:

 - `COMPUTE_WORKER_TOKEN`: bearer token expected by worker routes
- `REDIS_URL`: BullMQ Redis connection string
+- `NATS_URL`: NATS server connection string (JetStream enabled)
 - `S3_BUCKET`
 - `S3_REGION`
 - `S3_ACCESS_KEY_ID`
@ -37,7 +36,6 @@ Common optional:
 - `COMPUTE_WORKER_HOST=0.0.0.0`
 - `COMPUTE_WORKER_PORT=8081`
 - `COMPUTE_LOG_FORMAT=pretty` (default) or `json`
- `COMPUTE_QUEUE_MAX_DEPTH=64`
 - `COMPUTE_PREWARM_MODELS=true`

 ## App server environment variables (worker mode)
--- a/docs-site/docs/deploy/local-development.md
+++ b/docs-site/docs/deploy/local-development.md
@ -127,10 +127,10 @@ If you need mirrors or pinned artifact locations, set `WHISPER_MODEL_BASE_URL` i
 <details>
 <summary><strong>External compute worker dev stack (optional)</strong></summary>

-Use this when you want durable compute with Redis/BullMQ while keeping Next.js on native host `pnpm dev`.
-Full worker deployment details are in [Compute Worker (Redis + BullMQ)](./compute-worker).
+Use this when you want durable compute with NATS JetStream + KV while keeping Next.js on native host `pnpm dev`.
+Full worker deployment details are in [Compute Worker (NATS JetStream)](./compute-worker).

-Start only Redis + compute-worker via compose watch:
+Start only NATS + compute-worker via compose watch:

 ```bash
 docker compose --env-file compute/worker/.env -f compute/worker/docker-compose.yml up --watch
@ -243,7 +243,7 @@ S3_SECRET_ACCESS_KEY=your-secret-key
 ```

  </TabItem>
-  <TabItem value="worker-mode" label="External Worker + Redis">
+  <TabItem value="worker-mode" label="External Worker + NATS">

 ```env
 API_BASE=http://host.docker.internal:8880/v1
--- a/docs-site/docs/deploy/vercel-deployment.md
+++ b/docs-site/docs/deploy/vercel-deployment.md
@ -9,7 +9,7 @@ This guide covers deploying OpenReader to Vercel with external Postgres and S3-c
 - Documents (PDF/EPUB/TXT/MD) work with `POSTGRES_URL` + external S3 storage.
 - Audiobook routes work on Node.js serverless functions using `ffmpeg-static`.
 - Heavy compute features (Whisper alignment + PDF layout parsing) work through `COMPUTE_MODE=worker` with an external compute worker service.
- For worker setup details and worker-specific env vars, see [Compute Worker (Redis + BullMQ)](./compute-worker).
+- For worker setup details and worker-specific env vars, see [Compute Worker (NATS JetStream)](./compute-worker).

 :::warning DOCX Conversion Limitation
 `docx` conversion requires `soffice` (LibreOffice), which is not available in a standard Vercel runtime.
--- a/docs-site/docs/docker-quick-start.md
+++ b/docs-site/docs/docker-quick-start.md
@ -141,7 +141,7 @@ Visit [http://localhost:3003](http://localhost:3003) after startup.
 ## 3. Update Docker image

 Legacy image compatibility: `ghcr.io/richardr1126/openreader-webui:latest` remains available as an alias.
-For external compute mode image details, see [Compute Worker (Redis + BullMQ)](./deploy/compute-worker).
+For external compute mode image details, see [Compute Worker (NATS JetStream)](./deploy/compute-worker).

 ```bash
 docker stop openreader || true && \
--- a/docs-site/docs/introduction.md
+++ b/docs-site/docs/introduction.md
@ -15,7 +15,7 @@ It supports multiple TTS providers including OpenAI, Replicate, DeepInfra, and c
 - 🧱 **Layout-aware PDF Parsing**
  - PP-DocLayoutV3 (ONNX) detects structured blocks with cross-page stitching and geometry-based highlighting for precise read-along sync and clean TTS segmentation
 - ⏱️ **Word-by-word Highlighting** via ONNX Whisper alignment
-  - No external dependencies — runs in-process (`COMPUTE_MODE=local`) or offloaded to a scalable Redis-backed worker (`COMPUTE_MODE=worker`)
+  - No external dependencies — runs in-process (`COMPUTE_MODE=local`) or offloaded to a scalable NATS JetStream-backed worker (`COMPUTE_MODE=worker`)
 - ⚡ **Segment-based TTS Playback**
  - Sentence-aware generation with cached audio segments, background preloading, and resumable playback across EPUB, PDF, TXT, MD, and DOCX
 - 🎯 **Multi-Provider TTS Support**
--- a/docs-site/docs/reference/environment-variables.md
+++ b/docs-site/docs/reference/environment-variables.md
@ -357,11 +357,11 @@ Selects the backend for heavy compute features (ONNX word alignment + PDF layout
 - Default: `local`
 - Supported in v1:
  - `local`: run compute in-process on the app server
-  - `worker`: enqueue async jobs in an external durable compute worker (Redis + BullMQ)
+  - `worker`: enqueue async jobs in an external durable compute worker (NATS JetStream + NATS KV)
 - `worker` requires `COMPUTE_WORKER_URL` and `COMPUTE_WORKER_TOKEN`
 - `worker` assumes the external worker can directly reach shared object storage (S3-compatible endpoint)
 - `worker` is not compatible with non-exposed embedded `weed mini` storage topologies
- Worker service env vars are documented in [Compute Worker (Redis + BullMQ)](../deploy/compute-worker)
+- Worker service env vars are documented in [Compute Worker (NATS JetStream)](../deploy/compute-worker)

 ### COMPUTE_WORKER_URL

--- a/docs-site/docs/reference/stack.md
+++ b/docs-site/docs/reference/stack.md
@ -55,11 +55,11 @@ Monorepo packages under `compute/`:
  - Utilities: `jszip`, `ffmpeg-static`
 - **`@openreader/compute-worker`** — standalone Node.js worker service
  - HTTP server: [Fastify](https://fastify.dev/) v5
-  - Job queue: [BullMQ](https://bullmq.io/) + [ioredis](https://github.com/redis/ioredis) (queues: `whisper-align`, `pdf-layout`)
+  - Job queue + state: [NATS](https://nats.io/) JetStream WorkQueue pull consumers + NATS KV (`jobs.whisper`, `jobs.layout`)
  - Storage: AWS SDK v3 S3 client for reading/writing blobs
  - Logging: [Pino](https://getpino.io/)
  - Validation: [Zod](https://zod.dev/)
- Compute mode is controlled by `COMPUTE_MODE` env var: `local` (in-process) or `worker` (remote queue via HTTP + Redis)
+- Compute mode is controlled by `COMPUTE_MODE` env var: `local` (in-process) or `worker` (remote queue via HTTP + NATS)

 ## Tooling and testing

--- a/pnpm-lock.yaml
+++ b/pnpm-lock.yaml