fix: apply CodeRabbit review fixes for Phase F

- Apply post-review fixes to incremental/CDC/cross-engine data paths.
- Stop advancing the PG decode position from keepalive ServerWALEnd:
  track server WAL end separately so a keepalive crossing the bound
  cannot end a bounded incremental before the changes at that LSN are
  decoded and emitted (silent data loss); the catch-up stop now uses
  the separate server position.
- Remove the CDC ahead-of-stream periodic checkpoint: persisting the
  source's current cursor mid-stream could resume past un-applied
  changes after a crash. Checkpoint only the streamer's delivered
  position on exit; idempotent ApplyChange makes the replayed tail safe.
- Honor --table in cross-engine sync and CDC (snapshot + streamed
  changes) via shared filterSchemaTables/tableAllowed helpers.
- Surface mysqlbinlog tool failures: capture stderr, return non-zero
  exits that are not an intentional stop, and error when a bounded scan
  ends before reaching its stop position.
- Distinguish SQL NULL from the string 'NULL' in the binlog parser via
  a typed colVal token; strip type comments only outside quoted strings.
- Fail fast on unmapped Postgres/MySQL column types instead of silently
  coercing to text (would corrupt binary/BLOB families cross-engine).
- Preflight incremental-replay support before a Clean base restore;
  validate base profile/driver compatibility and a non-empty position
  before an incremental backup.
- Replay JSONL via json.Decoder to drop the 8 MiB Scanner line cap.
- Guard empty column sets in BuildUpdateSQL/BuildDeleteSQL; require both
  endpoints in the cdc CLI; restrict the orphan-slot sweep to physical
  slots.
- Update CDC docs and CHANGELOG to match the new checkpoint and table
  behavior.

Author: nixrajput

CHANGELOG.md

