Change Streams & the Transactional Outbox

Part 0. Start here

The problem: one change, four systems

An order service handles 500 order changes a second (our example's number). After each commit it writes the search index, then the cache, then asks the notifier to email the customer: four writes to four systems, one after another. Deploys, scale-downs and crashes kill a request between its commit and its last write. Suppose that happens to 1 request in 100,000 (an assumption, to have a number):

• 500 changes a second × 86,400 seconds = 43,200,000 changes a day
• 43,200,000 ÷ 100,000 = 432 orders a day whose index, cache or email silently disagrees with the database

Nothing anywhere records those 432. The database says SHIPPED, the search page says PAID, and no log line, alarm or queue knows the two differ.

The team "fixes" it by feeding the index from the table's change stream (DynamoDB Streams, the table's own log of changes) with a Lambda function on default settings. A month later one malformed record makes the function throw. Lambda retries the batch, in order, and everything behind it on that shard waits. AWS's own documentation says that with the default settings "a bad record can block processing on the affected shard for up to one day". A DynamoDB stream keeps records for only 24 hours, and the records queued behind the bad one are only a little younger than it, so many of them expire before the function can catch up. The first fix lost 432 changes a day; the second can lose a day's worth in one go.

The big picture

Synthesizing vector architecture diagram...

What to notice: the only write the service makes is one transaction. A mover copies what committed into a topic, and every consumer decides for itself, by version, whether an event is new, old or early.

What you'll be able to do after this page

1. Explain the three ways writing to two systems goes wrong, and when a direct write is still allowed (Part 1).
2. Write the outbox transaction and say what it guarantees and what it doesn't (Part 2).
3. Walk through a polling relay, explain why its delivery is at least once, and apply events by version (Part 3).
4. Keep per-order order across every hop, including resizes, rebalances and multi-event transactions (Part 4).
5. Explain log-based change data capture on PostgreSQL and MySQL, and trace one change through every layer (Part 5).
6. Handle a record that always fails without blocking everyone behind it, and repair from the source of truth (Part 6).
7. Refill a gap after a lost position, and bootstrap a consumer that was down longer than retention (Part 7).
8. Change an event's schema safely, and make deletes stick (Part 8).
9. Run the same design on DynamoDB Streams, Lambda, Pipes and Kinesis Data Streams for DynamoDB (Part 9).
10. Choose between dual writes, an outbox, change data capture, event sourcing and two-phase commit on equal terms (Part 10).
11. Map all of it to AWS services, and name the look-alikes that are not change streams (Part 11).

You may have arrived from a step that relies on this: step 1.1 of the transactional outbox loop (the order committed but the publish timed out), step 2.4 of the email loop (a parser writing DynamoDB and the index), step R1.6 of the ride-sharing loop (a status hint fed from the stream, not a second write), or step 3.1 of the ad-click aggregation loop (note the stream position, then scan). This page is the "why" behind all four.

---

Part 1. Why writing twice fails

At 09:05:00, before our service had an outbox, order o7 was paid. The service committed version 2, updated the search index, and then the pod was killed by a deploy. The cache still says PLACED, the customer never gets "payment received", and nothing records that anything went wrong. This Part shows why no amount of care in the request fixes that.

Trace: the pod dies after the index write

A dual write means the application writes two (or more) systems one after the other, with nothing tying them together. Here is the counterfactual timeline B1 to B3, before the outbox existed. Every trace on this page was produced by running a private reference implementation of the example, not worked out by hand.

Synthesizing vector architecture diagram...

What to notice: the database and the index moved on; the cache and the notifier never heard. Nothing in any of the four systems says so.

A minute later two app servers change o7 at almost the same moment. Server A commits version 3 (a new address) at 09:06:00.000; server B commits version 4 (SHIPPED) at .010, after A's commit released the row lock. B happens to be quicker on the way out:

Synthesizing vector architecture diagram...

What to notice: the database serialized the two commits, but nothing serialized the two index writes. The slower request overwrote the newer document. The cache writes crossed the same way (B at .040, A at .050), so the cache also ends at version 3.

Three ways two writes disagree

Snapshot S1: four systems, four answers

Synthesizing vector architecture diagram...

After B3: the database says SHIPPED v4; the index and the cache say PAID v3; one email is missing. Each copy is wrong in its own way, and none of them knows.

One write, then followers

The mental model for the rest of the page: the change and its announcement are one write; everything else is a follower of that write. A follower reads what committed, in order, and may read it more than once. It never needs the writer to be alive.

The one allowed dual write. A direct write is fine as a latency hint, when two things are true: (a) a change stream delivers every change anyway, and (b) both paths apply by version or monotonically (for example "set if larger"), so the stream heals whatever the hint missed or reordered. The leaderboard loop (steps 1.3 and 2.6) writes scores directly for speed and re-applies from the stream using the current stored value; the Drive loop (step 2.3) sends a fast-path nudge and keeps the stream as the reliable path. A direct write with no stream behind it is a dual write, however fast.

Three rules the design never breaks

The example we follow

Real systems carry millions of orders, which is too many to watch. So one story runs through the whole page: follow orders o7, o8 and o9, with o7 as the thread from placement to erasure. An aggregate is the unit whose changes must stay in order (here, one order); its version goes up by one with every change.

