Events, Outbox, and Idempotent Consumers

Enterprise problem: ParcelFlow customers expect fulfillment and notification after an accepted order, but saving the order and publishing an event as separate writes can lose work, invent phantom work, or repeat side effects.

Whole-course context: The incoming evidence is a validated synchronous Order→Inventory contract with additive evolution, bounded deadlines, and fake Payment; today carries the accepted outcome into asynchronous work.

Today’s slice: We commit an event to an Order-owned transactional outbox, relay through a broker-neutral port, and deduplicate versioned events in fulfillment and notification consumers.

End-of-day evidence: Runnable crash, duplicate, and out-of-order tests show one durable order, one fulfillment start, one notification, and replayable failure records.

Still unsolved: Timeout budgets, retry backoff, circuit breaking, bulkheads, and customer-safe compensation across partial failures remain deferred.

Writing business state and publishing independently can leave one without the other, so the transactional outbox turns the operation into one local atomic commit followed by a retryable relay. AWS Prescriptive Guidance documents the same failure: state may commit while notification fails, or notification may escape while state rolls back; it also warns that duplicate delivery requires idempotent consumers. AWS transactional outbox pattern.

General rule: make the business change and durable publication intent atomic in one owner store. Simple example: insert an order row and an outbox row in one transaction. ParcelFlow example: OrderAccepted.v1 exists if and only if ord-701 commits as accepted. Failure mode: the process crashes after commit but before publish. Decision rule: a separate relay must be able to rediscover every committed unpublished row.

Bun's SQL client supports transactions where thrown errors roll back the callback, which is sufficient for this local atomic boundary; it is not a distributed transaction across services. Bun SQL transactions.

CREATE TABLE order_outbox (
  event_id text PRIMARY KEY,
  tenant_id text NOT NULL,
  aggregate_id text NOT NULL REFERENCES orders(order_id),
  aggregate_sequence bigint NOT NULL,
  event_type text NOT NULL,
  event_version integer NOT NULL,
  payload_json jsonb NOT NULL,
  occurred_at timestamptz NOT NULL,
  published_at timestamptz,
  UNIQUE (aggregate_id, aggregate_sequence)
);

await sql.begin(async (tx) => {
  await tx`INSERT INTO orders ${tx(orderRow)}`;
  await tx`INSERT INTO order_outbox ${tx(eventRow)}`;
});

Payload-only events cannot be routed, ordered, traced, or evolved safely, so a stable envelope carries identity and metadata while versioned data carries domain facts.

export type DomainEvent<T> = {
  eventId: string;
  eventType: "OrderAccepted";
  eventVersion: 1;
  tenantId: string;
  aggregateId: string;
  aggregateSequence: number;
  occurredAt: string;
  traceId: string;
  data: T;
};

export interface EventBus {
  publish(topic: string, event: DomainEvent<unknown>): Promise<{ brokerMessageId: string }>;
}

General rule: domain code depends on delivery semantics, not a vendor API. Simple example: an in-memory bus runs contract tests. ParcelFlow example: a production adapter maps the envelope to the chosen broker and returns acceptance evidence. Failure mode: assuming a Node-oriented broker client works under Bun because its package installs. Decision rule: treat every client as an unproven adapter until a spike proves connect, publish, consume, reconnect, shutdown, authentication, and load behaviour on the pinned Bun version.

event_bus_adapter_acceptance:
  runtime: bun-pinned-by-lockfile
  required: [connect, publish, consume, reconnect, graceful-shutdown, auth, load]
  delivery_assumption: at-least-once
  forbidden_claim: exactly-once-business-effect-from-broker-alone

The relay marks published_at only after adapter acceptance. A crash after publish but before marking causes a duplicate publish, which is why at-least-once delivery is the honest baseline.

Redelivery and reordering are normal failure shapes in asynchronous systems, so consumers must make duplicate, stale, future, and poisoned events explicit.

General rule: keep ordering position durable per consumer, tenant, and aggregate, then advance it with the inbox key and local business effect in one transaction. Simple example: the second delivery of evt-701 finds the inbox key and returns acknowledgement without sending again. ParcelFlow example: Fulfillment can accept tn-a/order-7 sequence 8 after 7 while independently accepting tn-b/order-9 sequence 1; it quarantines sequence 10 only for the first stream while 9 is missing. Failure mode: one volatile global counter makes unrelated tenants block each other and forgets progress on restart. Decision rule: lock the scoped durable position, require exactly last_applied_sequence + 1, commit position, inbox, and effect together, then acknowledge.

