chatbot_v3/docs/adr/0014-durable-job-dispatch-with-rabbitmq.md

0014. Durable job dispatch with RabbitMQ

Status

Superseded by 0017

ADR-0017 defers this decision rather than rejecting it: ingestion currently runs inline in the POST /v1/files request, with no broker, no outbox, and no queue. This ADR remains the intended design for when ingestion becomes slow enough to need one — ADR-0017 lists the triggers, and names an in-process Postgres-claimed runner as the likely intermediate step before adopting a broker.

Context

ADR-0008 makes file ingestion job-shaped: POST /v1/files returns 202 Accepted with file_id, ingestion_job_id, and a queued/running status. ADR-0009 makes Postgres the durable source of truth for source_files, ingestion_jobs, and ingestion_job_events. ADR-0013 stores uploaded file bytes in MinIO so ingestion workers can fetch them after the request completes.

What is still missing is the delivery mechanism that tells workers which durable jobs are ready to process.

The broker decision has to preserve existing boundaries:

• Postgres remains authoritative for job status, progress, audit, tenancy, and retention-sensitive state.
• MinIO stores file bytes and derived blobs, not queue state.
• Broker messages carry ids and correlation metadata, not raw files or chunks.
• Chat runs are not queued by this service: ADR-0007/0008 allow one in-flight run per thread_id and return 409 Conflict for concurrent runs.
• Application-lifetime clients are created and closed by lifespan/startup owners under ADR-0012, not import-time globals.

Decision

Use RabbitMQ (via the aio-pika async SDK) as the durable pub/sub and job-dispatch broker for asynchronous ingestion and maintenance work.

RabbitMQ's work-queue model gives ingestion dispatch what it needs directly: durable exchanges/queues, manual acknowledgement, redelivery on nack/crash, and per-queue dead-lettering for poison messages, all while Postgres remains the source of truth for application state.

Start with ingestion jobs

The first broker-backed workflow is source-file ingestion:

POST /v1/files
  -> authenticate and resolve tenant
  -> store source bytes in MinIO (ADR-0013)
  -> create/update source_files row
  -> create ingestion_jobs row with status='queued'
  -> create an unpublished outbox_events row
  -> commit one transaction
  -> return 202 Accepted

outbox publisher
  -> publish the RabbitMQ message containing durable ids
  -> mark the outbox event published

A dedicated ingestion worker process consumes the message:

RabbitMQ message
  -> load ingestion_jobs/source_files from Postgres
  -> bind logging context from durable ids
  -> fetch source object from MinIO
  -> parse, chunk, embed, and mutate Qdrant points
  -> append ingestion_job_events
  -> mark ingestion_jobs succeeded/failed/cancelled
  -> ack or nack/reject the message

Worker-owned ingestion chunk CRUD

For file ingestion and re-ingestion, generated chunk/point CRUD is performed by the ingestion worker, not by the FastAPI publisher/request path.

The HTTP request path only owns the lightweight durable handoff:

• authenticate the caller and derive tenant context;
• store uploaded bytes in MinIO;
• create or update source_files;
• create ingestion_jobs(status='queued');
• create the corresponding unpublished outbox_events row;
• commit the Postgres transaction and return 202 Accepted.

The separate outbox-publisher process owns publication of the RabbitMQ message containing durable ids and records its outcome on the outbox event.

The ingestion worker owns the expensive and retryable side effects:

• parsing source files;
• chunking;
• embedding and reranking-vector generation when applicable;
• creating, updating, reordering, and soft-deleting generated Qdrant points;
• appending ingestion_job_events;
• updating ingestion job status, counters, and failure summaries.

Shared chunk/point mutation logic should live in a service/repository layer that can be reused by both direct API routes and ingestion workers. Direct /v1/points CRUD remains a synchronous API responsibility unless the operation is explicitly bulk or job-shaped. Ingestion-generated mutations are asynchronous worker responsibility because they must be safe to retry and resume by ingestion_job_id.

Keep messages small and versioned

Broker payloads carry identifiers and correlation metadata only. A typical message is:

{
  "type": "ingestion.job.created",
  "version": 1,
  "tenant_id": "...",
  "file_id": "...",
  "ingestion_job_id": "...",
  "request_id": "...",
  "api_key_id": "..."
}