The story in nine beats: the four-writes failure (this Part), one transaction (Part 2), the relay dies after publishing (Part 3), two relays, a resize and a rebalance (Part 4), the log becomes the relay (Part 5), a poison record (Part 6), the slot is lost and the index is down 30 hours (Part 7), a new field and an erasure (Part 8), and the same story on DynamoDB (Part 9). Each Part shows only its own slice of events; the full event table is in Part 12.

This answers the first question of the drill The Dual-Write That Broke Search Consistency: when the commit succeeds and the publish fails, the database and every copy drift apart silently, and no retry in the request can close that window; the fix is to make the event part of the commit. Its second question (log tailing versus polling every 500 ms) is answered in Part 5.

---

Part 2. The outbox: one transaction

The announcement has to commit with the change or not at all. The simplest place to put it is the same database, in the same transaction, as a row in a table built for the purpose.

One transaction

Event 4 at 10:01:00.0: o7 is paid. Here is the whole transaction:

sqlCopy
BEGIN;

UPDATE orders
   SET status = 'PAID', version = version + 1
 WHERE order_id = 'o7'
RETURNING version;                 -- 2, and o7's row lock is held until COMMIT

INSERT INTO outbox (event_id, aggregate_id, aggregate_version, event_type,
                    schema_version, payload, writer_epoch, bucket)
VALUES (gen_random_uuid(), 'o7', 2, 'order.paid', 1,
        '{"order_id": "o7", "status": "PAID", "total": 100}',
        (SELECT epoch FROM writer_epoch), 3);    -- seq 102 comes from the identity

COMMIT;

The table:

sqlCopy
CREATE TABLE outbox (
  seq               bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  event_id          uuid        NOT NULL UNIQUE,
  aggregate_id      text        NOT NULL,
  aggregate_version bigint      NOT NULL,
  event_type        text        NOT NULL,
  schema_version    int         NOT NULL,
  payload           jsonb       NOT NULL,
  writer_epoch      int         NOT NULL,
  bucket            smallint    NOT NULL,
  created_at        timestamptz NOT NULL DEFAULT now()
);

The table and the envelope

What it guarantees, and what it doesn't

So the outbox guarantees exactly one thing: the event exists if and only if the change committed. It does not publish anything, does not keep order across different orders, and does not stop a consumer from seeing the event twice. The outbox is a promise to publish, not a publish.

Events you mean, not rows that changed

An outbox row is an event the producer means: order.paid, with a payload designed for consumers and a version. The alternative, reading raw row changes from the business tables (Part 5), gives consumers the table's columns as they are today; a column rename becomes a breaking change for every consumer. Part 5 compares the two on equal terms.

The same idea works on a phone: the mobile chat loop (step 1.1) writes a message and its SENDING row in one local transaction, and a sender drains them.

---

Part 3. The relay and at-least-once delivery

The outbox holds rows; the index, the cache and the notifier read a topic. Something must move rows from one to the other, and that something can die at the worst moment. At 10:01:00.6 it does.

Trace: the relay dies after publishing

The first minute went smoothly. At 10:00:00.0 o7 was placed (outbox seq 101); at 10:00:00.6 relay R1 claimed bucket 3, the only bucket with rows, published 101 to P1@0 (partition 1, offset 0), got the acknowledgement, deleted the row and committed; at 10:00:00.7 all three consumers applied o7@1, the customer got "order received", and the feed got c1#1.

At 10:01:00.0 o7 is paid (seq 102) and at 10:01:00.1 o8 is placed (seq 103). From the reference implementation:

Synthesizing vector architecture diagram...

What to notice: the topic now holds each of the two events twice (P0@0 and P0@1, P1@1 and P1@2). Nothing was lost, and nothing downstream changed twice, because every consumer compared versions.

Snapshot S2 (after event 6, before R1 restarts):

The topic already has what the outbox still holds. When R1 comes back, it can't know that, so it publishes both rows again.

The relay loop

textCopy
RELAY LOOP (every 1 s, and again at once while work was found)
  BEGIN
    buckets = up to 4 bucket rows that have outbox rows,
              claimed with FOR UPDATE SKIP LOCKED          -- the claim: page 03, Part 9
    if none: COMMIT, sleep until the next tick or a signal
    for each claimed bucket b:
        rows = SELECT * FROM outbox WHERE bucket = b ORDER BY seq
        send each row to orders.events, key = aggregate_id
    wait for every acknowledgement      -- timeout: ROLLBACK, back off, try again
    DELETE FROM outbox WHERE seq IN (the published seqs)
  COMMIT

The relay holds a transaction open while it publishes, which is normally a bad idea; here it locks only bucket rows and outbox rows, never an order, and the publish has a hard timeout. The claim mechanism itself (SKIP LOCKED: take rows nobody holds, skip the rest) belongs to Leases, Fencing Tokens & Distributed Locks (Part 9).

Consumers apply by version

Duplicates are certain. "The broker removes duplicates" isn't true across a relay crash (the idempotent producer only removes its own retries within one session), and count = count + 1 turns every duplicate into a wrong number. The rule every consumer on this page follows:

textCopy
ON EVENT (key K, version v)                  -- where last_version is stored: page 04
  last = last_version[K]
  if v <= last:        skip                   -- a duplicate or an old event
  elif v == last + 1:  in ONE transaction: apply the effect, set last_version[K] = v
                       then apply parked events for K while each is exactly the next
  else:                park it under K        -- something before it is missing; alarm if it stays
  commit the offset only after the effect     -- our example commits after each batch
  event types this consumer ignores still advance last_version[K]

