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

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

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: %w", 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: %w", 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: %w", 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