cdc: massively speed replay with ordered concurrency and batching

.github/workflows/ci.yml

@@ -32,9 +32,29 @@ jobs:
       # integration-tagged testcontainers graph.
       - name: govulncheck
         run: go run golang.org/x/vuln/cmd/govulncheck@latest ./cmd/pgmigrate ./internal/...
-      # Compiles the integration-tagged files without running them; those and the
-      # Compose end-to-end suites need PostgreSQL and are run by hand.
+      # Fast tests do not start PostgreSQL; the focused replay integration and
+      # full Compose migration run in the e2e job below.
       - run: make test
       # Apply, the copy workers, and the CDC handoff are concurrent, and a race
       # between them is not something anyone reproduces by hand twice.
       - run: make race
+
+  e2e:
+    name: PostgreSQL migration e2e
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    steps:
+      - uses: actions/checkout@v7
+      - uses: actions/setup-go@v7
+        with:
+          go-version-file: go.mod
+      - name: Test concurrent, durable, and batched replay
+        run: go test -race -tags=integration ./internal/cdc -run '^TestPG17(ConcurrentReplayRunsIndependentTablesTogetherAndOrdersEachTable|ConcurrentReplayRecoversAnOutOfOrderDurableCommit|ReplayBatchCollapsesSerializedCommitLane|ReplayScalesAcrossIndependentCommitLanes)$' -count=1 -timeout=3m
+      - name: Run a live migration with concurrent replay
+        run: make e2e
+      - name: Print container logs on failure
+        if: failure()
+        run: docker compose -f test/e2e/compose.yaml logs --no-color
+      - name: Stop the test bed
+        if: always()
+        run: test/e2e/scripts/stop.sh

README.md

@@ -48,11 +48,14 @@ Change data capture uses `pgoutput`, the logical decoding plugin built into
 PostgreSQL, so there is no extension to install on the source. Decoded
 transactions are written to append-only checksummed segment files under the
 migration directory and fsynced at each transaction boundary; a transaction over
-256 MiB spills to temporary files beneath `cdc/spill`. Apply is serial, and
-target DML commits in the same transaction as
-`pgmigrate_internal.replication_progress` on the target, which is the
-authoritative apply position. Finalized segments that have been applied are
-pruned every `--segment-prune-interval`, retaining one safety segment.
+256 MiB spills to temporary files beneath `cdc/spill`. Transactions that touch
+different tables replay concurrently while each table retains source order.
+Contiguous transactions already serialized by the same tables share a bounded
+target commit. Each parallel DML commit atomically records a durable target
+receipt; `pgmigrate_internal.replication_progress` advances only across the
+contiguous receipt prefix and remains the authoritative apply position.
+Finalized segments that have been checkpointed are pruned every
+`--segment-prune-interval`, retaining one safety segment.
 
 `pgmigrate cutover` then emits a logical boundary message, drains exactly through
 it, advances target sequences with headroom, reverts what the migration changed on
@@ -88,7 +91,8 @@ There are six commands:
 
 Every command takes `--dir`. All but `status` also need source and target
 connection strings. `pgmigrate <command> --help` prints the defaults as resolved
-on the host, which for `--workers` and `--restore-jobs` depend on its CPU count.
+on the host, which for `--workers`, `--replay-workers`, `--replay-window`, and
+`--restore-jobs` depend on its CPU count.
 
 ## Example
 
@@ -111,7 +115,7 @@ With those understood, start the migration. It keeps running after the base copy
 finishes, following changes until you cut over:
 
 ```bash
-$ pgmigrate run --dir ./migration --ack-warnings --workers 8 --restore-jobs 4 --metrics :9187
+$ pgmigrate run --dir ./migration --ack-warnings --workers 8 --replay-workers 8 --restore-jobs 4 --metrics :9187
 ```
 
 `run` writes almost nothing to the terminal. Phase, progress, health, and error
@@ -278,6 +282,9 @@ directory's writer lock.
 | `--ack-warnings` | false | accept every current preflight warning, including consenting to `REPLICA IDENTITY FULL` where it is needed |
 | `--allow-collation-change` | false | proceed to a target that collates text differently from the source |
 | `--workers <n>` | host CPU count | parallel copy and index-build workers, and the cap on parts per table |
