Federation broker — AMQP queues

Why this exists / when to use what

The federation broker is a single RabbitMQ instance shared by every long-running batch pipeline in the fleet (gpumon-ocr, fraud-heuristics, edgar-cik-cli, and future projects). It exists so producers can publish a unit of work (an OCR job, a fraud row, a CIK, a filing) without knowing which worker on which machine will pick it up, and so workers can be horizontally scaled by adding replicas — the broker handles the fan-out.

Use the broker when: the work is expensive (many LLM calls per unit), batchable (thousands of units), and safe to retry (the consumer handler is idempotent). For a one-off LLM call, use ingress directly. For "what did I just process?" state, the consumer should write to gpumon-writer — the broker is fire-and-forget plus DLQ, not durable state.

Address

The password lives in .env (RABBITMQ_PASSWORD). Generate with:

openssl rand -base64 24 | tr -d '/+=' | head -c 32

vhost is federation (not the default /). All exchanges, queues, bindings, and policies live in this vhost.

Topology

Declared by services/broker/rabbitmq-definitions.json, loaded at boot via load_definitions. Producers and consumers MUST NOT redeclare queues with different arguments — let the definitions file own the shape.

Exchanges

Queues

All non-DLQ queues are:

• durable: true (survive broker restart)
• auto_delete: false
• bound to federation.work with routing key = queue name
• DLX-routed on dead-letter to federation.dlx with routing key <domain>.dlq

Policy ha-default

Applied to all queues in the vhost (pattern .*):

• message-ttl: 604800000 (7 days fallback if per-queue TTL not set)
• max-length: 100000 (fallback cap)

Per-queue x-message-ttl and x-max-length arguments override the policy.

Permissions

The federation user has configure=.* write=.* read=.* on the federation vhost. There is no per-queue ACL — same trust boundary as the rest of the LAN-only services.

Connection requirements

Heartbeat — 600s MANDATORY

Workers that run long-tail jobs (OCR pipeline 5-30 min, SEC downloads 1-10 min) MUST connect with heartbeat: 600 (seconds). The 60s default silently kills the connection mid-job:

• Broker stops receiving heartbeats while the handler awaits a slow operation.
• Broker drops the connection (after 2× heartbeat without traffic).
• Next ack() / nack() after the handler returns throws IllegalOperationError: Channel closed.
• Process crashes with no graceful re-queue.

This is not theoretical — it took down the edgar-filing-ingester on 2026-05-19. See /tmp/cane-audit-report.md finding 2.

Set via AMQP_HEARTBEAT_SEC=600 env var. The producer/consumer skeleton in client-pattern.md reads this var.

Prefetch

One channel per process with prefetch=1. Scale workers by adding replicas, not bumping prefetch. Reason: prefetch>1 means a slow in-flight job stalls N-1 others that are already reserved to this channel.

Confirm publishes

Producers MUST use createConfirmChannel() and await the publish confirm. Without confirms a transient broker hiccup silently drops the envelope.

TLS

Plain AMQP (amqp://). No TLS — LAN-only, behind the UDM firewall. If this ever needs to traverse a public link, switch to amqps:// and mount a CA bundle.

Sanity checks

# List queues + depths
ssh rooot@node-eleven 'docker exec gpumon-rabbitmq rabbitmqctl list_queues -p federation name messages consumers'

# Tail a queue without consuming (mgmt UI)
open http://node-eleven:15672/#/queues/federation/ocr.jobs

# Replay a DLQ message to its source queue (manual, see client-pattern.md)

Cross-references

• envelope.md — Envelope v1 + per-domain payload schemas.
• client-pattern.md — Reusable producer + consumer skeleton.
• ocr-api.md — ocr.jobs producer (HTTP frontend).
• Federation queue master design: ../specs/federation-queue-architecture.md