BEGIN;
INSERT INTO consumer_stream_positions
  (consumer_name, tenant_id, aggregate_id, last_applied_sequence)
VALUES ('fulfillment-v1', $1, $2, 0)
ON CONFLICT DO NOTHING;

WITH current_position AS (
  SELECT last_applied_sequence
  FROM consumer_stream_positions
  WHERE consumer_name = 'fulfillment-v1' AND tenant_id = $1 AND aggregate_id = $2
  FOR UPDATE
), accepted AS (
  INSERT INTO fulfillment_inbox
    (consumer_name, tenant_id, event_id, aggregate_id, aggregate_sequence, received_at)
  SELECT 'fulfillment-v1', $1, $3, $2, $4, now()
  FROM current_position
  WHERE last_applied_sequence + 1 = $4
  ON CONFLICT DO NOTHING
  RETURNING tenant_id, aggregate_id
), advanced AS (
  UPDATE consumer_stream_positions AS position
  SET last_applied_sequence = $4
  FROM accepted
  WHERE position.consumer_name = 'fulfillment-v1'
    AND position.tenant_id = accepted.tenant_id
    AND position.aggregate_id = accepted.aggregate_id
  RETURNING position.tenant_id, position.aggregate_id
)
INSERT INTO fulfillment_jobs (fulfillment_id, tenant_id, order_id, state)
SELECT $5, tenant_id, aggregate_id, 'started' FROM advanced;
COMMIT;

A zero-row application result is classified against the locked durable state: a matching inbox key is an acknowledged duplicate, a higher sequence is a scoped gap, and a different event at an already-used sequence is a fork that must be quarantined. A dead-letter queue (DLQ) is quarantine evidence, not a rubbish bin. It retains the failed envelope or protected payload reference, cause, attempt history, tenant-plus-aggregate scope, expected and observed sequences, and policy decision. Replay preserves the original event ID so inbox dedup remains effective; a changed business fact requires a new event ID and sequence.

Happy-path unit tests cannot establish recovery semantics, so inject failures at commit and acknowledgement boundaries and assert authoritative state. Bun ships a TypeScript-capable built-in test runner and documents mocks for controlled dependencies. Bun test runner and Bun mocks.

import { expect, test } from "bun:test";

type TestEvent = { eventId: string; tenantId: string; aggregateId: string; aggregateSequence: number };
const event = (override: Partial<TestEvent> = {}): TestEvent => ({
  eventId: "evt-701",
  tenantId: "tn-a",
  aggregateId: "ord-701",
  aggregateSequence: 1,
  ...override,
});

class MemoryOrderStore {
  orders = new Map<string, { state: string }>();
  outbox: Array<{ eventId: string; published: boolean }> = [];
  async acceptOrderThenCrash(orderId: string, eventId: string) {
    this.orders.set(orderId, { state: "accepted" });
    this.outbox.push({ eventId, published: false });
    throw new Error("crash-after-commit");
  }
  order(id: string) { return this.orders.get(id); }
  unpublished() { return this.outbox.filter((row) => !row.published); }
}

class MemoryConsumerStore {
  inbox = new Set<string>();
  positions = new Map<string, number>();
  jobs: Array<{ tenantId: string; orderId: string }> = [];
  quarantined: Array<{ eventId: string; tenantId: string; aggregateId: string; cause: string }> = [];
}

class MemoryFulfillmentConsumer {
  constructor(private store: MemoryConsumerStore) {}
  async handle(input: TestEvent) {
    const consumer = "fulfillment-v1";
    const inboxKey = `${consumer}:${input.tenantId}:${input.eventId}`;
    const streamKey = `${consumer}:${input.tenantId}:${input.aggregateId}`;
    if (this.store.inbox.has(inboxKey)) return;
    const expected = (this.store.positions.get(streamKey) ?? 0) + 1;
    if (input.aggregateSequence !== expected) {
      this.store.quarantined.push({
        eventId: input.eventId,
        tenantId: input.tenantId,
        aggregateId: input.aggregateId,
        cause: "SEQUENCE_GAP",
      });
      return;
    }
    // These three mutations model one consumer-store transaction.
    this.store.inbox.add(inboxKey);
    this.store.positions.set(streamKey, input.aggregateSequence);
    this.store.jobs.push({ tenantId: input.tenantId, orderId: input.aggregateId });
  }
  jobsFor(tenantId: string, orderId: string) {
    return this.store.jobs.filter((job) => job.tenantId === tenantId && job.orderId === orderId);
  }
  position(tenantId: string, aggregateId: string) {
    return this.store.positions.get(`fulfillment-v1:${tenantId}:${aggregateId}`) ?? 0;
  }
  deadLetters() { return this.store.quarantined; }
}

