CDC

The datum-cdc crate turns a PostgreSQL logical-replication stream into an ordinary Datum Source<ChangeEvent>. It is a single lightweight binary — one Tokio replication task feeding a bounded channel — rather than a JVM plus Kafka Connect plus a message broker. The crate is #![forbid(unsafe_code)]; the positioning claim against Debezium is measured, not asserted.

cargo add datum-cdc

The MVP decodes pgoutput protocol v1 with streaming transactions off, two-phase commit off, and text-mode tuple values. That covers the common case — insert/update/delete/truncate on published tables — and the deferred scope lists what it does not yet do.

How it is wired

The control plane (slot create/validate/drop, standby feedback, restart state, health) is actor-serialized. The data plane is a single-owner Tokio replication carrier that reads the CopyBoth stream, decodes pgoutput, and pushes ChangeEvents into a bounded Datum channel. Downstream backpressure is the channel: when your pipeline stalls, the carrier stops reading ahead and PostgreSQL retains WAL until feedback advances — see WAL retention. Building a CdcSource constructs a blueprint and connects nothing; the connection opens at materialization.

PostgreSQL setup

The server must run with logical decoding enabled (a restart is required to change wal_level):

# postgresql.conf
wal_level = logical

Create a publication for the tables you want to capture, and a pgoutput replication slot. The publication must publish insert, update, and delete (datum-cdc validates this at start unless you call .validate_publication(false)):

-- One publication for the tables you care about.
CREATE PUBLICATION orders_pub FOR TABLE public.orders
  WITH (publish = 'insert,update,delete,truncate');

-- FULL replica identity gives you complete before-images on UPDATE/DELETE.
-- The default identity only carries the primary-key columns in `before`.
ALTER TABLE public.orders REPLICA IDENTITY FULL;

-- A dedicated pgoutput slot per source (SlotLifecycle::Existing).
SELECT pg_create_logical_replication_slot('datum_orders', 'pgoutput');

The replication role needs the REPLICATION attribute (or superuser) and CONNECT on the database. You can skip the manual pg_create_logical_replication_slot call and let datum-cdc create the slot for you — see Slot lifecycle.

Building the source

CdcSource::postgres() returns a builder; .build() produces a Source<ChangeEvent, CdcHandle>, where CdcHandle is the materialized management handle:

use datum::Source;
use datum_cdc::{CdcHandle, CdcResult};

fn build_orders_source() -> CdcResult<Source<ChangeEvent, CdcHandle>> {
    CdcSource::postgres()
        // The MVP replication build is plaintext only (sslmode=disable/prefer);
        // see the TLS note in the guide.
        .connect(PostgresCdcConfig::from_url(
            "postgresql://datum_cdc@127.0.0.1:5432/datum_cdc?sslmode=disable",
        )?)
        .slot("datum_orders")
        .publication("orders_pub")
        .slot_lifecycle(SlotLifecycle::Existing)
        .buffer_capacity(8192)
        .build()
}

// Building the blueprint starts nothing — no PostgreSQL connection, no
// replication task. Materialization (`run`/`run_with`) is what connects.
let source = build_orders_source().expect("CDC blueprint builds");
let _ = source;

For at-least-once pipelines, prefer .build_with_context(). It returns a SourceWithContext<ChangeEvent, CdcOffset, CdcHandle> so the offset stays attached through context-preserving operators — you commit exactly the offset that belongs to each event:

use datum::SourceWithContext;
use datum_cdc::{CdcHandle, CdcOffset};

// `build_with_context()` keeps the CdcOffset attached to each event so it
// survives context-preserving operators — this is the recommended
// at-least-once shape. A durable checkpoint store lets the source resume
// after a process restart.
let source: SourceWithContext<ChangeEvent, CdcOffset, CdcHandle> = CdcSource::postgres()
    .connect(
        PostgresCdcConfig::from_url("postgresql://datum_cdc@127.0.0.1:5432/datum_cdc")
            .expect("valid url"),
    )
    .slot("datum_orders")
    .publication("orders_pub")
    .slot_lifecycle(SlotLifecycle::CreateIfMissing)
    .checkpoint_store(MemoryCheckpointStore::new())
    .build_with_context()
    .expect("CDC context blueprint builds");
