fix: update patroni etcd hosts on host removal

changes/unreleased/Fixed-20261102-000000.yaml

@@ -0,0 +1,3 @@
+kind: Fixed
+body: Removing a host now updates the Patroni configuration of every modifiable database so that it no longer refers to the removed Etcd server. Databases that can't be modified at that time are updated on their next database update.
+time: 2026-11-02T00:00:00.000000+00:00

server/internal/api/apiv1/post_init_handlers.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"net"
 	"net/url"
+	"slices"
 	"strconv"
 	"strings"
 	"time"
@@ -201,17 +202,17 @@ func (s *PostInitHandlers) RemoveHost(ctx context.Context, req *api.RemoveHostPa
 		return nil, err
 	}
 
-	dbs, err := s.dbSvc.GetDatabasesByHostId(ctx, hostID)
+	dbs, err := s.dbSvc.GetDatabases(ctx)
 	if err != nil {
 		return nil, apiErr(err)
 	}
 
-	specs, err := s.prepareDatabaseUpdates(ctx, hostID, dbs)
+	updated, err := s.prepareDatabaseUpdates(ctx, hostID, dbs)
 	if err != nil {
 		return nil, apiErr(err)
 	}
 
-	output, err := s.workflowSvc.RemoveHost(ctx, hostID, specs...)
+	output, err := s.workflowSvc.RemoveHost(ctx, hostID, updated...)
 	if err != nil {
 		return nil, apiErr(err)
 	}
@@ -249,23 +250,44 @@ func (s *PostInitHandlers) validateHostRemoval(ctx context.Context, hostID strin
 	return nil
 }
 
