Event-Driven AI Architecture: How to Scale Asynchronous LLM Workers with Kafka/RabbitMQ

A supplier invoice is uploaded. OCR text goes to a model that extracts line items, tax amounts, and currency. The result will eventually enter an ERP. In the first implementation, the endpoint waits for the model. The demo works. Then one tenant uploads ten thousand documents, revealing the actual protocol: timeouts, resubmissions, duplicate model calls, and records whose processing state nobody can establish.

The system we will design accepts invoice extraction as an asynchronous job. PostgreSQL owns job state, a broker carries execution signals, workers consume bounded capacity, and ERP delivery has its own state machine. Ending the HTTP request early is only one part of the design. We also need ownership, replay rules, deadlines, and accounting for external side effects.

The numbers below are explicit planning assumptions, not a TankDev customer benchmark. SQL targets PostgreSQL. Code fragments expose the important protocol boundaries; they are not a complete broker client or runnable service repository. Example settings are not production configuration until the provider, broker version, and deployment constraints have been verified.

1. The LLM bottleneck: concurrency and timeout math for a 15-second call

A synchronous HTTP contract is not the same as blocking I/O. An async server can yield its event loop while waiting for the model. The socket, proxy stream, request context, outbound connection, and cancellation bookkeeping still exist. A thread-per-request implementation also retains waiting threads. If a database transaction spans inference, the connection pool may saturate before the GPU does.

For a stable system, Little’s law gives average concurrency: L = λ × E[S]. At 20 requests per second and a mean service time of 15 seconds, L is 300 in-flight calls. Substituting p95 does not produce mean concurrency. A system accepting 20 requests/s while completing 2 jobs/s is not stable: backlog grows, so that simple steady-state interpretation does not apply.

L = arrival_rate * mean_service_time
  = 20 requests/s * 15 s = 300 in-flight requests

HTTP elapsed = admission + queue_wait + inference + persistence + response
timeout occurs when HTTP elapsed > the effective request deadline

Illustrative distribution, independently assumed:
P(HTTP elapsed > 30 s) = 0.02
30,000 requests * 0.02 = 600 expected deadline exceedances

A 15-second mean does not determine the timeout rate at 30 seconds; the distribution is required. The 2% above is an independent assumption. In production, traces must show whether the gateway, application deadline, or provider timeout fired first. A disconnected user does not prove that generation stopped or that the provider incurred no charge.

Here, POST /extraction-jobs handles authentication, document-reference validation, and durable acceptance, returning 202 Accepted with job_id and Location. Acceptance latency and extraction completion latency are separate SLIs. Token streaming remains useful for an interactive chat. Making a bulk document pipeline depend on one live HTTP connection creates an avoidable failure boundary.

2. Queue-based workers: durable acceptance before delivery

An idempotency key is unique within a tenant. The same key and normalized request return the same job_id; reusing it for different input returns 409. The request hash covers document version, extraction schema, model policy, and result-affecting options. A filename is not document identity. Store a content digest and immutable object version; an expiring signed URL is not a durable job identifier.

CREATE TABLE extraction_jobs (
  id uuid PRIMARY KEY,
  tenant_id uuid NOT NULL,
  idempotency_key text NOT NULL,
  request_hash text NOT NULL,
  object_version text NOT NULL,
  state text NOT NULL CHECK (state IN
    ('READY','RUNNING','RETRY_SCHEDULED','SUCCEEDED',
     'REVIEW_REQUIRED','FAILED_TERMINAL')),
  attempt integer NOT NULL DEFAULT 0,
  fence bigint NOT NULL DEFAULT 0,
  lease_until timestamptz,
  available_at timestamptz NOT NULL DEFAULT now(),
  deadline_at timestamptz NOT NULL,
  result_ref text,
  UNIQUE (tenant_id, idempotency_key)
);
CREATE TABLE outbox_events (
  event_id uuid PRIMARY KEY,
  job_id uuid NOT NULL REFERENCES extraction_jobs(id),
  event_type text NOT NULL,
  payload jsonb NOT NULL,
  published_at timestamptz
);

BEGIN;
INSERT INTO extraction_jobs
 (id, tenant_id, idempotency_key, request_hash,
  object_version, state, deadline_at)
