Introduction

DeltaForge is a versatile, high-performance Change Data Capture (CDC) engine built in Rust. It streams database changes into downstream systems like Kafka, Redis, NATS, HTTP endpoints, and S3 object storage - giving you full control over how events are routed, transformed, and delivered. Supports JSON and Avro encoding (with Confluent Schema Registry), transactional atomic-batch delivery via Kafka transactions (at-least-once across restart; consumers dedup on event id for exactly-once end to end), and built-in schema discovery that automatically infers and tracks the shape of your data as it flows through.

Pipelines are defined declaratively in YAML, making it straightforward to onboard new use cases without custom code.

Built with	Sources	Processors	Sinks	Output Formats
Rust	MySQL · PostgreSQL	JavaScript · Outbox · Flatten · Filter	Kafka · Redis · NATS · HTTP · S3

Why DeltaForge?

Core Capabilities

• ⚡ Powered by Rust : Predictable performance, memory safety, and minimal resource footprint.
• 🔌 Pluggable architecture : Sources, processors, and sinks are modular and independently extensible.
• 🧩 Declarative pipelines : Define sources, transforms, sinks, and commit policies in version-controlled YAML with environment variable expansion for secrets.
• 📦 Reliable checkpointing : Per-sink independent checkpoints. All sinks at-least-once; Kafka transactions add atomic-batch delivery (dedup on event id for exactly-once end to end). At-least-once with dedup for NATS and Redis.
• 🔁 Avro encoding : Confluent wire format with Schema Registry. DDL-derived schemas with exact types and nullability. Safe defaults for unsigned integers, enums, and timestamps.
• 🪦 Dead letter queue : Poison events routed to DLQ instead of blocking the pipeline. REST API for inspection, filtering, and replay.
• 🔀 Dynamic routing : Route events to per-table topics, streams, or subjects using templates or JavaScript logic.
• 📤 Transactional outbox : Publish domain events atomically with database writes. Per-aggregate routing, raw payload delivery, zero polling.
• 🛠️ Cloud-native ready : Single binary, Docker images, JSON logs, Prometheus metrics, and liveness/readiness probes for Kubernetes.

Schema Intelligence

• 🔍 Schema sensing : Automatically infer and track schema from event payloads, including deep inspection of nested JSON structures.
• 🗺️ High-cardinality handling : Detect and normalize dynamic map keys (session IDs, trace IDs) to prevent false schema evolution events.
• 🏷️ Schema fingerprinting : SHA-256 based change detection with schema-to-checkpoint correlation for reliable replay.
• 🗃️ Source-owned semantics : Preserves native database types (PostgreSQL arrays, MySQL JSON, etc.) instead of normalizing to a universal type system.

Operational Features

• 🔄 Graceful failover : Handles source failover with automatic schema revalidation - no manual intervention needed.
• 🧬 Zero-downtime schema evolution : Detects DDL changes and reloads schemas automatically, no pipeline restart needed.
• 🎯 Flexible table selection : Wildcard patterns (db.*, schema.prefix%) for easy onboarding.
• 📀 Transaction boundaries : Optionally keep source transactions intact across batches.
• ⚙️ Commit policies : Control checkpoint behavior with all, required, or quorum modes across multiple sinks.
• 🔧 Live pipeline management : Pause, resume, patch, and inspect running pipelines via the REST API.
• 🗄️ Safe initial snapshot : Consistent parallel backfill of existing tables before streaming begins, with binlog/WAL retention validation, background guards, and crash-resume at table granularity.

Use Cases

DeltaForge is designed for:

• Real-time data synchronization : Keep caches, search indexes, and analytics systems in sync with your primary database.
• Event-driven architectures : Stream database changes to Kafka or NATS for downstream microservices.
• Transactional messaging : Use the outbox pattern to publish domain events atomically with database writes - no distributed transactions needed.
• Audit trails and compliance : Capture every mutation with full before/after images for SOC2, HIPAA, or GDPR requirements.
• Lightweight ETL : Transform, filter, and route data in-flight with JavaScript processors - no Spark or Flink cluster needed.

DeltaForge is not a DAG-based stream processor. It is a focused CDC engine meant to replace tools like Debezium when you need a lighter, cloud-native, and more customizable runtime.

Getting Started

Quick Example

metadata:
  name: orders-to-kafka
  tenant: acme

spec:
  source:
    type: mysql
    config:
      dsn: ${MYSQL_DSN}
      tables: [shop.orders]

  processors:
    - type: javascript
      inline: |
        function processBatch(events) {
          for (const event of events) {
            event.processed_at = Date.now();
          }
          return events;
        }

  sinks:
    - type: kafka
      config:
        brokers: ${KAFKA_BROKERS}
        topic: order-events

Installation

Docker (recommended):

docker pull ghcr.io/vnvo/deltaforge:latest

From source:

git clone https://github.com/vnvo/deltaforge.git
cd deltaforge
cargo build --release

See the Development Guide for detailed build instructions and available Docker image variants.

License

DeltaForge is dual-licensed under MIT and Apache 2.0.

Change Data Capture (CDC)

Change Data Capture (CDC) is the practice of streaming database mutations as ordered events so downstream systems can stay in sync without periodic full loads. Rather than asking “what does my data look like now?”, CDC answers “what changed, and when?”

DeltaForge is a CDC engine built to make log-based change streams reliable, observable, and operationally simple in modern, containerized environments. It focuses on row-level CDC from MySQL binlog and Postgres logical replication to keep consumers accurate and latency-aware while minimizing impact on source databases.

In short: DeltaForge tails committed transactions from MySQL and Postgres logs, preserves ordering and transaction boundaries, and delivers events to Kafka and Redis with checkpointed delivery, configurable batching, and Prometheus metrics - without requiring a JVM or distributed coordinator.

Why CDC matters

Traditional data integration relies on periodic batch jobs that query source systems, compare snapshots, and push differences downstream. This approach worked for decades, but modern architectures demand something better.

The batch ETL problem: A nightly sync means your analytics are always a day stale. An hourly sync still leaves gaps and hammers your production database with expensive SELECT * queries during business hours. As data volumes grow, these jobs take longer, fail more often, and compete for the same resources your customers need.

CDC flips the model. Instead of pulling data on a schedule, you subscribe to changes as they happen.

These benefits compound as systems scale and teams decentralize:

• Deterministic replay: Ordered events with stable idempotency keys let consumers reconstruct state and dedup replays, so at-least-once delivery becomes effectively exactly-once at the consumer.
• Polyglot delivery: The same change feed can serve caches, queues, warehouses, and search indexes simultaneously without additional source queries.

How CDC works

All CDC implementations share a common goal: detect changes and emit them as events. The approaches differ in how they detect those changes.

Log-based CDC

Databases maintain transaction logs (MySQL binlog, Postgres WAL) that record every committed change for durability and replication. Log-based CDC reads these logs directly, capturing changes without touching application tables.

┌─────────────┐    commits    ┌─────────────┐    tails     ┌─────────────┐
│ Application │──────────────▶│  Database   │─────────────▶│ CDC Engine  │
└─────────────┘               │   + WAL     │              └──────┬──────┘
                              └─────────────┘                     │
                                                                  ▼
                                                           ┌─────────────┐
                                                           │   Kafka /   │
                                                           │   Redis     │
                                                           └─────────────┘

Advantages:

• Zero impact on source table performance
• Captures all changes including those from triggers and stored procedures
• Preserves transaction boundaries and ordering
• Can capture deletes without soft-delete columns

Trade-offs:

• Requires database configuration (replication slots, binlog retention)
• Schema changes need careful handling
• Log retention limits how far back you can replay

DeltaForge uses log-based CDC exclusively. This allows DeltaForge to provide stronger ordering guarantees, lower source impact, and simpler operational semantics than hybrid approaches that mix log tailing with polling or triggers.

Trigger-based CDC

Database triggers fire on INSERT, UPDATE, and DELETE operations, writing change records to a shadow table that a separate process polls.

Advantages:

• Works on databases without accessible transaction logs
• Can capture application-level context unavailable in logs

Trade-offs:

• Adds write overhead to every transaction
• Triggers can be disabled or forgotten during schema migrations
• Shadow tables require maintenance and can grow unbounded

Polling-based CDC

A process periodically queries tables for rows modified since the last check, typically using an updated_at timestamp or incrementing ID.

Advantages:

• Simple to implement
• No special database configuration required

Trade-offs:

• Cannot reliably detect deletes
• Requires updated_at columns on every table
• Polling frequency trades off latency against database load
• Clock skew and transaction visibility can cause missed or duplicate events

Anatomy of a CDC event

A well-designed CDC event contains everything downstream consumers need to process the change correctly.

{
  "id": "evt_01J7K9X2M3N4P5Q6R7S8T9U0V1",
  "source": {
    "database": "shop",
    "table": "orders",
    "server_id": "mysql-prod-1"
  },
  "operation": "update",
  "timestamp": "2025-01-15T14:32:01.847Z",
  "transaction": {
    "id": "gtid:3E11FA47-71CA-11E1-9E33-C80AA9429562:42",
    "position": 15847293
  },
  "before": {
    "id": 12345,
    "status": "pending",
    "total": 99.99,
    "updated_at": "2025-01-15T14:30:00.000Z"
  },
  "after": {
    "id": 12345,
    "status": "shipped",
    "total": 99.99,
    "updated_at": "2025-01-15T14:32:01.000Z"
  },
  "schema_version": "v3"
}

Key components:

Not all fields are present for every operation-before is omitted for INSERTs, after is omitted for DELETEs - but the event envelope and metadata fields are consistent across MySQL and Postgres sources.

Real-world use cases

Cache invalidation and population

Problem: Your Redis cache serves product catalog data, but cache invalidation logic is scattered across dozens of services. Some forget to invalidate, others invalidate too aggressively, and debugging staleness issues takes hours.

CDC solution: Stream changes from the products table to a message queue. A dedicated consumer reads the stream and updates or deletes cache keys based on the operation type. This centralizes invalidation logic, provides replay capability for cache rebuilds, and removes cache concerns from application code entirely.

┌──────────┐     CDC      ┌─────────────┐   consumer   ┌─────────────┐
│  MySQL   │─────────────▶│   Stream    │─────────────▶│ Cache Keys  │
│ products │              │  (Kafka/    │              │ product:123 │
└──────────┘              │   Redis)    │              └─────────────┘
                          └─────────────┘

Event-driven microservices

Problem: Your order service needs to notify inventory, shipping, billing, and analytics whenever an order changes state. Direct service-to-service calls create tight coupling and cascade failures.

CDC solution: Publish order changes to a durable message queue. Each downstream service subscribes independently, processes at its own pace, and can replay events after failures or during onboarding. The order service doesn’t need to know about its consumers.

┌─────────────┐     CDC      ┌─────────────┐
│   Orders    │─────────────▶│   Message   │
│   Database  │              │   Queue     │
└─────────────┘              └──────┬──────┘
                    ┌───────────────┼───────────────┐
                    ▼               ▼               ▼
             ┌───────────┐   ┌───────────┐   ┌───────────┐
             │ Inventory │   │ Shipping  │   │ Analytics │
             └───────────┘   └───────────┘   └───────────┘

Search index synchronization

Problem: Your Elasticsearch index drifts from the source of truth. Full reindexing takes hours and blocks search updates during the process.

CDC solution: Stream changes continuously to keep the index synchronized. Use the before image to remove old documents and the after image to index new content.

Data warehouse incremental loads

Problem: Nightly full loads into your warehouse take 6 hours and block analysts until noon. Business users complain that dashboards show yesterday’s numbers.

CDC solution: Stream changes to a staging topic, then micro-batch into your warehouse every few minutes. Analysts get near real-time data without impacting source systems.

Audit logging and compliance

Problem: Regulations require you to maintain a complete history of changes to sensitive data. Application-level audit logging is inconsistent and can be bypassed.

CDC solution: The transaction log captures every committed change regardless of how it was made - application code, admin scripts, or direct SQL. Stream these events to immutable storage for compliance.

Cross-region replication

Problem: You need to replicate data to a secondary region for disaster recovery, but built-in replication doesn’t support the transformations you need.

CDC solution: Stream changes through a CDC pipeline that filters, transforms, and routes events to the target region’s databases or message queues.

Architecture patterns

The outbox pattern

When you need to update a database and record the intent to publish an event atomically, the outbox pattern ties the event to the same transaction as the business write - the event is captured if and only if the transaction commits, with no dual-write gap and no distributed transaction. Delivery of that captured event downstream is still at-least-once; consumers dedup on the event id.

┌─────────────────────────────────────────┐
│              Single Transaction         │
│  ┌─────────────┐    ┌─────────────────┐ │
│  │ UPDATE      │    │ INSERT INTO     │ │
│  │ orders SET  │ +  │ outbox (event)  │ │
│  │ status=...  │    │                 │ │
│  └─────────────┘    └─────────────────┘ │
└─────────────────────────────────────────┘
                      │
                      ▼ CDC
               ┌─────────────┐
               │   Kafka     │
               └─────────────┘

1. The application writes business data and an event record in the same transaction.
2. CDC tails the outbox table and publishes events to Kafka.
3. A cleanup process (or CDC itself) removes processed outbox rows.

This guarantees that events are published if and only if the transaction commits.

When to skip the outbox: If your only requirement is to react to committed database state (not application intent), direct CDC from business tables is often simpler than introducing an outbox. The outbox pattern adds value when you need custom event payloads, explicit event versioning, or when the event schema differs significantly from table structure.

Event sourcing integration

CDC complements event sourcing by bridging legacy systems that weren’t built event-first. Stream changes from existing tables into event stores, then gradually migrate to native event sourcing.

CQRS (Command Query Responsibility Segregation)

CDC naturally supports CQRS by populating read-optimized projections from the write model. Changes flow through the CDC pipeline to update denormalized views, search indexes, or cache layers.

Challenges and solutions

Schema evolution

Databases change. Columns get added, renamed, or removed. Types change. CDC pipelines need to handle this gracefully.

Strategies:

• Schema registry: Store and version schemas centrally (e.g., Confluent Schema Registry with Avro).
• Forward compatibility: Add columns as nullable; avoid removing columns that consumers depend on.
• Consumer tolerance: Design consumers to ignore unknown fields and handle missing optional fields.
• Processor transforms: Use DeltaForge’s JavaScript processors to normalize schemas before sinks.

Ordering guarantees

Events must arrive in the correct order for consumers to reconstruct state accurately. A DELETE arriving before its corresponding INSERT would be catastrophic.

DeltaForge guarantees:

• Source-order preservation per table partition (always enabled)
• Transaction boundary preservation when respect_source_tx: true is configured in batch settings

Kafka sink uses consistent partitioning by primary key to maintain ordering within a partition at the consumer.

Delivery guarantees

Network failures, process crashes, and consumer restarts can cause duplicates. DeltaForge delivers at-least-once and gives every event a stable idempotency key so consumers can collapse duplicates into an effectively exactly-once result.

DeltaForge approach:

• Checkpoints track the last committed position in the source log.
• Configurable commit policies (all, required, quorum) control when checkpoints advance.
• Kafka: transactional atomic-batch delivery via the transactional producer (exactly_once: true); consumers set isolation.level=read_committed to avoid partial batches. Still at-least-once across restart - a replayed batch is a new transaction - so consumers dedup on event id.
• NATS: at-least-once with server-side dedup via Nats-Msg-Id header within duplicate_window.
• Redis: at-least-once with consumer-side dedup via idempotency_key field.

Default behavior: DeltaForge provides at-least-once delivery out of the box. Kafka exactly_once: true adds transactional atomic-batch visibility on top; it is not end-to-end exactly-once on its own. See the Guarantees page for the full delivery tier matrix.

High availability

Production CDC pipelines need to handle failures without data loss or extended downtime.

Best practices:

• Run multiple pipeline instances with leader election.
• Store checkpoints in durable storage (DeltaForge persists to local files, mountable volumes in containers).
• Monitor lag between source position and checkpoint position.
• Set up alerts for pipeline failures and excessive lag.

Expectations: DeltaForge checkpoints ensure no data loss on restart, but does not currently include built-in leader election. For HA deployments, use external coordination (Kubernetes leader election, etcd locks) or run active-passive with health-check-based failover.

Backpressure

When sinks can’t keep up with the change rate, pipelines need to slow down gracefully rather than dropping events or exhausting memory.

DeltaForge handles backpressure through:

• Configurable batch sizes (max_events, max_bytes, max_ms).
• In-flight limits (max_inflight) that bound concurrent sink writes.
• Blocking reads from source when batches queue up.

Performance considerations

Batching trade-offs

Start with DeltaForge defaults and tune based on observed latency and throughput.

Source database impact

Log-based CDC has minimal impact, but consider:

• Replication slot retention: Paused pipelines cause WAL/binlog accumulation.
• Connection limits: Each pipeline holds a replication connection.
• Network bandwidth: High-volume tables generate significant log traffic.

Sink throughput

• Kafka: Tune batch.size, linger.ms, and compression in client_conf.
• Redis: Use pipelining and connection pooling for high-volume streams.

Monitoring and observability

CDC pipelines are long-running, stateful processes; without metrics and alerts, failures are silent by default. A healthy pipeline requires visibility into lag, throughput, and errors.

Key metrics to track

DeltaForge exposes Prometheus metrics on the configurable metrics endpoint (default :9000).

Health checks

• GET /health: Liveness probe - is the process running?
• GET /ready: Readiness probe - are pipelines connected and processing?
• GET /pipelines: Detailed status of each pipeline including configuration.

Choosing a CDC solution

When evaluating CDC tools, consider:

Where DeltaForge fits

DeltaForge intentionally avoids the operational complexity of JVM-based CDC stacks (Kafka Connect-style deployments with Zookeeper, Connect workers, and converter configurations) while remaining compatible with Kafka-centric architectures.

DeltaForge is designed for teams that want:

• Lightweight runtime: Single binary, minimal memory footprint, no JVM warmup.
• Config-driven pipelines: YAML specs instead of code for common patterns.
• Inline transformation: JavaScript processors for custom logic without recompilation.
• Container-native operations: Built for Kubernetes with health endpoints and Prometheus metrics.

DeltaForge is not designed for:

• Complex DAG-based stream processing with windowed aggregations
• Stateful joins across multiple streams
• Sources beyond MySQL and Postgres (currently)
• Built-in schema registry integration (use external registries)

If you need those capabilities, consider dedicated stream processors or the broader Kafka ecosystem. DeltaForge excels at getting data out of databases and into your event infrastructure reliably and efficiently.

Getting started

Ready to try CDC with DeltaForge? Head to the Quickstart guide to run your first pipeline in minutes.

For production deployments, review the Development guide for container builds and operational best practices.

Quickstart

Get DeltaForge running in minutes.

1. Prepare a pipeline spec

Create a YAML file that defines your CDC pipeline. Environment variables are expanded at runtime, so secrets stay out of version control.

metadata:
  name: orders-mysql-to-kafka
  tenant: acme

spec:
  source:
    type: mysql
    config:
      id: orders-mysql
      dsn: ${MYSQL_DSN}
      tables:
        - shop.orders

  processors:
    - type: javascript
      id: transform
      inline: |
        function processBatch(events) {
          for (const event of events) {
            event.tags = ["processed"];
          }
          return events;
        }

  sinks:
    - type: kafka
      config:
        id: orders-kafka
        brokers: ${KAFKA_BROKERS}
        topic: orders
        required: true

  batch:
    max_events: 500
    max_bytes: 1048576
    max_ms: 1000

2. Start DeltaForge

Using Docker (recommended):

docker run --rm \
  -p 8080:8080 -p 9000:9000 \
  -e MYSQL_DSN="mysql://user:pass@host:3306/db" \
  -e KAFKA_BROKERS="kafka:9092" \
  -v $(pwd)/pipeline.yaml:/etc/deltaforge/pipeline.yaml:ro \
  -v deltaforge-checkpoints:/app/data \
  ghcr.io/vnvo/deltaforge:latest \
  --config /etc/deltaforge/pipeline.yaml

From source:

cargo run -p runner -- --config ./pipeline.yaml

Runner options

3. Verify it’s running

Check health and pipeline status:

# Liveness probe
curl http://localhost:8080/health

# Readiness with pipeline status
curl http://localhost:8080/ready

# List all pipelines
curl http://localhost:8080/pipelines

4. Manage pipelines

Control pipelines via the REST API:

# Pause a pipeline
curl -X POST http://localhost:8080/pipelines/orders-mysql-to-kafka/pause

# Resume a pipeline
curl -X POST http://localhost:8080/pipelines/orders-mysql-to-kafka/resume

# Stop a pipeline
curl -X POST http://localhost:8080/pipelines/orders-mysql-to-kafka/stop

Next steps

• CDC Overview : Understand how Change Data Capture works
• Configuration : Full pipeline spec reference
• Sources : MySQL and Postgres setup
• Sinks : Kafka, Redis, and NATS configuration
• Dynamic Routing : Route events to per-table destinations
• Envelopes : Native, Debezium, and CloudEvents output formats
• Development : Build from source, run locally

Production Quick Start

A minimal, production-shaped deployment on Kubernetes using the bundled Helm chart. It uses the supported single-instance topology, persistent storage, referenced secrets, readiness/liveness probes, explicit resource limits, and the preflight command as a deployment gate.

This is a starting point, not a configuration reference - see Configuration for every option and the Capacity & Resource Envelope for sizing.

Prerequisites

• A Kubernetes cluster with a default StorageClass (for the checkpoint volume).
• A reachable source database, already provisioned per Supported Deployment Envelope → Required privileges. For PostgreSQL, create the publication (DeltaForge creates and owns the slot).
• Source (and sink) credentials in a Kubernetes Secret - referenced, never inlined in the config.

1. Create the credentials Secret

Keep secrets out of the pipeline config; the config references them with ${VAR}.

kubectl create secret generic mysql-creds \
  --from-literal=MYSQL_USER=cdc_user \
  --from-literal=MYSQL_PASSWORD=s3cret

2. Write a values file

# values.prod.yaml
replicaCount: 1                     # single instance (required)

pipeline:
  config: |
    apiVersion: deltaforge/v1
    kind: Pipeline
    metadata:
      name: orders
      tenant: default
    spec:
      source:
        type: mysql
        config:
          id: orders-src
          dsn: "mysql://mysql-primary:3306/orders"
          credentials:                # typed secret references (protected resolution)
            username:
              provider: env
              location: MYSQL_USER
            password:
              provider: env
              location: MYSQL_PASSWORD
          tables: ["orders.*"]
      sinks:
        - type: kafka
          config:
            id: orders-kafka
            brokers: kafka:9092
            topic: cdc.orders
            exactly_once: true
      commit_policy:
        mode: required
      journal:
        enabled: true               # DLQ on, so a poison event never blocks

secrets:
  existingSecrets:                  # referenced, not created from values
    - name: mysql-creds

persistence:
  enabled: true                    # durable checkpoints + DLQ (SQLite on a PVC)
  size: 5Gi

preflight:
  enabled: true                    # gate the deploy on preflight (default)

resources:
  requests: { cpu: 250m, memory: 512Mi }
  limits:   { cpu: "2",  memory: 2Gi }   # no aggregate memory cap exists; size per Capacity Envelope

Probes (/health liveness, /ready readiness) and the single-instance StatefulSet come from the chart defaults; you do not need to set them.

The deployment is gated on preflight automatically: with preflight.enabled: true the chart runs deltaforge preflight as an initContainer - with the same config, secrets, and --storage-* settings as the pipeline container - before the pipeline starts. A failing check fails the pod, so a misconfigured deployment never begins streaming, and it re-runs on every rollout.