Do not put raw file bytes, extracted text, chunks, embeddings, raw prompts, model outputs, or secrets in RabbitMQ messages.

Workers must treat message metadata as routing/correlation input, not as the only authority. Before doing tenant-scoped work, workers reload the durable job and file rows from Postgres and verify the ids are consistent.

Use explicit routing keys, a topic exchange, and durable queues

Use stable, dot-separated routing keys. Initial routing keys:

ingestion.job.created
ingestion.job.retry_requested
maintenance.retention.requested
maintenance.erasure.requested

Create an application-owned durable topic exchange for dispatching work messages, and durable (quorum-type) queues bound to it by pattern, for example:

exchange: chatbot.jobs (topic, durable)

queue: ingestion.jobs   <- binding pattern ingestion.job.*
queue: maintenance.jobs <- binding pattern maintenance.*

Multiple worker processes share a queue as competing consumers, so work is distributed without each process independently receiving every message. Set consumer prefetch (QoS) explicitly rather than relying on the client default, so one slow consumer cannot starve the others or accumulate unbounded unacked messages.

Ack messages only after the worker has persisted the resulting job state and progress events to Postgres. Redelivery must be safe: ingestion processors should be idempotent by ingestion_job_id and Qdrant point ids/upsert semantics.

Run subscribers as worker processes, not FastAPI side effects

The outbox-publisher process owns the RabbitMQ publisher connection/channel and publishes committed outbox events. FastAPI routes create durable outbox intent only; they do not publish ingestion notifications directly.

Subscribers should run in separate worker process entrypoints. Do not start a consumer inside every FastAPI web worker by default: with multiple Uvicorn/Gunicorn workers, each process has its own lifespan and could create surprising duplicate consumers or resource pressure.

Worker processes create their own application-lifetime resources at startup:

• SQLAlchemy engine/sessionmaker;
• MinIO/S3 object-storage client;
• Qdrant client;
• RabbitMQ connection/channel (aio-pika, via aio_pika.connect_robust for automatic reconnection);
• observability/logging clients;
• ingestion model clients.

Each message gets its own request/job-lifetime SQLAlchemy AsyncSession and explicit transaction boundaries, following ADR-0012.

Use a transactional outbox for job dispatch

Use a transactional outbox to atomically record the ingestion job and the intent to dispatch it. The HTTP publisher does not publish directly to RabbitMQ after committing the ingestion-job transaction.

Instead, the same Postgres transaction creates or updates the durable job state and inserts an unpublished outbox event:

same DB transaction:
  insert/update source_files
  insert ingestion_jobs(status='queued')
  insert outbox_events(type='ingestion.job.created', payload, published_at=NULL)
  commit

A separate outbox publisher process reads unpublished events, publishes the versioned message to RabbitMQ, and records the publication outcome:

outbox publisher:
  claim unpublished outbox event
  publish to RabbitMQ (persistent message, publisher confirms enabled)
  mark outbox event published once the broker confirms receipt

This removes the failure window where an ingestion_jobs row commits but the API process crashes before recording that it must be dispatched. If the transaction commits, the job and its durable dispatch intent both exist; the outbox publisher can resume publication after a restart.

An outbox event must have a stable UUID event id. Include that id in the RabbitMQ message so consumers and operators can correlate it back to the outbox record. RabbitMQ has no broker-native publish-deduplication mechanism comparable to some other brokers: publisher confirms guarantee the broker accepted the message, but the publisher can still crash after the broker confirms and before published_at is recorded, causing a duplicate publish on retry. There is no broker-side safety net for that case here — workers must be idempotent by ingestion_job_id, and Qdrant mutations must use deterministic point ids/upsert semantics, as the sole guarantee against duplicate processing.

Outbox events are delivery records, not a replacement for ingestion_jobs or ingestion_job_events. Postgres continues to own application job state and progress. A stale-job scanner remains a repair/operational check for jobs or outbox events that have not advanced as expected, not the primary dispatch path.

Add the outbox_events schema through an Alembic migration. Do not create or alter the table at FastAPI startup, as required by ADR-0009.

Configure retries, dead letters, and observability

