Messaging Patterns

The Dual-Write Problem

The most common reliability mistake: writing to a database AND publishing an event in two separate operations.

// WRONG — dual write
async function createOrder(order: Order) {
	await db.insert('orders', order); // succeeds
	await kafka.publish('orders', order); // crash here → event never sent
	// OR:
	await kafka.publish('orders', order); // succeeds
	await db.insert('orders', order); // crash here → event sent but no DB record
}

If the process crashes between the two operations, you have an inconsistency: DB and message broker are out of sync.

Outbox Pattern

Write the event to the database in the same transaction as the business data. A separate process reads undelivered events and publishes them.

-- outbox table
CREATE TABLE outbox (
  id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  topic       TEXT NOT NULL,
  key         TEXT,
  payload     JSONB NOT NULL,
  created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  published_at TIMESTAMPTZ   -- NULL until delivered
);

// Atomic: business write + event write in one transaction
async function createOrder(order: Order) {
	await db.transaction(async (tx) => {
		await tx.query('INSERT INTO orders (id, customer_id, total) VALUES ($1, $2, $3)', [
			order.id,
			order.customerId,
			order.total
		]);

		await tx.query('INSERT INTO outbox (topic, key, payload) VALUES ($1, $2, $3)', [
			'orders',
			order.id,
			JSON.stringify({ event: 'order.created', ...order })
		]);
	});
}

// Outbox publisher — runs separately, polls for undelivered events
async function publishOutbox() {
	while (true) {
		const rows = await db.query(
			`SELECT * FROM outbox
       WHERE published_at IS NULL
       ORDER BY created_at
       LIMIT 100
       FOR UPDATE SKIP LOCKED`
		);

		for (const row of rows.rows) {
			await kafka.publish(row.topic, { key: row.key, value: row.payload });

			await db.query('UPDATE outbox SET published_at = NOW() WHERE id = $1', [row.id]);
		}

		await sleep(1000);
	}
}

FOR UPDATE SKIP LOCKED lets multiple publisher instances run without duplicate publishing — each row is claimed by one publisher.

Cleanup: delete published rows after a retention window:

DELETE FROM outbox WHERE published_at < NOW() - INTERVAL '7 days';

Use Debezium for the publisher instead of polling — CDC watches the Postgres WAL and publishes outbox rows to Kafka automatically (zero polling delay).

Inbox Pattern

Prevent duplicate processing when a consumer receives the same message twice (at-least-once delivery):

