From 2cf65203d512c75efb47ecb491a76d7c8d20e8d8 Mon Sep 17 00:00:00 2001 From: Siva Date: Fri, 15 May 2026 23:06:37 +0530 Subject: [PATCH 1/2] test(e2e): add WaitForReplication after first switchover/failover --- e2e/database_test.go | 93 +++++++++++++++++++++++++++++------------- e2e/failover_test.go | 23 +++++++---- e2e/switchover_test.go | 51 +++++++++++++++++++---- 3 files changed, 124 insertions(+), 43 deletions(-) diff --git a/e2e/database_test.go b/e2e/database_test.go index 52f29367..0c8a15dd 100644 --- a/e2e/database_test.go +++ b/e2e/database_test.go @@ -8,10 +8,9 @@ import ( "fmt" "iter" "testing" + "time" "github.com/jackc/pgx/v5" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" controlplane "github.com/pgEdge/control-plane/api/apiv1/gen/control_plane" "github.com/pgEdge/control-plane/client" @@ -356,49 +355,85 @@ func (d *DatabaseFixture) waitForTask(ctx context.Context, task *controlplane.Ta func (f *DatabaseFixture) WaitForReplication(ctx context.Context, t testing.TB, username, password string) { tLog(t, "waiting for replication to catch up on all nodes") - // Execute sync_event on all primary nodes - nodeSyncMap := make(map[string]string) - for _, node := range f.Spec.Nodes { - primaryOpts := ConnectionOptions{ - Matcher: And(WithNode(node.Name), WithRole("primary")), - Username: username, - Password: password, + // After a primary change (switchover/failover), Spock subscriptions need + // time to reconnect to the new primary. Retry with fresh sync markers + // until all nodes are in sync or the 10-minute deadline is exceeded. + // The 10-minute budget accommodates Spock's reconnect delay when the + // provider node's primary changes (the logical replication slot must be + // re-created on the new primary, which takes additional time in CI). + // Each iteration calls wait_for_sync_event (30s timeout) plus a 15s + // sleep, so roughly 77s per retry; 10 minutes allows ~7 retries. + deadline := time.Now().Add(10 * time.Minute) + for { + if err := f.Refresh(ctx); err != nil { + t.Errorf("failed to refresh database state while waiting for replication: %v", err) + return + } + ok, reason := f.pollReplication(ctx, username, password) + if ok { + return + } + if time.Now().After(deadline) { + t.Errorf("replication did not catch up on all nodes within 10 minutes (last failure: %s)", reason) + return + } + tLogf(t, "replication not yet in sync, retrying in 15s... (%s)", reason) + select { + case <-ctx.Done(): + t.Errorf("replication wait aborted: %v", ctx.Err()) + return + case <-time.After(15 * time.Second): } - - f.WithConnection(ctx, primaryOpts, t, func(conn *pgx.Conn) { - var syncLSN string - - row := conn.QueryRow(ctx, "SELECT spock.sync_event();") - require.NoError(t, row.Scan(&syncLSN)) - - assert.NotEmpty(t, syncLSN) - - nodeSyncMap[node.Name] = syncLSN - }) } +} - // Verify wait_for_sync_event on all other nodes +// pollReplication inserts a sync marker on every primary, then checks that +// each marker has been delivered to every other node's primary. Returns +// (true, "") only when all cross-node checks pass. On failure it returns +// (false, reason) where reason identifies the failing pair so callers can +// log it. Never calls t.Fatal/t.Error so it can be used safely in a retry loop. +func (f *DatabaseFixture) pollReplication(ctx context.Context, username, password string) (bool, string) { + nodeSyncMap := make(map[string]string) for _, node := range f.Spec.Nodes { - primaryOpts := ConnectionOptions{ + conn, err := f.ConnectToInstance(ctx, ConnectionOptions{ Matcher: And(WithNode(node.Name), WithRole("primary")), Username: username, Password: password, + }) + if err != nil { + return false, fmt.Sprintf("connect to %s primary: %v", node.Name, err) + } + var syncLSN string + err = conn.QueryRow(ctx, "SELECT spock.sync_event();").Scan(&syncLSN) + conn.Close(ctx) + if err != nil || syncLSN == "" { + return false, fmt.Sprintf("sync_event on %s: %v", node.Name, err) } + nodeSyncMap[node.Name] = syncLSN + } + for _, node := range f.Spec.Nodes { for peerNode, lsn := range nodeSyncMap { if peerNode == node.Name { continue } - - f.WithConnection(ctx, primaryOpts, t, func(conn *pgx.Conn) { - var synced bool - - row := conn.QueryRow(ctx, "CALL spock.wait_for_sync_event(true, $1, $2::pg_lsn, 30);", peerNode, lsn) - require.NoError(t, row.Scan(&synced)) - assert.True(t, synced) + conn, err := f.ConnectToInstance(ctx, ConnectionOptions{ + Matcher: And(WithNode(node.Name), WithRole("primary")), + Username: username, + Password: password, }) + if err != nil { + return false, fmt.Sprintf("connect to %s primary for peer check: %v", node.Name, err) + } + var synced bool + err = conn.QueryRow(ctx, "CALL spock.wait_for_sync_event(true, $1, $2::pg_lsn, 30);", peerNode, lsn).Scan(&synced) + conn.Close(ctx) + if err != nil || !synced { + return false, fmt.Sprintf("%s waiting for %s events (LSN %s): synced=%v err=%v", node.Name, peerNode, lsn, synced, err) + } } } + return true, "" } func (d *DatabaseFixture) SwitchoverDatabaseNode(ctx context.Context, req *controlplane.SwitchoverDatabaseNodePayload) error { diff --git a/e2e/failover_test.go b/e2e/failover_test.go index 03e30e9d..7e936b40 100644 --- a/e2e/failover_test.go +++ b/e2e/failover_test.go @@ -17,11 +17,13 @@ import ( func TestFailoverScenarios(t *testing.T) { t.Parallel() - host1 := fixture.HostIDs()[0] - host2 := fixture.HostIDs()[1] - host3 := fixture.HostIDs()[2] + hostIDs := fixture.HostIDs() + require.GreaterOrEqual(t, len(hostIDs), 3, "fixture must provide at least 3 hosts") + host1 := hostIDs[0] + host2 := hostIDs[1] + host3 := hostIDs[2] - ctx, cancel := context.WithTimeout(context.Background(), 8*time.Minute) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute) defer cancel() db := fixture.NewDatabaseFixture(ctx, t, &controlplane.CreateDatabaseRequest{ @@ -42,6 +44,10 @@ func TestFailoverScenarios(t *testing.T) { Name: "n1", HostIds: []controlplane.Identifier{controlplane.Identifier(host1), controlplane.Identifier(host2), controlplane.Identifier(host3)}, }, + { + Name: "n2", + HostIds: []controlplane.Identifier{controlplane.Identifier(host1)}, + }, }, }, }) @@ -56,15 +62,17 @@ func TestFailoverScenarios(t *testing.T) { return inst.ID } - waitFor(func() bool { - db.Refresh(ctx) + require.True(t, waitFor(func() bool { + require.NoError(t, db.Refresh(ctx)) for _, inst := range db.Instances { if inst.NodeName == "n1" && (inst.State == "modifying" || inst.State == "creating") { return false } } return true - }, 60*time.Second) + }, 60*time.Second)) + + waitForFailoverSlots(ctx, t, db) // Returns a non-primary instance that's ready/available, or "" if none found within timeout. waitForReadyReplica := func(timeout time.Duration) string { @@ -138,6 +146,7 @@ func TestFailoverScenarios(t *testing.T) { "[auto] primary did not change within timeout (still %s)", origPrimary) newPrimary := getPrimaryInstanceID() t.Logf("[auto] new primary: %s", newPrimary) + db.WaitForReplication(ctx, t, "admin", "password") }) t.Run("failover to a specific candidate", func(t *testing.T) { diff --git a/e2e/switchover_test.go b/e2e/switchover_test.go index ef5a74ab..b1494b2c 100644 --- a/e2e/switchover_test.go +++ b/e2e/switchover_test.go @@ -9,19 +9,23 @@ import ( "time" "github.com/google/uuid" + "github.com/jackc/pgx/v5" "github.com/stretchr/testify/require" controlplane "github.com/pgEdge/control-plane/api/apiv1/gen/control_plane" + "github.com/pgEdge/control-plane/client" ) func TestSwitchoverScenarios(t *testing.T) { t.Parallel() - host1 := fixture.HostIDs()[0] - host2 := fixture.HostIDs()[1] - host3 := fixture.HostIDs()[2] + hostIDs := fixture.HostIDs() + require.GreaterOrEqual(t, len(hostIDs), 3, "fixture must provide at least 3 hosts") + host1 := hostIDs[0] + host2 := hostIDs[1] + host3 := hostIDs[2] - ctx, cancel := context.WithTimeout(context.Background(), 8*time.Minute) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute) defer cancel() db := fixture.NewDatabaseFixture(ctx, t, &controlplane.CreateDatabaseRequest{ @@ -42,6 +46,10 @@ func TestSwitchoverScenarios(t *testing.T) { Name: "n1", HostIds: []controlplane.Identifier{controlplane.Identifier(host1), controlplane.Identifier(host2), controlplane.Identifier(host3)}, }, + { + Name: "n2", + HostIds: []controlplane.Identifier{controlplane.Identifier(host1)}, + }, }, }, }) @@ -60,15 +68,17 @@ func TestSwitchoverScenarios(t *testing.T) { return inst.ID } - waitFor(func() bool { - db.Refresh(ctx) + require.True(t, waitFor(func() bool { + require.NoError(t, db.Refresh(ctx)) for _, inst := range db.Instances { if inst.NodeName == "n1" && (inst.State == "modifying" || inst.State == "creating") { return false } } return true - }, 60*time.Second) + }, 60*time.Second)) + + waitForFailoverSlots(ctx, t, db) // Returns a non-primary instance that's ready/available, or "" if none found within timeout. waitForReadyReplica := func(curPrimary string, timeout time.Duration) string { @@ -140,6 +150,7 @@ func TestSwitchoverScenarios(t *testing.T) { "[auto] primary did not change within timeout (still %s)", origPrimary) newPrimary := getPrimaryInstanceID() t.Logf("[auto] new primary: %s", newPrimary) + db.WaitForReplication(ctx, t, "admin", "password") }) t.Run("switchover to a specific candidate", func(t *testing.T) { @@ -261,6 +272,32 @@ func waitFor(cond func() bool, timeout time.Duration) bool { return false } +func waitForFailoverSlots(ctx context.Context, t *testing.T, db *DatabaseFixture) { + require.NoError(t, db.Refresh(ctx)) + replicas := db.GetInstances(And(WithRole(client.RoleReplica))) + require.NotEmpty(t, replicas) + for replica := range replicas { + opts := ConnectionOptions{ + Username: "admin", + Password: "password", + Instance: replica, + } + db.WithConnection(ctx, opts, t, func(conn *pgx.Conn) { + tLogf(t, "waiting for failover slot to exist on replica instance %s", replica.ID) + start := time.Now() + + require.True(t, waitFor(func() bool { + var failoverSlotExists bool + row := conn.QueryRow(ctx, `select exists (select 1 from pg_replication_slots where plugin = 'spock_output')`) + require.NoError(t, row.Scan(&failoverSlotExists)) + return failoverSlotExists + }, 60*time.Second)) + + tLogf(t, "failover slot exists in replica instance %s after %.2f seconds", replica.ID, time.Since(start).Seconds()) + }) + } +} + func serverNowUTC(t *testing.T, baseURL string) time.Time { req, err := http.NewRequest(http.MethodHead, baseURL+"/v1/version", nil) require.NoError(t, err) From 78670c3ec731c3bcf10878fb17912d810044d9ba Mon Sep 17 00:00:00 2001 From: Siva Date: Mon, 1 Jun 2026 17:47:25 +0530 Subject: [PATCH 2/2] addressing review comments --- e2e/database_test.go | 93 +++++++++++++----------------------------- e2e/failover_test.go | 2 +- e2e/switchover_test.go | 2 +- 3 files changed, 31 insertions(+), 66 deletions(-) diff --git a/e2e/database_test.go b/e2e/database_test.go index 0c8a15dd..52f29367 100644 --- a/e2e/database_test.go +++ b/e2e/database_test.go @@ -8,9 +8,10 @@ import ( "fmt" "iter" "testing" - "time" "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" controlplane "github.com/pgEdge/control-plane/api/apiv1/gen/control_plane" "github.com/pgEdge/control-plane/client" @@ -355,85 +356,49 @@ func (d *DatabaseFixture) waitForTask(ctx context.Context, task *controlplane.Ta func (f *DatabaseFixture) WaitForReplication(ctx context.Context, t testing.TB, username, password string) { tLog(t, "waiting for replication to catch up on all nodes") - // After a primary change (switchover/failover), Spock subscriptions need - // time to reconnect to the new primary. Retry with fresh sync markers - // until all nodes are in sync or the 10-minute deadline is exceeded. - // The 10-minute budget accommodates Spock's reconnect delay when the - // provider node's primary changes (the logical replication slot must be - // re-created on the new primary, which takes additional time in CI). - // Each iteration calls wait_for_sync_event (30s timeout) plus a 15s - // sleep, so roughly 77s per retry; 10 minutes allows ~7 retries. - deadline := time.Now().Add(10 * time.Minute) - for { - if err := f.Refresh(ctx); err != nil { - t.Errorf("failed to refresh database state while waiting for replication: %v", err) - return - } - ok, reason := f.pollReplication(ctx, username, password) - if ok { - return - } - if time.Now().After(deadline) { - t.Errorf("replication did not catch up on all nodes within 10 minutes (last failure: %s)", reason) - return - } - tLogf(t, "replication not yet in sync, retrying in 15s... (%s)", reason) - select { - case <-ctx.Done(): - t.Errorf("replication wait aborted: %v", ctx.Err()) - return - case <-time.After(15 * time.Second): - } - } -} - -// pollReplication inserts a sync marker on every primary, then checks that -// each marker has been delivered to every other node's primary. Returns -// (true, "") only when all cross-node checks pass. On failure it returns -// (false, reason) where reason identifies the failing pair so callers can -// log it. Never calls t.Fatal/t.Error so it can be used safely in a retry loop. -func (f *DatabaseFixture) pollReplication(ctx context.Context, username, password string) (bool, string) { + // Execute sync_event on all primary nodes nodeSyncMap := make(map[string]string) for _, node := range f.Spec.Nodes { - conn, err := f.ConnectToInstance(ctx, ConnectionOptions{ + primaryOpts := ConnectionOptions{ Matcher: And(WithNode(node.Name), WithRole("primary")), Username: username, Password: password, - }) - if err != nil { - return false, fmt.Sprintf("connect to %s primary: %v", node.Name, err) - } - var syncLSN string - err = conn.QueryRow(ctx, "SELECT spock.sync_event();").Scan(&syncLSN) - conn.Close(ctx) - if err != nil || syncLSN == "" { - return false, fmt.Sprintf("sync_event on %s: %v", node.Name, err) } - nodeSyncMap[node.Name] = syncLSN + + f.WithConnection(ctx, primaryOpts, t, func(conn *pgx.Conn) { + var syncLSN string + + row := conn.QueryRow(ctx, "SELECT spock.sync_event();") + require.NoError(t, row.Scan(&syncLSN)) + + assert.NotEmpty(t, syncLSN) + + nodeSyncMap[node.Name] = syncLSN + }) } + // Verify wait_for_sync_event on all other nodes for _, node := range f.Spec.Nodes { + primaryOpts := ConnectionOptions{ + Matcher: And(WithNode(node.Name), WithRole("primary")), + Username: username, + Password: password, + } + for peerNode, lsn := range nodeSyncMap { if peerNode == node.Name { continue } - conn, err := f.ConnectToInstance(ctx, ConnectionOptions{ - Matcher: And(WithNode(node.Name), WithRole("primary")), - Username: username, - Password: password, + + f.WithConnection(ctx, primaryOpts, t, func(conn *pgx.Conn) { + var synced bool + + row := conn.QueryRow(ctx, "CALL spock.wait_for_sync_event(true, $1, $2::pg_lsn, 30);", peerNode, lsn) + require.NoError(t, row.Scan(&synced)) + assert.True(t, synced) }) - if err != nil { - return false, fmt.Sprintf("connect to %s primary for peer check: %v", node.Name, err) - } - var synced bool - err = conn.QueryRow(ctx, "CALL spock.wait_for_sync_event(true, $1, $2::pg_lsn, 30);", peerNode, lsn).Scan(&synced) - conn.Close(ctx) - if err != nil || !synced { - return false, fmt.Sprintf("%s waiting for %s events (LSN %s): synced=%v err=%v", node.Name, peerNode, lsn, synced, err) - } } } - return true, "" } func (d *DatabaseFixture) SwitchoverDatabaseNode(ctx context.Context, req *controlplane.SwitchoverDatabaseNodePayload) error { diff --git a/e2e/failover_test.go b/e2e/failover_test.go index 7e936b40..3cd4f5e5 100644 --- a/e2e/failover_test.go +++ b/e2e/failover_test.go @@ -23,7 +23,7 @@ func TestFailoverScenarios(t *testing.T) { host2 := hostIDs[1] host3 := hostIDs[2] - ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute) + ctx, cancel := context.WithTimeout(context.Background(), 8*time.Minute) defer cancel() db := fixture.NewDatabaseFixture(ctx, t, &controlplane.CreateDatabaseRequest{ diff --git a/e2e/switchover_test.go b/e2e/switchover_test.go index b1494b2c..4021d71d 100644 --- a/e2e/switchover_test.go +++ b/e2e/switchover_test.go @@ -25,7 +25,7 @@ func TestSwitchoverScenarios(t *testing.T) { host2 := hostIDs[1] host3 := hostIDs[2] - ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute) + ctx, cancel := context.WithTimeout(context.Background(), 8*time.Minute) defer cancel() db := fixture.NewDatabaseFixture(ctx, t, &controlplane.CreateDatabaseRequest{