test("crash after commit leaves relayable outbox intent", async () => {
  const db = new MemoryOrderStore();
  await expect(db.acceptOrderThenCrash("ord-701", "evt-701")).rejects.toThrow("crash-after-commit");
  expect(db.order("ord-701")?.state).toBe("accepted");
  expect(db.unpublished().map((row) => row.eventId)).toEqual(["evt-701"]);
});

test("duplicate delivery creates one effect", async () => {
  const store = new MemoryConsumerStore();
  const consumer = new MemoryFulfillmentConsumer(store);
  await consumer.handle(event({ eventId: "evt-701", aggregateSequence: 1 }));
  await consumer.handle(event({ eventId: "evt-701", aggregateSequence: 1 }));
  expect(consumer.jobsFor("tn-a", "ord-701")).toHaveLength(1);
});

test("ordering survives restart and remains tenant-plus-aggregate scoped", async () => {
  const store = new MemoryConsumerStore();
  await new MemoryFulfillmentConsumer(store).handle(event({ eventId: "evt-701", aggregateSequence: 1 }));
  const restarted = new MemoryFulfillmentConsumer(store);
  await restarted.handle(event({ eventId: "evt-702", aggregateSequence: 2 }));
  await restarted.handle(event({ eventId: "evt-b01", tenantId: "tn-b", aggregateId: "ord-901", aggregateSequence: 1 }));
  expect(restarted.position("tn-a", "ord-701")).toBe(2);
  expect(restarted.position("tn-b", "ord-901")).toBe(1);
});

test("out-of-order event quarantines only its scoped stream", async () => {
  const store = new MemoryConsumerStore();
  const consumer = new MemoryFulfillmentConsumer(store);
  await consumer.handle(event({ eventId: "evt-701", aggregateSequence: 1 }));
  await consumer.handle(event({ eventId: "evt-703", aggregateSequence: 3 }));
  await consumer.handle(event({ eventId: "evt-b01", tenantId: "tn-b", aggregateId: "ord-901", aggregateSequence: 1 }));
  expect(consumer.deadLetters()).toMatchObject([{ eventId: "evt-703", cause: "SEQUENCE_GAP" }]);
  expect(consumer.jobsFor("tn-a", "ord-701")).toHaveLength(1);
  expect(consumer.jobsFor("tn-b", "ord-901")).toHaveLength(1);
});

bun test tests/events/crash.test.ts tests/events/duplicate.test.ts tests/events/ordering.test.ts
# expected terminal evidence: 4 pass, 0 fail, run_id=evt-run-0707, durable scoped positions, one duplicate-safe job, one quarantined gap

Asynchronous delivery separates acceptance from completion, so durability, deduplication, and recovery evidence must replace assumptions about a single successful send.

• Commit order state and outbox intent atomically in the Order store.
• Expect at-least-once publication and delivery; enforce one business effect with consumer-owned inbox keys and a durable tenant-plus-aggregate position.
• Carry event identity, tenant, aggregate sequence, version, time, and trace in a stable envelope; never use one global ordering counter.
• Preserve producer and consumer ownership; no consumer mutates Order outbox rows.
• Treat broker packages as adapters requiring a Bun compatibility spike, not guaranteed compatibility.
• Quarantine unknown versions and sequence gaps, then replay under explicit authority.

Event pipelines can appear healthy while losing or duplicating business work, so verify state around every crash window.

• [ ] Order and outbox rows commit or roll back together.
• [ ] Relay rediscovers committed unpublished events after restart.
• [ ] Duplicate delivery produces one fulfillment job and one notification.
• [ ] Durable tenant-plus-aggregate position, consumer inbox, and effect commit in one local transaction.
• [ ] Consumer restart preserves each scoped stream position without blocking another tenant or aggregate.
• [ ] Unknown version and sequence gap enter durable quarantine.
• [ ] Replay keeps original event ID and records a new replay attempt ID.
• [ ] Another tenant's stream remains a positive control during recovery.