let _ = source;

The ChangeEvent

Each element is one row change with the metadata you need to route, transform, and de-duplicate it:

Field	Type	Notes
source	SourceMetadata	{ database, slot, publication }
schema / table	String	e.g. public / orders
op	ChangeOperation	Insert | Update | Delete | Truncate
before / after	Option<RowData>	tuple values in relation-column order
truncate	Option<TruncateOptions>	{ cascade, restart_identity } for truncates
lsn	CdcOffset	the committable/dedup offset (below)
tx	TransactionMeta	{ xid, commit_time_micros, event_index, event_count }
relation	RelationMetadata	column names, type OIDs, replica identity

Read column values by name against the relation metadata:

use datum_cdc::ChangeOperation;

if event.op == ChangeOperation::Insert {
    if let Some(row) = &event.after {
        let id = row.get_text(&event.relation, "id");
        let total = row.get(&event.relation, "total");
        // ...
    }
}

ChangeEvent is serde::{Serialize, Deserialize}, so it drops straight into a JSON sink, a KafkaSink, or your own encoder.

Checkpointing and at-least-once delivery

The source is at-least-once. PostgreSQL standby feedback (the confirmed-flush LSN) advances only after downstream code calls CdcCheckpointHandle::checkpoint for every event in the next contiguous committed transaction. The FeedbackCoordinator tracks per-transaction completion and advances the watermark only when a transaction is fully checkpointed and there are no un-checkpointed transactions before it — so feedback never runs ahead of durably-processed data.

If a process crashes after an event is delivered but before its checkpoint feedback reaches PostgreSQL, those rows are replayed on restart. Downstream code must be idempotent or de-duplicate on the stable key (slot, tx_end_lsn, xid, event_index) — every field is on CdcOffset.

The commit is a single call on the checkpoint handle you take from the materialized CdcHandle. This example is pg-gated — it needs a live PostgreSQL and is adapted from crates/datum-cdc/tests/pg_integration.rs:

// Requires PostgreSQL. Adapted from tests/pg_integration.rs.
use datum::{Keep, Sink};

let source = CdcSource::postgres()
    .connect(PostgresCdcConfig::from_url(url)?)
    .slot("datum_orders")
    .publication("orders_pub")
    .slot_lifecycle(SlotLifecycle::CreateOwned)
    .checkpoint_store(FileCheckpointStore::new("/var/lib/datum/cdc")) // durable resume
    .build()?;

let (handle, completion) = source
    .take(1)
    .to_mat(Sink::collect(), Keep::both)
    .run()?;

let events = completion.wait()?;

// Announce the checkpoint only after the event is durably processed.
handle
    .checkpoint_handle()
    .checkpoint(events[0].lsn.clone())?;

checkpoint(offset) persists the offset to the configured store (if any) and tells the replication task it may advance feedback for that transaction. checkpoint_async(offset) is the async sibling with the same semantics.

Resume and restart

Point the builder at a durable CdcCheckpointStore and datum-cdc resumes from the last checkpointed offset instead of the slot's confirmed LSN:

• FileCheckpointStore::new(path) — one JSON file per slot (path is a directory) or an exact file; writes are atomic (write-temp-then-rename).
• MemoryCheckpointStore::new() — for tests and ephemeral readers.
• Implement CdcCheckpointStore yourself (load/store) to checkpoint into your own transactional store alongside your downstream writes.

Where the source starts is CdcStart:

CdcStart	Behavior
CheckpointOrSlot (default)	durable checkpoint if present, else the slot's confirmed LSN
SlotConfirmed	always the slot's confirmed LSN
Lsn(PgLsn)	an explicit LSN

On start, datum-cdc refuses to move backwards past what the slot can still serve: if the requested start LSN is older than the slot's restart_lsn (WAL recycled), or the slot's confirmed LSN is ahead of a durable checkpoint (a competing consumer or possible loss), it fails loudly rather than silently skipping data.