+| `--replay-workers <n>` | host CPU count clamped to 8–32 | target sessions that replay independent tables concurrently. Transactions touching the same table wait for their predecessor; independent durable commits may finish out of order while authoritative progress advances only through their contiguous source-order prefix |
+| `--replay-batch-size <n>` | `128` | maximum contiguous dependent source transactions combined into one durable target transaction. Independent table lanes are never combined, and encoded batch data is capped at 16 MiB |
+| `--replay-window <n>` | 8 times `--replay-workers` | source transactions searched for independent table work; also bounds scheduler memory |
 | `--split-threshold <bytes>` | `1073741824` (1 GiB) | desired bytes per copy part. A table is split into at most `--workers` parts, so a table far larger than the threshold produces larger parts |
 | `--restore-jobs <n>` | half the host CPU count, at least 1 | parallel `pg_restore` jobs for the schema restore |
 | `--pg-dump <path>` | found on `PATH` | `pg_dump` executable |
@@ -454,13 +461,13 @@ about what a partial part left behind.
 ### Apply progress lives on the target, not beside the tool
 
 `pgmigrate_internal.replication_progress` on the target is the authoritative
-apply position, and it commits in the same transaction as the DML it describes.
-The local SQLite database is a low-rate control plane whose apply LSN is
-display-only. A position recorded anywhere but next to the rows can disagree
-with them after a crash, and then replay either loses transactions or repeats
-them. A source-and-filter-derived stream generation binds copied data to that
-progress, and a resume refuses progress that is missing or belongs to another
-stream.
+apply position. Serial replay commits it with DML. Parallel replay commits a
+durable receipt with each independent DML transaction, then atomically advances
+progress and removes only the contiguous receipt prefix. A crash before that
+checkpoint leaves receipts that make already-committed DML unambiguous on
+resume. The local SQLite database is a low-rate control plane whose apply LSN is
+display-only. A source-and-filter-derived stream generation binds copied data,
+receipts, and progress; a resume refuses missing or mismatched durable identity.
 
 Every connection that reads or executes a catalog definition pins `search_path`
 to the empty path, so definitions are fully qualified and mean the same thing on
@@ -684,9 +691,11 @@ Re-run `pgmigrate run` with the same DSNs, filter, and directory.
 
 - A torn `.partial` CDC tail is scanned and truncated to the last valid frame.
   Receiving resumes from the latest fsynced transaction EndLSN.
-- Target DML and authoritative progress commit atomically, so a reconnect or
-  restart skips transactions already recorded on the target. Missing or
-  mismatched stream generation or progress is fatal once copied data exists.
+- Serial target DML and progress commit atomically. Parallel DML commits with a
+  durable receipt, and checkpoint progress advances while deleting the
+  contiguous receipt prefix in one target transaction. A restart therefore
+  skips every committed transaction without guessing. Missing or mismatched
+  stream generation or progress is fatal once progress has started.
 - Restarts from `indexes`, `catchup`, or `follow` retain the completed base copy
   and recover staged CDC.
 - Restarts from `setup`, `schema`, or `copy` deliberately discard **all**
@@ -783,7 +792,10 @@ schema.
   findings and need an operator plan.
 - The target is assumed not to receive independent application traffic before
   cutover. Replay divergence stops the run.
-- Apply is serial.
+- Replay parallelism comes from transactions that touch independent tables.
+  A workload whose every transaction touches the same table remains serial by
+  design so that updates and deletes observe source order, but bounded batches
+  amortize its target commit cost.
 - The delivered e2e bed is PostgreSQL 17 to 17. Cross-major compatibility has
   focused integration probes but no full cross-major Compose migration.
 - Verification samples, and reports 64-bit server-side hashes rather than a

internal/app/app.go

@@ -1361,6 +1361,7 @@ func runApplierToFollow(
 	}
 	applier, err := cdc.NewApplier(cdc.ApplierConfig{
 		ConnString: cfg.Target, Directory: filepath.Join(cfg.Dir, "cdc"),
+		Workers: cfg.ReplayWorkers, BatchSize: cfg.ReplayBatchSize, Window: cfg.ReplayWindow,
 		StreamID: snapshot.Slot, StreamGeneration: streamGeneration(
 			migration.SourceFingerprint, migration.FilterFingerprint,
 		), TargetHasCopiedData: true, Durable: durable, EndPosition: endPosition(store),