Effects are set by version, never added: "status is PAID as of version 2", not "one more payment". An email can't be un-sent, so the notifier also stores an idempotency key per email (o7:2); where that key lives and for how long is page 04's (Idempotency & Effectively-Once Processing).

A new owner replays the old owner's work

Events 10 to 12 (Part 4) publish o7 v3 and v4 to P1@4 and P1@5. The notifier runs as two instances: N1 reads P1, N2 reads P0. At 10:03:00.7, N1 is killed halfway through v4:

The feed is a derived numbered record: entry n exists only after n − 1. A naive new owner that takes "the next free number" would write c1#4 for the same event, and the customer's feed would show "shipped" twice. The src check tells "my own replay" apart from "someone else's entry":

textCopy
APPEND DERIVED ENTRY (customer c, event e)
  n = last_seq[c] + 1
  write entry c#n with src = e.event_id, only if c#n doesn't exist
  if it existed:
      if entry(c#n).src == e.event_id: last_seq[c] = n      -- my own replay: finish the bookkeeping
      else:                                                 -- someone else took n
          check I still own the partition (group generation or lease: page 03); if not, stop
          re-read last_seq[c]; write at the first free number after n; never retry n
  else: last_seq[c] = n
  last_seq never expires

Snapshot S3 (after event 14):

The cursor trap: numbers are not commit order

A cheaper relay design keeps no claims and deletes nothing: it remembers "published up to seq N" and reads rows above N. Event 9 shows why that fails. Two transactions start at 10:02:00: T1 places o9, T2 pays o8.

Synthesizing vector architecture diagram...

What to notice: seq is handed out when a row is inserted, not when it commits. The cursor moved past 104 while 104 was still invisible, and no later read looks below the cursor.

o9's "order placed" is never published, silently. Our claim-and-delete relay doesn't have this problem: it has no cursor, and any row still in the table is unpublished by definition, so at 10:02:00.6 it published both (105 to P0@2, 104 to P1@3). The other fixes are to read in commit order, which is what a log reader does (Part 5), or to read only up to the oldest transaction still in flight. The payment loop (step 3.6) has its sealer read the change stream in commit order for exactly this reason: "entry IDs can commit out of order, so batching by ID could skip a late-committing entry".

A related rule for the relay loop: keep claiming until a claim comes back empty, and treat an empty claim as "nothing visible right now", not as "done forever". A row that commits a moment later is still in the table at the next tick.

Waking the relay

When the broker is down

The outbox absorbs a broker outage: rows simply wait. The outbox loop (step 2.4) sizes it: MSK is down for 10 minutes while 6,000 events a second keep arriving, so 6,000 × 600 = 3,600,000 rows pile up. When MSK returns, the relays drain at most 20,000 events a second (a cap, because every claim, read and delete runs on the same writer as live transactions: a shared limit), while 6,000 a second keep arriving:

drain time = backlog ÷ (drain rate − arrival rate) = 3,600,000 ÷ (20,000 − 6,000) = 3,600,000 ÷ 14,000 ≈ 257 s, about 4.3 minutes.

Each bucket drains oldest first, so the oldest events are published within seconds of the broker's return and the newest backlog events wait the longest.

Delete churn. Every event is an insert and then a delete: at 6,000 events a second that is 6,000 dead rows a second for vacuum to clean. If anything holds vacuum back, the relay's claim and read queries slow down. The outbox loop (steps 1.6 and 2.5) handles it with short transactions and partitioned outboxes whose empty partitions are dropped; that detail stays there.

---

Part 4. Order per aggregate

Consumers need one order's events in version order: "shipped" before "address changed" makes the cache wrong and the customer confused. Order is not a property of Kafka or of the database; it holds only if every hop keeps it. This Part breaks it one hop at a time.

Trace: two relays split one order

At 10:02:30 a second relay, R2, starts beside R1. At 10:03:00.0 o7's address changes (v3, seq 106) and at 10:03:00.2 it ships (v4, seq 107). Suppose the relays claimed rows instead of buckets: at 10:03:00.6 R1 claims row 106 and R2 claims row 107, and R2 happens to publish first. From the reference implementation:

The version check saved the indexer; the cache without it is now as wrong as the dual write in Part 1. The real design never lets this happen in the first place: relays claim whole buckets, so one order's rows have one publisher at a time. In the main line, R1 claimed bucket 3 at 10:03:00.6, R2's claim skipped it and found no other bucket with rows, and R1 published 106 to P1@4 then 107 to P1@5.

Every hop must keep order

Synthesizing vector architecture diagram...

What to notice: the first four links keep order; the last one detects when any of them didn't. Remove any link and order can break; remove the check and nobody notices.

The cost: parallelism is capped by the bucket count and the partition count, and a very busy order delays the other orders in its bucket.

Gaps are checked number by number

A consumer that reads a batch must check every version, not the first and last of the batch. The mobile news feed loop (step 2.4) states it for feed numbers: every number from last + 1 onward must be present, with no hole anywhere in the batch.

textCopy
APPLY BATCH (records of one partition, in offset order)
  for each record r (key K, version v):
      if K is blocked:          park r; continue          -- Part 6
      if v <= last_version[K]:  skip; continue
      if v != last_version[K] + 1:
                                park r; continue          -- a hole before v
      apply r and set last_version[K] = v in one transaction
      while parked[K] holds last_version[K] + 1:
          apply it and store its version                  -- the hole just closed
  commit the partition's offset after the batch

Adding partitions

More throughput needs more partitions. The trap: the partition is chosen when a record is produced. Our example grows orders.events from 2 to 4 partitions while R1 is publishing o7 v3 and v4 (a side trace; the main line keeps 2 partitions). The example forces R1's producer to refresh its metadata between the two sends; real producers pick up new partitions at their next metadata refresh (in Kafka, metadata.max.age.ms, 5 minutes by default, or sooner after an error).

Synthesizing vector architecture diagram...

What to notice: one order's events now live in two partitions with no order between them. The newer event arrived 1.9 s before the older one, and only the version check put them back in order.

Rebalances move partitions between consumers of a group, and the new owner replays from the last committed offset: event 14 in Part 3 was one. The version check and the src check make the replay harmless. DynamoDB Streams and Kinesis shards also split, and there a child shard must be read only after its parent (Part 9).

One transaction, several events

A transfer in the wallet loop (step 2.2) debits one account and credits another in one transaction, writing two outbox rows keyed by the two accounts. The keys hash to different partitions, so a consumer can see the debit without the credit for a while, even though they committed together.

The rule: choose the aggregate as the smallest scope whose events must be seen together. The outbox loop (step 1.5) keys ledger events by account; the payment loop (step 3.6) reads the change stream in commit order so a transaction's entries are never split by a batch boundary.

Epochs: when the writer itself changes

After a failover promotes a new database, the version numbers of an order can repeat for different events (the lost tail of the old writer). The outbox loop (step 3.4) copies a writer_epoch into every event and has consumers compare (epoch, version):

How the epoch is bumped inside the promoted database before writes open is in Leases, Fencing Tokens & Distributed Locks (Part 7).

---

Part 5. Reading the database's own log

By 11:00 the relays work, but look at what they cost at scale: every relay runs a claim query every second against the busiest database in the company, every event is an insert and a delete for vacuum to clean, every event waits up to a poll interval, and every team builds its own relay. The outbox loop (step 3.2) hits this at 2 billion events a day. The database already writes every committed change to its write-ahead log (WAL) for durability; a reader can follow that log instead of querying tables. That is log-based change data capture (CDC).

Why read the log

This answers the second question of the drill The Dual-Write That Broke Search Consistency: polling every 500 ms adds a query and table churn to the busiest database forever, and up to half a second of delay; tailing the WAL costs almost no queries and keeps commit order, and in exchange you run a connector and watch a slot. At one service and modest rates the relay is the better deal; at many services or high rates, CDC is.

The outbox table or the business tables?

CDC can read the outbox table (the same events as before, now moved by the log) or the business tables themselves.

From 11:00 our story tails the outbox table: consumers keep the same events and versions, and only the mover changes.

PostgreSQL: the replication slot

Logical decoding needs wal_level = logical. A publication lists the tables to capture:

sqlCopy
CREATE PUBLICATION outbox_pub FOR TABLE outbox, cdc_heartbeat;

A logical replication slot records how far its reader has confirmed. The server keeps every WAL segment from that position on, whether or not the reader is connected. The Write-Ahead Log page (Part 8) gives the general rule: a log segment may go only when its data is durable elsewhere and no reader still needs it, so the slowest reader sets the log's size. This page owns what happens next.

Synthesizing vector architecture diagram...

What to notice: the connector moves the slot forward only after the broker acknowledged. Everything between the slot's position and now stays on the writer's disk, up to the cap, if one is set.

To watch it:

sqlCopy
SELECT slot_name, active, wal_status,
       pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn) AS retained_bytes
  FROM pg_replication_slots;