-// prepareDatabaseUpdates prepares workflow inputs for all databases affected by host removal.
-func (s *PostInitHandlers) prepareDatabaseUpdates(ctx context.Context, hostID string, dbs []*database.Database) ([]*database.Spec, error) {
-	specs := make([]*database.Spec, 0, len(dbs))
+// prepareDatabaseUpdates prepares all databases in the cluster to be updated
+// by the remove host workflow. Databases with instances on the removed host
+// have the host removed from their specs. Every other database is updated with
+// its current spec so that its Patroni configuration no longer refers to the
+// removed host if it was an Etcd server.
+func (s *PostInitHandlers) prepareDatabaseUpdates(ctx context.Context, hostID string, dbs []*database.Database) ([]*database.Database, error) {
+	updated := make([]*database.Database, 0, len(dbs))
 	for _, db := range dbs {
-		if ok := db.Spec.RemoveHost(hostID); !ok {
+		if !hasInstanceOnHost(db, hostID) {
+			if !database.DatabaseStateModifiable(db.RawState) {
+				// This database will still pick up the Etcd membership change
+				// the next time it's updated.
+				s.logger.Warn().
+					Str("database_id", db.DatabaseID).
+					Str("host_id", hostID).
+					Str("state", string(db.RawState)).
+					Msg("skipping update for database that is not in a modifiable state")
+				continue
+			}
+		} else if ok := db.Spec.RemoveHost(hostID); !ok {
 			return nil, fmt.Errorf("%s host id not found/removed", hostID)
 		}
 
-		_, err := s.dbSvc.UpdateDatabase(ctx, database.DatabaseStateModifying, db.Spec)
+		updatedDB, err := s.dbSvc.UpdateDatabase(ctx, database.DatabaseStateModifying, db.Spec)
 		if err != nil {
 			return nil, fmt.Errorf("failed to update database: %w", err)
 		}
 
-		specs = append(specs, db.Spec)
+		updated = append(updated, updatedDB)
 	}
 
-	return specs, nil
+	return updated, nil
+}
+
+func hasInstanceOnHost(db *database.Database, hostID string) bool {
+	return slices.ContainsFunc(db.Instances, func(instance *database.Instance) bool {
+		return instance.HostID == hostID
+	})
 }
 
 // ListDatabases fetches all databases from the database service and converts them to API format.

server/internal/orchestrator/common/patroni_config.go

@@ -14,6 +14,7 @@ import (
 	clientv3 "go.etcd.io/etcd/client/v3"
 
 	"github.com/pgEdge/control-plane/server/internal/database"
+	"github.com/pgEdge/control-plane/server/internal/ds"
 	"github.com/pgEdge/control-plane/server/internal/filesystem"
 	"github.com/pgEdge/control-plane/server/internal/patroni"
 	"github.com/pgEdge/control-plane/server/internal/pgbackrest"
@@ -29,6 +30,11 @@ type PatroniConfig struct {
 	ParentID   string                  `json:"parent_id"`
 	OwnerUID   int                     `json:"owner_uid"`
 	OwnerGID   int                     `json:"owner_gid"`
+	// EtcdHostsOutdated is set by Refresh when the Etcd hosts in the existing
+	// config file no longer match the Etcd cluster membership, e.g. after an
+	// Etcd server host is removed. It's never set in the desired state, so
+	// setting it produces a diff that causes the config to be rewritten.
+	EtcdHostsOutdated bool `json:"etcd_hosts_outdated,omitempty"`
 }
 
 func (c *PatroniConfig) Dependencies() []resource.Identifier {
@@ -74,9 +80,26 @@ func (c *PatroniConfig) Refresh(ctx context.Context, rc *resource.Context) error
 		return fmt.Errorf("%w: failed to unmarshal patroni config", resource.ErrNotFound)
 	}
 
+	etcdClient, err := do.Invoke[*clientv3.Client](rc.Injector)
+	if err != nil {
+		return err
+	}
+	etcdHosts, err := patroni.EtcdHosts(ctx, etcdClient)
+	if err != nil {
+		return fmt.Errorf("failed to get etcd hosts: %w", err)
+	}
+	c.EtcdHostsOutdated = etcdHostsOutdated(&config, etcdHosts)
+
 	return nil
 }
 
+func etcdHostsOutdated(config *patroni.Config, etcdHosts []string) bool {
+	if config.Etcd3 == nil || config.Etcd3.Hosts == nil {
+		return true
+	}
+	return !ds.NewSet(*config.Etcd3.Hosts...).Equal(ds.NewSet(etcdHosts...))
+}
+
 func (c *PatroniConfig) Create(
 	ctx context.Context,
 	rc *resource.Context,
@@ -207,6 +230,7 @@ func (c *PatroniConfig) create(
 	if err := fs.Chown(configPath, c.OwnerUID, c.OwnerGID); err != nil {
 		return nil, fmt.Errorf("failed to change ownership for %s: %w", configPath, err)
 	}
+	c.EtcdHostsOutdated = false
 
 	return cfg, nil
 }

server/internal/orchestrator/common/patroni_config_internal_test.go

@@ -0,0 +1,71 @@
+package common
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/pgEdge/control-plane/server/internal/patroni"
+)
+
+func TestEtcdHostsOutdated(t *testing.T) {
+	configWithHosts := func(hosts ...string) *patroni.Config {
+		return &patroni.Config{
+			Etcd3: &patroni.Etcd{Hosts: &hosts},
+		}
+	}
+
+	for _, tc := range []struct {
+		name      string
+		config    *patroni.Config
+		etcdHosts []string
+		expected  bool
+	}{
+		{
+			name:      "same hosts",
+			config:    configWithHosts("host-1:2379", "host-2:2379", "host-3:2379"),
+			etcdHosts: []string{"host-1:2379", "host-2:2379", "host-3:2379"},
+			expected:  false,
+		},
+		{
+			name:      "same hosts in different order",
+			config:    configWithHosts("host-3:2379", "host-1:2379", "host-2:2379"),
+			etcdHosts: []string{"host-1:2379", "host-2:2379", "host-3:2379"},
+			expected:  false,
+		},
+		{
+			name:      "host removed",
+			config:    configWithHosts("host-1:2379", "host-2:2379", "host-3:2379", "host-4:2379"),
+			etcdHosts: []string{"host-1:2379", "host-2:2379", "host-3:2379"},
+			expected:  true,
+		},
+		{
+			name:      "host added",
+			config:    configWithHosts("host-1:2379", "host-2:2379"),
+			etcdHosts: []string{"host-1:2379", "host-2:2379", "host-3:2379"},
+			expected:  true,
+		},
+		{
+			name:      "host replaced",
+			config:    configWithHosts("host-1:2379", "host-2:2379", "host-3:2379"),
+			etcdHosts: []string{"host-1:2379", "host-2:2379", "host-4:2379"},
+			expected:  true,
+		},
+		{
+			name:      "missing etcd3 section",
+			config:    &patroni.Config{},
+			etcdHosts: []string{"host-1:2379"},
+			expected:  true,
+		},
+		{
+			name:      "missing hosts",
+			config:    &patroni.Config{Etcd3: &patroni.Etcd{}},
+			etcdHosts: []string{"host-1:2379"},
+			expected:  true,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.Equal(t, tc.expected, etcdHostsOutdated(tc.config, tc.etcdHosts))
+		})
+	}
+}

server/internal/orchestrator/swarm/patroni_config_test.go

@@ -0,0 +1,49 @@
+package swarm
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/pgEdge/control-plane/server/internal/orchestrator/common"
+	"github.com/pgEdge/control-plane/server/internal/resource"
+)
+
+func TestPatroniConfigEtcdHostsOutdatedDiff(t *testing.T) {
+	newConfig := func(outdated bool) *PatroniConfig {
+		return &PatroniConfig{
+			DatabaseID:          "database-1",
+			DatabaseNetworkName: "database-1-database",
+			Base: &common.PatroniConfig{
+				InstanceID:        "database-1-n1-host-1",
+				HostID:            "host-1",
+				NodeName:          "n1",
+				ParentID:          "database-1-n1-host-1-configs",
+				Generator:         &common.PatroniConfigGenerator{},
+				EtcdHostsOutdated: outdated,
+			},
+		}
+	}
+
+	desired, err := resource.ToResourceData(newConfig(false))
+	require.NoError(t, err)
+
+	t.Run("current", func(t *testing.T) {
+		current, err := resource.ToResourceData(newConfig(false))
+		require.NoError(t, err)
+
+		diff, err := current.Diff(desired)
+		require.NoError(t, err)
+		assert.Empty(t, diff)
+	})
+
+	t.Run("outdated", func(t *testing.T) {
+		current, err := resource.ToResourceData(newConfig(true))
+		require.NoError(t, err)
+
+		diff, err := current.Diff(desired)
+		require.NoError(t, err)
+		assert.NotEmpty(t, diff)
+	})
+}