3. Install

helm install creates the ConfigMap and Secret wiring and runs the preflight initContainer as the authoritative deployment gate:

helm install orders ./deploy/helm/deltaforge -f values.prod.yaml

4. Verify

# The preflight initContainer must complete before the app container starts.
kubectl get pods -l app.kubernetes.io/instance=orders
kubectl logs orders-0 -c preflight        # the preflight report (the deploy gate)

# Readiness gates traffic; liveness restarts a wedged process:
kubectl get pod orders-0 -o jsonpath='{.status.conditions[?(@.type=="Ready")].status}'

# Health/metrics:
kubectl port-forward svc/orders 8080:8080 &
curl -fsS localhost:8080/ready

If preflight fails, the pod stays in Init and kubectl logs orders-0 -c preflight shows the hard errors to fix before the pipeline can start.

Notes

• Single instance is required. The guarantee that only one process holds the state store comes from replicaCount: 1, the StatefulSet’s one-pod-at-a-time behavior, and the documented prohibition against a second installation against the same store - not from ReadWriteOnce (RWO restricts volume attachment to one node, but multiple pods on that node could still mount it). Do not scale the replica count or point a second install at the same volume - see the single-owner rule.
• Persistent storage. SQLite checkpoints and the DLQ live on the PVC. Treat the state volume as durable production data. Back it up using the documented SQLite-safe procedure. If it is lost, stop the pipeline and recover the state or explicitly reinitialize/re-snapshot; do not assume automatic checkpoint recovery (a lost volume loses the durable checkpoint, and exact recovery may be impossible once source logs expire). For a shared store, use the postgres storage backend (still single-instance).
• Upgrades. A single-replica StatefulSet terminates the old pod before starting the new one; the preflight initContainer re-gates every rollout.

Supported Deployment Envelope

This page defines the bounded support envelope for the current DeltaForge release: the configurations DeltaForge is validated for, what operators must provide, and what is explicitly out of scope. Anything not listed here as supported should be treated as unvalidated - test it in a disposable environment first, and talk to us before relying on it.

The delivery and correctness guarantees referenced below are defined in Guarantees & Correctness. Read that page together with this one.

Build and freeze

• Pin your deployment to a specific release commit or tag rather than tracking main, so your environment does not drift.
• Toolchain: built with the Rust stable toolchain, edition 2024. Single static binary; no JVM, no runtime GC.
• Report issues against the exact commit/tag so we can reproduce them.

Supported versions

Other sinks (Redis, NATS, HTTP, S3-compatible object storage, ClickHouse, Elasticsearch) are supported as documented on their per-sink pages; run your target versions through a disposable pipeline during onboarding.

Topology: single owner per source

Run one DeltaForge instance per source (per replication slot / binlog reader). DeltaForge does not yet provide a cluster-wide lock or lease that prevents two instances from being started against the same source; single ownership is enforced at the slot and producer level, not globally.

Single-instance requirement. For the supported single-instance deployment, run exactly one DeltaForge process against a given checkpoint/state store. Within one process, lifecycle operations (start, stop, delete, patch, resume) are serialized and a durable per-source-id claim rejects a second pipeline that reuses an active source id, so two pipelines can never share a source id and corrupt each other’s checkpoints (checkpoints are keyed by source id). Across processes, the store gate makes a second server against the same state store refuse to start. Deploy DeltaForge as a single instance (a single replica; if orchestrated, replicas: 1 with strategy: Recreate, not rolling), and do not point a second process at the same state store.

• PostgreSQL: the replication slot has durable, DeltaForge-recorded ownership. DeltaForge only drops or recreates a slot it can prove it owns, whose lineage matches, and that is inactive; a foreign, ambiguously owned, or active slot fails closed with remediation rather than being taken over. A slot created outside DeltaForge (no ownership record) fails closed on re-snapshot - drop it and let DeltaForge recreate it, or run snapshot.mode = never.
• Kafka: with exactly_once: true, a second producer using the same transactional.id fences the first. Fencing is a fatal error that stops the pipeline. This is a safety net, not a substitute for running a single instance.
• MySQL: each instance derives its replication server_id from the source id; give each source a unique id to avoid server_id collisions on the same MySQL server.
• Failover: for planned HA, use GTID (MySQL) and a slot-aware HA tool (for example Patroni permanent_slots) so the replication position survives a primary change. See Failover Handling.

Deployment preflight

Before deploying a pipeline, validate its config against the live source:

deltaforge preflight <config-file-or-dir>      # human-readable report
deltaforge preflight <config-file-or-dir> --json   # machine-readable report

Preflight resolves the config’s secrets and source DSN exactly as startup does, connects to the source, and runs the source checks plus local config validation. It exits non-zero (fail closed) if any check fails, so it is safe to gate a deploy on it in CI/CD.

• PostgreSQL: wal_level = logical; the publication exists and has tables; WAL-retention capacity vs estimated snapshot size. Slot validation is mode-aware: DeltaForge creates and owns the replication slot in every supported mode, so an absent slot on a fresh deployment is reported as informational, not a failure. An existing slot is checked under the same ownership rules as startup - a slot that is invalidated, WAL-lost, currently active (another consumer), or foreign (no matching ownership record for this server/database) fails closed.
• MySQL: log_bin on, binlog_format = ROW, gtid_mode = ON, the RELOAD privilege (when a snapshot will run), captured tables on InnoDB, binlog retention window.
• Both: connectivity; source and sink credential references are resolved (a missing/invalid secret or an inline+reference conflict fails here); commit-policy validity vs sink count; at least one sink configured; and source ids are unique across the supplied config(s).

Wildcard table patterns are reported as not validated per-table (server-level checks still run). Preflight validates credentials and configuration, not endpoints: sink endpoint reachability is not probed (a documented follow-up), and the slot-ownership check requires pointing preflight at the deployment’s storage backend (the same --storage-* flags the server uses).

Store gate

Breaking change on upgrade. Earlier releases let several processes share one PostgreSQL-backed state store. That topology now refuses to start; give each process its own store. Upgrade by stopping the old version cleanly before starting the new one; if the old process crashed instead, release its gate with deltaforge store-gate break as described below.

The state store carries one durable store gate. A DeltaForge server acquires it at startup, before it reads any schema history, and holds it until it shuts down cleanly; deltaforge schema-migrate --apply holds it while it writes. Only one holder can exist, so a second server, or a migration apply while a server runs, refuses to start with an error that names the holder (role, owner id, host, pid, since) and the command to release it.

• Clean shutdown (SIGTERM/SIGINT) stops the API, stops every pipeline and waits for its tasks, then releases the gate. This also holds for a signal that arrives while the server is still starting: the signal handlers are installed before the gate is acquired, and startup stops at the next step. Give the process enough time to do this (the Helm chart’s terminationGracePeriodSeconds is 30).

• A crash keeps the gate held. Nothing expires. After an OOM kill, a SIGKILL (including one sent after the grace period), or a host loss, the next start fails until an operator releases the gate. In Kubernetes this shows as the restarted container exiting with the store-gate error (CrashLoopBackOff). This is deliberate: DeltaForge will not guess that the previous holder is gone.

• Recovery:

  deltaforge store-gate status            # who holds it (add --json for scripts)
  deltaforge store-gate break --owner <owner-id>
  

  Before breaking, confirm that the recorded process (host and pid) is no longer running. break releases the gate only if that owner still holds it, so a stale or mistyped owner id changes nothing. Pass the same --storage-backend/--storage-path/--storage-dsn flags the server uses.

• A migration’s gate cannot be broken. A schema migration that failed or crashed may have written part of its plan, and a server must never read partially migrated history, so store-gate break refuses it. Finish the migration instead (see below); the error message prints the exact command.

Migrating pre-upgrade schema history

Schema history written by earlier releases is kept but not used automatically: a table’s history is only trusted once it is tied to the source’s verified lineage (PostgreSQL system identifier and database OID; MySQL server UUID). deltaforge schema-migrate adopts it into the lineage-scoped history for tables you list explicitly. It never deletes the old records.

1. Start the pipeline once on the new release so it records the source lineage, then stop the server. deltaforge schema-migrate needs the lineage hash of each source; the dry run prints the recorded one.

2. Write a mapping file that lists each table (no wildcards):

   mappings:
     - tenant: acme
       source_id: orders-pg
       lineage_hash: <recorded lineage hash>
       tables:
         - { db: public, table: orders }
         - { db: public, table: order_items }
   

3. Dry run (the default; read-only, safe while the server runs):

   deltaforge schema-migrate --mapping mapping.yaml [--tenant T] [--source S] [--json]
   

   It classifies every table as migrate, already migrated, ambiguous or rejected (with the reason: lineage missing or different from the asserted hash, empty or corrupt history, a version that conflicts with existing history, or a marker from a different migration), and prints a proof digest over everything it would adopt.

4. With the server stopped, apply exactly that plan:

   deltaforge schema-migrate --mapping mapping.yaml --apply --expect-proof <proof>
   

   The apply recomputes the plan under the store gate and refuses if the proof differs, so anything that changed since the review stops it (nothing is written and the gate is released). Tables already migrated by a different reviewed proof are reported as rejected, with that migration’s details, rather than counted as done.

5. If the apply fails or is interrupted, the store gate stays held (role migration, recording the proof) and the server cannot start, because part of the plan may already be written. After confirming the failed process is no longer running and fixing the cause, finish the same plan:

   deltaforge schema-migrate --mapping mapping.yaml --apply --expect-proof <proof> --resume-owner <owner-id>
   

   The owner id is in the error message and in deltaforge store-gate status. The resume takes the gate over in one step (it is never unlocked in between), accepts only the recorded owner and the same proof, treats the work already done as progress, and releases the gate once the migration completes.

Required privileges

PostgreSQL

• A role with LOGIN REPLICATION.
• GRANT CONNECT ON DATABASE <db>.
• GRANT USAGE ON SCHEMA <schema> and GRANT SELECT on the tables to be captured (snapshot reads and schema introspection).
• pg_hba.conf must permit both regular and replication connections for the role/host.
• The publication is not auto-created - create it before starting the pipeline. The replication slot is created by DeltaForge.
• REPLICA IDENTITY FULL is recommended on captured tables for complete before-images.

MySQL

• REPLICATION SLAVE, REPLICATION CLIENT ON *.* (always).
• BACKUP_ADMIN ON *.* and SELECT ON performance_schema.log_status for a first start without a checkpoint and without a snapshot: the capture boundary (binlog position and GTID set) is read there in one consistent query. Without them that start fails closed. Validated on MySQL 8.4; managed services are not yet qualified for these grants.
• SELECT, SHOW VIEW on the captured database(s) for reads and introspection.
• RELOAD ON *.* only when an initial snapshot runs (needed for the brief FLUSH TABLES WITH READ LOCK that anchors the snapshot). A CDC-only pipeline (snapshot.mode = never) does not need RELOAD.
• Server settings enforced fail-closed at snapshot preflight: gtid_mode = ON, enforce_gtid_consistency = ON, binlog_format = ROW, and captured tables on InnoDB. binlog_row_image = FULL is recommended.
• On managed MySQL (for example RDS/Aurora) where RELOAD / FLUSH TABLES WITH READ LOCK is unavailable, the snapshot fails closed with a typed error - there is no silent unsafe fallback. Grant RELOAD, or run snapshot.mode = never and perform the initial load out of band.

Sinks

Sink credentials and auth modes are documented per sink (see Sinks). Where a sink is configured to auto-create its target object (for example ClickHouse tables or Elasticsearch indices/templates), the sink user needs the corresponding create privilege on the target system; otherwise disable auto-create and provision the target yourself.

Supported credential modes

DeltaForge resolves credentials from typed secret references and never stores resolved secrets in serialized config. See Secrets & Credentials.

• Secret providers: env, file, and vault (Vault KV v2). Kubernetes is not a separate provider - inject secrets as env (secretKeyRef) or as files (projected volume) and reference them with the env or file provider.
• Vault: static KV references are supported (the runner must be built with the vault feature; a Vault reference fails closed at startup otherwise). Dynamic Vault database credentials (lease-driven reconnect) are not live in the current release - the lease lifecycle exists but does not yet drive a DSN swap.
• Rotation: treat credential rotation as restart-required. Change the secret, then restart the affected pipeline or the runner for it to take effect. An opt-in live-reconnect path for file-backed source database credentials exists in the code (MySQL requires GTID mode), but it is outside the supported deployment envelope - validate it in a disposable environment before considering it. Environment-variable credentials are process-immutable and never rotate live; sink credentials are restart-required.

Network and TLS

Plan the network on the basis that DeltaForge’s source database connections and its own HTTP endpoints are not encrypted or authenticated in this release. Deploy accordingly.

