Exactly-once is a per-sink promise

2026-06-21 • by Vahid Negahdari

This is the second post in a series on the hard parts of building a Change Data Capture engine, told through DeltaForge, a CDC engine I’m writing in Rust. The first post was about understanding the data. This one is about delivering it, and the promise everyone wants you to make about that.

Worth saying up front: most deployments do not need multiple sinks in one pipeline. A single sink per pipeline, or a separate pipeline per destination, is common and often the right choice. This post is about the harder case, where one pipeline fans out to several sinks at once, and the delivery guarantees stop being simple.

The word everybody wants

“Exactly-once delivery” is the phrase every CDC evaluation asks for. It sounds like a property of the tool: either it has it or it doesn’t. So the honest answer tends to disappoint, because the moment a single pipeline writes to more than one place, exactly-once stops being one thing you can promise.

A normal DeltaForge pipeline fans out. The same stream of row changes might land in Kafka for the application, Redis for a cache, and S3 as Parquet for the lake, all from one source, all at once. Ask “is this exactly-once” and the only correct answer is a question back: exactly-once where?

Because these sinks do not have the same powers. Kafka has transactions. S3 has atomic file writes and nothing else. Redis and a webhook have neither. You cannot hand all of them a single uniform guarantee, and any tool that claims you can is selling you the word, not the property.

DeltaForge’s guarantees documentation opens with a rule I like: the word “exactly-once” is used only where the engine guarantees no duplicates without the consumer’s help. Everywhere else it says “at-least-once” and names the dedup mechanism. That rule is the whole design in one sentence. The rest of this post is what it takes to earn it.

The floor: never lose anything

Before you can talk about duplicates, you have to rule out loss. The mechanism is a single invariant, and it is almost boring: a checkpoint is saved only after the sink acknowledges the batch, never before.

let result = sink.send_batch(&batch).await?;   // wait for the ack
if result.ok {
    checkpoint.save(sink_id, batch.end_lsn);    // only now
}

That ordering is the entire at-least-once guarantee. If DeltaForge crashes after the sink accepted a batch but before the checkpoint is written, the checkpoint still points at the old position, so on restart the source replays that batch. The sink sees it twice. Nobody sees it zero times. Loss is impossible by construction; duplicates are the price, and the whole game from here is deciding who pays it.

The twist: every sink keeps its own checkpoint

Here is where the multi-sink reality bites. If the pipeline kept one checkpoint for the whole batch, a single slow sink would define the speed of everything. An S3 sink rolling a 256 MB Parquet file would hold back a Kafka sink doing a hundred thousand events a second. That is unacceptable, so DeltaForge gives every sink its own checkpoint, keyed by {source_id}::sink::{sink_id}. Each one advances on its own.

Which immediately raises the question: if the sinks are at different positions, where does the source replay from after a crash? It has one read position, and it has to be safe for everyone. So it replays from the minimum checkpoint across all sinks. Anything less would lose data for the sink that is furthest behind.

That single choice, replay from MIN, has a consequence worth sitting with: the sinks that were ahead of the minimum re-receive every event between MIN and their own checkpoint. They already processed those. They are, by design, going to see them again.

The demo below is that mechanism. Three sinks ack at different speeds, so their checkpoints drift apart. The dashed line is MIN, the only place the source can rewind to. Press play, let them spread out, then crash it. Try stalling S3 first to make the gap bigger.

Stall S3 for a while and the lesson gets loud: the slowest sink sets the replay point for the entire pipeline, and the faster a sink runs ahead of it, the more it re-receives when something restarts. Independent checkpoints buy you throughput. They do not buy you a free lunch on recovery.

So push exactly-once to the edge

At-least-once with replay is the floor. Getting to exactly-once means eliminating those duplicates, and you can only do that where the sink itself can help. So instead of one pipeline-wide promise, DeltaForge makes a specific one per sink:

Kafka is the only row that gets the strong word, and it earns it. With exactly_once: true, each batch is wrapped in a Kafka transaction, begin_transaction to commit_transaction. A consumer reading with isolation.level=read_committed never sees a batch that was not committed, so a crash mid-batch is invisible: the transaction aborts and replays from the same checkpoint, and the consumer only ever observes the clean version. That is exactly-once with no cooperation required, which is why it gets the label. It costs about 7 to 11 percent in throughput with tuned batches, and that number is honest too.

