From 08e917b42b58cb5973e927285e8df5aeada1ec78 Mon Sep 17 00:00:00 2001
From: mihir-datazip <mihir@datazip.io>
Date: Mon, 10 Aug 2026 15:08:33 +0530
Subject: [PATCH] fix(postgres): retry slot advance until confirmed to prevent
 first-sync lsn mismatch

pg_replication_slot_advance silently clamps its target to the flushed WAL
position, so under concurrent write load the slot stopped short of the
pg_current_wal_lsn() bookmark PreCDC had just written to state, and the next
validation failed with "lsn mismatch, please proceed with clear destination"
on a healthy pipeline. AdvanceLSN now reads the end_lsn the advance returns
and re-advances until the slot confirms the target; PreCDC seeds state only
after that confirmation.
---
 drivers/postgres/internal/cdc.go |  6 ++++--
 pkg/waljs/replicator.go          | 33 ++++++++++++++++++++++++++------
 2 files changed, 31 insertions(+), 8 deletions(-)

diff --git a/drivers/postgres/internal/cdc.go b/drivers/postgres/internal/cdc.go
index 7b4bfb961..97c5f5eb8 100644
--- a/drivers/postgres/internal/cdc.go
+++ b/drivers/postgres/internal/cdc.go
@@ -49,11 +49,13 @@ func (p *Postgres) PreCDC(ctx context.Context, streams []types.StreamInterface)
 
 	globalState := p.state.GetGlobal()
 	if globalState == nil || globalState.State == nil {
-		p.state.SetGlobal(waljs.WALState{LSN: slot.CurrentLSN.String()})
-		p.state.ResetStreams()
+		// advance before seeding state: AdvanceLSN returns only once the slot has confirmed
+		// the target, so the state file and confirmed_flush_lsn agree by construction
 		if err := waljs.AdvanceLSN(ctx, p.client, p.cdcConfig.ReplicationSlot, slot.CurrentLSN.String()); err != nil {
 			return err
 		}
+		p.state.SetGlobal(waljs.WALState{LSN: slot.CurrentLSN.String()})
+		p.state.ResetStreams()
 	}
 	p.streams = streams
 	return nil
diff --git a/pkg/waljs/replicator.go b/pkg/waljs/replicator.go
index a1e12bf3d..f530bcbf2 100644
--- a/pkg/waljs/replicator.go
+++ b/pkg/waljs/replicator.go
@@ -130,14 +130,35 @@ func NewReplicator(ctx context.Context, config *Config, slot ReplicationSlot, re
 	}
 }
 
-// advanceLSN advances the logical replication position to the current WAL position.
+// AdvanceLSN moves the slot to currentWalPos and retries until postgres confirms it there:
+// pg_replication_slot_advance silently clamps its target to the flushed WAL position.
 func AdvanceLSN(ctx context.Context, db *sqlx.DB, slot, currentWalPos string) error {
-	// Get replication slot position
-	if _, err := db.ExecContext(ctx, fmt.Sprintf(AdvanceLSNTemplate, slot, currentWalPos)); err != nil {
-		return fmt.Errorf("failed to advance replication slot: %s", err)
+	target, err := pglogrepl.ParseLSN(currentWalPos)
+	if err != nil {
+		return fmt.Errorf("failed to parse advance target lsn[%s]: %s", currentWalPos, err)
+	}
+
+	timeoutCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
+	defer cancel()
+	for {
+		var slotName string
+		var endLSN pglogrepl.LSN
+		if err := db.QueryRowContext(timeoutCtx, fmt.Sprintf(AdvanceLSNTemplate, slot, currentWalPos)).Scan(&slotName, &endLSN); err != nil {
+			return fmt.Errorf("failed to advance replication slot: %s", err)
+		}
+		if endLSN >= target {
+			logger.Debugf("advanced LSN to %s", currentWalPos)
+			return nil
+		}
+		// end_lsn short of the request means the advance was clamped to the flushed position;
+		// wait for the walwriter to catch up (bounded by ~3x wal_writer_delay) and re-advance
+		logger.Debugf("slot advance clamped at %s (target %s), retrying", endLSN, target)
+		select {
+		case <-timeoutCtx.Done():
+			return fmt.Errorf("slot advance stopped at %s and could not reach target %s: %s", endLSN, target, timeoutCtx.Err())
+		case <-time.After(100 * time.Millisecond):
+		}
 	}
-	logger.Debugf("advanced LSN to %s", currentWalPos)
-	return nil
 }
 
 // Confirm that Logs has been recorded