feat: add native failover-slot support for Spock 6

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/monitor/instance_monitor.go

@@ -5,24 +5,36 @@ import (
 	"crypto/tls"
 	"errors"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/rs/zerolog"
 
 	"github.com/pgEdge/control-plane/server/internal/certificates"
 	"github.com/pgEdge/control-plane/server/internal/database"
+	"github.com/pgEdge/control-plane/server/internal/ds"
 	"github.com/pgEdge/control-plane/server/internal/patroni"
 	"github.com/pgEdge/control-plane/server/internal/postgres"
 	"github.com/pgEdge/control-plane/server/internal/utils"
 )
 
+// patroniRequestTimeout bounds the Patroni REST calls
+// reconcileSynchronizedStandbySlots makes. patroni.NewClient's default
+// http.Client has no request timeout of its own, and this reconciliation
+// tends to have work to do right after a failover -- exactly when
+// Patroni's REST API is most likely to be transiently unresponsive
+// (mid-election) -- so an explicit bound here keeps a stalled connection
+// from blocking this instance's status collection indefinitely.
+const patroniRequestTimeout = 10 * time.Second
+
 type InstanceMonitor struct {
 	statusMonitor *Monitor
 	databaseID    string
 	instanceID    string
 	dbName        string
 	dbSvc         *database.Service
 	certSvc       *certificates.Service
+	logger        zerolog.Logger
 }
 
 func NewInstanceMonitor(
@@ -39,6 +51,7 @@ func NewInstanceMonitor(
 		dbName:     dbName,
 		dbSvc:      dbSvc,
 		certSvc:    certSvc,
+		logger:     logger,
 	}
 	m.statusMonitor = NewMonitor(
 		logger,
@@ -175,6 +188,126 @@ func (m *InstanceMonitor) populateFromDbConn(
 				Status:       sub.Status,
 			})
 		}
