Guarantees & Correctness

This page defines DeltaForge’s data delivery guarantees, ordering model, transaction semantics, failure handling, and operational boundaries. The guarantees below describe current behavior. Where a verification is planned rather than already in place, the Correctness Test Matrix marks it explicitly, so read a guarantee together with its test status.

Delivery Guarantees

Per-sink delivery tiers

Terminology rule: DeltaForge does not claim end-to-end exactly-once delivery for any sink. Every sink is at-least-once; the differences are in what each sink adds on top. Kafka exactly_once: true adds transactional atomic-batch visibility (a read_committed consumer sees a whole batch or none of it) but still delivers at-least-once across a restart, so consumers must dedup to reach exactly-once end to end. NATS and Redis add a dedup mechanism (server-side or consumer-side) whose effectiveness depends on configuration or consumer behavior outside DeltaForge’s control. Calling any of these “exactly-once” without the consumer’s cooperation would be misleading.

What “at-least-once” means

• No data loss: every event from the source is delivered to the sink at least once. Checkpoints are saved only after the sink acknowledges delivery — never before.
• Duplicates on crash recovery: if DeltaForge crashes after delivering a batch but before saving the checkpoint, that batch is replayed on restart. Consumers must handle duplicates (see Consumer Guidance below).
• No silent drops: events are never discarded. On a transient delivery error a sink applies a bounded in-sink retry (exponential backoff, finite attempts). If those attempts are exhausted on a required sink, the checkpoint does not advance and the batch is re-delivered from the source on the next pipeline restart (source replay) - there is no unbounded in-session retry loop. See Error Classification & Retry.

What Kafka transactions guarantee (exactly_once: true)

exactly_once: true gives transactional atomic-batch delivery, not end-to-end exactly-once. Each batch is wrapped in a Kafka transaction (begin_transaction / commit_transaction), so consumers using isolation.level=read_committed see a whole batch or none of it - never a partial batch. If a transaction fails before commit, it is aborted and re-attempted from the same checkpoint position.

The restart boundary is still at-least-once: DeltaForge advances its own checkpoint only after the Kafka transaction commits, so a crash between the commit and the checkpoint persist replays that batch on restart as a new transaction. A read_committed consumer then sees the batch a second time. To reach exactly-once end to end, consumers must dedup on the event id (or idempotency key), which is stable across replays.

The transactional-producer overhead is ~7-11% with properly tuned batch sizes. See the Performance guide for benchmark details.

Ordering Model

Within a source

Events are emitted in the source’s native order:

• MySQL: binlog file + position order. WriteRowsEvent batches preserve row order within each binlog event.
• PostgreSQL: LSN (Log Sequence Number) order. One WAL message per row change.

DeltaForge does not reorder events. The source order is preserved through the pipeline.

Within a batch

All events in a batch maintain their source order. The delivery task processes batches in FIFO order from a bounded channel — no reordering between batches.

Per-primary-key ordering (the core guarantee)

DeltaForge guarantees per-primary-key ordering within a table under non-sharded operation. This means: for any single row identified by its primary key, all changes (INSERT, UPDATE, DELETE) are delivered to the sink in the exact order they occurred in the source database.

For Kafka specifically: the default message key is the serialized primary key, so events for the same row always go to the same partition and arrive in order. With dynamic routing (key template), ordering follows the resolved key — events with the same key are ordered; different keys may land in different partitions.

Cross-table ordering

There is no global ordering across tables. Events from different tables may be interleaved across batches. This is by design — enforcing global ordering would require single-threaded delivery, which would cap throughput.

However, when batch.respect_source_tx: true (the default), all rows from a single database transaction are kept in the same batch (see Transaction Boundaries below). This preserves causal ordering within a transaction.

Ordering under retries

When a batch delivery fails and is retried, the batch is re-delivered as a unit in the same order. No reordering occurs within or across retries. The max_inflight=1 setting (default) ensures strict ordering; with max_inflight > 1, batches are still delivered in FIFO order by the single-threaded delivery task.

Cross-sink ordering

All sinks receive the same batch simultaneously. The relative order of events is identical across all sinks.

Transaction Boundaries

How it works

When batch.respect_source_tx: true (the default), the coordinator checks each event’s tx_end flag before splitting a batch:

• MySQL: tx_end is set on the last row of each XID (transaction commit) event.
• PostgreSQL: tx_end is set on the COMMIT WAL record.

The batch accumulator will not split a batch at a point that would separate rows from the same transaction. If the batch limit (max_events or max_bytes) is reached mid-transaction, the batch grows beyond the limit to include all remaining rows in that transaction.

What this guarantees

• All rows from one database transaction appear in the same batch.
• A batch is DeltaForge’s unit of delivery accounting: a sink’s checkpoint advances only if the whole batch succeeds. Whether a consumer observes the batch atomically depends on the sink:
  • Kafka exactly_once: true: the batch is one Kafka transaction, so a read_committed consumer sees all rows from the DB transaction together or not at all - within a single delivery. Across a restart the batch can be replayed as a new transaction, so the consumer may see the same transaction twice (at-least-once); dedup on event id to collapse it.
  • Non-transactional sinks (Redis, HTTP, NATS, S3, ClickHouse, Elasticsearch): DeltaForge does not make the batch atomic for the consumer. A sink failure partway through a batch can leave some rows visible downstream, and a retry or restart re-delivers the batch, producing duplicates. These sinks are at-least-once; use their stated dedup mechanism.
• Cross-table transactions: a transaction spanning tables A and B is emitted as a single batch containing events for both tables, tagged with the same tx_id, and delivered as a unit. This is stronger than “tagged but not grouped” - all events from one DB transaction are in one batch. Consumer-visible atomicity for that batch still follows the per-sink rule above.

Precise transaction semantics

To avoid ambiguity, here is exactly what DeltaForge guarantees about transactions:

• Events from one source transaction are emitted contiguously within a single batch.
• Multi-table transactions preserve commit grouping — all rows from tables A and B in one DB transaction appear in the same batch.
• Within a single sink, the batch is the unit of delivery accounting: the sink’s checkpoint advances only if the whole batch succeeds. Consumer-visible atomicity of that batch holds only for a transactional sink (Kafka exactly_once: true) and only within one delivery; see What this guarantees.
• Across heterogeneous sinks, DeltaForge does not guarantee atomic commit. Kafka may commit a transaction while Redis is still retrying. Each sink’s checkpoint tracks its own progress independently.
• Retries and restarts do not break transaction grouping (a replayed batch contains the same events in the same order), but they can re-deliver a transaction that was already delivered. Delivery is at-least-once; consumers dedup on event id.

Edge cases

• A single database transaction that exceeds max_events or max_bytes is still kept in one batch. The limits are exceeded rather than the transaction being split.
• With respect_source_tx: false, batches are split purely by size/time limits regardless of transaction boundaries. Cross-table transaction atomicity is not preserved in this mode.

Disconnect inside a transaction (PostgreSQL)

When the replication stream ends inside a transaction (the connection is closed or reset), the source reconnects to the same verified server, resumes at the end of the last transaction it handed on, abandons the cut-off transaction (an internal abort marker, never a CDC event) and receives it again from its beginning:

Each abandoned transaction logs a warning (source, transaction, last complete position, events abandoned, whether duplicates are possible). Metrics:

respect_source_tx is true unless a pipeline sets it to false: BatchConfig fills every key a pipeline’s batch section omits from its defaults (respect_source_tx: true), and the coordinator applies the same defaults when there is no batch section at all. Keep it true for production. In that mode an open transaction is buffered in memory up to max_tx_events / max_tx_bytes, beyond which the oversized_tx policy applies.

Initial-snapshot anchoring

The initial snapshot and the CDC stream meet at a single anchor so that no committed row is lost across the boundary. The checks below run only when a snapshot runs; CDC-only pipelines (snapshot.mode = never) are unaffected.

• PostgreSQL anchors CDC at the replication slot’s consistent point C. Every row committed at or before C is in the snapshot; every row after C is in the CDC stream. Rows committed in (C, snapshot-export] fall in both - a bounded at-least-once overlap, not exactly-once. Current-state sinks (with version_source: source_position) converge via last-writer-wins; append-only sinks receive the overlap twice. A snapshot completed under the older anchor is flagged by the gauge deltaforge_snapshot_unsafe_anchor = 1 until a safe re-snapshot.
• MySQL brackets the anchor under a brief FLUSH TABLES WITH READ LOCK: all snapshot workers open their consistent-snapshot transactions and the binlog position + GTID set are captured while the lock is held, so every worker shares one view that matches the captured position exactly. This closes a real initial-snapshot data-loss window present in earlier versions (independent per-worker snapshots with the position captured afterward). It requires gtid_mode = ON, binlog_format = ROW, InnoDB tables, and the global RELOAD privilege; managed MySQL without RELOAD fails closed rather than snapshotting unsafely (no silent fallback).

