feat: use common resources in swarm

changes/unreleased/Fixed-20260611-090000.yaml

@@ -1,3 +1,3 @@
 kind: Fixed
-body: Fixed global Patroni parameters not being applied to running systemd-managed database clusters — Dynamic configuration changes are now patched directly via the Patroni API on the primary instance, ensuring parameters that cannot be changed via config file reload take effect immediately.
+body: Fixed global Patroni parameters not being applied to running database clusters. Dynamic configuration changes are now patched directly via the Patroni API on the primary instance, ensuring parameters that cannot be changed via config file reload take effect immediately.
 time: 2026-06-11T09:00:00.000000-04:00

e2e/database_test.go

@@ -262,6 +262,18 @@ func (d *DatabaseFixture) WithConnection(ctx context.Context, opts ConnectionOpt
 	}
 }
 
+func (d *DatabaseFixture) HeartBeat(t testing.TB, username, password string, nodeNames ...string) {
+	if len(nodeNames) == 0 {
+		for _, node := range d.Spec.Nodes {
+			RunHeartBeat(t, d, username, password, node.Name)
+		}
+	} else {
+		for _, nodeName := range nodeNames {
+			RunHeartBeat(t, d, username, password, nodeName)
+		}
+	}
+}
+
 type InstanceMatcher func(inst *controlplane.Instance) bool
 
 func WithID(instanceID string) InstanceMatcher {

e2e/db_update_add_node_test.go

@@ -9,9 +9,10 @@ import (
 	"time"
 
 	"github.com/jackc/pgx/v5"
-	controlplane "github.com/pgEdge/control-plane/api/apiv1/gen/control_plane"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+
+	controlplane "github.com/pgEdge/control-plane/api/apiv1/gen/control_plane"
 )
 
 func TestUpdateAddNode23(t *testing.T) {
@@ -28,7 +29,7 @@ func testUpdateAddNode(t *testing.T, nodeCount int, deployReplicas bool) {
 	username := "admin"
 	password := "password"
 
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+	ctx, cancel := context.WithTimeout(context.Background(), 8*time.Minute)
 	defer cancel()
 
 	logDeploymentConfig(t, nodeCount, deployReplicas)
@@ -86,7 +87,7 @@ func buildNodeSpecs(nodeCount int) []*controlplane.DatabaseNodeSpec {
 
 func createDatabaseFixture(ctx context.Context, t *testing.T, username, password string, nodes []*controlplane.DatabaseNodeSpec) *DatabaseFixture {
 	t.Helper()
-	return fixture.NewDatabaseFixture(ctx, t, &controlplane.CreateDatabaseRequest{
+	db := fixture.NewDatabaseFixture(ctx, t, &controlplane.CreateDatabaseRequest{
 		Spec: &controlplane.DatabaseSpec{
 			DatabaseName: "test_db_create",
 			DatabaseUsers: []*controlplane.DatabaseUserSpec{{
@@ -100,6 +101,11 @@ func createDatabaseFixture(ctx context.Context, t *testing.T, username, password
 			Nodes:       nodes,
 		},
 	})
+	// The peer sync and replication slot advance steps can take an _extremely_
+	// long time in an idle database. This heartbeat provides consistent
+	// activity and makes the test run faster.
+	db.HeartBeat(t, username, password, nodes[0].Name)
+	return db
 }
 
 func verifyPrimaryNode(ctx context.Context, t *testing.T, db *DatabaseFixture, node *controlplane.DatabaseNodeSpec, username, password string, deployReplicas bool) {

e2e/heartbeat_test.go

@@ -0,0 +1,113 @@
+//go:build e2e_test
+
+package e2e
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/stretchr/testify/require"
+)
+
+// HeartBeat is a helper to create a low, consistent volume of activity on a
+// node to avoid replication stalls and waits.
+type HeartBeat struct {
+	DB       *DatabaseFixture
+	NodeName string
+	Username string
+	Password string
+	wg       sync.WaitGroup
+	cancel   context.CancelFunc
+	done     chan struct{}
+}
+
+// RunHeartBeat starts a heartbeat for the given node. The heartbeat is
+// automatically stopped during test cleanup.
+func RunHeartBeat(t testing.TB, db *DatabaseFixture, username, password, nodeName string) {
+	t.Helper()
+
+	heartbeat := &HeartBeat{
+		DB:       db,
+		NodeName: nodeName,
+		Username: username,
+		Password: password,
+	}
+	heartbeat.Start(t)
+	t.Cleanup(func() {
+		heartbeat.Stop(t)
+	})
+}
+
+func (l *HeartBeat) Start(t testing.TB) {
+	t.Helper()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	l.cancel = cancel
+	l.done = make(chan struct{})
+	l.wg.Go(func() {
+		defer l.cancel()
+
+		l.DB.WithConnection(ctx, l.connOpts(), t, func(conn *pgx.Conn) {
+			// Create table if not exists
+			l.createTable(ctx, t, conn)
+
+			// Persist heartbeat until Stop() is called
+			l.heartbeat(ctx, t, conn)
+		})
+	})
+}
+
+func (l *HeartBeat) Stop(t testing.TB) {
+	tLogf(t, "HeartBeat: stopping heartbeat on primary instance of node %s", l.NodeName)
+
+	close(l.done)
+	l.wg.Wait()
+}
+
+func (l *HeartBeat) connOpts() ConnectionOptions {
+	return ConnectionOptions{
+		Matcher: And(
+			WithNode(l.NodeName),
+			WithRole("primary"),
+		),
+		Username: l.Username,
+		Password: l.Password,
+	}
+}
+
+func (l *HeartBeat) createTable(ctx context.Context, t testing.TB, conn *pgx.Conn) {
+	sql := `CREATE TABLE IF NOT EXISTS heartbeat (
+		node_name TEXT PRIMARY KEY,
+		updated_at TIMESTAMPTZ
+	);`
+
+	_, err := conn.Exec(ctx, sql)
+	require.NoError(t, err)
+}
+
+func (l *HeartBeat) heartbeat(ctx context.Context, t testing.TB, conn *pgx.Conn) {
+	tLogf(t, "HeartBeat: starting heartbeat on primary instance of node %s", l.NodeName)
+
+	sql := `INSERT INTO heartbeat (node_name, updated_at)
+			VALUES (@node_name, now())
+			ON CONFLICT (node_name)
+			DO UPDATE SET updated_at = EXCLUDED.updated_at;`
+
+	ticker := time.NewTicker(500 * time.Millisecond)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-l.done:
+			return
+		case <-ticker.C:
+			_, err := conn.Exec(ctx, sql, pgx.NamedArgs{"node_name": l.NodeName})
+			require.NoError(t, err)
+		}
+	}
+}

server/internal/database/instance_resource.go

@@ -16,7 +16,6 @@ import (
 	"github.com/pgEdge/control-plane/server/internal/patroni"
 	"github.com/pgEdge/control-plane/server/internal/postgres"
 	"github.com/pgEdge/control-plane/server/internal/resource"
-	"github.com/pgEdge/control-plane/server/internal/utils"
 )
 
 var _ resource.Resource = (*InstanceResource)(nil)
@@ -213,15 +212,6 @@ func (r *InstanceResource) initializeInstance(ctx context.Context, rc *resource.
 		return nil
 	}
 
-	// Enable failsafe mode if this instance is the only one in the node.
-	// Otherwise, disable it.
-	_, err := r.patroniClient().PatchDynamicConfig(ctx, &patroni.DynamicConfig{
-		FailsafeMode: utils.PointerTo(r.Spec.NodeSize == 1),
-	})
-	if err != nil {
-		return fmt.Errorf("failed to configure failsafe mode: %w", err)
-	}
-
 	conn, err := r.Connection(ctx, rc, "postgres")
 	if err != nil {
 		return err

server/internal/docker/docker.go

@@ -601,10 +601,10 @@ func (d *Docker) ensureDockerImage(ctx context.Context, img string) error {
 }
 
 type NetworkInfo struct {
-	Name    string
-	ID      string
-	Subnet  netip.Prefix
-	Gateway netip.Addr
+	Name    string       `json:"name"`
+	ID      string       `json:"id"`
+	Subnet  netip.Prefix `json:"subnet"`
+	Gateway netip.Addr   `json:"gateway"`
 }
 
 func ExtractNetworkInfo(info network.Inspect) (*NetworkInfo, error) {

server/internal/orchestrator/common/patroni_config.go

@@ -16,6 +16,7 @@ import (
 	"github.com/pgEdge/control-plane/server/internal/database"
 	"github.com/pgEdge/control-plane/server/internal/filesystem"
 	"github.com/pgEdge/control-plane/server/internal/patroni"
+	"github.com/pgEdge/control-plane/server/internal/pgbackrest"
 	"github.com/pgEdge/control-plane/server/internal/postgres/hba"
 	"github.com/pgEdge/control-plane/server/internal/resource"
 )
@@ -37,6 +38,12 @@ func (c *PatroniConfig) Dependencies() []resource.Identifier {
 		PatroniMemberResourceIdentifier(c.InstanceID),
 		PatroniClusterResourceIdentifier(c.NodeName),
 	}
+	if c.Generator.ArchiveCommand != "" {
+		deps = append(deps, PgBackRestConfigIdentifier(c.InstanceID, pgbackrest.ConfigTypeBackup))
+	}
+	if c.Generator.RestoreCommand != "" {
+		deps = append(deps, PgBackRestConfigIdentifier(c.InstanceID, pgbackrest.ConfigTypeRestore))
+	}
 	return deps
 }
 

server/internal/orchestrator/common/patroni_config_generator.go

@@ -195,6 +195,16 @@ func (p *PatroniConfigGenerator) Generate(
 	}
 }
 
+func (p *PatroniConfigGenerator) AuthMethod() hba.AuthMethod {
+	encryption, ok := utils.TypedFromMap[string](p.parameters(), "password_encryption")
+	if !ok {
+		// ScramSHA256 is the default password encryption for all of our
+		// supported Postgres versions.
+		return hba.AuthMethodScramSHA256
+	}
+	return hba.AuthMethod(encryption)
+}
+
 func (p *PatroniConfigGenerator) parameters() map[string]any {
 	parameters := postgres.DefaultGUCs()
 	maps.Copy(parameters, postgres.SpockDefaultGUCs())
@@ -319,12 +329,8 @@ func (p *PatroniConfigGenerator) postgreSQL(
 	}
 
 	// The catch-all authenticates non-system users by password, so its auth
-	// method follows password_encryption to match how passwords are stored. It
-	// defaults to md5 (our DefaultGUCs value) when unset.
-	passwordAuthMethod := hba.AuthMethodMD5
-	if pe, ok := parameters["password_encryption"].(string); ok && pe != "" {
-		passwordAuthMethod = hba.AuthMethod(pe)
-	}
+	// method follows password_encryption to match how passwords are stored.
+	passwordAuthMethod := p.AuthMethod()
 
 	return &patroni.PostgreSQL{
 		ConnectAddress:                         utils.PointerTo(net.JoinHostPort(p.FQDN, strconv.Itoa(p.PostgresPort))),