fix(postgres): establish logical slot for base

- 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.

Author: nixrajput

internal/app/cross_engine_integration_test.go

@@ -9,8 +9,10 @@ import (
 	"testing"
 	"time"
 
+	tc "github.com/testcontainers/testcontainers-go"
 	mysqlctr "github.com/testcontainers/testcontainers-go/modules/mysql"
 	pgctr "github.com/testcontainers/testcontainers-go/modules/postgres"
+	"github.com/testcontainers/testcontainers-go/wait"
 
 	mysqlcommon "github.com/nixrajput/siphon/internal/driver/_mysqlcommon"
 
@@ -56,10 +58,20 @@ func startIntegPG(t *testing.T) (host string, port int, cleanup func()) {
 func startIntegMySQL(t *testing.T) (host string, port int, cleanup func()) {
 	t.Helper()
 	ctx := context.Background()
+	// Explicit, generous startup wait: this test boots Postgres + MySQL in one
+	// run alongside the per-driver suites, so under concurrent cold-image pulls
+	// MySQL's two-phase init (init → restart → second "ready for connections")
+	// can exceed the module's default 60s timeout. Wait for that second ready
+	// log with a 180s ceiling so a slow-but-healthy boot is tolerated, not failed.
 	c, err := mysqlctr.Run(ctx, "mysql:8.0",
 		mysqlctr.WithDatabase("test"),
 		mysqlctr.WithUsername("root"),
 		mysqlctr.WithPassword("rootpass"),
+		tc.WithWaitStrategy(
+			wait.ForLog("port: 3306  MySQL Community Server").
+				WithOccurrence(1).
+				WithStartupTimeout(180*time.Second),
+		),
 	)
 	if err != nil {
 		t.Fatalf("start mysql container: %v", err)

internal/driver/postgres/changestream.go

@@ -67,7 +67,7 @@ func (c *Conn) streamWithStop(ctx context.Context, from canonical.Position, stop
 	}
 	defer func() { _ = repl.Close(context.Background()) }()
 
-	if err := ensureLogicalSlot(ctx, repl); err != nil {
+	if _, err := ensureLogicalSlot(ctx, repl); err != nil {
 		return canonical.Position{}, err
 	}
 
@@ -375,16 +375,50 @@ func (c *Conn) ensurePublication(ctx context.Context) error {
 // ensureLogicalSlot creates the pgoutput logical slot on the replication
 // connection if absent. CreateReplicationSlot errors with "already exists" when
 // the slot is present; that is tolerated so repeated streams reuse the slot.
-func ensureLogicalSlot(ctx context.Context, repl *pgconn.PgConn) error {
-	_, err := pglogrepl.CreateReplicationSlot(ctx, repl, siphonSlot, outputPlugin,
+//
+// It returns the slot's consistent point — the LSN from which the freshly
+// created slot guarantees every subsequent change is retained and decodable —
+// when it creates the slot, and "" when the slot already existed (the existing
+// slot is already retaining WAL, so the caller has no new anchor to record).
+func ensureLogicalSlot(ctx context.Context, repl *pgconn.PgConn) (string, error) {
+	res, err := pglogrepl.CreateReplicationSlot(ctx, repl, siphonSlot, outputPlugin,
 		pglogrepl.CreateReplicationSlotOptions{Temporary: false})
 	if err != nil {
 		if strings.Contains(err.Error(), "already exists") {
-			return nil
+			return "", nil
 		}
-		return &errs.Error{Op: "postgres.stream", Code: errs.CodeSystem, Cause: err}
+		return "", &errs.Error{Op: "postgres.stream", Code: errs.CodeSystem, Cause: err}
 	}
-	return nil
+	return res.ConsistentPoint, nil
+}
+
+// establishLogicalSlot opens a short-lived replication connection, ensures the
+// publication and logical slot exist, and returns the slot's consistent point
+// (empty when the slot already existed).
+//
+// This is the slot-establishment primitive used to anchor WAL retention BEFORE
+// a workload runs: a logical slot only retains and decodes WAL produced after
+// its own creation, so the slot must exist before the changes a later
+// incremental or CDC run wants to capture. CurrentPosition calls this so a base
+// backup leaves the slot in place, and the changes that follow are retained.
+func (c *Conn) establishLogicalSlot(ctx context.Context) (string, error) {
+	if err := c.requireLogicalWAL(ctx); err != nil {
+		return "", err
+	}
+	if err := c.ensurePublication(ctx); err != nil {
+		return "", err
+	}
+	repl, err := pgconn.Connect(ctx, c.replicationDSN())
+	if err != nil {
+		return "", &errs.Error{
+			Op:    "postgres.stream",
+			Code:  errs.CodeSystem,
+			Cause: errs.ErrConnectionFailed,
+			Hint:  "logical replication connect: " + err.Error(),
+		}
+	}
+	defer func() { _ = repl.Close(context.Background()) }()
+	return ensureLogicalSlot(ctx, repl)
 }
 
 // resolveStartLSN returns the LSN to start replication from. A non-empty

internal/driver/postgres/changestream_integration_test.go

@@ -72,14 +72,12 @@ func TestStreamChanges_Bounded(t *testing.T) {
 
 	conn := &Conn{db: db, p: prof}
 
-	// Capture the start LSN before any DML so the stream sees exactly our changes.
-	var startLSN string
-	if err := db.QueryRowContext(ctx, "SELECT pg_current_wal_lsn()::text").Scan(&startLSN); err != nil {
-		t.Fatalf("capture lsn: %v", err)
-	}
-	// The publication+slot must exist before the DML so the slot retains the WAL.
-	if err := conn.ensurePublication(ctx); err != nil {
-		t.Fatalf("ensure publication: %v", err)
+	// Capture the start position before any DML so the stream sees exactly our
+	// changes. CurrentPosition also establishes the logical slot — a slot only
+	// retains WAL produced after its creation, so it must exist before the DML.
+	startPos, err := conn.CurrentPosition(ctx)
+	if err != nil {
+		t.Fatalf("capture start position: %v", err)
 	}
 
 	// Apply the three operations.
@@ -109,7 +107,7 @@ func TestStreamChanges_Bounded(t *testing.T) {
 		return nil
 	}
 
-	if _, err := conn.StreamChanges(streamCtx, canonical.Position{LSN: startLSN}, emit); err != nil {
+	if _, err := conn.StreamChanges(streamCtx, startPos, emit); err != nil {
 		t.Fatalf("StreamChanges: %v", err)
 	}
 

internal/driver/postgres/incremental_change.go

@@ -14,10 +14,28 @@ import (
 var _ driver.IncrementalBackuper = (*Conn)(nil)
 var _ driver.BasePositioner = (*Conn)(nil)
 
-// CurrentPosition returns the server's current WAL position via
-// pg_current_wal_lsn(). app.Backup calls this right after a full backup so the
-// base dump's Envelope records where the first incremental should resume from.
+// CurrentPosition records the resume anchor for the first incremental after a
+// base backup. app.Backup calls this right after a full backup so the base
+// dump's Envelope carries where the first incremental should resume from.
+//
+// It also ESTABLISHES the logical replication slot if it does not yet exist. A
+// logical slot only retains and decodes WAL produced after the slot's own
+// creation, so the slot must exist from the recorded anchor forward or the
+// first incremental silently captures nothing (it would resume from an LSN the
+// slot never retained). When this call creates the slot, the slot's consistent
+// point is the correct anchor — it is the exact LSN from which the slot
+// guarantees decodable WAL. When the slot already exists it is already
+// retaining WAL, so pg_current_wal_lsn() is a safe anchor.
 func (c *Conn) CurrentPosition(ctx context.Context) (canonical.Position, error) {
+	consistent, err := c.establishLogicalSlot(ctx)
+	if err != nil {
+		return canonical.Position{}, err
+	}
+	if consistent != "" {
+		// Slot freshly created: anchor to its consistent point, the boundary
+		// from which it retains every subsequent change.
+		return canonical.Position{LSN: consistent}, nil
+	}
 	var lsn string
 	if err := c.db.QueryRowContext(ctx, "SELECT pg_current_wal_lsn()::text").Scan(&lsn); err != nil {
 		return canonical.Position{}, &errs.Error{Op: "postgres.current_position", Code: errs.CodeSystem, Cause: err}

internal/driver/postgres/incremental_integration_test.go

@@ -46,15 +46,13 @@ func TestBackupIncremental_BoundedCaptureAndReplay(t *testing.T) {
 
 	conn := &Conn{db: db, p: prof}
 
-	// Publication + slot must exist before the DML so the slot retains the WAL.
-	if err := conn.ensurePublication(ctx); err != nil {
-		t.Fatalf("ensure publication: %v", err)
-	}
-
-	// Capture the base-end LSN: the incremental resumes from exactly here.
-	var baseEnd string
-	if err := db.QueryRowContext(ctx, "SELECT pg_current_wal_lsn()::text").Scan(&baseEnd); err != nil {
-		t.Fatalf("capture base-end lsn: %v", err)
+	// Capture the base-end position: the incremental resumes from exactly here.
+	// CurrentPosition also establishes the logical slot — a slot only retains
+	// WAL produced after its creation, so it must exist before the post-base DML
+	// or the incremental captures nothing.
+	basePos, err := conn.CurrentPosition(ctx)
+	if err != nil {
+		t.Fatalf("capture base-end position: %v", err)
 	}
 
 	// Post-base changes: one insert, one update, one delete.
@@ -72,7 +70,7 @@ func TestBackupIncremental_BoundedCaptureAndReplay(t *testing.T) {
 	capCtx, cancel := context.WithTimeout(ctx, 60*time.Second)
 	defer cancel()
 	var buf bytes.Buffer
-	endPos, err := conn.BackupIncremental(capCtx, canonical.Position{LSN: baseEnd}, &buf)
+	endPos, err := conn.BackupIncremental(capCtx, basePos, &buf)
 	if err != nil {
 		t.Fatalf("BackupIncremental: %v", err)
 	}