Client pattern — producer + consumer skeletons

Why this exists / when to use what

This file is the copy-paste reference for new producers and consumers that talk to the federation broker. The shape is taken verbatim from the production reference impls (services/ocr-api/src/amqp.ts for the producer, services/ocr-worker/src/amqp.ts for the consumer). Both files are ~100 lines each — small enough to duplicate per project rather than factor into a shared library. Premature factoring across repo boundaries is worse than measured duplication.

Use these skeletons when standing up a new worker (fraud-enricher-v2, edgar-cik-resolver, edgar-filing-ingester). Read broker.md first for topology and envelope.md for payload shapes.

Connection requirements (recap)

Producer skeleton

// producer.ts — generic federation broker producer.
//
// Pattern (copied from services/ocr-api/src/amqp.ts):
//   - One confirm channel.
//   - Reconnect on close (idempotent connect()).
//   - publish() awaits the confirm — fail-loud on broker hiccup.

import amqplib, { type Channel, type ChannelModel } from "amqplib";
import { randomUUID } from "node:crypto";

export interface EnvelopeV1<P = unknown> {
  version: 1; id: string; ts: number; attempt: number; payload: P;
}

const URL_REDACT = /:\/\/[^:]+:[^@]+@/;

export interface ProducerOpts {
  url:        string;   // amqp://federation:****@node-eleven:5672/federation
  exchange:   string;   // "federation.work"
  routingKey: string;   // queue name, e.g. "ocr.jobs" / "fraud.rows"
}

export class AmqpProducer {
  private conn: ChannelModel | null = null;
  private channel: Channel | null = null;
  constructor(private opts: ProducerOpts) {}

  redactedUrl(): string { return this.opts.url.replace(URL_REDACT, "://***:***@"); }
  isReady():    boolean { return !!this.channel; }

  async connect(): Promise<void> {
    if (this.channel) return;
    const heartbeat = Number(process.env.AMQP_HEARTBEAT_SEC ?? 600);
    const conn    = await amqplib.connect(this.opts.url, { heartbeat });
    const channel = await conn.createConfirmChannel();
    conn.on("close", () => { this.conn = null; this.channel = null; });
    conn.on("error", (e) => console.error(`[amqp.producer] ${e.message}`));
    this.conn = conn; this.channel = channel;
    console.log(`[amqp.producer] connected exchange=${this.opts.exchange} rk=${this.opts.routingKey} heartbeat=${heartbeat}s`);
  }

  async publish<P>(payload: P, attempt = 1): Promise<EnvelopeV1<P>> {
    if (!this.channel) await this.connect();
    const env: EnvelopeV1<P> = {
      version: 1,
      id:      randomUUID(),
      ts:      Math.floor(Date.now() / 1000),
      attempt,
      payload,
    };
    const buf = Buffer.from(JSON.stringify(env));
    await new Promise<void>((resolve, reject) => {
      this.channel!.publish(
        this.opts.exchange,
        this.opts.routingKey,
        buf,
        { contentType: "application/json", deliveryMode: 2, messageId: env.id, correlationId: env.id },
        (err) => err ? reject(err) : resolve(),
      );
    });
    return env;
  }

  async close(): Promise<void> {
    try { await this.channel?.close(); } catch {}
    try { await this.conn?.close(); }    catch {}
    this.channel = null; this.conn = null;
  }
}

Consumer skeleton

// consumer.ts — generic federation broker consumer.
//
// Pattern (copied from services/ocr-worker/src/amqp.ts):
//   - One channel, prefetch=1, manual ack.
//   - Heartbeat 600s — non-negotiable.
//   - safeAck/safeNack swallow IllegalOperationError if the channel
//     closed between handler-await and ack call.
//   - Transient failure → republish with attempt+1, ack original.
//     This keeps retries off the DLX so the broker UI shows queue-depth
//     fluctuation, not DLQ growth.
//   - attempt > MAX_ATTEMPTS → nack(requeue=false), broker DLX-routes
//     to federation.dlx → <domain>.dlq for human inspection.

import amqplib, { type Channel, type ChannelModel, type ConsumeMessage } from "amqplib";

export interface EnvelopeV1<P = unknown> {
  version: 1; id: string; ts: number; attempt: number; payload: P;
}

const URL_REDACT = /:\/\/[^:]+:[^@]+@/;

export interface ConsumerOpts {
  url:         string;
  exchange:    string;   // "federation.work"
  queue:       string;   // "ocr.jobs" / "fraud.rows" / ...
  routingKey:  string;   // == queue name
  prefetch?:   number;   // default 1
}

export type HandlerResult = "ack" | "retry" | "dlq";

export class AmqpConsumer {
  private conn: ChannelModel | null = null;
  private channel: Channel | null = null;

  constructor(private opts: ConsumerOpts) {}

  redactedUrl(): string { return this.opts.url.replace(URL_REDACT, "://***:***@"); }

  async connect(): Promise<void> {
    if (this.channel) return;
    const heartbeat = Number(process.env.AMQP_HEARTBEAT_SEC ?? 600);
    const conn    = await amqplib.connect(this.opts.url, { heartbeat });
    const channel = await conn.createConfirmChannel();
    await channel.prefetch(this.opts.prefetch ?? 1);
    await channel.checkQueue(this.opts.queue);   // existence check only — do NOT redeclare
    conn.on("close", () => { this.conn = null; this.channel = null; });
    conn.on("error", (e) => console.error(`[amqp.consumer] ${e.message}`));
    this.conn = conn; this.channel = channel;
    console.log(`[amqp.consumer] connected queue=${this.opts.queue} prefetch=${this.opts.prefetch ?? 1} heartbeat=${heartbeat}s`);
  }