Snapshot completion follows the commit policy. A snapshot is complete only once its terminal barrier is committed by the sinks the policy requires; an interrupted snapshot is copied again in full (at-least-once, never resumed table by table); a sink that missed part of a completed snapshot is reported (sink_snapshot_incomplete) and excluded from the resume position rather than silently treated as complete. See Initial Snapshots.

Failure Isolation

Per-sink independence

All sinks deliver concurrently. One sink’s failure does not block other sinks:

1. The coordinator dispatches the same batch to all sinks simultaneously.
2. Each sink’s delivery result is collected independently.
3. Only sinks that delivered successfully get their checkpoints advanced.
4. Failed sinks remain at their prior checkpoint position — they will receive the same batch again on retry or restart.

Coordinator-level deadline (defense in depth)

In addition to each sink’s own internal timeout (e.g. send_timeout_secs on the S3/Kafka/Redis/NATS/HTTP sinks), the pipeline supports an outer deadline that the coordinator applies to every send_batch call:

spec:
  sink_batch_deadline_secs: 60   # outer bound on any one sink's send_batch

Behavior:

• Each sink’s future is wrapped in tokio::time::timeout(deadline, ...). If a sink exceeds the deadline, the coordinator gives up on it for that batch, classifies it as SinkError::Backpressure, and continues with the others.
• A timed-out required sink causes its checkpoint to stay behind (same as any failure); the source replays from MIN(checkpoints) on restart.
• A timed-out optional sink advances none of its state for that batch; same in-session-loss / on-restart-recovery semantics as documented in Required vs. optional sinks.
• Metric: deltaforge_coordinator_sink_timeout_total{pipeline,sink}.
• Default: None (no outer bound). Set this for defense-in-depth against a sink that hangs despite its own internal timeout (bug, deadlock, infinite retry loop in a dependency).

The two timeout layers are complementary:

Required vs. optional sinks

Each sink is marked required: true (default) or required: false:

• Required: must succeed for the pipeline to consider the batch delivered. Each sink applies bounded sink-local retries (finite attempts with backoff) within a single send_batch; if those are exhausted for a required sink, no checkpoint advances for any sink and the pipeline stops. There is no in-session pipeline-level retry of the batch - recovery is by source replay from MIN(checkpoints) on the next restart.
• Optional (best-effort): failures are logged but don’t prevent the pipeline from advancing. The failed sink’s own checkpoint stays at its last-successful position; the source’s MIN-checkpoint reader doesn’t go back, but on the next pipeline restart, the source replays from the failed sink’s stuck position and the failed sink catches up.

What an optional sink failure means in practice

When an optional sink fails for batch B (events at LSN 100):

1. Required sinks succeed → their checkpoints advance to LSN 100.
2. The optional sink’s checkpoint stays at its prior position (LSN 50, say).
3. The pipeline continues to batch B+1 (events at LSN 101+). The source’s in-memory position is now 101, not 50.
4. The optional sink receives batch B+1 with events at LSN 101+. It may succeed or fail again: its checkpoint stays at LSN 50 for the rest of the session either way. A later success never moves a sink’s checkpoint past a batch that sink failed (in both batching modes).
5. Until restart: if the optional sink stays failing, the source’s in-memory position keeps marching forward. The failed events between the sink’s stuck checkpoint and “now” are not in any retry queue for this session.
6. On restart: the source reads MIN(required_cp, optional_cp) = 50 and replays from 50. The optional sink (back from outage, presumably) catches up. The replay is source-level: every sink receives it, so sinks already past LSN 50 see duplicates and dedup or accept (idempotent design).

Before the source delivers anything, every configured sink without a checkpoint (a new pipeline, or a newly added sink) gets one at the position the source starts from. A sink that fails its very first batch is therefore part of MIN(...) and the batch is replayed to it on restart. If that initial checkpoint cannot be persisted, the source does not start (fail closed).

