fix(postgres): prevent first-sync lsn mismatch under concurrent write load

[mihir-datazip]

Description

On a pipeline's first CDC sync, PreCDC bookmarks the resume position by reading pg_current_wal_lsn() into the state file and advancing the replication slot to the same value. Two PostgreSQL behaviours break the intended state == confirmed_flush_lsn invariant:

1. pg_current_wal_lsn() returns the WAL write pointer, which runs ahead of the fsynced flush pointer whenever WAL is written faster than it is flushed (bulk statements, async commits).
2. pg_replication_slot_advance() silently clamps its target to the flushed position, and reports where the slot really stopped only in its end_lsn result column — which AdvanceLSN discarded (db.ExecContext, result ignored).

So under concurrent write load the state file records the write pointer while the slot stops at the flush pointer. validateGlobalState demands strict equality, so the sync dies at its CDC phase — and every retry after it — with:

FATAL ... failed with non retryable error: lsn mismatch, please proceed with clear destination.
lsn saved in state [13/4964E000] current lsn [13/495EFFD8]

…telling the user to wipe the destination and re-backfill a perfectly healthy pipeline. Only the bootstrap path is affected: AcknowledgeLSN already polls the slot until it confirms, so post-bootstrap syncs are safe.

When it appears — measured (stock postgres:15, wal_buffers = 4 MB default, local NVMe)

Sync-commit workloads — the trigger is any single statement/transaction writing more WAL than wal_buffers:

Concurrent load beside the first sync	Result
Small synchronous commits, any rate	Never — 0 of 150,000 sampled instants had the write pointer ahead of flush (the commit path publishes both pointers together)
Bulk statements below wal_buffers (0.5–2 MB WAL/stmt, ~100–170k rows/s)	Never — 0 open-window reads across 532 bootstrap attempts
Bulk statements at/above wal_buffers (4 MB WAL/stmt, ~182k rows/s)	Reproduced on attempt 7 — state 385,064 bytes ahead of the slot

Per-bootstrap hit probability measured ~3.5% beside 30 MB statements — consistent with the field observation of a 432-byte mismatch roughly once per ~60 first syncs on a near-idle test database.

Async-commit workloads (synchronous_commit = off, a documented and common knob for ingest/ETL pipelines) — commits stop closing the window, so any rate is exposed:

Concurrent load beside the first sync	Result
100 KB statements, lowest ladder level	Reproduced on attempt 8, in under 0.1 s — state 111,048 bytes ahead of the slot

Local NVMe is the conservative case: slower (cloud) storage widens the write-vs-flush window.

The fix