Every other sink is at-least-once, and the table says so plainly, along with the one thing the consumer has to do to close the gap. NATS can dedup server-side inside a duplicate_window. Redis and HTTP consumers dedup on an id you carry in the payload. S3 gives you atomic files, so a reader never sees a half-written Parquet file, but a replay produces a new file for the same events, and you dedup in the query with a MERGE or an event_id. None of these is exactly-once, and calling them that would be a lie that fails in production at the worst possible time.

Wiring it up is declarative. The guarantee per sink is a property of the sink, right next to where it is defined:

sinks:
  - type: kafka
    config:
      id: app-kafka
      brokers: ${KAFKA_BROKERS}
      topic: orders
      exactly_once: true      # batch = Kafka transaction
      required: true

  - type: redis
    config:
      id: cache-redis
      uri: ${REDIS_URI}
      stream: orders
      required: false         # best effort, must not stall the pipeline

  - type: s3
    config:
      id: lake-s3
      bucket: analytics
      format: parquet
      required: true

Keeping the source transaction whole

There is a subtlety underneath all of this. A CDC event is not really the unit that matters; a database transaction is. If an order and its line items were written in one transaction, delivering the order without the items, even briefly, is a lie about what the database did.

So by default (respect_source_tx: true), DeltaForge refuses to split a batch in the middle of a source transaction. The accumulator watches a tx_end flag on each event, set on the MySQL commit event or the Postgres COMMIT record, and if a batch hits its size limit mid-transaction it grows past the limit rather than cut the transaction in half. Every row from one database transaction lands in the same batch, and the batch is the atomic unit each sink either takes or does not.

For the Kafka sink with exactly_once: true, that composes into something strong: one database transaction becomes one Kafka transaction, and a read_committed consumer sees all of its rows atomically or none of them. What DeltaForge does not claim is atomicity across heterogeneous sinks. Kafka can commit a transaction while Redis is still retrying the same batch. There is no two-phase commit spanning a Kafka broker and an S3 bucket, because there cannot be. Each sink’s checkpoint tracks its own progress, and the honesty is in not pretending otherwise.

Who has to say yes

The last piece is deciding what “delivered” even means when one sink is down. DeltaForge makes it a policy rather than an accident.

Every sink is required: true or required: false. A required sink must ack or the pipeline does not advance. An optional sink is best-effort: if it fails, it is logged, the pipeline keeps moving, and that sink’s own checkpoint simply stays put and catches up on the next restart. That is exactly the S3-versus-Kafka tradeoff from the demo, turned into a config flag: make S3 optional and Kafka keeps flowing through an S3 outage, at the cost of S3 re-delivering a larger gap once it comes back.

On top of that, a commit policy decides how many sinks must agree before any checkpoint moves:

spec:
  commit_policy:
    mode: required        # required | all | quorum
  sink_batch_deadline_secs: 60   # outer bound; a hung sink cannot stall forever

The important detail is when the policy is evaluated: before any checkpoint is written, not after. If the policy is not satisfied, not a single sink advances. That prevents the quietly corrupting failure mode where a fast optional sink checkpoints past a batch that a required sink never accepted, so the required sink’s replay would skip it. Check first, then commit, all or nothing at the policy boundary.

The honest version

Put it together and the promise DeltaForge actually makes is not “exactly-once delivery.” It is a set of specific per-sink guarantees:

• nothing is ever lost, because checkpoints follow acks
• each sink runs at its own speed, with its own checkpoint, and replays only its own gap
• exactly-once is real where the sink can enforce it (Kafka transactions) and claimed nowhere else
• a source transaction stays whole within any single sink
• and the pipeline advances only when your stated policy says it may

That is a longer sentence than “exactly-once,” and it is the true one. In distributed systems the useful guarantees are almost never the one-word ones, and a tool that gives you the honest, boring, per-sink version is worth more than one that gives you the word you asked for and a data integrity incident eighteen months later.

Next in the series

Delivery assumes you already caught every change. The next post is about the moment before that: the initial snapshot, and the quiet way it corrupts itself when the transaction log you were going to resume from gets purged out from under you while the snapshot is still running.

The code is at github.com/vnvo/deltaforge; the delivery guarantees live in the runner coordinator and the checkpoints crate, and the full correctness spec is in the docs. It is pre-release, and the delivery path is the part I most want people to try to break.