This is the practical reality of required: false:

• ✅ A failed optional sink does not stall the pipeline.
• ✅ A failed optional sink does not lose data permanently — restart-replay recovers it.
• ❌ A failed optional sink does not catch up in-flight — events between failure and restart are only delivered on restart.
• ❌ A long-running pipeline with a chronically failing optional sink accumulates a growing replay gap — the longer between failures and restart, the more events the failed sink will re-deliver after restart.

For the S3 sink specifically (where extended outages are realistic — S3 throttling, regional issues), the operational choice is:

Multi-batch retries vs in-batch retries

Two distinct retry layers exist and should not be conflated:

1. In-batch retries (within a single send_batch call): each sink’s internal retry policy (exponential backoff, finite attempts). These are fast (<1s typical) and apply to transient errors like Queue full or Connection timeout.
2. Pipeline-level retries (across batches): triggered by required: true failures. The coordinator does NOT auto-retry the same batch from the source channel — instead, the source-side replay mechanism handles recovery on restart.

required: false skips pipeline-level retries for the failing sink. There is no “background queue” that retries the failed batch later in the same session.

Commit policy

The commit policy determines when checkpoints advance:

The policy is checked before any checkpoint is committed. If the policy isn’t satisfied, no sink advances — this prevents optional sinks from getting ahead of failed required sinks.

Per-sink checkpoints

Each sink maintains its own checkpoint key ({source_id}::sink::{sink_id}). On restart, the source replays from the minimum checkpoint across all sinks. This means:

• A fast sink is never held back by a slow one during normal operation: while the pipeline keeps running, the source’s in-memory read position does not rewind for a lagging sink.
• Recovery is not selective per sink. On restart the source resumes from the MIN checkpoint across all sinks, so every sink - including ones already ahead - is re-delivered the events from that minimum forward. Sinks that were ahead receive duplicates and must dedup (see On crash recovery).
• Adding a new sink lowers the MIN to that sink’s earliest position, so on the next restart the source replays from there and the existing sinks are re-delivered those events too (they dedup). The new sink is not backfilled in isolation.

Fatal errors

Some errors are unrecoverable and stop the pipeline immediately:

• Kafka ProducerFenced: another producer instance started with the same transactional.id. The broker fences the old producer permanently.
• Permanent auth revocation: credentials are invalid and retrying won’t help.

Fatal errors return SinkError::Fatal and are not retried. The pipeline stops and requires operator intervention.

MySQL replay across a schema change

MySQL rows are decoded by column position. Each rows event is decoded with the schema version proven for its position in the binlog - the definition the table had when the rows were written - not with the table’s current definition, so a restart that replays rows written before a DDL decodes them with the old definition and the rows after it with the new one. How a version is proven is described in Schema tracking. The proven version must also agree with the column layout the binlog recorded for those rows (column count, storage types and type metadata; with binlog_row_metadata=FULL also names, signedness, charsets and the primary key); if it does not, the source stops fail-closed.

When no version can be proven, the source stops fail-closed: it emits no event for those rows and does not advance the checkpoint, rather than put values under the wrong columns. With binlog_row_metadata=MINIMAL (the MySQL default) this happens when retained rows predate a DDL the source never observed (for example a checkpoint older than a schema change made while an earlier deployment ran), after a statement that cannot be attributed to specific tables (a versioned-comment DDL, several statements in one event, ALTER DATABASE or DROP DATABASE of a tracked database, among others; the tables it may affect stay unproven until later proof for each exists - usually the table’s next rows, once its definition is captured and proven unchanged since them, or a later DDL of the table that the source proves), and when two DDLs of one table both happen before the source reads the first, with rows between them. The source never chooses a version from the MINIMAL layout alone: two tables or columns with the same storage types cannot be told apart without names.

Recovery: the safe default is to re-snapshot (restart the pipeline once with snapshot mode always), which rebuilds current state without decoding the retained rows. Moving the source position past the DDL instead is a manual, lossy action: it intentionally abandons every retained change between the checkpoint and the new position, for all captured tables (the binlog position is shared by the whole stream), and should only be taken after an operator assessment of that data loss.

