diff --git a/e2e/minor_version_upgrade_test.go b/e2e/minor_version_upgrade_test.go index dfbd8e57..69d0e7f3 100644 --- a/e2e/minor_version_upgrade_test.go +++ b/e2e/minor_version_upgrade_test.go @@ -14,6 +14,174 @@ import ( controlplane "github.com/pgEdge/control-plane/api/apiv1/gen/control_plane" ) +// TestRollingUpdatePreservesReplication verifies that Spock subscriptions +// remain functional after a rolling update on a multi-node database with +// replica instances (PLAT-665). It writes data before and after the update +// and asserts that both directions of replication still work. +func TestRollingUpdatePreservesReplication(t *testing.T) { + t.Parallel() + + fixture.SkipIfUpgradesUnsupported(t) + + host1 := fixture.HostIDs()[0] + host2 := fixture.HostIDs()[1] + host3 := fixture.HostIDs()[2] + host4 := fixture.HostIDs()[3] + + username := "admin" + password := "password" + + fromVersion := "18.3" + toVersion := "18.4" + + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Minute) + defer cancel() + + t.Log("creating two-node database with replicas") + + db := fixture.NewDatabaseFixture(t.Context(), t, &controlplane.CreateDatabaseRequest{ + Spec: &controlplane.DatabaseSpec{ + DatabaseName: "repl_test", + PostgresVersion: &fromVersion, + Port: pointerTo(0), + PatroniPort: pointerTo(0), + DatabaseUsers: []*controlplane.DatabaseUserSpec{ + { + Username: username, + Password: pointerTo(password), + DbOwner: pointerTo(true), + Attributes: []string{"LOGIN", "SUPERUSER"}, + }, + }, + Nodes: []*controlplane.DatabaseNodeSpec{ + {Name: "n1", HostIds: []controlplane.Identifier{ + controlplane.Identifier(host1), + controlplane.Identifier(host2), + }}, + {Name: "n2", HostIds: []controlplane.Identifier{ + controlplane.Identifier(host3), + controlplane.Identifier(host4), + }}, + }, + }, + }) + + n1Opts := ConnectionOptions{Matcher: And(WithNode("n1"), WithRole("primary")), Username: username, Password: password} + n2Opts := ConnectionOptions{Matcher: And(WithNode("n2"), WithRole("primary")), Username: username, Password: password} + + // Write a row on n1 and wait for it to replicate to n2 before the update. + t.Log("writing pre-update data to n1") + + var preLSN string + db.WithConnection(ctx, n1Opts, t, func(conn *pgx.Conn) { + _, err := conn.Exec(ctx, "CREATE TABLE items (id INT PRIMARY KEY, data TEXT);") + require.NoError(t, err) + + _, err = conn.Exec(ctx, "INSERT INTO items VALUES (1, 'before-update');") + require.NoError(t, err) + + require.NoError(t, conn.QueryRow(ctx, "SELECT spock.sync_event();").Scan(&preLSN)) + }) + + t.Log("waiting for pre-update row to replicate to n2") + + db.WithConnection(ctx, n2Opts, t, func(conn *pgx.Conn) { + var synced bool + require.NoError(t, conn.QueryRow(ctx, + "CALL spock.wait_for_sync_event(true, $1, $2::pg_lsn, 30);", "n1", preLSN, + ).Scan(&synced)) + require.True(t, synced, "pre-update row did not replicate to n2 before update") + + var data string + require.NoError(t, conn.QueryRow(ctx, "SELECT data FROM items WHERE id = 1;").Scan(&data)) + assert.Equal(t, "before-update", data) + }) + + // Perform the rolling update — version bump changes the image, which + // triggers a container restart on each instance. + t.Logf("performing rolling update from %s to %s", fromVersion, toVersion) + + err := db.Update(ctx, UpdateOptions{ + Spec: &controlplane.DatabaseSpec{ + DatabaseName: "repl_test", + PostgresVersion: &toVersion, + Port: pointerTo(0), + PatroniPort: pointerTo(0), + DatabaseUsers: []*controlplane.DatabaseUserSpec{ + { + Username: username, + Password: pointerTo(password), + DbOwner: pointerTo(true), + Attributes: []string{"LOGIN", "SUPERUSER"}, + }, + }, + Nodes: []*controlplane.DatabaseNodeSpec{ + {Name: "n1", HostIds: []controlplane.Identifier{ + controlplane.Identifier(host1), + controlplane.Identifier(host2), + }}, + {Name: "n2", HostIds: []controlplane.Identifier{ + controlplane.Identifier(host3), + controlplane.Identifier(host4), + }}, + }, + }, + }) + require.NoError(t, err) + + t.Log("asserting both nodes have an active primary after update") + require.NotNil(t, db.GetInstance(And(WithNode("n1"), WithRole("primary")))) + require.NotNil(t, db.GetInstance(And(WithNode("n2"), WithRole("primary")))) + + // Write a new row on n1 after the update and verify it reaches n2. + // This is the core regression check: if the second switchover broke the + // n1→n2 subscription, this insert will never replicate. + t.Log("writing post-update data to n1, verifying replication to n2") + + var postLSN string + db.WithConnection(ctx, n1Opts, t, func(conn *pgx.Conn) { + _, err := conn.Exec(ctx, "INSERT INTO items VALUES (2, 'after-update');") + require.NoError(t, err) + + require.NoError(t, conn.QueryRow(ctx, "SELECT spock.sync_event();").Scan(&postLSN)) + }) + + db.WithConnection(ctx, n2Opts, t, func(conn *pgx.Conn) { + var synced bool + require.NoError(t, conn.QueryRow(ctx, + "CALL spock.wait_for_sync_event(true, $1, $2::pg_lsn, 30);", "n1", postLSN, + ).Scan(&synced)) + require.True(t, synced, "post-update row did not replicate n1→n2") + + var data string + require.NoError(t, conn.QueryRow(ctx, "SELECT data FROM items WHERE id = 2;").Scan(&data)) + assert.Equal(t, "after-update", data) + }) + + // Also verify the reverse direction: n2→n1. + t.Log("writing post-update data to n2, verifying replication to n1") + + var reverseLSN string + db.WithConnection(ctx, n2Opts, t, func(conn *pgx.Conn) { + _, err := conn.Exec(ctx, "INSERT INTO items VALUES (3, 'after-update-n2');") + require.NoError(t, err) + + require.NoError(t, conn.QueryRow(ctx, "SELECT spock.sync_event();").Scan(&reverseLSN)) + }) + + db.WithConnection(ctx, n1Opts, t, func(conn *pgx.Conn) { + var synced bool + require.NoError(t, conn.QueryRow(ctx, + "CALL spock.wait_for_sync_event(true, $1, $2::pg_lsn, 30);", "n2", reverseLSN, + ).Scan(&synced)) + require.True(t, synced, "post-update row did not replicate n2→n1") + + var data string + require.NoError(t, conn.QueryRow(ctx, "SELECT data FROM items WHERE id = 3;").Scan(&data)) + assert.Equal(t, "after-update-n2", data) + }) +} + func TestMinorVersionUpgrade(t *testing.T) { t.Parallel() @@ -21,6 +189,8 @@ func TestMinorVersionUpgrade(t *testing.T) { host1 := fixture.HostIDs()[0] host2 := fixture.HostIDs()[1] + host3 := fixture.HostIDs()[2] + host4 := fixture.HostIDs()[3] username := "admin" password := "password" @@ -54,15 +224,20 @@ func TestMinorVersionUpgrade(t *testing.T) { controlplane.Identifier(host2), }, }, + { + Name: "n2", HostIds: []controlplane.Identifier{ + controlplane.Identifier(host3), + controlplane.Identifier(host4), + }, + }, }, }, }) t.Log("asserting that the primary is running on the first host") - // Assert that the primary is running on the first host in the host_ids - // list. - primary := db.GetInstance(WithRole("primary")) + // Assert that the primary is running on the first host in the host_ids list. + primary := db.GetInstance(And(WithNode("n1"), WithRole("primary"))) require.NotNil(t, primary) assert.Equal(t, host1, primary.HostID) @@ -107,18 +282,33 @@ func TestMinorVersionUpgrade(t *testing.T) { controlplane.Identifier(host2), }, }, + { + Name: "n2", HostIds: []controlplane.Identifier{ + controlplane.Identifier(host3), + controlplane.Identifier(host4), + }, + }, }, }, }) require.NoError(t, err) - t.Log("asserting that the primary hasn't changed") + t.Log("asserting that both nodes have an active primary after update") + + // The primary may have moved to a different host during the rolling update. + // Rolling updates no longer force a switchover back to the original primary + // (PLAT-665), so we only assert that each node has some primary. + updatedN1 := db.GetInstance(And(WithNode("n1"), WithRole("primary"))) + require.NotNil(t, updatedN1) + updatedN2 := db.GetInstance(And(WithNode("n2"), WithRole("primary"))) + require.NotNil(t, updatedN2) - // Assert that the primary hasn't changed - updated := db.GetInstance(WithRole("primary")) - require.NotNil(t, updated) - assert.Equal(t, primary.ID, updated.ID) - assert.Equal(t, primary.HostID, updated.HostID) + // Use the current n1 primary for the version check. + opts = ConnectionOptions{ + InstanceID: updatedN1.ID, + Username: username, + Password: password, + } t.Logf("validating that the version is updated to %s", toVersion) diff --git a/server/internal/database/instance_resource.go b/server/internal/database/instance_resource.go index 32ba8256..c93528a7 100644 --- a/server/internal/database/instance_resource.go +++ b/server/internal/database/instance_resource.go @@ -204,6 +204,12 @@ func (r *InstanceResource) initializeInstance(ctx context.Context, rc *resource. } if r.Spec.InstanceID != r.PrimaryInstanceID { + // Spock 5.x takes up to 60 s to create failover replication slots after a + // switchover. Wait for them before marking this replica available so that + // subscribers connected to this node do not lose their slots. + if err := r.waitForReplicationSlots(ctx, rc); err != nil { + return r.recordError(ctx, rc, err) + } err := r.updateInstanceRecord(ctx, rc, &InstanceUpdateOptions{State: InstanceStateAvailable}) if err != nil { return r.recordError(ctx, rc, err) @@ -246,6 +252,60 @@ func (r *InstanceResource) initializeInstance(ctx context.Context, rc *resource. return nil } +// waitForReplicationSlots polls pg_replication_slots on this instance until the +// replication slot for every subscription where this node is the provider exists. +// In Spock 5.x the worker process on a replica takes up to 60 s to create +// failover slots after a switchover; this blocks until they are ready. +func (r *InstanceResource) waitForReplicationSlots(ctx context.Context, rc *resource.Context) error { + subs, err := resource.AllFromContext[*SubscriptionResource](rc, ResourceTypeSubscription) + if err != nil { + return fmt.Errorf("failed to list subscriptions: %w", err) + } + + var relevant []*SubscriptionResource + for _, sub := range subs { + if sub.ProviderNode == r.Spec.NodeName { + relevant = append(relevant, sub) + } + } + if len(relevant) == 0 { + return nil + } + + conn, err := r.Connection(ctx, rc, r.Spec.DatabaseName) + if err != nil { + return fmt.Errorf("failed to connect for replication slot check: %w", err) + } + defer conn.Close(ctx) + + const ( + pollInterval = 2 * time.Second + slotWaitTimeout = 2 * time.Minute + ) + ctx, cancel := context.WithTimeout(ctx, slotWaitTimeout) + defer cancel() + + for _, sub := range relevant { + for { + exists, err := postgres.ReplicationSlotExists(sub.DatabaseName, sub.ProviderNode, sub.SubscriberNode). + Scalar(ctx, conn) + if err != nil { + return fmt.Errorf("failed to check replication slot for %s→%s: %w", + sub.ProviderNode, sub.SubscriberNode, err) + } + if exists { + break + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(pollInterval): + } + } + } + return nil +} + func (r *InstanceResource) updateInstanceRecord(ctx context.Context, rc *resource.Context, opts *InstanceUpdateOptions) error { svc, err := do.Invoke[*Service](rc.Injector) if err != nil { diff --git a/server/internal/database/operations/common.go b/server/internal/database/operations/common.go index 82e47a1f..af2a8c31 100644 --- a/server/internal/database/operations/common.go +++ b/server/internal/database/operations/common.go @@ -20,15 +20,6 @@ type NodeResources struct { Scripts database.Scripts } -func (n *NodeResources) primaryInstance() *database.InstanceResources { - for _, instance := range n.InstanceResources { - if instance.InstanceID() == n.PrimaryInstanceID { - return instance - } - } - - return nil -} func (n *NodeResources) nodeResourceState() (*resource.State, error) { var instanceIDs []string diff --git a/server/internal/database/operations/end.go b/server/internal/database/operations/end.go index 8ced1f80..6ee3d2f4 100644 --- a/server/internal/database/operations/end.go +++ b/server/internal/database/operations/end.go @@ -4,7 +4,6 @@ import ( "fmt" "github.com/pgEdge/control-plane/server/internal/database" - "github.com/pgEdge/control-plane/server/internal/patroni" "github.com/pgEdge/control-plane/server/internal/resource" ) @@ -35,21 +34,6 @@ func EndState(nodes []*NodeResources, services []*ServiceResources) (*resource.S } end.Merge(databaseState) - if len(node.InstanceResources) > 1 { - primary := node.primaryInstance() - if primary != nil { - // Primary will be non-nil for existing nodes. Adding the - // switchover resource to the end state prevents "permadrift" - // where this resource is created and deleted even if the update - // is a no-op. - resources = append(resources, &database.SwitchoverResource{ - HostID: primary.HostID(), - InstanceID: primary.InstanceID(), - TargetRole: patroni.InstanceRolePrimary, - }) - } - } - for _, peer := range nodes { if peer.NodeName == node.NodeName { continue diff --git a/server/internal/database/operations/golden_test/TestRestoreDatabase/single_node_restore_in_two-node_db_with_replica.json b/server/internal/database/operations/golden_test/TestRestoreDatabase/single_node_restore_in_two-node_db_with_replica.json index c216cbcd..4493ff17 100644 --- a/server/internal/database/operations/golden_test/TestRestoreDatabase/single_node_restore_in_two-node_db_with_replica.json +++ b/server/internal/database/operations/golden_test/TestRestoreDatabase/single_node_restore_in_two-node_db_with_replica.json @@ -187,14 +187,6 @@ "reason": "does_not_exist", "diff": null } - ], - [ - { - "type": "create", - "resource_id": "database.switchover::n1-instance-1-id", - "reason": "does_not_exist", - "diff": null - } ] ] ] \ No newline at end of file diff --git a/server/internal/database/operations/golden_test/TestUpdateDatabase/adding_a_replica.json b/server/internal/database/operations/golden_test/TestUpdateDatabase/adding_a_replica.json index c16a14f7..90d5724e 100644 --- a/server/internal/database/operations/golden_test/TestUpdateDatabase/adding_a_replica.json +++ b/server/internal/database/operations/golden_test/TestUpdateDatabase/adding_a_replica.json @@ -63,14 +63,6 @@ "reason": "does_not_exist", "diff": null } - ], - [ - { - "type": "create", - "resource_id": "database.switchover::n1-instance-1-id", - "reason": "does_not_exist", - "diff": null - } ] ] ] \ No newline at end of file diff --git a/server/internal/database/operations/golden_test/TestUpdateDatabase/adding_a_replica_with_primary_update.json b/server/internal/database/operations/golden_test/TestUpdateDatabase/adding_a_replica_with_primary_update.json index a1ad69e6..62d86b24 100644 --- a/server/internal/database/operations/golden_test/TestUpdateDatabase/adding_a_replica_with_primary_update.json +++ b/server/internal/database/operations/golden_test/TestUpdateDatabase/adding_a_replica_with_primary_update.json @@ -125,14 +125,6 @@ "reason": "does_not_exist", "diff": null } - ], - [ - { - "type": "create", - "resource_id": "database.switchover::n1-instance-1-id", - "reason": "does_not_exist", - "diff": null - } ] ] ] \ No newline at end of file diff --git a/server/internal/database/operations/golden_test/TestUpdateDatabase/adding_multiple_replicas_concurrent.json b/server/internal/database/operations/golden_test/TestUpdateDatabase/adding_multiple_replicas_concurrent.json index 5590eb51..0f53d552 100644 --- a/server/internal/database/operations/golden_test/TestUpdateDatabase/adding_multiple_replicas_concurrent.json +++ b/server/internal/database/operations/golden_test/TestUpdateDatabase/adding_multiple_replicas_concurrent.json @@ -99,20 +99,6 @@ "reason": "does_not_exist", "diff": null } - ], - [ - { - "type": "create", - "resource_id": "database.switchover::n1-instance-1-id", - "reason": "does_not_exist", - "diff": null - }, - { - "type": "create", - "resource_id": "database.switchover::n2-instance-1-id", - "reason": "does_not_exist", - "diff": null - } ] ] ] \ No newline at end of file diff --git a/server/internal/database/operations/golden_test/TestUpdateDatabase/adding_multiple_replicas_rolling.json b/server/internal/database/operations/golden_test/TestUpdateDatabase/adding_multiple_replicas_rolling.json index 022d3516..09ba543c 100644 --- a/server/internal/database/operations/golden_test/TestUpdateDatabase/adding_multiple_replicas_rolling.json +++ b/server/internal/database/operations/golden_test/TestUpdateDatabase/adding_multiple_replicas_rolling.json @@ -125,20 +125,6 @@ "reason": "does_not_exist", "diff": null } - ], - [ - { - "type": "create", - "resource_id": "database.switchover::n1-instance-1-id", - "reason": "does_not_exist", - "diff": null - }, - { - "type": "create", - "resource_id": "database.switchover::n2-instance-1-id", - "reason": "does_not_exist", - "diff": null - } ] ] ] \ No newline at end of file diff --git a/server/internal/database/operations/update_nodes.go b/server/internal/database/operations/update_nodes.go index 951b4647..951eaf1a 100644 --- a/server/internal/database/operations/update_nodes.go +++ b/server/internal/database/operations/update_nodes.go @@ -4,8 +4,6 @@ import ( "fmt" "slices" - "github.com/pgEdge/control-plane/server/internal/database" - "github.com/pgEdge/control-plane/server/internal/patroni" "github.com/pgEdge/control-plane/server/internal/resource" ) @@ -16,7 +14,6 @@ import ( func UpdateNode(start *resource.State, node *NodeResources) ([]*resource.State, error) { var primary *resource.State var replicaUpdates, replicaAdds []*resource.State - var primaryHostID string for _, inst := range node.InstanceResources { state, err := inst.InstanceState() @@ -30,7 +27,6 @@ func UpdateNode(start *resource.State, node *NodeResources) ([]*resource.State, return nil, fmt.Errorf("invalid state: node %s exists, but its primary instance '%s' hasn't been created yet", node.NodeName, node.PrimaryInstanceID) } primary = state - primaryHostID = inst.HostID() case start.HasResources(inst.Instance.Identifier()): replicaUpdates = append(replicaUpdates, state) default: @@ -46,19 +42,6 @@ func UpdateNode(start *resource.State, node *NodeResources) ([]*resource.State, return nil, fmt.Errorf("node %s has no primary instance", node.NodeName) } - // This condition is true when we have existing replicas - if len(replicaUpdates) != 0 { - // Ensure that we always switch back to the original primary - err := primary.AddResource(&database.SwitchoverResource{ - HostID: primaryHostID, - InstanceID: node.PrimaryInstanceID, - TargetRole: patroni.InstanceRolePrimary, - }) - if err != nil { - return nil, fmt.Errorf("failed to add switchover resource to replica state: %w", err) - } - } - // Existing replicas should be updated first and new replicas should be // added last. states := slices.Concat(replicaUpdates, []*resource.State{primary}, replicaAdds) diff --git a/server/internal/database/operations/update_nodes_test.go b/server/internal/database/operations/update_nodes_test.go index 697652b8..c91bac08 100644 --- a/server/internal/database/operations/update_nodes_test.go +++ b/server/internal/database/operations/update_nodes_test.go @@ -8,7 +8,6 @@ import ( "github.com/pgEdge/control-plane/server/internal/database" "github.com/pgEdge/control-plane/server/internal/database/operations" - "github.com/pgEdge/control-plane/server/internal/patroni" "github.com/pgEdge/control-plane/server/internal/resource" ) @@ -105,11 +104,6 @@ func TestUpdateNode(t *testing.T) { instance2.InstanceID(), }, }, - &database.SwitchoverResource{ - HostID: instance1.HostID(), - InstanceID: instance1.InstanceID(), - TargetRole: patroni.InstanceRolePrimary, - }, }, instance1.InstanceDependencies, ), @@ -173,11 +167,6 @@ func TestUpdateNode(t *testing.T) { instance3.InstanceID(), }, }, - &database.SwitchoverResource{ - HostID: instance1.HostID(), - InstanceID: instance1.InstanceID(), - TargetRole: patroni.InstanceRolePrimary, - }, }, instance1.InstanceDependencies, ), @@ -438,11 +427,6 @@ func TestRollingUpdateNodes(t *testing.T) { n1Instance2.InstanceID(), }, }, - &database.SwitchoverResource{ - HostID: n1Instance1.HostID(), - InstanceID: n1Instance1.InstanceID(), - TargetRole: patroni.InstanceRolePrimary, - }, }, n1Instance1.InstanceDependencies, ), @@ -527,11 +511,6 @@ func TestRollingUpdateNodes(t *testing.T) { n1Instance2.InstanceID(), }, }, - &database.SwitchoverResource{ - HostID: n1Instance1.HostID(), - InstanceID: n1Instance1.InstanceID(), - TargetRole: patroni.InstanceRolePrimary, - }, }, n1Instance1.InstanceDependencies, ), @@ -551,11 +530,6 @@ func TestRollingUpdateNodes(t *testing.T) { n2Instance2.InstanceID(), }, }, - &database.SwitchoverResource{ - HostID: n2Instance1.HostID(), - InstanceID: n2Instance1.InstanceID(), - TargetRole: patroni.InstanceRolePrimary, - }, }, n2Instance1.InstanceDependencies, ), @@ -820,11 +794,6 @@ func TestConcurrentUpdateNodes(t *testing.T) { n1Instance2.InstanceID(), }, }, - &database.SwitchoverResource{ - HostID: n1Instance1.HostID(), - InstanceID: n1Instance1.InstanceID(), - TargetRole: patroni.InstanceRolePrimary, - }, }, n1Instance1.InstanceDependencies, ), @@ -911,16 +880,6 @@ func TestConcurrentUpdateNodes(t *testing.T) { n2Instance2.InstanceID(), }, }, - &database.SwitchoverResource{ - HostID: n1Instance1.HostID(), - InstanceID: n1Instance1.InstanceID(), - TargetRole: patroni.InstanceRolePrimary, - }, - &database.SwitchoverResource{ - HostID: n2Instance1.HostID(), - InstanceID: n2Instance1.InstanceID(), - TargetRole: patroni.InstanceRolePrimary, - }, }, slices.Concat( n1Instance1.InstanceDependencies, diff --git a/server/internal/postgres/create_db.go b/server/internal/postgres/create_db.go index 5a52bb2b..d7148dee 100644 --- a/server/internal/postgres/create_db.go +++ b/server/internal/postgres/create_db.go @@ -366,6 +366,19 @@ func IsReplicationSlotActive(databaseName, providerNode, subscriberNode string) } } +// ReplicationSlotExists checks whether the replication slot for the given +// subscription exists. Used to poll for Spock 5.x failover slot creation after +// a switchover, which can take up to 60 seconds. +func ReplicationSlotExists(databaseName, providerNode, subscriberNode string) Query[bool] { + slotName := ReplicationSlotName(databaseName, providerNode, subscriberNode) + return Query[bool]{ + SQL: "SELECT EXISTS (SELECT 1 FROM pg_replication_slots WHERE slot_name = @slot_name);", + Args: pgx.NamedArgs{ + "slot_name": slotName, + }, + } +} + func GetReplicationSlotLSNFromCommitTS(databaseName, providerNode, subscriberNode string, commitTS time.Time) Query[string] { slotName := ReplicationSlotName(databaseName, providerNode, subscriberNode)