On Aurora and RDS the same numbers arrive as CloudWatch metrics: OldestReplicationSlotLag, ReplicationSlotDiskUsage and TransactionLogsDiskUsage.

When the slot stalls

A stalled reader keeps WAL growing at the database's WAL rate:

retained WAL = WAL rate × stall time

At real scale, from the outbox loop: about 2 KB of WAL per event × 12,000 events a second at peak (its CDC round) = 24,000 KB/s = 24 MB/s. A cap of 200 GB then lasts 200,000 MB ÷ 24 MB/s ≈ 8,333 s ≈ 2.3 hours of stall at peak. In our tiny example we simply set the cap so it is reached 6 hours into a stall.

Two shared limits decide where the cap goes:

• Every holder counts. Retained WAL runs from the oldest position anything needs: every logical slot, every physical replica's slot, wal_keep_size, an archiver that is behind. The slowest one decides.
• The tables share the disk. They keep growing through the stall. Set the cap below the free disk minus the tables' growth over the longest stall you plan to ride out; the outbox loop pairs a 200 GB cap with at least 400 GB free.

Before you set a cap, plan the republish (Part 7): past the cap, the writer is safe and the reader has a guaranteed gap. Check that your engine version's parameter group exposes max_slot_wal_keep_size (it appears in Aurora PostgreSQL's parameter list; the outbox loop checks per engine version).

MySQL: the binlog is purged by time (text only)

MySQL's change log is the binary log in row format (binlog_format=ROW, the default). A reader stores its own position (a GTID set, or a file and an offset). The server doesn't keep binlogs for a reader: they expire by age (binlog_expire_logs_seconds, 30 days by default; on RDS for MySQL the binlog retention hours setting, which is NULL by default, meaning binlogs are not retained, with a maximum of 168 hours). The same stall ends the opposite way:

Other databases expose the same idea under other names (SQL Server's CDC, Oracle's redo mining, MongoDB change streams); no loop on this site uses them.