• waljs.AdvanceLSN now reads the end_lsn the advance returns. When it lands short of the request the advance was clamped, so it waits 100 ms (the walwriter closes the gap within ~3× wal_writer_delay) and re-advances until the slot confirms the target. A 30 s cap turns a pathological stall into a clear, retryable error instead of a bricked pipeline. Function signature unchanged.
• PreCDC seeds the state after the slot has confirmed, so state == confirmed_flush_lsn holds by construction — and a failed advance no longer leaves a seeded state behind (the write-ordering gap fix(postgres): reverse write ordering in PreCDC/PostCDC to prevent LSN mismatch #854 flagged on this path).
• Composes with the read-replica slot query (fix(postgres): pick the slot LSN source by recovery state so CDC works on a read-replica #1070): on a standby the target is the replay LSN and the same retry converges against replay progress.

Type of change

• Bug fix (non-breaking change which fixes an issue)

How Has This Been Tested?

Verified with a two-test repro harness, kept out of this PR's diff — full source in the collapsed block at the bottom of this section (drop cdc_bootstrap_repro_test.go into drivers/postgres/internal/ and run the command below; it skips without OLAKE_PG_REPRO_DSN). Both tests replay the exact PreCDC bootstrap plus the next sync's validation against a live postgres while a load generator runs, ramping load level by level and failing with full forensics on the first divergence:

// first sync bootstrap, verbatim from PreCDC
slot, _ := waljs.GetSlotPosition(ctx, db, reproSlotName)
state := waljs.WALState{LSN: slot.CurrentLSN.String()}          // what PreCDC bookmarks
_ = waljs.AdvanceLSN(ctx, db, reproSlotName, slot.CurrentLSN.String())

// next sync: StreamChanges re-reads the slot and validates
after, _ := waljs.GetSlotPosition(ctx, db, reproSlotName)
if verr := validateGlobalState(state, after.LSN); verr != nil {
    // REPRODUCED — reports both LSNs, the gap, and the end_lsn the old code discarded
}

• TestReproPreCDCBootstrapLSNDivergence — default synchronous_commit, ramps bulk-statement size 0.5 → 32 MB WAL/stmt across the wal_buffers boundary.
• TestReproPreCDCBootstrapLSNDivergenceAsyncCommit — load session under synchronous_commit = off, ramps small-statement rate from 5 stmt/s to unthrottled.

make olake.postgres.refresh
OLAKE_PG_REPRO_DSN='postgres://postgres:secret1234@localhost:5433/postgres?sslmode=disable' \
  go -C drivers/postgres test ./internal/ -run TestReproPreCDCBootstrapLSNDivergence -v -count=1

• Before the fix: both ladders reproduce (sync at the 4 MB rung, async at the lowest rung)
• After the fix: every level of both ladders holds the invariant — including levels where the window was demonstrably open ("saw write ahead of flush" up to 107 of 156 reads), i.e. the bootstrap now rides through the window instead of depending on a quiet instant
• gofmt / go vet clean on pkg/waljs and drivers/postgres

Before the fix — both ladders reproduce

=== RUN   TestReproPreCDCBootstrapLSNDivergence
    cdc_bootstrap_repro_test.go:145: sync-commit level 500 rows/stmt (~0.5 MB WAL/stmt, ~96408 rows/s achieved): 218 bootstrap attempts, 0 saw write ahead of flush — invariant held
    cdc_bootstrap_repro_test.go:145: sync-commit level 1000 rows/stmt (~1.0 MB WAL/stmt, ~126207 rows/s achieved): 218 bootstrap attempts, 0 saw write ahead of flush — invariant held
    cdc_bootstrap_repro_test.go:145: sync-commit level 2000 rows/stmt (~2.0 MB WAL/stmt, ~168014 rows/s achieved): 96 bootstrap attempts, 0 saw write ahead of flush — invariant held
    cdc_bootstrap_repro_test.go:144: REPRODUCED at sync-commit level 4000 rows/stmt (~4.0 MB WAL/stmt, ~182348 rows/s achieved)
          attempt 7 of this level; 2 of 7 reads saw write ahead of flush
          wal pointers at bootstrap: write=13/4964E000 flush=13/495EFFD8
          state seeded from pg_current_wal_lsn: 13/4964E000
          slot confirmed_flush after advance:   13/495EFFD8 (385064 bytes short)
          next sync fails with: failed with non retryable error: lsn mismatch, please proceed with clear destination. lsn saved in state [13/4964E000] current lsn [13/495EFFD8]
          follow-up advance requested=13/4BD3E078 returned end_lsn=13/4BD3E078, slot confirmed_flush=13/4BD3E078 (end_lsn is the correct seed)
--- FAIL: TestReproPreCDCBootstrapLSNDivergence (63.93s)
=== RUN   TestReproPreCDCBootstrapLSNDivergenceAsyncCommit
    cdc_bootstrap_repro_test.go:209: REPRODUCED at async-commit level target 5 stmt/s × 100 rows/stmt (achieved ~80 stmt/s ≈ 8.0 MB/s WAL)
          attempt 8 of this level; 1 of 8 reads saw write ahead of flush
          wal pointers at bootstrap: write=13/4E4D2000 flush=13/4E4B6E38
          state seeded from pg_current_wal_lsn: 13/4E4D2000
          slot confirmed_flush after advance:   13/4E4B6E38 (111048 bytes short)
          next sync fails with: failed with non retryable error: lsn mismatch, please proceed with clear destination. lsn saved in state [13/4E4D2000] current lsn [13/4E4B6E38]
          follow-up advance requested=13/4E4D2000 returned end_lsn=13/4E4B6E38, slot confirmed_flush=13/4E4B6E38 (end_lsn is the correct seed)
--- FAIL: TestReproPreCDCBootstrapLSNDivergenceAsyncCommit (0.06s)
FAIL
FAIL    github.com/datazip-inc/olake/drivers/postgres/internal  64.727s

After the fix — invariant held at every level of both ladders

=== RUN   TestReproPreCDCBootstrapLSNDivergence
    cdc_bootstrap_repro_test.go:145: sync-commit level 500 rows/stmt (~0.5 MB WAL/stmt, ~103042 rows/s achieved): 238 bootstrap attempts, 0 saw write ahead of flush — invariant held
    cdc_bootstrap_repro_test.go:145: sync-commit level 1000 rows/stmt (~1.0 MB WAL/stmt, ~125900 rows/s achieved): 82 bootstrap attempts, 0 saw write ahead of flush — invariant held
    cdc_bootstrap_repro_test.go:145: sync-commit level 2000 rows/stmt (~2.0 MB WAL/stmt, ~173602 rows/s achieved): 82 bootstrap attempts, 0 saw write ahead of flush — invariant held
    cdc_bootstrap_repro_test.go:145: sync-commit level 4000 rows/stmt (~4.0 MB WAL/stmt, ~200698 rows/s achieved): 75 bootstrap attempts, 19 saw write ahead of flush — invariant held
    cdc_bootstrap_repro_test.go:145: sync-commit level 8000 rows/stmt (~8.0 MB WAL/stmt, ~212806 rows/s achieved): 55 bootstrap attempts, 23 saw write ahead of flush — invariant held
    cdc_bootstrap_repro_test.go:145: sync-commit level 16000 rows/stmt (~16.0 MB WAL/stmt, ~248634 rows/s achieved): 31 bootstrap attempts, 14 saw write ahead of flush — invariant held
    cdc_bootstrap_repro_test.go:145: sync-commit level 32000 rows/stmt (~32.0 MB WAL/stmt, ~175899 rows/s achieved): 20 bootstrap attempts, 13 saw write ahead of flush — invariant held
    cdc_bootstrap_repro_test.go:147: no sync-commit load level reproduced the divergence; invariant held at every level
--- PASS: TestReproPreCDCBootstrapLSNDivergence (142.10s)
=== RUN   TestReproPreCDCBootstrapLSNDivergenceAsyncCommit
    cdc_bootstrap_repro_test.go:210: async-commit level target 5 stmt/s × 100 rows/stmt (achieved ~5 stmt/s ≈ 0.5 MB/s WAL): 183 bootstrap attempts, 46 saw write ahead of flush — invariant held
    cdc_bootstrap_repro_test.go:210: async-commit level target 10 stmt/s × 100 rows/stmt (achieved ~10 stmt/s ≈ 1.0 MB/s WAL): 175 bootstrap attempts, 79 saw write ahead of flush — invariant held
    cdc_bootstrap_repro_test.go:210: async-commit level target 25 stmt/s × 100 rows/stmt (achieved ~23 stmt/s ≈ 2.3 MB/s WAL): 97 bootstrap attempts, 90 saw write ahead of flush — invariant held
    cdc_bootstrap_repro_test.go:210: async-commit level target 50 stmt/s × 100 rows/stmt (achieved ~44 stmt/s ≈ 4.4 MB/s WAL): 101 bootstrap attempts, 56 saw write ahead of flush — invariant held
    cdc_bootstrap_repro_test.go:210: async-commit level target 100 stmt/s × 100 rows/stmt (achieved ~84 stmt/s ≈ 8.4 MB/s WAL): 131 bootstrap attempts, 73 saw write ahead of flush — invariant held
    cdc_bootstrap_repro_test.go:210: async-commit level target 250 stmt/s × 100 rows/stmt (achieved ~174 stmt/s ≈ 17.4 MB/s WAL): 156 bootstrap attempts, 107 saw write ahead of flush — invariant held
    cdc_bootstrap_repro_test.go:210: async-commit level target max (unthrottled) × 100 rows/stmt (achieved ~801 stmt/s ≈ 80.1 MB/s WAL): 44 bootstrap attempts, 30 saw write ahead of flush — invariant held
    cdc_bootstrap_repro_test.go:212: no async-commit rate reproduced the divergence; invariant held at every level
--- PASS: TestReproPreCDCBootstrapLSNDivergenceAsyncCommit (142.83s)
PASS
ok      github.com/datazip-inc/olake/drivers/postgres/internal  285.418s

Full repro harness — cdc_bootstrap_repro_test.go

package driver

import (
	"context"
	"fmt"
	"os"
	"sync/atomic"
	"testing"
	"time"

	"github.com/datazip-inc/olake/pkg/waljs"
	"github.com/jmoiron/sqlx"
)

// TEMPORARY repro file (delete after fix): PreCDC seeds state from the WAL write pointer
// while pg_replication_slot_advance clamps to flush — both ladder tests hunt the trigger load.
const reproSlotName = "olake_repro_slot"

// setupReproInfra connects, recreates the repro slot and load table, and registers cleanup.
func setupReproInfra(ctx context.Context, t *testing.T) *sqlx.DB {
	t.Helper()
	dsn := os.Getenv("OLAKE_PG_REPRO_DSN")
	if dsn == "" {
		t.Skip("set OLAKE_PG_REPRO_DSN (compose default: postgres://postgres:secret1234@localhost:5433/postgres?sslmode=disable)")
	}

	db, err := sqlx.ConnectContext(ctx, "pgx", dsn)
	if err != nil {
		t.Fatalf("connect: %s", err)
	}
	t.Cleanup(func() { db.Close() })

	_, _ = db.ExecContext(ctx, fmt.Sprintf("SELECT pg_drop_replication_slot('%s')", reproSlotName))
	if _, err := db.ExecContext(ctx, fmt.Sprintf("SELECT pg_create_logical_replication_slot('%s', 'pgoutput')", reproSlotName)); err != nil {
		t.Fatalf("create slot: %s", err)
	}
	t.Cleanup(func() {
		_, _ = db.ExecContext(context.Background(), fmt.Sprintf("SELECT pg_drop_replication_slot('%s')", reproSlotName))
	})

	if _, err := db.ExecContext(ctx, "CREATE TABLE IF NOT EXISTS olake_repro_load (id bigserial PRIMARY KEY, payload text)"); err != nil {
		t.Fatalf("create load table: %s", err)
	}
	t.Cleanup(func() {
		_, _ = db.ExecContext(context.Background(), "DROP TABLE IF EXISTS olake_repro_load")
	})
	return db
}

// runBootstrapAttempts replays the PreCDC bootstrap plus the next sync's validation until
// deadline; on divergence it fails the test with full forensics, headlined by describe().
func runBootstrapAttempts(ctx context.Context, t *testing.T, db *sqlx.DB, deadline time.Time, describe func() string) (attempts, writeAhead int) {
	t.Helper()
	for time.Now().Before(deadline) {
		attempts++

		var writePtr, flushPtr string
		var gapOpen bool
		if err := db.QueryRowContext(ctx, "SELECT w::text, f::text, w > f FROM (SELECT pg_current_wal_lsn() w, pg_current_wal_flush_lsn() f) s").Scan(&writePtr, &flushPtr, &gapOpen); err != nil {
			t.Fatalf("attempt %d: read wal pointers: %s", attempts, err)
		}
		if gapOpen {
			writeAhead++
		}

		// first sync bootstrap, verbatim from PreCDC (cdc.go:52-56)
		slot, err := waljs.GetSlotPosition(ctx, db, reproSlotName)
		if err != nil {
			t.Fatalf("attempt %d: get slot position: %s", attempts, err)
		}
		state := waljs.WALState{LSN: slot.CurrentLSN.String()}
		if err := waljs.AdvanceLSN(ctx, db, reproSlotName, slot.CurrentLSN.String()); err != nil {
			t.Fatalf("attempt %d: advance: %s", attempts, err)
		}

		// next sync: StreamChanges re-reads the slot and validates (cdc.go:133,152)
		after, err := waljs.GetSlotPosition(ctx, db, reproSlotName)
		if err != nil {
			t.Fatalf("attempt %d: re-read slot: %s", attempts, err)
		}
		if verr := validateGlobalState(state, after.LSN); verr != nil {
			// hypothesis part 2: the end_lsn AdvanceLSN discards reports where the slot really stopped
			var name, endLSN, confirmed string
			requested := after.CurrentLSN.String()
			if err := db.QueryRowContext(ctx, fmt.Sprintf(waljs.AdvanceLSNTemplate, reproSlotName, requested)).Scan(&name, &endLSN); err != nil {
				endLSN = "scan failed: " + err.Error()
			}
			if err := db.QueryRowContext(ctx, fmt.Sprintf("SELECT confirmed_flush_lsn FROM pg_replication_slots WHERE slot_name = '%s'", reproSlotName)).Scan(&confirmed); err != nil {
				confirmed = "scan failed: " + err.Error()
			}
			t.Fatalf("REPRODUCED at %s\n"+
				"  attempt %d of this level; %d of %d reads saw write ahead of flush\n"+
				"  wal pointers at bootstrap: write=%s flush=%s\n"+
				"  state seeded from pg_current_wal_lsn: %s\n"+
				"  slot confirmed_flush after advance:   %s (%d bytes short)\n"+
				"  next sync fails with: %s\n"+
				"  follow-up advance requested=%s returned end_lsn=%s, slot confirmed_flush=%s (end_lsn is the correct seed)",
				describe(), attempts, writeAhead, attempts, writePtr, flushPtr,
				state.LSN, after.LSN, uint64(slot.CurrentLSN-after.LSN), verr,
				requested, endLSN, confirmed)
		}
	}
	return attempts, writeAhead
}

// Default synchronous_commit: ramps rows per bulk statement — the window only opens once one
// statement's WAL exceeds wal_buffers (4 MB stock), so the ladder brackets that boundary.
func TestReproPreCDCBootstrapLSNDivergence(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
	defer cancel()
	db := setupReproInfra(ctx, t)

	loadConn, err := db.Connx(ctx)
	if err != nil {
		t.Fatalf("load conn: %s", err)
	}
	var rowsPerStmt, rowsWritten atomic.Int64
	loadCtx, stopLoad := context.WithCancel(ctx)
	loadDone := make(chan struct{})
	go func() {
		defer close(loadDone)
		for i := 0; loadCtx.Err() == nil; i++ {
			if i%3 == 0 {
				_, _ = loadConn.ExecContext(loadCtx, "TRUNCATE olake_repro_load")
			}
			n := rowsPerStmt.Load()
			if _, err := loadConn.ExecContext(loadCtx, fmt.Sprintf("INSERT INTO olake_repro_load (payload) SELECT repeat('x', 1000) FROM generate_series(1, %d)", n)); err == nil {
				rowsWritten.Add(n)
			}
		}
	}()
	defer func() { stopLoad(); <-loadDone; loadConn.Close() }()

	levels := []int64{500, 1000, 2000, 4000, 8000, 16000, 32000}
	rowsPerStmt.Store(levels[0])
	for _, n := range levels {
		rowsPerStmt.Store(n)
		rowsWritten.Store(0)
		levelStart := time.Now()
		describe := func() string {
			rps := float64(rowsWritten.Load()) / time.Since(levelStart).Seconds()
			return fmt.Sprintf("sync-commit level %d rows/stmt (~%.1f MB WAL/stmt, ~%.0f rows/s achieved)", n, float64(n)/1000.0, rps)
		}
		attempts, writeAhead := runBootstrapAttempts(ctx, t, db, levelStart.Add(20*time.Second), describe)
		t.Logf("%s: %d bootstrap attempts, %d saw write ahead of flush — invariant held", describe(), attempts, writeAhead)
	}
	t.Logf("no sync-commit load level reproduced the divergence; invariant held at every level")
}

// Async-commit ingest (SET synchronous_commit = off, a documented knob for bulk/event pipelines):
// ramps the rate of small statements — commits stop closing the window, only the walwriter is left.
func TestReproPreCDCBootstrapLSNDivergenceAsyncCommit(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
	defer cancel()
	db := setupReproInfra(ctx, t)

	loadConn, err := db.Connx(ctx)
	if err != nil {
		t.Fatalf("load conn: %s", err)
	}
	if _, err := loadConn.ExecContext(ctx, "SET synchronous_commit = off"); err != nil {
		t.Fatalf("set async commit: %s", err)
	}

	// ~100 KB per statement, far below wal_buffers on purpose: any window at these levels
	// comes from async commits piling up, not from single-statement buffer pressure
	const rowsEach = 100
	var intervalNS, stmtsDone atomic.Int64
	loadCtx, stopLoad := context.WithCancel(ctx)
	loadDone := make(chan struct{})
	go func() {
		defer close(loadDone)
		for i := 0; loadCtx.Err() == nil; i++ {
			if i%100 == 0 {
				_, _ = loadConn.ExecContext(loadCtx, "TRUNCATE olake_repro_load")
			}
			if _, err := loadConn.ExecContext(loadCtx, fmt.Sprintf("INSERT INTO olake_repro_load (payload) SELECT repeat('x', 1000) FROM generate_series(1, %d)", rowsEach)); err == nil {
				stmtsDone.Add(1)
			}
			if d := intervalNS.Load(); d > 0 {
				select {
				case <-loadCtx.Done():
					return
				case <-time.After(time.Duration(d)):
				}
			}
		}
	}()
	defer func() { stopLoad(); <-loadDone; loadConn.Close() }()

	// target statements/sec; 0 = unthrottled back-to-back
	levels := []int64{5, 10, 25, 50, 100, 250, 0}
	for _, rate := range levels {
		if rate > 0 {
			intervalNS.Store(int64(time.Second) / rate)
		} else {
			intervalNS.Store(0)
		}
		stmtsDone.Store(0)
		levelStart := time.Now()
		target := "max (unthrottled)"
		if rate > 0 {
			target = fmt.Sprintf("%d stmt/s", rate)
		}
		describe := func() string {
			s := float64(stmtsDone.Load()) / time.Since(levelStart).Seconds()
			return fmt.Sprintf("async-commit level target %s × %d rows/stmt (achieved ~%.0f stmt/s ≈ %.1f MB/s WAL)", target, rowsEach, s, s*float64(rowsEach)/1000.0)
		}
		attempts, writeAhead := runBootstrapAttempts(ctx, t, db, levelStart.Add(20*time.Second), describe)
		t.Logf("%s: %d bootstrap attempts, %d saw write ahead of flush — invariant held", describe(), attempts, writeAhead)
	}
	t.Logf("no async-commit rate reproduced the divergence; invariant held at every level")
}

Screenshots or Recordings

N/A — terminal outputs attached above.

Documentation

• N/A (bug fix, refactor, or test changes only)

Related PR's (If Any):

• Bug: PostCDC write ordering causes irrecoverable LSN mismatch between OLake state and PostgreSQL replication slot #853 / fix(postgres): reverse write ordering in PreCDC/PostCDC to prevent LSN mismatch #854 — earlier diagnosis of the same fatal as a crash window between the state write and the slot call; the PostCDC half has since been fixed, and this PR's advance-before-seed ordering closes the PreCDC half as a side effect. The clamp bug fixed here needs no crash at all — it happens on a fully successful advance.
• fix(postgres): pick the slot LSN source by recovery state so CDC works on a read-replica #1070 — read-replica slot query; this fix composes with it (the retry converges against the replay position on a standby).
• question: full load on lsn mismatch or fail sync? #321 — fail-vs-full-load policy discussion for the "lsn mismatch" error; this PR removes the most common spurious trigger for that flow.