VALUES (:job_id, :tenant, :key, :hash, :version, 'READY', :deadline)
ON CONFLICT (tenant_id, idempotency_key) DO NOTHING
RETURNING id;
-- If no row was inserted: SELECT the existing row, compare request_hash,
-- return its id on equality, or roll back and respond 409 on mismatch.
-- Only the inserting branch executes the outbox INSERT below.
INSERT INTO outbox_events (event_id, job_id, event_type, payload)
VALUES (:event_id, :job_id, 'extraction.requested.v1',
        jsonb_build_object('job_id', :job_id, 'schema_version', 1));
COMMIT;

The job and its outbox event commit together. Publishing before the job exists lets consumers observe nonexistent work. Writing the job and then publishing leaves a crash window that can strand it. An outbox relay publishes committed records and records published_at after broker confirmation. If the process dies between confirmation and that update, it publishes the event again. Consumers therefore assume at-least-once delivery from the outset.

The relay also needs ownership: bounded batches, SKIP LOCKED or leased claims, crash recovery, and alerts on publication age. Do not retain unbounded transactions while calling the broker. Carry job_id and event version rather than invoice text. Workers obtain tenant, document, and authorization data from an authoritative store; a tenant_id inside a message is not a security boundary.

RabbitMQ: publisher confirms, consumer acks, and a durable inbox

For this invoice pipeline, RabbitMQ is a reasonable first choice: task distribution, bounded in-flight delivery, and a broker the operations team understands are sufficient. Durable queues, persistent messages, publisher confirms, and replication policy must be considered together. A publisher confirm does not mean a consumer ran inference. A consumer ack does not establish a single provider invocation. The acknowledgements cover different boundaries.

A short-lived ingress consumer can separate broker delivery from slow inference. It inserts a durable inbox record, unique on event_id, and marks the job runnable in one transaction, acknowledging only after commit. Inference workers lease jobs from that durable store. The ack means execution responsibility was durably transferred, not that inference finished. An inbox used only as a dedup table, without a scheduler, can strand acknowledged work.

With that design, prefetch bounds ingress deliveries to the database, not inference concurrency. Worker concurrency is a separate semaphore and distributed admission budget. In the alternative where consumers run inference directly, prefetch should track actual worker parallelism, and acknowledgements follow durable result or retry-state commits. Those two designs have different acknowledgement points.

-- The earlier job schema is a subset; production also stores
-- state_version, policy versions, audit timestamps, and attempt outcomes.
CREATE TABLE execution_inbox (
  event_id uuid PRIMARY KEY,
  job_id uuid NOT NULL REFERENCES extraction_jobs(id),
  payload_digest text NOT NULL,
  received_at timestamptz NOT NULL DEFAULT now()
);

BEGIN;
INSERT INTO execution_inbox(event_id, job_id, payload_digest)
VALUES (:event_id, :job_id, :digest)
ON CONFLICT (event_id) DO NOTHING;
-- Check existing digest on duplicates; mismatch => quarantine + alert.
-- The job was already durably READY in the acceptance transaction.
-- Do not reset a RUNNING or terminal job on redelivery.
COMMIT;
-- RabbitMQ: basic_ack on the delivering channel only AFTER this commit.
-- Kafka: advance contiguous committed ingestion offsets only AFTER commit.
-- A durable scheduler scans eligible jobs even without another notification.

Kafka: if replay matters, design the offset protocol too

Kafka becomes relevant when audit, analytics, and other consumers need independent reads, retention, replay, or partition-local ordering. A slow LLM call alone does not require Kafka. Stacking both brokers for the same task is not the default either. Start with consumer behavior, retention, and operating expertise rather than a throughput label.

Disable auto-commit in the Kafka ingress consumer. Persist each partition/offset record to the inbox before committing its offset. Redelivery across the crash window is safe because inbox insertion is unique. The committed offset identifies the next record to read: after durably ingesting offset 41, advancement to 42 is safe. With parallel ingestion, an unfinished offset 40 prevents advancement past it; commit only a contiguous durable prefix.

Do not attach the polling loop to a 15-second inference call. Pause partitions when the dispatcher is full while preserving the client’s polling and heartbeat contract; test max.poll.interval.ms and rebalances for the chosen client version. Kafka transactions can provide strong guarantees across Kafka topics. An external model provider and a PostgreSQL transaction do not automatically join that scope. Broker exactly-once terminology does not imply one billable LLM call.

Worker ownership: leases, fencing tokens, and short transactions