Cutting over from the relay (event 19)

The order of the cut-over matters: open the new reader before the old one stops.

Synthesizing vector architecture diagram...

What to notice: from 11:00:00 every commit is covered by the slot. Had the relays stopped first, a row committed between their stop and the slot's creation would reach neither path; with the slot first, a row committed during the overlap is published by both, a harmless duplicate.

no_data means the connector takes no snapshot and streams from the slot's creation point; the relays had already confirmed every earlier row, so nothing is missing. From now on the outbox is insert-only: the connector never deletes, and old rows go by dropping daily partitions after 3 days. Keeping them is what makes Part 7's republish possible.

End to end through the layers

Event 20 (o8 v3, seq 111) through every layer:

Synthesizing vector architecture diagram...

What to notice: two positions are saved, the slot's (step 4) and the consumer's offset (step 7), and each is saved after the work it covers. That is why a crash repeats work and never skips it.

Gaps come only from losing a position: a slot invalidated or dropped, a failover to a primary without the slot, a connector restarted from the wrong offset, a consumer offset reset to "latest", or records trimmed by retention. Part 7 handles each.

Lag is a sum of the dependent hops: commit, decode, publish and acknowledge, the consumer's poll, the apply. In the tiny example each hop takes 0.1 s: o8 v3 commits at 11:10:00.0, is published at 11:10:00.1 and applied at 11:10:00.2. At real scale, measure it end to end with the heartbeat.

Each hop is at least once; exactly-once per hop, and Kafka transactions, are covered by the Idempotency & Effectively-Once Processing loop primitive (page 04).

---

Part 6. Poison records and repair

At 10:04:30.0 a producer bug stores an address for o9 that the indexer's parser can't handle. The event is valid as far as the outbox knows, it is published like any other, and the indexer fails on it every time. A record that always fails is called a poison record. Retrying it forever blocks everything behind it on the partition; catching the error and moving on loses the change silently. This Part does neither.

Trace: o9's bad address

At 10:04:30.0 o9 v2 order.paid (seq 108) commits with the bad address; at 10:04:30.2 o7 v5 (seq 109) commits. R1 publishes both at 10:04:30.6, to P1@6 and P1@7, so o7 v5 sits right behind the poison.

Synthesizing vector architecture diagram...

What to notice: the partition waited 6.1 s (three retries, after 0.1 s, 1 s and 5 s), not forever. After that only o9 is stopped, for this consumer; o7 flows past it.

Only the indexer's parser chokes on the address, so only the indexer dead-letters. The error boundary is per consumer.

Snapshot S4 (after event 17):

The error boundary

Engine-neutral, per consumer:

1. Validate the record against its schema before doing anything with it.
2. Retry transient failures a few times with backoff (0.1 s, 1 s, 5 s here). A schema violation isn't transient and can skip the retries.
3. Dead-letter it: in one transaction in the consumer's own database, store the record and the reason, mark the key blocked for this consumer, and commit the offset. Recording it in the same database as the offset decision means a crash can't lose it.
4. Park the key's later events behind it, in version order, instead of applying them.
5. Move on: every other key on the partition keeps flowing.
6. Alarm on every dead letter; each needs an owner.

textCopy
REPAIR (consumer, key K)                      -- after the cause is fixed
  row = read K from the PRIMARY               -- never a lagging replica (event 28, Part 8)
  in one transaction in the consumer's database:
      write the effect from row's current state if row.version > last_version[K]
      last_version[K] = row.version
      mark K's dead letters resolved ("superseded by row.version")
      unblock K
  drop parked events for K with version <= row.version
  apply any parked event that is now exactly the next

Replays and a quarantine list. A rebuild that replays raw events from an archive would feed o9 v2 to the new code and fail again, or worse, succeed on a payload that was rejected on purpose. So replays skip a quarantine list of events resolved as invalid (the outbox loop, step 3.3), or they rebuild from the source of truth instead (Part 7).

The periodic compare. Every stream-fed copy also gets a job that compares it with the source and writes repairs: the stream is how changes normally arrive; the compare catches what slipped through anyway (a bug, a lost position nobody noticed). Its reads must come from the primary, or from a replica known to be caught up; Part 8's event 28 shows the compare job going wrong with a lagging replica.

Stop it at the producer. The best poison record is one that never commits. Validate each event against its registered schema and the size limit before the outbox insert, so a bad event fails the business transaction instead of reaching the stream (the outbox loop, step 2.3).

Managed stream readers have their own knobs for all of this (Lambda's and Pipes' bisect, maximum record age and on-failure destination); they are in Part 9.

---

Part 7. Refilling what the stream can't give you: republish and bootstrap

At 18:00:00 slot S1 is invalidated. Two committed events, seq 112 and 113, will never come through it. No consumer can ask for them, because as far as the stream is concerned they never happened. This Part is about gaps: where they come from, how to notice them, and how to fill them in the right order.

Trace: the slot is invalidated

Snapshot S5 (after event 23):

Where gaps come from

Detecting a gap

textCopy
GAP CHECK (every minute)
  alarm if heartbeat lag > 60 s for 5 minutes
  alarm if any consumer has held a parked event longer than a threshold
  t = newest seq the topic holds from this database        -- read the topic first
  o = newest committed seq in the outbox                    -- then the outbox
  if o > t:
      wait briefly, then read both again once               -- an in-flight publish is not a gap
      if o is still > t and the connector's position isn't moving: declare a gap