+
+		// Logged and swallowed rather than propagated: this is a
+		// best-effort background reconciliation, not a health signal.
+		// Letting it fail the whole status collection here would report
+		// an otherwise-healthy primary as errored (discarding the
+		// version/subscription data this same pass already collected)
+		// over what's usually a transient Patroni REST hiccup -- and
+		// since the reconciliation is self-correcting on every poll (see
+		// its own doc comment), nothing is lost by retrying next tick
+		// instead of surfacing this as an instance-level error now.
+		if err := m.reconcileSynchronizedStandbySlots(ctx, conn, info, pgVersion, spockVersion); err != nil {
+			m.logger.Err(err).
+				Str("database_id", m.databaseID).
+				Str("instance_id", m.instanceID).
+				Msg("failed to reconcile synchronized_standby_slots")
+		}
+	}
+
+	return nil
+}
+
+// reconcileSynchronizedStandbySlots keeps Postgres 17+'s
+// synchronized_standby_slots GUC in sync with this node's actual current
+// physical standby topology, on the current primary only. This is what
+// makes native failover slots (see postgres.NeedsNativeFailoverSlots)
+// safe to fail over onto: without it, a promoted replica's logical slots
+// have no guarantee the outgoing primary's not-yet-decoded WAL was ever
+// received by the physical standby that just became primary.
+//
+// This runs here, in the same 5s poll that already detects a role change
+// (rather than e.g. a Patroni on_role_change callback), because it's the
+// one thing in this codebase that already knows a role change happened
+// -- Control Plane's own spec-driven reconciliation never runs on its
+// own initiative when Patroni autonomously promotes a replica, and
+// wiring a callback into the Postgres/Patroni container image would be
+// new plumbing (script delivery, auth back to Control Plane) with no
+// existing precedent anywhere in this codebase. See the design doc for
+// the fuller comparison.
+//
+// Deliberately idempotent and self-correcting rather than cached: it
+// re-derives the desired value and compares against the GUC's own live
+// setting on every call, so a prior partial failure (e.g. the DCS patch
+// below succeeds but the reload doesn't) is retried on the very next
+// tick rather than silently stuck behind an in-memory "already handled"
+// flag.
+func (m *InstanceMonitor) reconcileSynchronizedStandbySlots(
+	ctx context.Context,
+	conn postgres.Executor,
+	info *database.ConnectionInfo,
+	pgVersionStr, spockVersionStr string,
+) error {
+	// A malformed version string here isn't treated as an error condition
+	// -- it fails open to "not eligible," the same way
+	// needsOutputPluginLibraries and nativeFailoverSlotMajors (see
+	// postgres/gucs.go) treat an unresolvable version as "doesn't need
+	// this" rather than an error. In practice this path is unreachable:
+	// pgVersionStr/spockVersionStr come from GetPostgresVersion()/
+	// GetSpockVersion(), which only ever produce clean, well-formed
+	// version strings for a live connection.
+	pgVersion, err := ds.ParseVersion(pgVersionStr)
+	if err != nil {
+		return nil
+	}
+	spockVersion, err := ds.ParseVersion(spockVersionStr)
+	if err != nil {
+		return nil
+	}
+	pgMajor, ok := pgVersion.Major()
+	if !ok {
+		return nil
+	}
+	spockMajor, ok := spockVersion.Major()
+	if !ok {
+		return nil
+	}
+	if !postgres.NeedsNativeFailoverSlots(spockMajor, pgMajor) {
+		// Not a native-failover-slot cluster (e.g. Spock 5.x, or PG < 17)
+		// -- leave synchronized_standby_slots alone entirely. Its
+		// Postgres default is an empty string (no synchronization
+		// requirement), so there's nothing to reconcile toward.
+		return nil
+	}
+
+	slotNames, err := postgres.PhysicalReplicationSlotNames().Scalars(ctx, conn)
+	if err != nil {
+		return fmt.Errorf("failed to list physical replication slots: %w", err)
+	}
+	desired := strings.Join(slotNames, ",")
+
+	current, err := postgres.CurrentSynchronizedStandbySlots().Scalar(ctx, conn)
+	if err != nil {
+		return fmt.Errorf("failed to read current synchronized_standby_slots: %w", err)
+	}
+	if current == desired {
+		return nil
+	}
+
+	// Bounded independently of the caller's context: this reconciliation
+	// is most likely to have work to do right after a failover, which is
+	// exactly when Patroni's own REST API is most likely to be
+	// transiently unresponsive (mid-election). http.DefaultClient (what
+	// patroni.NewClient falls back to) has no request timeout of its
+	// own, so without this a stalled connection here could block this
+	// instance's status collection well past its usual 5s cadence.
+	patchCtx, cancel := context.WithTimeout(ctx, patroniRequestTimeout)
+	defer cancel()
+
+	client := patroni.NewClient(info.PatroniURL(), nil)
+	_, err = client.PatchDynamicConfig(patchCtx, &patroni.DynamicConfig{
+		PostgreSQL: &patroni.DynamicPostgreSQLConfig{
+			Parameters: utils.PointerTo(map[string]any{
+				"synchronized_standby_slots": desired,
+			}),
+		},
+	})
+	if err != nil {
+		return fmt.Errorf("failed to patch synchronized_standby_slots to %q: %w", desired, err)
+	}
+	if err := client.Reload(patchCtx); err != nil {
+		return fmt.Errorf("failed to reload after patching synchronized_standby_slots: %w", err)
 	}
 
 	return nil

server/internal/patroni/gucs.go

@@ -16,6 +16,14 @@ var dynamicGUCs = ds.NewSet(
 	"max_replication_slots",
 	"wal_keep_segments",
 	"wal_keep_size",
+	// Never generated as a static default (see postgres.DefaultGUCs' doc
+	// comment) -- its correct value depends on live replication topology,
+	// so it's only ever pushed here directly via the Patroni REST client's
+	// PatchDynamicConfig, by InstanceMonitor's runtime reconciliation (see
+	// server/internal/monitor/instance_monitor.go). Listed here purely so
+	// this file stays the one place documenting every GUC this codebase
+	// manages through Patroni's DCS, reload-safe.
+	"synchronized_standby_slots",
 )
 
 // ExtractPatroniControlledGUCs extracts the GUCs that Patroni controls into a

server/internal/postgres/create_db.go

@@ -323,11 +323,26 @@ func ReplicationSlotNeedsCreate(databaseName, providerNode, subscriberNode strin
 	}
 }
 
-func CreateReplicationSlot(databaseName, providerNode, subscriberNode string) ConditionalStatement {
+// CreateReplicationSlot creates the logical replication slot backing a
+// peer subscription. failover should be true only when
+// postgres.NeedsNativeFailoverSlots reports the managed database's Spock
+// and Postgres majors both require it -- when false, the statement is
+// byte-for-byte identical to the pre-failover-slot-support form, so
+// clusters that don't need this see no behavior change at all.
+// pg_create_logical_replication_slot's failover parameter was only added
+// in PG17, which is exactly the same version floor
+// NeedsNativeFailoverSlots already requires, so there's no separate PG
+// major check needed here -- failover=true never happens on an older
+// Postgres where the 5-arg form wouldn't exist.
+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),
 		},
 	}
@@ -408,6 +423,42 @@ func ReplicationSlotExists(databaseName, providerNode, subscriberNode string) Qu
 	}
 }
 