server/internal/workflows/activities/activities.go

@@ -40,6 +40,7 @@ func (a *Activities) Register(work *worker.Worker) error {
 		work.RegisterActivity(a.PersistState),
 		work.RegisterActivity(a.PlanRefresh),
 		work.RegisterActivity(a.RemoveHost),
+		work.RegisterActivity(a.RemoveHostFromEtcd),
 		work.RegisterActivity(a.RestartInstance),
 		work.RegisterActivity(a.SelectCandidate),
 		work.RegisterActivity(a.StartInstance),

server/internal/workflows/activities/remove_host.go

@@ -14,6 +14,51 @@ import (
 	"github.com/pgEdge/control-plane/server/internal/utils"
 )
 
+type RemoveHostFromEtcdInput struct {
+	HostID string `json:"host_id"`
+}
+
+type RemoveHostFromEtcdOutput struct{}
+
+func (a *Activities) ExecuteRemoveHostFromEtcd(
+	ctx workflow.Context,
+	input *RemoveHostFromEtcdInput,
+) workflow.Future[*RemoveHostFromEtcdOutput] {
+	options := workflow.ActivityOptions{
+		Queue: utils.HostQueue(a.Config.HostID),
+		RetryOptions: workflow.RetryOptions{
+			MaxAttempts: 1,
+		},
+	}
+	return workflow.ExecuteActivity[*RemoveHostFromEtcdOutput](ctx, options, a.RemoveHostFromEtcd, input)
+}
+
+// RemoveHostFromEtcd removes the host's Etcd cluster membership, if any, and
+// its Etcd credentials.
+func (a *Activities) RemoveHostFromEtcd(ctx context.Context, input *RemoveHostFromEtcdInput) (*RemoveHostFromEtcdOutput, error) {
+	logger := activity.Logger(ctx)
+	if input == nil {
+		return nil, errors.New("input is nil")
+	}
+	logger = logger.With(
+		"host_id", input.HostID,
+	)
+	logger.Info("starting remove host from etcd activity")
+
+	etcdClient, err := do.Invoke[etcd.Etcd](a.Injector)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get etcd client: %w", err)
+	}
+
+	err = etcdClient.RemoveHost(ctx, input.HostID)
+	if err != nil {
+		return nil, fmt.Errorf("failed to remove host from etcd: %w", err)
+	}
+
+	logger.Info("remove host from etcd activity completed")
+	return &RemoveHostFromEtcdOutput{}, nil
+}
+
 type RemoveHostInput struct {
 	HostID string `json:"host_id"`
 }
@@ -33,6 +78,8 @@ func (a *Activities) ExecuteRemoveHost(
 	return workflow.ExecuteActivity[*RemoveHostOutput](ctx, options, a.RemoveHost, input)
 }
 
+// RemoveHost removes the host's record from the host service. The host must
+// already have been removed from Etcd via RemoveHostFromEtcd.
 func (a *Activities) RemoveHost(ctx context.Context, input *RemoveHostInput) (*RemoveHostOutput, error) {
 	logger := activity.Logger(ctx)
 	if input == nil {
@@ -43,21 +90,11 @@ func (a *Activities) RemoveHost(ctx context.Context, input *RemoveHostInput) (*R
 	)
 	logger.Info("starting remove host activity")
 
-	etcdClient, err := do.Invoke[etcd.Etcd](a.Injector)
-	if err != nil {
-		return nil, fmt.Errorf("failed to get etcd client: %w", err)
-	}
-
 	hostSvc, err := do.Invoke[*host.Service](a.Injector)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get host service: %w", err)
 	}
 
-	err = etcdClient.RemoveHost(ctx, input.HostID)
-	if err != nil {
-		return nil, fmt.Errorf("failed to remove host from etcd: %w", err)
-	}
-
 	err = hostSvc.RemoveHost(ctx, input.HostID)
 	if err != nil {
 		return nil, fmt.Errorf("failed to remove host from host service: %w", err)