diff --git a/server/internal/database/replication_slot_create_resource.go b/server/internal/database/replication_slot_create_resource.go index 8db34a3c..4c52f6dd 100644 --- a/server/internal/database/replication_slot_create_resource.go +++ b/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) } diff --git a/server/internal/monitor/instance_monitor.go b/server/internal/monitor/instance_monitor.go index 2385fa0f..2fb5181c 100644 --- a/server/internal/monitor/instance_monitor.go +++ b/server/internal/monitor/instance_monitor.go @@ -5,17 +5,28 @@ 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 @@ -23,6 +34,7 @@ type InstanceMonitor struct { 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 diff --git a/server/internal/patroni/gucs.go b/server/internal/patroni/gucs.go index e39cfc76..44ed7a88 100644 --- a/server/internal/patroni/gucs.go +++ b/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 diff --git a/server/internal/postgres/create_db.go b/server/internal/postgres/create_db.go index 917ef979..8dea2e77 100644 --- a/server/internal/postgres/create_db.go +++ b/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 diff --git a/server/internal/postgres/create_db_test.go b/server/internal/postgres/create_db_test.go index d0616805..4e50387c 100644 --- a/server/internal/postgres/create_db_test.go +++ b/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") +} diff --git a/server/internal/postgres/gucs.go b/server/internal/postgres/gucs.go index 58a17ece..2c44bafc 100644 --- a/server/internal/postgres/gucs.go +++ b/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", diff --git a/server/internal/postgres/gucs_test.go b/server/internal/postgres/gucs_test.go index dd7d1606..6eed458a 100644 --- a/server/internal/postgres/gucs_test.go +++ b/server/internal/postgres/gucs_test.go @@ -41,6 +41,73 @@ func TestDefaultGUCsOutputPluginLibraries(t *testing.T) { } } +func TestNeedsNativeFailoverSlots(t *testing.T) { + for _, tc := range []struct { + name string + spockMajor uint64 + pgMajor uint64 + expected bool + }{ + {name: "spock 5 pg16", spockMajor: 5, pgMajor: 16, expected: false}, + {name: "spock 5 pg17", spockMajor: 5, pgMajor: 17, expected: false}, + {name: "spock 5 pg18", spockMajor: 5, pgMajor: 18, expected: false}, + {name: "spock 6 pg16", spockMajor: 6, pgMajor: 16, expected: false}, + {name: "spock 6 pg17", spockMajor: 6, pgMajor: 17, expected: true}, + {name: "spock 6 pg18", spockMajor: 6, pgMajor: 18, expected: true}, + {name: "spock 7 pg17", spockMajor: 7, pgMajor: 17, expected: true}, + } { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.expected, postgres.NeedsNativeFailoverSlots(tc.spockMajor, tc.pgMajor)) + }) + } +} + +func TestNeedsNativeFailoverSlotsForVersion(t *testing.T) { + for _, tc := range []struct { + name string + version *ds.PgEdgeVersion + expected bool + }{ + {name: "nil version", version: nil, expected: false}, + {name: "spock 5 pg18", version: ds.MustParsePgEdgeVersion("18.4", "5"), expected: false}, + {name: "spock 6 pg16", version: ds.MustParsePgEdgeVersion("16.10", "6"), expected: false}, + {name: "spock 6 pg17", version: ds.MustParsePgEdgeVersion("17.0", "6"), expected: true}, + {name: "spock 6 pg18", version: ds.MustParsePgEdgeVersion("18.4", "6"), expected: true}, + } { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.expected, postgres.NeedsNativeFailoverSlotsForVersion(tc.version)) + }) + } +} + +func TestDefaultGUCsSyncReplicationSlots(t *testing.T) { + for _, tc := range []struct { + name string + version *ds.PgEdgeVersion + expectedPresent bool + }{ + {name: "nil version", version: nil, expectedPresent: false}, + {name: "spock 5 pg18", version: ds.MustParsePgEdgeVersion("18.4", "5"), expectedPresent: false}, + {name: "spock 6 pg16", version: ds.MustParsePgEdgeVersion("16.10", "6"), expectedPresent: false}, + {name: "spock 6 pg17", version: ds.MustParsePgEdgeVersion("17.0", "6"), expectedPresent: true}, + {name: "spock 6 pg18", version: ds.MustParsePgEdgeVersion("18.4", "6"), expectedPresent: true}, + } { + t.Run(tc.name, func(t *testing.T) { + gucs := postgres.DefaultGUCs(tc.version) + value, ok := gucs["sync_replication_slots"] + assert.Equal(t, tc.expectedPresent, ok) + if tc.expectedPresent { + assert.Equal(t, "on", value) + } + // synchronized_standby_slots is never a static default -- its + // value depends on live topology, computed at runtime instead + // (see InstanceMonitor.reconcileSynchronizedStandbySlots). + _, hasSyncSlots := gucs["synchronized_standby_slots"] + assert.False(t, hasSyncSlots) + }) + } +} + func TestDefaultTunableGUCs(t *testing.T) { for _, tc := range []struct { name string