The replication carrier reconnects on transient replication/connection/IO errors with bounded exponential backoff (ReconnectSettings: 200 ms → 5 s, 60 s total by default). .disable_reconnect() turns this off; restart_factory() returns a Fn() -> Source<…> you can hand to a supervised restart policy or a datum-agent job.

WAL retention

An un-consumed slot pins WAL. Because backpressure is the bounded channel, a stalled downstream translates directly into retained WAL on the server. Monitor both sides:

• CdcHandle::lag() samples CdcLag { retained_wal_bytes, confirmed_lag_bytes, slot_active } from pg_replication_slots.
• CdcHandle::health() returns CdcHealth { running, last_error, last_received_lsn, latest_server_wal_end, last_feedback_lsn, emitted_events, reconnects }.
• Server-side, watch pg_replication_slots.restart_lsn, confirmed_flush_lsn, wal_status, and safe_wal_size.

Bound the blast radius with max_slot_wal_keep_size on PostgreSQL. If a slot is marked lost, datum-cdc refuses to start against it — create a new slot after a fresh snapshot.

Slot lifecycle

SlotLifecycle controls whether datum-cdc creates or drops the slot:

Variant	Creates missing slot?	Droppable via handle?
Existing (default, production)	no	no (requires force_drop_slot)
CreateIfMissing	yes	no (requires force_drop_slot)
CreateOwned	yes	drop_slot()
Temporary	yes (temporary slot)	drop_slot()

CdcHandle::drop_slot() works only for CreateOwned/Temporary; force_drop_slot() terminates the active backend and drops any slot (use with care). stop() shuts the carrier down without dropping the slot.

TLS

The MVP replication build is plaintext only (sslmode=disable or prefer). .build() rejects require/verify-ca/verify-full because the workspace pins pgwire-replication without its optional TLS feature. Datum's network stack uses rustls with the ring provider (Rust plus assembly), so its TLS/crypto path avoids aws-lc-rs, aws-lc-sys, and CMake-built C crypto. That does not make every optional dependency in the workspace C-free. Wiring and testing replication TLS remains a deferred item; until then, run the replication connection over a trusted network path (loopback, private subnet, or a tunnel).

Performance vs Debezium

The "lightweight" claim is measured on the same PostgreSQL 17 instance and loadgen workload — a narrow single-table mixed insert/update/delete stream at 100 tx/s for 60 s (6,000 committed ops). Both consumers were correctness-clean: every committed op observed exactly once, per-key ordered, zero duplicates/missing/mismatched. Full method and raw captures are in roadmap/benchmarks/cdc.md.

Metric	Debezium 3.6 (Kafka Connect)	datum-cdc (pgoutput v1)
End-to-end latency p50	519.43 ms	0.14 ms
End-to-end latency p99	969.19 ms	4.91 ms
Throughput	98.13 events/s	100.02 events/s
Consumer CPU	9.77 core-s	6.69 core-s (~1.5× less)
Consumer peak RSS	1,600.76 MiB	9.08 MiB (~176× less)
Max confirmed slot lag	873,072 bytes	4,816 bytes

The gap is structural: no JVM, no Kafka Connect worker, no broker hop between PostgreSQL and your code. Latency is measured commit-to-delivery, so Debezium's Connect + broker pipeline shows up directly.

Apples-to-apples caveat

Debezium ships an initial-snapshot mode, exactly-once via Kafka transactions, schema history, and a mature connector ecosystem that this MVP does not (see below). This row measures the steady-state streaming path both tools share — it is not a claim of feature parity.

Deferred scope

The MVP intentionally defers, in rough priority order:

• Initial snapshots (COPY + snapshot-LSN handoff) — start from an existing slot today.
• pgoutput v2–v4: large in-progress (streamed) transactions, two-phase commit, and parallel streaming.
• Binary tuple decoding (text mode only for now).
• Schema/DDL capture.
• Multi-slot merge ordering.
• Replication over TLS (see TLS).

Per-slot total order is preserved today; that is the ordering guarantee the MVP makes.

See also

• Context Propagation — SourceWithContext and offset handling.
• Kafka — pair CDC with a Kafka sink for a change pipeline.
• Agent — run the source as a supervised, drainable job.