feat: add native failover-slot support for Spock 6

client/http.go

@@ -3,11 +3,17 @@ package client
 import (
 	"net/http"
 	"net/url"
+	"time"
 
 	"github.com/pgEdge/control-plane/api/apiv1/gen/http/control_plane/client"
 	goahttp "goa.design/goa/v3/http"
 )
 
+// defaultHTTPTimeout bounds every request through an HTTPServerConfig's
+// client, matching NewMQTTServerConfig's own default maxWait -- without one,
+// an unresponsive server blocks the caller indefinitely.
+const defaultHTTPTimeout = 30 * time.Second
+
 // HTTPServerConfig configures a connection to a Control Plane server via HTTP.
 type HTTPServerConfig struct {
 	url *url.URL
@@ -24,10 +30,11 @@ func NewHTTPServerConfig(hostID string, url *url.URL) ServerConfig {
 }
 
 func (c *HTTPServerConfig) newClient() *client.Client {
+	httpClient := &http.Client{Timeout: defaultHTTPTimeout}
 	return client.NewClient(
 		c.url.Scheme,
 		c.url.Host,
-		http.DefaultClient,
+		httpClient,
 		goahttp.RequestEncoder,
 		goahttp.ResponseDecoder,
 		false,

server/internal/database/replication_slot_create_resource.go

@@ -88,7 +88,9 @@ func (r *ReplicationSlotCreateResource) Create(ctx context.Context, rc *resource
 	}
 	defer conn.Close(ctx)
 
-	stmt := postgres.CreateReplicationSlot(r.DatabaseName, r.ProviderNode, r.SubscriberNode)
+	failover := postgres.NeedsNativeFailoverSlotsForVersion(instance.Spec.PgEdgeVersion)
+
+	stmt := postgres.CreateReplicationSlot(r.DatabaseName, r.ProviderNode, r.SubscriberNode, failover)
 	if err := stmt.Exec(ctx, conn); err != nil {
 		return fmt.Errorf("failed to create replication slot: %w", err)
 	}

server/internal/database/spec.go

@@ -586,6 +586,9 @@ type InstanceSpec struct {
 	OrchestratorOpts *OrchestratorOpts `json:"orchestrator_opts,omitempty"`
 	InPlaceRestore   bool              `json:"in_place_restore,omitempty"`
 	AllHostIDs       []string          `json:"all_host_ids"` // All host IDs in the database
+	// PeerInstanceIDs are this instance's physical HA peers in the same
+	// node, excluding itself. See postgres.DefaultGUCs.
+	PeerInstanceIDs []string `json:"peer_instance_ids,omitempty"`
 }
 
 func (s *InstanceSpec) CopySettingsFrom(current *InstanceSpec) {
@@ -641,6 +644,7 @@ func (s *InstanceSpec) Clone() *InstanceSpec {
 		NodeSize:         s.NodeSize,
 		OrchestratorOpts: s.OrchestratorOpts.Clone(),
 		AllHostIDs:       slices.Clone(s.AllHostIDs),
+		PeerInstanceIDs:  slices.Clone(s.PeerInstanceIDs),
 	}
 }
 
@@ -746,6 +750,17 @@ func (s *Spec) NodeInstances() ([]*NodeInstances, error) {
 			}
 		}
 
+		// Second pass: needs every instance's InstanceID already assigned.
+		for hostIdx, instance := range instances {
+			peers := make([]string, 0, len(instances)-1)
+			for otherIdx, other := range instances {
+				if otherIdx != hostIdx {
+					peers = append(peers, other.InstanceID)
+				}
+			}
+			instance.PeerInstanceIDs = peers
+		}
+
 		nodes[nodeIdx] = &NodeInstances{
 			DatabaseID:    s.DatabaseID,
 			DatabaseOwner: owner,

server/internal/orchestrator/common/patroni_config_generator.go

@@ -60,6 +60,9 @@ type PatroniConfigGenerator struct {
 	PatroniAllowlist []string `json:"patroni_allowlist"`
 	// PatroniPort is the port that Patroni will listen on.
 	PatroniPort int `json:"patroni_port"`
+	// PeerInstanceIDs are this instance's physical HA peers in the same
+	// Spock node, used to compute synchronized_standby_slots.
+	PeerInstanceIDs []string `json:"peer_instance_ids,omitempty"`
 	// PgEdgeVersion is the Postgres/Spock version for this instance. This is
 	// used to gate version-specific default Postgres parameters.
 	PgEdgeVersion *ds.PgEdgeVersion `json:"pg_edge_version,omitempty"`
@@ -148,6 +151,7 @@ func NewPatroniConfigGenerator(opts PatroniConfigGeneratorOptions) *PatroniConfi
 		NodeSize:               opts.Instance.NodeSize,
 		OrchestratorParameters: opts.OrchestratorParameters,
 		PatroniPort:            opts.PatroniPort,
+		PeerInstanceIDs:        opts.Instance.PeerInstanceIDs,
 		PgEdgeVersion:          opts.Instance.PgEdgeVersion,
 		PostgresCertsDir:       opts.Paths.Instance.PostgresCertificates(),
 		PostgresPort:           opts.PostgresPort,
@@ -212,7 +216,7 @@ func (p *PatroniConfigGenerator) AuthMethod() hba.AuthMethod {
 }
 
 func (p *PatroniConfigGenerator) parameters() map[string]any {
-	parameters := postgres.DefaultGUCs(p.PgEdgeVersion)
+	parameters := postgres.DefaultGUCs(p.PgEdgeVersion, p.PeerInstanceIDs)
 	maps.Copy(parameters, postgres.SpockDefaultGUCs())
 	maps.Copy(parameters, postgres.DefaultTunableGUCs(p.MemoryBytes, p.CPUs, p.ClusterSize))
 	maps.Copy(parameters, map[string]any{

server/internal/postgres/create_db.go

@@ -323,11 +323,18 @@ func ReplicationSlotNeedsCreate(databaseName, providerNode, subscriberNode strin
 	}
 }
 
-func CreateReplicationSlot(databaseName, providerNode, subscriberNode string) ConditionalStatement {
+// CreateReplicationSlot creates the logical replication slot backing a peer
+// subscription. Pass failover from NeedsNativeFailoverSlots; false is
+// byte-for-byte identical to the pre-failover-slot-support statement.
+func CreateReplicationSlot(databaseName, providerNode, subscriberNode string, failover bool) ConditionalStatement {
+	sql := fmt.Sprintf("SELECT pg_create_logical_replication_slot(%s, 'spock_output');", slotNameExpr)
+	if failover {
+		sql = fmt.Sprintf("SELECT pg_create_logical_replication_slot(%s, 'spock_output', false, false, true);", slotNameExpr)
+	}
 	return ConditionalStatement{
 		If: ReplicationSlotNeedsCreate(databaseName, providerNode, subscriberNode),
 		Then: Statement{
-			SQL:  fmt.Sprintf("SELECT pg_create_logical_replication_slot(%s, 'spock_output');", slotNameExpr),
+			SQL:  sql,
 			Args: slotNameArgs(databaseName, providerNode, subscriberNode),
 		},
 	}

server/internal/postgres/create_db_test.go

@@ -94,3 +94,35 @@ func TestWaitForSyncEvent(t *testing.T) {
 		})
 	}
 }
+
+func TestCreateReplicationSlot(t *testing.T) {
+	for _, tc := range []struct {
+		name        string
+		failover    bool
+		expectedSQL string
+	}{
+		{
+			name:     "failover",
+			failover: true,
+			expectedSQL: "SELECT pg_create_logical_replication_slot(" +
+				"spock.spock_gen_slot_name(@slot_dbname, @slot_provider_node, @slot_sub_name), " +
+				"'spock_output', false, false, true);",
+		},
+		{
+			name:     "no failover",
+			failover: false,
+			expectedSQL: "SELECT pg_create_logical_replication_slot(" +
+				"spock.spock_gen_slot_name(@slot_dbname, @slot_provider_node, @slot_sub_name), " +
+				"'spock_output');",
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			slot := postgres.CreateReplicationSlot("db", "n1", "n2", tc.failover)
+			then, ok := slot.Then.(postgres.Statement)
+			if !ok {
+				t.Fatalf("expected slot.Then to be a postgres.Statement, got %T", slot.Then)
+			}
+			assert.Equal(t, tc.expectedSQL, then.SQL)
+		})
+	}
+}

server/internal/postgres/gucs.go

@@ -1,7 +1,10 @@
 package postgres
 
 import (
+	"fmt"
 	"math"
+	"sort"
+	"strings"
 
 	"github.com/pgEdge/control-plane/server/internal/ds"
 )
@@ -37,7 +40,7 @@ func needsOutputPluginLibraries(version *ds.PgEdgeVersion) bool {
 	return pgVersion.Compare(minVersion) >= 0
 }
 
-func DefaultGUCs(version *ds.PgEdgeVersion) map[string]any {
+func DefaultGUCs(version *ds.PgEdgeVersion, peerInstanceIDs []string) map[string]any {
 	gucs := map[string]any{
 		"archive_command":              "/bin/true",
 		"archive_mode":                 "on",
@@ -64,9 +67,84 @@ func DefaultGUCs(version *ds.PgEdgeVersion) map[string]any {
 	if needsOutputPluginLibraries(version) {
 		gucs["output_plugin_libraries"] = "pgoutput, test_decoding, spock_output"
 	}
+	if NeedsNativeFailoverSlotsForVersion(version) {
+		gucs["sync_replication_slots"] = "on"
+		if slots := synchronizedStandbySlots(peerInstanceIDs); slots != "" {
+			gucs["synchronized_standby_slots"] = slots
+		}
+	}
 	return gucs
 }
 
+// synchronizedStandbySlots computes this instance's synchronized_standby_slots
+// value from its peer instances' IDs, sorted for a stable result.
+func synchronizedStandbySlots(peerInstanceIDs []string) string {
+	if len(peerInstanceIDs) == 0 {
+		return ""
+	}
+	names := make([]string, len(peerInstanceIDs))
+	for i, id := range peerInstanceIDs {
+		names[i] = patroniSlotNameFromInstanceID(id)
+	}
+	sort.Strings(names)
+	return strings.Join(names, ",")
+}
+
+// patroniSlotNameFromInstanceID must exactly reproduce Patroni's own
+// slot_name_from_member_name (patroni/dcs/__init__.py), since that's what
+// actually names each peer's physical replication slot: lowercase, "-"/"."
+// become "_", anything else invalid becomes "uNNNN" (its zero-padded
+// ordinal), truncated to 63 bytes.
+func patroniSlotNameFromInstanceID(instanceID string) string {
+	var b strings.Builder
+	for _, r := range strings.ToLower(instanceID) {
+		switch {
+		case r == '-' || r == '.':
+			b.WriteByte('_')
+		case (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') || r == '_':
+			b.WriteRune(r)
+		default:
+			fmt.Fprintf(&b, "u%04d", r)
+		}
+	}
+	s := b.String()
+	if len(s) > 63 {
+		s = s[:63]
+	}
+	return s
+}
+
+// NeedsNativeFailoverSlots reports whether the given Spock and Postgres
+// major versions require PG17+'s native logical-slot-failover mechanism
+// (see DefaultGUCs). Gated on Spock major >= 6 only: 5.x had this behind an
+// opt-in GUC at times, but Control Plane never manages that GUC, so no 5.x
+// minor should trigger this.
+func NeedsNativeFailoverSlots(spockMajor, pgMajor uint64) bool {
+	return spockMajor >= 6 && pgMajor >= 17
+}
+
+// NeedsNativeFailoverSlotsForVersion is NeedsNativeFailoverSlots for a
+// declared *ds.PgEdgeVersion; false if either major is unresolvable.
+func NeedsNativeFailoverSlotsForVersion(version *ds.PgEdgeVersion) bool {
+	spockMajor, pgMajor, ok := nativeFailoverSlotMajors(version)
+	return ok && NeedsNativeFailoverSlots(spockMajor, pgMajor)
+}
+
+func nativeFailoverSlotMajors(version *ds.PgEdgeVersion) (spockMajor, pgMajor uint64, ok bool) {
+	if version == nil || version.SpockVersion == nil || version.PostgresVersion == nil {
+		return 0, 0, false
+	}
+	spockMajor, ok = version.SpockVersion.Major()
+	if !ok {
+		return 0, 0, false
+	}
+	pgMajor, ok = version.PostgresVersion.MajorMinorVersion().Major()
+	if !ok {
+		return 0, 0, false
+	}
+	return spockMajor, pgMajor, true
+}
+
 func SpockDefaultGUCs() map[string]any {
 	return map[string]any{
 		"spock.enable_ddl_replication":   "on",