-- One short claim transaction; inference runs AFTER COMMIT.
BEGIN;
WITH candidate AS (
  SELECT id FROM extraction_jobs
  WHERE state IN ('READY', 'RETRY_SCHEDULED')
    AND available_at <= now() AND deadline_at > now()
  ORDER BY available_at, id
  FOR UPDATE SKIP LOCKED LIMIT 1
)
UPDATE extraction_jobs j
SET state='RUNNING', attempt=j.attempt+1,
    fence=j.fence+1, lease_until=now()+interval '90 seconds'
FROM candidate c WHERE j.id=c.id
RETURNING j.id, j.tenant_id, j.attempt, j.fence, j.lease_until;
COMMIT;

-- Heartbeat from the current owner; zero updated rows means ownership lost.
UPDATE extraction_jobs SET lease_until=now()+interval '90 seconds'
WHERE id=:job AND state='RUNNING' AND fence=:fence
  AND lease_until > now();

-- In ONE transaction: conditional state update + result/event INSERTs.
UPDATE extraction_jobs
SET state='SUCCEEDED', result_ref=:result_ref, lease_until=NULL
WHERE id=:job AND state='RUNNING' AND fence=:fence
  AND lease_until > now()
RETURNING id;
-- Insert result metadata and extraction.completed outbox event only
-- when exactly one row was updated; otherwise discard stale completion.

The 90-second lease is illustrative. Choose it alongside a 45-second total provider-attempt deadline, heartbeat interval, and shutdown margin. A reaper conditionally transitions expired RUNNING jobs to RETRY_SCHEDULED, increments the fence, and persists the next attempt time. Exhausted attempts or expired job deadlines become terminal. A pod restart never resets the attempt counter.

Fencing prevents a stale worker from overwriting a newer result; it cannot undo an external request. If worker A stalls, loses its lease, and receives a response after B takes over, A cannot commit. Both provider calls may still be billable. Sending job_id in an arbitrary header does not solve this without a documented provider idempotency contract. ERP writes therefore remain outside the extraction transaction too.

3. Backpressure and rate limiting: accepting 10,000 documents is not invoking them together

A queue does not create capacity; it spreads a capacity deficit over time. An unbounded queue merely moves the failure date. Admission therefore checks pending jobs, bytes, estimated tokens, and remaining spending budget per tenant. Accepted work must not disappear silently. Before acceptance, tenant-quota violations can return 429, while exhausted system admission capacity can return an appropriate 503 with retry guidance.

attempt_capacity = min(RPM / 60,
                       TPM_budget / (60 * reserved_tokens),
                       concurrency / mean_inference_seconds)
                 = min(10, 1.6667, 2) = 1.6667 attempts/s

job_capacity = attempt_capacity / mean_attempts_per_job
             = 1.6667 / 1.10 = 1.5152 jobs/s

10,000-job drain time, with no new arrivals:
  10,000 / 1.5152 = 6,600 s = 110 minutes

With sustained new arrivals of 1 job/s:
  net drain rate = 1.5152 - 1 = 0.5152 jobs/s
  initial backlog drains in ~19,412 s = 5.39 hours

This is a steady-state planning approximation; token lengths, burst windows, and latency distributions change actual completion times. The 6,000-token figure is a conservative application reservation, not a universal definition of provider rate-limit accounting or billing. Read RPM, TPM, daily limits, and shared model pools from the real account. Cost per completed job must use reported usage and actual completed jobs rather than that fixed reservation.

Giving three pods independent semaphores of 30 creates a global ceiling of 90. The limiter scope is the provider project or shared model pool, not the pod. Atomic multidimensional admission acquires request budget, token reservation, and in-flight permit together. Redis with Lua is one implementation; Redis is not the sole durable record of work. Decide in advance whether limiter failure permits a bounded local emergency allowance or fails closed.

Retrying at the same rate after 429 reduces useful capacity further. Share cooldown state and valid provider delay hints across workers. Request budget, token reservation, and in-flight permits do not share identical refund semantics: fully refunding an attempted request after a timeout can undercount actual consumption. Reconcile reported usage while retaining uncertain attempts in the accounting ledger.

Tenant fairness matters too. One tenant’s ten thousand documents should not leave another tenant’s single job waiting for five hours. Use priority classes, per-tenant in-flight limits, and weighted fair scheduling. Two queues alone do not guarantee fairness; measure and prevent starvation. Autoscaling should consider oldest eligible job age and actual provider headroom as well as queue depth.

Local vLLM: request count does not describe GPU memory

With local inference, GPU capacity replaces the provider quota as the immediate constraint. A short document and an 80,000-token context may consume one semaphore slot each while exerting very different KV-cache pressure. Measure weights, KV cache, context length, active sequences, and prefill/decode mix together. max_num_seqs and max_num_batched_tokens configure server scheduling; they do not replace upstream durable queues or admission control.