  private safeAck(msg: ConsumeMessage):  void { try { this.channel?.ack(msg); }              catch (e) { console.warn(`[amqp] safe-ack: ${(e as Error).message}`); } }
  private safeNack(msg: ConsumeMessage, requeue: boolean): void {
    try { this.channel?.nack(msg, false, requeue); } catch (e) { console.warn(`[amqp] safe-nack: ${(e as Error).message}`); }
  }

  async consume<P>(handler: (env: EnvelopeV1<P>) => Promise<HandlerResult>): Promise<void> {
    if (!this.channel) await this.connect();
    await this.channel!.consume(this.opts.queue, async (msg) => {
      if (!msg) return;
      let env: EnvelopeV1<P>;
      try {
        env = JSON.parse(msg.content.toString("utf-8"));
      } catch (e) {
        console.error(`[consumer] bad JSON, sending to DLQ: ${(e as Error).message}`);
        return this.safeNack(msg, false);
      }
      if (env.version !== 1) {
        console.error(`[consumer] unknown envelope.version=${(env as any).version}, sending to DLQ`);
        return this.safeNack(msg, false);
      }
      try {
        const result = await handler(env);
        if (result === "ack")   return this.safeAck(msg);
        if (result === "dlq")   return this.safeNack(msg, false);
        if (result === "retry") {
          // Republish with attempt+1, then ack the original.
          // Caller's handler must NOT have done a partial side-effect
          // it can't repeat (e.g. don't half-write to SQLite then retry).
          // …producer.publish(env.payload, env.attempt + 1)…
          return this.safeAck(msg);
        }
      } catch (e) {
        console.error(`[consumer] handler threw: ${(e as Error).message}`);
        return this.safeNack(msg, false);
      }
    }, { noAck: false });
  }

  async close(): Promise<void> {
    try { await this.channel?.close(); } catch {}
    try { await this.conn?.close(); }    catch {}
    this.channel = null; this.conn = null;
  }
}

Handler skeleton (worker side)

const consumer = new AmqpConsumer({
  url:        process.env.AMQP_URL!,
  exchange:   "federation.work",
  queue:      "fraud.rows",
  routingKey: "fraud.rows",
});
const producer = new AmqpProducer({
  url:        process.env.AMQP_URL!,
  exchange:   "federation.work",
  routingKey: "fraud.rows",
});

const MAX_ATTEMPTS = Number(process.env.MAX_ATTEMPTS ?? 5);

await consumer.consume<FraudRowPayload>(async (env) => {
  const { id, attempt, payload } = env;
  try {
    // 1. Do the work (may take minutes — heartbeat 600s covers it).
    await enrichFraudRow(payload);

    // 2. Persist outcome via the writer.
    //    (Same shape as services/ocr-worker/src/writer-client.ts)
    await writer.finishFraudRow(payload.corpus_id, payload.row_no, result);

    return "ack";
  } catch (e) {
    console.error(`[handler] ${id} attempt=${attempt} failed: ${(e as Error).message}`);
    if (attempt >= MAX_ATTEMPTS) {
      await writer.failFraudRow(payload.corpus_id, payload.row_no, (e as Error).message);
      return "dlq";
    }
    await producer.publish(payload, attempt + 1);  // transient retry
    return "retry";
  }
});

Operational notes

Heartbeat audit lesson

services/ocr-worker/src/amqp.ts had a heartbeat fix landed in commit 19eb545. The same bug was independently observed on the edgar-filing-ingester (Cane audit 2026-05-19, finding 2, /tmp/cane-audit-report.md). Any new consumer that omits the 600s heartbeat will hit this within hours of going to prod. It is not a "maybe", it is "when".

DLQ replay

When a message lands in <domain>.dlq:

1. Inspect via mgmt UI (http://node-eleven:15672/#/queues/federation/ocr.dlq).

2. If the failure was a fixable bug — fix the worker, redeploy, then shovel the DLQ back to its source queue:

   ssh rooot@node-eleven 'docker exec gpumon-rabbitmq rabbitmqadmin -V federation \
     get queue=ocr.dlq ackmode=ack_requeue_false count=100 -f raw_json' \
     | jq -r '.[].payload' | while read env; do
       # republish via your producer, attempt=1
       …
     done
   

3. If the failure was data-poison (malformed PDF, dead CIK), leave it in the DLQ. It is a forensic record.

Reconnect resilience

Both skeletons leave the close / error handlers no-op on the connection itself — the next connect() call re-establishes. Wrap your top-level driver in a while(true) with a 5s backoff:

for (;;) {
  try {
    await consumer.consume(handler);
    // consume() returns immediately; the handler callback drives the loop
    await new Promise(r => setTimeout(r, 60_000));  // keep-alive
  } catch (e) {
    console.error(`[driver] fatal: ${(e as Error).message}`);
    await consumer.close();
    await new Promise(r => setTimeout(r, 5000));
  }
}

Cross-references

• Reference producer impl: services/ocr-api/src/amqp.ts
• Reference consumer impl: services/ocr-worker/src/amqp.ts
• Writer client impl (the "persist outcome" side): services/ocr-worker/src/writer-client.ts
• Envelope spec: envelope.md
• Broker topology: broker.md
• Audit lesson (heartbeat bug): /tmp/cane-audit-report.md