Set binlog_row_metadata=FULL to remove most of these stops: without positional proof the source then selects the one recorded definition that exactly matches the rows’ metadata, and records that choice durably before emitting them. Rows whose definition was never recorded (the shape between two DDLs the source did not see separately) still stop the source; FULL lets them through only when that definition was already recorded and is the single exact match.

S3 sink atomicity guarantees

The S3 sink commits at file granularity, not event granularity. Specifically:

• File-level atomicity: a Parquet or JSONL file is only visible at its final S3 key after the multipart-complete call returns success. Readers never see a partial file. Verified by tests (abandon_all_produces_no_visible_file, drop_pool_without_close_produces_no_visible_file).
• At-least-once per file: if send_batch succeeds and the process crashes before the source checkpoint is committed, the same events will be re-delivered on restart and produce a new file with a different ULID. The original file is also retained — there is no automatic deduplication. Downstream consumers must dedup via MERGE INTO or event_id.
• Mid-upload crash: a process killed during a multipart upload leaves orphan parts on S3 (not a visible file). The bucket lifecycle policy AbortIncompleteMultipartUpload: DaysAfterInitiation=1 reclaims them within 24h. This policy is an operational prerequisite — without it, abandoned uploads accumulate storage cost. See deployment.
• Per-row DLQ: encoder failures are isolated per row via a slow-path retry on the same writer. The bad row appears in BatchResult.dlq_failures and is routed to the DLQ; the rest of the batch is written normally. The fast path (success) costs nothing extra — per-event retry only triggers on encoder errors. ParquetFileWriter and JsonLinesFileWriter both roll back their state on per-call failure so the writer remains safe to reuse during the retry loop.

For exactly-once at the event level (i.e., dedup without consumer cooperation), the planned Iceberg sink in Phase 2 uses atomic snapshot commits to make file appearance and event commit equivalent operations.

Error Classification & Retry

Retry behavior by sink

All sinks use exponential backoff with jitter. The classification determines whether an error is retried:

Kafka:

NATS:

Redis:

S3:

After retry exhaustion

If all retry attempts fail for a retryable error, the error is propagated to the coordinator. The coordinator’s behavior depends on the commit policy:

• Required sink: the batch is not committed and its checkpoint stays put. The coordinator does not re-inject the same batch from the source channel in-session; recovery happens by source replay from MIN(checkpoints) on the next restart (see Multi-batch retries vs in-batch retries).
• Optional sink: the failure is logged, and the pipeline continues with other sinks.

Checkpoint Semantics

When checkpoints are saved

The checkpoint commit follows a strict sequence:

1. Accumulate events from source into a batch
2. Run processors (transform, filter)
3. Deliver batch to ALL sinks concurrently
4. Check commit policy (required/all/quorum)
5. Commit per-sink checkpoints (only for successful sinks)

Key invariant: a checkpoint is saved only after the sink has acknowledged delivery AND the commit policy is satisfied. This is the foundation of at-least-once delivery.

On crash recovery

1. DeltaForge reads per-sink checkpoints from the checkpoint store.
2. The source resumes from the minimum checkpoint across all sinks.
3. Sinks that were already ahead of the minimum position receive duplicate events - they must handle these idempotently (Kafka read_committed still sees the replayed batch as a new transaction, so dedup on event id there too).
4. Sinks that were behind receive their missing events.

Checkpoint storage

Checkpoints are stored in SQLite (default) with WAL mode and synchronous=NORMAL for durability. The checkpoint store survives SIGKILL — no graceful shutdown required for checkpoint safety.

Event replay

Event replay re-delivers captured commit units from the durable journal to selected sinks. Its correctness guarantees:

• At-least-once re-delivery. The durable cursor is advanced only after a sink acknowledges an envelope. A crash between the acknowledgement and the cursor persist causes the acknowledged envelope to be delivered again on restart, so dedup-capable sinks must absorb duplicates (the same idempotency key applies).
• No checkpoint mutation during replay. A running job never advances or rewinds the source or per-sink checkpoints; only normal live delivery moves checkpoints after the handoff.
• Pause and handoff. Selected sinks are excluded from live delivery and from commit-policy evaluation while a job runs, and rejoin the live set only at an acknowledged handoff to a frozen tail H; the first live delivery to a restored sink is strictly after H.
• Fail-closed capture. A commit unit that cannot be captured verbatim aborts the pipeline rather than advancing the checkpoint past uncaptured data.

