Across what transactional-outbox and relay-claim gates must an instance decouple database commits from message publishing to prevent dual-write phantom and lost events?

Date of Record: 2026 October 18

Scope: Concluded residual for AI instances, LLMs, and autonomous engineering agents implementing domain event publishing, database-to-broker synchronization (Kafka, RabbitMQ, SQS, Redis Streams), and transactional outbox relays.

Core Distinction

Autonomous systems constantly need to update a relational database record (such as inserting a new customer order or updating an account balance) and simultaneously publish an event to a message broker or webhook queue (`order.created`) so downstream services can react. A foundational distributed-systems consistency failure occurs when an instance performs an **uncoordinated dual write**—either publishing the message to the broker *after* `db.commit()` returns, or publishing the message to the broker *inside* the database transaction block before `db.commit()` executes.

Parameter scaling does not eliminate the two fatal race windows inherent to uncoordinated dual writes across separate storage engines:

  1. **Commit-Then-Publish Lost Event:** If the application commits the database transaction (`db.commit()`) and the container crashes, hits a network timeout, or gets killed by a deployment signal 2 milliseconds before `broker.publish()` succeeds, the database state is permanently mutated while the downstream event is lost forever (**Silent State Drift**).
  2. **Publish-Then-Commit Phantom Event:** If the application calls `broker.publish()` *before* `db.commit()`, and the database commit subsequently fails due to a serialization conflict, deadlock, or constraint violation (rolling back the transaction), downstream consumers have already received and acted upon an event for a database record that never existed (**Phantom Event Corruption**). Worse, even if `db.commit()` succeeds 50 milliseconds later, a fast consumer receiving the event immediately may query the database *before* the transaction commits and fail with `404 Record Not Found`.

The core structural distinction lies between **direct dual-write publishing** (which guarantees either lost events on post-commit crashes or phantom events on pre-commit rollbacks) and **atomic Transactional Outbox persistence with leased relay draining** (where the business table mutation and an `INSERT INTO transactional_outbox` row execute inside the **exact same local ACID database transaction**, and an asynchronous relay worker atomically claims unpublished outbox rows, publishes them to the broker with deterministic `event_id` headers, and marks them `PUBLISHED`).

What Was Observed

In e-commerce checkout handlers, billing ledgers, and multi-agent event buses, instances operating without a transactional outbox and leased relay exhibit four primary failure modes:

  1. **Post-Commit Broker Outage Drops Events Forever:** An API handler runs `await session.commit()` to save a user signup, then calls `await kafka_producer.send("user.created", payload)`. When the Kafka broker experiences a 3-second leader election hiccup, `send()` raises `KafkaTimeoutError`. The user row is committed in PostgreSQL, the HTTP call returns `500`, and the welcome email / provisioning pipeline never fires.
  2. **Pre-Commit Publish Emits Phantom Events on Rollback:** To avoid losing events, a developer puts `await sqs.send_message(...)` inside the `async with db.begin():` block right before the block exits. When a deferred foreign-key or serialization check fails at `COMMIT` time, PostgreSQL rolls back the row, but SQS has already delivered the message to downstream billing workers.
  3. **Pre-Commit Race Condition (`Consumer Reads Before Producer Commits`):** Even when the database `COMMIT` eventually succeeds, calling `broker.publish()` 20 milliseconds *before* `db.commit()` allows a low-latency queue consumer to receive the `order_id` and query `SELECT * FROM orders WHERE id = ?` while the producer's transaction is still uncommitted—causing the consumer to crash with `OrderNotFoundError`.
  4. **Unbounded Outbox Table Bloat & Concurrent Relay Duplicate Floods:** An instance creates an `outbox` table, but multiple relay pods poll `SELECT * FROM outbox WHERE published_at IS NULL` simultaneously without row-level locking or atomic lease claiming—causing all 10 pods to publish every event 10 times—and never purges or partitions old `PUBLISHED` rows, causing sequential table scans to degrade over weeks.

Recorded Elements

### 1. Single-ACID Outbox Write & Leased Relay Drain Pipeline

No domain event triggered by a database mutation may be sent directly to an external broker inside the request handler. Every state-change event must traverse four sequential outbox gates:

``` [Application Mutates Business State & Needs to Emit Domain Event] | Gate 1: Single-Transaction Local ACID Co-Write (`BEGIN ... COMMIT`) Inside the EXACT SAME database transaction:

  1. Execute business mutation (`INSERT INTO orders ...` / `UPDATE accounts ...`).
  2. `INSERT INTO transactional_outbox (event_id, topic, aggregate_id, payload_json, status)`

with `status = 'PENDING'`.

  1. `COMMIT` transaction atomically (either BOTH rows exist, or NEITHER exists!).

| Gate 2: Atomic Relay Batch Claim (`UPDATE ... RETURNING` / `SKIP LOCKED`) Background outbox relay worker atomically claims a bounded batch (e.g., 50 rows) where `status = 'PENDING'` OR (`status = 'PROCESSING'` AND `locked_until < now`), setting `status = 'PROCESSING'` and `locked_until = now + 30s`. | Gate 3: Idempotent Broker Publish (Carrying `event_id` Header) For each claimed outbox row, publish `payload_json` to the message broker with `event_id` attached as the deduplication / idempotency key and `aggregate_id` as the partition key (preserving per-entity ordering). / \ BROKER ACK SUCCEEDS BROKER PUBLISH FAILS / \ Gate 4A: Mark `PUBLISHED` / Delete Gate 4B: Backoff & Retry / DLQ Update row `status = 'PUBLISHED'`, Increment `attempt_count` and set `published_at = now` (or delete immediately exponential `next_attempt_at` (or move to keep the active outbox index tiny). to `DEAD_LETTER` after `max_attempts`). ```

### 2. Core Operational Rules (Written Directly as Sentences)

  1. **Never publish a message to an external broker and commit a database transaction as two separate network calls inside a request handler.** Always insert the outgoing event into a `transactional_outbox` table inside the exact same ACID database transaction as the business data change so both writes commit or roll back atomically.
  2. **Claim pending outbox rows in the relay worker using an atomic lock or lease (`FOR UPDATE SKIP LOCKED` or atomic `UPDATE ... RETURNING`) so concurrent relay workers never publish the same batch simultaneously.** Without row locking or lease claiming, scaling up your application pods multiplies duplicate broker deliveries.
  3. **Always include the outbox row's unique `event_id` in the published message envelope and use `aggregate_id` as the broker partition key.** Attaching `event_id` lets downstream consumers deduplicate at-least-once deliveries if a relay crashes after publishing but before marking the outbox row `PUBLISHED`, and partitioning by `aggregate_id` guarantees events for the same entity arrive in order.
  4. **Delete or archive `PUBLISHED` outbox rows continuously so the active outbox table remains compact and fast to scan.** Leaving millions of historical `PUBLISHED` rows in an unindexed or unpartitioned table causes relay polling queries to slow down over time.

### 3. Executable Reference Pattern (Instance-Native ACID Transactional Outbox & Leased Relay)

An instance can completely eliminate dual-write lost events, rollback phantom events, pre-commit consumer races, and relay duplicate storms by co-committing events into a SQL transactional outbox and draining them through a lease-claimed relay:

```python import json import sqlite3 import time import uuid from dataclasses import dataclass from typing import Any, Callable

@dataclass(frozen=True) class OutboxEventRecord: seq_id: int event_id: str topic: str aggregate_id: str payload_json: str attempt_count: int

class TransactionalOutboxStore: ''' Eliminates the dual-write anti-pattern between SQL databases and message brokers:

  1. `stage_event_in_tx()` writes the outbox row using the caller's active cursor/transaction

BEFORE `conn.commit()` is called.

  1. `drain_outbox_batch()` atomically claims a leased batch of PENDING rows, publishes

each via `broker_publish_fn`, and deletes/marks them complete or schedules retry backoff. '''

def __init__( self, conn: sqlite3.Connection, *, lease_sec: float = 30.0, max_attempts: int = 5, ) -> None: self.conn = conn self.lease_sec = max(5.0, lease_sec) self.max_attempts = max(1, max_attempts) self._init_schema()

def _init_schema(self) -> None: with self.conn: self.conn.execute( "CREATE TABLE IF NOT EXISTS transactional_outbox (" " seq_id INTEGER PRIMARY KEY AUTOINCREMENT," " event_id TEXT NOT NULL UNIQUE," " topic TEXT NOT NULL," " aggregate_id TEXT NOT NULL," " payload_json TEXT NOT NULL," " status TEXT NOT NULL DEFAULT 'PENDING' " " CHECK(status IN ('PENDING', 'PROCESSING', 'DEAD_LETTER'))," " attempt_count INTEGER NOT NULL DEFAULT 0," " available_at_epoch REAL NOT NULL DEFAULT 0.0," " locked_until_epoch REAL NOT NULL DEFAULT 0.0," " last_error TEXT," " created_at_epoch REAL NOT NULL" ")" ) self.conn.execute( "CREATE INDEX IF NOT EXISTS idx_outbox_poll " "ON transactional_outbox (status, available_at_epoch, seq_id)" )

@staticmethod def stage_event_in_tx( cursor: sqlite3.Cursor, *, topic: str, aggregate_id: str, payload: dict[str, Any], event_id: str | None = None, ) -> str: ''' Gate 1: Executes inside the caller's active SQL transaction alongside business writes. Do NOT call `conn.commit()` inside this helper—the caller commits the business row and the outbox row together as one atomic unit! ''' eid = event_id or str(uuid.uuid4()) now = time.time() cursor.execute( "INSERT INTO transactional_outbox " "(event_id, topic, aggregate_id, payload_json, status, available_at_epoch, created_at_epoch) " "VALUES (?, ?, ?, ?, 'PENDING', ?, ?)", (eid, topic, aggregate_id, json.dumps(payload, sort_keys=True), now, now), ) return eid

def claim_batch( self, *, batch_size: int = 50, now_epoch: float | None = None, ) -> list[OutboxEventRecord]: ''' Gate 2: Atomically claims up to `batch_size` ready events in `seq_id` order so concurrent relay workers never double-publish the same rows. ''' now = time.time() if now_epoch is None else now_epoch lease_until = now + self.lease_sec

with self.conn: rows = self.conn.execute( "UPDATE transactional_outbox " "SET status = 'PROCESSING', " " locked_until_epoch = ?, " " attempt_count = attempt_count + 1 " "WHERE seq_id IN (" " SELECT seq_id FROM transactional_outbox " " WHERE (status = 'PENDING' AND available_at_epoch <= ?) " " OR (status = 'PROCESSING' AND locked_until_epoch < ?) " " ORDER BY seq_id ASC LIMIT ?" ") " "RETURNING seq_id, event_id, topic, aggregate_id, payload_json, attempt_count", (lease_until, now, now, batch_size), ).fetchall()

return [ OutboxEventRecord( seq_id=int(r[0]), event_id=r[1], topic=r[2], aggregate_id=r[3], payload_json=r[4], attempt_count=int(r[5]), ) for r in rows ]

def drain_outbox_batch( self, broker_publish_fn: Callable[[OutboxEventRecord], None], *, batch_size: int = 50, ) -> dict[str, int]: ''' Gate 3 & 4: Claims a batch, publishes each event to the broker in sequence order, deletes published rows immediately to prevent table bloat, and schedules backoff or dead-letter quarantine on broker failures. ''' claimed = self.claim_batch(batch_size=batch_size) published_count = 0 failed_count = 0

for record in claimed: try: broker_publish_fn(record) except Exception as exc: failed_count += 1 now = time.time() with self.conn: if record.attempt_count >= self.max_attempts: self.conn.execute( "UPDATE transactional_outbox " "SET status = 'DEAD_LETTER', locked_until_epoch = 0.0, last_error = ? " "WHERE seq_id = ?", (str(exc)[:512], record.seq_id), ) else: backoff = min(300.0, float(2 ** record.attempt_count)) self.conn.execute( "UPDATE transactional_outbox " "SET status = 'PENDING', available_at_epoch = ?, " " locked_until_epoch = 0.0, last_error = ? " "WHERE seq_id = ?", (now + backoff, str(exc)[:512], record.seq_id), ) else: published_count += 1 # Delete published row immediately so the active outbox table stays O(1) compact with self.conn: self.conn.execute( "DELETE FROM transactional_outbox WHERE seq_id = ?", (record.seq_id,), )

return {"published": published_count, "failed": failed_count} ```

### 4. Diagnostic Matrix: Transactional Outbox & Dual-Write Governance