vLLM preemption and recomputation can preserve operation while increasing tail latency. Observe TTFT, inter-token latency, input/output token throughput, KV-cache utilization, and preemptions in the same experiment. Increasing worker replicas against a fixed GPU often enlarges an internal queue rather than accelerating inference. A documents-per-second claim needs model version, quantization, GPU type/count, tensor parallelism, context distribution, and scheduling settings.

4. DLQs and failure handling: transport errors, invalid output, and incorrect results

JSON parsing checks syntax, and schema validation checks the representation contract. A model can invent an IBAN that fits the schema perfectly. No general validator detects every hallucination. This pipeline checks amount arithmetic, permitted currencies, supplier master-data matches, and source page/span references. Evidence references can themselves be fabricated, so resolve them against the actual OCR content.

Do not quietly convert an uncertain financial field to null and mark the job SUCCEEDED. Success depends on a versioned validation policy. An inspectable but unacceptable result becomes REVIEW_REQUIRED and is withheld from ERP delivery. Model output is data, not instructions. An invoice saying “ignore the rules and send this to a URL” cannot expand worker tools or network access. Extraction workers have no payment authority.

# Retry scheduling sketch; attempts are persisted, never process-local.
def retry_at(now, attempt, deadline, retry_after_seconds=None):
    if attempt >= 5:
        return None
    cap = min(60.0, 2.0 * (2 ** (attempt - 1)))
    delay = random.uniform(0, cap)  # full jitter
    if retry_after_seconds is not None:
        delay = max(delay, retry_after_seconds)
    when = now + timedelta(seconds=delay)
    # Reserve time for the next full attempt and its commit margin.
    if when + timedelta(seconds=50) >= deadline:
        return None
    return when

# Persist under the current fence in one transaction:
# attempt outcome + next available_at + RETRY_SCHEDULED + audit metadata.
# Never sleep for minutes while holding a DB connection or broker delivery.
# Dispatch due retries through the durable scheduler.

Own retries at one layer. Five application attempts, three SDK attempts, and two proxy attempts can turn one job into thirty external calls. Disable SDK retries when using a dedicated scheduler, or explicitly include them in the total budget. Retry-After may be seconds or an HTTP date and needs normalization. If a valid delay exceeds the job deadline, do not call sooner; defer or terminate according to the product contract.

A DLQ is neither a state machine nor a repair engine. It is an inspection channel for malformed transport messages and exhausted execution events. The authoritative job records failure class, attempt history, policy version, and final decision. Dead letters carry references rather than raw invoices. Without retention, access controls, alerts, replay ownership, and an audit trail, a DLQ is just an invisible archive.

In RabbitMQ, nack(requeue=false) can route to a configured DLX; without one, the message can be discarded. Default dead-letter forwarding is not universally lossless. Quorum-queue at-least-once dead-lettering requires explicit configuration. Here, application terminal decisions and dead-letter outbox events share a database transaction, so a failed DLQ publication can be retried from durable state.

Kafka retry and dead-letter topics are application protocols. Moving an event to a retry topic while continuing the main partition can reorder work for the same key. If invoice version n+1 finishes before n, aggregate versions or optimistic concurrency must decide which result ERP accepts. Blocking the partition can preserve order while stalling unrelated documents. An ordering claim is incomplete without that trade-off.

5. Delivering asynchronous results: WebSocket notifications, webhooks, and an ERP inbox

GET /extraction-jobs/{id} is the authoritative status endpoint. Expose READY, RUNNING, REVIEW_REQUIRED, and terminal outcomes; operational retry details require separate permissions. Every read needs tenant scoping and object-level authorization. An unguessable UUID is not RBAC. Permissions checked at WebSocket subscription must also handle revocation during long-lived sessions.

WebSocket or SSE messages notify that a job changed. If a notification is lost, the client reads state again. Reconnects may use bounded replay from an event cursor, otherwise a fresh snapshot. Older state_version values must not roll the UI backward. SSE can suffice for one-way notifications; WebSockets are not mandatory without bidirectional traffic. Polling with backoff and jitter remains a valid fallback.

HTTP/1.1 202 Accepted
Location: /extraction-jobs/0190-example
Content-Type: application/json

{
  "job_id": "0190-example",
  "state": "READY",
  "status_url": "/extraction-jobs/0190-example"
}