+// PhysicalReplicationSlotNames lists every permanent (non-temporary)
+// physical replication slot currently on this instance -- i.e. the slots
+// backing this node's own physical (Patroni-managed HA) standbys, as
+// distinct from the logical spock_output slots backing peer
+// subscriptions. Used to compute synchronized_standby_slots: Patroni
+// creates and names these itself (permanent member slots, PG11+'s
+// use_slots), Control Plane never creates or names a physical slot
+// directly, so the live catalog is the only source of truth for "which
+// slot names exist right now" -- there's no Go-side naming convention to
+// reproduce instead.
+//
+// Temporary slots are deliberately excluded: they're scoped to whatever
+// session created them (e.g. a one-off basebackup helper bootstrapping a
+// new replica) and vanish the moment that session ends. Including one
+// here could reference a slot name in synchronized_standby_slots that's
+// already gone by the time Postgres reloads -- not harmful (Postgres
+// treats a missing slot name as simply never satisfied, not an error),
+// but pointless churn that a temporary slot, by definition, was never
+// meant to be depended on for.
+func PhysicalReplicationSlotNames() Query[string] {
+	return Query[string]{
+		SQL: "SELECT slot_name FROM pg_replication_slots WHERE slot_type = 'physical' AND NOT temporary ORDER BY slot_name;",
+	}
+}
+
+// CurrentSynchronizedStandbySlots reports the live value of Postgres 17+'s
+// synchronized_standby_slots GUC on this connection, as a plain
+// comma-separated string (its raw on-disk/runtime representation) --
+// used by InstanceMonitor to detect drift from the desired value without
+// forcing an unconditional reload on every check.
+func CurrentSynchronizedStandbySlots() Query[string] {
+	return Query[string]{
+		SQL: "SELECT setting FROM pg_settings WHERE name = 'synchronized_standby_slots';",
+	}
+}
+
 func GetReplicationSlotLSNFromCommitTS(databaseName, providerNode, subscriberNode string, commitTS time.Time) Query[string] {
 	args := slotNameArgs(databaseName, providerNode, subscriberNode)
 	args["commit_ts"] = commitTS

server/internal/postgres/create_db_test.go

@@ -94,3 +94,42 @@ 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)
+		})
+	}
+}
+
+func TestPhysicalReplicationSlotNames(t *testing.T) {
+	query := postgres.PhysicalReplicationSlotNames()
+	assert.Contains(t, query.SQL, "slot_type = 'physical'")
+	assert.Contains(t, query.SQL, "NOT temporary",
+		"must exclude temporary slots -- they vanish with their creating session and shouldn't be depended on")
+}

server/internal/postgres/gucs.go

@@ -64,9 +64,64 @@ func DefaultGUCs(version *ds.PgEdgeVersion) map[string]any {
 	if needsOutputPluginLibraries(version) {
 		gucs["output_plugin_libraries"] = "pgoutput, test_decoding, spock_output"
 	}
+	if NeedsNativeFailoverSlotsForVersion(version) {
+		// synchronized_standby_slots is deliberately NOT set here even
+		// though the gate passed: its correct value is the current set of
+		// physical standby slot names, which isn't knowable at
+		// config-generation/bootstrap time (no instances exist yet, let
+		// alone standbys) and changes over the node's lifetime as replicas
+		// are added/removed or a failover promotes a different primary.
+		// That value is instead computed from live replication state and
+		// kept in sync at runtime — see
+		// server/internal/monitor/instance_monitor.go.
+		gucs["sync_replication_slots"] = "on"
+	}
 	return gucs
 }
 
+// NeedsNativeFailoverSlots reports whether the given Spock and Postgres
+// major versions require PG17+'s native logical-slot-failover mechanism:
+// replication slots created with failover => true, plus
+// sync_replication_slots (see DefaultGUCs) and synchronized_standby_slots
+// (see server/internal/monitor/instance_monitor.go) kept in sync.
+//
+// The FAILOVER-flag history this needs to account for is more specific
+// than "5.x doesn't have it, 6.0 does": 5.0.7 had it on unconditionally,
+// 5.0.8-5.0.10 removed it entirely, 5.0.11 brought it back opt-in behind
+// spock.use_native_failover_slots (default off), and 6.0.0 made it
+// unconditional again with that GUC removed. Deliberately gated on Spock
+// major >= 6 only, never on any 5.x minor (including 5.0.11's opt-in
+// GUC) — existing 5.x deployments must see zero behavior change from
+// this, and Control Plane doesn't manage that opt-in GUC either way.
+func NeedsNativeFailoverSlots(spockMajor, pgMajor uint64) bool {
+	return spockMajor >= 6 && pgMajor >= 17
+}
+
+// NeedsNativeFailoverSlotsForVersion is NeedsNativeFailoverSlots for
+// callers that have a declared *ds.PgEdgeVersion on hand (e.g. an
+// instance's spec) rather than already-extracted major versions. Returns
+// false for a nil version or either major being unresolvable, matching
+// how the rest of this file treats an unknown/unparseable version.
+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",