# Making CDC snapshot merges correct In part 2 of this 3-part series on rebuilding stateful CDC pipelines, we cover the keyed-state fix that keeps live data winning after a rebuild. **URL Source:** https://www.brex.com/journal/making-cdc-snapshot-merges-correct/part-2 --- Making CDC snapshot merges correct Part 2 of a three-part series on how Brex rebuilds and operates stateful CDC pipelines. **The problem created by isolation** In [Part 1](https://www.brex.com/journal/rebuilding-stateful-cdc-pipelines/part-1), we described why we separated historical snapshot data from the shared CDC mutation stream. Each application consumes: - a shared topic carrying live `INSERT`, `UPDATE`, and `DELETE` events - a private topic carrying snapshot `READ` events for that application A custom Flink source combines them into one logical stream. This architecture isolates rebuild traffic, but it removes any assumption that records from the two inputs arrive in a useful order. A live mutation can arrive before its corresponding snapshot row, or it can arrive after the snapshot row. The snapshot topic may process faster than the mutation topic, or vice versa. And a restart can change the timing all over again. The merge must remain correct under each and every one of those schedules. **The invariant** Suppose a source table contains a row for key `K`. The snapshot captures K at some earlier database position. While the snapshot is still running, the row may be updated or deleted. When both records reach the Flink job, the newer live mutation must win. That gives us the invariant: **For any key, a snapshot `READ` may be emitted only if no newer live mutation for that key has already been observed and then forgotten.** The last three words matter because remembering a mutation is easy, but remembering it for long enough is the actual correctness boundary. **Why this is not a conventional join** At first glance, the problem looks like a two-stream join: - join snapshot records with live mutations by primary key - choose the newest value A conventional windowed join is not a good fit; we do not want to buffer both streams until an event-time window closes. Snapshot duration can be long, and the two sources do not provide a useful shared arrival order. We also do not need to retain complete records from both inputs. For each key, we just need to answer one question: _Have we already seen a live mutation that makes this snapshot row stale?_ That can be represented with a small piece of keyed state. **The merge operator** The operator has three steps. **1. Partition by primary key** Both snapshot and mutation events are shuffled by the same primary key. That guarantees all records for key `K` reach the same parallel operator instance, even though they originated from different Kafka topics. **2. Record live mutations** When an `INSERT`, `UPDATE`, or `DELETE` arrives from the shared mutation stream, the operator: 1. writes a marker for the key into TTL-backed keyed state 2. emits the mutation downstream The marker does not need to contain the full entity value. Its purpose is to record that a newer live event has been observed for the key. Conceptually: ![code](https://brand.brex.com/transform/4292e62f-5db2-49e0-b707-b0b0c036e467/CDC-blog_inline-9) **3. Filter stale snapshot rows** When a snapshot READ arrives, the operator checks keyed state. ![code](https://brand.brex.com/transform/c1ef2b5e-eb9f-4f5a-a7b9-81d9cc76d524/CDC-blog_inline-10) If a marker exists, a newer mutation has already superseded the historical row. If no marker exists, the snapshot row fills in state that the application has not yet observed from the live stream. As such, the algorithm deliberately does not depend on ordering between the Kafka topics. **Updates and deletes use the same rule** Updates are the easiest case to visualize. Imagine the snapshot contains: ![code](https://brand.brex.com/transform/beb2e1b4-0457-4593-a270-82002d6a89b0/CDC-blog_inline-11) The live stream later contains: ![code](https://brand.brex.com/transform/9462e79d-3d64-45be-9317-ee62d59b392a/CDC-blog_inline-12) If the mutation arrives first, the operator marks `K5`, emits `value_at_t10`, and drops the stale snapshot row upon arrival. Deletes work the same way. Suppose a row existed when the snapshot was taken but was deleted afterward. The delete mutation records the key and emits the deletion downstream. A later snapshot `READ` for that key is dropped, preventing the historical row from resurrecting an entity that no longer exists. The marker represents precedence, not merely the presence of an update. **The role of TTL** Without expiration, mutation markers would grow forever. Every key ever updated would remain in state, even after the snapshot completed. We therefore store the markers with a time-to-live. The TTL bounds the extra state introduced by a rebuild, but it also creates the system's most important failure mode. Consider this sequence: 1. The snapshot captures `K5 = value_at_t1`. 2. A live update produces `K5 = value_at_t10`. 3. The operator records a marker for `K5`. 4. The marker expires. 5. The delayed snapshot `READ` for `K5 = value_at_t1` arrives. 6. The operator no longer remembers the mutation and emits the stale value. Unless another mutation for `K5` arrives later, the reconstructed output remains incorrect. Correctness, therefore, depends on the following operational guarantee: _The mutation-marker TTL must exceed the maximum time between the snapshot boundary and the processing of the final relevant snapshot row, with a safety margin._ In our platform, the default TTL is three days and can be dynamically extended for larger snapshots. **Reasoning about the state window** It is useful to model the operator's memory as a moving range of retained mutation positions. Let: - `ts` be the write-ahead-log position represented by the snapshot - `tm-s` be the oldest mutation position still represented in keyed state - `tm-e` be the newest mutation position the operator has processed The snapshot can fall in one of three places relative to that range. **Case 1: The snapshot is inside the retained window** ![code](https://brand.brex.com/transform/c64602e3-278f-4297-8791-94db9dbfb4f2/CDC-blog_inline-13) The mutation stream has advanced beyond the snapshot boundary, and the operator still remembers mutations that occurred around that boundary. Suppose the snapshot contains: ![code](https://brand.brex.com/transform/93b0038f-0cf2-4496-abb8-9bb2170de449/CDC-blog_inline-14) The operator has already seen `K5_t10`, a mutation after the snapshot's `K5_t1`, so it keeps `K5_t10` and fills the keys absent from the keyed state (`K3` and `K4`) from the snapshot. Every key resolves to its true latest value. ![tm-s ≤ ts ≤ tm-e: ts falls inside the state window, so every key resolves to its latest value.](https://brand.brex.com/transform/e15f449a-edcd-471f-b5ec-c9640c68756b/CDC-blog_inline-1) *tm-s ≤ ts ≤ tm-e: ts falls inside the state window, so every key resolves to its latest value.* This is the state we want the system to remain in until the snapshot completes. Case 2: The snapshot is ahead of the mutation stream ![code](https://brand.brex.com/transform/ff826c29-2ad4-48e7-86a3-e36f2ab2dfad/CDC-blog_inline-15) The change stream has not caught up to ts yet. The snapshot's `K2_t9 `sits ahead of `tm-e`, so it is emitted early as a future value (yellow) and re-delivered once the change stream advances to `t9`, at which point the output converges. Because downstream consumption is idempotent in our jobs, this is harmless. ![ts > tm-e: ts falls ahead of the state window; K2_t9 is emitted early as a future value (yellow), then re-delivered as the stream advances.](https://brand.brex.com/transform/42a2e1ca-8956-46e4-bbb6-d3ebc21cf15f/CDC-blog_inline-2) *ts > tm-e: ts falls ahead of the state window; K2_t9 is emitted early as a future value (yellow), then re-delivered as the stream advances.* This case is convergent, but it makes the sink requirement explicit: A pipeline that can observe the same logical value twice must use idempotent, upsert, deduplicating, or otherwise replay-safe output semantics. A non-idempotent side effect cannot be treated as harmless. Case 3: The snapshot falls behind the retained window ![code](https://brand.brex.com/transform/5899fc03-0620-40b4-be9a-827404078c11/CDC-blog_inline-16) A mutation made between `ts` and `tm-s` has already aged out of the keyed state. The later update to `K5` is no longer retained, so the operator does not know `K5` changed and emits the snapshot's stale `K5_t1` (red) instead of the correct `K5_t10`, and never corrects it. ![ts < tm-s: ts falls behind the state window, so the stale K5_t1 (red) slips through and is never corrected.](https://brand.brex.com/transform/3766b5f6-65ff-4d7c-a8be-819633fe7f81/CDC-blog_inline-3) *ts < tm-s: ts falls behind the state window, so the stale K5_t1 (red) slips through and is never corrected.* This is why the TTL is not merely a performance setting. It is part of the correctness contract. The platform must keep the system out of this case. Explicit watermarks versus retained mutation state This design was inspired by [DBLog's watermark-based CDC framework](https://arxiv.org/abs/2010.12597). DBLog writes low- and high-watermark records to the source database and observes them in the transaction log, creating explicit boundaries around each snapshot chunk. It uses those boundaries to interleave selected rows with live log events and suppress stale snapshot values precisely. Conceptually, DBLog's watermark interval corresponds to our safe `tm-s <= ts <= tm-e `case, but our boundaries are implicit: the operator derives them from the live mutations represented in TTL-backed keyed state. We size the TTL to outlive the snapshot, keeping the system out of the unsafe` ts < tm-s` case. We also tolerate `ts > tm-e`; the snapshot value may be emitted first, but later live mutations restore the latest value under our replay-safe sink semantics. What the algorithm assumes The merge is intentionally small, but its guarantees rely on several assumptions. **`ts` is a logical position, not an LSN** `ts` is a conceptual approximation of where a snapshot value falls in the source database's mutation history. It is not a PostgreSQL LSN carried by the snapshot event, nor is it a value compared by the Flink operator. Federated snapshots deliberately avoid LSNs. They are PostgreSQL internals, may vary across versions, and would couple the design to whether the snapshot runs against a primary or secondary instance. The three cases above describe where a snapshot `READ` effectively falls relative to the mutation window still retained in TTL-backed keyed state when it reaches the operator. The operator implements this model through marker presence and expiration, not by calculating `ts` or comparing database positions. **All records for a key reach the same operator instance** Snapshot and live records must use the same primary-key partitioning. If a schema or keying change routes the two inputs differently, the marker and the snapshot row can end up on different operator instances, causing the algorithm to fail. **Mutation markers survive recovery** The keyed state is checkpointed with the job. After a restart, the operator must restore both source positions and mutation markers consistently enough that it does not forget which keys have been superseded while replaying snapshot records. The exact delivery guarantee may still be at-least-once, but the output must converge under the sink's replay semantics. **Downstream publication is replay-safe** Case 2 can produce duplicate values, and blue/green cutover can briefly produce duplicate output from two job versions. This system is designed for materialized-state and upsert-style consumers. Pipelines with irreversible or non-idempotent side effects require additional deduplication or a transactional boundary. **Snapshot completion** The second input eventually becomes idle, but the control plane still needs an authoritative completion signal. A robust readiness decision should distinguish at least three states: 1. the snapshot connector has finished reading the source 2. the snapshot topic has been fully consumed 3. the Flink job has checkpointed after processing the final snapshot records and caught up to the live mutation stream Only then is the reconstructed state ready for a blue/green promotion. **Benchmarking the invariant** We did not want correctness to depend only on a convincing argument. We built an end-to-end benchmark that reconstructs a table through the composite source while concurrent clients continuously modify the source database. The key property is that the snapshot is triggered _while writes are still in flight_, so the snapshot READ stream and the live mutation stream overlap on the same keys — the exact condition the invariant governs. **The harness** The workload is driven by `pgbench` against a source Postgres table, `public.hello(id, greeting)`, using this custom transaction: ![code](https://brand.brex.com/transform/058d39de-a42f-4451-8938-04243bc5b20f/CDC-blog_inline-17) Because `id` is drawn uniformly from a fixed key space of 10,000 keys, every transaction is either an `INSERT` for a new key or an `UPDATE` for an existing one. Mutations concentrate on the same keys the snapshot is capturing, maximizing the overlap between the two streams. The e2e script sequences the run so the snapshot and mutation streams are forced to interleave: 1. Start the `pgbench` write workload against the source for 100 seconds. 2. Wait 10 seconds, then trigger a federated (incremental) snapshot while writes continue. 3. Let writes run for the remaining ~90 seconds. 4. Wait an additional 30 seconds for the snapshot and mutation streams to fully drain through the Flink job. 5. Compare the reconstructed sink table against the source-of-truth `hello` table with a set-difference query. Any surviving stale or resurrected row shows up as a differing row. **The run** ![the run table](https://brand.brex.com/transform/80867892-5a1c-4b8a-997d-4e5586a4c40d/The-Run-Table-2) After the streams drained, the set-difference query returned **0 rows**: the state reconstructed through the composite source was identical to the source-of-truth table. **What this run does and does not cover** This workload is upsert-only, so it exercises the `INSERT` and `UPDATE` precedence paths (Case 1 and Case 2 above) under real concurrency, but it does not issue DELETEs and does not inject a job restart or a marker expiry. Those scenarios are covered separately in the operator's unit tests; extending the randomized e2e harness to include deletes, restarts, and forced marker expiry is the natural next step for publication-grade evidence. **Costs and tradeoffs** The keyed-state approach is deliberately simple, but it is not free. During a backfill: - each recently mutated key creates additional state - checkpoints and savepoints grow - RocksDB disk use can increase - checkpoint duration may increase - the job consumes from an additional Kafka topic - a longer TTL raises the state bound The state size is related to the number of distinct keys mutated during the TTL window, not the total size of the source table. That is an attractive trade-off for our workloads: we retain a compact precedence marker for recently changed keys rather than buffering or versioning the entire snapshot. The system also creates more Kafka topics and storage objects. We accepted that infrastructure cost in exchange for isolating rebuild traffic and allowing each pipeline to choose its own backfill capacity. **What we would make explicit in a reusable design** For another team implementing a similar system, we would treat the following as first-class platform controls rather than documentation alone: - verify that snapshot and mutation events use identical keys - require replay-safe sink semantics - calculate a conservative snapshot-duration budget - reject or warn on TTL configurations below that budget - alert as elapsed snapshot time approaches the retention boundary - expose snapshot-topic lag and live-topic lag separately - track keyed-state size and checkpoint growth during backfills - persist an authoritative snapshot-generation identifier - make completion and promotion criteria machine-readable - test deletes and restarts, not only updates The merge algorithm is only a few lines of logic. Most of the engineering work is making its assumptions observable and enforceable. **The broader lesson** The central correctness problem is not that two streams can arrive out of order. Stream-processing systems already live with reordering. The dangerous condition is **forgetting precedence too early**. Once a live mutation has established that a historical row is stale, the system must retain that knowledge until the historical input can no longer produce the row. In our design, keyed state carries that knowledge, and TTL bounds its lifetime. That gives us a compact operator with a clear contract: - live mutations always take precedence over historical reads - the merge does not depend on cross-topic ordering - temporary duplicates are handled by replay-safe sinks - correctness fails if mutation memory expires before the snapshot is complete A clear invariant made the architecture easier to reason about, benchmark, and operate. ## Related Articles ### [Operating hundreds of stateful flink pipelines](https://www.brex.com/journal/operating-hundreds-of-stateful-flink-pipelines/part-3) In part 3 of this 3-part series on rebuilding stateful CDC pipelines, we cover how rebuilds became self-service instead of expert-only work. ### [CrabTrap: an LLM-as-a-judge HTTP proxy to secure agents in production](https://www.brex.com/journal/building-crabtrap-open-source) Here's what we learned in building and open-sourcing CrabTrap, and why we believe it’s a major step forward in the security of agent harnesses. ### [When your infrastructure tool's best interface is a web page, you have an agent problem](https://www.brex.com/journal/terminal-ui-became-an-agent-interface) We built a terminal UI to replace a browser tab. Then we realized we'd built something agents could use too. ### [Not all MCPs are created equal](https://www.brex.com/journal/not-all-mcps-are-created-equal) Two MCP servers can look identical on paper and behave nothing alike. Here are the three questions that tell them apart. ### [AI cost visibility is a problem for every company. Here's how we're solving it.](https://www.brex.com/journal/magpie-ai-cost-visibility-dashboard) Traditional spend management can't handle the complexity of AI costs, so we built Magpie, an internal tool that provides real-time visibility of AI spending. ### [Building autonomous agents for technical tasks: 5 lessons learned](https://www.brex.com/journal/building-autonomous-agents-for-technical-tasks) Brex built a general-purpose template for delegating engineering work to agents with confidence. Here's how we did it and what we learned.