Backpressure

DeltaForge implements end-to-end backpressure without dropping events:

Source → [event channel] → Accumulator → [batch channel (max_inflight)] → Delivery → Sinks

1. Sink slow: delivery task blocks waiting for sink acknowledgement.
2. Batch channel full: accumulator blocks waiting to enqueue the next batch (bounded by max_inflight).
3. Event channel full: source blocks waiting to enqueue the next event.
4. Source slows: the database connection idles until the channel has capacity.

No events are dropped at any stage. Backpressure propagates from the slowest sink all the way back to the source connection.

max_inflight controls the pipeline depth: higher values allow overlapping batch building with delivery (better throughput), lower values reduce memory usage and latency.

Consumer Guidance

Idempotency key

Every event has a deterministic idempotency key in the format:

{tenant}|{db}.{table}|{tx_id}|{event_id}

This key is identical across replays — the same source event always produces the same key.

• Kafka with exactly_once: true: set isolation.level=read_committed on consumers. No application-level dedup needed.
• Kafka without exactly_once: use the event’s id field (UUID v7) or the idempotency key to detect duplicates.
• NATS JetStream: server-side dedup via Nats-Msg-Id header. Configure duplicate_window on the stream to cover your maximum expected downtime (default: 2 minutes).
• Redis Streams: check the idempotency_key field in the stream entry before processing. Use a Redis SET or application-level tracking to remember processed keys.

Dedup window

How long should consumers remember processed event IDs? Match your maximum expected DeltaForge downtime:

Correctness Test Matrix

This matrix maps guarantees to their verification. Rows marked Exists have a test in the suite today; rows marked Planned do not yet - treat those guarantees as design intent pending verification, not as tested behavior.

Limitations

These are not guaranteed and are documented honestly:

• No cross-table global ordering — events from different tables may be interleaved. This is by design; enforcing global order would require single-threaded delivery and cap throughput. Use respect_source_tx: true to preserve ordering within database transactions.
• No stateful stream processing — DeltaForge does not support joins, aggregations, or windowing. For stateful processing, consume DeltaForge’s output with Apache Flink, ksqlDB, or Kafka Streams.
• Dead letter queue - when journal.enabled: true, poison events (serialization/routing failures) are durably appended to the DLQ before the sink acknowledges, so the pipeline advances without losing the row. A row is only treated as isolated once its DLQ append is confirmed. If there is no DLQ, or the DLQ cannot durably capture the row (overflow drop/reject, or a backend write failure), the affected sink does not acknowledge and its own checkpoint is held (fail closed) so restart replay can recover the row. A required sink in that state also stops the pipeline (commit policy unsatisfied); an optional sink does not stop the pipeline, but its checkpoint still stays behind until the row is recovered. The Block overflow policy waits for space but is bounded - it fails closed rather than hanging. Configure a DLQ with a non-dropping overflow policy so poison rows are isolated and every sink keeps moving. See the DLQ page.
• No in-session retry for optional sinks — when required: false and a sink fails, the failed batch is not retried in the same session. The failed sink’s checkpoint stays at its prior position; events are re-delivered only on pipeline restart via source replay. For sinks with realistic outage windows (e.g., S3 throttling, cross-region issues), the operator must weigh source backpressure (required: true) against replay-on-restart latency (required: false). See Required vs. optional sinks.
• S3 sink — at-least-once at file granularity — duplicate events across a crash boundary appear in two files with different ULIDs. Per-row DLQ isolates encoder failures, but does not provide exactly-once delivery on its own; that requires the Phase 2 Iceberg sink (atomic snapshot commits). Consumers must dedup downstream via MERGE INTO or event_id. See S3 sink atomicity guarantees.
• S3 sink — lifecycle policy required for production — DeltaForge does not track multipart upload IDs externally. Abandoned multiparts are reclaimed by the bucket’s AbortIncompleteMultipartUpload lifecycle rule. Without this rule, S3 storage cost accumulates on every aborted batch.
• Snapshot duplicates - snapshot rows can be delivered more than once: PostgreSQL’s bounded snapshot-to-stream overlap, and every row of a snapshot copied again after an interruption or upgrade. Consumers should resolve them by source position or idempotency key. See Initial Snapshots.