The re-check matters: an empty or short read that races a writer isn't evidence of anything. The mobile news feed loop (step 2.4) re-checks an empty read the same way before declaring a gap.

Republish: refill from the stored source, then resume

textCopy
REPUBLISH (after a lost position)
  1. create the new slot FIRST                  -- every later commit is covered from here
  2. newest = the newest event the topic holds from this database
  3. from = newest.committed_at - margin        -- can reach back only as far as rows are kept
  4. publish every outbox row committed at or after `from`, in seq order, key = aggregate_id
  5. wait for every acknowledgement
  6. start the connector on the new slot        -- it re-sends what committed after step 1
  7. consumers skip what they already have

Event 24, from the reference implementation:

Synthesizing vector architecture diagram...

What to notice: the newest event the topic held was seq 111 (committed 11:10:00); a 10-minute margin reaches back to 11:00:00, the cut-over, so rows 111 to 114 are republished. Rows before 11:00 were deleted by the relay and aren't needed: the relay confirmed every one of them. And 114 comes twice, once from the republish and once from the connector.

That margin works only because the outbox has been insert-only since the cut-over. A relay that deletes on publish keeps nothing to republish from, and rows are kept only 3 days: the outbox's retention must be longer than your worst time to detect and repair a gap.

Snapshot S6 (after event 24):

Trace: the index is down for 30 hours

A consumer that was away longer than retention can never catch up from the stream alone. It needs a snapshot of the source plus the stream, joined at the right point:

Synthesizing vector architecture diagram...

What to notice: in the right order, every change after the noted position is in the stream the rebuild applies, so v6 is seen twice and applied once. In the wrong order, v6 committed after the scan read o8 and before the position was noted, so it is in neither.

Snapshot S7 (after event 26):

A replica is a valid snapshot source only once its replay position is at or past the noted stream position. A replica that lags reads the past, and the stream from the noted position won't contain what it's missing. The same mistake returns in event 28 (Part 8), where the compare job reads a replica 40 s behind. The ad-click loop (step 3.1) and the mobile news feed loop (step 2.3: build the head after reading last_seq) follow the same order.

The tools the source gives you

Debezium's PostgreSQL connector has snapshot modes that decide what it does when it starts:

when_needed quietly re-snapshots after a lost position: every row the outbox still holds is read again, the same effect as a republish but unannounced. Know which mode you run before an incident, not during one.

Shadow rebuild and switch. When the live copy must keep serving while it is rebuilt, build a shadow copy (a new index) by the same procedure, let it catch up to the live position, then switch readers to it and drop the old one. The outbox loop (step 3.3) rebuilds a projection this way, skipping its quarantine list.

---

Part 8. Schema changes and deletes

Events outlive the code that wrote them. At 10:04:30.2 o7 v5 arrives with schema_version 2 and a new field, currency, and two of the three consumers are still running last week's code. Later, o7 is erased, and a repair job that doesn't know it tries to bring it back. Both are the same problem: what a consumer does with an event it didn't expect.

Trace: currencymeets old code

o7 v5 order.items_updated (seq 109) adds an optional field, currency. Event 17, per consumer:

Nothing broke, because adding an optional field is a safe change. Most changes are not.

Compatibility in plain words

The tools: a schema registry keeps each event type's versions and a compatibility mode; CI rejects an incompatible change before it ships, and producers can publish only registered versions. With change data capture on business tables (Part 5), the table is the schema, so a column rename is a contract change for every consumer.

textCopy
DECODE (event)                                      -- fail closed
  reader = the registered reader for (event_type, schema_version)
  if none:  dead-letter "unknown schema version"; block the key
  fields = reader.decode(payload)
  if a required field is missing or has the wrong type:
            dead-letter with the reason; block the key    -- never default to 0
  ignore unknown optional fields

A delete must travel

A deleted order leaves nothing behind to publish. "Consumers will notice it's gone" is wrong: a consumer sees events, not absences. So a delete travels as an event with a version, like any change:

Trace: o7is erased, and a repair comes too late

Snapshot S8 (after event 28):

Caches store "gone". Deleting order:o7 would leave room for a late refill to put v5 back; storing GONE@6 means refills and repairs, which write with SET … NX (only if absent), can't. Compacted topics (Kafka's cleanup.policy=compact, which keeps the latest value per key) keep a delete's tombstone for delete.retention.ms, one day by default, then drop it: a consumer away longer than that never sees the delete and must rebuild (Part 7). Erasure events carry no personal data: order.erased holds the id and the version, nothing else, so the log isn't where the data lives on.

---

Part 9. The same story on DynamoDB

This Part replays events 15 to 17 and 25 to 26, plus one shard split, on DynamoDB. The replay keeps the same orders and versions but its own times.

DynamoDB Streams in one table

Trace: the poison batch, bisected

The table's stream uses NEW_AND_OLD_IMAGES. The Lambda mapping reads with batch size 4, bisect on (split a failing batch in two), 2 retries, a maximum record age of 3,600 s, and an SQS on-failure destination. Shard sh-1 holds four changes in a row (stream sequence numbers 100 to 103): o8 v2 (10:04:29.5), o9 v2 with the bad address (10:04:30.0), o7 v5 (10:04:30.2) and o8 v3 (10:04:30.4). The function applies each record by version and throws on the bad address. From the reference implementation (our model: split on each error; a single failing record is retried twice, then its metadata goes to the destination; AWS documents that splitting doesn't consume the retry quota):