CREATE TABLE inbox (
  message_id  TEXT PRIMARY KEY,
  topic       TEXT NOT NULL,
  processed_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

async function handleOrder(msg: KafkaMessage) {
	const messageId = `${msg.topic}-${msg.partition}-${msg.offset}`;

	await db.transaction(async (tx) => {
		// Check if already processed
		const result = await tx.query(
			'INSERT INTO inbox (message_id, topic) VALUES ($1, $2) ON CONFLICT DO NOTHING RETURNING message_id',
			[messageId, msg.topic]
		);

		// No rows returned = conflict = already processed
		if (result.rows.length === 0) return;

		const order = JSON.parse(msg.value!.toString());
		await processOrder(tx, order);
	});
}

The ON CONFLICT DO NOTHING combined with RETURNING makes the duplicate check atomic. No separate SELECT needed.

Choreography

Services react to events from other services — no central coordinator.

OrderService publishes order.created
  → PaymentService (subscribes) charges card, publishes payment.completed
    → FulfillmentService (subscribes) ships order, publishes order.shipped
      → NotificationService (subscribes) sends email

Pros: loose coupling, no SPOF coordinator, easy to add new services.

Cons: hard to trace a saga across services, hard to answer “what’s the current state of order 123?”, failure recovery requires each service to handle compensating events.

// Each service is autonomous
class PaymentService {
	async onOrderCreated(event: OrderCreated) {
		try {
			const payment = await chargeCard(event.customerId, event.total);
			await publish('payment.completed', { orderId: event.orderId, paymentId: payment.id });
		} catch {
			await publish('payment.failed', { orderId: event.orderId, reason: 'card_declined' });
		}
	}

	async onOrderCancelled(event: OrderCancelled) {
		// Compensation: refund if payment was taken
		await refundPayment(event.orderId);
	}
}

Orchestration (Saga)

A central coordinator (the saga) tracks the state of a distributed transaction and directs each step.

// Saga state machine
interface OrderSagaState {
	orderId: string;
	step: 'payment' | 'inventory' | 'fulfillment' | 'completed' | 'failed';
	paymentId?: string;
	compensations: Array<() => Promise<void>>;
}

class OrderSaga {
	async execute(order: Order): Promise<void> {
		const state: OrderSagaState = {
			orderId: order.id,
			step: 'payment',
			compensations: []
		};

		try {
			// Step 1: Payment
			const payment = await paymentClient.charge(order);
			state.paymentId = payment.id;
			state.compensations.push(() => paymentClient.refund(payment.id));

			// Step 2: Reserve inventory
			await inventoryClient.reserve(order.items);
			state.compensations.push(() => inventoryClient.release(order.items));

			// Step 3: Fulfill
			await fulfillmentClient.ship(order);
			state.step = 'completed';
		} catch (err) {
			state.step = 'failed';
			// Run compensations in reverse order
			for (const compensate of state.compensations.reverse()) {
				await compensate().catch(console.error); // best-effort
			}
			throw err;
		}
	}
}

Pros: clear state, easy to reason about, one place to handle failures.

Cons: the saga coordinator is a SPOF (mitigated by persisting state), tighter coupling to step order.

For durable sagas (survive process restart), persist state to a database:

CREATE TABLE sagas (
  id         UUID PRIMARY KEY,
  type       TEXT NOT NULL,
  state      JSONB NOT NULL,
  status     TEXT NOT NULL DEFAULT 'running',  -- running, completed, failed
  created_at TIMESTAMPTZ DEFAULT NOW(),
  updated_at TIMESTAMPTZ DEFAULT NOW()
);

Temporal (temporal.io) is a purpose-built durable workflow engine that makes saga implementation with automatic replay, retries, and state persistence trivial.

Competing Consumers

Scale message processing by running multiple worker instances against the same queue:

Queue: [msg1, msg2, msg3, msg4, msg5]
  Worker 1 processes: msg1, msg3, msg5
  Worker 2 processes: msg2, msg4

Works automatically with:

• RabbitMQ: multiple consumers on the same queue
• Kafka: multiple consumers in the same consumer group (up to partition count)
• NATS JetStream: multiple pull consumers on same durable

The key invariant: each message processed by exactly one worker. Guaranteed by the broker’s locking semantics.

Fan-Out

One event consumed by multiple independent services:

Per-consumer queues (RabbitMQ):

// Exchange with one binding per service
await ch.assertExchange('orders', 'topic', { durable: true });

// Each service gets its own queue
await ch.assertQueue('orders.payment', { durable: true });
await ch.assertQueue('orders.analytics', { durable: true });
await ch.assertQueue('orders.notifications', { durable: true });

await ch.bindQueue('orders.payment', 'orders', 'created');
await ch.bindQueue('orders.analytics', 'orders', 'created');
await ch.bindQueue('orders.notifications', 'orders', 'created');

Kafka: multiple consumer groups automatically achieve fan-out. Each group reads all messages independently.

// payment-service group — reads all messages
const paymentConsumer = kafka.consumer({ groupId: 'payment-service' });

// analytics-service group — reads same messages independently
const analyticsConsumer = kafka.consumer({ groupId: 'analytics-service' });

Poison Pills

A message that always causes consumer failure, blocking the queue.

Detection:

ch.consume('orders', async (msg) => {
	const attempt = (msg.properties.headers['x-attempt'] || 0) as number;

	try {
		await processOrder(JSON.parse(msg.content.toString()));
		ch.ack(msg);
	} catch (err) {
		if (attempt >= 3) {
			// Poison pill — move to DLQ with diagnostic headers
			ch.publish('orders.dlx', 'created', msg.content, {
				headers: {
					...msg.properties.headers,
					'x-failed-reason': err.message,
					'x-failed-at': new Date().toISOString()
				}
			});
			ch.ack(msg);
		} else {
			// Retry
			ch.publish('orders', 'created', msg.content, {
				headers: { 'x-attempt': attempt + 1 }
			});
			ch.ack(msg);
		}
	}
});

Always have a DLQ. A queue without a DLQ eventually blocks on a poison pill indefinitely.