• Source DB connections are not encrypted. Neither the PostgreSQL nor the MySQL source establishes TLS to the database in this build. Run DeltaForge on a trusted/private network segment with the source database, or place an encrypted tunnel (for example a service mesh sidecar, stunnel, or a cloud private link) between DeltaForge and the database. Do not run source traffic across an untrusted network.
• REST API is unauthenticated and binds all interfaces by default. The default --api-addr is 0.0.0.0:8080 (every interface), so an unrestricted deployment exposes the unauthenticated API on the network. Bind it to loopback (--api-addr 127.0.0.1:8080) where the client is local, or confine it with a firewall / Kubernetes NetworkPolicy; do not expose it publicly.
• Metrics endpoint is unauthenticated and defaults to 0.0.0.0:9000 (all interfaces). Bind it to loopback (--metrics-addr 127.0.0.1:9000) where the scraper is local, or confine it with a firewall/NetworkPolicy. See Observability.
• Sink TLS/auth is supported where the sink provides it: Kafka (SASL + SASL_SSL via client_conf), Elasticsearch (https:// with tls.ca_file, basic/API-key auth), HTTP (https:// with header-based auth), NATS (TLS + credentials/token). Confirm Redis and ClickHouse TLS in your environment before relying on it.

State and backup requirements

DeltaForge keeps all runtime state (checkpoints, schema registry, snapshot progress, DLQ/journal) in one storage backend. See Storage and Checkpoints.

• Backends: sqlite (default; single-instance production) and postgres (a shared storage backend, beta - not yet given the same crash/recovery validation as SQLite). One state store serves exactly one DeltaForge server at a time (enforced by the store gate); the PostgreSQL backend does not make source processing highly available. Use with caution. memory is for testing only and is lost on restart.

• Back up the state store regularly. For SQLite the store is the deltaforge.db file (default under ./data/), which holds both checkpoints and schema history; losing it means losing resume position and schema lineage. The store runs in WAL mode, so do not copy deltaforge.db on its own while DeltaForge is running - committed data may still be in the -wal file, and a bare file copy can be inconsistent. Use one of:

  • SQLite’s online-backup API or VACUUM INTO 'backup.db' against the live database;
  • stop DeltaForge cleanly, checkpoint the WAL (PRAGMA wal_checkpoint(TRUNCATE)), then copy the file;
  • a storage-level consistent snapshot that captures the database together with its -wal and -shm files.

  For the PostgreSQL backend, back up that database with your normal PostgreSQL backup tooling (pg_dump or a consistent base backup).

• Durability: the SQLite store runs in WAL mode with synchronous = NORMAL and survives SIGKILL without a graceful shutdown; a checkpoint is persisted only after the sink acknowledges delivery and the commit policy is satisfied (the basis of at-least-once).

• Restoring from a backup rewinds to that backup’s position; expect at-least-once re-delivery (duplicates) of everything after the backup point, which consumers dedup on the event id.

Known limitations

These are documented in full on Guarantees & Correctness; the summary:

• Delivery is at-least-once. No sink is end-to-end exactly-once. Kafka exactly_once: true adds transactional atomic-batch delivery (a read_committed consumer never sees a partial batch) but is still at-least-once across a restart. Consumers must dedup on the event id for exactly-once end to end.
• Snapshot to CDC has a bounded at-least-once overlap - rows committed between the anchor and the snapshot export are delivered by both the snapshot and CDC. Current-state sinks converge; append-only sinks see duplicates.
• No cross-table global ordering. Per-primary-key ordering within a table is guaranteed; different tables may interleave.
• No stateful stream processing (no joins/aggregations/windowing). Do that downstream (Flink, ksqlDB, Kafka Streams).
• Optional sinks (required: false) have no in-session retry. A failed optional sink is re-delivered only on the next restart via source replay; a chronic optional-sink outage accumulates a growing replay gap.
• S3 sink is at-least-once at file granularity and requires an AbortIncompleteMultipartUpload bucket lifecycle policy in production to reclaim orphaned multiparts.
• DLQ must be enabled (journal.enabled: true) to isolate poison events; without it a single unprocessable event blocks the pipeline.
• PostgreSQL start_position is not implemented - a newly created slot always starts at its consistent point (the current WAL position when it is created).
• Changing a pipeline’s sink set via PATCH is disabled. Adding or removing a sink changes the per-sink checkpoint keys, which is not crash-safe; a PATCH that alters the sink set is rejected. Patches that leave the sink set unchanged are allowed. To change sinks, delete and recreate the pipeline.
• Deleting a pipeline is fail-closed. Delete stops the source, cleans up its checkpoints, and only then releases the source-id claim; if checkpoint cleanup fails, the delete fails and the source id stays locked (safe: it blocks reuse rather than exposing stale checkpoints). Retry the delete once the store is reachable.
• Cross-primary position safety at failover depends on GTID (MySQL) and slot-aware HA (PostgreSQL); see Failover Handling.
• MySQL replays binlog rows across a DDL only when the schema in effect can be proven. Rows are decoded with the version proven for their position. When none can be proven (retained rows older than a DDL the source never observed, a statement that cannot be attributed to tables such as a versioned-comment DDL, or two DDLs of one table before the source reads the first), the source stops fail-closed (no event, no checkpoint advance). The immediate remediation is a re-snapshot (restart once with snapshot mode always); after an unattributable statement, the table’s next rows (once its definition is captured and proven unchanged since them) or a later proven DDL of the table also restore proof. Moving the source position past the DDL intentionally abandons all retained changes in between, for every captured table, and requires an operator assessment of that loss. Set binlog_row_metadata=FULL to remove most of these stops. See Guarantees.

Out of scope

Not part of the supported deployment envelope (do not rely on these):

• Multi-instance HA for a single source (no cluster-wide ownership lock/lease yet).
• TLS directly to the source database (use a trusted network or an external tunnel).
• Authenticated REST API or metrics endpoint (use network controls).
• Live/dynamic credential rotation as a supported contract (treat rotation as restart-required; dynamic Vault DB credentials are not live).
• End-to-end exactly-once delivery, deterministic snapshot+replay rebuild, and any feature marked planned on the Roadmap.

Capacity & Resource Envelope

This page states DeltaForge’s resource behaviour so operators can size a deployment conservatively. It is a description of how the engine behaves today, not a set of guaranteed maximums.

Every bound below is tagged with how we know it:

• [measured] - observed in a test/benchmark run (environment-dependent; not a guarantee).
• [code-derived] - a constant or algorithm in the source; exact, but a mechanism, not a tested ceiling.
• [external] - enforced by the source/sink database, not by DeltaForge.
• [operator] - a configuration default the operator can change.
• [unknown] - not yet benchmarked; treat with caution.

These are not product ceilings. Where a number is [code-derived] or [operator] it describes a default or a mechanism, not a validated limit. Do not quote them as maximums. Run the preflight command and a disposable soak in your environment before committing to any number.

At-a-glance conservative starting guidance

Everything else is detailed below.

Memory and backpressure

DeltaForge has no aggregate (pipeline-wide or process-wide) memory or byte budget [code-derived - confirmed absent]. Resident memory is bounded only indirectly by the levers below, so you must provision headroom rather than rely on a cap.

• Source→coordinator channel: 32,768 items, item-count bound only, no byte cap [code-derived]. A burst of large change events can hold up to 32,768 SourceItems resident with no byte ceiling. This is the primary backpressure lever and is not operator-configurable. Worst-case channel memory ≈ 32768 × (largest event size); size RAM for your widest rows/transactions accordingly.
• In-flight batch bytes: bounded per batch by batch.max_bytes (default 16 MiB [operator]) and per source transaction by batch.max_tx_bytes (default 512 MiB [operator], only when respect_source_tx = true). If an operator sets these to unset/None, the effective cap becomes unbounded (usize::MAX) [code-derived] - do not disable them.
• Schema cache: only the latest version of tables in use is cached, process-wide, bounded by --schema-cache-max-bytes (default 64 MiB, a conservative estimate of resident bytes) and --schema-cache-max-entries (default 50,000) [operator]. Older versions and unused tables stay in the state store and are read on demand. See Schema registry at scale.
• Guidance: budget ≈ (channel depth × typical event size) + max_tx_bytes per active pipeline + the schema cache budget, with generous headroom. There is no backstop if you under-provision. Actual RSS under load is [unknown] pending a soak in your environment.

Transaction and batch size

BatchConfig defaults [operator], all overridable:

A single source transaction larger than max_tx_events/max_tx_bytes is failed closed by default (oversized_tx = Fail). Very large transactions on the source are the main way to blow past the byte budget - keep the defaults.

Concurrency and in-flight

• max_inflight default 1 [operator]. It bounds pipeline batch depth, not concurrent sink delivery: delivery is processed FIFO by a single delivery task to preserve checkpoint ordering, so raising it does not parallelize sinks [code-derived]. With respect_source_tx = true (the default), max_inflight must be 1 (startup fails otherwise) [code-derived]. So the effective default is single-in-flight.
• Snapshot parallelism [operator]: discovery_page_size 1,000 (catalog rows per discovery query), max_parallel_tables 8, chunk_size 10,000 rows, intra_table_parallel false, max_parallel_chunks 4 (only when intra-table parallel), lock_timeout_secs 10. Effective table concurrency is min(max_parallel_tables, table_count).

Queue and DLQ bytes

• DLQ (when journal.enabled = true, default off): max_entries 10,000 [operator], max_age_secs 7 days [operator], overflow_policy DropOldest [operator] (Reject/Block also available). The Block policy waits at most 60 s before failing closed [code-derived]. Journal max_event_bytes default 256 KB [operator].
• Replay stream (default off): max_entries/max_bytes default 0 = unbounded [operator]; max_envelope_bytes 8 MiB [operator]; retention_secs 24 h.
• Secret material cap: 1 MiB per secret [code-derived].

Source-log retention

Retention windows are [external] (enforced by the source DB); preflight only estimates risk.

• PostgreSQL: one replication slot retains WAL. max_slot_wal_keep_size governs the cap. Preflight estimates snapshot WAL as ≈ 2× data bytes and warns at ≥50% / flags HIGH risk at ≥80% of max_slot_wal_keep_size [code-derived heuristic]. wal_status = lost/unreserved is reported. Set max_slot_wal_keep_size generously or the slot can be invalidated mid-snapshot.
• MySQL: binlog_expire_logs_seconds (fallback expire_logs_days) governs binlog retention. Preflight estimates snapshot duration and warns at ≥50% / HIGH at ≥80% of the retention window [code-derived heuristic]; post-snapshot it fails closed if the captured binlog file was purged during the snapshot.
• Snapshot throughput estimate used by both: 20 MB/s per worker [code-derived heuristic, not measured].
• During a snapshot, retention the anchor needs (the PostgreSQL slot, the MySQL anchor binlog file) and the anchor’s age (snapshot.max_anchor_age_secs, default 24 h) are checked continuously: a warning incident at 80%, and at the limit the generation stops and blocks before anything is lost (snapshot_anchor_unavailable) [code-derived, tested]. See Bounds and blocking.

Database connections

Per pipeline [code-derived; scales with snapshot config]:

• Snapshot connection cap [operator]: every snapshot connection, across all pipelines of the process, is taken from one pool of --max-snapshot-connections (default 64); each source’s share is snapshot.max_snapshot_connections (default max_parallel_tables x max_parallel_chunks + 2). A snapshot that cannot get its 3 base connections queues holding nothing (deltaforge_snapshot_queued). See Connection limits.
• PostgreSQL - snapshot: 1 anchor + min(max_parallel_tables, tables) workers (up to 8) + up to ×max_parallel_chunks per worker if intra_table_parallel (worst case with defaults + intra-table ≈ ~41), within the source’s share. Steady CDC: 1 replication connection (= 1 walsender) + occasional short-lived control connections. No source-side connection pool - each is an individual connection.
• MySQL - snapshot: 1 lock connection + min(max_parallel_tables, tables) workers + 1 binlog-position guard, within the source’s share. Steady CDC: 1 binlog stream (+ one per proof scan in flight) + the source’s control-plane pool of at most proofs.control_connections (default 4, at most one kept idle), within the process budget --max-mysql-control-connections (default 32, 4 reserved for cancellations) [code-derived, tested]. See Control-plane connections.
• Storage backend (PostgreSQL): one connection pool per backend instance (shared across pipelines), max_size = deadpool default ≈ CPU × 4 [library-default], plus one background TTL-sweep task.

Size max_connections on the source and on the PostgreSQL storage DB for the snapshot peak, not steady state: at most --max-snapshot-connections snapshot connections across the process.

PostgreSQL slots and walsenders

• One replication slot and one walsender per PostgreSQL source [code-derived]. Provision max_replication_slots and max_wal_senders ≥ number of PG source pipelines on that server, with headroom for other consumers.

Checkpoint-store and schema-registry load

• Checkpoint writes: one put_raw per checkpointed sink per committed batch, with no coalescing [code-derived]:

  checkpoint writes/sec = committed batches/sec × checkpointed sinks
  

  This is workload-dependent and has no configured QPS ceiling. max_ms = 50 is a time-based flush ceiling, not a rate cap: batches also flush on max_events, max_bytes, and when the input drains (at most once per min_flush_interval_ms while nothing is in flight), so committed-batches/sec can be much higher than 1000/max_ms under load. Size the checkpoint/storage backend for the actual committed-batch rate × checkpointed sinks across all pipelines. Reads are change-driven (the source is notified on commit), not polled.

• Checkpoint reads are scoped to the source [code-derived, measured]: a source reads its per-sink checkpoints by key prefix, answered by the store as a bounded key range (PostgreSQL: a partial index on the checkpoint namespace). Cost does not grow with the number of other sources or pipelines sharing the store.

• Schema registry startup reads one record [measured]: no namespace scan and no history replay; each table’s latest schema is read on first use and cached within the budget above. See below.

Schema registry at scale

Measured with the dense-catalog harness (crates/scale-harness, registry-scale; see its README to reproduce) at commit 664838a, on a developer machine, single runs, SQLite state store. The numbers are indicative, not guarantees; the shape (what does and does not grow) is the point.

Synthetic registries: N tables per source × 2 sources with the same table names (colliding across sources), 2-3 versions each, generated through the normal registration path. 100 active tables are looked up.

• PostgreSQL state store (100K tables × 2 versions × 2 sources, local container): the same operation counts as SQLite - startup 1 key read (1.4 ms), a cold lookup 1 read (p50 205 µs, p99 816 µs, network round trips), cached lookups no store access, 1 read for 64 concurrent first lookups, 0 reads of other sources. Migration of 10K tables: ~56-60 s at 1 version and ~185 s at 5 versions per table (vs ~4 s / ~14 s on SQLite); memory the same ~2.4-2.5 KB per mapped table. Registrations ran at ~250/s against ~3,600/s on SQLite.
• Time to first CDC event: the source’s own registry holds N synthetic tables (they are not in the source database); a row committed while the pipeline was stopped is timed from source start until the sink receives it. The storage calls in that window are identical at 100K and 1M tables (PostgreSQL: 12 key reads, 8 prefix-scoped checkpoint listings, 1 write, 1 schema read; MySQL: 5, 1, 1, 1) with no registry scan. This isolates the registry.
• CDC restart against a large live catalog [measured]: the source’s table pattern matches 20 or 1,000 real tables in the database (only one of them changes). A CDC-only restart does the same work at both sizes, measured up to the arrival of the first event: the same storage operations, calls and returned and written bytes per primitive (PostgreSQL 11 key reads, 3 latest-version reads; MySQL 9 key reads; the timer-driven rereads of the sink checkpoints for WAL feedback are not counted), no registry enumeration, and the same number of statements on the server (PostgreSQL 9, MySQL 27); no catalog enumeration and no schema load for tables without changes. Both runs create the same 1,000 tables; only how many the pattern matches differs. Reproduce with cargo test -p scale-harness --test live_ttfce -- --include-ignored or registry-scale --live postgres,mysql --live-catalog-tables N.
• Per-source schema cache [code-derived]: the resident (heavyweight) table schemas are capped at 4,096 tables or about 64 MiB of serialized schema per source, whichever is reached first, with least-recently-used eviction. The limit is fixed (not an operator setting). An evicted table is rebuilt from durable history as exactly the version it resolved to, so the source keeps a compact record per table resolved in the current lineage generation (table key, version, sequence, fingerprint). That record is not bounded: it grows with the distinct tables used since the source started (or last changed lineage), and the 64 MiB does not include it or the cache’s key and map overhead. Bounding or paging these records is a later capacity improvement.
• MySQL schema selection caches [code-derived, measured]: each MySQL source caches at most 4,096 tables’ validated timelines and 256 MiB of timelines, shared barrier streams (once each), row-schema selections and read-ahead versions, evicted least recently used first (deltaforge_mysql_selection_cache_evictions_total{pipeline,source}); the estimate and the timelines held are exported as deltaforge_mysql_selection_cache_bytes and deltaforge_mysql_selection_cache_timelines, under the same labels. The bytes cover resident memory (resident measured at 0.66 to 0.75 of the estimate). A load in flight adds at most one timeline’s limit (64 MiB, 100,000 records) and one record. An evicted entry is rebuilt from the durable registry and activation records, never from the live catalog, in batches of up to 256 tables. The bounds are fixed (not operator settings). See Activation histories.
• Pre-upgrade schema history migration (deltaforge schema-migrate) [measured]: memory grows with the number of mapped tables, not with history length: about 2.4-2.8 KB per mapped table (10K tables: ~39 MiB peak; 100K tables: ~243 MiB peak), unchanged between 1 and 5 versions per table and between page sizes. Time: 100K tables took ~36 s at 1 version and ~135 s at 5 versions per table. For very large catalogs, split the migration with --tenant / --source or several mapping files.

Metric cardinality

• Default: independent of the table count [measured]. No metric carries a table label unless the pipeline enables metrics.per_table; a live test records identical series sets for pipelines over 10 and 1,000 tables (apart from batch flush causes, a closed set that varies with timing). See Observability.
• Opt-in per-table detail [code-derived, measured]: at most max_tables + 1 table values (default 100, up to 10,000) per metric and remaining label set; the per-table lag holds at most max_tables + 1 series per pipeline and drops a table’s series after lag_idle_secs without events. The histogram (deltaforge_schema_sensing_seconds) multiplies this by its buckets. Budget Prometheus accordingly before raising max_tables.

Table and source-unit counts

• No coded limit on tables per source or pipelines per instance [confirmed absent]. Practical limits come from: initial snapshot time (linear in the tables and rows copied; the durable plan is bounded by snapshot.max_plan_bytes / max_plan_items), and connection/slot math above. A CDC start does no per-table work (see below). The schema registry is no longer one of them ([measured] to 1M tables per source).
• Single-instance requirement: run exactly one DeltaForge process against a given state store (see Supported Deployment Envelope) [code-derived containment].
• Conservative starting configuration (unvalidated, not a supported limit): a small number of pipelines (single digits) and tens-to-low-hundreds of tables per source is a reasonable place to start, and grow only after a soak in your environment. We have not measured enough to claim either that larger configurations are unsupported or that this range is universally safe - both directions are [unknown] pending benchmark.

Throughput

Throughput is environment-dependent and is [measured] only on a developer machine (single runs, heavy desktop contention observed), so no headline number is published as a guarantee. The figures in Performance Tuning are such development-machine observations, useful for the relative effect of settings only; comparative benchmarks are defined in the benchmark design and run after rc.1. Treat sustained throughput as [unknown] for your environment until you run a soak. Backpressure bounds queued item count and batch/transaction payloads where their byte limits are enabled, but it does not provide a process-wide memory bound.

Known scaling caveats (pending benchmark)

These are [unknown] and should be validated before scaling up:

• Aggregate/RSS memory under sustained load (no coded cap).
• Initial snapshot time for sources capturing many tables: discovery reads the catalog in keyset pages (discovery_page_size, one query per page) and each table’s schema is resolved once, from the schema registry when known or the source catalog otherwise (a CDC restart does neither; see above).
• Initial snapshots of large catalogs [measured to 10,000 tables]: snapshot state is a durable, paged plan with constant-size boundaries and no progress rewrites; resident queue state is bounded by discovery_page_size and max_parallel_tables. From 1,000 to 10,000 tables, boundary bytes grew 10x and wall time 10.5x (PostgreSQL) and 12.2x (MySQL). Total process memory still grows with the tables captured, through the schema caches (MySQL peak RSS growth 17.9 MiB at 1K, 48.4 MiB at 10K). Figures, environment and scope in Initial Snapshots. Qualification beyond 10,000 tables per source (up to 400,000 per source, and fleets of 50 to 120 million tables) is a milestone after rc.1, not a current support claim.
• MySQL mysql_async control-pool sizing under many concurrent pipelines.

Configuration

Pipelines are defined as YAML documents that map directly to the PipelineSpec type. Environment variables are expanded before parsing using ${VAR} syntax, so secrets and connection strings can be injected at runtime. Unknown variables (e.g. ${source.table}) pass through for use as routing templates.

For credentials, prefer references (env, file, Kubernetes projected volume, or Vault KV) over inline values and ${ENV} expansion. See Secrets & Credentials.

Document structure

apiVersion: deltaforge/v1
kind: Pipeline
metadata:
  name: <pipeline-name>
  tenant: <tenant-id>
spec:
  source: { ... }
  processors: [ ... ]
  sinks: [ ... ]
  batch: { ... }
  commit_policy: { ... }
  schema_sensing: { ... }

Metadata

Spec fields

Sources

MySQL

Captures row-level changes via binlog replication. See MySQL source documentation for prerequisites and detailed configuration.

source:
  type: mysql
  config:
    id: orders-mysql
    dsn: ${MYSQL_DSN}
    tables:
      - shop.orders
      - shop.order_items
    outbox:
      tables: ["shop.outbox"]
    snapshot:
      mode: initial
    on_schema_drift: adapt

Table patterns (the initial snapshot and CDC select exactly the same tables):

• db.table - exact match
• db.prefix% or db.prefix* - tables matching the prefix (a wildcard counts only as the last character; _ is literal)
• db.% or db.* - all tables in database
• table (no qualifier) - that table in every database/schema

PostgreSQL

Captures row-level changes via logical replication. See PostgreSQL source documentation for prerequisites and detailed configuration.

source:
  type: postgres
  config:
    id: users-postgres
    dsn: ${POSTGRES_DSN}
    slot: deltaforge_users
    publication: users_pub
    tables:
      - public.users
      - public.sessions
    start_position: earliest
    outbox:
      prefixes: ["outbox", "order_outbox_%"]
    snapshot:
      mode: initial
    on_schema_drift: adapt

Note: The source-level outbox field only tags matching events with the __outbox sentinel. Routing and transformation are handled by the outbox processor.

Processors

Processors transform events between source and sinks. They run in order and can filter, enrich, or modify events.

JavaScript

processors:
  - type: javascript
    id: transform
    inline: |
      function processBatch(events) {
        return events.map(e => {
          e.tags = ["processed"];
          return e;
        });
      }
    limits:
      cpu_ms: 50
      mem_mb: 128
      timeout_ms: 500

Flatten

processors:
  - type: flatten
    id: flat
    separator: "__"
    max_depth: 3
    on_collision: last
    empty_object: preserve
    lists: preserve
    empty_list: drop

Filter

processors:
  - type: filter
    id: only-active-orders
    ops: [create, update]
    tables:
      include: ["shop.orders"]
      exclude: ["*.tmp"]
    fields:
      - path: status
        op: eq
        value: "active"
      - path: total
        op: gte
        value: 100
    match: all

Outbox

Transforms raw outbox events into routed, sink-ready events. Requires the source to have outbox configured so events are tagged before reaching this processor. See Outbox pattern documentation for full details.

processors:
  - type: outbox
    id: outbox
    topic: "${aggregate_type}.${event_type}"
    default_topic: "events.unrouted"
    raw_payload: true
    columns:
      payload: data
    additional_headers:
      x-trace-id: trace_id

Sinks

Sinks deliver events to downstream systems. Each sink supports configurable envelope formats and wire encodings to match consumer expectations. See the Sinks documentation for detailed information on multi-sink patterns, commit policies, and failure handling.

Envelope and encoding

The Kafka, Redis, NATS, and HTTP sinks support these serialization options. (The S3 sink uses format/compression instead — see S3.)

Envelope types:

• native - Direct Debezium payload structure (default, most efficient)
• debezium - Full {"payload": ...} wrapper
• cloudevents - CloudEvents 1.0 specification (requires type_prefix)

# Native (default)
envelope:
  type: native

# Debezium wrapper
envelope:
  type: debezium

# CloudEvents
envelope:
  type: cloudevents
  type_prefix: "com.example.cdc"

Kafka

See Kafka sink documentation for detailed configuration options and best practices.

sinks:
  - type: kafka
    config:
      id: orders-kafka
      brokers: ${KAFKA_BROKERS}
      topic: orders
      envelope:
        type: debezium
      encoding: json
      required: true
      exactly_once: false
      send_timeout_secs: 30
      client_conf:
        security.protocol: SASL_SSL

CloudEvents example:

sinks:
  - type: kafka
    config:
      id: events-kafka
      brokers: ${KAFKA_BROKERS}
      topic: events
      envelope:
        type: cloudevents
        type_prefix: "com.acme.cdc"
      encoding: json

Redis

See Redis sink documentation for detailed configuration options and best practices.

sinks:
  - type: redis
    config:
      id: orders-redis
      uri: ${REDIS_URI}
      stream: orders
      envelope:
        type: native
      encoding: json
      required: true

NATS

See NATS sink documentation for detailed configuration options and best practices.

sinks:
  - type: nats
    config:
      id: orders-nats
      url: ${NATS_URL}
      subject: orders.events
      stream: ORDERS
      envelope:
        type: native
      encoding: json
      required: true
      send_timeout_secs: 5
      batch_timeout_secs: 30

HTTP / Webhook

See HTTP sink documentation for full configuration. POST/PUT events to any URL with custom headers, URL templates, and batch mode.

sinks:
  - type: http
    config:
      id: orders-webhook
      url: "https://example.com/hooks/{source.table}"
      method: POST

S3 / Object storage

See S3 sink documentation for full configuration. Writes Parquet or JSON Lines to any S3-compatible store, Hive-partitioned by table + date. Uses format/compression (not envelope/encoding).

sinks:
  - type: s3
    config:
      id: lake
      bucket: my-lake
      prefix: cdc
      endpoint: "http://localhost:9000"
      region: us-east-1
      format: parquet
      compression: snappy
      required: true

Batching

batch:
  max_events: 500
  max_bytes: 1048576
  max_ms: 1000
  respect_source_tx: true
  max_inflight: 2

Commit policy

commit_policy:
  mode: required

# For quorum mode:
commit_policy:
  mode: quorum
  quorum: 2

Schema sensing

Schema sensing automatically infers and tracks schema from event payloads. See the Schema Sensing documentation for detailed information on how it works, drift detection, and API endpoints.

Performance tip: Schema sensing can be CPU-intensive, especially with deep JSON inspection. Consider your throughput requirements when configuring:

• Set enabled: false if you don’t need runtime schema inference
• Limit deep_inspect.max_depth to avoid traversing deeply nested structures
• Increase sampling.sample_rate to analyze fewer events (e.g., 1 in 100 instead of 1 in 10)
• Reduce sampling.warmup_events if you’re confident in schema stability

schema_sensing:
  enabled: true
  deep_inspect:
    enabled: true
    max_depth: 3
    max_sample_size: 500
  sampling:
    warmup_events: 50
    sample_rate: 5
    structure_cache: true
    structure_cache_size: 50
  high_cardinality:
    enabled: true
    min_events: 100
    stable_threshold: 0.5
    min_dynamic_fields: 5

Metrics

By default no metric carries a table label, so a pipeline’s series count does not depend on how many tables it captures. Per-table detail is opt-in:

metrics:
  per_table:
    enabled: true
    max_tables: 100      # tables with their own series (1-10,000)
    lag_idle_secs: 300   # per-table lag disappears after this long without events (10-86,400)

Out-of-range values are rejected when the pipeline is created and by preflight.

Complete examples

MySQL to Kafka with Debezium envelope

apiVersion: deltaforge/v1
kind: Pipeline
metadata:
  name: orders-mysql-to-kafka
  tenant: acme

spec:
  source:
    type: mysql
    config:
      id: orders-mysql
      dsn: ${MYSQL_DSN}
      tables:
        - shop.orders

  processors:
    - type: javascript
      id: transform
      inline: |
        function processBatch(events) {
          return events.map(event => {
            event.tags = (event.tags || []).concat(["normalized"]);
            return event;
          });
        }
      limits:
        cpu_ms: 50
        mem_mb: 128
        timeout_ms: 500

  sinks:
    - type: kafka
      config:
        id: orders-kafka
        brokers: ${KAFKA_BROKERS}
        topic: orders
        envelope:
          type: debezium
        encoding: json
        required: true
        exactly_once: false
        client_conf:
          message.timeout.ms: "5000"

  batch:
    max_events: 500
    max_bytes: 1048576
    max_ms: 1000
    respect_source_tx: true
    max_inflight: 2

  commit_policy:
    mode: required

PostgreSQL to Kafka with CloudEvents

apiVersion: deltaforge/v1
kind: Pipeline
metadata:
  name: users-postgres-to-kafka
  tenant: acme

spec:
  source:
    type: postgres
    config:
      id: users-postgres
      dsn: ${POSTGRES_DSN}
      slot: deltaforge_users
      publication: users_pub
      tables:
        - public.users
        - public.user_sessions
      start_position: earliest

  sinks:
    - type: kafka
      config:
        id: users-kafka
        brokers: ${KAFKA_BROKERS}
        topic: user-events
        envelope:
          type: cloudevents
          type_prefix: "com.acme.users"
        encoding: json
        required: true

  batch:
    max_events: 500
    max_ms: 1000
    respect_source_tx: true

  commit_policy:
    mode: required

Multi-sink with different formats

apiVersion: deltaforge/v1
kind: Pipeline
metadata:
  name: orders-multi-sink
  tenant: acme

spec:
  source:
    type: mysql
    config:
      id: orders-mysql
      dsn: ${MYSQL_DSN}
      tables:
        - shop.orders

  sinks:
    # Kafka Connect expects Debezium format
    - type: kafka
      config:
        id: connect-sink
        brokers: ${KAFKA_BROKERS}
        topic: connect-events
        envelope:
          type: debezium
        required: true

    # Lambda expects CloudEvents
    - type: kafka
      config:
        id: lambda-sink
        brokers: ${KAFKA_BROKERS}
        topic: lambda-events
        envelope:
          type: cloudevents
          type_prefix: "com.acme.cdc"
        required: false

    # Redis cache uses native format
    - type: redis
      config:
        id: cache-redis
        uri: ${REDIS_URI}
        stream: orders-cache
        envelope:
          type: native
        required: false

  batch:
    max_events: 500
    max_ms: 1000
    respect_source_tx: true

  commit_policy:
    mode: required

MySQL to NATS

apiVersion: deltaforge/v1
kind: Pipeline
metadata:
  name: orders-mysql-to-nats
  tenant: acme

spec:
  source:
    type: mysql
    config:
      id: orders-mysql
      dsn: ${MYSQL_DSN}
      tables:
        - shop.orders
        - shop.order_items

  sinks:
    - type: nats
      config:
        id: orders-nats
        url: ${NATS_URL}
        subject: orders.events
        stream: ORDERS
        envelope:
          type: native
        encoding: json
        required: true

  batch:
    max_events: 500
    max_ms: 1000
    respect_source_tx: true

  commit_policy:
    mode: required

Secrets & Credentials

DeltaForge resolves every credential in a deployment - source database, sinks, schema registry, and the storage backend - from references rather than inline values. A reference names where a secret lives (an environment variable, a file, or a Vault KV path); the runtime reads it at startup, holds it in a zeroized buffer, and hands it straight to the client that needs it. The secret never enters the parsed config, so it cannot leak through the status API, logs, or Debug output.

Inline secrets and ${ENV} expansion still work for backward compatibility, but they are deprecated. Prefer references for anything sensitive.

Reference shape

A reference is a small object:

{ provider: file, location: /run/secrets/db/password }

Providers

Environment variables (env)

{ provider: env, location: DF_PG_PASSWORD }

Covers Kubernetes env.valueFrom.secretKeyRef, which projects a Secret key into an environment variable.

Files (file)

{ provider: file, location: /run/secrets/db/password }

The path must be absolute. By default the file must be a regular file (not a symlink), which suits operator-mounted secrets.

Kubernetes projected volumes

A projected Secret volume mounts each key as a symlink into a versioned directory. To follow those symlinks safely, set a trusted root so the resolved target must stay within the mounted volume:

• Sources and sinks: set projected_file_root under spec.secrets (see Pipeline-level secret providers). A source file rotation trigger’s trusted_root is honored as a fallback.
• Storage backend: set secret_trusted_root on the storage config.

storage:
  backend: postgres
  dsn: postgres://pg.internal:5432/deltaforge      # password-less base
  secret_trusted_root: /var/run/secrets
  credentials:
    username: { provider: file, location: /var/run/secrets/store/username }
    password: { provider: file, location: /var/run/secrets/store/password }

HashiCorp Vault KV (vault)

A Vault reference points at a KV v2 record; location is <mount>/<path> (the KV v2 data/ segment is added automatically, so do not include it) and selector picks the field:

{ provider: vault, location: secret/deltaforge/pg, selector: password }

Vault KV references require a Vault connection. Configure it in the scope that owns the reference: the pipeline (spec.secrets) for source and sink references, and the storage config for the storage backend. The two are independent.

storage:
  backend: postgres
  dsn: postgres://pg.internal:5432/deltaforge
  vault:
    address: https://vault.internal:8200
    auth:
      method: kubernetes
      role: deltaforge-storage
      jwt_path: /var/run/secrets/kubernetes.io/serviceaccount/token
  dsn_secret: { provider: vault, location: secret/deltaforge/store, selector: dsn }

The runner must be built with the vault feature; otherwise a Vault-referenced credential fails closed at startup with a clear error.

Pipeline-level secret providers

spec.secrets assembles the single resolver shared by both source and sink credential resolution. Configure it here whenever any source or sink reference needs Vault or a Kubernetes projected volume - the source does not need to use credential rotation for a sink’s Vault or projected-file reference to resolve.

spec:
  secrets:
    # Enables `vault` references on the source and any sink.
    vault:
      address: https://vault.internal:8200
      auth:
        method: kubernetes
        role: deltaforge-pipeline
        jwt_path: /var/run/secrets/kubernetes.io/serviceaccount/token
    # Enables projected-volume (symlink) `file` references for the source and any sink.
    projected_file_root: /var/run/secrets

When spec.secrets is absent, the resolver falls back to the source’s rotation configuration (backward compatible): a source file rotation trigger supplies the projected-file root, and a source vault trigger supplies the Vault connection.

Where references apply

Source (PostgreSQL / MySQL)

Three mutually exclusive forms:

source:
  type: postgres
  config:
    dsn_secret: { provider: vault, location: secret/pg, selector: dsn }   # whole DSN

or a password-less base DSN plus credential references:

    dsn: postgres://pg.internal:5432/orders
    credentials:
      username: { provider: env,  location: DF_PG_USER }
      password: { provider: file, location: /run/secrets/pg/password }

Storage backend

dsn_secret (whole DSN) or a password-less dsn plus credentials (username/password). Resolved before the backend is opened.

Sinks

For Kafka and HTTP, a key set both inline and in secret_refs is rejected - never silently overridden.

Schema registry (Avro)

Any sink using encoding: { type: avro } resolves the registry’s basic-auth password from a reference:

encoding:
  type: avro
  schema_registry_url: https://schema-registry.internal:8081
  username: sr-user
  password_ref: { provider: vault, location: secret/sr, selector: password }

S3 and ambient AWS identity

Omitting all S3 keys is a first-class, no-static-secret mode: the sink uses the ambient AWS credential chain (IAM instance/role, environment, or profile). Provide access_key_id_ref and secret_access_key_ref together or not at all; a partial pair is rejected.

Rules the runtime enforces

All of these fail closed at startup, before any network connection:

• A reference that resolves to nothing, an empty value, or invalid UTF-8 is an error.
• Setting both an inline value and its reference for the same field is rejected.
• Partial credential sets (a username without a password, one S3 key without the other) are rejected.
• A base DSN that already embeds a password cannot also take credential references.
• Storage credential resolution happens before the backend is opened; source and sink resolution happens before any client is constructed.

Redaction

Resolved secrets are held in zeroized buffers and never written back into the serializable config. Inline secrets that remain in config for compatibility are redacted everywhere they could surface - the status API, the pipeline Debug representation, and logs all show ***REDACTED*** in place of secret values.

Rotation is restart-required

This baseline does not hot-swap credentials in a live client. Rotating any referenced credential (updating the Secret, file, or Vault KV value) takes effect when the affected pipeline or the runner restarts. Plan rotations as a rolling restart.

Zero-downtime rotation of dynamic Vault database credentials for PostgreSQL and MySQL sources is designed as a separate capability and is not part of this baseline.

Complete example: no inline secrets

apiVersion: deltaforge/v1
kind: Pipeline
metadata:
  name: orders-cdc
  tenant: retail
spec:
  source:
    type: postgres
    config:
      id: pg
      dsn: postgres://pg.internal:5432/orders
      publication: df_pub
      slot: df_slot
      tables: [public.orders]
      credentials:
        username: { provider: env,  location: DF_PG_USER }
        password: { provider: file, location: /run/secrets/pg/password }
  sinks:
    - type: kafka
      config:
        id: kafka
        brokers: broker.internal:9092
        topic: orders.events
        secret_refs:
          sasl.password: { provider: file, location: /run/secrets/kafka/sasl_password }
        encoding:
          type: avro
          schema_registry_url: https://schema-registry.internal:8081
          username: sr-user
          password_ref: { provider: file, location: /run/secrets/sr/password }
    - type: s3
      config:
        id: archive
        bucket: cdc-archive
        prefix: orders/
        # no keys: uses the pod's IAM role (ambient AWS identity)

REST API Reference

DeltaForge exposes a REST API for health checks, pipeline management, schema inspection, and drift detection. All endpoints return JSON.

Base URL

Default: http://localhost:8080

Configure with --api-addr:

deltaforge --config pipelines.yaml --api-addr 0.0.0.0:9090

Health Endpoints

Liveness Probe

GET /health

Liveness only: returns 200 while the process and its API are running, and reports how many pipelines have failed. A failed pipeline is a readiness concern (see below), not a reason to restart the process - a pipeline that stopped fail-closed would fail again after every restart.

Response: 200 OK

{"status": "healthy", "pipelines": 3, "failed_pipelines": 1}

Readiness Probe

GET /ready

Use for Kubernetes readiness probes. Returns 503 while any pipeline has failed, carries a blocking incident that is open or only acknowledged (acknowledging an incident does not clear it), has a pending recovery operation, or is starting, and names them (starting_pipelines lists the starting ones). Paused and stopped pipelines are operator intent and do not hold readiness back. A pipeline whose initial snapshot runs is starting until its snapshot generation completes, so /ready stays 503 for that whole time: in Kubernetes the pod is out of the Service’s endpoints while it snapshots. Liveness (/health) stays 200, so the pod is not restarted.

Response: 503 Service Unavailable

{
  "status": "not_ready",
  "failed_pipelines": ["orders-cdc"],
  "starting_pipelines": [],
  "blocked_pipelines": [
    {
      "name": "orders-cdc",
      "status": "failed",
      "primary_incident": "3f9c...",
      "durability_pending": false,
      "store_unavailable": false,
      "blocking_incidents": [
        {"incident_id": "3f9c...", "reason_code": "unclassified_failure", "status": {"state": "acknowledged"}, "durable": true}
      ],
      "overflow_blocking": 0
    }
  ]
}

Response: 200 OK

{
  "status": "ready",
  "pipelines": [
    {
      "name": "orders-cdc",
      "status": "running",
      "spec": { ... }
    }
  ]
}

Pipeline Management

List Pipelines

GET /pipelines
GET /pipelines?label=env:prod
GET /pipelines?label=env:prod&label=team:platform

Returns all pipelines with current status. Filter by labels with AND logic. Key-only filter (?label=env) matches any value.

Response: 200 OK

[
  {
    "name": "orders-cdc",
    "status": "running",
    "spec": {
      "metadata": { "name": "orders-cdc", "tenant": "acme" },
      "spec": { ... }
    }
  }
]

Get Pipeline

GET /pipelines/{name}

Returns a single pipeline by name with operational status.

status is starting, running, paused, stopped, deleting or failed. A started pipeline is starting until its current run’s capture boundary is fixed and durable, then running, and capture_start names that boundary: changes to the configured tables committed after it are captured. It says nothing about delivery latency, tables excluded by configuration, or delivery while paused. Its origin is first_start (no checkpoint, no snapshot: the server’s position when it was read), resume (the committed resume position), snapshot (a snapshot generation’s anchor, once the generation is durably completed) or maintenance (a PostgreSQL resume moved by a recorded publication maintenance decision). A restart reports starting again until its run publishes its boundary. See design.

"capture_start": {"source": "mysql", "origin": "first_start", "lineage": "…",
                  "server_uuid": "…", "file": "mysql-bin.000007", "pos": 685,
                  "gtid_set": "…:1-721031"}
"capture_start": {"source": "postgres", "origin": "resume", "lineage": "…",
                  "system_identifier": 7412…, "slot": "df_slot", "lsn": "0/16B3780"}

Response: 200 OK

{
  "name": "orders-cdc",
  "status": "running",
  "spec": { ... },
  "ops": {
    "uptime_seconds": 3600.5,
    "dlq_entries": 0,
    "sink_errors": {},
    "checkpoints": [
      {"sink_id": "kafka-primary", "position": {"file": "mysql-bin.000005", "pos": 12345}, "age_seconds": 0.3}
    ]
  }
}

Errors:

• 404 Not Found - Pipeline doesn’t exist

Create Pipeline

POST /pipelines
Content-Type: application/json

Creates a new pipeline from a full spec.

Request:

{
  "metadata": {
    "name": "orders-cdc",
    "tenant": "acme"
  },
  "spec": {
    "source": {
      "type": "mysql",
      "config": {
        "id": "mysql-1",
        "dsn": "mysql://user:pass@host/db",
        "tables": ["shop.orders"]
      }
    },
    "processors": [],
    "sinks": [
      {
        "type": "kafka",
        "config": {
          "id": "kafka-1",
          "brokers": "localhost:9092",
          "topic": "orders"
        }
      }
    ]
  }
}

Response: 200 OK

{
  "name": "orders-cdc",
  "status": "running",
  "spec": { ... }
}

Errors:

• 409 Conflict - Pipeline already exists

Update Pipeline

PATCH /pipelines/{name}
Content-Type: application/json

Applies a partial update to an existing pipeline. The spec is merged, the pipeline is restarted from its last saved checkpoint, and the new config takes effect immediately. Only the fields present in the request body are changed — omitted fields retain their current values.

If the pipeline is currently stopped, PATCH applies the new config and restarts it from the saved checkpoint. This is the recommended way to tune throughput settings before resuming after a planned stop.

Request:

{
  "spec": {
    "batch": {
      "max_events": 1000,
      "max_ms": 500
    }
  }
}

Response: 200 OK

{
  "name": "orders-cdc",
  "status": "running",
  "spec": { ... }
}

Errors:

• 404 Not Found - Pipeline doesn’t exist
• 400 Bad Request - Invalid field value or name mismatch in patch

Pause Pipeline

POST /pipelines/{name}/pause

Suspends event processing while keeping the source connection alive. No new events are consumed from the binlog/WAL. Resume restarts processing from exactly where it paused — no events are missed.

Response: 200 OK

{
  "name": "orders-cdc",
  "status": "paused",
  "spec": { ... }
}

Resume Pipeline

POST /pipelines/{name}/resume

Resumes a paused or stopped pipeline.

• From paused — restarts event processing immediately; source connection was kept alive.
• From stopped — reconnects to the source and replays from the last saved checkpoint; any events written to the binlog/WAL while stopped are replayed in order.
• From failed: restarts it like a stopped pipeline. A resume never acknowledges or resolves its incidents: an unclassified_failure is resolved (pipeline_recovered) only once the restarted pipeline has reached verified running (the source finished its startup checks and opened its stream while the coordinator runs and nothing has failed).

Response: 200 OK

{
  "name": "orders-cdc",
  "status": "running",
  "spec": { ... }
}

Stop Pipeline

POST /pipelines/{name}/stop

Gracefully stops a pipeline: flushes in-flight events, saves the binlog/WAL checkpoint, and disconnects from the source. The pipeline remains in the registry and can be resumed with POST /pipelines/{name}/resume or by issuing a PATCH with updated config.

Use stop (rather than delete) when you intend to restart the pipeline later — for example, before a planned maintenance window or when tuning config for a backlog drain.

Response: 200 OK

{
  "name": "orders-cdc",
  "status": "stopped",
  "spec": { ... }
}

Delete Pipeline

DELETE /pipelines/{name}

Permanently removes a pipeline from the runtime. The checkpoint is not preserved. Use stop first if you may want to restart the pipeline later.

Response: 204 No Content

Errors:

• 404 Not Found - Pipeline doesn’t exist

Schema Management

List Database Schemas

GET /pipelines/{name}/schemas

Returns all tracked database schemas for a pipeline. These are the schemas loaded directly from the source database.

Response: 200 OK

[
  {
    "database": "shop",
    "table": "orders",
    "column_count": 5,
    "primary_key": ["id"],
    "fingerprint": "sha256:a1b2c3d4e5f6...",
    "registry_version": 2
  },
  {
    "database": "shop",
    "table": "customers",
    "column_count": 8,
    "primary_key": ["id"],
    "fingerprint": "sha256:f6e5d4c3b2a1...",
    "registry_version": 1
  }
]

Get Schema Details

GET /pipelines/{name}/schemas/{db}/{table}

Returns detailed schema information including all columns.

Response: 200 OK

{
  "database": "shop",
  "table": "orders",
  "columns": [
    {
      "name": "id",
      "column_type": "bigint(20) unsigned",
      "data_type": "bigint",
      "nullable": false,
      "ordinal_position": 1,
      "default_value": null,
      "extra": "auto_increment",
      "is_primary_key": true
    },
    {
      "name": "customer_id",
      "column_type": "bigint(20)",
      "data_type": "bigint",
      "nullable": false,
      "ordinal_position": 2,
      "default_value": null,
      "extra": null,
      "is_primary_key": false
    }
  ],
  "primary_key": ["id"],
  "fingerprint": "sha256:a1b2c3d4..."
}

Schema Sensing

Schema sensing automatically infers schema structure from JSON event payloads. This is useful for sources that don’t provide schema metadata or for detecting schema evolution in JSON columns.

List Inferred Schemas

GET /pipelines/{name}/sensing/schemas

Returns all schemas inferred via sensing for a pipeline.

Response: 200 OK

[
  {
    "table": "orders",
    "fingerprint": "sha256:abc123...",
    "sequence": 3,
    "event_count": 1500,
    "stabilized": true,
    "first_seen": "2025-01-15T10:30:00Z",
    "last_seen": "2025-01-15T14:22:00Z"
  }
]

Get Inferred Schema Details

GET /pipelines/{name}/sensing/schemas/{table}

Returns detailed inferred schema including all fields.

Response: 200 OK

{
  "table": "orders",
  "fingerprint": "sha256:abc123...",
  "sequence": 3,
  "event_count": 1500,
  "stabilized": true,
  "fields": [
    {
      "name": "id",
      "types": ["integer"],
      "nullable": false,
      "optional": false
    },
    {
      "name": "metadata",
      "types": ["object"],
      "nullable": true,
      "optional": false,
      "nested_field_count": 5
    },
    {
      "name": "tags",
      "types": ["array"],
      "nullable": false,
      "optional": true,
      "array_element_types": ["string"]
    }
  ],
  "first_seen": "2025-01-15T10:30:00Z",
  "last_seen": "2025-01-15T14:22:00Z"
}

Export JSON Schema

GET /pipelines/{name}/sensing/schemas/{table}/json-schema

Exports the inferred schema as a standard JSON Schema document.

Response: 200 OK

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "orders",
  "type": "object",
  "properties": {
    "id": { "type": "integer" },
    "metadata": { "type": ["object", "null"] },
    "tags": {
      "type": "array",
      "items": { "type": "string" }
    }
  },
  "required": ["id", "metadata"]
}

