# CDC-out (reverse-ETL) > Declaratively mirror table changes outward to external sinks (webhook, plus a CdcSink interface for custom sinks) through a durable outbox — ordered per pipe, at-least-once from enqueue, retried with backoff, dead-lettered. --- ## CDC-out (reverse-ETL) _Declaratively mirror table changes outward to external sinks (webhook, plus a CdcSink interface for custom sinks) through a durable outbox — ordered per pipe, at-least-once from enqueue, retried with backoff, dead-lettered._ # CDC-out — declarative reverse-ETL `@voltro/plugin-cdc-out` streams your table changes OUTWARD to an external sink. Declare `{ table → sink, map }` and the plugin mirrors every change through a **durable outbox** in your app's own database — no separate Debezium / Fivetran pipeline. The data-team wedge: app data in the warehouse, declared in code, composed onto the change stream you already have. ## Wiring ```ts // app.config.ts import { cdcOutPlugin, webhookSink } from '@voltro/plugin-cdc-out' export default { type: 'api' as const, name: 'api', plugins: [ cdcOutPlugin({ sinks: [ { table: 'orders', sink: webhookSink('https://warehouse.example.com/ingest'), // Map a row → the outbound record body (default: the row unchanged). map: (row) => ({ id: row.id, total: row.total, status: row.status }), // Optional: only mirror some changes. filter: (event) => event.op !== 'delete', batchSize: 500, // max records per delivered batch (default 100) }, ], maxAttempts: 5, // per-record delivery attempts before dead-letter backoffBaseMs: 200, // first retry delay; doubles per attempt, jittered deliveryTimeoutMs: 10_000, // per-attempt timeout — aborts the sink call leaseTtlMs: 15_000, // leader lease; heartbeat renews at ttl/3 // Fleet-scope handoff (see "Delivery guarantees"): dedupWindowMs: 60_000, // how long an enqueue claim is kept — must exceed leaseTtlMs handoffBufferMs: 60_000, // how far back each replica buffers for a takeover to drain handoffBufferSize: 10_000, // hard ceiling on buffered changes per replica }), ], } ``` One sink config per table per instance — a duplicate `table` is a **loud boot error**. For multiple sinks on one table, wire a second instance with a `name` (`cdcOutPlugin({ name: 'analytics', sinks: […] })`); it suffixes the plugin name (`@voltro/plugin-cdc-out#analytics`) and the inspect mount (`/_voltro/inspect/plugins/cdc-out--analytics/…`). ## Sinks - **`memorySink()`** — in-process, records every delivered batch. Dev + tests. - **`webhookSink(url, { headers? })`** — POSTs each batch as `{ records: [...] }` JSON, honoring the engine's per-attempt abort signal. Its host is declared as a `network:outbound:` permission automatically. - **Warehouse / Kafka** — implement the `CdcSink` interface (`{ name, deliver(batch, ctx), outboundHost? }`). `deliver` may return a **Promise or an Effect** — both compose without wrapping. The engine is connector-agnostic; the sink is the only thing that changes. ## Delivery guarantees — exactly what holds Every mirrored change becomes a row in `_voltro_cdcout_outbox` (contributed via `extendSchema`, migrated by `voltro dev`). The row's TypeID id **is** the record's `deliveryKey` — unique across replicas, stable across restarts and retries. - **Enqueue is de-duplicated per observed change, fleet-wide, across leadership handovers — with one hole: a replica that dies between winning a change's claim and inserting its outbox row loses that change, because the claim survives and nothing rescans orphan claims.** (Those are two statements with no transaction around them, which is why this does not say "exactly-once".) On `changeScope: 'local'` stores each replica enqueues only its OWN commits (injected cross-replica events are skipped), so a change is observed by exactly one process and needs nothing further. On `'fleet'` stores (postgres `changeStrategy: 'cdc'` LISTEN/NOTIFY, mysql binlog) EVERY replica sees the full stream, so every replica **buffers** it in memory (`handoffBufferMs`, capped at `handoffBufferSize` entries, oldest dropped first). The holder of the leader lease (`_voltro_cdcout_leases`, TTL-heartbeat) enqueues as it goes; a replica that WINS the lease drains the window its predecessor never got to. Enqueue is **idempotent per change identity**: each change is keyed by a `changeKey` every replica computes alike — a digest of `(pipe, op, row id, new image, old image)` plus an occurrence counter that keeps two byte-identical changes to one row apart — and claimed in `_voltro_cdcout_claims` under `unique(pipe, changeKey)`. So the rows the dying leader already wrote collapse instead of duplicating, and the ones it never reached are written by its successor. A replica that boots into a fleet that is **already running** adopts the fleet's occurrence counters from the claims already in the database before it keys anything — otherwise its first sighting of an already-claimed change would key occurrence 0, collide, and be dropped as a duplicate it is not. `GET /_voltro/inspect/plugins/cdc-out/sinks` reports `handoff.seeded` and `handoff.awaitingSeed` so you can see that happen rather than assume it. What still bounds it, stated plainly: a change **no surviving replica observed** is gone (the transport delivered it only to the dead process); a handoff that takes longer than `handoffBufferMs` loses whatever aged out of the buffer, and that count is reported as `handoff.dropped` on `GET /_voltro/inspect/plugins/cdc-out/sinks` rather than dropped silently. `dedupWindowMs` (default `max(60_000, 4 × leaseTtlMs)`) is how long a claim is kept; it **must exceed `leaseTtlMs`** — a claim that expires mid-handoff is a duplicate window, and the plugin refuses to boot with one. - **At-least-once FROM ENQUEUE.** The tap is post-commit — a crash in the narrow window between commit and the outbox insert loses that one event; the plugin does not claim better. From enqueue on, delivery survives restarts, retries with exponential backoff + jitter, and re-sends the SAME `deliveryKey`s — a sink that upserts on them dedupes safely. - **Ordered per pipe.** One in-flight batch per pipe, consumed strictly in commit (id) order; the queue is head-blocking, so a batch waiting out its retry backoff is never overtaken. - **Dead-letter, never silent drop.** A record that exhausts `maxAttempts` moves to status `dead` with its last error — queryable at `GET /_voltro/inspect/plugins/cdc-out/dead-letter` — and unblocks the pipe. - **Bounded storage.** Delivered/dead rows are purged by the framework retention sweep after `retentionHours` (default 72, env `CDCOUT_RETENTION_HOURS`); pending rows are never purged. Enqueue claims are short-lived by design — purged after `dedupWindowMs`, which only has to outlive a leadership handoff. ## Multi-tenancy The change tap fires for **every tenant's rows**, and a sink is app-global, developer-authored config: whatever `map` returns (default: the raw row, including `tenantId`) is mirrored to ONE external endpoint. Use `filter` / `map` to scope or strip tenant data, and only ever point sinks at **operator-controlled** URLs — never at a tenant-supplied one. ## Backfill Seed a pipe with existing rows once — they flow through the SAME durable outbox as live changes (ordered, retried, dead-lettered, stable per-row `deliveryKey`s). Call from a `*.startup.tsx` or a CLI with the app's store: ```ts import { enqueueBackfill } from '@voltro/plugin-cdc-out' import { sinkConfig } from './cdc' const n = await enqueueBackfill(store, sinkConfig, rows) // every current row, enqueued as an insert ```