fix: prevent subscription breakage on rolling updates

e2e/minor_version_upgrade_test.go

@@ -14,13 +14,183 @@ 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()
 
 	fixture.SkipIfUpgradesUnsupported(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)
 

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 {

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

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

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
-      }
     ]
   ]
 ]

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
-      }
     ]
   ]
 ]