Declare queues with an explicit dead-letter exchange (x-dead-letter-exchange) pointing at a durable chatbot.jobs.dlx fanout exchange bound to a durable dead-letter queue. A message lands there when a worker rejects/nacks it with requeue=False. Workers should nack with requeue=False after a bounded local retry count for a given delivery, rather than looping redelivery indefinitely. Poison messages or repeatedly failing jobs should end as durable Postgres failures with an ingestion_job_events error entry, and the dead-letter queue gives operators a place to inspect the raw message for messages that failed before any Postgres state could be written.

At worker ingress, bind structured logging context from durable identifiers:

• tenant_id;
• request_id;
• api_key_id;
• file_id;
• ingestion_job_id;
• RabbitMQ exchange/routing key/delivery tag when available.

Use stable event names such as:

• ingestion.job.dispatched;
• ingestion.job.received;
• ingestion.job.completed;
• ingestion.job.failed;
• broker.publish.failed;
• broker.message.redelivered.

Do not queue chat runs through RabbitMQ yet

ADR-0007 and ADR-0008 intentionally reject queueing concurrent chat runs per thread: one run may be in flight, and concurrent runs return 409 Conflict. This ADR does not change that decision.

A future ADR may introduce broker-backed work for chat-adjacent tasks such as summarization, retention, erasure, or offline evaluation, but synchronous chat run execution remains owned by the FastAPI/LangGraph boundary unless superseded.

Consequences

Positive

• File ingestion can move out of HTTP request handling into scalable worker processes without changing the REST contract from ADR-0008.
• RabbitMQ gives durable delivery, manual ack/redelivery, per-queue dead-lettering, and competing-consumer work distribution without adopting Kafka-level operational complexity.
• Postgres remains the auditable source of truth for jobs, progress, tenant ownership, failure state, and the durable intent to dispatch each job.
• The transactional outbox prevents a committed job from losing its initial dispatch intent when the API process fails before a direct broker publish.
• MinIO, Postgres, Qdrant, and the broker each have distinct responsibilities: bytes, metadata/state, vector search, and work delivery.
• Separate worker processes avoid surprising subscriptions in every FastAPI web worker and make GPU/model-heavy ingestion resources easier to size.

Negative

• Adds another external service to run, secure, monitor, and back up where queue/exchange durability is required.
• Developers must reason about redelivery and idempotency; handlers may run more than once for the same ingestion_job_id.
• Adds an outbox_events table, Alembic migration, outbox publisher process, and monitoring for unpublished or stuck events.
• RabbitMQ has no broker-native publish deduplication. Unlike brokers that offer one, duplicate publication after a publisher crash between broker confirmation and outbox-event marking is not mitigated by the broker at all; idempotent consumers are the only protection.
• RabbitMQ exchange, queue, binding, and dead-letter-exchange configuration must be managed explicitly per environment.
• The app now needs web, ingestion-worker, and outbox-publisher deployment/runbook conventions.

Alternatives Considered

• NATS JetStream: rejected. JetStream keeps the broker footprint lighter and offers built-in publish deduplication, but the project wants RabbitMQ's more mature work-queue/routing model (topic exchanges, per-queue DLX, prefetch-based fair dispatch) and the aio-pika SDK for asyncio integration.
• Redis Streams: rejected for this decision. Redis Streams would work for an MVP and has simple local deployment, but RabbitMQ is a cleaner dedicated broker for durable pub/sub and worker coordination without also mixing cache responsibilities into the same service.
• Kafka: rejected. Kafka is an excellent durable event log at high scale, but its operational overhead is unnecessary for this service's initial ingestion and maintenance jobs.
• FastAPI BackgroundTasks or in-process asyncio tasks: rejected as the durable design. They are simple, but work can be lost on process restart and they do not coordinate multiple workers cleanly.
• Post-commit direct RabbitMQ publish plus a repair scanner: rejected as the primary dispatch design. It is simpler, but a crash after the job transaction commits and before the publish is recorded can strand a queued job until the scanner finds it. The transactional outbox persists the dispatch intent in the same transaction as the job.
• Database polling only: rejected as the main dispatch mechanism. Polling Postgres for queued jobs is a useful repair path, but relying on polling alone adds latency and unnecessary database load once a broker is available.
• Use the broker as the source of truth for job state: rejected. Broker state is delivery state. Application job state, progress, audit, and tenant ownership remain in Postgres.