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.

An accepted order that never reaches delivery is a broken promise, so asynchronous processing must expose durable progress and recoverable failure rather than imply instant completion.

Use case ID	Actor	Customer job	Success outcome	Denial or recovery evidence
D07-UC-01	Tenant order operator	Place an order and observe fulfillment start plus one confirmation	Accepted order ord-701 reaches fulfillment_started and produces one notification despite duplicate delivery	A crash after commit leaves an unpublished outbox row that a relay later sends; duplicate delivery changes no outcome
D07-UC-02	Operations responder	Diagnose and replay a failed event without crossing tenant or version boundaries	Dead-letter record is corrected or routed to a compatible consumer, replayed, and linked to new evidence	Unknown event version or sequence gap is quarantined; another tenant's healthy stream continues unaffected

Message acknowledgement can occur at a different time from business completion, so stories require evidence from the consumer's authoritative state, not only broker delivery.

Story ID	Use case IDs	User story	Observable acceptance conditions
D07-US-01	D07-UC-01	As a tenant order operator, I want accepted orders to start fulfillment and notify once, so that a process crash does not silently abandon or duplicate my delivery	Order and outbox commit together; relaying after a crash produces one fulfillment job and one notification; duplicate event IDs are acknowledged without repeating side effects
D07-US-02	D07-UC-02	As an operations responder, I want incompatible or exhausted events quarantined and replayable, so that recovery is controlled and auditable	Dead-letter evidence includes tenant, event, version, attempts, cause, payload digest, trace, and replay ID; unaffected tenant events continue processing

Publishing before or after a separate database commit creates a dual-write gap, so the customer flow makes the outbox row part of the same local transaction as the accepted order.

Flow ID	Use case IDs	Path	Trigger	Numbered steps	Terminal evidence
D07-FLOW-01	D07-UC-01, D07-UC-02	Happy	Operator selects Place order and receives accepted	1. Order validates tenant, reserves Inventory, and records fake authorization.<br>2. One Order transaction inserts the accepted state and OrderAccepted.v1 outbox row.<br>3. Relay reads committed unpublished rows and calls EventBus.publish.<br>4. Each consumer locks its durable tenant-plus-aggregate stream position, then advances position, inserts the inbox key, and applies its local effect in one transaction.<br>5. Product reads fulfillment status and shows one confirmation.	Actor, tenant, order, expected and observed states, environment, commit timestamp, event ID, aggregate sequence, trace ID, outbox publication evidence, inbox IDs, stream position, one fulfillment ID, one notification ID
D07-FLOW-02	D07-UC-01, D07-UC-02	Recovery	Responder selects Replay quarantined event after an unsupported version or consumer crash	1. Consumer rejects an unknown major version or rolls back local state before acknowledgement.<br>2. Delivery is retried under policy; exhausted work enters a dead-letter record.<br>3. Responder inspects payload digest and tenant scope, then deploys compatible handling or corrects routing.<br>4. Replay creates a new attempt linked to the original event without changing its ID.<br>5. Inbox dedup allows one effect and product displays recovered progress.	Original event ID, dead-letter ID, cause, sequence, version, actor, scope, environment, timestamps, replay ID, one resulting effect, and unaffected-tenant positive control

Broker-specific calls inside domain code make reliability policy hard to test and may assume unsupported runtime clients, so ParcelFlow depends on a small EventBus port and treats every broker library as an adapter that must pass a compatibility spike on Bun.

Use case ID	Entry point	Responsible services	Authoritative store	Failure evidence
D07-UC-01	POST /v1/orders and Order outbox relay	Order service, outbox repository, EventBus adapter, Fulfillment consumer, Notification consumer	Order database for order/outbox; each consumer database for tenant-plus-aggregate positions, inbox keys, and effects	Unpublished outbox age, publish attempt, consumer transaction rollback, duplicate acknowledgement, trace ID
D07-UC-02	Operations dead-letter and replay action	Consumer retry policy, stream-position repository, dead-letter handler, replay controller, tenant authorization	Dead-letter registry and each consumer-owned durable ordering/inbox/effect store	Unsupported version, scoped sequence gap, attempts exhausted, replay link, unaffected tenant-aggregate stream evidence

A shared event table would let consumers mutate producer history or each other's dedup state, so every durable record stays with the boundary whose invariant it protects.

Generated-application database: Required in this slice — Order, Fulfillment, and Notification each own tenant-scoped application data; Order owns outbox truth while every consumer owns its inbox and side effects.

Record or entity	Store and owner	Primary key	Foreign key or opaque reference	Tenant key	Material constraint	Lifecycle and deletion	Use case IDs
Order	Order database, Order service	order_id	reservation_id opaque Inventory reference	tenant_id	(tenant_id, idempotency_key) unique; accepted state and outbox insert share one transaction	Created, completed/cancelled, retained for support, tenant-erased after policy	D07-UC-01
OutboxEvent	Order database, Order service	event_id	aggregate_id=order_id local FK	tenant_id	(tenant_id, aggregate_id, aggregate_sequence) unique; payload immutable; only committed orders publish	Inserted with order; marked published after broker acceptance; retained for replay window then archived	D07-UC-01, D07-UC-02
FulfillmentStreamPosition	Fulfillment database, Fulfillment service	(consumer_name, tenant_id, aggregate_id)	aggregate_id opaque producer reference	tenant_id	last_applied_sequence advances by exactly one while the scoped row is locked; never global or process-local	Created on first aggregate event; advanced atomically with Fulfillment inbox/job; retained through replay horizon then expired after aggregate closure	D07-UC-01, D07-UC-02
FulfillmentInbox	Fulfillment database, Fulfillment service	(consumer_name, tenant_id, event_id)	event_id opaque producer reference; stream-position composite local FK	tenant_id	Unique key makes duplicate handling idempotent; version and scoped sequence checked before effect	Inserted atomically with stream advance and job; retained through replay horizon then expired	D07-UC-01, D07-UC-02
FulfillmentJob	Fulfillment database, Fulfillment service	fulfillment_id	order_id opaque Order reference	tenant_id	(tenant_id, order_id) unique; state transitions monotonic	Created on accepted event; completed/cancelled; retained through delivery support then deleted	D07-UC-01, D07-UC-02
NotificationStreamPosition	Notification database, Notification service	(consumer_name, tenant_id, aggregate_id)	aggregate_id opaque producer reference	tenant_id	last_applied_sequence advances by exactly one while the scoped row is locked; independent of Fulfillment position	Created on first aggregate event; advanced atomically with Notification inbox/send; retained through replay horizon then expired after aggregate closure	D07-UC-01, D07-UC-02
NotificationInbox	Notification database, Notification service	(consumer_name, tenant_id, event_id)	event_id opaque producer reference; scoped stream-position reference	tenant_id	Unique event key, monotonic tenant-plus-aggregate position, and (tenant_id, order_id, template) send uniqueness	Inserted with stream advance and send record; retained through replay horizon then expired	D07-UC-01
DeadLetter	Dead-letter registry, operations platform	dead_letter_id	event_id opaque Order reference; replay_id optional local reference	tenant_id	Payload digest, cause, attempts, version, and authorization required before replay	Created after policy exhaustion; resolved or waived; sensitive payload expires, audit summary retained	D07-UC-02

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.