Get Sensing Cache Statistics

GET /pipelines/{name}/sensing/stats

Returns cache performance statistics for schema sensing.

Response: 200 OK

{
  "tables": [
    {
      "table": "orders",
      "cached_structures": 3,
      "max_cache_size": 100,
      "cache_hits": 1450,
      "cache_misses": 50
    }
  ],
  "total_cache_hits": 1450,
  "total_cache_misses": 50,
  "hit_rate": 0.9667
}

Drift Detection

Drift detection compares expected database schema against observed data patterns to detect mismatches, unexpected nulls, and type drift.

Get Drift Results

GET /pipelines/{name}/drift

Returns drift detection results for all tables in a pipeline.

Response: 200 OK

[
  {
    "table": "orders",
    "has_drift": true,
    "columns": [
      {
        "column": "amount",
        "expected_type": "decimal(10,2)",
        "observed_types": ["string"],
        "mismatch_count": 42,
        "examples": ["\"99.99\""]
      }
    ],
    "events_analyzed": 1500,
    "events_with_drift": 42
  }
]

Get Table Drift

GET /pipelines/{name}/drift/{table}

Returns drift detection results for a specific table.

Response: 200 OK

{
  "table": "orders",
  "has_drift": false,
  "columns": [],
  "events_analyzed": 1000,
  "events_with_drift": 0
}

Errors:

• 404 Not Found - Table not found or no drift data available

Dead Letter Queue

See the DLQ page for full documentation.

Peek DLQ Entries

GET /pipelines/{name}/journal/dlq?limit=50&sink_id=kafka-primary&error_kind=serialization

Returns DLQ entries (oldest first). All query params are optional.

DLQ Count

GET /pipelines/{name}/journal/dlq/count

Response: 200 OK

{"count": 42}

Acknowledge DLQ Entries

POST /pipelines/{name}/journal/dlq/ack
Content-Type: application/json

{"up_to_seq": 42}

Permanently removes entries from the head up to the given sequence number.

Response: 200 OK

{"acked": 12}

Purge DLQ

DELETE /pipelines/{name}/journal/dlq

Response: 200 OK

{"purged": 42}

Checkpoint Inspection

Get Checkpoints

GET /pipelines/{name}/checkpoints

Returns per-sink checkpoint positions and ages.

Response: 200 OK

[
  {"sink_id": "kafka-primary", "position": {"file": "mysql-bin.000005", "pos": 12345}, "age_seconds": 0.3},
  {"sink_id": "redis-cache", "position": {"file": "mysql-bin.000005", "pos": 11000}, "age_seconds": 2.1}
]

Incidents

A pipeline that stops (or cannot do something safely) records an incident: a structured description with a stable reason_code, retryability, safety_state (halted_safe, halted_uncertain, running_degraded), allow-listed evidence, recommended_actions and an explanation generated from them. Incidents never contain error text, DSNs, SQL or row values. The same condition is the same incident across retries and restarts (its occurrences grow); a later independent occurrence is a new one.

Lifecycle: open -> acknowledged -> resolved. Acknowledging means an operator has seen it; the incident stays open, keeps blocking the pipeline and readiness, and is resolved only by a verified check (or a future recovery operation). Every transition is recorded in an audit log, including reclassified: the same condition raised again with another retryability or safety state (an automatic retry that exhausted its budget now needs an operator) updates the incident rather than opening a new one. A pipeline keeps at most 63 open incidents individually; beyond that one incident_overflow incident counts the rest by reason code (blocking ones first).

Reason codes:

Conditions resolved by a verified start (unclassified_failure, pg_different_cluster, the two position reasons) are bound to a recovery generation: the same failure before the restarted pipeline reaches verified running updates the same incident; after a genuine recovery it is a new one.

Status (GET /pipelines/{name}, GET /pipelines, /ready) carries an incidents summary: primary (the blocking incident of a failed pipeline), primary_final, blocking, overflow_blocking, durability_pending (an incident is known but its durable write is still being retried - it is not yet auditable) and store_unavailable.

List Incidents

GET /pipelines/{name}/incidents

Every incident of the pipeline (open, acknowledged, recently resolved, and any not yet durable), with the primary incident.

Get Incident

GET /pipelines/{name}/incidents/{id}

Response: 200 OK

{
  "incident_id": "3f9c...",
  "reason_code": "unclassified_failure",
  "component": {"kind": "source", "id": "mysql"},
  "retryability": "auto_retry",
  "safety_state": "halted_uncertain",
  "cause_code": "source_connect",
  "explanation": "source mysql failed (source_connect). The pipeline stopped; this failure is not classified yet, see the logs for details.",
  "evidence": {},
  "recommended_actions": ["inspect_logs"],
  "status": {"state": "open"},
  "blocking": true,
  "occurrences": 2,
  "durable": true,
  "audit_pending": false
}

Acknowledge Incident

POST /pipelines/{name}/incidents/{id}/acknowledge
Content-Type: application/json

{"asserted_actor": "alice", "reason": "investigating the source outage"}

Moves an open incident to acknowledged and records the actor, the request’s peer address (origin) and the reason in the audit log. The API has no authenticated identity yet: asserted_actor is recorded as caller-supplied (actor_verified: false). It does not change the pipeline, its source, sinks or checkpoints, and does not resolve the incident.

Responses: 200 OK with the incident; 400 when a field is empty; 404 for an unknown incident; 409 when it is already resolved, or not durable yet (retry once it is).

System Endpoints

Log Level

GET /log-level

Returns the current RUST_LOG value.

Response: 200 OK

{"level": "deltaforge=info,sources=info,sinks=info,warn"}

Validate Config

POST /validate
Content-Type: application/json

Dry-run validation of a pipeline config without creating it.

Response: 200 OK — config is valid

{"valid": true, "pipeline": "orders-cdc", "source_type": "mysql", "sink_count": 2}

Response: 400 Bad Request — config has errors

{"valid": false, "error": "spec: missing field `processors` at line 7 column 3"}

Error Responses

All error responses return structured JSON:

{
  "code": "PIPELINE_NOT_FOUND",
  "message": "pipeline orders-cdc not found"
}

Pipelines

Each pipeline is created from a single PipelineSpec. The runtime spawns the source, processors, and sinks defined in the spec and coordinates them with batching and checkpointing.

• 🔄 Live control: pause, resume, or stop pipelines through the REST API without redeploying.
• 📦 Coordinated delivery: batching and commit policy keep sinks consistent even when multiple outputs are configured.

Lifecycle controls

The REST API addresses pipelines by metadata.name and returns PipeInfo records containing the live spec and status.

• GET /health - liveness probe.
• GET /ready - readiness with pipeline states.
• GET /pipelines - list pipelines.
• POST /pipelines - create from a full spec.
• PATCH /pipelines/{name} - merge a partial spec (for example, adjust batch thresholds) and restart the pipeline.
• POST /pipelines/{name}/pause - pause ingestion and coordination.
• POST /pipelines/{name}/resume - resume a paused pipeline.
• POST /pipelines/{name}/stop - stop a running pipeline.

Pausing halts both source ingestion and the coordinator. Resuming re-enables both ends so buffered events can drain cleanly.

Processors

Processors run in the declared order for each batch. The built-in processor type is JavaScript, powered by deno_core.

• type: javascript
  • id: processor label.
  • inline: JS source. Export a processBatch(events) function that returns the transformed batch.
  • limits (optional): resource guardrails (cpu_ms, mem_mb, timeout_ms).

Batching

The coordinator builds batches using soft thresholds:

• max_events: flush after this many events.
• max_bytes: flush after the serialized size reaches this limit.
• max_ms: flush after this much time has elapsed since the batch started.
• min_flush_interval_ms (default 40): when the input drains and no batch is being delivered, a batch of whole transactions is delivered at once, but no sooner than this long after the previous such delivery. A quiet source’s transaction goes immediately; a steady trickle is batched over this window; under load, batches fill to the limits above.
• respect_source_tx: when true, never split a single source transaction across batches.
• max_inflight: cap the number of batches being processed concurrently.

Commit policy

When multiple sinks are configured, checkpoints can wait for different acknowledgement rules:

• all: every sink must acknowledge.
• required (default): only sinks marked required: true must acknowledge; others are best-effort.
• quorum: checkpoint after at least quorum sinks acknowledge.

Sources

DeltaForge captures database changes through pluggable source connectors. Each source is configured under spec.source with a type field and a config object. Environment variables are expanded before parsing using ${VAR} syntax.

Supported Sources

Source	Status	Description
mysql	✅ Production	MySQL binlog CDC with GTID support
postgres	✅ Production	PostgreSQL logical replication via pgoutput

Common Behavior

All sources share these characteristics:

• Checkpointing: Progress is automatically saved and resumed on restart
• Schema tracking: Table schemas are loaded and fingerprinted for change detection
• At-least-once delivery: Events may be redelivered after failures; sinks should be idempotent
• Batching: Events are batched according to the pipeline’s batch configuration
• Transaction boundaries: respect_source_tx: true (default) keeps source transactions intact

Source Interface

Sources implement a common trait that provides:

#![allow(unused)]
fn main() {
trait Source {
    fn checkpoint_key(&self) -> &str;
    async fn run(&self, tx: Sender<Event>, checkpoint_store: Arc<dyn CheckpointStore>) -> SourceHandle;
}
}