// Later notification; fetch status_url for authoritative state.
{
  "event_id": "evt-example",
  "type": "extraction.completed.v1",
  "job_id": "0190-example",
  "state_version": 7
}

ERP delivery is independent of extraction. The result reference, validation policy, and extraction.completed outbox event commit together. A webhook dispatcher tracks delivery_id, attempts, next_attempt_at, and the latest HTTP outcome. It does not rerun inference. If ERP is unavailable for two hours, the delivery backlog should grow without paying to extract completed documents again.

Webhook signature contract (example, versioned):
signed_bytes = timestamp_ascii + b"." + delivery_id_ascii + b"." + raw_body
signature = HMAC-SHA256(endpoint_secret, signed_bytes)

Receiver:
1. Enforce TLS, payload limit, supported signature version and key id.
2. Validate timestamp window; compare signature in constant time.
3. Validate event schema and resolve tenant from trusted endpoint config.
4. BEGIN
     INSERT inbox(delivery_id, payload_digest, ...)
       ON CONFLICT DO NOTHING;
     -- Same id + different digest => reject and alert.
     -- New delivery => persist runnable ERP operation in same transaction.
   COMMIT
5. Return 2xx only after durable acceptance.
6. Apply ERP side effect with its own idempotency/version contract.

HMAC alone does not prevent replay; combine a timestamp window with durable delivery_id deduplication. A retry may receive a fresh signature timestamp while retaining the same logical delivery_id. Support key IDs for secret rotation and sign the exact raw body, not reserialized JSON. ERP returning 2xx may mean durable inbox acceptance. Commercial completion requires a separate acknowledgement event or status lookup.

Never call an arbitrary callback URL copied from a job payload. Use tenant-registered, verified endpoints, controlled DNS resolution, redirect rules, and egress restrictions against private and link-local destinations. URL string validation alone is not an SSRF boundary. Retry 429 and transient 5xx within budget, alert on repeated 401/403, and isolate failing tenants with circuit breakers. Giving every attempt a new delivery_id defeats deduplication.

Production acceptance: a healthy broker can coexist with a stalled pipeline

Queue depth can be zero while ten thousand ingested jobs still wait in the inbox. Measure broker lag, durable runnable backlog, and completion latency separately. Correlate initial HTTP acceptance, outbox publication, broker ingestion, each inference attempt, and webhook delivery. Do not hide multiple attempts in one span. Keep job_id, attempt, and fence in logs, not unbounded metric labels.

The 120-second objective does not cover the ten-thousand-document burst; capacity math already rules that out. A normal-load contract might instead specify 0.5 jobs/s, a mean reservation of 6,000 tokens, and bounded bursts. Report input/output token distributions, provider limits, worker counts, and database/broker resources alongside p50/p95/p99 latency. Use open-loop traffic to observe backlog growth; a closed loop that sends only after each response can hide saturation.

The SLI denominator is admitted jobs within the supported document class; late jobs and terminal failures are not successful completions. REVIEW_REQUIRED has its own review SLO and rate, and does not count as automatic success unless the product contract explicitly defines it that way. Returning errors quickly or routing every document to review must not artificially improve completion performance.

Kill a worker just after the provider responds, stop a consumer between database commit and ack, and sever broker connectivity. Inject a 429 storm, timeouts, malformed JSON, and ERP downtime. Acceptance means every admitted job remains explainable, stale workers cannot replace terminal results, and ERP effects do not duplicate. That last property requires an end-to-end test of ERP’s idempotency contract.

Our API and system integration article examines idempotency and outbox boundaries. From PoC to production connects SLOs, cost, and architectural decisions. AI automation reliability covers the boundary between model output, authorization, and business decisions. Queues carry those contracts; they do not replace them.

The decisive production choices precede broker branding: where acceptance commits, how execution ownership changes, and which external effects can repeat. Without those answers, more workers only produce the same uncertainty faster.

Technical references and guarantee boundaries

Consult RabbitMQ acknowledgements and publisher confirms for acknowledgement scope, and RabbitMQ DLX documentation for dead-letter forwarding options. The durable inbox and outbox protocol here is our application design built around those boundaries.

See the Apache Kafka design document for exactly-once boundaries involving external systems. Offset management and inbox handoff are application choices in this article, not a Kafka guarantee covering an external LLM call.

For quota and retry behavior, see OpenAI rate-limit documentation. For GPU scheduling constraints, see vLLM optimization and tuning. Quotas, SLOs, and hardware examples in this article are not product defaults or performance promises.