Background Job Queue

Work that outlives the request runs on a queue

Two kinds of work run off the request path through one JobQueue interface — BullMQ on the server (Redis-backed, persistent, retry + concurrency control), inline fire-and-forget on Desktop:

• The agent run itself — execute() enqueues task.run onto the agent queue; a worker runs it and produces into the Redis stream buffer that the response reads back (see Task Orchestration).
• Post-turn jobs — once a turn’s stream drains (finalizeExecution, in task-runner.ts), three fire-and-forget jobs: credit billing (credit.consume), memory extraction (memory.extraction), and resource indexing (resource.index).

How BullMQ Works

BullMQ is a Redis-backed job queue for Node.js. Understanding its lifecycle is essential for reasoning about retry behavior and failure modes.

Three actors:

• Producer — the API server calls Queue.add(name, payload) to enqueue a job. This writes the job data to a Redis Stream and returns immediately. The producer never executes the job.
• Redis — stores all job state. Jobs live in Redis Streams (waiting list) and Sorted Sets (delayed/priority). Redis persistence (AOF/RDB) ensures jobs survive Redis restarts.
• Worker — a separate Node.js process that pulls jobs via blocking reads (BRPOPLPUSH). Each worker has a concurrency limit — it processes at most N jobs in parallel per queue.

Job state flow: waiting → active → completed or failed. On failure, if retries remain, the job moves to delayed (with exponential or fixed backoff), then back to waiting. When all retries are exhausted, the job stays in failed permanently — visible in the bull-board dashboard and triggering a webhook alert.

Why not just setTimeout or Promise?

• Process crash during setTimeout or void promise = job lost forever. BullMQ jobs persist in Redis.
• No retry with setTimeout. BullMQ retries with configurable backoff.
• No concurrency control with raw promises. BullMQ limits parallel execution per queue.
• No observability with fire-and-forget. BullMQ provides state, duration, attempt count, and failure reason.

JobQueue Interface

Following the project’s infra pattern (TaskLock, StreamBuffer, KeyEncryption), the queue is defined as an interface in @zapvol/backend/src/infra/job-queue.ts with two implementations:

export interface JobQueue {
  enqueue(
    jobName: string,
    payload: Record<string, unknown>,
    execute: () => Promise<void>,
    options?: JobEnqueueOptions,
  ): void;
}

The key design: enqueue takes both a serializable payload and an execute closure:

Implementation	payload	execute
createBullMQJobQueue()	Serialized to Redis	Ignored (worker reconstructs from payload)
createInlineJobQueue()	Ignored	Called directly (fire-and-forget)

The caller provides everything; the implementation picks what it needs. Both paths work from the same call site (finalizeExecution in task-orchestrator.ts).

Injection

// Server — BullMQ when Redis available, inline fallback
const queues = getQueues();
const jobQueue = queues ? createBullMQJobQueue(queues) : createInlineJobQueue(log);

// Desktop — always inline
const jobQueue = createInlineJobQueue(log);

Queue Topology

Six named queues (QUEUE_NAMES in apps/server/src/lib/queues.ts), split so one workload can’t starve another:

Queue	Jobs	Why it is isolated
agent	task.run / chat.run	The agent runs themselves — long-lived, IO-bound (awaiting LLM/tools); high concurrency, kept off the others so a burst of runs can’t starve billing / indexing
critical	credit.consume	Billing must drain fast — never starved by slow LLM work
llm	memory.extraction	Low concurrency — stays within provider rate-limits, caps token spend
indexing	resource.index	CPU / DB-bound — independent scaling
scheduling	scheduled task fires	wait_and_resume / cron continuations
nudge	nudge fires	Time-triggered nudges

The lazy-singleton Queue instances live in queues.ts, shared by the job queue, the bull-board dashboard, and the metrics endpoint. Each queue’s Worker is created in worker.ts with its own concurrency.

Worker Process

A single worker process (apps/server/src/worker.ts) hosts one Worker per queue and dispatches by job name via JOB_PROCESSORS. Run separately from the API server: pnpm worker or pnpm worker:dev.

apps/server/src/
  worker.ts                 → Entry point, one Worker per queue (dispatch via JOB_PROCESSORS)
  jobs/
    task-run.ts             → task.run (the agent run → produce into the stream buffer)
    credit-consumer.ts      → credit.consume
    memory-worker.ts        → memory.extraction
    resource-indexer.ts     → resource.index
    schedule-fire-runner.ts → scheduled task fires
    nudge-fire-runner.ts    → nudge fires

Each processor creates service instances at module level (same inline-assembly pattern as route files), then exports a plain async function:

export async function processCreditConsume(job: Job<CreditConsumePayload>) {
  const { userId, taskId, messageId, totalTokens } = job.data;
  await creditService.consume(userId, taskId, messageId, totalTokens);
}

Payload Contract

Payloads contain only serializable IDs, never runtime objects. Workers reconstruct service dependencies (sandbox, model factory, repos) from these IDs — closures and NodeSandbox file handles cannot be serialized.

Workers load messages via taskService.loadTaskData(taskId) and locate the target assistant message by assistantMessageId (not array position — a new round may start between enqueue and worker execution).

Graceful Shutdown

worker.close() waits for in-flight jobs, then stops pulling. SIGTERM/SIGINT trigger coordinated shutdown of all three workers + Redis connection.

Idempotency

Two layers of dedup:

1. Enqueue-level — BullMQ jobId (e.g., credit:{taskId}:{messageId}) prevents duplicate enqueue. Same job cannot be added twice.

2. Retry-level — If a job fails mid-execution and BullMQ retries it, the processor runs again. For credit.consume, the repository uses insert-first with a unique referenceId constraint on the ledger table (ON CONFLICT DO NOTHING). If the ledger row exists, balance deduction is skipped entirely — no TOCTOU race. Other processors are naturally idempotent (upsert/overwrite).

Observability

Three components:

• Bull-board dashboard — /admin/queues (admin auth). Full web UI for inspecting jobs, retrying failures, viewing payloads. File: apps/server/src/routes/admin-queues.ts.

• Queue metrics API — GET /api/admin/queue-metrics (admin auth). Returns per-queue counts: active, waiting, delayed, failed, completedTotal, failedTotal.

• Failure webhook — JOB_FAILURE_WEBHOOK_URL env var (optional). On final failure (all retries exhausted), sends a POST with job metadata. Worker also emits structured logs for active, completed, failed, stalled, and error events.

File Map

File	Role
packages/backend/src/infra/job-queue.ts	JobQueue interface + createInlineJobQueue()
apps/server/src/lib/queues.ts	Shared Queue instances (lazy singleton)
apps/server/src/infra/bullmq-job-queue.ts	createBullMQJobQueue()
apps/server/src/worker.ts	Worker entry point
apps/server/src/jobs/*.ts	Three job processors
apps/server/src/routes/admin-queues.ts	Bull-board dashboard route

Design Constraints

1. Serializable payloads only. Closures and file handles cannot cross the serialization boundary.
2. Message identity by ID. Workers find messages via assistantMessageId, not array position.
3. Desktop uses inline queue. Same JobQueue interface, fire-and-forget execution, no Redis.
4. Redis optional in dev. Server falls back to createInlineJobQueue() when REDIS_URL is unset.
5. MCP disconnect stays in-process. Depends on mcpClientManager state, not queued.