The returned SourceHandle supports pause/resume and graceful cancellation.

Adding Custom Sources

The source interface is pluggable. To add a new source:

1. Implement the Source trait
2. Add configuration parsing in deltaforge-config
3. Register the source type in the pipeline builder

See existing sources for implementation patterns.

MySQL source

DeltaForge tails the MySQL binlog to capture row-level changes with automatic checkpointing and schema tracking.

Prerequisites

MySQL Server Configuration

Ensure your MySQL server has binary logging enabled with row-based format:

-- Required server settings (my.cnf or SET GLOBAL)
log_bin = ON
binlog_format = ROW
gtid_mode = ON                    -- required when the initial snapshot runs
enforce_gtid_consistency = ON     -- required with gtid_mode = ON
log_replica_updates = ON          -- required (the default); see below
binlog_row_image = FULL  -- Recommended for complete before-images
binlog_row_metadata = FULL  -- Recommended: decodes rows across unobserved schema changes

log_replica_updates = ON is required on every source, primary or replica (log_slave_updates before MySQL 8.0.26; the default since 8.0). With it OFF, transactions the server applies through replication - from a channel configured now or at any later time - change its tables and advance its executed GTID set without reaching the binlog DeltaForge reads, so changes would be missed and a schema change could go unseen. DeltaForge refuses such a server before it writes anything durable, reads any table or runs a snapshot, and checks again each time a binlog stream opens (startup, every reconnect, credential rotation) before reading from it, and before a failover candidate is reconciled. The setting is read-only: changing it requires a server restart, which ends the stream, so the reconnect afterwards is checked. Changing it on a running server is otherwise not possible and not supported. The check fails closed when the setting cannot be read.

If binlog_row_image is not FULL, DeltaForge will warn at startup and before-images on UPDATE/DELETE events may be incomplete. For binlog_row_metadata, see Schema Tracking.

Initial-snapshot requirements. When a snapshot runs (snapshot.mode = initial or always), DeltaForge fails closed at preflight unless: gtid_mode = ON, binlog_format = ROW, every snapshotted table uses the InnoDB storage engine, and the user holds the global RELOAD privilege (see Snapshot). These are checked only when a snapshot actually runs; a CDC-only pipeline (snapshot.mode = never) is never rejected for them.

User Privileges

Create a dedicated replication user with the required grants:

-- Create the user with the server's default authentication plugin
-- (caching_sha2_password on MySQL 8.x)
CREATE USER 'deltaforge'@'%' IDENTIFIED BY 'your_password';

-- Replication privileges (required)
GRANT REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'deltaforge'@'%';

-- The capture boundary of a first start without a checkpoint and without a
-- snapshot: the binlog position and GTID set, read together
GRANT BACKUP_ADMIN ON *.* TO 'deltaforge'@'%';
GRANT SELECT ON performance_schema.log_status TO 'deltaforge'@'%';

-- Global RELOAD (required only when the initial snapshot runs; used for the
-- brief FLUSH TABLES WITH READ LOCK that brackets the consistent anchor)
GRANT RELOAD ON *.* TO 'deltaforge'@'%';

-- Schema introspection (required for table discovery)
GRANT SELECT, SHOW VIEW ON your_database.* TO 'deltaforge'@'%';

FLUSH PRIVILEGES;

For capturing all databases, grant SELECT on *.* instead. RELOAD is only required if the pipeline will run an initial snapshot; a CDC-only pipeline (snapshot.mode = never) does not need it.

Configuration

Set spec.source.type to mysql and provide a config object:

Table Patterns

The tables field takes table patterns; the initial snapshot and CDC select exactly the same tables:

tables:
  - shop.orders          # exact match: database "shop", table "orders"
  - shop.order_%         # prefix: tables starting with "order_" in "shop" (same as shop.order_*)
  - analytics.*          # wildcard: all tables in "analytics" database
  - %.audit_log          # cross-database: "audit_log" table in any database
  # use an empty list [] to capture all user tables (excludes system schemas)

A wildcard (* or %) only counts as the last character of a part (a prefix match); _ and any other character are literal. A pattern without a qualifier matches that table name in every database/schema. The initial snapshot copies exactly the tables these patterns capture during CDC.

System schemas (mysql, information_schema, performance_schema, sys) are always excluded.

Example

source:
  type: mysql
  config:
    id: orders-mysql
    dsn: ${MYSQL_DSN}
    tables:
      - shop.orders
      - shop.order_items

Schema proof bounds

proofs bounds the work that proves the schema version of a table’s first rows (see Rows are decoded with the definition in effect when they were written). Every field is optional. A value outside its range is refused when the configuration is loaded:

source:
  type: mysql
  config:
    id: orders-mysql
    dsn: ${MYSQL_DSN}
    tables: ["shop.*"]
    proofs:
      activation_window_ms: 50
      max_seek_ms: 10000

Activation histories

A table’s schema version at its rows is chosen from its activation timeline: the records of its own stream and the barriers that apply to it (its lineage’s and its database’s), kept in the state store.

• Cached and bounded: a source caches at most 4,096 tables’ timelines and 256 MiB of them (records, barrier streams once each, and the schemas it read), whichever comes first. The bytes cover resident memory. Admission evicts the least recently used entries first; an evicted table is read again exactly from the store.

• Read in batches: a table the cache does not hold is read with the other uncached tables the stream is about to reach (the events an activation window read ahead or holds, or a short read-ahead within the window’s bounds), at most 256 tables per statement, each barrier stream once. Known tables returning after eviction cost a few store calls per batch, not per table.

• Barriers: a barrier invalidates only the cached timelines and selections of the tables it applies to: a database barrier, that database’s; a lineage barrier, every table’s.

• The per-timeline limit: a timeline is never loaded whole beyond 100,000 records or 64 MiB (stored bytes), its barrier streams included. A table over the limit stops the source with a mysql_timeline_too_large incident before any of its rows is dispatched; the checkpoint stays before them. It needs an operator: the timeline does not shrink by itself. Before upgrading, this read-only query on the state store (PostgreSQL or SQLite) lists the activation and barrier streams that are large, as records and approximate bytes:

  SELECT ns, key, COUNT(*) AS records,
         SUM(LENGTH(val) + COALESCE(LENGTH(capture_id), 0)
             + COALESCE(LENGTH(content_hash), 0) + 32) AS bytes
  FROM df_log
  WHERE ns IN ('schemas.v1.activation', 'schemas.v1.activation.barrier')
  GROUP BY ns, key
  HAVING COUNT(*) > 10000 OR SUM(LENGTH(val)) > 8388608
  ORDER BY bytes DESC;
  

  A table’s timeline is its activation stream (key is the table’s key) plus its lineage’s and its database’s barrier streams (keys that prefix it).

Control-plane connections

Every SQL connection a MySQL source opens besides its binlog stream and its snapshot copy (schema captures, catalog reads, position and identity queries, health checks) comes from the source’s pool of proofs.control_connections, within one budget for the whole process, --max-mysql-control-connections (default 32, of which 4 are reserved for cancelling captures):

• A connection holds its permits from before it opens until it has closed.
• A returned connection is kept as the source’s one idle connection for at most 10 s; any other closes. When the process budget is full, every idle connection closes at once to make room.
• Waits are first come, first served. A capture waits at most its capture_lock_wait_secs; a position query or a health check at most 5 s, one health check at a time per source; a catalog read at most 30 s.
• An activation window’s, a failover drift check’s or a DDL’s forward proof capture that gets no connection within its wait stops the source with a mysql_control_plane_saturated incident and is retried by the restart: without it, rows could not be decoded, drift decided, or the rows after the DDL proven. A forward proof stops before its DDL is emitted or checkpointed; the restart replays the DDL and proves it.
• A snapshot baseline that gets no connection keeps its existing safe deferral, as for any failed capture: it only proves early what the table’s lazy proof proves at its next rows, so the table is proven there before any of those rows is delivered. Each deferral is logged and counted in deltaforge_mysql_control_saturation_deferrals_total{kind} (kind="baseline"); it never raises an incident.
• Cancelling a capture never waits for a pool: it draws on the reserve.
• After a credential rotation or a failover, connections opened before it are never reused: the idle one closes at once, the others when returned.

deltaforge_mysql_connections{source, kind} reports a source’s open connections by kind (replication, control, cancellation, snapshot_copy). deltaforge_mysql_control_permit_wait_seconds{class}, deltaforge_mysql_control_permit_timeouts_total{class, scope} and deltaforge_mysql_control_idle_evictions_total report the waits.

A DDL’s forward proof

When the stream reads a DDL on a tracked table, the source proves the table’s new shape (a capture of it and a scan of the binlog since the DDL) before the DDL event is emitted. A proof that fails for a transient reason is retried in place, with the stream paused at the DDL:

• Retried: a metadata-lock wait timeout or deadlock (another session holds or queues the table’s lock), busy capture slots, and connection failures (refused, reset, timed out, closed, server shutting down).
• Bounds: at most 4 attempts within 120 s from the first, with 1, 4 and 16 s between them. A capture starts only with at least 1 s left, and waits for the lock at most 30 s or the time left. Each attempt captures and scans afresh; a table proven in one attempt is committed then and not captured again.
• Exhausted: the source stops before the DDL is emitted or checkpointed, with a mysql_forward_proof_unavailable incident. A restart replays the DDL and proves it again; tables proven earlier are recognized by their recorded binding. While the proof retries, the same incident is open as auto_retry; it is resolved when the proof succeeds.
• Not retried: an authentication or authorization failure stops the source the same way at once (operator action). A connection to another server stops it as a lineage error, and a saturated control plane as mysql_control_plane_saturated, as before. A refusal on the evidence (a DDL on the table in the scanned interval, a purged binlog, a scan bound) leaves the DDL pending as before; the table’s next rows are then proven by its lazy proof, which can still succeed.
• Deadline: no capture, scan, store batch or repair of the proof starts after its deadline. Store work already in progress is not cut: planning reads, a repair, a batch (including the resolution of an unacknowledged commit), and binding or freshness reads, so the proof can end after the deadline by as long as those take. A capture or scan cut at the deadline is ended on the server. A stop during the last batch keeps what it committed and still stops the source before the DDL.

deltaforge_mysql_forward_proof_retries_total{class} counts retried failures (lock_wait_timeout, deadlock, capture_slots_busy, transport, read, not_reached, deadline), and deltaforge_mysql_forward_proof_outcomes_total{outcome} the ends of proofs that retried (recovered, ended_deferred, exhausted, cancelled, ended_by_terminal).

Resume Behavior

DeltaForge automatically checkpoints progress and resumes from the last position on restart. The resume strategy follows this priority:

1. GTID: Preferred if the MySQL server has GTID enabled. Provides the most reliable resume across binlog rotations and failovers.
2. File:position: Used when GTID is not available. Resumes from the exact binlog file and byte offset.
3. Capture boundary: On first run with no checkpoint and no snapshot, the source reads the binlog position and executed GTID set together in one performance_schema.log_status query (MySQL documents them as consistent with each other) and persists it before the stream opens. The pipeline reports running only after that, with the position as capture_start: changes committed after it are captured, earlier ones are not (no historical replay). Every retry and restart resumes from it, never from a newer tail. Reading it needs BACKUP_ADMIN and SELECT on performance_schema.log_status (both, only for this first start; a start with a checkpoint or a snapshot does not read it); without them the start fails closed with a permission error naming both grants, with no fallback:

GRANT BACKUP_ADMIN ON *.* TO 'deltaforge'@'%';
GRANT SELECT ON performance_schema.log_status TO 'deltaforge'@'%';

performance_schema.log_status exists from MySQL 8.0.14. The requirement is validated on MySQL 8.4 only; other versions and managed services (where BACKUP_ADMIN may not be grantable) are not yet qualified.

Checkpoints are stored using the id field as the key.

GTID sets are exact

A GTID checkpoint holds exactly the transactions DeltaForge has seen or resumed past, gaps included. Gaps in a server’s executed set are normal:

• Group Replication and InnoDB Cluster members assign GTIDs of the group’s UUID from their own blocks (group_replication_gtid_assignment_block_size, 1,000,000 by default). A group whose members both write has an executed set such as group:1-3:1000001, and the gap fills as the first member commits more.
• An explicitly assigned GTID (SET gtid_next) leaves the numbers below it free for later transactions.

A checkpoint that claimed such a gap would make the server skip the transactions that fill it on resume. Earlier versions stored group:1-3:1000001 as group:1-1000001; on a group, every row the first member committed after a restart was then lost. Such checkpoints are refused (see MySQL GTID state from earlier versions).

Snapshot (Initial Load)

DeltaForge can perform a consistent initial snapshot of existing table data before starting binlog streaming. This is the recommended approach for migrating existing tables without manual backfills.

How it works

DeltaForge establishes a single consistent anchor under a brief global read lock. It holds FLUSH TABLES WITH READ LOCK on a dedicated connection only while it opens a bounded pool of REPEATABLE READ START TRANSACTION WITH CONSISTENT SNAPSHOT workers and captures the binlog position + GTID set, then releases the lock. Because no transaction can commit while the lock is held, every worker shares one consistent view that matches the captured position exactly - so a row committed during snapshot setup is never lost from both the snapshot and the CDC stream. Workers then read in parallel (bounded by max_parallel_tables), each reading its tables under its single snapshot and committing once.

The lock window is bounded by snapshot.lock_timeout_secs (default 10s); if the lock cannot be acquired in time, the snapshot fails closed rather than stalling writes. The lock is always released - the dedicated lock connection is dropped on any error, panic, or timeout, which releases it server-side.

Correctness note. This closes a real initial-snapshot data-loss window in earlier versions, where per-worker snapshots were opened independently and the position was captured afterward, so a change committed during snapshot setup could be absent from both the snapshot and the CDC stream.

Requirements (fail closed at preflight when a snapshot runs): gtid_mode = ON, binlog_format = ROW, all snapshotted tables InnoDB, and the global RELOAD privilege.

Managed MySQL (RDS/Aurora and similar). Where FLUSH TABLES WITH READ LOCK is unavailable (RELOAD restricted), the snapshot fails closed with a typed permission error - DeltaForge does not silently fall back to an unsafe anchor. Remediation:

• grant the global RELOAD privilege if your provider allows it; or
• run without an initial load by setting snapshot.mode = never to stream only changes from the current position (this is not a complete initial load); or
• perform the initial load out of band (e.g. a provider snapshot/dump) and start DeltaForge in never mode.

Additional privileges

-- Required for snapshot: SELECT (introspection + reads) and global RELOAD.
GRANT SELECT ON your_database.* TO 'deltaforge'@'%';
GRANT RELOAD ON *.* TO 'deltaforge'@'%';

Configuration

source:
  type: mysql
  config:
    id: orders-mysql
    dsn: ${MYSQL_DSN}
    tables:
      - shop.orders
    snapshot:
      mode: initial           # initial | always | never (default: never)
      max_parallel_tables: 8  # tables snapshotted concurrently (also caps workers under the lock)
      chunk_size: 10000       # rows per chunk for integer-PK tables
      lock_timeout_secs: 10   # bound on the brief FLUSH TABLES WITH READ LOCK setup window

Snapshot events

Snapshot rows are emitted as Op::Read events (Debezium op: "r"), distinguishable from live CDC Op::Create events. The binlog position captured at snapshot time becomes the CDC resume point, so no rows are missed. A snapshot copied again after a restart delivers its rows again (at-least-once).

Resume after interruption

An interrupted snapshot is copied again in full on the next restart, as a new snapshot generation with a new anchor; it is never resumed table by table. Losing any worker’s consistent-snapshot connection loses the generation’s read view and replaces it the same way.

A snapshot counts as complete only once its terminal barrier is committed by the sinks the commit policy requires; a restart after that completes it without copying. Completion, lagging sinks, bounds and blocking are described in Initial Snapshots.

Binlog retention safety

DeltaForge validates binlog retention before starting a snapshot and monitors it throughout. This prevents the silent failure mode where a long snapshot completes successfully but CDC startup then fails because the captured binlog position was purged.

Preflight checks (before any rows are read):

• Fails hard if log_bin=0 or binlog_format != ROW
• Estimates snapshot duration from table sizes and max_parallel_tables
• Warns at ≥50% of binlog_expire_logs_seconds usage; HIGH RISK at ≥80%

During snapshot:

• Background task polls SHOW BINARY LOGS every 30s
• Cancels the snapshot immediately if the captured file is purged

After all tables complete:

• Synchronous final check before the generation records its rows as produced: a snapshot whose anchor binlog file was purged never completes, and blocks with snapshot_anchor_unavailable (see Bounds and blocking)

If you see retention risk warnings, the recommended actions are:

1. Increase binlog_expire_logs_seconds to cover the estimated snapshot duration
2. Use a read replica as the snapshot source to avoid load on the primary

Server ID Handling

MySQL replication requires each replica to have a unique server_id. DeltaForge derives this automatically from the source id using a CRC32 hash:

server_id = 1 + (CRC32(id) % 4,000,000,000)

When running multiple DeltaForge instances against the same MySQL server, ensure each has a unique id to avoid server_id conflicts.

Schema Tracking

DeltaForge has a built-in schema registry to track table schemas, per source. For MySQL source:

• Schemas are loaded when a table is first used, not at startup: startup work does not grow with the number of matched tables (only an initial snapshot reads the tables it copies)
• Each schema is fingerprinted using SHA-256 for change detection
• Events carry schema_version (the fingerprint of the version the row was decoded with) and schema_sequence (that version’s registry sequence)
• Schema-to-checkpoint correlation enables reliable replay

A DDL forgets the cached shapes of the tables it affects; their new versions are registered only by a positional proof (below), never reloaded from the live catalog. See Schema Registry: DDL Handling.

DDL events

Each DDL statement emits one event (source.table = "_ddl", operation r). Its payload (format 2) identifies what the statement acts on:

{
  "ddl_format": 2,
  "scope": "tables",
  "objects": [
    {"kind": "table", "database": "shop", "name": "orders"},
    {"kind": "table", "database": "archive", "name": "orders"}
  ],
  "database": null,
  "session_database": "shop",
  "sql": "RENAME TABLE shop.orders TO archive.orders"
}

• scope: what the statement does to base-table shapes:
  • tables: it may change the listed tables;
  • same_shape: TRUNCATE;
  • database: a database barrier;
  • lineage: unattributable;
  • none: no base table changes (views, account statements).
• objects: {kind, database, name} for each table, view or barrier database, in statement order. A rename lists source then target. The list is empty when the statement cannot be attributed. Under lower_case_table_names = 2 table statements are applied as lineage barriers but keep their objects.
• database and source.db: name the one database every object belongs to. A cross-database statement or one without objects has database: null and source.db: "".
• session_database: the connection’s active database when the statement ran (null when none). It is context only, never the attribution.
• Account statements (CREATE USER, ALTER USER, SET PASSWORD) are scope: none with no objects. Their SQL is never emitted (operation, redacted: true instead).
• No DDL event, as before: plain GRANT and REVOKE.

Rows are decoded with the definition in effect when they were written

Binlog row events carry values by column position, not by name. DeltaForge decodes each rows event with the schema version proven for its position in the binlog, never simply with the table’s current definition, so replaying retained binlog across a DDL (after a restart from an older checkpoint, or while catching up) puts every value under the right column. A version is proven by:

• the table’s definition captured when its first rows without other proof arrive, and proven unchanged between those rows and the capture (no DDL of the table and no unattributable statement in between);
• a DDL’s resulting definition, captured right after the source reads the DDL and proven unchanged since;
• with binlog_row_metadata = FULL and neither of the above: the single recorded definition that exactly matches the row’s binlog metadata (column names, types, signedness, charsets, primary key), recorded durably before the row is emitted.

Capturing a definition never requires the server to be idle. In one read-only transaction the source takes the table’s shared metadata lock (the lock any SELECT takes: inserts, updates and deletes proceed, DDL of that table or its database waits), then reads the binlog position, the table’s definition (several INFORMATION_SCHEMA queries) and the position again, then releases the lock. A DDL already running on the table finishes before the lock is granted, so the definition read is never older than the position; a DDL issued meanwhile waits a few milliseconds and lands after the capture. The source then proves from the binlog that nothing between the two positions could have changed that table, such as an unattributable statement. Ordinary writes, to any table, and DDL of other tables do not matter, so a busy server proves tables as readily as a quiet one. Only a relevant statement inside that short interval causes a retry; a binlog that cannot be read (purged, unknown events, scan limits) or a lock not granted within 30 seconds stops the proof instead. The capture needs SELECT on the table. Table and database names are compared as lower_case_table_names defines: exactly with 0, case-insensitively with 1, and with 2 (names stored as written, compared case-insensitively) every DDL counts as relevant.

When no version can be proven, the source stops with a schema error naming the table, emits nothing for those rows and does not advance its checkpoint. Under the default binlog_row_metadata = MINIMAL this happens when:

• retained rows predate a DDL the source never observed (for example, a checkpoint restored from before a schema change made by an earlier deployment);
• a statement DeltaForge cannot attribute to specific tables was executed (versioned comments such as /*!50100 ALTER TABLE ... */, several statements in one event, ALTER DATABASE or DROP DATABASE of a tracked database, among others): every table it may affect stays unproven until later proof for that table exists - usually its next rows, once the source captures the table’s definition and proves nothing changed it since them, or a later DDL of the table that the source proves - or a re-snapshot;
• a table’s first rows (or its first rows after an unattributable statement) are followed, before the source reads them, by a DDL of that table or another unattributable statement: the definition those rows were written under can no longer be captured (this needs the source to lag behind the database);
• two DDLs of one table both happen before the source reads the first, with rows written between them (the definition between the two was never captured; with binlog_row_metadata = FULL those rows proceed only if that definition was already recorded and is the single exact match).

Re-snapshot is the immediate remediation that always works.

Many newly active tables at once. When rows of a table without proof arrive, the source first reads ahead (an activation window, at most proofs.activation_window_ms, without advancing its position or emitting anything) and collects every other table whose first rows follow. A statement that can change a schema, or an idle stream, closes the window early. One read-only transaction then captures all of these tables: it locks their metadata in sorted order, which avoids deadlocking against a multi-table DDL, and reads every definition between the same two binlog positions. Meanwhile the source keeps reading the stream, holding what arrives for the next window without emitting it, so tables first touched during an admission are not held for a second full window. The next window’s capture and proof scan also start while the current window is still being admitted. It is decided and written only after the current window commits, in stream order: tables it captured that turn out to be proven already are dropped, and any table it should have captured but did not stops the source (an internal error, never repaired by guessing). At most two windows exist at a time. One binlog scan proves them all, and each table is admitted or refused exactly as it would be on its own. The admitted tables’ schema versions and proofs are then written in one state-store transaction: all of them or none. A failover drift check runs for every table of the window first, so a drift incident stops the whole window with nothing of it written. If the store cannot commit the window (after a few retries of the whole transaction), the source stops; it never falls back to admitting the tables one by one. If the locks are not granted within proofs.capture_lock_wait_secs on three attempts, the source stops with a mysql_capture_contention incident: nothing of the window is emitted, the checkpoint stays before it, and a restart retries once the schema change holding the lock is done. There is no fallback to one capture per table.

