Complete Phase F: wire incremental, cross-engine, and CDC end-to-end#6
Merged
Conversation
…nal/canonical leaf - Move types (CanonicalType, CanonicalColumn, CanonicalTable, CanonicalSchema, CanonicalRow) and MapToNative from internal/app/canonical.go to new leaf package internal/canonical/canonical.go - Add PrimaryKey bool field to CanonicalColumn - Add new types: ChangeOp (OpInsert/OpUpdate/OpDelete), CanonicalChange, Position - Export quoteIdent→QuoteIdent and placeholder→Placeholder - Move pure db-free builders (BuildSelectSQL, BuildCreateTableSQL, BuildInsertSQL, WriteJSONL, NormalizeScanned) to internal/canonical - Update canonical_emit.go and canonical_consume.go to import and qualify canonical.* symbols; db-touching orchestration stays in internal/app - Move pure-function tests to internal/canonical/canonical_test.go - internal/canonical imports stdlib only (leaf package invariant)
- Add BuildUpdateSQL and BuildDeleteSQL to internal/canonical. - Add PRIMARY KEY clause to BuildCreateTableSQL when PK columns exist. - Add changeColumns and ApplyChange to internal/app for CDC/restore. - Add CanonicalChange JSON round-trip test in canonical package.
…engine - Add ChangeColumns + ValidateChangeKey exported fns to canonical pkg. - Add driver.SchemaInspector + driver.CanonicalTransfer optional interfaces. - Implement InspectSchema, EmitCanonical, ConsumeCanonical, ApplyChange on postgres.Conn and mysqlcommon.Conn (engine string field added to Conn). - Flip CrossEngineSource/Target = true in all three driver Capabilities. - Rewrite runCrossEngineSync to use the real inspect→emit→consume pipeline. - Delete app canonical_emit/consume/test; logic now lives in driver pkgs.
…pplyChange guard test
- Add driver.ChangeStreamer interface: stream engine-neutral CanonicalChanges from a Position, returning the final Position. - Postgres: pglogrepl pgoutput (proto v1) logical decoding over a replication-mode conn; ensure siphon_pub publication + siphon_logical slot; decode Insert/Update/Delete with PK-flag key extraction; periodic standby acks; ctx cancel is a clean stop. - MySQL/MariaDB: shell out to the fork binlog tool with --verbose --base64-output=DECODE-ROWS and parse the ### ROW pseudo-SQL into CanonicalChanges; map @n to column names + PK via information_schema. - Tests: offline binlog-parser unit tests; bounded Postgres logical stream integration test (wal_level=logical container). - Capabilities unchanged: verb wiring + cap flips belong to Task 5/6.
- Add driver.IncrementalBackuper: bounded change capture from a base position to the engine's current end, emitted as JSONL CanonicalChange records. Postgres bounds the pgoutput stream by pg_current_wal_lsn(); MySQL/MariaDB by the current binlog offset. - Wire app.Backup --incremental: read the base envelope end position, capture to a temp body, then write envelope(end pos) + body and a Meta linking BaseID/ParentID. - Replay incremental restore links via ApplyChange; base links still restore natively. - Add Postgres SweepOrphanSlots: drop inactive siphon_ physical slots, preserving the persistent logical resume slot. - Un-gate the CLI and flip the Postgres Incremental capability. - Update restore-chain tests, README, CHANGELOG, INCREMENTAL.md.
…ntal resumes correctly - Add driver.BasePositioner; implement CurrentPosition on Postgres (pg_current_wal_lsn) and MySQL/MariaDB (binlog file+offset) Conns with compile assertions. - app.Backup full path now streams the dump body to a temp file, captures the engine position just after Backup returns, and stamps WALEnd / BinlogFile+BinlogEnd into the base envelope before assembling envelope+body. Previously full base envelopes carried no position, so the first incremental off a full base started from "now" and silently dropped changes committed between the base dump and the incremental run. - Make incremental-replay INSERTs idempotent via canonical.BuildIdempotentInsertSQL (ON CONFLICT DO NOTHING / INSERT IGNORE) so a change captured in both base and first incremental re-applies harmlessly. - Add covering unit test that a full backup whose driver implements BasePositioner stamps a non-empty WALEnd into the base envelope, read back via basePosition; plus canonical idempotent-INSERT tests. - Document the base-position stamping and capture-after-dump rationale in docs/INCREMENTAL.md.
- Replace RunCDC polling scaffold with unbounded StreamChanges to ApplyChange on the target; works same-engine and cross-engine via engine-neutral CanonicalChange. - Add initial snapshot to stream handoff: capture source position before snapshot, then stream changes committed after it. - Resume from a state file keyed by a stable per source/target job ID; checkpoint every 100 changes plus on clean exit. At-least-once delivery is safe because ApplyChange is idempotent. - Route sync --continuous to RunCDC; add siphon cdc command and register it in root. - Flip CDC capability to true on postgres, mysql, mariadb. - Add integration tests: same-engine, cross-engine, and resume. - Update docs/CDC.md, README, CHANGELOG to reflect shipped CDC.
|
Warning Review limit reached
More reviews will be available in 51 minutes and 37 seconds. Learn how PR review limits work. Your organization has run out of usage credits. Purchase more credits in the billing tab to continue. ⌛ How to resolve this issue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based credits. 🚦 How do rate limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan refill rate. For paid Pro and Pro+ PR reviews, CodeRabbit uses rolling per-developer review limits. Reviews become available again as older review attempts age out of the rolling limit window. Please see our Fair Usage Limits Policy for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (2)
📝 WalkthroughWalkthroughPhase F is wired end-to-end: a new ChangesPhase F: Incremental Backup, CDC, and Cross-Engine Sync
Sequence Diagram(s)sequenceDiagram
participant CLI
participant RunCDC
participant cdcSnapshot
participant ChangeStreamer
participant ApplyChange
participant CDCState
rect rgba(100, 149, 237, 0.5)
note over CLI,CDCState: CDC Continuous Sync (same-engine or cross-engine)
CLI->>RunCDC: siphon cdc src dst (or siphon sync --continuous src dst)
RunCDC->>CDCState: load saved position
alt first run (no saved position)
RunCDC->>cdcSnapshot: InspectSchema → EmitCanonical → ConsumeCanonical
cdcSnapshot-->>RunCDC: initial snapshot complete
end
RunCDC->>ChangeStreamer: StreamChanges(ctx, resumePos, emit)
loop per CanonicalChange from streamer
ChangeStreamer-->>RunCDC: emit(CanonicalChange)
RunCDC->>ApplyChange: ApplyChange(ch) on target
RunCDC->>CDCState: checkpoint every 100 changes
end
note over RunCDC: context cancel or stream end
RunCDC->>CDCState: persist final Position
RunCDC-->>CLI: nil (or error if non-cancel failure)
end
sequenceDiagram
participant CLI
participant Backup
participant IncrementalBackuper
participant Restore
participant ApplyChange
rect rgba(144, 238, 144, 0.5)
note over CLI,ApplyChange: Incremental Backup
CLI->>Backup: siphon backup --incremental --base baseID
Backup->>Backup: read baseID envelope, extract WALEnd/BinlogFile/BinlogEnd
Backup->>IncrementalBackuper: BackupIncremental(ctx, sincePos, writer)
IncrementalBackuper-->>Backup: endPos + stream of JSONL CanonicalChanges
Backup->>Backup: write EnvelopeIncremental(parentID, rootID, endPos)
end
rect rgba(255, 165, 0, 0.5)
note over CLI,ApplyChange: Incremental Restore
CLI->>Restore: siphon restore chainID
Restore->>Restore: walk chain, detect EnvelopeIncremental
loop for each incremental envelope in chain
Restore->>Restore: decode JSONL CanonicalChange records
Restore->>ApplyChange: ApplyChange(ch) insert/update/delete
end
Restore-->>CLI: restore complete
end
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 19
🧹 Nitpick comments (1)
internal/driver/_mysqlcommon/changestream_test.go (1)
11-24: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd regressions for the parser edge cases fixed above.
Please cover
@N='NULL'vs@N=NULL, quoted values containing/*, and a single-row transcript followed only by a non-DML# at/Xid marker. Those cases exercise the real CDC data-loss/corruption paths.Also applies to: 61-82
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@internal/driver/_mysqlcommon/changestream_test.go` around lines 11 - 24, Add regression test cases to the tests slice in TestParseColAssign to cover the edge cases mentioned: add a test case where the assigned value is the string literal 'NULL' (not the actual NULL keyword) to distinguish from the existing `@4`=NULL case, add a test case for a quoted string value that contains a forward slash and asterisk (like 'value/*comment') to ensure the parser correctly handles quoted values with comment-like syntax inside them, and consider adding test cases for non-DML markers following row assignments. These cases should be appended to the existing tests slice in the struct initialization to exercise the CDC data-loss and corruption edge cases.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@internal/app/backup.go`:
- Around line 77-96: The cursor position is being captured after the conn.Backup
method returns, which causes writes committed during the dump to be skipped in
subsequent incremental backups. Move the engine cursor capture to occur before
conn.Backup is called to ensure the position reflects the exact snapshot point
of the dump, or enhance the driver API to return the snapshot cursor as part of
the Backup result. Ensure the position used for the envelope is captured at or
before the dump snapshot, not after it completes.
- Around line 230-245: After reading the base metadata using
d.Dumps.ReadMeta(opt.BaseID) into the base variable, add validation checks to
ensure the base is compatible with the current driver and profile before
proceeding with the incremental backup. Verify that the base has a recorded
stream position (WALEnd or binlog position) available. If the base is from a
different profile/driver or lacks a valid stream position, return an appropriate
error before the code reaches the driver.IncrementalBackuper type assertion.
This prevents invalid backup chains and ensures incremental backups don't
silently start from a zero cursor.
In `@internal/app/cdc.go`:
- Around line 190-202: The RunCDC function does not respect the SyncOpts.Tables
filter, causing it to snapshot and apply changes for all tables regardless of
what tables were specified. Pass the opt.Tables filter to the cdcSnapshot
function to limit the initial snapshot to only the requested tables, and add
filtering logic in the emit2 closure to prevent ApplyChange from processing
streamed changes that are outside the requested table set. Use the table filter
from the options to determine which changes should be applied.
- Around line 186-190: The stream cursor is being recorded at line 186 with
srcPositioner.CurrentPosition, but the actual stream is not established until
the StreamChanges call around line 219, leaving a gap where changes are
unprotected. Refactor the code to establish the stream cursor position before
calling cdcSnapshot instead of after, either by moving the
srcPositioner.CurrentPosition call closer to the StreamChanges invocation or by
implementing a driver handoff that reserves the stream position through
src.Driver before the snapshot begins, ensuring all changes are protected by the
logical slot from the moment the snapshot starts.
- Around line 210-214: The periodic checkpoint logic that executes when
applied%cdcCheckpointEvery equals 0 is using srcPositioner.CurrentPosition()
which returns a position ahead of changes that haven't been applied yet,
creating a data loss risk on restart. Remove the entire conditional block
containing the srcPositioner.CurrentPosition() call, state.setPosition() call,
and saveCDCState() call to eliminate ahead-of-stream checkpointing until the
streamer exposes per-change positions that can be safely used for checkpointing.
In `@internal/app/restore.go`:
- Around line 87-96: The check for CanonicalTransfer driver capability is
performed too late in the restore process—it only occurs when an incremental
envelope is encountered during restoration, meaning the base dump may have
already been applied and potentially wiped the target. Move the driver
capability validation earlier by scanning the entire envelope chain before
starting any restore operations to ensure the driver supports incremental
replay. Verify that CanonicalTransfer is supported by the driver upfront if any
incremental envelopes exist in the chain, rather than discovering the
unsupported-driver error after the base has already been restored.
- Around line 127-143: Replace the bufio.Scanner-based line scanning approach
with json.Decoder to remove the 8 MiB line-size limitation that incorrectly
marks valid large changes as corrupt. Remove the sc := bufio.NewScanner(r)
initialization and sc.Buffer(...) call, then replace the for sc.Scan() loop by
creating a json.Decoder from the reader and using decoder.Decode(&ch) to
directly unmarshal each JSON object into the canonical.CanonicalChange variable.
This eliminates the manual json.Unmarshal call and the Scanner.Err() check,
instead relying on the Decoder to handle EOF and parsing errors appropriately.
In `@internal/app/sync.go`:
- Around line 178-187: After inspecting the full schema via
srcInsp.InspectSchema, filter the schema.Tables to only include the tables
specified in opt.Tables before passing the schema to srcXfer.EmitCanonical. This
ensures that cross-engine sync respects the user-specified table filter (passed
as opt.Tables) rather than emitting all tables in the schema, maintaining
consistency with native sync behavior.
In `@internal/canonical/canonical.go`:
- Around line 228-272: The BuildUpdateSQL and BuildDeleteSQL functions generate
malformed SQL when their column slices are empty (e.g., when setCols is empty in
BuildUpdateSQL or keyCols is empty in BuildDeleteSQL). Add guard clauses at the
start of each function to validate that the required column slices are not
empty, and return an error with a descriptive message if they are. For
BuildUpdateSQL, check that setCols has at least one element, and for
BuildDeleteSQL, check that keyCols has at least one element before proceeding
with the SQL construction logic.
In `@internal/cli/cdc.go`:
- Around line 19-35: The RunE function in the cobra command handler accepts up
to two arguments but does not validate that both required CDC endpoints are
provided, allowing empty fromName or toName values to be passed to app.RunCDC.
Add validation after processing the command arguments to ensure both fromName
and toName are non-empty strings. If either endpoint is missing, return a cobra
usage error with a clear message before calling buildDeps(), so the CLI fails
fast with a user-facing error that explains both endpoints are required.
In `@internal/driver/_mysqlcommon/changestream.go`:
- Around line 109-110: The parseColAssign function currently returns a string
value that conflates bare SQL NULL with the literal string "NULL", causing
binlogValue to incorrectly convert both to nil and corrupt legitimate string
values. Additionally, the parser strips mysqlbinlog type comments before
respecting quoted strings, truncating values like 'a /* b'. Refactor
parseColAssign to return a typed token (such as a struct with value and isNull
fields) instead of a plain string, update binlogValue and all other callers at
lines 271-287, 387-395, and 433-435 to work with this new token type, and ensure
that comment stripping for /* ... */ patterns occurs only outside of quoted
string boundaries to preserve quoted content correctly.
- Around line 65-88: The code currently ignores the exit status from cmd.Wait()
by assigning it to a blank identifier, and does not capture stderr from the
binlog tool. If the binlog tool exits early due to bad flags, authentication
issues, or missing files, parseBinlogRows may hit EOF without a scanner error
and the code returns success incorrectly. Modify the command setup to capture
stderr (similar to how stdout is captured via StdoutPipe), then after calling
cmd.Wait() on the binlog command, check the returned error and include the
captured stderr in the error response if the command exits with a non-zero
status. Additionally, for bounded streams where stopAt is specified, verify that
the endPos returned from parseBinlogRows actually reached the expected stopAt
position, otherwise return an error indicating the stream did not reach the
expected position.
- Around line 125-157: The pending row events should be flushed at each binlog
"# at" marker boundary to improve responsiveness in unbounded CDC streams.
Currently, the flush() call only executes when the stopAt condition is met,
causing rows to remain buffered until a later event or EOF. Move the flush()
call to execute whenever parseAtMarker successfully detects a binlog position
marker (when ok is true), before the stopAt condition check, so that accumulated
changes are emitted promptly at each event boundary. The stopAt boundary
condition should remain after the flush to ensure correctness for bounded scans.
In `@internal/driver/_mysqlcommon/inspect_schema.go`:
- Around line 47-50: The mapMySQLType function is silently defaulting unmapped
MySQL/MariaDB types to CTText, which corrupts schema semantics for types like
BLOB. Modify the mapMySQLType function to return an error (in addition to the
type) when it encounters an unmapped dataType or columnType parameter instead of
defaulting to CTText. Then update all call sites of mapMySQLType to check and
handle this error by returning an error from the parent function, ensuring
unsupported types surface as explicit failures rather than being silently
converted to CTText.
In `@internal/driver/postgres/changestream_integration_test.go`:
- Around line 75-83: The test captures startLSN before creating the logical
replication slot, which fails to validate the WAL retention contract since the
slot is not instantiated until StreamChanges is called later. Reorder the
operations to create the logical replication slot first by establishing a
replication connection using pgconn.Connect with replicationDSN and calling
ensureLogicalSlot, then capture the startLSN using the SELECT pg_current_wal_lsn
query, and finally ensure the publication exists by calling ensurePublication.
This ensures the test properly validates that the slot correctly anchors WAL
retention from the captured start LSN point.
In `@internal/driver/postgres/changestream.go`:
- Around line 111-128: The keepalive message handler is incorrectly advancing
clientXLogPos by assigning ServerWALEnd to it, which represents the server's
current position rather than data this client has decoded. Find and remove the
assignment statement that sets clientXLogPos to pkm.ServerWALEnd (the
conditional block checking if pkm.ServerWALEnd > clientXLogPos), keeping
clientXLogPos advancement only through successful decodeWALData processing so
that bounded incremental stops with stopLSN work correctly without persisting
state prematurely.
- Around line 20-26: The clientXLogPos is being advanced from keepalive
ServerWALEnd at lines 167-169 without corresponding decoded changes, causing the
stop condition at lines 120-124 to trigger prematurely and skip changes. Remove
the code that updates clientXLogPos from keepalive messages (ServerWALEnd), and
only update it from decoded XLogData messages at lines 182-185. Additionally,
the shared siphonSlot constant creates cross-job interference across multiple
CDC and incremental jobs. Replace the single siphonSlot constant with logic that
generates a unique slot identifier per job, allocate it before capturing the
start position, and ensure each CDC/incremental job uses its own isolated slot
instead of the shared siphon_logical slot.
In `@internal/driver/postgres/incremental_change.go`:
- Around line 85-86: The SQL query in the SweepOrphanSlots function currently
matches all inactive siphon_* slots regardless of their type, but it should only
match physical slots to avoid accidentally dropping logical slots and losing
replication resume state. Add an explicit filter `AND slot_type = 'physical'` to
the WHERE clause of the SELECT query that filters pg_replication_slots to
restrict the sweep to physical slots only.
In `@internal/driver/postgres/inspect_schema.go`:
- Around line 45-49: The mapPGType function is silently returning CTText as a
fallback for unsupported Postgres types instead of explicitly failing. Modify
the mapPGType function to return an error (or use a result type) when it
encounters an unknown data type rather than defaulting to CTText. Update the
code that calls mapPGType (both at line 45 where canonical.CanonicalColumn is
created and at the other location mentioned at lines 117-139) to handle and
propagate this error case explicitly instead of allowing silent type coercion,
ensuring that unknown types cause the schema inspection to fail fast with a
clear error message about the unsupported type.
---
Nitpick comments:
In `@internal/driver/_mysqlcommon/changestream_test.go`:
- Around line 11-24: Add regression test cases to the tests slice in
TestParseColAssign to cover the edge cases mentioned: add a test case where the
assigned value is the string literal 'NULL' (not the actual NULL keyword) to
distinguish from the existing `@4`=NULL case, add a test case for a quoted string
value that contains a forward slash and asterisk (like 'value/*comment') to
ensure the parser correctly handles quoted values with comment-like syntax
inside them, and consider adding test cases for non-DML markers following row
assignments. These cases should be appended to the existing tests slice in the
struct initialization to exercise the CDC data-loss and corruption edge cases.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 8609c0e2-f40f-402d-9d90-3fd022eb6e76
⛔ Files ignored due to path filters (1)
go.sumis excluded by!**/*.sum
📒 Files selected for processing (44)
CHANGELOG.mdREADME.mddocs/CDC.mddocs/CROSS_ENGINE.mddocs/INCREMENTAL.mdgo.modinternal/app/backup.gointernal/app/backup_test.gointernal/app/canonical.gointernal/app/canonical_consume.gointernal/app/canonical_emit.gointernal/app/cdc.gointernal/app/cdc_integration_test.gointernal/app/cross_engine_integration_test.gointernal/app/restore.gointernal/app/restore_chain_test.gointernal/app/sync.gointernal/canonical/canonical.gointernal/canonical/canonical_test.gointernal/cli/backup.gointernal/cli/cdc.gointernal/cli/root.gointernal/cli/sync.gointernal/driver/_mysqlcommon/canonical.gointernal/driver/_mysqlcommon/canonical_test.gointernal/driver/_mysqlcommon/changestream.gointernal/driver/_mysqlcommon/changestream_test.gointernal/driver/_mysqlcommon/conn.gointernal/driver/_mysqlcommon/incremental.gointernal/driver/_mysqlcommon/inspect_schema.gointernal/driver/_mysqlcommon/inspect_schema_test.gointernal/driver/driver.gointernal/driver/mariadb/driver.gointernal/driver/mysql/driver.gointernal/driver/postgres/canonical.gointernal/driver/postgres/changestream.gointernal/driver/postgres/changestream_integration_test.gointernal/driver/postgres/driver.gointernal/driver/postgres/incremental.gointernal/driver/postgres/incremental_change.gointernal/driver/postgres/incremental_integration_test.gointernal/driver/postgres/inspect_schema.gointernal/driver/postgres/inspect_schema_test.gointernal/errs/errs.go
💤 Files with no reviewable changes (4)
- internal/app/canonical_consume.go
- internal/app/canonical_emit.go
- internal/app/canonical.go
- internal/driver/postgres/incremental.go
- Make CurrentPosition create the pgoutput logical slot if absent and anchor to its consistent point, so WAL from the base forward is retained. A logical slot only decodes WAL produced after its own creation, so creating it lazily at stream time meant the first incremental and the bounded change stream captured zero changes. - Add establishLogicalSlot helper; ensureLogicalSlot now returns the consistent point on fresh creation, "" when the slot already exists. - Fix the two Postgres integration tests to take the start/base position via CurrentPosition before the DML, matching the real ordering. - Give the cross-engine MySQL container a 180s startup wait so a slow concurrent boot is tolerated rather than failing the run.
- 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.
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
internal/app/cdc.go (1)
184-192: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winTreat context cancellation during the initial snapshot as a clean stop.
If
ctxis cancelled whilecdcSnapshotis still running (e.g. user aborts during a long first-run snapshot),snapErrwrapscontext.Canceledand is returned as a hard error. That contradicts the documented contract that ctx cancellation is the normal clean stop signal (handled only afterStreamChangesat Line 230). Guard the snapshot error path the same way.🛠️ Proposed guard
if snapErr := cdcSnapshot(ctx, srcConn, dstXfer, src.Driver, opt.Tables); snapErr != nil { + if ctx.Err() != nil { + emit(jobs.Event{Phase: jobs.PhaseProgress, Message: "CDC stopped"}) + return nil + } return snapErr }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@internal/app/cdc.go` around lines 184 - 192, The error handling after the cdcSnapshot function call treats context cancellation as a hard error that is immediately returned, but context cancellation should instead be treated as a clean stop signal consistent with how it is handled elsewhere in the CDC flow (around Line 230 with StreamChanges). Guard the snapErr check by inspecting whether the error wraps context.Canceled using errors.Is, and if so, break or exit the initialization block cleanly without returning the error as a failure. Only return snapErr as a hard error if it represents an actual failure condition unrelated to context cancellation.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@internal/app/cdc.go`:
- Around line 184-192: The error handling after the cdcSnapshot function call
treats context cancellation as a hard error that is immediately returned, but
context cancellation should instead be treated as a clean stop signal consistent
with how it is handled elsewhere in the CDC flow (around Line 230 with
StreamChanges). Guard the snapErr check by inspecting whether the error wraps
context.Canceled using errors.Is, and if so, break or exit the initialization
block cleanly without returning the error as a failure. Only return snapErr as a
hard error if it represents an actual failure condition unrelated to context
cancellation.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 6d224921-4764-4854-b4e7-32c191d00f2f
📒 Files selected for processing (19)
CHANGELOG.mddocs/CDC.mdinternal/app/backup.gointernal/app/cdc.gointernal/app/cross_engine_integration_test.gointernal/app/restore.gointernal/app/sync.gointernal/canonical/canonical.gointernal/cli/cdc.gointernal/driver/_mysqlcommon/changestream.gointernal/driver/_mysqlcommon/changestream_test.gointernal/driver/_mysqlcommon/inspect_schema.gointernal/driver/_mysqlcommon/inspect_schema_test.gointernal/driver/postgres/changestream.gointernal/driver/postgres/changestream_integration_test.gointernal/driver/postgres/incremental_change.gointernal/driver/postgres/incremental_integration_test.gointernal/driver/postgres/inspect_schema.gointernal/driver/postgres/inspect_schema_test.go
✅ Files skipped from review due to trivial changes (3)
- internal/driver/postgres/inspect_schema.go
- docs/CDC.md
- CHANGELOG.md
🚧 Files skipped from review as they are similar to previous changes (11)
- internal/driver/_mysqlcommon/changestream_test.go
- internal/driver/postgres/incremental_change.go
- internal/cli/cdc.go
- internal/app/sync.go
- internal/driver/postgres/changestream_integration_test.go
- internal/app/cross_engine_integration_test.go
- internal/driver/postgres/incremental_integration_test.go
- internal/canonical/canonical.go
- internal/app/backup.go
- internal/driver/postgres/changestream.go
- internal/driver/_mysqlcommon/changestream.go
- Flush a completed row event at each "# at" marker instead of waiting for the next ### op-line or EOF. In an unbounded CDC stream the next op-line may never arrive during a quiet period, so the last change would otherwise stay buffered and never reach the target. - The marker is the existing event boundary already used for the bounded stopAt check; flushing there is safe and does not double-emit (flush nils the pending event). - Add a regression test asserting events flush at the following marker.
This was referenced Jun 23, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Completes Phase F by wiring the three advanced-transfer modes that previously shipped as gated machinery (Phase F was 🟡 Partial). All three now work end-to-end and are validated by integration tests in CI:
backup --incremental --base <id>) — captures a bounded change set since the base dump's recorded end position, serialized as engine-neutral JSONLCanonicalChangerecords.restorereplays incremental links viaApplyChange(base links restore natively).sync --cross-engine, e.g. Postgres → MySQL) — typed schema introspection viaSchemaInspectorbuilds aCanonicalSchema; the source emits canonical JSONL rows, the target re-creates tables (with primary keys) and inserts them, translating types viaMapToNative.siphon cdc <from> <to>/sync --continuous) — unboundedChangeStreamer.StreamChanges→ApplyChange, with snapshot→stream handoff on first run and resume-from-saved-position on restart. Same-engine and cross-engine. Replaces the polling scaffold.Architecture
Everything converges on one engine-neutral type,
canonical.CanonicalChange{Op, Table, Key, Values}. Incremental backup is bounded change capture; CDC is unbounded; both decode through the same loop. Cross-engine is engine-neutral by construction. Incremental restore is replay, notpg_restore.DB access is encapsulated behind a new
CanonicalTransferdriver interface (emit/consume/apply), so*sql.DBnever escapes the driver layer. Five optional, type-asserted driver interfaces were added (coredriver.Conncontract unchanged):SchemaInspector,CanonicalTransfer,ChangeStreamer,IncrementalBackuper,BasePositioner. A new stdlib-onlyinternal/canonicalleaf holds the shared types and pure SQL builders, satisfying thedriver !import appdepguard constraint.New dependency:
github.com/jackc/pglogrepl(Postgres logical decoding / pgoutput).Safety properties
ApplyChangeis idempotent (ON CONFLICT DO NOTHING/INSERT IGNORE, UPDATE/DELETE by PK).BasePositioner.CurrentPosition) so the first incremental against a normal full base doesn't silently drop changes.ctx.Canceled).Testing
make test— 15 packages, all pass;make lint— 0 issues.//go:build integration) compile locally; the live DB paths (pglogrepl decoding, MySQL binlog streaming, cross-engine round-trip, CDC resume) run in CI against real Postgres/MySQL/MariaDB, which is their first runtime execution.Docs
README Phase F row 🟡 → ✅ Complete; CHANGELOG updated;
docs/INCREMENTAL.md,docs/CROSS_ENGINE.md,docs/CDC.mdreconciled to the honest as-built behavior.Known follow-ups (tracked, non-blocking)
'NULL'-string vs SQL NULL aliasing in canonical decode.🤖 Generated with Claude Code
Summary by CodeRabbit
backup --incremental --base <id>produces bounded change dumps and supports incremental restore replay.siphon cdc <from> <to>andsiphon sync --continuouscontinuously stream and apply changes with resumable checkpoints.