Checkpoints

Checkpoints record pipeline progress so ingestion can resume from the last successfully delivered position. DeltaForge guarantees at-least-once delivery by saving checkpoints only after events have been acknowledged by sinks.

Core Guarantee: At-Least-Once Delivery

Checkpoints are only saved after events have been successfully delivered to sinks.

If a checkpoint were saved before delivery, a crash between those two points would silently lose events. DeltaForge prevents this by always checkpointing after sink acknowledgment.

┌─────────────┐     ┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│   Source    │────▶│  Processor  │────▶│    Sink     │────▶│ Checkpoint  │
│   (read)    │     │ (transform) │     │  (deliver)  │     │   (save)    │
└─────────────┘     └─────────────┘     └─────────────┘     └─────────────┘

On crash: events since the last checkpoint are replayed. Consumers should be idempotent or use deduplication.

Storage

Checkpoints are stored via the unified StorageBackend - see Storage for backend configuration. All pipelines share the same storage backend; each pipeline’s checkpoint is keyed by its source ID.

Checkpoint Contents

MySQL

Tracks binlog position:

file:     binlog.000042
pos:      12345
gtid_set: (optional, if GTID replication is enabled)

PostgreSQL

Tracks replication stream LSN:

lsn:    0/1A2B3C4D
tx_id:  (optional)

Checkpoint-Schema Correlation

When a schema change is detected, the current checkpoint position is recorded alongside the new schema version. This ensures that during replay, each event is interpreted with the schema that was active when it was produced - even if the table has since been altered.

Commit Policy

When multiple sinks are configured, the commit policy controls when the checkpoint advances:

Set required: true only on sinks where delivery is mandatory for correctness. Optional sinks can fail without blocking the checkpoint.

Operations

Inspecting Checkpoints

sqlite3 ./data/deltaforge.db \
  "SELECT key, length(val) FROM df_kv WHERE ns = 'checkpoints';"

Resetting a Pipeline

Checkpoints are never edited or deleted by hand: a deleted or rewritten checkpoint can skip or lose changes, and the snapshot and continuity state that refers to it would no longer match. To copy a source’s data again, use the reviewed, audited resnapshot recovery operation (it keeps every checkpoint and moves each sink into the new snapshot through its start barrier). To start a source from scratch, configure a new source id (it starts with a snapshot).

MySQL GTID state from earlier versions

Before exact GTID sets, a MySQL GTID checkpoint could claim transactions the server had not executed (a set with a gap, u:1-8:13, was stored as u:1-13), and resuming from it skipped them once they were committed. Such state cannot be repaired later, so this version refuses it: a stored MySQL GTID checkpoint without gtid_format: 2 does not load, older MySQL GTID durable watermarks (before version 3) are incomparable, and older activation records and failover anchors are refused. The source stops with an error naming the cause. Configure it under a new source id to start it again with a snapshot. File/position checkpoints are unaffected.

Best Practices

• Back up deltaforge.db regularly - it contains both checkpoints and schema history
• Monitor checkpoint lag to detect stuck pipelines
• Use smaller batch sizes for more frequent checkpoints at the cost of throughput
• Test recovery by killing the process and verifying no events are lost or duplicated