Proof scans start where their interval starts. A proof scan normally opens the binlog at a position the source observed in its own stream, either the transaction of the rows or the end of the last scan, and checks the first events the server sends before trusting it. Without such a position, for example a snapshot baseline, the server seeks from the start of the binlog file. That seek is bounded by proofs.max_seek_ms: a seek that exceeds it stops the proof rather than wait, and the position the scan then observes positions the next proof at the same boundary. GTID membership still decides what a proof contains, so the start position never changes a proof, only how long it takes.

Recommended: set binlog_row_metadata = FULL on the server. It removes the first two cases whenever the row’s definition is still recorded, at a small cost in binlog size.

Schema versions record per-column detail (character octet length, collation, fractional-seconds precision, primary-key prefix). After upgrading from a release that did not, the first capture of each table may register one new version even without a DDL; events then carry that version’s schema_version and schema_sequence.

Timeouts and Heartbeats

These values are currently fixed. Reconnection uses exponential backoff with jitter.

Event Format

Each captured row change produces an event with:

• op: insert, update, or delete
• before: Previous row state (updates and deletes only, requires binlog_row_image = FULL)
• after: New row state (inserts and updates only)
• table: Fully qualified table name (database.table)
• tx_id: GTID if available
• checkpoint: Binlog position for resume
• schema_version: Schema fingerprint
• schema_sequence: Monotonic sequence for schema correlation

Troubleshooting

Connection Issues

DeltaForge authenticates with the server’s default plugin (caching_sha2_password on MySQL 8.x); no particular plugin is required. MySQL 8.4 disables mysql_native_password by default, so a user created IDENTIFIED WITH mysql_native_password fails to log in with an error mentioning that plugin. Move the user to the server’s default plugin:

ALTER USER 'deltaforge'@'%' IDENTIFIED BY 'password';

Missing Before-Images

If UPDATE/DELETE events have incomplete before data:

SET GLOBAL binlog_row_image = 'FULL';

Binlog Not Enabled

Check binary logging status:

SHOW VARIABLES LIKE 'log_bin';
SHOW VARIABLES LIKE 'binlog_format';
SHOW BINARY LOG STATUS;  -- or SHOW MASTER STATUS on older versions

Privilege Issues

Verify grants for your user:

SHOW GRANTS FOR 'deltaforge'@'%';

Required grants include REPLICATION SLAVE and REPLICATION CLIENT, plus BACKUP_ADMIN and SELECT on performance_schema.log_status for a first start without a checkpoint and without a snapshot.

PostgreSQL source

DeltaForge captures row-level changes from PostgreSQL using logical replication with the pgoutput plugin.

Prerequisites

PostgreSQL Server Configuration

Enable logical replication in postgresql.conf:

# Required settings
wal_level = logical
max_replication_slots = 10    # At least 1 per DeltaForge pipeline
max_wal_senders = 10          # At least 1 per DeltaForge pipeline

Restart PostgreSQL after changing these settings.

User Privileges

Create a replication user with the required privileges:

-- Create user with replication capability
CREATE ROLE deltaforge WITH LOGIN REPLICATION PASSWORD 'your_password';

-- Grant connect access
GRANT CONNECT ON DATABASE your_database TO deltaforge;

-- Grant schema usage and table access for schema introspection
GRANT USAGE ON SCHEMA public TO deltaforge;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO deltaforge;

-- The replication SLOT is auto-created on first run using the REPLICATION
-- attribute granted above. The PUBLICATION is NOT auto-created — create it
-- yourself (see "Replication Slot and Publication" below).

The same role reads the catalog for capture-time annotations and checks. Each such read first proves, in its own snapshot, that it reaches the node and the live walsender serving the stream: it reads pg_replication_slots, pg_stat_activity (the walsender’s row; visible to the same role, otherwise grant pg_read_all_stats) and pg_control_system(). A capture takes the table’s ACCESS SHARE lock (granted by SELECT on the table) for a few milliseconds: inserts, updates and deletes are not blocked, DDL on that table waits until the capture ends.

Unsupported: a tracked table with a stored generated column (GENERATED ALWAYS AS ... STORED) stops the source before its next row (pg_table_unsupported): pgoutput does not publish generated columns. Exclude the table or drop the generation.

pg_hba.conf

Ensure your pg_hba.conf allows replication connections:

# TYPE  DATABASE        USER            ADDRESS                 METHOD
host    replication     deltaforge      0.0.0.0/0               scram-sha-256
host    your_database   deltaforge      0.0.0.0/0               scram-sha-256

Replication Slot and Publication

DeltaForge creates the replication slot on first run and records durable ownership of it (bound to the server’s system_identifier, the database, the slot name, and the plugin). The publication is never created by DeltaForge: create it, then register it.

CREATE PUBLICATION my_pub FOR TABLE public.orders, public.order_items;

The publication contract

A publication DeltaForge streams from is immutable while it is registered. Registration is a superuser step, done before the source’s first start:

deltaforge pg-publication --dsn-file superuser.dsn register --publication my_pub
# review the database-wide impact printed, then repeat with --confirm

Registering a publication:

• requires an explicit FOR TABLE list of ordinary tables, with all four publish operations, no row filter, no column list, no publish_via_partition_root and no publish_generated_columns. FOR ALL TABLES, TABLES IN SCHEMA and partitioned tables are refused: their membership changes without publication DDL;
• transfers the publication to the NOLOGIN role deltaforge_publication_owner and records its canonical digest and the WAL position of the registration;
• installs enforcement for the whole database. While any publication of the database is registered, every ALTER PUBLICATION and DROP PUBLICATION is refused for every role, superusers included, and any drop that would remove a registered publication’s table (DROP TABLE, DROP SCHEMA ... CASCADE, DROP OWNED, …) is refused inside its transaction. The database enforces this while DeltaForge is stopped.

At startup and on every reconnect, before it snapshots, decodes, emits or advances a checkpoint, the source verifies that enforcement is intact, the publication is registered and its live digest equals the registration it accepted, and that it does not stream from WAL older than the registration. Otherwise it stops with pg_publication_changed or pg_publication_enforcement.

Changing a publication (membership, options, a table drop) is a database-wide maintenance procedure:

1. stop every DeltaForge PostgreSQL source using the database;
2. for each, record how it continues: deltaforge recover apply <pipeline> pg-publication-maintenance --arg mode=resnapshot (then the resnapshot operation) or --arg mode=abandon (the backlog before the new registration is skipped and never delivered; the applied plan is the audit record);
3. deltaforge pg-publication ... unregister --publication <each>;
4. deltaforge pg-publication ... uninstall (refused while any registration remains, naming them);
5. apply the change;
6. register the publications again (new digests);
7. restart the sources.

Outside the guarantee: a superuser can disable or drop the enforcement triggers. The source detects it at its next startup or reconnect and stops, but changes made between the tampering and that check are not guaranteed to have been captured completely.

Replica Identity

For complete before-images on UPDATE and DELETE operations, set tables to REPLICA IDENTITY FULL:

ALTER TABLE public.orders REPLICA IDENTITY FULL;
ALTER TABLE public.order_items REPLICA IDENTITY FULL;

Without this setting:

• FULL: Complete row data in before-images
• DEFAULT (primary key): Only primary key columns in before-images
• NOTHING: No before-images at all

DeltaForge warns at startup if tables don’t have REPLICA IDENTITY FULL.

Unchanged TOAST values

PostgreSQL stores large values out of line (TOAST). When an UPDATE leaves such a value unchanged, PostgreSQL does not send it in the new row.

• Under REPLICA IDENTITY FULL PostgreSQL sends the complete old row, including those values (verified on PostgreSQL 14 through 18). DeltaForge fills each unchanged value from the old row, exactly. The event’s after is complete.
• Otherwise the value is not available. Sinks that accept partial rows (native JSON: Kafka, Redis, NATS, HTTP, S3 JSON Lines) leave it out of after and list it in unchanged_columns.

Sinks that need complete rows (Avro encoding on any sink, S3 Parquet, ClickHouse, Elasticsearch) cannot represent a missing value. A pipeline with at least one of them requires REPLICA IDENTITY FULL on every captured table that has a TOAST-capable column (any variable-length type stored with a strategy other than PLAIN, domains and arrays included, read from the catalog).

• At start: a bulk check over the publication’s tables, before anything is snapshotted or streamed, refuses the pipeline with pg_table_unsupported (class toast_without_full_identity). The source log names every such table with its TOAST-capable columns and the statement that fixes it, for example ALTER TABLE "public"."docs" REPLICA IDENTITY FULL;. The refusal contains table and column names only, never row values, and the checkpoint does not move.
• While running: the check is repeated whenever PostgreSQL announces a table’s definition (stream start, after DDL, after reconnect or failover), so changing a table’s identity away from FULL stops the source before that table’s next row.

A pipeline whose sinks all accept partial rows does not require FULL; one sink that needs complete rows is enough to require it. deltaforge_pg_unchanged_toast_total{pipeline,source,outcome} counts values filled from the old row (reconstructed) and values left unavailable (unavailable).

WAL cost of FULL. Under FULL, every UPDATE and DELETE writes the complete old row to the WAL, and the old row includes its out-of-line values even when the UPDATE leaves them unchanged. The cost grows with row width:

More WAL also means more WAL decoded and sent to the source, and more retained for the slot while the source is behind. For wide, frequently updated tables, weigh this against delivering those tables only to sinks that accept partial rows.

Configuration

Set spec.source.type to postgres and provide a config object:

Message size limit

PostgreSQL sends each row change as one replication message, so a single wide row (or a large pg_logical_emit_message() payload) must fit under max_message_bytes. A larger message is refused before any of it is buffered, and the source stops fail-closed with the incident pg_message_too_large:

• the session that received it is abandoned and the source does not reconnect: a reconnect with the same limit would meet the same message again;
• the checkpoint stays before the message’s transaction, so nothing of it is acknowledged or delivered partially;
• the evidence is the message’s size (message_bytes), the limit (message_limit) and the source, never the payload.

Raise max_message_bytes above the reported size and restart the pipeline: it resumes from the held checkpoint and delivers the whole transaction. The limit bounds one message, not a transaction or total memory.

DSN Formats

DeltaForge accepts both URL-style and key=value DSN formats:

# URL style
dsn: "postgres://user:pass@localhost:5432/mydb"

# Key=value style
dsn: "host=localhost port=5432 user=deltaforge password=pass dbname=mydb"

Table Patterns

The tables field supports flexible pattern matching:

tables:
  - public.orders          # exact match: schema "public", table "orders"
  - public.order_%         # prefix: tables starting with "order_" (same as public.order_*)
  - myschema.*             # wildcard: all tables in "myschema"
  - %.audit_log            # cross-schema: "audit_log" table in any schema
  - orders                 # table "orders" in every schema

A wildcard (* or %) only counts as the last character of a part (a prefix match); _ and any other character are literal. A pattern without a qualifier matches that table name in every database/schema. The initial snapshot copies exactly the tables these patterns capture during CDC.

System schemas (pg_catalog, information_schema, pg_toast) are always excluded.

Start Position

⚠️ Not yet implemented. start_position is accepted by the config parser but currently ignored — a new slot always starts from its consistent point (the current WAL position when it is created). The forms below are aspirational.

Intended to control where replication begins when no checkpoint exists:

# Start from the earliest available position (slot's restart_lsn)
start_position: earliest

# Start from current WAL position (skip existing data)
start_position: latest

# Start from a specific LSN
start_position:
  lsn: "0/16B6C50"

Example

source:
  type: postgres
  config:
    id: orders-postgres
    dsn: ${POSTGRES_DSN}
    slot: deltaforge_orders
    publication: orders_pub
    tables:
      - public.orders
      - public.order_items

Resume Behavior

DeltaForge checkpoints progress using PostgreSQL’s LSN (Log Sequence Number):

1. With checkpoint: Resumes from the stored LSN
2. Without checkpoint: Uses the slot’s confirmed_flush_lsn or restart_lsn
3. New slot: Starts from the slot’s consistent point (the LSN its creation returns): changes committed after it are captured.

The pipeline reports running only once the slot exists, the start position is persisted for every sink and the stream is open, with that position as capture_start. A retry or restart before then starts from the slot again, never from a newer position.

Checkpoints are stored using the id field as the key.

Snapshot (Initial Load)

DeltaForge performs a consistent initial snapshot using PostgreSQL’s exported snapshot mechanism before starting logical replication.

How it works

DeltaForge anchors CDC at the replication slot’s consistent point C - the LSN pg_create_logical_replication_slot returns when the slot is created - rather than a separately sampled pg_current_wal_lsn(). A coordinator connection then exports an MVCC snapshot for worker mutual consistency; worker connections import it into their own REPEATABLE READ transactions, so all workers see one consistent DB state with no locks held on the source. Tables with a single integer primary key use PK-range chunking; others fall back to ctid page-range chunking.

Anchoring at C (which precedes the snapshot read) closes the snapshot-to-CDC seam: no committed row is lost. The trade-off is a bounded at-least-once overlap - a change committed between C and the snapshot export is present in both the snapshot and the CDC stream from C, so it is delivered more than once. This is at-least-once, not exactly-once (see Snapshot events).

Correctness note. Earlier versions sampled pg_current_wal_lsn() decoupled from the exported snapshot, which admitted a (small) window where a row could be absent from both the snapshot and the CDC stream. The slot-consistent-point anchor removes that window, trading the loss risk for the bounded overlap above.

Configuration

source:
  type: postgres
  config:
    id: orders-postgres
    dsn: ${POSTGRES_DSN}
    slot: deltaforge_orders
    publication: orders_pub
    tables:
      - public.orders
    snapshot:
      mode: initial           # initial | always | never (default: never)
      max_parallel_tables: 8  # tables snapshotted concurrently
      chunk_size: 10000       # rows per chunk for integer-PK tables

Snapshot events

Snapshot rows are emitted as Op::Read events (Debezium op: "r"), distinguishable from live CDC Op::Create events. The slot’s consistent point C is the CDC resume position, so no rows are missed. Rows committed in (C, snapshot-export] are delivered by both the snapshot (as Op::Read) and the CDC stream (as their change op) - the bounded at-least-once overlap. The two copies carry distinct event identities:

• Current-state / idempotent sinks (Elasticsearch, ClickHouse upsert, with version_source: source_position) converge to the correct current state - the duplicate is absorbed by last-writer-wins.
• Append-only sinks (Kafka, Redis, HTTP, ClickHouse changelog, legacy S3) receive the overlapping rows twice, by design.

This is at-least-once delivery; DeltaForge does not claim exactly-once for the snapshot-to-CDC boundary.

Legacy-anchor warning and metric

A pipeline whose initial snapshot was completed under the older (pre-hardening) anchor is flagged so you can decide to re-snapshot:

• a structured WARN at startup, and
• the gauge deltaforge_snapshot_unsafe_anchor{pipeline,source} held at 1.

Re-snapshotting under this version records the safe anchor and resets the gauge to 0. See Upgrade guidance.

Resume after interruption

An interrupted snapshot is copied again in full on the next restart, as a new snapshot generation with a new anchor; it is never resumed table by table. When DeltaForge can prove it owns the now-inactive slot, it re-anchors (drops and recreates its slot for a fresh C).

A snapshot counts as complete only once its terminal barrier is committed by the sinks the commit policy requires; a restart after that completes it without copying. Completion, lagging sinks, bounds and blocking are described in Initial Snapshots.

WAL slot retention safety

DeltaForge validates replication slot health before starting a snapshot and monitors it throughout. This prevents the slot from being invalidated during a long snapshot, which would make the captured LSN unreachable for CDC resume.

Preflight checks (before any rows are read):

• Fails hard if the slot does not exist or is already invalidated
• Fails hard if wal_status=lost
• Warns if wal_status=unreserved (WAL retention no longer guaranteed)
• Estimates WAL generated during snapshot (~2× data size) against max_slot_wal_keep_size; warns at ≥50%, HIGH RISK at ≥80%

During snapshot:

• Background task polls pg_replication_slots every 30s
• Cancels immediately on slot invalidation or disappearance
• Warns but continues on wal_status=unreserved

After all tables complete:

• Synchronous final check before the generation records its rows as produced: a snapshot whose slot is lost, invalidated or missing never completes, and blocks with snapshot_anchor_unavailable (see Bounds and blocking)

If you see WAL retention risk warnings:

ALTER SYSTEM SET max_slot_wal_keep_size = '10GB';
SELECT pg_reload_conf();

Upgrade guidance

The slot-consistent-point anchor and durable slot ownership change how existing slots are handled when a snapshot runs. Steady-state CDC resume from a checkpoint is unaffected - the notes below apply only when a snapshot runs.

• Existing slots created before this version have no ownership record. A re-snapshot on such a slot (mode: always, or mode: initial after clearing the checkpoint) cannot prove ownership and fails closed with remediation: drop the slot manually (SELECT pg_drop_replication_slot('<slot>')) so DeltaForge recreates it with ownership, then restart. New pipelines and first-run snapshots are unaffected.
• Ownership ambiguity / foreign slots. If the slot exists but DeltaForge cannot prove exclusive ownership (missing, partial, or mismatched record - e.g. a different server system_identifier after a restore/failover) or the slot is active, it fails closed and never drops the slot. Remediation: confirm no other consumer uses it, drop it, and restart; or set snapshot.mode = never to stream from the current position without an initial load.
• Interrupted snapshots. An owned, inactive slot left by a snapshot that was interrupted before its first checkpoint is re-anchored (dropped and recreated for a fresh consistent point) and fully re-snapshotted automatically.
• Safe re-snapshotting. To take a fresh, correctly-anchored snapshot, stop the pipeline and apply the resnapshot recovery operation (or set mode: always), then resume. Checkpoints are kept; a slot this source owns is re-anchored, and a lost one is recreated only under the operation’s explicit authorization. A completed, safely-anchored snapshot resets deltaforge_snapshot_unsafe_anchor to 0.

Type Handling

DeltaForge preserves PostgreSQL’s native type semantics:

Event Format

Each captured row change produces an event with:

• op: insert, update, delete, or truncate
• before: Previous row state (updates and deletes, requires appropriate replica identity)
• after: New row state (inserts and updates)
• table: Fully qualified table name (schema.table)
• tx_id: PostgreSQL transaction ID (xid)
• checkpoint: LSN position for resume
• schema_version: Schema fingerprint
• schema_sequence: Monotonic sequence for schema correlation

WAL Management

Logical replication slots prevent WAL segments from being recycled until the consumer confirms receipt. To avoid disk space issues:

1. Monitor slot lag: Check pg_replication_slots.restart_lsn vs pg_current_wal_lsn()
2. Set retention limits: Configure max_slot_wal_keep_size (PostgreSQL 13+)
3. Handle stale slots: Drop unused slots with pg_drop_replication_slot('slot_name')

-- Check slot status and lag
SELECT slot_name, 
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) as lag
FROM pg_replication_slots;

Troubleshooting

Connection Issues

If you see authentication errors:

-- Verify user has replication privilege
SELECT rolname, rolreplication FROM pg_roles WHERE rolname = 'deltaforge';

-- Check pg_hba.conf allows replication connections
-- Ensure the line type includes "replication" database

Missing Before-Images

If UPDATE/DELETE events have incomplete before data:

-- Check current replica identity
SELECT relname, relreplident 
FROM pg_class 
WHERE relname = 'your_table';
-- d = default, n = nothing, f = full, i = index

-- Set to FULL for complete before-images
ALTER TABLE your_table REPLICA IDENTITY FULL;

Slot/Publication Not Found

-- List existing publications
SELECT * FROM pg_publication;

-- List existing slots
SELECT * FROM pg_replication_slots;

-- Create if missing
CREATE PUBLICATION my_pub FOR TABLE public.orders;
SELECT pg_create_logical_replication_slot('my_slot', 'pgoutput');

WAL Disk Usage Growing

-- Check slot lag
SELECT slot_name, 
       active,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) as lag
FROM pg_replication_slots;

-- If slot is inactive and not needed, drop it
SELECT pg_drop_replication_slot('unused_slot');

Logical Replication Not Enabled

-- Check wal_level
SHOW wal_level;  -- Should be 'logical'

-- If not, update postgresql.conf and restart PostgreSQL
-- wal_level = logical

Initial Snapshots

An initial snapshot copies the existing rows of every captured table, then hands over to the change stream at a single anchor. This page covers what PostgreSQL and MySQL share: generations, restarts, completion under the commit policy, bounds and blocking, connection limits, the upgrade from earlier releases, and measured scale. The engine-specific mechanics (anchors, locks, privileges) are on the MySQL and PostgreSQL pages.

Generations

Each snapshot is a generation: one complete copy of one sealed table plan, read in one database read view and anchored at one stream position.

• The source discovers its tables in catalog pages and writes each table to a durable plan in the state store. The plan is sealed at the anchor and never changes.
• Before any row of a generation is published, every sink of the pipeline moves its own checkpoint into that generation (the start barrier). Every sink must accept, including optional ones. A sink that is unavailable, or whose stored position cannot move into the generation (another snapshot chain, a later generation, an unknown format), fails the start: the pipeline stops, and each later start tries again with a new generation.
• Rows are published chunk by chunk. Each boundary records a constant-size snapshot position, not a per-table cursor.
• When every table is read and every final check passed, the source sends the terminal barrier, and each sink commits it once all earlier rows are durably delivered.
• The generation is completed when the sinks that acknowledged the terminal satisfy the commit policy. The stream then continues from the anchor.

Generation numbers never repeat. Generations of one source belong to a snapshot chain, so a sink can tell a newer generation from an older one; positions of different chains are never compared.

Restarts copy the whole snapshot again

A generation lives and dies with its read view. An interrupted snapshot is never resumed table by table or chunk by chunk: a process restart, a lost worker connection or a lost read view replaces the generation, and the next generation reads every table again under a new anchor. Sinks receive the rows already sent once more (at-least-once; never loss). The replacement raises a non-blocking snapshot_replaced incident.

The exceptions:

• If the terminal barrier was already acknowledged by enough sinks to satisfy the frozen policy (a crash after the acknowledgement), the restart completes the generation without copying and streams.
• A blocked generation is never replaced automatically (see Blocking).
• A completed generation streams changes and never copies rows, except with snapshot.mode: always (a new generation on every start).
• A change to the sink cohort or commit policy before completion replaces the generation (snapshot_replaced, class policy_changed): a terminal is only judged by the cohort that started it.

Plan restart time accordingly: a crash late in a long snapshot costs a full copy.

Completion and lagging sinks

At allocation the generation freezes the pipeline’s commit policy: the mode, the quorum and every sink with its required flag. Completion is judged only against that frozen cohort:

The completion record keeps the acknowledging sinks; later configuration changes never reinterpret it.

A sink that lags, failing or missing a batch of the generation, does not hold completion under required or quorum:

• A sink that missed any batch since the generation started cannot acknowledge its terminal: it would claim rows it never delivered. It is left behind the terminal.
• Once the generation completes, every sink behind it gets a non-blocking sink_snapshot_incomplete incident naming the sink and generation, with the action rebootstrap_sink. It is excluded from the resume position, so it never pulls the stream back into the snapshot, and it continues with changes from the resume position. It does not receive the rest of that snapshot: give it a fresh baseline (a new generation, or an out-of-band load).
• Under all, a failing sink stops the pipeline instead. Once the sink recovers and the pipeline resumes, the next start replaces the generation and every sink receives the whole snapshot.

A sink added after completion starts behind it and is treated the same way.

Bounds and blocking

Every generation runs inside explicit bounds. Crossing 80% of a bound raises a non-blocking snapshot_bound_warning. Crossing the bound stops the generation before anything is lost:

The plan bound is the state store’s resource; WAL and binlog retention are the source’s. They are separate settings and separate incidents.