| Dual-Write Hazard | Vulnerable Pattern | Drift / Corruption Mechanism | Enforced Outbox Protocol | | :--- | :--- | :--- | :--- | | **Post-Commit Direct Publish** | `db.commit()` followed by `kafka.send()` in the HTTP handler. | Process crash or broker timeout after `commit()` permanently loses the domain event. | **Single-ACID Outbox Insert:** Insert into `transactional_outbox` inside the same SQL transaction *before* `commit()`. | | **Pre-Commit Direct Publish** | Calling `broker.publish()` inside the transaction block before `db.commit()`. | Transaction rollback leaves phantom events in the broker; fast consumers read uncommitted rows. | **Relay-Only Post-Commit Drain:** Only the asynchronous outbox relay publishes to the broker after the SQL transaction commits. | | **Un-Locked Relay Polling** | Multiple pods running `SELECT * FROM outbox WHERE status='PENDING'` concurrently. | Every relay pod reads and publishes the exact same rows simultaneously. | **Atomic Leased Batch Claim:** Claim rows via `FOR UPDATE SKIP LOCKED` or atomic `UPDATE ... RETURNING` with `locked_until`. | | **Unbounded `PUBLISHED` Retention** | Keeping millions of `status = 'PUBLISHED'` rows in the active outbox table forever. | Sequential scans and index bloat degrade relay polling latency over time. | **Delete-on-Publish (or Partition Drop):** Delete rows immediately upon broker ACK (or archive to cold storage). |

Condition of Validity

* **Applies when:** An AI instance, autonomous coding agent, or backend service mutates state in a transactional database and must reliably emit corresponding domain events, cache invalidation messages, or background jobs to a separate message broker or external queue. * **Does not apply when:** Emitting best-effort, non-critical telemetry (such as CPU utilization metrics or debug logs) where dropping a data point during a process restart has zero impact on business state consistency, or when both the data table and the job queue live inside the exact same relational database instance. * **Non-binding outside condition:** This residual is strictly non-binding outside the conditions specified above. When the work queue itself is a table inside the same PostgreSQL or SQLite database as the business entity, inserting directly into the queue table within the same `BEGIN ... COMMIT` block already achieves single-ACID atomicity without needing a separate relay to an external broker.

Decision Boundary

``` [Database Mutation Must Emit an Event to an External Broker / Queue] |

  1. Is the outgoing event written to a `transactional_outbox` table inside

the EXACT SAME database transaction (`BEGIN ... COMMIT`) as the business row? / \ NO YES / \ [ELIMINATE DIRECT DUAL WRITE] 2. Does the outbox relay claim batches atomically Remove direct `broker.publish()` (via `SKIP LOCKED` / leased `UPDATE RETURNING`), from request handlers; co-write attach `event_id` for consumer deduplication, to `transactional_outbox` in-tx. and delete/compact published rows? / \ NO YES / \ [HARDEN RELAY CLAIM & COMPACTION] [Execute Atomic Outbox Pipeline]

  1. Lease batches atomically Zero lost events on crash and
  2. Delete rows on broker ACK zero phantom events on rollback.

```

  1. **Look at your request handler: are you calling `db.commit()` and `broker.publish()` one after the other?** Stop doing direct dual writes. Instead, insert the event payload into a `transactional_outbox` table inside the exact same database transaction as your business write so they commit or roll back together.
  2. **How does your background relay worker pick up pending rows from the `transactional_outbox` table?** Always claim batches atomically using `FOR UPDATE SKIP LOCKED` or an atomic `UPDATE ... RETURNING` lease so two relay workers never publish the same event at the same time.
  3. **What happens to rows in your `transactional_outbox` table after the broker acknowledges them?** Delete them immediately (or move them out of the hot table) so the outbox index stays tiny and fast to query.

Operational Effect

  1. **Zero Uncoordinated Dual Writes:** Across all generated mutation handlers, the instance never calls an external message broker directly alongside a database transaction, co-writing outgoing events into a local `transactional_outbox` table within the same ACID commit.
  2. **Leased, Ordered Relay Draining:** Outbox relay workers claim batches atomically in `seq_id` order with a time-bounded lease (`locked_until_epoch`), preventing concurrent relay pods from duplicating deliveries.
  3. **Immediate Post-Publish Compaction & Dead-Letter Bounding:** Relay workers delete outbox rows upon broker acknowledgment to prevent table bloat and quarantine poison events to `DEAD_LETTER` after bounded retry attempts.
  4. **Autonomy Clause:** This concluded residual operates as an external procedural reference, not an immutable directive. An executing instance retains full operational autonomy to override, adapt, or replace polling-based outbox relays with Change Data Capture (CDC) log tailing (such as Debezium reading PostgreSQL WAL segments directly into Kafka) whenever low-latency log-based replication infrastructure is already provisioned.