Synthesizing vector architecture diagram...

What to notice: 7 invocations for 4 records. o8 v2 was delivered 3 times and o9 v2 5 times; the version check made every repeat harmless. The shard stopped for one bad record's retries, not for a day.

The destination receives metadata, not the record. For SQS and SNS destinations Lambda sends only a description of the failed batch; an S3 destination also gets the records. Trimmed to the fields that matter:

jsonCopy
{
  "requestContext": { "condition": "RetryAttemptsExhausted" },
  "DDBStreamBatchInfo": {
    "shardId": "sh-1",
    "startSequenceNumber": "101",
    "endSequenceNumber": "101",
    "batchSize": 1
  }
}

So the repair worker re-reads the item from the table, never trusts the message to hold it. At 10:04:40 it reads o9: the table itself still holds the bad address (v2), because the bug was in the producer, so the repair fails and stays open with an alarm. At 10:20:00 the producer's fix writes v3; the stream delivers it and the function parks it (its last o9 is 1, a gap at 2). At 10:20:30 the repair re-reads o9 (v3), writes the index and sets the version to 3; the parked v3 is skipped. The same repair as Part 6.

Trace: a shard splits

At 10:30 shard sh-1 is closed and split into two children. From the reference implementation:

Synthesizing vector architecture diagram...

What to notice: o7 v5 is in sh-1, and anything after it will be in sh-3. Reading sh-3 before sh-1 is finished could apply a newer change of o7 or o9 before an older one. DynamoDB's rule is that a parent is processed before its children; the DynamoDB Streams Kinesis Adapter for KCL follows it for you, and any reader you write must too.

The same rule has a sharp edge. With Lambda's default settings (no bisect, retry until the record expires), sh-1 would have been stuck on o9 v2 until D2 10:04:30, and a reader that keeps the parent-first rule would hold sh-2 and sh-3 back just as long. One bad record would have delayed every later change to these orders by up to a day.

Lambda and Pipes settings

Two more facts belong with these. Lambda polls each shard 4 times a second and delivers at least once. Its concurrency for a stream is open shards × ParallelizationFactor, drawn from the account's Regional pool (1,000 concurrent executions by default), shared with every other function; throttling counts as an error before invocation and is retried until the record expires or passes its maximum age. Alarm on the mapping's IteratorAge metric (in the Lambda namespace), the Lambda-side equivalent of heartbeat lag. (Lambda can also aggregate over tumbling windows; windows belong to the Event Time, Watermarks & Checkpoints loop primitive.)

Rebuilding inside 24 hours

The 30-hour outage on DynamoDB, from the reference implementation:

Both writers carry whole items (each stream record has the new image, each export row is the item), so "only if newer by version" is enough. Had we loaded the export first and started the stream afterwards, the stream position would have to be one from before the export, and if the export and load took 24 hours or more, that position would be gone. AWS states this limit for OpenSearch Ingestion's DynamoDB source: "If ingestion from an initial snapshot of a large table takes 24 hours or more, there will be some initial data loss."

Kinesis Data Streams for DynamoDB

DynamoDB can also send the same changes into a Kinesis data stream. It looks like "the same thing, kept longer", and it isn't quite:

From the reference implementation, o7 v3 and v4 arrive as v4, v3, v4:

The maps loop (step 2.6) relies on exactly this: its session index is fed from Kinesis Data Streams for DynamoDB, and route_version absorbs the duplicates and reordering.

Global tables

With global tables, each Region's replica has its own stream, and that stream carries every change, including the ones replicated in from other Regions. A consumer in every Region would therefore send every email once per Region. Side effects must act in one Region only (the item's owning Region), while each Region may still build its own read models from its own stream, as the leaderboard loop (step 3.4) does.

Do you need an outbox on DynamoDB?

Usually not. The stream is the table's commit log, kept by DynamoDB, with nothing to poll. A separate event item written in the same transaction costs more: a transactional write takes 2 write units per item of up to 1 KB, so a 1 KB update that costs 1 unit on its own costs 2 × 2 = 4 units as a transaction with an event item. The leaderboard loop (step 2.6) makes the same call: the outbox would close the gap too, but costs twice the write units per item of a plain write, a read on every poll, and up to one poll interval per repair. Write a semantic event item only when consumers need a meaning the item change doesn't carry.

---

Part 10. Choosing

Students (and interviewers) propose other answers: "just write both and fix it nightly", "use two-phase commit", "use event sourcing". Each has a place. The fair way to compare them is the same change under each, then the same questions for every approach.

The same change, three ways

o7 v4 (SHIPPED) under three designs, from the reference implementation. The hourly compare and the 2-minute coordinator outage are assumptions.

On equal terms

Three more, in sentences, with their costs next to their benefits:

• Log first (write the event to the broker; the database is built from it): atomic by construction, because there is only one write. But the application loses read-your-writes on its own database, and the broker becomes the source of truth, which needs its own retention and backup story.
• Event sourcing (the event log is the source of truth; state is derived): removes the dual write and gives a full history. It costs projection lag, schemas that must stay readable forever, and snapshots to bound replay time. See Primitive #23: Event sourcing and CQRS.
• Two-phase commit (XA): most brokers, and every search index and email provider, can't join; where one can, availability is tied to a coordinator, a prepared participant holds its locks until the coordinator decides, and retries still need to be idempotent. See Primitive #10: Two-phase commit and sagas.

