From 5b53dd06bf60c2360653352a531dd3a7b274ab7b Mon Sep 17 00:00:00 2001 From: moizpgedge Date: Thu, 20 Aug 2026 19:35:34 +0500 Subject: [PATCH 1/3] feat: add native failover-slot support for Spock 6 Implements PG17+'s native logical-slot-failover mechanism for Spock 6 clusters, fully gated so Spock 5.x deployments see zero behavior change: - postgres.NeedsNativeFailoverSlots gates all of the below behind Spock major >= 6 && Postgres major >= 17. - CreateReplicationSlot takes a failover bool; the SQL is byte-identical to before when false. - NativeFailoverSlotGUCs sets sync_replication_slots = on as a static, spec-known default via PatroniConfigGenerator. - InstanceMonitor.reconcileSynchronizedStandbySlots keeps synchronized_standby_slots in sync with each node's actual live physical standby slots, on the primary only. Self-correcting on every 5s poll rather than cached, so a partial failure (DCS patch succeeds, reload doesn't) retries on the next tick instead of getting stuck. Temporary slots are excluded, since they vanish with their creating session and were never meant to be depended on. synchronized_standby_slots' maintenance is Control-Plane-side monitoring rather than a Patroni on_role_change callback: the live-GUC-push mechanism and role-polling this needs already exist in this codebase, while a callback would need new script-delivery and auth plumbing into the Postgres/Patroni container with no existing precedent. Full comparison in ADR-0003 (internal-design-docs). Verified live end to end, including two independent real Patroni failovers on separate hosts with roles reversed between runs: synchronized_standby_slots was already correct for the new primary by the time each failover task completed, and replication continued in both directions through the transition. PLAT-719 --- .../replication_slot_create_resource.go | 4 +- server/internal/monitor/instance_monitor.go | 92 +++++++++++++++++++ .../common/patroni_config_generator.go | 1 + server/internal/patroni/gucs.go | 14 +++ server/internal/postgres/create_db.go | 55 ++++++++++- server/internal/postgres/create_db_test.go | 39 ++++++++ server/internal/postgres/gucs.go | 64 +++++++++++++ server/internal/postgres/gucs_test.go | 67 ++++++++++++++ 8 files changed, 333 insertions(+), 3 deletions(-) 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..c3a853fc 100644 --- a/server/internal/monitor/instance_monitor.go +++ b/server/internal/monitor/instance_monitor.go @@ -5,12 +5,14 @@ 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" @@ -175,6 +177,96 @@ func (m *InstanceMonitor) populateFromDbConn( Status: sub.Status, }) } + + if err := m.reconcileSynchronizedStandbySlots(ctx, conn, info, pgVersion, spockVersion); err != nil { + return fmt.Errorf("failed to reconcile synchronized_standby_slots: %w", err) + } + } + + 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 { + pgVersion, err := ds.ParseVersion(pgVersionStr) + if err != nil { + return fmt.Errorf("failed to parse postgres version %q: %w", pgVersionStr, err) + } + spockVersion, err := ds.ParseVersion(spockVersionStr) + if err != nil { + return fmt.Errorf("failed to parse spock version %q: %w", spockVersionStr, err) + } + pgMajor, ok := pgVersion.Major() + if !ok { + return fmt.Errorf("failed to determine postgres major version from %q", pgVersionStr) + } + spockMajor, ok := spockVersion.Major() + if !ok { + return fmt.Errorf("failed to determine spock major version from %q", spockVersionStr) + } + 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 + } + + client := patroni.NewClient(info.PatroniURL(), nil) + _, err = client.PatchDynamicConfig(ctx, &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(ctx); err != nil { + return fmt.Errorf("failed to reload after patching synchronized_standby_slots: %w", err) } return nil diff --git a/server/internal/orchestrator/common/patroni_config_generator.go b/server/internal/orchestrator/common/patroni_config_generator.go index 7735a3c8..1701dfb0 100644 --- a/server/internal/orchestrator/common/patroni_config_generator.go +++ b/server/internal/orchestrator/common/patroni_config_generator.go @@ -232,6 +232,7 @@ func (p *PatroniConfigGenerator) parameters() map[string]any { }) } maps.Copy(parameters, postgres.SnowflakeLolorGUCs(p.NodeOrdinal)) + maps.Copy(parameters, postgres.NativeFailoverSlotGUCs(p.PgEdgeVersion)) maps.Copy(parameters, p.SpecParameters) return parameters diff --git a/server/internal/patroni/gucs.go b/server/internal/patroni/gucs.go index e39cfc76..2119d53d 100644 --- a/server/internal/patroni/gucs.go +++ b/server/internal/patroni/gucs.go @@ -16,6 +16,20 @@ var dynamicGUCs = ds.NewSet( "max_replication_slots", "wal_keep_segments", "wal_keep_size", + // Reload-safe, kept identical across every instance in the node via + // DCS rather than each instance's own static config -- see + // postgres.NativeFailoverSlotGUCs. Set once at config-generation time + // for every instance regardless of current role, since Patroni can + // promote any of them to primary later. + "sync_replication_slots", + // Never generated as a static default (see NativeFailoverSlotGUCs' + // 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..5212a913 100644 --- a/server/internal/postgres/gucs.go +++ b/server/internal/postgres/gucs.go @@ -67,6 +67,70 @@ func DefaultGUCs(version *ds.PgEdgeVersion) map[string]any { 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 and synchronized_standby_slots kept in sync via +// Patroni (see NativeFailoverSlotGUCs and +// server/internal/monitor/instance_monitor.go respectively). +// +// 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 +} + +// NativeFailoverSlotGUCs returns the static, spec-known GUCs needed once a +// database's declared version crosses the NeedsNativeFailoverSlots gate. +// +// synchronized_standby_slots is deliberately NOT set here even when the +// gate passes: 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. +func NativeFailoverSlotGUCs(version *ds.PgEdgeVersion) map[string]any { + if !NeedsNativeFailoverSlotsForVersion(version) { + return map[string]any{} + } + return map[string]any{ + "sync_replication_slots": "on", + } +} + 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..fff0de82 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 TestNativeFailoverSlotGUCs(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.NativeFailoverSlotGUCs(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 From 655704c72fb6d71625277d63537257f16e0293ec Mon Sep 17 00:00:00 2001 From: moizpgedge Date: Fri, 21 Aug 2026 00:21:08 +0500 Subject: [PATCH 2/3] fix: don't let synchronized_standby_slots checks flip instance health reconcileSynchronizedStandbySlots errors were propagating all the way out of populateFromDbConn, so a transient Patroni REST hiccup (most likely to happen right after a failover, exactly when this reconciliation has real work to do) would report an otherwise healthy primary as errored and discard the version/subscription data the same pass had already collected. Since the reconciliation is self-correcting on every 5s poll, nothing is lost by logging and retrying instead of surfacing it as an instance-level error. Also: - Fail open (return nil) on an unparseable version string, matching needsOutputPluginLibraries/nativeFailoverSlotMajors' existing convention of treating an unresolvable version as "not eligible" rather than an error. Unreachable in practice today, since the version strings here always come from a live, successful query. - Bound the Patroni PatchDynamicConfig/Reload calls with their own 10s timeout. patroni.NewClient's default http.Client has no request timeout of its own, so a stalled connection during a Patroni election could otherwise block this instance's status collection indefinitely. Verified live against a real Patroni failover (Lima fixture, two separate hosts): confirmed instance health stayed available/no-error on both instances throughout the transition, and re-ran every other PLAT-719 scenario from scratch with these changes applied with no regressions. PLAT-719 --- server/internal/monitor/instance_monitor.go | 55 ++++++++++++++++++--- 1 file changed, 48 insertions(+), 7 deletions(-) diff --git a/server/internal/monitor/instance_monitor.go b/server/internal/monitor/instance_monitor.go index c3a853fc..2fb5181c 100644 --- a/server/internal/monitor/instance_monitor.go +++ b/server/internal/monitor/instance_monitor.go @@ -18,6 +18,15 @@ import ( "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 @@ -25,6 +34,7 @@ type InstanceMonitor struct { dbName string dbSvc *database.Service certSvc *certificates.Service + logger zerolog.Logger } func NewInstanceMonitor( @@ -41,6 +51,7 @@ func NewInstanceMonitor( dbName: dbName, dbSvc: dbSvc, certSvc: certSvc, + logger: logger, } m.statusMonitor = NewMonitor( logger, @@ -178,8 +189,20 @@ func (m *InstanceMonitor) populateFromDbConn( }) } + // 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 { - return fmt.Errorf("failed to reconcile synchronized_standby_slots: %w", err) + m.logger.Err(err). + Str("database_id", m.databaseID). + Str("instance_id", m.instanceID). + Msg("failed to reconcile synchronized_standby_slots") } } @@ -216,21 +239,29 @@ func (m *InstanceMonitor) reconcileSynchronizedStandbySlots( 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 fmt.Errorf("failed to parse postgres version %q: %w", pgVersionStr, err) + return nil } spockVersion, err := ds.ParseVersion(spockVersionStr) if err != nil { - return fmt.Errorf("failed to parse spock version %q: %w", spockVersionStr, err) + return nil } pgMajor, ok := pgVersion.Major() if !ok { - return fmt.Errorf("failed to determine postgres major version from %q", pgVersionStr) + return nil } spockMajor, ok := spockVersion.Major() if !ok { - return fmt.Errorf("failed to determine spock major version from %q", spockVersionStr) + return nil } if !postgres.NeedsNativeFailoverSlots(spockMajor, pgMajor) { // Not a native-failover-slot cluster (e.g. Spock 5.x, or PG < 17) @@ -254,8 +285,18 @@ func (m *InstanceMonitor) reconcileSynchronizedStandbySlots( 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(ctx, &patroni.DynamicConfig{ + _, err = client.PatchDynamicConfig(patchCtx, &patroni.DynamicConfig{ PostgreSQL: &patroni.DynamicPostgreSQLConfig{ Parameters: utils.PointerTo(map[string]any{ "synchronized_standby_slots": desired, @@ -265,7 +306,7 @@ func (m *InstanceMonitor) reconcileSynchronizedStandbySlots( if err != nil { return fmt.Errorf("failed to patch synchronized_standby_slots to %q: %w", desired, err) } - if err := client.Reload(ctx); err != nil { + if err := client.Reload(patchCtx); err != nil { return fmt.Errorf("failed to reload after patching synchronized_standby_slots: %w", err) } From 91966ceda149f0e362714cb8d534af18b9533f38 Mon Sep 17 00:00:00 2001 From: moizpgedge Date: Fri, 21 Aug 2026 23:41:36 +0500 Subject: [PATCH 3/3] fix: move sync_replication_slots handling into DefaultGUCs --- .../common/patroni_config_generator.go | 1 - server/internal/patroni/gucs.go | 20 ++++------ server/internal/postgres/gucs.go | 37 +++++++------------ server/internal/postgres/gucs_test.go | 4 +- 4 files changed, 23 insertions(+), 39 deletions(-) diff --git a/server/internal/orchestrator/common/patroni_config_generator.go b/server/internal/orchestrator/common/patroni_config_generator.go index 1701dfb0..7735a3c8 100644 --- a/server/internal/orchestrator/common/patroni_config_generator.go +++ b/server/internal/orchestrator/common/patroni_config_generator.go @@ -232,7 +232,6 @@ func (p *PatroniConfigGenerator) parameters() map[string]any { }) } maps.Copy(parameters, postgres.SnowflakeLolorGUCs(p.NodeOrdinal)) - maps.Copy(parameters, postgres.NativeFailoverSlotGUCs(p.PgEdgeVersion)) maps.Copy(parameters, p.SpecParameters) return parameters diff --git a/server/internal/patroni/gucs.go b/server/internal/patroni/gucs.go index 2119d53d..44ed7a88 100644 --- a/server/internal/patroni/gucs.go +++ b/server/internal/patroni/gucs.go @@ -16,19 +16,13 @@ var dynamicGUCs = ds.NewSet( "max_replication_slots", "wal_keep_segments", "wal_keep_size", - // Reload-safe, kept identical across every instance in the node via - // DCS rather than each instance's own static config -- see - // postgres.NativeFailoverSlotGUCs. Set once at config-generation time - // for every instance regardless of current role, since Patroni can - // promote any of them to primary later. - "sync_replication_slots", - // Never generated as a static default (see NativeFailoverSlotGUCs' - // 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. + // 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", ) diff --git a/server/internal/postgres/gucs.go b/server/internal/postgres/gucs.go index 5212a913..2c44bafc 100644 --- a/server/internal/postgres/gucs.go +++ b/server/internal/postgres/gucs.go @@ -64,15 +64,26 @@ 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 and synchronized_standby_slots kept in sync via -// Patroni (see NativeFailoverSlotGUCs and -// server/internal/monitor/instance_monitor.go respectively). +// 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, @@ -111,26 +122,6 @@ func nativeFailoverSlotMajors(version *ds.PgEdgeVersion) (spockMajor, pgMajor ui return spockMajor, pgMajor, true } -// NativeFailoverSlotGUCs returns the static, spec-known GUCs needed once a -// database's declared version crosses the NeedsNativeFailoverSlots gate. -// -// synchronized_standby_slots is deliberately NOT set here even when the -// gate passes: 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. -func NativeFailoverSlotGUCs(version *ds.PgEdgeVersion) map[string]any { - if !NeedsNativeFailoverSlotsForVersion(version) { - return map[string]any{} - } - return map[string]any{ - "sync_replication_slots": "on", - } -} - 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 fff0de82..6eed458a 100644 --- a/server/internal/postgres/gucs_test.go +++ b/server/internal/postgres/gucs_test.go @@ -80,7 +80,7 @@ func TestNeedsNativeFailoverSlotsForVersion(t *testing.T) { } } -func TestNativeFailoverSlotGUCs(t *testing.T) { +func TestDefaultGUCsSyncReplicationSlots(t *testing.T) { for _, tc := range []struct { name string version *ds.PgEdgeVersion @@ -93,7 +93,7 @@ func TestNativeFailoverSlotGUCs(t *testing.T) { {name: "spock 6 pg18", version: ds.MustParsePgEdgeVersion("18.4", "6"), expectedPresent: true}, } { t.Run(tc.name, func(t *testing.T) { - gucs := postgres.NativeFailoverSlotGUCs(tc.version) + gucs := postgres.DefaultGUCs(tc.version) value, ok := gucs["sync_replication_slots"] assert.Equal(t, tc.expectedPresent, ok) if tc.expectedPresent {