Complete Phase F: wire incremental, cross-engine, and CDC end-to-end (#6)

Author: nixrajput

CHANGELOG.md

@@ -22,7 +22,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   - `mysql` and `mariadb` drivers as thin wrappers that inject the fork-specific binary names and declare capabilities honestly (`Parallel: false` — the dump tools are single-threaded). Registered via side-effect import in `internal/app/drivers.go`.
   - Integration suites for both engines run on the Phase D `RunDriverSuite` harness via testcontainers (`mysql:8.0`, `mariadb:11`).
   - CI installs the Postgres/MySQL/MariaDB client tools and runs the full integration suite on the Ubuntu runner (where Docker is available); unit tests continue to run on Linux/macOS/Windows.
-- Phase F advanced-transfer machinery (some CLI paths gated pending follow-up wiring — see the `docs/INCREMENTAL.md`, `docs/CROSS_ENGINE.md`, `docs/CDC.md` Status sections):
+- Phase F advanced transfer — all four modes work end-to-end (incremental, cross-engine, CDC, bounded streaming); live DB paths are integration-tested in CI. See `docs/INCREMENTAL.md`, `docs/CROSS_ENGINE.md`, `docs/CDC.md`.
   - **Working today:** every dump is prefixed with a 4 KB `SIPH` JSON envelope (type, base/parent IDs, WAL/binlog positions, checksum); `restore` resolves the base→incremental chain and applies it in order, with `--up-to <id>` to stop early (cycle + broken-chain detection). Sync now streams through a bounded `jobs.Stream` (default 64×1 MB) instead of `io.Pipe`, exposing a `FillPercent` backpressure metric and propagating backup failures to the restore side via `CloseErr` (no truncated dump committed as clean).
-  - **Machinery in place, CLI gated (follow-up):** incremental backup capture (Postgres WAL replication slot + LSN; MySQL/MariaDB binlog position) — `backup --incremental` returns a clear "not yet wired" error; cross-engine type-mapping (canonical schema + JSONL emit/consume with per-engine identifier quoting and placeholders) — `sync --cross-engine` is capability-gated off until typed schema introspection exists; CDC continuous mode (state-file persistence + resume + capability gating) ships as a polling scaffold, not real logical-replication streaming.
-  - Known follow-up gates before incremental backup ships to users: Postgres orphan-replication-slot cleanup, validation of the `pg_receivewal`/`mysqlbinlog` streaming invocations against live servers (compile-checked only here).
+  - **Incremental backup wired end-to-end (`backup --incremental --base <id>`):** captures a **bounded** change set since the base dump's recorded end position, serialized as engine-neutral JSONL `CanonicalChange` records. Postgres bounds the pgoutput logical-decoding stream by `pg_current_wal_lsn()` captured at backup time; MySQL/MariaDB bound the binlog-tool decode by the current binlog file+offset. The incremental dump's envelope carries this capture's end position so the next incremental resumes exactly there. `restore` replays incremental links via `ApplyChange` (base links still restore natively). Postgres adds an orphan replication-slot sweep (`SweepOrphanSlots`) run before each capture — drops inactive `siphon_*` physical slots while preserving the persistent logical resume slot. `Incremental` capability is now `true` for all three drivers. Live-server behavior is integration-tested in CI (`wal_level=logical`); compile-checked locally.
+  - **CDC continuous sync wired end-to-end (`siphon cdc <from> <to>` / `sync --continuous`):** unbounded `ChangeStreamer.StreamChanges` → `CanonicalTransfer.ApplyChange` on the target, with an initial schema+data snapshot→stream handoff on first run (consistent start position captured before the snapshot) and resume-from-saved-position on restart (stable per-(source,target) job ID). Works same-engine and cross-engine (changes are engine-neutral `CanonicalChange`s) and honors `--table` in both the initial snapshot and the streamed changes. Persists the streamer's delivered position on exit (no ahead-of-stream periodic checkpoint, which could resume past un-applied changes after a crash); at-least-once delivery is safe because `ApplyChange` is idempotent (INSERT upsert, UPDATE/DELETE by PK). ctx cancel is the normal clean stop. `CDC` capability is now `true` for all three drivers. Same-engine, cross-engine, and resume paths are integration-tested in CI.
+  - **Cross-engine sync wired end-to-end (`sync --cross-engine`, e.g. Postgres → MySQL):** typed schema introspection via `driver.SchemaInspector` (`information_schema`/`pg_catalog`, incl. primary keys) builds a `CanonicalSchema`; the source emits canonical JSONL rows and the target re-creates tables (with primary keys) and inserts them, translating types via `MapToNative` — all with per-engine identifier quoting and parameterized values. Honors `--table`. Unmapped source column types now fail fast with an explicit error instead of silently coercing to `text` (which would corrupt binary/BLOB families). `CrossEngineSource`/`CrossEngineTarget` capabilities are now `true` for all three drivers. Scope is data + table structure + primary keys (secondary indexes, FKs, triggers, views, functions not translated). Integration-tested in CI (Postgres → MySQL); compile-checked locally.

README.md

@@ -17,7 +17,7 @@ A single binary that turns the painful, error-prone sprawl of `pg_dump` → `pg_
 ---
 
 > [!WARNING]
-> **Pre-1.0 — active development.** Postgres, MySQL, and MariaDB backup/restore/sync/verify/inspect work end-to-end today (Phases B + E), and bare `siphon` opens an interactive multi-panel dashboard (Phase C). The driver layer is hardened with a shared cross-driver test harness, capability gating, and connection retry (Phase D). Incremental backups, cross-engine sync, and ops features are on the [roadmap](#roadmap). APIs, flags, and the on-disk dump format may change before 1.0. Track progress via the milestone tags (`phase-a`, `phase-b`, `phase-c`, `phase-d`, `phase-e`, …).
+> **Pre-1.0 — active development.** Postgres, MySQL, and MariaDB backup/restore/sync/verify/inspect work end-to-end today (Phases B + E), and bare `siphon` opens an interactive multi-panel dashboard (Phase C). The driver layer is hardened with a shared cross-driver test harness, capability gating, and connection retry (Phase D). Advanced transfer is complete (Phase F): incremental backup/restore, cross-engine sync, and CDC continuous replication all work end-to-end. Ops features (cloud storage, secret backends, retention, telemetry) are on the [roadmap](#roadmap). APIs, flags, and the on-disk dump format may change before 1.0. Track progress via the milestone tags (`phase-a` … `phase-f`).
 
 ## Table of contents
 
@@ -54,7 +54,7 @@ A single binary that turns the painful, error-prone sprawl of `pg_dump` → `pg_
 | **C** — TUI dashboard             | Multi-panel Bubble Tea dashboard (profiles · dumps · jobs) with live job progress, backup/restore modal forms, and snapshot tests                                                                                                                                                                                                                                                                                                       | ✅ Complete |
 | **D** — Driver hardening          | Shared cross-driver test harness (`RunDriverSuite`), capability-gating helper (`RequireCapability`), Postgres connection-probe retry, and a `docs/DRIVERS.md` contributor guide                                                                                                                                                                                                                                                         | ✅ Complete |
 | **E** — MySQL + MariaDB           | Both drivers via a shared `_mysqlcommon` package (shared `Conn`, `mysqldump`/`mariadb-dump` backup, client-pipe restore), exercised by the Phase D `RunDriverSuite` harness                                                                                                                                                                                                                                                             | ✅ Complete |
-| **F** — Advanced transfer         | Bounded-buffer streaming sync + chain-walking incremental restore work today; the dump envelope, incremental WAL/binlog capture, cross-engine type-mapping, and CDC state machinery are in place but their CLI paths (`backup --incremental`, `sync --cross-engine`, CDC) are gated pending follow-up wiring — see [docs/INCREMENTAL.md](docs/INCREMENTAL.md), [docs/CROSS_ENGINE.md](docs/CROSS_ENGINE.md), [docs/CDC.md](docs/CDC.md) | 🟡 Partial  |
+| **F** — Advanced transfer         | All four advanced-transfer modes work end-to-end: bounded-buffer streaming sync; **incremental** backup/restore (`backup --incremental --base <id>` captures a bounded change set via Postgres logical decoding / MySQL-MariaDB binlog, `restore` replays the base→incremental chain, Postgres orphan-slot sweep); **cross-engine** sync (`sync --cross-engine` — typed `SchemaInspector` introspection → canonical type-mapping, e.g. Postgres → MySQL); and **CDC** (`siphon cdc` / `sync --continuous` — unbounded change streaming with snapshot→stream handoff, resumable, same- and cross-engine). Live DB paths are integration-tested in CI — see [docs/INCREMENTAL.md](docs/INCREMENTAL.md), [docs/CROSS_ENGINE.md](docs/CROSS_ENGINE.md), [docs/CDC.md](docs/CDC.md) | ✅ Complete |
 | **G** — Ops features              | Cloud storage, secret backends, profile groups + 2FA, team mode, audit log, retention, telemetry                                                                                                                                                                                                                                                                                                                                        | ⏳ Planned  |
 | **H** — Distribution              | GoReleaser, Homebrew tap, Scoop bucket, install script, docs site                                                                                                                                                                                                                                                                                                                                                                       | ⏳ Planned  |
 
@@ -222,7 +222,7 @@ Contributions are welcome. Please read [CONTRIBUTING.md](CONTRIBUTING.md), keep
 
 Adding a new database engine? See [docs/DRIVERS.md](docs/DRIVERS.md) for the driver contributor guide.
 
-Concept docs (honest as-built status — several paths are documented follow-ups): [docs/INCREMENTAL.md](docs/INCREMENTAL.md) (incremental restore works; the backup path is a follow-up), [docs/CROSS_ENGINE.md](docs/CROSS_ENGINE.md) (type-map matrix; `--cross-engine` is gated off pending typed introspection), and [docs/CDC.md](docs/CDC.md) (scaffold only; not yet runnable).
+Concept docs: [docs/INCREMENTAL.md](docs/INCREMENTAL.md) (incremental backup + restore), [docs/CROSS_ENGINE.md](docs/CROSS_ENGINE.md) (cross-engine sync + the type-map matrix), and [docs/CDC.md](docs/CDC.md) (continuous CDC sync, same- and cross-engine). All three work end-to-end; live DB behavior is integration-tested in CI.
 
 ## License
 

docs/CDC.md

@@ -2,19 +2,21 @@
 
 CDC ("continuous sync" / follow mode) keeps a target database in step with a
 source by streaming changes as they happen, rather than re-running a full
-backup/restore. In siphon this is **scaffolding today** — the resume-state
-plumbing and capability gating are in place, but continuous streaming is a
-future deliverable and does not run.
+backup/restore. siphon streams the source's logical change stream and applies
+each change to the target continuously, until interrupted. It works
+**same-engine** (Postgres→Postgres, MySQL→MySQL) and **cross-engine**
+(Postgres→MySQL) alike, because changes are carried as engine-neutral
+`CanonicalChange`s and replayed natively on the target.
 
 ## Table of contents
 
-- [What exists](#what-exists)
+- [How it works](#how-it-works)
 - [The CLI surface](#the-cli-surface)
 - [Status](#status)
 - [Resume state](#resume-state)
 - [Limitations](#limitations)
 
-## What exists
+## How it works
 
 `internal/app/cdc.go` holds:
 
@@ -24,49 +26,60 @@ future deliverable and does not run.
 - **`saveCDCState` / `loadCDCState`** — JSON state files under a per-user
   directory resolved from `SIPHON_STATE_HOME`, then `XDG_STATE_HOME`, then
   `$HOME/.local/state/siphon/cdc`.
-- **`RunCDC`** — a capability-gated entry point. It resolves the source and
-  target profiles, then requires both drivers to advertise `CapCDC`. Its body is
-  a **polling scaffold**: a 10-second ticker that persists state each tick and
-  resumes from a prior run's state on restart. It is **not** real change
-  streaming.
+- **`RunCDC`** — the capability-gated entry point. Both drivers must advertise
+  `CapCDC`. It connects source and target, then:
+  1. **First run (no prior state):** captures the source's current stream
+     position (`BasePositioner.CurrentPosition`), takes an initial schema+data
+     snapshot via the canonical transfer pipe (`InspectSchema` →
+     `EmitCanonical` → `ConsumeCanonical`), persists the start position, then
+     streams changes committed after it.
+  2. **Restart (prior state exists):** resumes from the saved position, no
+     snapshot.
+  3. **Stream loop:** `ChangeStreamer.StreamChanges` (unbounded) emits each
+     `CanonicalChange`; `RunCDC` applies it via `CanonicalTransfer.ApplyChange`
+     on the target and persists the streamer's delivered position on exit.
+
+The `job_id` is a stable hash of `(from, to)`, so re-running the same continuous
+sync resumes from the same state file.
 
 `CDCState` save/load round-trip, the state-dir resolution, and the
-no-capability rejection are unit-tested (`internal/app/cdc_test.go`).
+no-capability rejection are unit-tested (`internal/app/cdc_test.go`). End-to-end
+streaming (same-engine, cross-engine, and resume) is covered by integration
+tests (`internal/app/cdc_integration_test.go`, `-tags integration`).
 
 ## The CLI surface
 
-There is **no `siphon cdc` subcommand wired today.** `RunCDC` exists as an
-application function but is not registered as a Cobra command
-(`internal/cli/root.go` wires `backup`, `restore`, `sync`, `verify`, `inspect`,
-`profile`, `dumps`, `config`, `schedule`, `tunnel` — no `cdc`).
-
-`sync --continuous` exposes the flag but does not run CDC; it returns a clear
-error pointing at the (not-yet-wired) `siphon cdc` follow-up:
+Two equivalent entry points:
 
 ```bash
+# Dedicated command
+siphon cdc <from> <to>
+siphon cdc --from pg-prod --to pg-replica
+
+# sync follow mode (equivalent)
 siphon sync --from pg-prod --to pg-replica --continuous
-# Error: continuous CDC sync is not wired here; use `siphon cdc` (Phase F Task 10)
 ```
 
+Both stream continuously until interrupted. Press Ctrl-C to stop cleanly —
+ctx cancellation is the normal termination signal; the final position is
+persisted on exit so a later run resumes without a gap.
+
 ## Status
 
 | Capability | Status |
 | --- | --- |
 | `CDCState` persistence (save/load) | ✅ Works (unit-tested) |
 | State-dir resolution (`SIPHON_STATE_HOME` / `XDG_STATE_HOME`) | ✅ Works (unit-tested) |
-| Resume from prior state on restart | ✅ Works (in the scaffold loop) |
-| Capability gating on `CapCDC` | ✅ Works |
-| `siphon cdc` CLI subcommand | ❌ Not wired (no Cobra command) |
-| Continuous change streaming | ⚠️ Scaffold only (polling tick; not real CDC) |
-
-CDC does **not run today**. No driver declares `CapCDC` true, so `RunCDC` is
-rejected with `ErrDriverUnsupported`. Even if a driver enabled it, the loop is a
-polling scaffold, not logical-replication streaming.
+| Resume from prior state on restart | ✅ Works |
+| Capability gating on `CapCDC` | ✅ Works (true on postgres, mysql, mariadb) |
+| `siphon cdc` CLI subcommand | ✅ Wired |
+| `sync --continuous` follow mode | ✅ Routes to `RunCDC` |
+| Initial snapshot → stream handoff | ✅ Works |
+| Continuous change streaming (same + cross-engine) | ✅ Works |
 
 ## Resume state
 
-When CDC streaming is built, `RunCDC` resumes from the last persisted position.
-State files live at:
+`RunCDC` resumes from the last persisted position. State files live at:
 
 ```text
 $SIPHON_STATE_HOME/cdc/<job-id>.state        # if SIPHON_STATE_HOME is set
@@ -77,12 +90,25 @@ $HOME/.local/state/siphon/cdc/<job-id>.state  # default
 Each file is JSON: source/target profiles plus the last applied position (LSN
 for Postgres, binlog file + offset for MySQL/MariaDB).
 
+**Checkpoint/resume granularity is "since the streamer's last delivered
+position."** RunCDC persists the position the change streamer reports when the
+stream stops — a position tied to what was actually delivered, never ahead of
+it. There is deliberately **no** ahead-of-stream periodic checkpoint: writing
+the source's *current* WAL/binlog end mid-stream would, after a crash, resume
+past changes that were streamed but never applied — silent data loss. After a
+crash it resumes from the last persisted position and re-applies the tail. This
+is safe because delivery is **at-least-once** and `ApplyChange` is
+**idempotent**: INSERT is idempotent (upsert), and UPDATE/DELETE target by
+primary key. Re-applying the tail is therefore a no-op — no gaps, no duplicates.
+
 ## Limitations
 
-- **Not enabled on any driver** — `CapCDC` is `false` everywhere, so CDC is
-  rejected up front.
-- **No real streaming** — the scaffold polls on a ticker. Real logical-
-  replication streaming (`pglogrepl` for Postgres, binlog tailing for
-  MySQL/MariaDB) is deferred to a future revision.
-- **No CLI entry point** — `RunCDC` is internal application scaffolding, not a
-  user-facing command yet.
+- **Postgres source requires `wal_level=logical`** and sufficient
+  `max_replication_slots` / `max_wal_senders`; MySQL/MariaDB source requires
+  row-based binlogging. These are the same prerequisites as incremental backup.
+- **At-least-once, not exactly-once** — see resume granularity above. Correct
+  because apply is idempotent; a target with non-idempotent side effects
+  (triggers, etc.) is out of scope.
+- **Snapshot consistency window** — the start position is captured before the
+  snapshot, so changes committed during the snapshot are re-streamed and
+  re-applied idempotently rather than lost.