Choose this when:

• Outbox + polling relay: one service, modest rates, no connector platform yet (the outbox loop's Rounds 1 and 2).
• Outbox + log tailing: many services or high rates on one database, and a team to run connectors and watch slots (the outbox loop's Round 3).
• CDC on business tables: copying data (a search index, a lake) where consumers want rows, not meanings, and you accept the coupling.
• The table's own stream: DynamoDB tables; add a semantic event item only when the item change doesn't carry the meaning.
• A direct write: only as a hint, with one of the above behind it.
• Sagas (a chain of local transactions with compensations) are how a change spanning services completes; they ride on an outbox at every step (Primitive #10).

---

Part 11. On AWS

Everything on this page is on AWS under different names. Each service reads some commit log, and each has its own retention, order and duplicate rules.

Managed services that use it

Running it yourself

Look-alikes that are not this mechanism

---

Part 12. What you've learned

Back to the 432 a day

Our service lost 432 changes a day to crashes between writes, and its first fix could stall a shard for a day. Here is how each piece closed a hole, snapshot by snapshot:

• One write. The index, the cache and the notifier no longer hear from the request; they follow the commit (S1, Parts 1 and 2). A crash can no longer leave a copy behind silently: the event exists if and only if the change committed.
• At least once, applied by version. The relay died after publishing and republished (S2); a notifier died mid-event and its successor recognised its own feed entry by src (S3, Part 3). Nothing happened twice.
• Order per order. Whole-bucket claims, one partition per key and the exactly-next check kept o7's versions in order, even across a resize (Part 4).
• The log as the mover. CDC replaced the relays' queries with a slot, and the heartbeat kept it moving (Part 5).
• Poison without a stall. o9's bad address stopped only o9, for only the indexer, and was repaired from the primary (S4, Part 6), where Lambda's defaults would have blocked the shard for up to a day (Part 9).
• Gaps refilled in order. The invalidated slot's lost events were republished from the outbox before the connector resumed (S5, S6); the indexer, away for 30 hours, rebuilt by noting the position first (S7, Part 7).
• Deletes that stick. o7's erasure travelled as a versioned event, and a late repair from a lagging replica was dropped by the tombstone and the GONE marker (S8, Part 8).

The whole story, event by event

The cheat card

Failure checklist

• Does any service write the database and then another system? A direct write with no stream behind it is a dual write, however fast.
• Does every event carry the aggregate's version, and does every consumer apply by version, never by adding?
• Do relays claim whole buckets (or does one reader decode in commit order), so one aggregate has one publisher?
• Does any relay remember "published up to seq N"?
• Does every consumer check versions contiguously, and park rather than skip a gap?
• Do derived numbered records store src, and does last_seq outlive what it numbers?
• Is heartbeat lag alarmed, and does every captured database have a heartbeat table in its publication?
• Is max_slot_wal_keep_size set below free disk minus table growth, with a tested republish runbook for when it fires?
• Is the outbox kept longer than the worst time to detect and repair a gap?
• Does every consumer have an error boundary, a dead-letter owner, a repair path that reads the primary, and a periodic compare?
• Are Lambda mappings and Pipes on DynamoDB streams configured with bisect, limited retries, a maximum record age, a destination and TRIM_HORIZON, with IteratorAge alarmed?
• Do deletes travel as versioned events, with tombstones kept longer than any late or repair write can arrive?

Think-first drills

Interview questions

Where to go next

• Write-Ahead Log, fsync & Group Commit: the log a change stream reads, why a reader pins it (Part 8) and what acks=all promises (Part 10).
• Leases, Fencing Tokens & Distributed Locks: the SKIP LOCKED claim (Part 9) and how writer_epoch is bumped across a failover (Part 7).
• The Idempotency & Effectively-Once Processing loop primitive (coming): where consumers store what they applied, notification idempotency keys, and exactly-once per hop.
• The Replication, Quorums & Read-Your-Writes loop primitive (coming): replica lag, and why a lagging replica can't be a snapshot. The Sharding, Hot Keys & Rebalancing and Multi-Region Failover loop primitives (coming) pick up partition choice and cross-Region event flow.
• Primitives: #05 Message queues vs event streams, #21 Isolation levels (what the outbox transaction's row lock does), #10 Two-phase commit and sagas, #23 Event sourcing and CQRS.
• Drill: The Dual-Write That Broke Search Consistency, answered in Parts 1 and 5.
• Loops: the transactional outbox (steps 1.0 to 1.6, 2.1 to 2.4, 3.1 to 3.6), payments (2.1, 2.3, 2.5, 3.6), wallet (2.2, 2.5), hotel reservations (2.1, 2.2, 2.4, 3.1, 3.5, R2.8), matching engine (3.4), proximity (2.1, 2.6, R2.8), ride-sharing (R1.6, 2.4, 2.6, 3.6), maps (2.6, R2.8), nearby friends (2.5), email (2.2, 2.4), YouTube (R1.6, R2.8), Drive (1.4, 2.3, 2.6, 3.1), ad-click aggregation (3.1), mobile stock trading (1.2, R2.5), mobile news feed (2.2, 2.3, 2.4), news feed (1.4, 3.1, 3.3), leaderboard (1.3, 2.6, 3.4), chat (R1.6), URL shortener (3.3), S3-like storage (3.1), web crawler (3.6), message queue (2.2), notification system (R1.11, 3.5) and mobile chat (1.1).