Two related stops are not bounds:

• Corrupt or unknown-format snapshot state stops the source with snapshot_state_invalid and is left untouched; every start stops on it again.
• A second process found owning the source before a publish (concurrent_owner) stops only the stale run, with a non-blocking incident; it never changes the current owner’s generation. One owner per source remains a deployment requirement (see Supported Deployment Envelope).

A blocked generation stays halted across restarts. Every start re-raises its incident and stops before any row or replacement. These failures need an operator decision (raise the bound, narrow the tables, restore retention), so nothing retries them automatically.

Recovery is the explicit, proof-bound resnapshot recovery operation, which replaces the blocked generation with a recovery allocation of the next one, audited with actor and reason. Raise the bound or restore what the incident names first; then plan, review and apply it, and resume the pipeline.

Connection limits

Every connection a snapshot opens (PostgreSQL coordinator and workers; MySQL lock connection, binlog guard and workers; intra-table readers) is taken from one process-wide pool:

• --max-snapshot-connections (default 64) caps snapshot connections across every pipeline of the process.
• snapshot.max_snapshot_connections caps one source’s share (default max_parallel_tables x max_parallel_chunks + 2).
• A snapshot takes its base connections (3 on either engine) atomically, before opening any transaction, lock or anchor. Extra workers are taken when free and released when done; the base worker always progresses.
• A snapshot that cannot get its base queues, holding nothing, and can be stopped while queued. Queued snapshots are visible in deltaforge_snapshot_queued{pipeline}; the pool in deltaforge_snapshot_connection_permits{state}.
• A source whose share cannot hold its base, or whose share exceeds the process cap, fails its configuration check.

Oversubscribing the process cap with many sources is allowed: their snapshots queue.

Upgrading from an earlier release: one-time recopy

Snapshot state from releases before durable generations is classified at the first start after the upgrade, with its source lineage verified first:

• Proven complete (the sink checkpoints prove the completion under the commit policy): upgraded in place; nothing is copied.
• Not proven (an interrupted snapshot, sinks behind its anchor, a MySQL progress record without its anchor, and every PostgreSQL snapshot taken before the slot-anchor hardening): copied once more, in full, as the first generation, before streaming continues. A non-blocking snapshot_replaced incident (class legacy_recopy) records it.

This recopy reads every table again and can take as long as the original initial load. Plan the upgrade window for large catalogs. With snapshot.mode: never an unproven snapshot stops the source with an actionable error instead. State of an unknown format is refused and left untouched.

Rollback to an earlier release is safe only once every sink checkpoint of the source is a stream position (after completion, and after every sink committed a change past it). Before that, an earlier release refuses the new state rather than misreading it.

Scale

What is bounded

The durable queue bounds the snapshot’s own state, not the whole process:

• Per boundary: one constant-size position (about 190 bytes on PostgreSQL, 240 on MySQL), independent of the table count. No per-table cursor vector and no progress record are written.
• Durable operations per generation (counted on the storage path at 10, 30 and 90 tables): the control record is created once and updated 4 times; each table’s plan entry is written once and deleted once after completion; one control read per publish (every chunk and both barriers) checks ownership; a constant number of other control reads; plan reads are whole page walks, linear in pages.
• Resident queue state: at most one discovery page of plan entries (discovery_page_size, default 1,000) and at most max_parallel_tables table tasks (default 8). The plan itself stays in the state store.

The whole process is not bounded by the queue. Schema caches and schema registry state grow with the tables a source uses, each within its own bounds (see Capacity envelope). Do not read the figures below as constant total memory.

1K and 10K tables

Measured once on the release candidate: single-row tables, default snapshot settings, a debug build, SQLite state store and schema registry, PostgreSQL 17 and MySQL 8.4 in local containers, on an Intel i7-1355U laptop (12 threads, 31 GB). Indicative, not a guarantee:

No progress writes in any run. Acceptance gates (10K against 1K), all met:

Peak RSS growth during the snapshot, reported, not gated: PostgreSQL 2.8 MiB at 1K and 2.5 MiB at 10K; MySQL 17.9 MiB at 1K and 48.4 MiB at 10K (2.7x). The MySQL growth is schema state, not queue state: each table’s schema is loaded once while the plan is prepared, into the source’s schema cache (up to 4,096 tables or about 64 MiB) and the schema registry’s cache, plus a compact per-table record. It grows with the tables captured, within those caches’ bounds.

For comparison, the same measurement before the durable queue (in-memory checkpoint store, same machine) took 505 s for 10,000 MySQL tables and wrote 9.6 GiB of boundaries and 1.2 GiB of progress.

PostgreSQL catalog queries

The measurement also found a quadratic cost in PostgreSQL schema loading: the query reading a table’s columns through information_schema.columns scanned the whole catalog on every call (44 ms per table at 10,000 tables). It now also filters by the requested schema and table names, so PostgreSQL resolves the table through its catalog name index: 0.4 ms per table, with identical results. Preparing 10,000 PostgreSQL tables went from 440 s to 169 s. The same query serves CDC schema loads.

What is not claimed

These measurements cover 1,000 and 10,000 tables per source. Qualification at up to 400,000 tables per source, and for fleets of 50 to 120 million tables, is a separate milestone after rc.1. It is not a current support claim. See Capacity envelope for the supported starting range.

Processors

Processors can transform, modify or take extra action per event batch between the source and sinks. They run in the order listed in spec.processors, and each receives the output of the previous one. A processor can modify events, filter them out, or add routing metadata.

Available Processors

JavaScript

Run arbitrary JavaScript against each event batch. Uses the V8 engine via deno_core for near-native speed.

processors:
  - type: javascript
    id: enrich
    inline: |
      function processBatch(events) {
        return events.map(e => {
          e.tags = ["processed"];
          return e;
        });
      }
    limits:
      cpu_ms: 50
      mem_mb: 128
      timeout_ms: 500

⚠️ limits are not currently enforced. The fields deserialize but the JS runtime applies only a fixed ~5s reply timeout per batch — there is no CPU-time or heap limiting yet.

Performance

JavaScript processing adds a fixed overhead per batch plus linear scaling per event. Benchmarks show ~70K+ events/sec throughput with typical transforms — roughly 61× slower than native Rust processors, but sufficient for most workloads. If you need higher throughput, keep transform logic simple or move heavy processing downstream.

Tips

• Return an empty array to drop all events in a batch.
• Return multiple copies of an event to fan out.
• 64-bit integers (BIGINT, bigint) reach scripts as exact decimal strings, so no value passes through a JavaScript number.

Values and types

Scripts see each event’s native JSON rendering (see Data Types). Column types stay outside the script, from the event’s schema version, and each returned row is typed again:

• a value the script left as it was comes back exactly as it was;
• a changed value of a table column (including one that was NULL or a PostgreSQL unchanged TOAST value) must be a valid value of the column’s declared type (ENUM and SET values must be declared labels), or the batch fails;
• a removed field is removed;
• a field the table does not have becomes a document extension: native JSON shows it in the row; typed sinks carry it in their extension slot (after_extensions, ClickHouse _extensions), never as a column.

If the event’s schema version cannot be resolved, a changed or added value fails the batch rather than losing its type.

Outbox

The outbox processor transforms events captured by the outbox pattern into routed, sink-ready events. It extracts business fields from the raw outbox payload, resolves the destination topic, and passes through all non-outbox events unchanged.

processors:
  - type: outbox
    topic: "${aggregate_type}.${event_type}"
    default_topic: "events.unrouted"

Column Defaults

See the Outbox Pattern guide for source configuration, complete examples, and multi-outbox routing.

Flatten

Flattens nested JSON objects in event payloads into top-level keys joined by a configurable separator. Works on every object-valued field present on the event without assuming any particular envelope structure - CDC before/after, outbox business payloads, or any custom fields introduced by upstream processors are all handled uniformly.

processors:
  - type: flatten
    id: flat
    separator: "__"
    max_depth: 3
    on_collision: last
    empty_object: preserve
    lists: preserve
    empty_list: drop

Input:

{
  "after": {
    "order_id": "abc",
    "customer": {
      "id": 1,
      "address": { "city": "Berlin", "zip": "10115" }
    },
    "tags": ["vip"],
    "meta": {}
  }
}

Output (with defaults — separator: "__", empty_object: preserve, lists: preserve):

{
  "after": {
    "order_id": "abc",
    "customer__id": 1,
    "customer__address__city": "Berlin",
    "customer__address__zip": "10115",
    "tags": ["vip"],
    "meta": {}
  }
}

Configuration

max_depth in practice

max_depth counts nesting levels from the top of the payload object. Objects at the boundary are treated as opaque leaves rather than expanded further, still subject to empty_object policy.

# max_depth: 2

depth 0: customer               -> object, recurse
depth 1: customer__address      -> object, recurse
depth 2: customer__address__geo -> STOP, kept as leaf {"lat": 52.5, "lng": 13.4}

Without max_depth, a deeply nested or recursive payload could produce a large number of keys. Setting a limit is recommended for payloads with variable or unknown nesting depth.

Collision policy

A collision occurs when two input paths produce the same flattened key - typically when a payload already contains a pre-flattened key (e.g. "a__b": 1) alongside a nested object ("a": {"b": 2}).

• last - the last path to write a key wins (default, never fails)
• first - the first path to write a key wins, subsequent writes are ignored
• error - the batch fails immediately, useful in strict pipelines where collisions indicate a schema problem

What flattens

Flatten expands JSON documents: outbox payload fields and values a processor added (document extensions, which flatten into extensions). A table’s own columns, a JSON column included, are typed values and are never expanded.

Working with outbox payloads

After the outbox processor runs, event.after holds the extracted business payload - there is no before. The flatten processor handles this naturally since it operates on whatever fields are present:

processors:
  - type: outbox
    topic: "${aggregate_type}.${event_type}"
  - type: flatten
    id: flat
    separator: "."
    empty_list: drop

Envelope interaction

The flatten processor runs on the raw Event struct before sink delivery. Envelope wrapping happens inside the sink, after all processors have run. This means the envelope always wraps already-flattened data, no special configuration needed.

Source → [flatten processor] → Sink (envelope → bytes)

All envelope formats work as expected:

// Native
{ "before": null, "after": { "customer__id": 1, "customer__address__city": "Berlin" }, "op": "c" }

// Debezium
{ "payload": { "before": null, "after": { "customer__id": 1, "customer__address__city": "Berlin" }, "op": "c" } }

// CloudEvents
{ "specversion": "1.0", ..., "data": { "before": null, "after": { "customer__id": 1, "customer__address__city": "Berlin" } } }

Outbox + raw_payload: true

When the outbox processor is configured with raw_payload: true, the sink delivers event.after directly, bypassing the envelope entirely. If the flatten processor runs after outbox, the raw payload delivered to the sink is the flattened object — which is the intended behavior for analytics sinks that can’t handle nested JSON.

processors:
  - type: outbox
    topic: "${aggregate_type}.${event_type}"
    raw_payload: true       # sink delivers event.after directly, no envelope
  - type: flatten
    id: flat
    separator: "__"
    empty_list: drop

The flatten processor runs second, so by the time the sink delivers the raw payload it is already flat.

Analytics sink example

When sending to column-oriented sinks (ClickHouse, BigQuery, S3 Parquet) that don’t handle nested JSON:

processors:
  - type: flatten
    id: flat
    separator: "__"
    lists: index          # expand arrays to indexed columns
    empty_object: drop    # remove sparse marker objects
    empty_list: drop      # remove empty arrays

Filter

Drops events that do not pass configured criteria. Each criterion is independent — omit any of them to skip that check entirely. An event must pass all configured checks to be forwarded.

processors:
  - type: filter
    id: only-active-orders
    ops: [create, update]
    tables:
      include: ["shop.orders"]
      exclude: ["*.tmp"]
    fields:
      - path: status
        op: eq
        value: "active"
      - path: total
        op: gte
        value: 100
    match: all

Field operators

The path field is a dot-separated path into event.after, e.g. "status" or "order.total".

Notes

Numeric equality — integers compare exactly at any width (BIGINT included, though native JSON renders it as a string); an integer and a float compare by value, so 42 and 42.0 are equal. A path can descend into a JSON column (payload.status). A predicate on a PostgreSQL column whose value an update did not send (unchanged TOAST) fails the batch; changed treats it as unchanged.

not_in with a missing field — if the field is absent from the event, the event passes. Absence is not membership in any set.

regex on non-string fields — silently does not match. Use exists first if the field may be absent or non-string.

changed and the before image — the predicate reads event.before, which is only populated for update operations from sources configured with full row images. Verify your source has REPLICA IDENTITY FULL (PostgreSQL) or binlog_row_image = FULL (MySQL) before using changed in production.

Performance

The filter processor is pure Rust with no serialization overhead. Op and table checks are O(1) to O(patterns). Field predicate evaluation reads event.after directly. Regex patterns are compiled once at construction time. Put a filter early in the processor chain to reduce the batch size before heavier processors (JavaScript, flatten) run.

Examples

Drop deletes and snapshot reads:

- type: filter
  id: no-deletes
  ops: [create, update]

Alert stream - pass if status is failed OR retry count exhausted:

- type: filter
  id: alert-worthy
  match: any
  fields:
    - path: status
      op: eq
      value: "failed"
    - path: retry_count
      op: gte
      value: 3

Filter before enrichment to avoid paying JavaScript overhead on the full stream:

processors:
  - type: filter
    id: low-stock
    ops: [create, update]
    tables:
      include: ["inventory.products"]
    fields:
      - path: stock_qty
        op: lt
        value: 10
  - type: javascript
    id: enrich
    inline: |
      function processBatch(events) {
        return events.map(e => {
          e.after.alert_level = e.after.stock_qty === 0 ? "critical" : "warning";
          return e;
        });
      }

Only forward rows where a status column actually changed value (suppresses no-op updates):

- type: filter
  id: real-status-changes
  ops: [update]
  fields:
    - path: status
      op: changed

Processor Chain

Processors execute in order. Events flow through each processor sequentially:

Source → [Processor 1] → [Processor 2] → ... → Sinks

Each processor receives a Vec<Event> and returns a Vec<Event>. This means processors can:

• Transform: Modify event fields in place
• Filter: Return a subset of events (drop unwanted ones)
• Fan-out: Return more events than received
• Route: Set event.routing.topic to override sink defaults
• Enrich: Add headers via event.routing.headers

Non-outbox events always pass through the outbox processor untouched, so it is safe to combine it with JavaScript processors in any order.

Adding Custom Processors

The processor interface is a simple trait:

#![allow(unused)]
fn main() {
#[async_trait]
pub trait Processor: Send + Sync {
    fn id(&self) -> &str;
    async fn process(&self, events: Vec<Event>) -> Result<Vec<Event>>;
}
}

To add a new processor:

1. Implement the Processor trait in crates/processors
2. Add a config variant to ProcessorCfg in deltaforge-config
3. Register the build step in build_processors()

Sinks

Sinks receive batches from the coordinator after processors run. Each sink lives under spec.sinks and can be marked as required or best-effort via the required flag. Checkpoint behavior is governed by the pipeline’s commit policy.

Envelope and Encoding

All sinks support configurable envelope formats and wire encodings. See the Envelopes and Encodings page for detailed documentation.

Quick example:

sinks:
  - type: kafka
    config:
      id: events-kafka
      brokers: localhost:9092
      topic: events
      envelope:
        type: cloudevents
        type_prefix: "com.example.cdc"
      encoding: json

Available Sinks

	Sink	Description
	kafka	Kafka producer sink
	nats	NATS JetStream sink
	redis	Redis stream sink
	http	HTTP/Webhook sink
	s3	S3-compatible object storage (Parquet / JSON Lines) — AWS S3, MinIO, GCS, Azure, local FS
	clickhouse	ClickHouse (HTTP + RowBinary) — change-log or current-state (ReplacingMergeTree), auto table creation

Multiple sinks in one pipeline

You can combine multiple sinks in one pipeline to fan out events to different destinations. However, multi-sink pipelines introduce complexity that requires careful consideration.

Why multiple sinks are challenging

Different performance characteristics: Kafka might handle 100K events/sec while a downstream HTTP webhook processes 100/sec. The slowest sink becomes the bottleneck for the entire pipeline.

Independent failure modes: Each sink can fail independently. Redis might be healthy while Kafka experiences broker failures. Without proper handling, a single sink failure could block the entire pipeline or cause data loss.

No distributed transactions: DeltaForge cannot atomically commit across heterogeneous systems. If Kafka succeeds but Redis fails mid-batch, you face a choice: retry Redis (risking duplicates in Kafka) or skip Redis (losing data there).

Checkpoint semantics: The checkpoint represents “how far we’ve processed from the source.” With multiple sinks, when is it safe to advance? After one sink succeeds? All of them? A majority?

Read the required and commit_policy sections below for options to manage these challenges.

The required flag

The required flag on each sink determines whether that sink must acknowledge successful delivery before the checkpoint advances:

sinks:
  - type: kafka
    config:
      id: primary-kafka
      required: true    # Must succeed for checkpoint to advance
      
  - type: redis
    config:
      id: cache-redis
      required: false   # Best-effort; failures don't block checkpoint

When required: true (default): The sink must acknowledge the batch before the checkpoint can advance. If this sink fails, the pipeline blocks and retries until it succeeds or the operator intervenes.

When required: false: The sink is best-effort. Failures are logged but don’t prevent the checkpoint from advancing. Use this for non-critical destinations where some data loss is acceptable.

Commit policy

The commit_policy works with the required flag to determine checkpoint behavior:

commit_policy:
  mode: required   # Only wait for required sinks

sinks:
  - type: kafka
    config:
      required: true   # Checkpoint waits for this
  - type: redis  
    config:
      required: false  # Checkpoint doesn't wait for this
  - type: nats
    config:
      required: true   # Checkpoint waits for this

Per-sink independent checkpoints

Each sink maintains its own checkpoint, committed independently after successful delivery. This means:

• Faster sinks are not held back by slower ones while the pipeline runs - each sink advances its own checkpoint and the source’s read position does not rewind for a lagging sink in-session
• The source replays from the minimum checkpoint across all sinks. Recovery is not selective per sink: on restart every sink, including ones already ahead, is re-delivered from that minimum forward, so sinks that were ahead receive duplicates and must dedup
• Adding a new sink lowers the minimum to that sink’s earliest position, so on the next restart the source replays from there and existing sinks are re-delivered those events too (they dedup); the new sink is not backfilled in isolation
• Removing a sink cleans up its checkpoint automatically on the next pipeline patch

This architecture avoids the common CDC pitfall where the slowest sink becomes a bottleneck for all other sinks.

Delivery guarantee tiers

DeltaForge does not claim end-to-end exactly-once for any sink. Every sink is at-least-once; Kafka exactly_once: true adds transactional atomic-batch visibility on top, and NATS/Redis add a dedup mechanism. Reaching exactly-once end to end always requires consumer cooperation (dedup on the event id).

Practical patterns

Primary + secondary: One critical sink (Kafka for durability) marked required: true, with secondary sinks (Redis for caching, testing or experimentation) marked required: false.

Quorum for redundancy: Three sinks with commit_policy.mode: quorum and quorum: 2. Checkpoint advances when any two succeed, providing fault tolerance.

All-or-nothing checkpointing: Use commit_policy.mode: all when every destination is critical - the checkpoint advances only once all sinks have acknowledged (throughput is gated by the slowest sink). This gates checkpoint advancement across sinks; it is not an atomic cross-sink commit - a batch can still land in some sinks and not others within a cycle, and those are reconciled by replay on restart.

Multi-format fan-out

For sending the same events to different consumers that expect different formats:

sinks:
  # Kafka Connect expects Debezium format
  - type: kafka
    config:
      id: connect-sink
      brokers: ${KAFKA_BROKERS}
      topic: connect-events
      envelope:
        type: debezium
      required: true

  # Lambda expects CloudEvents
  - type: kafka
    config:
      id: lambda-sink
      brokers: ${KAFKA_BROKERS}
      topic: lambda-events
      envelope:
        type: cloudevents
        type_prefix: "com.acme.cdc"
      required: false

  # Analytics wants raw events
  - type: redis
    config:
      id: analytics-redis
      uri: ${REDIS_URI}
      stream: analytics
      envelope:
        type: native
      required: false

This allows each consumer to receive events in their preferred format without post-processing.

Redis sink

The Redis sink publishes events to a Redis Stream for real-time consumption with consumer groups.

When to use Redis

Redis Streams shine when you need low-latency event delivery with simple operational requirements and built-in consumer group support.

Real-world applications

Pros and cons

Configuration

sinks:
  - type: redis
    config:
      id: orders-redis
      uri: ${REDIS_URI}
      stream: orders.events
      required: true

Idempotency

Every XADD command includes an idempotency_key field in the stream entry payload. This key is deterministic — the same source event always produces the same key, even across replays.

Redis Streams does not deduplicate server-side (entry IDs are auto-generated), so consumers are responsible for dedup using this field. A simple approach:

# Check idempotency_key before processing
seen = redis.smembers("processed_keys")
if data[b"idempotency_key"] in seen:
    r.xack(stream, group, msg_id)  # Skip duplicate
    continue
process(event)
redis.sadd("processed_keys", data[b"idempotency_key"])

This makes Redis a Tier 2 delivery guarantee: at-least-once with idempotency keys for consumer-side dedup.

Consuming events

Consumer groups (recommended)

# Create consumer group
redis-cli XGROUP CREATE orders.events mygroup $ MKSTREAM

# Read as consumer (blocking)
redis-cli XREADGROUP GROUP mygroup consumer1 BLOCK 5000 COUNT 10 STREAMS orders.events >

# Acknowledge processing
redis-cli XACK orders.events mygroup 1234567890123-0

# Check pending (unacknowledged) messages
redis-cli XPENDING orders.events mygroup

Simple subscription

# Read latest entries
redis-cli XREAD COUNT 10 STREAMS orders.events 0-0

# Block for new entries
redis-cli XREAD BLOCK 5000 STREAMS orders.events $

Go consumer example

import "github.com/redis/go-redis/v9"

rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379"})

// Create consumer group (once)
rdb.XGroupCreateMkStream(ctx, "orders.events", "mygroup", "0")

for {
    streams, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
        Group:    "mygroup",
        Consumer: "worker1",
        Streams:  []string{"orders.events", ">"},
        Count:    10,
        Block:    5 * time.Second,
    }).Result()
    
    if err != nil {
        continue
    }
    
    for _, stream := range streams {
        for _, msg := range stream.Messages {
            var event Event
            json.Unmarshal([]byte(msg.Values["df-event"].(string)), &event)
            process(event)
            rdb.XAck(ctx, "orders.events", "mygroup", msg.ID)
        }
    }
}

Rust consumer example

#![allow(unused)]
fn main() {
use redis::AsyncCommands;

let client = redis::Client::open("redis://localhost:6379")?;
let mut con = client.get_async_connection().await?;

// Create consumer group
let _: () = redis::cmd("XGROUP")
    .arg("CREATE").arg("orders.events").arg("mygroup").arg("0").arg("MKSTREAM")
    .query_async(&mut con).await.unwrap_or(());

loop {
    let results: Vec<StreamReadReply> = con.xread_options(
        &["orders.events"],
        &[">"],
        &StreamReadOptions::default()
            .group("mygroup", "worker1")
            .count(10)
            .block(5000)
    ).await?;
    
    for stream in results {
        for msg in stream.ids {
            let event: Event = serde_json::from_str(&msg.map["df-event"])?;
            process(event);
            con.xack("orders.events", "mygroup", &[&msg.id]).await?;
        }
    }
}
}