@@ -25,5 +25,5 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - 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).
   - **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). Checkpoints state every 100 changes plus on clean exit; 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. `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.
+  - **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.

docs/CDC.md

@@ -37,7 +37,7 @@ each change to the target continuously, until interrupted. It works
      snapshot.
   3. **Stream loop:** `ChangeStreamer.StreamChanges` (unbounded) emits each
      `CanonicalChange`; `RunCDC` applies it via `CanonicalTransfer.ApplyChange`
-     on the target and checkpoints state periodically.
+     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.
@@ -90,10 +90,14 @@ $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 last checkpoint."** RunCDC
-checkpoints every 100 applied changes plus on clean exit. After a crash it
-resumes from the last checkpoint and re-applies any changes that landed after
-it. This is safe because delivery is **at-least-once** and `ApplyChange` is
+**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.
 

internal/app/backup.go

@@ -237,10 +237,32 @@ func runIncrementalBackup(ctx context.Context, d Deps, opt BackupOpts, resolved
 		}
 	}
 
+	// Reject a base that does not belong to this profile/driver: an incremental
+	// chain is only meaningful against the same engine and source, and a
+	// cross-driver base would produce changes the target cannot replay.
+	if base.Driver != resolved.Driver || base.Profile != opt.Profile {
+		return &errs.Error{
+			Op:    "app.backup.incremental",
+			Code:  errs.CodeUser,
+			Cause: errs.ErrIncompatibleEngine,
+			Hint:  "base dump " + opt.BaseID + " was created for profile " + base.Profile + " (" + base.Driver + "), not " + opt.Profile + " (" + resolved.Driver + ")",
+		}
+	}
+
 	since, err := basePosition(d, opt.BaseID)
 	if err != nil {
 		return err
 	}
+	// A base with no recorded change-stream position cannot anchor an incremental;
+	// capturing from a zero cursor would silently start at "now" and drop every
+	// change committed since the base. Require a real position.
+	if since.LSN == "" && since.BinlogFile == "" {
+		return &errs.Error{
+			Op:   "app.backup.incremental",
+			Code: errs.CodeUser,
+			Hint: "base dump " + opt.BaseID + " has no recorded change-stream position; take a fresh full backup and use that as --base",
+		}
+	}
 
 	inc, ok := conn.(driver.IncrementalBackuper)
 	if !ok {

internal/app/cdc.go

@@ -93,13 +93,6 @@ func cdcJobID(from, to string) string {
 	return "cdc-" + hex.EncodeToString(sum[:8])
 }
 
-// cdcCheckpointEvery bounds how many applied changes pass between resume-state
-// checkpoints. At-least-once delivery plus idempotent ApplyChange makes coarse
-// checkpointing safe: on restart we resume from the last checkpoint and re-apply
-// any changes that landed after it, which is a no-op for INSERT (idempotent),
-// UPDATE, and DELETE (both by primary key).
-const cdcCheckpointEvery = 100
-
 // RunCDC starts (or resumes) a continuous, unbounded sync: it tails the source's
 // logical change stream and applies each engine-neutral CanonicalChange to the
 // target. It works same-engine and cross-engine alike because CanonicalChange is
@@ -110,10 +103,11 @@ const cdcCheckpointEvery = 100
 // changes committed after that position. On a restart it resumes from the saved
 // position with no snapshot.
 //
-// Resume granularity is "since the last checkpoint": StreamChanges reports the
-// position reached after each applied change via the emit callback, and RunCDC
-// checkpoints every cdcCheckpointEvery changes plus on clean exit. Re-applying the
-// tail after a crash is safe (see cdcCheckpointEvery).
+// Resume granularity is "since the last clean exit": RunCDC persists the
+// streamer's returned final position when the stream stops (the position is tied
+// to what was actually delivered, never ahead of it). Re-applying the tail after
+// a crash is safe because ApplyChange is idempotent (INSERT upserts; UPDATE and
+// DELETE target by primary key).
 //
 // ctx cancellation is the normal stop signal (matching the bounded-stream
 // convention): on cancel RunCDC persists final state and returns nil; only a
@@ -187,7 +181,7 @@ func RunCDC(parent context.Context, d Deps, opt SyncOpts) (<-chan jobs.Event, st
 				if err != nil {
 					return err
 				}
-				if snapErr := cdcSnapshot(ctx, srcConn, dstXfer, src.Driver); snapErr != nil {
+				if snapErr := cdcSnapshot(ctx, srcConn, dstXfer, src.Driver, opt.Tables); snapErr != nil {
 					return snapErr
 				}
 				state.setPosition(from)
@@ -199,6 +193,12 @@ func RunCDC(parent context.Context, d Deps, opt SyncOpts) (<-chan jobs.Event, st
 
 			applied := 0
 			emit2 := func(ch canonical.CanonicalChange) error {
+				// Honor --table: the streamer surfaces changes for every table (the
+				// source's publication/binlog is database-wide), so skip any change
+				// outside the requested set before applying it to the target.
+				if !tableAllowed(ch.Table, opt.Tables) {
+					return nil
+				}
 				if applyErr := dstXfer.ApplyChange(ctx, ch); applyErr != nil {
 					return applyErr
 				}
@@ -207,12 +207,13 @@ func RunCDC(parent context.Context, d Deps, opt SyncOpts) (<-chan jobs.Event, st
 					Phase:   jobs.PhaseProgress,
 					Message: fmt.Sprintf("applied %s on %s (%d total)", ch.Op, ch.Table, applied),
 				})
-				if applied%cdcCheckpointEvery == 0 {
-					if pos, posErr := srcPositioner.CurrentPosition(ctx); posErr == nil {
-						state.setPosition(pos)
-						_ = saveCDCState(state)
-					}
-				}
+				// No periodic mid-stream checkpoint here: srcPositioner.CurrentPosition
+				// returns the source's CURRENT WAL/binlog end, which is ahead of the
+				// change we just applied. Persisting it would, after a crash, resume
+				// PAST changes that streamed but were never applied — silent data loss.
+				// We checkpoint only the streamer's returned final position on exit
+				// (below), which is tied to what was actually delivered; at-least-once
+				// redelivery on resume is safe because ApplyChange is idempotent.
 				return nil
 			}
 
@@ -239,7 +240,7 @@ func RunCDC(parent context.Context, d Deps, opt SyncOpts) (<-chan jobs.Event, st
 // the canonical transfer pipe (the same pattern as runCrossEngineSync). Both the
 // source's SchemaInspector+CanonicalTransfer and the target's CanonicalTransfer
 // are required.
-func cdcSnapshot(ctx context.Context, srcConn driver.Conn, dstXfer driver.CanonicalTransfer, srcDriverName string) error {
+func cdcSnapshot(ctx context.Context, srcConn driver.Conn, dstXfer driver.CanonicalTransfer, srcDriverName string, tables []string) error {
 	srcInsp, ok := srcConn.(driver.SchemaInspector)
 	if !ok {
 		return cdcUnsupported(srcDriverName, "SchemaInspector")
@@ -253,6 +254,9 @@ func cdcSnapshot(ctx context.Context, srcConn driver.Conn, dstXfer driver.Canoni
 	if err != nil {
 		return err
 	}
+	// Snapshot only the requested tables; the streamed-change phase applies the
+	// same filter in emit2.
+	schema = filterSchemaTables(schema, tables)
 
 	stream := jobs.NewStream(64)
 	errCh := make(chan error, 1)

internal/app/restore.go

@@ -1,9 +1,9 @@
 package app
 
 import (
-	"bufio"
 	"context"
 	"encoding/json"
+	"errors"
 	"io"
 	"os"
 
@@ -58,6 +58,22 @@ func Restore(parent context.Context, d Deps, opt RestoreOpts) (<-chan jobs.Event
 			}
 			defer func() { _ = conn.Close() }()
 
+			// Preflight: if the chain contains any incremental link, the driver must
+			// support change replay (CanonicalTransfer). Assert this BEFORE applying
+			// the base — a Clean base restore wipes the target, so discovering the
+			// unsupported-driver error only when we reach the first incremental link
+			// would leave the target destroyed and the chain half-applied.
+			if chainHasIncremental(chain) {
+				if _, ok := conn.(driver.CanonicalTransfer); !ok {
+					return &errs.Error{
+						Op:    "restore",
+						Code:  errs.CodeUser,
+						Cause: errs.ErrDriverUnsupported,
+						Hint:  resolved.Driver + " cannot replay incremental change links",
+					}
+				}
+			}
+
 			for i, m := range chain {
 				emit(jobs.Event{Phase: jobs.PhaseProgress, Message: "applying " + m.ID})
 				f, err := os.Open(d.Dumps.Path(m.ID))
@@ -120,28 +136,40 @@ func Restore(parent context.Context, d Deps, opt RestoreOpts) (<-chan jobs.Event
 	})
 }
 
+// chainHasIncremental reports whether any link in the resolved chain is an
+// incremental dump (one that carries a JSONL change body replayed via
+// ApplyChange). Incremental links record a non-empty ParentID in their metadata;
+// the chain root (a base or full dump) does not.
+func chainHasIncremental(chain []dumps.Meta) bool {
+	for _, m := range chain {
+		if m.ParentID != "" {
+			return true
+		}
+	}
+	return false
+}
+
 // replayChanges decodes a JSONL stream of CanonicalChanges from r and applies
-// each to the target via ApplyChange, in order. A malformed line aborts the
+// each to the target via ApplyChange, in order. A malformed record aborts the
 // replay so a corrupt incremental never partially applies undetected.
+//
+// A json.Decoder streams successive JSON values directly off the reader, so —
+// unlike a bufio.Scanner — there is no per-record size ceiling that would reject
+// a valid change carrying a large row value as "corrupt".
 func replayChanges(ctx context.Context, applier driver.CanonicalTransfer, r io.Reader) error {
-	sc := bufio.NewScanner(r)
-	sc.Buffer(make([]byte, 0, 64*1024), 8*1024*1024)
-	for sc.Scan() {
-		line := sc.Bytes()
-		if len(line) == 0 {
-			continue
-		}
+	dec := json.NewDecoder(r)
+	for {
 		var ch canonical.CanonicalChange
-		if err := json.Unmarshal(line, &ch); err != nil {
+		if err := dec.Decode(&ch); err != nil {
+			if errors.Is(err, io.EOF) {
+				break
+			}
 			return &errs.Error{Op: "restore.replay", Code: errs.CodeSystem, Cause: err, Hint: "incremental change body is corrupt"}
 		}
 		if err := applier.ApplyChange(ctx, ch); err != nil {
 			return err
 		}
 	}
-	if err := sc.Err(); err != nil {
-		return &errs.Error{Op: "restore.replay", Code: errs.CodeSystem, Cause: err}
-	}
 	return nil
 }
 

internal/app/sync.go

@@ -3,6 +3,7 @@ package app
 import (
 	"context"
 
+	"github.com/nixrajput/siphon/internal/canonical"
 	"github.com/nixrajput/siphon/internal/driver"
 	"github.com/nixrajput/siphon/internal/errs"
 	"github.com/nixrajput/siphon/internal/jobs"
@@ -179,6 +180,10 @@ func runCrossEngineSync(parent context.Context, d Deps, opt SyncOpts) (<-chan jo
 			if err != nil {
 				return err
 			}
+			// Honor --table: native sync passes opt.Tables into Backup, so the
+			// cross-engine path must filter the canonical schema the same way or it
+			// would copy every table regardless of the requested subset.
+			schema = filterSchemaTables(schema, opt.Tables)
 
 			stream := jobs.NewStream(64)
 			errCh := make(chan error, 1)
@@ -200,3 +205,32 @@ func runCrossEngineSync(parent context.Context, d Deps, opt SyncOpts) (<-chan jo
 		},
 	})
 }
+
+// tableAllowed reports whether table t passes a --table filter. An empty filter
+// (the default) allows every table.
+func tableAllowed(t string, filter []string) bool {
+	if len(filter) == 0 {
+		return true
+	}
+	for _, want := range filter {
+		if want == t {
+			return true
+		}
+	}
+	return false
+}
+
+// filterSchemaTables returns a schema containing only the tables that pass the
+// --table filter. An empty filter returns the schema unchanged.
+func filterSchemaTables(schema *canonical.CanonicalSchema, filter []string) *canonical.CanonicalSchema {
+	if len(filter) == 0 {
+		return schema
+	}
+	out := &canonical.CanonicalSchema{}
+	for _, t := range schema.Tables {
+		if tableAllowed(t.Name, filter) {
+			out.Tables = append(out.Tables, t)
+		}
+	}
+	return out
+}

internal/canonical/canonical.go

@@ -226,6 +226,12 @@ func BuildCreateTableSQL(engine string, t CanonicalTable) (string, error) {
 // come first (1-based placeholders), then key columns. Identifiers are always
 // quoted via QuoteIdent; values are bound, not interpolated.
 func BuildUpdateSQL(engine, table string, setCols, keyCols []string) (string, error) {
+	if len(setCols) == 0 {
+		return "", fmt.Errorf("cross-engine: update %s: no columns to set", table)
+	}
+	if len(keyCols) == 0 {
+		return "", fmt.Errorf("cross-engine: update %s: no key columns", table)
+	}
 	qt, err := QuoteIdent(engine, table)
 	if err != nil {
 		return "", err
@@ -256,6 +262,9 @@ func BuildUpdateSQL(engine, table string, setCols, keyCols []string) (string, er
 // key columns in the WHERE clause. Identifiers are always quoted; values are
 // bound (1-based for Postgres, ? for MySQL/MariaDB).
 func BuildDeleteSQL(engine, table string, keyCols []string) (string, error) {
+	if len(keyCols) == 0 {
+		return "", fmt.Errorf("cross-engine: delete %s: no key columns", table)
+	}
 	qt, err := QuoteIdent(engine, table)
 	if err != nil {
 		return "", err

internal/cli/cdc.go

@@ -4,6 +4,7 @@ import (
 	"github.com/spf13/cobra"
 
 	"github.com/nixrajput/siphon/internal/app"
+	"github.com/nixrajput/siphon/internal/errs"
 )
 
 func newCdcCmd() *cobra.Command {
@@ -24,6 +25,13 @@ func newCdcCmd() *cobra.Command {
 			if len(args) >= 2 {
 				toName = args[1]
 			}
+			if fromName == "" || toName == "" {
+				return &errs.Error{
+					Op:   "cdc",
+					Code: errs.CodeUser,
+					Hint: "cdc requires both source and target profiles (positional args or --from/--to)",
+				}
+			}
 			deps, err := buildDeps()
 			if err != nil {
 				return err