1.3 bullmq

1.3. BullMQ Async Processing

3F never does inference on the HTTP thread. All heavy work — prompt processing, media ingestion — runs as BullMQ jobs on KeyDB (a Redis-compatible store), with persistence of terminal failures in PostgreSQL.

Queue topology

Piece	Implementation
Broker	KeyDB (infra.compose.yml, configured by server/keydb.conf)
Module	@ehildt/nestjs-bullmq + @ehildt/nestjs-bullmq-logger
Producer	HarnessQueueService (harness/services/harness-queue.service.ts)
Consumer	HarnessProcessor (harness/processors/harness.processor.ts) on the HARNESS_QUEUE
Queue name constant	modules/bullmq/constants/bullmq.constants.ts

The single HARNESS_QUEUE is deliberately the only hot path; queue-depth scaling happens by adding workers behind it, not by multiplying queues.

Job envelope

HarnessQueueService.emit enqueues { buffers, meta, filters }:

• buffers/meta — image payloads and their metadata destined for MinIO, deduplicated by content hash;
• filters — everything the processor needs without touching the request: model, prompt, identity (requestId, sessionId, conversationId), routing (roomId, event), and parameters (stream, think, numCtx, sessionMetadata).

Large binary content moves through MinIO references, not through the queue payload.

Retry & backoff

Defaults come from BULLMQ_* env knobs (validated Joi config): attempts, backoff type/delay, LIFO/priority options, stack-trace limits, worker concurrency, TLS. Effective values are introspectable via GET /api/v1/bullmq/retry-config; live state via GET /api/v1/bullmq/live (the SysCtl console polls these).

Failure handling — the persisted DLQ

BullMQ's own failed set is volatile. 3F therefore mirrors terminally failed jobs into HarnessDlq (Prisma/PostgreSQL) with the full envelope and error context:

Operation	Where
Inspect records	GET /api/v1/dlq/:requestId (dashboard DLQ area)
Edit before replay	PATCH /api/v1/dlq/:requestId / …/upsert
Re-queue	POST /api/v1/dlq/reinstate
Purge	DELETE /api/v1/dlq/:requestId

A LifecycleService (modules/dead-letter/services/lifecycle.service.ts) coordinates DLQ bookkeeping.

Crash recovery

On boot, JobReinstatementService (modules/minio/services/job-reinstatement.service.ts) reconciles jobs that were interrupted mid-flight (container restart, OOM) against stored payloads and re-queues them. Combined with image payload persistence in MinIO, no user request is lost to a restart — worst case it is replayed.

Cancellation

POST /api/v1/harness/cancel writes a cooperative cancellation marker for the requestId (KeyDB). The step engine checks it at boundaries; the queue entry is consumed normally and terminates early. See 1.2 (Cancellation).

Scaling notes

• More throughput: run additional replicas of the server image as pure workers — the queue is external, no code change required.
• KeyDB being Redis-protocol compatible means Redis can stand in without code changes (config only).
• Buffer-heavy jobs (multi-image chats) stay cheap for the queue because image bytes live in MinIO; payloads carry hashes.