Python consumer example

import redis
import json

r = redis.Redis.from_url("redis://localhost:6379")

# Create consumer group (once)
try:
    r.xgroup_create("orders.events", "mygroup", id="0", mkstream=True)
except redis.ResponseError:
    pass  # Group already exists

# Consume events
while True:
    events = r.xreadgroup("mygroup", "worker1", {"orders.events": ">"}, count=10, block=5000)
    for stream, messages in events:
        for msg_id, data in messages:
            event = json.loads(data[b"df-event"])
            process(event)
            r.xack("orders.events", "mygroup", msg_id)

Failure modes

In the table below, “retries” means a bounded in-sink retry (exponential backoff, finite attempts) inside a single send_batch call. There is no unbounded pipeline-level retry loop: once in-sink attempts are exhausted on a required sink, the checkpoint is held and the batch is recovered by source replay on the next restart. Redis Streams delivery is at-least-once (XADD is not transactional across the batch); dedup on the event idempotency_key.

Failure scenarios and data guarantees

Redis OOM during batch delivery

1. DeltaForge sends a batch of 100 events
2. 50 events are written, then Redis hits maxmemory and the rest fail (XADD is not atomic across the batch, so the 50 already-written entries remain)
3. send_batch returns an error; the checkpoint does not advance
4. DeltaForge applies bounded in-sink retry; if OOM persists, the required-sink failure holds the checkpoint and backpressures until memory is available
5. On recovery/restart the batch is re-delivered from the source, so the already-written entries reappear as duplicates - consumers dedup on idempotency_key
6. Checkpoint is saved only after the full batch is acknowledged

DeltaForge crash after XADD, before checkpoint

1. Batch written to Redis stream successfully
2. DeltaForge crashes before saving checkpoint
3. On restart: replays from last checkpoint
4. Result: Duplicate events in stream (at-least-once)
5. Consumer must handle idempotently (check event.id)

Redis failover (Sentinel/Cluster)

1. Master fails, Sentinel promotes replica
2. In-flight XADD may fail with connection error
3. DeltaForge reconnects to new master
4. Bounded in-sink retry re-sends the batch (or it is replayed from the source on restart)
5. Possible duplicates if the original write succeeded - consumers dedup on idempotency_key

Handling duplicates in consumers

# Idempotent consumer using event ID
processed_ids = set()  # Or use Redis SET for distributed dedup

for msg_id, data in messages:
    event = json.loads(data[b"df-event"])
    event_id = event["id"]
    
    if event_id in processed_ids:
        r.xack("orders.events", "mygroup", msg_id)
        continue  # Skip duplicate
    
    process(event)
    processed_ids.add(event_id)
    r.xack("orders.events", "mygroup", msg_id)

Monitoring

DeltaForge exposes these metrics for Redis sink monitoring:

# DeltaForge sink metrics (exposed at /metrics on port 9000)
deltaforge_sink_events_total{pipeline,sink}      # Events delivered
deltaforge_sink_batch_total{pipeline,sink}       # Batches delivered  
deltaforge_sink_latency_seconds{pipeline,sink}   # Delivery latency histogram
deltaforge_stage_latency_seconds{pipeline,stage="sink"}  # Stage timing

For Redis server visibility, use Redis’s built-in monitoring:

# Monitor commands in real-time
redis-cli MONITOR

# Get server stats
redis-cli INFO stats

# Check memory usage
redis-cli INFO memory

# Stream-specific info
redis-cli XINFO STREAM orders.events
redis-cli XINFO GROUPS orders.events

Stream management

# Check stream length
redis-cli XLEN orders.events

# Trim to last 10000 entries (approximate)
redis-cli XTRIM orders.events MAXLEN ~ 10000

# Trim to exact length
redis-cli XTRIM orders.events MAXLEN 10000

# View consumer group info
redis-cli XINFO GROUPS orders.events

# Check pending messages
redis-cli XPENDING orders.events mygroup

# Claim stuck messages (after 60 seconds)
redis-cli XCLAIM orders.events mygroup worker2 60000 <message-id>

# Delete processed messages (careful!)
redis-cli XDEL orders.events <message-id>

Notes

• Redis Streams provide at-least-once delivery with consumer group acknowledgements
• Use MAXLEN ~ trimming to prevent unbounded memory growth (approximate is faster)
• Consider Redis Cluster for horizontal scaling with multiple streams
• Combine with Redis pub/sub for fan-out to ephemeral subscribers
• For durability, enable AOF persistence with appendfsync everysec or always
• Monitor memory usage closely; Redis will reject writes when maxmemory is reached

Kafka sink

The Kafka sink publishes batches to a Kafka topic using rdkafka.

When to use Kafka

Kafka excels as the backbone for event-driven architectures where durability, ordering, and replay capabilities are critical.

Real-world applications

Pros and cons

Configuration

sinks:
  - type: kafka
    config:
      id: orders-kafka
      brokers: ${KAFKA_BROKERS}
      topic: orders
      required: true
      exactly_once: false
      client_conf:
        message.timeout.ms: "5000"
        acks: "all"

Transactional atomic-batch delivery (exactly_once: true)

When exactly_once: true, the Kafka sink uses a transactional producer. Each batch is wrapped in a Kafka transaction (begin_transaction / commit_transaction), so a read_committed consumer sees either all events in the batch or none - never a partial batch.

This is not end-to-end exactly-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, and a read_committed consumer sees it twice. Delivery is at-least-once across a restart; dedup on the event id (stable across replays) to reach exactly-once end to end. The config field is named exactly_once for historical reasons; read it as “transactional producer.”

sinks:
  - type: kafka
    config:
      id: orders-kafka
      brokers: ${KAFKA_BROKERS}
      topic: orders
      exactly_once: true

How it works

1. DeltaForge assigns a stable transactional.id per pipeline-sink pair (deltaforge-{pipeline}-{sink_id})
2. On startup, init_transactions() registers with the broker and fences any zombie producer from a previous instance
3. Each batch: begin_transaction() → produce messages → commit_transaction()
4. If the commit fails before completing, the transaction is aborted and the batch is re-attempted from the same checkpoint (bounded in-sink retry); a persistent failure holds the checkpoint and recovers by source replay on restart
5. Consumers using isolation.level=read_committed only see committed batches; a batch replayed after a restart appears as a new committed transaction (duplicate), so consumers still dedup on event id

Requirements

• Kafka 2.5+ (transaction support)
• isolation.level=read_committed on consumers (otherwise they see uncommitted messages)
• Only one DeltaForge instance per pipeline at a time (the broker will fence duplicates)

Transactional overrides

When exactly_once: true, DeltaForge automatically configures:

You do not need to set these — they are applied automatically as defaults. Your client_conf is merged last, so any value you set there takes precedence (override at your own risk; these defaults are chosen for correct transactional behavior).

Fatal errors

If the broker fences the producer (another instance started with the same transactional.id), DeltaForge treats this as a fatal error and stops the pipeline. This is not retryable — resolve the duplicate instance before restarting.

Performance impact

Transactions add ~1-3ms overhead per batch for the two-phase commit. With properly sized batches (max_events=16000, max_bytes=16MB), throughput impact is ~7-11%. See the Performance guide for tuning details and benchmark results.

Avro encoding threads

An Avro sink encodes each batch in full before it enqueues any record, so a batch whose schema cannot be resolved publishes nothing. With encode_workers: N (N > 1) the batch is split into N contiguous slices, encoded on N threads, and enqueued in its original order: the records, their order, the batch’s failure behavior and its acknowledgement are the same as with one thread. A record whose schema is not registered yet (a table’s first batch, a new schema version) is encoded on the delivery task, which registers the schema once. A record that cannot be encoded is handled as with one thread. Only one batch is encoded at a time, so memory holds one batch’s encoded records whatever the thread count; the threads exist only while a batch is encoded. JSON sinks ignore the setting.

One thread (the default) keeps encoding on the delivery task and uses the least CPU. More threads are a capacity option for saturated Avro streams on hosts with free cores. Measured on one host (a 15 W laptop CPU, MySQL to Kafka, saturated backlog, five interleaved repetitions):

The extra CPU per row with more threads (1.45 to 2.44 CPU seconds per 100k narrow rows at four) is mostly lower clock speed on a power-limited CPU running more cores: the instructions per row do not grow. Records per Kafka request stay the same (about 970-980 narrow, 660 wide).

Linger defaults by format

client_conf overrides both.

Recommended client_conf settings

Consuming events

Kafka CLI

# Consume from beginning
kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic orders \
  --from-beginning

# Consume with consumer group
kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic orders \
  --group deltaforge-consumers

Go consumer example

config := sarama.NewConfig()
config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRoundRobin
config.Consumer.Offsets.Initial = sarama.OffsetOldest

group, _ := sarama.NewConsumerGroup([]string{"localhost:9092"}, "my-group", config)

for {
    err := group.Consume(ctx, []string{"orders"}, &handler{})
    if err != nil {
        log.Printf("Consumer error: %v", err)
    }
}

type handler struct{}

func (h *handler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
    for msg := range claim.Messages() {
        var event Event
        json.Unmarshal(msg.Value, &event)
        process(event)
        session.MarkMessage(msg, "")
    }
    return nil
}

Rust consumer example

#![allow(unused)]
fn main() {
use rdkafka::consumer::{Consumer, StreamConsumer};
use rdkafka::Message;

let consumer: StreamConsumer = ClientConfig::new()
    .set("bootstrap.servers", "localhost:9092")
    .set("group.id", "my-group")
    .set("auto.offset.reset", "earliest")
    .create()?;

consumer.subscribe(&["orders"])?;

loop {
    match consumer.recv().await {
        Ok(msg) => {
            let payload = msg.payload_view::<str>().unwrap()?;
            let event: Event = serde_json::from_str(payload)?;
            process(event);
            consumer.commit_message(&msg, CommitMode::Async)?;
        }
        Err(e) => eprintln!("Kafka error: {}", e),
    }
}
}

Python consumer example

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'orders',
    bootstrap_servers=['localhost:9092'],
    group_id='my-group',
    auto_offset_reset='earliest',
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

for message in consumer:
    event = message.value
    process(event)
    consumer.commit()

Failure modes

In the table below, “retries” means a bounded in-sink retry (exponential backoff, finite attempts) inside a single send_batch call. DeltaForge does not run an unbounded pipeline-level retry loop: once in-sink attempts are exhausted on a required sink, the checkpoint is held and the batch is recovered by source replay on the next restart. “Blocks checkpoint” means the source backpressures until the sink recovers.

Failure scenarios and data guarantees

Broker failure during batch delivery

1. DeltaForge sends batch of 100 events
2. 50 events delivered, broker crashes
3. rdkafka detects failure, retries remaining 50
4. If idempotence enabled: no duplicates
5. If not: possible duplicates of events near failure point
6. Checkpoint only saved after ALL events acknowledged

DeltaForge crash after Kafka ack, before checkpoint

1. Batch delivered to Kafka successfully
2. DeltaForge crashes before saving checkpoint
3. On restart: replays from last checkpoint
4. Without exactly_once: duplicate events in Kafka (at-least-once); consumer must handle idempotently
5. With exactly_once: replayed batch gets a new transaction; consumers with read_committed see duplicates but each batch is atomic

Monitoring recommendations

DeltaForge exposes these metrics for Kafka sink monitoring:

# DeltaForge sink metrics (exposed at /metrics on port 9000)
deltaforge_sink_events_total{pipeline,sink}      # Events delivered
deltaforge_sink_batch_total{pipeline,sink}       # Batches delivered
deltaforge_sink_latency_seconds{pipeline,sink}   # Delivery latency histogram
deltaforge_stage_latency_seconds{pipeline,stage="sink"}  # Stage timing

For deeper Kafka broker visibility, monitor your Kafka cluster directly:

• Broker metrics via JMX or Kafka’s built-in metrics
• Consumer lag via kafka-consumer-groups.sh
• Topic throughput via broker dashboards

Note: Internal rdkafka producer statistics (message queues, broker RTT, etc.) are not currently exposed by DeltaForge. This is a potential future enhancement.

Notes

• Combine Kafka with other sinks to fan out data; use commit policy to control checkpoint behavior
• For transactional atomic-batch delivery, set exactly_once: true and ensure your Kafka cluster supports transactions (2.5+); this is not end-to-end exactly-once, so consumers still dedup on event id across restarts
• With exactly_once: false (default), idempotent production is still enabled — duplicates are prevented during retries but not across DeltaForge restarts
• Adjust client_conf for durability (acks=all) or performance based on your requirements
• Consider partitioning strategy for ordering guarantees within partitions

NATS sink

The NATS sink publishes events to a NATS JetStream stream for durable, at-least-once delivery.

When to use NATS

NATS JetStream is ideal when you need a lightweight, high-performance messaging system with persistence, without the operational overhead of Kafka.

Real-world applications

Pros and cons

Configuration

sinks:
  - type: nats
    config:
      id: orders-nats
      url: ${NATS_URL}
      subject: orders.events
      stream: ORDERS
      required: true
      send_timeout_secs: 5
      batch_timeout_secs: 30

Authentication options

# Credentials file
credentials_file: /etc/nats/creds/user.creds

# Username/password
username: ${NATS_USER}
password: ${NATS_PASSWORD}

# Token
token: ${NATS_TOKEN}

Deduplication

DeltaForge sets the Nats-Msg-Id header on every published message using a deterministic idempotency key derived from the event’s source, transaction, and row identity. JetStream uses this header for server-side deduplication within the stream’s duplicate_window.

This means: if DeltaForge replays a batch after a crash, duplicate messages are automatically discarded by the server — no consumer-side dedup needed (within the dedup window).

Configure duplicate_window on the stream to match your maximum expected replay window:

nats stream add ORDERS \
  --subjects "orders.>" \
  --duplicate-window 5m    # Dedup messages replayed within 5 minutes

The default duplicate_window is 2 minutes. Increase this if DeltaForge might be down longer before restarting.

JetStream setup

Before using the NATS sink with JetStream, create a stream that captures your subject:

# Using NATS CLI
nats stream add ORDERS \
  --subjects "orders.>" \
  --retention limits \
  --storage file \
  --replicas 3 \
  --max-age 7d

# Verify stream
nats stream info ORDERS

Consuming events

NATS CLI

# Subscribe to subject (ephemeral)
nats sub "orders.>"

# Create durable consumer
nats consumer add ORDERS orders-processor \
  --pull \
  --ack explicit \
  --deliver all \
  --max-deliver 3 \
  --filter "orders.events"

# Consume messages
nats consumer next ORDERS orders-processor --count 10

Go consumer example

nc, _ := nats.Connect("nats://localhost:4222")
js, _ := nc.JetStream()

// Create or bind to consumer
sub, _ := js.PullSubscribe("orders.events", "orders-processor",
    nats.Durable("orders-processor"),
    nats.AckExplicit(),
)

for {
    msgs, _ := sub.Fetch(10, nats.MaxWait(5*time.Second))
    for _, msg := range msgs {
        var event Event
        json.Unmarshal(msg.Data, &event)
        process(event)
        msg.Ack()
    }
}

Rust consumer example

#![allow(unused)]
fn main() {
use async_nats::jetstream;

let client = async_nats::connect("nats://localhost:4222").await?;
let js = jetstream::new(client);

let stream = js.get_stream("ORDERS").await?;
let consumer = stream.get_consumer("orders-processor").await?;

let mut messages = consumer.messages().await?;
while let Some(msg) = messages.next().await {
    let msg = msg?;
    let event: Event = serde_json::from_slice(&msg.payload)?;
    process(event);
    msg.ack().await?;
}
}

Monitoring

DeltaForge exposes these metrics for NATS sink monitoring:

# DeltaForge sink metrics (exposed at /metrics on port 9000)
deltaforge_sink_events_total{pipeline,sink}      # Events delivered
deltaforge_sink_batch_total{pipeline,sink}       # Batches delivered
deltaforge_sink_latency_seconds{pipeline,sink}   # Delivery latency histogram
deltaforge_stage_latency_seconds{pipeline,stage="sink"}  # Stage timing

For NATS server visibility, use the NATS CLI or monitoring endpoint:

# Server info
nats server info

# JetStream account info
nats account info

# Stream statistics
nats stream info ORDERS

# Consumer statistics  
nats consumer info ORDERS orders-processor

# Real-time event monitoring
nats events

NATS also exposes a monitoring endpoint (default :8222) with JSON stats:

• http://localhost:8222/varz - General server stats
• http://localhost:8222/jsz - JetStream stats
• http://localhost:8222/connz - Connection stats

Subject design patterns

Failure modes

In the table below, “retries” means a bounded in-sink retry (exponential backoff, finite attempts) inside a single send_batch call. There is no unbounded pipeline-level retry loop: once in-sink attempts are exhausted on a required sink, the checkpoint is held and the batch is recovered by source replay on the next restart. JetStream delivery is at-least-once; server-side dedup via Nats-Msg-Id within duplicate_window collapses duplicates.

Failure scenarios and data guarantees

NATS server restart during batch delivery

1. DeltaForge sends batch of 100 events
2. 50 events published, server restarts
3. async_nats detects disconnect, starts reconnecting
4. After reconnect, DeltaForge retries remaining 50
5. JetStream deduplication prevents duplicates (if enabled)
6. Checkpoint only saved after ALL events acknowledged

DeltaForge crash after JetStream ack, before checkpoint

1. Batch published to JetStream successfully
2. DeltaForge crashes before saving checkpoint
3. On restart: replays from last checkpoint
4. Result: Duplicate events in stream (at-least-once)
5. Consumer must handle idempotently (check event.id)

Stream storage exhausted

1. JetStream stream hits max_bytes or max_msgs limit
2. With discard: old → oldest messages removed, publish succeeds
3. With discard: new → publish rejected
4. Bounded in-sink retry runs; if the stream stays full, the required-sink failure holds the checkpoint and backpressures (retrying alone will not clear a full stream)
5. Resolution: Increase limits or enable discard: old

JetStream acknowledgement levels

# Stream configuration affects durability
nats stream add ORDERS \
  --replicas 3 \           # R=3 for production
  --retention limits \     # or 'workqueue' for single consumer
  --discard old \          # Remove oldest when full
  --max-age 7d \           # Auto-expire after 7 days
  --storage file           # Persistent (vs memory)

Handling duplicates in consumers

// Use event ID for idempotency
processedIDs := make(map[string]bool)  // Or use Redis/DB

for _, msg := range msgs {
    var event Event
    json.Unmarshal(msg.Data, &event)
    
    if processedIDs[event.ID] {
        msg.Ack()  // Already processed
        continue
    }
    
    if err := process(event); err == nil {
        processedIDs[event.ID] = true
    }
    msg.Ack()
}

Notes

• When stream is specified, events are published to that JetStream stream (the stream must already exist; the sink does not create or verify it)
• Without stream, events are published to core NATS (no persistence guarantees)
• Connection pooling ensures efficient reuse across batches
• Use replicated streams (--replicas 3) for production durability
• Combine with other sinks to fan out data; use commit policy to control checkpoint behavior
• JetStream provides at-least-once delivery with server-side deduplication via Nats-Msg-Id within the configured duplicate_window

HTTP/Webhook sink

The HTTP sink delivers CDC events via HTTP POST (or PUT) to any URL — internal services, serverless functions, third-party APIs, webhooks.

When to use HTTP

HTTP is the universal integration point. Use it when:

• Your consumer doesn’t run Kafka/NATS/Redis
• You need to call a REST API or webhook on every database change
• You want the simplest possible setup (DeltaForge + any HTTP server)
• You’re integrating with serverless functions (AWS Lambda, Cloud Functions)

Pros and cons

Configuration

sinks:
  - type: http
    config:
      id: my-webhook
      url: "https://api.example.com/events"
      method: POST
      headers:
        Authorization: "Bearer ${API_TOKEN}"
        X-Source: deltaforge
      batch_mode: false
      required: true
      send_timeout_secs: 10

Authentication

All auth is handled via the headers map. Values support ${ENV_VAR} shell expansion.

# Bearer token
headers:
  Authorization: "Bearer ${API_TOKEN}"

# Basic auth
headers:
  Authorization: "Basic ${BASIC_AUTH_B64}"

# Custom API key
headers:
  X-API-Key: "${MY_API_KEY}"

# HMAC signature (computed externally, injected via env)
headers:
  X-Signature: "${WEBHOOK_SIGNATURE}"

URL templates

Route events to different URLs based on event fields:

# Per-table endpoint
url: "https://api.example.com/cdc/${source.table}"
# → https://api.example.com/cdc/orders
# → https://api.example.com/cdc/customers

# Per-database endpoint
url: "https://${source.db}.api.internal/events"

# Static URL (most common)
url: "https://api.example.com/webhook"

Batch mode

By default, the sink sends one HTTP request per event. Enable batch_mode: true to send a JSON array of events in a single request:

# Per-event mode (default): one POST per event
batch_mode: false
# Body: {"id": "...", "op": "c", "after": {...}}

# Batch mode: one POST with JSON array
batch_mode: true
# Body: [{"id": "...", "op": "c", ...}, {"id": "...", "op": "u", ...}]

Batch mode reduces HTTP overhead but means the consumer must handle arrays.

Retry behavior

Failure modes

In the table below, “retries” means a bounded in-sink retry (exponential backoff, finite attempts) inside a single send_batch call. There is no unbounded pipeline-level retry loop: once in-sink attempts are exhausted on a required sink, the checkpoint is held and the batch is recovered by source replay on the next restart. HTTP delivery is at-least-once with no server-side dedup, so the endpoint must be idempotent (dedup on event id).

Consuming events

Node.js / Express

app.post('/webhook', (req, res) => {
  const event = req.body;
  console.log(`${event.op} on ${event.source.table}: ${JSON.stringify(event.after)}`);
  res.sendStatus(200);
});

Python / Flask

@app.route('/webhook', methods=['POST'])
def webhook():
    event = request.json
    process(event)
    return '', 200

Go

http.HandleFunc("/webhook", func(w http.ResponseWriter, r *http.Request) {
    var event Event
    json.NewDecoder(r.Body).Decode(&event)
    process(event)
    w.WriteHeader(http.StatusOK)
})

Notes

• Connection pooling is automatic — reqwest reuses TCP connections to the same host
• The Content-Type header is set to application/json by default
• At-least-once delivery: on crash, events may be re-sent. Consumers should be idempotent.
• For per-event dedup, use the id field in the event payload (UUID v7, stable across replays)

S3 sink (Parquet / JSON Lines)

The S3 sink writes CDC events as Parquet (default) or JSON Lines files to S3-compatible object storage - AWS S3, MinIO, GCS via S3-interop, Azure Blob, R2, Wasabi, or local filesystem. Files are partitioned by table and UTC date in Hive style, with atomic multipart commits.

When to use S3

The S3 sink is the lakehouse-first path out of DeltaForge. CDC events land directly in queryable Parquet without a Kafka detour.

Real-world applications

Pros and cons

Configuration

sinks:
  - type: s3
    config:
      id: orders-lake
      bucket: my-data-lake
      prefix: cdc/orders
      region: us-east-1
      access_key_id: ${S3_ACCESS_KEY}
      secret_access_key: ${S3_SECRET_KEY}
      format: parquet
      compression: snappy
      file_roll:
        max_bytes: 268435456
        max_events: 1000000
        max_age_secs: 300
        idle_age_secs: 600
      send_timeout_secs: 60
      required: true