From 5fac61500e09e1d91d570b49c92c2f935298d607 Mon Sep 17 00:00:00 2001 From: Siva Date: Fri, 14 Aug 2026 22:36:03 +0530 Subject: [PATCH 1/7] feat: add foundational Spock 6 cluster support --- e2e/spock6_add_node_test.go | 101 ++++++++++++ ...wo_nodes_to_three_nodes_with_populate.json | 53 ++++++- .../database/operations/populate_nodes.go | 62 ++++++++ .../database/operations/update_database.go | 8 + .../database/peer_catchup_resource.go | 8 +- .../database/reconcile_versions_test.go | 53 +++++++ server/internal/database/resources.go | 1 + .../internal/database/sync_event_resource.go | 25 ++- ...erify_subscription_replicating_resource.go | 117 ++++++++++++++ server/internal/ds/versions.go | 50 +++++- server/internal/ds/versions_test.go | 63 +++++++- .../orchestrator/swarm/manifest_loader.go | 24 ++- .../swarm/manifest_loader_test.go | 149 ++++++++++++++++++ .../orchestrator/swarm/version-manifest.json | 6 + server/internal/postgres/create_db.go | 20 ++- server/internal/postgres/create_db_test.go | 33 ++++ 16 files changed, 743 insertions(+), 30 deletions(-) create mode 100644 e2e/spock6_add_node_test.go create mode 100644 server/internal/database/verify_subscription_replicating_resource.go diff --git a/e2e/spock6_add_node_test.go b/e2e/spock6_add_node_test.go new file mode 100644 index 00000000..cf9a048b --- /dev/null +++ b/e2e/spock6_add_node_test.go @@ -0,0 +1,101 @@ +//go:build e2e_test + +package e2e + +import ( + "context" + "testing" + "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" +) + +// spock6DevImage is a floating/mutable tag tracking the latest Spock 6 +// development build. Not pinned to a specific build number: nightly CI +// re-running this test picks up whatever the tag currently resolves to, +// with no extra plumbing needed to point CI at "latest." +const spock6DevImage = "ghcr.io/pgedge/pgedge-postgres:18-spock6-standard" + +// TestSpock6AddNode validates the add-node workflow end-to-end against a +// real Spock 6 cluster: creates a 2-node database pinned to a Spock 6 dev +// image via orchestrator_opts.swarm.image (bypassing manifest version +// constraints, since spock6 manifest entries are deliberately "dev" +// stability and never auto-selected), adds a 3rd node, and confirms the +// full mesh reaches "replicating" — exercising the Spock-major-gated +// spock.progress query (PeerCatchupResource) and the verify-replicating +// step (VerifySubscriptionReplicatingResource) added in this same ticket. +func TestSpock6AddNode(t *testing.T) { + t.Parallel() + + const ( + username = "admin" + password = "password" + dbName = "spock6_add_node_db" + ) + + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Minute) + defer cancel() + + hostIDs := fixture.HostIDs() + + nodeSpec := func(name, hostID string) *controlplane.DatabaseNodeSpec { + return &controlplane.DatabaseNodeSpec{ + Name: name, + HostIds: []controlplane.Identifier{controlplane.Identifier(hostID)}, + OrchestratorOpts: &controlplane.OrchestratorOpts{ + Swarm: &controlplane.SwarmOpts{Image: pointerTo(spock6DevImage)}, + }, + } + } + + t.Log("Step 1: Creating 2-node Spock 6 database fixture") + db := fixture.NewDatabaseFixture(ctx, t, &controlplane.CreateDatabaseRequest{ + Spec: &controlplane.DatabaseSpec{ + DatabaseName: dbName, + PostgresVersion: pointerTo("18.4"), + SpockVersion: pointerTo("6"), + Port: pointerTo(0), + PatroniPort: pointerTo(0), + DatabaseUsers: []*controlplane.DatabaseUserSpec{{ + Username: username, + Password: pointerTo(password), + DbOwner: pointerTo(true), + Attributes: []string{"LOGIN", "SUPERUSER"}, + }}, + Nodes: []*controlplane.DatabaseNodeSpec{ + nodeSpec("n1", hostIDs[0]), + nodeSpec("n2", hostIDs[1]), + }, + }, + }) + t.Logf("Database created: %s", db.ID) + + t.Log("Step 2: Adding n3 node with n1 as source") + db.Spec.Nodes = append(db.Spec.Nodes, func() *controlplane.DatabaseNodeSpec { + n := nodeSpec("n3", hostIDs[2]) + n.SourceNode = pointerTo("n1") + return n + }()) + require.NoError(t, db.Update(ctx, UpdateOptions{Spec: db.Spec})) + t.Log("Add-node completed successfully against Spock 6") + + t.Log("Step 3: Waiting for full mesh replication") + db.WaitForReplication(ctx, t, username, password) + t.Log("Replication complete") + + t.Log("Step 4: Verifying spock.spock_version() reports major 6 on the new node") + n3Opts := ConnectionOptions{ + Matcher: And(WithNode("n3"), WithRole("primary")), + Username: username, + Password: password, + } + db.WithConnection(ctx, n3Opts, t, func(conn *pgx.Conn) { + var version string + err := conn.QueryRow(ctx, "SELECT spock.spock_version();").Scan(&version) + require.NoError(t, err) + assert.Regexp(t, `^6\.`, version, "expected node n3 to be running Spock 6, got %q", version) + }) +} diff --git a/server/internal/database/operations/golden_test/TestUpdateDatabase/two_nodes_to_three_nodes_with_populate.json b/server/internal/database/operations/golden_test/TestUpdateDatabase/two_nodes_to_three_nodes_with_populate.json index 869f0e4a..7bb8c13e 100644 --- a/server/internal/database/operations/golden_test/TestUpdateDatabase/two_nodes_to_three_nodes_with_populate.json +++ b/server/internal/database/operations/golden_test/TestUpdateDatabase/two_nodes_to_three_nodes_with_populate.json @@ -159,6 +159,40 @@ } ] ], + [ + [ + { + "type": "update", + "resource_id": "database.subscription::n2:n3:test", + "reason": "has_diff", + "diff": [ + { + "value": false, + "op": "replace", + "path": "/disabled" + }, + { + "value": [ + { + "id": "n2:n3:test", + "type": "database.replication_origin_advance" + } + ], + "op": "replace", + "path": "/extra_dependencies" + } + ] + } + ], + [ + { + "type": "create", + "resource_id": "database.verify_subscription_replicating::n2:n3:test", + "reason": "does_not_exist", + "diff": null + } + ] + ], [ [ { @@ -203,9 +237,9 @@ "reason": "has_diff", "diff": [ { - "value": false, + "value": null, "op": "replace", - "path": "/disabled" + "path": "/extra_dependencies" } ] }, @@ -225,12 +259,12 @@ [ { "type": "delete", - "resource_id": "database.replication_origin_advance::n2:n3:test", + "resource_id": "database.roles_source::n3", "diff": null }, { "type": "delete", - "resource_id": "database.roles_source::n3", + "resource_id": "database.verify_subscription_replicating::n2:n3:test", "diff": null } ], @@ -239,7 +273,16 @@ "type": "delete", "resource_id": "database.dump_roles::n1", "diff": null - }, + } + ], + [ + { + "type": "delete", + "resource_id": "database.replication_origin_advance::n2:n3:test", + "diff": null + } + ], + [ { "type": "delete", "resource_id": "database.replication_slot_advance_from_cts::n2:n3:test", diff --git a/server/internal/database/operations/populate_nodes.go b/server/internal/database/operations/populate_nodes.go index 9aa84eef..3bb587e7 100644 --- a/server/internal/database/operations/populate_nodes.go +++ b/server/internal/database/operations/populate_nodes.go @@ -85,6 +85,68 @@ func PopulateNodes(existing, new []*NodeResources) (*resource.State, error) { return merged, nil } +// EnablePeerSubscriptions returns a diff that enables the peer subscriptions +// addPeerResources creates disabled. It must be applied as a separate, later +// phase than PopulateNodes' own state. +// +// addPeerResources creates each peer->new-node subscription disabled so the +// peer-catchup chain (SyncEvent -> WaitForSyncEvent -> PeerCatchup -> +// LagTracker -> ReplicationSlotAdvanceFromCTS -> ReplicationOriginAdvance) +// can run without a live subscriber racing that setup. But a single +// resource.State can only express one desired value per identifier, so that +// same state can never also declare "now enable it" — nothing else in the +// codebase ever does, which left these subscriptions permanently disabled. +// This returns a second state that re-declares the same SubscriptionResource +// identifiers with Disabled: false; applied after the populate phase is +// fully persisted, its diff sees disabled->enabled and calls Update, which +// is what actually flips sub_enabled in spock.subscription. +func EnablePeerSubscriptions(existing, new []*NodeResources) (*resource.State, error) { + existingNodeNames := make([]string, len(existing)) + for i, n := range existing { + existingNodeNames[i] = n.NodeName + } + + enable := resource.NewState() + for _, node := range new { + if node.SourceNode == "" { + continue + } + dbName := node.DatabaseName + for _, peer := range existingNodeNames { + if peer == node.NodeName || peer == node.SourceNode { + continue + } + err := enable.AddResource( + &database.SubscriptionResource{ + DatabaseName: dbName, + SubscriberNode: node.NodeName, + ProviderNode: peer, + Disabled: false, + ExtraDependencies: []resource.Identifier{ + database.ReplicationOriginAdvanceResourceIdentifier(peer, node.NodeName, dbName), + }, + }, + // Verify the enable actually took effect. Same phase, not a + // separate one: this is a new resource type/identifier, not + // a re-declaration of an existing one, so it can safely + // depend on the SubscriptionResource declared just above + // within this same state and run after it in the same + // apply pass. + &database.VerifySubscriptionReplicatingResource{ + DatabaseName: dbName, + SubscriberNode: node.NodeName, + ProviderNode: peer, + }, + ) + if err != nil { + return nil, fmt.Errorf("failed to add peer-enable resource to 'enable' state: %w", err) + } + } + } + + return enable, nil +} + func addPeerResources( state *resource.State, dbName string, diff --git a/server/internal/database/operations/update_database.go b/server/internal/database/operations/update_database.go index 3ec57341..3252545f 100644 --- a/server/internal/database/operations/update_database.go +++ b/server/internal/database/operations/update_database.go @@ -192,6 +192,14 @@ func addNodesStates(updates, adds []*NodeResources) ([]*resource.State, error) { states = append(states, populate) } + enable, err := EnablePeerSubscriptions(updates, adds) + if err != nil { + return nil, err + } + if enable != nil { + states = append(states, enable) + } + return states, nil } diff --git a/server/internal/database/peer_catchup_resource.go b/server/internal/database/peer_catchup_resource.go index 597b0e7f..117a1919 100644 --- a/server/internal/database/peer_catchup_resource.go +++ b/server/internal/database/peer_catchup_resource.go @@ -77,6 +77,12 @@ func (r *PeerCatchupResource) Refresh(ctx context.Context, rc *resource.Context) } defer conn.Close(ctx) + spockVersion, err := getLiveSpockVersion(ctx, conn) + if err != nil { + return fmt.Errorf("failed to check spock version on source node %q: %w", r.SourceNode, err) + } + spockMajor, _ := spockVersion.Major() + const pollInterval = 500 * time.Millisecond for { @@ -84,7 +90,7 @@ func (r *PeerCatchupResource) Refresh(ctx context.Context, rc *resource.Context) return ctx.Err() } - reached, err := postgres.SpockProgressReachedLSN(r.PeerNode, syncEvent.SyncEventLsn). + reached, err := postgres.SpockProgressReachedLSN(spockMajor, r.PeerNode, syncEvent.SyncEventLsn). Scalar(ctx, conn) if err != nil { return fmt.Errorf("failed to query spock progress for peer %q: %w", r.PeerNode, err) diff --git a/server/internal/database/reconcile_versions_test.go b/server/internal/database/reconcile_versions_test.go index 92ffb774..3493a1e8 100644 --- a/server/internal/database/reconcile_versions_test.go +++ b/server/internal/database/reconcile_versions_test.go @@ -122,6 +122,59 @@ func TestReconcileVersions(t *testing.T) { }, }, }, + { + // Before ds.ParseVersion accepted pre-release suffixes, this + // instance would be silently skipped at reconcile_versions.go's + // ds.ParsePgEdgeVersion call (line 141) and never appear in + // updatedInstances. It should now reconcile normally, with the + // pre-release suffix dropped by Normalize(). + name: "spock beta version is not silently skipped", + spec: &database.StoredSpec{ + Spec: &database.Spec{ + PostgresVersion: "17.4", + SpockVersion: "5", + Nodes: []*database.Node{ + {Name: "n1", HostIDs: []string{"host-1"}}, + }, + }, + }, + instances: []*database.StoredInstance{ + { + InstanceID: "n1-host-1", + NodeName: "n1", + HostID: "host-1", + PgEdgeVersion: ds.MustParsePgEdgeVersion("17.4", "5"), + }, + }, + statuses: []*database.StoredInstanceStatus{ + { + InstanceID: "n1-host-1", + Status: &database.InstanceStatus{ + StatusUpdatedAt: utils.PointerTo(time.Now()), + Role: utils.PointerTo(patroni.InstanceRolePrimary), + PostgresVersion: utils.PointerTo("17.5"), + SpockVersion: utils.PointerTo("6.0.0-beta.1"), + }, + }, + }, + expectedSpec: &database.StoredSpec{ + Spec: &database.Spec{ + PostgresVersion: "17.5", + SpockVersion: "6", + Nodes: []*database.Node{ + {Name: "n1", HostIDs: []string{"host-1"}}, + }, + }, + }, + expectedInstances: []*database.StoredInstance{ + { + InstanceID: "n1-host-1", + NodeName: "n1", + HostID: "host-1", + PgEdgeVersion: ds.MustParsePgEdgeVersion("17.5", "6"), + }, + }, + }, { name: "all nodes updated spock only", spec: &database.StoredSpec{ diff --git a/server/internal/database/resources.go b/server/internal/database/resources.go index 998b1190..511bd54b 100644 --- a/server/internal/database/resources.go +++ b/server/internal/database/resources.go @@ -13,6 +13,7 @@ func RegisterResourceTypes(registry *resource.Registry) { resource.RegisterResourceType[*LagTrackerCommitTimestampResource](registry, ResourceTypeLagTrackerCommitTS) resource.RegisterResourceType[*ReplicationSlotAdvanceFromCTSResource](registry, ResourceTypeReplicationSlotAdvanceFromCTS) resource.RegisterResourceType[*ReplicationOriginAdvanceResource](registry, ResourceTypeReplicationOriginAdvance) + resource.RegisterResourceType[*VerifySubscriptionReplicatingResource](registry, ResourceTypeVerifySubscriptionReplicating) resource.RegisterResourceType[*PeerCatchupResource](registry, ResourceTypePeerCatchup) resource.RegisterResourceType[*SwitchoverResource](registry, ResourceTypeSwitchover) resource.RegisterResourceType[*PostgresDatabaseResource](registry, ResourceTypePostgresDatabase) diff --git a/server/internal/database/sync_event_resource.go b/server/internal/database/sync_event_resource.go index 9a6377bb..c9989888 100644 --- a/server/internal/database/sync_event_resource.go +++ b/server/internal/database/sync_event_resource.go @@ -13,18 +13,31 @@ import ( var minSpockVersionForSyncEventArgs = ds.MustParseVersion(postgres.MinSpockVersionForSyncEventArgs) +// getLiveSpockVersion queries conn directly for the Spock version actually +// running on this connection, rather than trusting spec/stored state — spec +// state can lag behind what's really deployed (e.g. mid add-node, mid +// upgrade), and every SQL-shape decision gated on Spock version needs to +// match what's really there. +func getLiveSpockVersion(ctx context.Context, conn *pgx.Conn) (*ds.Version, error) { + versionStr, err := postgres.GetSpockVersion().Scalar(ctx, conn) + if err != nil { + return nil, fmt.Errorf("failed to get spock version: %w", err) + } + version, err := ds.ParseVersion(versionStr) + if err != nil { + return nil, fmt.Errorf("failed to parse spock version %q: %w", versionStr, err) + } + return version, nil +} + // spockSupportsSyncEventArgs reports whether conn's Spock version is new // enough for spock.sync_event(boolean) and the 5-arg // spock.wait_for_sync_event(..., wait_if_disabled) — see // postgres.MinSpockVersionForSyncEventArgs. func spockSupportsSyncEventArgs(ctx context.Context, conn *pgx.Conn) (bool, error) { - versionStr, err := postgres.GetSpockVersion().Scalar(ctx, conn) - if err != nil { - return false, fmt.Errorf("failed to get spock version: %w", err) - } - version, err := ds.ParseVersion(versionStr) + version, err := getLiveSpockVersion(ctx, conn) if err != nil { - return false, fmt.Errorf("failed to parse spock version %q: %w", versionStr, err) + return false, err } return version.Compare(minSpockVersionForSyncEventArgs) >= 0, nil } diff --git a/server/internal/database/verify_subscription_replicating_resource.go b/server/internal/database/verify_subscription_replicating_resource.go new file mode 100644 index 00000000..d3de7594 --- /dev/null +++ b/server/internal/database/verify_subscription_replicating_resource.go @@ -0,0 +1,117 @@ +package database + +import ( + "context" + "fmt" + "time" + + "github.com/pgEdge/control-plane/server/internal/postgres" + "github.com/pgEdge/control-plane/server/internal/resource" +) + +var _ resource.Resource = (*VerifySubscriptionReplicatingResource)(nil) + +const ResourceTypeVerifySubscriptionReplicating resource.Type = "database.verify_subscription_replicating" + +func VerifySubscriptionReplicatingResourceIdentifier(providerNode, subscriberNode, databaseName string) resource.Identifier { + return resource.Identifier{ + Type: ResourceTypeVerifySubscriptionReplicating, + ID: fmt.Sprintf("%s:%s:%s", providerNode, subscriberNode, databaseName), + } +} + +// VerifySubscriptionReplicatingResource polls a subscription's status until +// it reaches "replicating", failing loudly if it doesn't within a bounded +// wait. Mirrors Spock's own zodan.sql reference add-node flow +// (spock.verify_subscription_replicating), which Control Plane's pipeline +// otherwise has no equivalent of: nothing previously checked that an +// enabled subscription's apply worker actually started, so a subscription +// that never starts replicating could silently leave a node missing data +// with no error anywhere. This resource doesn't fix that underlying +// possibility — an apply worker failing to start is Spock's own concern — +// it turns it from silent, permanent data loss into a visible, actionable +// task failure instead. +type VerifySubscriptionReplicatingResource struct { + DatabaseName string `json:"database_name"` + ProviderNode string `json:"provider_node"` + SubscriberNode string `json:"subscriber_node"` +} + +func (r *VerifySubscriptionReplicatingResource) ResourceVersion() string { return "1" } +func (r *VerifySubscriptionReplicatingResource) DiffIgnore() []string { return nil } + +// Subscription status is local to the subscriber (spock.sub_show_status() +// reports on incoming subscriptions), so this must run on that host. +func (r *VerifySubscriptionReplicatingResource) Executor() resource.Executor { + return resource.PrimaryExecutor(r.SubscriberNode) +} + +func (r *VerifySubscriptionReplicatingResource) Identifier() resource.Identifier { + return VerifySubscriptionReplicatingResourceIdentifier(r.ProviderNode, r.SubscriberNode, r.DatabaseName) +} + +func (r *VerifySubscriptionReplicatingResource) Dependencies() []resource.Identifier { + return []resource.Identifier{ + SubscriptionResourceIdentifier(r.ProviderNode, r.SubscriberNode, r.DatabaseName), + } +} + +func (r *VerifySubscriptionReplicatingResource) TypeDependencies() []resource.Type { return nil } + +func (r *VerifySubscriptionReplicatingResource) Refresh(ctx context.Context, rc *resource.Context) error { + subscriber, err := GetPrimaryInstance(ctx, rc, r.SubscriberNode) + if err != nil { + return fmt.Errorf("failed to get subscriber instance for node %q: %w", r.SubscriberNode, err) + } + conn, err := subscriber.Connection(ctx, rc, r.DatabaseName) + if err != nil { + return fmt.Errorf("failed to connect to subscriber %q: %w", r.SubscriberNode, err) + } + defer conn.Close(ctx) + + // Matches Spock's own verify_subscription_replicating default wait + // (120s) with headroom, since ours is the second such check in the + // pipeline (after whatever WaitForSyncEventResource already waited + // for) rather than the only one. + const ( + pollInterval = 2 * time.Second + waitTimeout = 3 * time.Minute + ) + waitCtx, cancel := context.WithTimeout(ctx, waitTimeout) + defer cancel() + + var lastStatus string + for { + status, err := postgres.GetSubscriptionStatus(r.ProviderNode, r.SubscriberNode).Scalar(waitCtx, conn) + if err != nil { + if postgres.IsSpockNodeNotConfigured(err) { + return resource.ErrNotFound + } + return fmt.Errorf("failed to check subscription status: %w", err) + } + if status == postgres.SubStatusReplicating { + return nil + } + lastStatus = status + + select { + case <-waitCtx.Done(): + return fmt.Errorf( + "subscription %s->%s did not reach %q status within %s (last status: %q)", + r.ProviderNode, r.SubscriberNode, postgres.SubStatusReplicating, waitTimeout, lastStatus) + case <-time.After(pollInterval): + } + } +} + +func (r *VerifySubscriptionReplicatingResource) Create(ctx context.Context, rc *resource.Context) error { + return r.Refresh(ctx, rc) +} + +func (r *VerifySubscriptionReplicatingResource) Update(ctx context.Context, rc *resource.Context) error { + return r.Refresh(ctx, rc) +} + +func (r *VerifySubscriptionReplicatingResource) Delete(ctx context.Context, rc *resource.Context) error { + return nil +} diff --git a/server/internal/ds/versions.go b/server/internal/ds/versions.go index 5128dd3f..597c1b10 100644 --- a/server/internal/ds/versions.go +++ b/server/internal/ds/versions.go @@ -47,6 +47,11 @@ var _ encoding.TextUnmarshaler = (*Version)(nil) type Version struct { Components []uint64 `json:"components"` + // PreRelease is the optional suffix after a "-" (e.g. "beta.1" in + // "6.0.0-beta.1"). It is an opaque string, not decomposed into SemVer + // precedence rules, and is intentionally dropped by MajorVersion and + // MajorMinorVersion since a major/major-minor bucket never carries one. + PreRelease string `json:"pre_release,omitempty"` } func (v *Version) Major() (uint64, bool) { @@ -64,6 +69,8 @@ func (v *Version) MajorString() (string, bool) { return strconv.FormatUint(major, 10), true } +// MajorVersion returns just the major component. PreRelease is intentionally +// dropped: a major-version bucket never carries a pre-release suffix. func (v *Version) MajorVersion() *Version { if len(v.Components) == 0 { return &Version{} @@ -73,6 +80,9 @@ func (v *Version) MajorVersion() *Version { } } +// MajorMinorVersion returns the major.minor components. PreRelease is +// intentionally dropped: a major.minor bucket never carries a pre-release +// suffix. func (v *Version) MajorMinorVersion() *Version { components := slices.Clone(v.Components) if len(components) > 2 { @@ -88,12 +98,17 @@ func (v *Version) String() string { for i, c := range v.Components { components[i] = strconv.FormatUint(c, 10) } - return strings.Join(components, ".") + s := strings.Join(components, ".") + if v.PreRelease != "" { + s += "-" + v.PreRelease + } + return s } func (v *Version) Clone() *Version { return &Version{ Components: slices.Clone(v.Components), + PreRelease: v.PreRelease, } } @@ -107,6 +122,7 @@ func (v *Version) UnmarshalText(data []byte) error { return err } v.Components = parsed.Components + v.PreRelease = parsed.PreRelease return nil } @@ -138,11 +154,32 @@ func (v *Version) UnmarshalJSON(data []byte) error { } } +// Compare orders by numeric components first. If those are equal, a +// pre-release never compares equal to its release counterpart (a release +// sorts after any pre-release of the same numeric version), and two +// different pre-releases of the same numeric version fall back to a plain +// string comparison. This is NOT full SemVer pre-release precedence — it's +// just enough to avoid falsely claiming two different versions are equal. func (v *Version) Compare(other *Version) int { - return slices.Compare(v.Components, other.Components) + if c := slices.Compare(v.Components, other.Components); c != 0 { + return c + } + switch { + case v.PreRelease == other.PreRelease: + return 0 + case v.PreRelease == "": + return 1 + case other.PreRelease == "": + return -1 + default: + return strings.Compare(v.PreRelease, other.PreRelease) + } } -var semverRegexp = regexp.MustCompile(`^\d+(\.\d+){0,2}$`) +// semverRegexp matches "[.[.]][-]". The +// pre-release group accepts dot/hyphen-separated alphanumeric identifiers +// (e.g. "beta", "beta.1", "rc.2") but not SemVer build metadata ("+..."). +var semverRegexp = regexp.MustCompile(`^(\d+(?:\.\d+){0,2})(?:-([0-9A-Za-z]+(?:[.-][0-9A-Za-z]+)*))?$`) func MustParseVersion(s string) *Version { v, err := ParseVersion(s) @@ -153,10 +190,11 @@ func MustParseVersion(s string) *Version { } func ParseVersion(s string) (*Version, error) { - if !semverRegexp.MatchString(s) { + m := semverRegexp.FindStringSubmatch(s) + if m == nil { return nil, fmt.Errorf("invalid version format: %q", s) } - parts := strings.Split(s, ".") + parts := strings.Split(m[1], ".") components := make([]uint64, len(parts)) for i, p := range parts { c, err := strconv.ParseUint(p, 10, 64) @@ -165,7 +203,7 @@ func ParseVersion(s string) (*Version, error) { } components[i] = c } - return &Version{Components: components}, nil + return &Version{Components: components, PreRelease: m[2]}, nil } type PgEdgeVersion struct { diff --git a/server/internal/ds/versions_test.go b/server/internal/ds/versions_test.go index 7b19dd90..a6996031 100644 --- a/server/internal/ds/versions_test.go +++ b/server/internal/ds/versions_test.go @@ -61,9 +61,25 @@ func TestParseVersion(t *testing.T) { expectedErr: "invalid version format", }, { - // Intentionally not supporting pre-release identifiers because they - // are not comparable. - input: "5.0.0-beta", + // Pre-release identifiers are accepted so we can tolerate a + // live-observed beta version (e.g. from spock_version()) without + // silently dropping the instance from reconciliation. This is not + // full SemVer precedence — see Version.Compare. + input: "5.0.0-beta", + expected: &ds.Version{Components: []uint64{5, 0, 0}, PreRelease: "beta"}, + }, + { + input: "6.0.0-beta.1", + expected: &ds.Version{Components: []uint64{6, 0, 0}, PreRelease: "beta.1"}, + }, + { + // Still rejected: empty pre-release suffix. + input: "5.0.0-", + expectedErr: "invalid version format", + }, + { + // Still rejected: SemVer build metadata is out of scope. + input: "5.0.0-beta+build1", expectedErr: "invalid version format", }, } { @@ -86,6 +102,7 @@ func TestVersion(t *testing.T) { "17", "17.6", "5.0.0", + "6.0.0-beta.1", } { t.Run(tc, func(t *testing.T) { out, err := ds.ParseVersion(tc) @@ -96,6 +113,14 @@ func TestVersion(t *testing.T) { } }) + t.Run("MajorVersion and MajorMinorVersion drop PreRelease", func(t *testing.T) { + v, err := ds.ParseVersion("6.0.0-beta.1") + require.NoError(t, err) + + assert.Equal(t, &ds.Version{Components: []uint64{6}}, v.MajorVersion()) + assert.Equal(t, &ds.Version{Components: []uint64{6, 0}}, v.MajorMinorVersion()) + }) + t.Run("Compare", func(t *testing.T) { for _, tc := range []struct { a *ds.Version @@ -166,6 +191,30 @@ func TestVersion(t *testing.T) { b: &ds.Version{Components: []uint64{1, 0, 0}}, expected: -1, }, + { + // A pre-release must never compare equal to its release + // counterpart, even though the numeric Components match. + a: &ds.Version{Components: []uint64{6, 0, 0}, PreRelease: "beta.1"}, + b: &ds.Version{Components: []uint64{6, 0, 0}}, + expected: -1, + }, + { + a: &ds.Version{Components: []uint64{6, 0, 0}}, + b: &ds.Version{Components: []uint64{6, 0, 0}, PreRelease: "beta.1"}, + expected: 1, + }, + { + a: &ds.Version{Components: []uint64{6, 0, 0}, PreRelease: "beta.1"}, + b: &ds.Version{Components: []uint64{6, 0, 0}, PreRelease: "beta.1"}, + expected: 0, + }, + { + // Two different pre-releases of the same numeric version: not + // full SemVer precedence, just guaranteed non-equal. + a: &ds.Version{Components: []uint64{6, 0, 0}, PreRelease: "beta.1"}, + b: &ds.Version{Components: []uint64{6, 0, 0}, PreRelease: "beta.2"}, + expected: -1, + }, } { t.Run(fmt.Sprintf("%s and %s", tc.a.String(), tc.b.String()), func(t *testing.T) { result := tc.a.Compare(tc.b) @@ -228,6 +277,14 @@ func TestNewPgEdgeVersion(t *testing.T) { spockVersion: "invalid", expectedErr: "invalid spock version", }, + { + postgresVersion: "17.6", + spockVersion: "6.0.0-beta.1", + expected: &ds.PgEdgeVersion{ + PostgresVersion: &ds.Version{Components: []uint64{17, 6}}, + SpockVersion: &ds.Version{Components: []uint64{6, 0, 0}, PreRelease: "beta.1"}, + }, + }, } { t.Run(tc.postgresVersion+"_"+tc.spockVersion, func(t *testing.T) { result, err := ds.ParsePgEdgeVersion(tc.postgresVersion, tc.spockVersion) diff --git a/server/internal/orchestrator/swarm/manifest_loader.go b/server/internal/orchestrator/swarm/manifest_loader.go index 0f1a0adf..01417e47 100644 --- a/server/internal/orchestrator/swarm/manifest_loader.go +++ b/server/internal/orchestrator/swarm/manifest_loader.go @@ -382,16 +382,36 @@ func buildVersions(cfg config.Config, mf *versionManifest) (*Versions, error) { } img := &Images{ PgEdgeImage: serviceImageTag(cfg, e.Image), + Stability: e.Stability, } versions.addImage(pv, img) if e.Default { + if e.Stability != "" && e.Stability != "stable" { + return nil, fmt.Errorf("invalid version entry {postgres:%s spock:%s}: a %q-stability entry cannot be marked default", + e.PostgresVersion, e.SpockVersion, e.Stability) + } defaultVer = pv } } if defaultVer == nil { - // Fall back to the last entry if no default is marked. - defaultVer = versions.supportedVersions[len(versions.supportedVersions)-1] + // Fall back to the last stable entry if no default is marked. A + // non-stable (e.g. "dev") entry must never become the default just + // because it happens to be last in the manifest. + for i := len(entries) - 1; i >= 0 && defaultVer == nil; i-- { + if entries[i].Stability != "" && entries[i].Stability != "stable" { + continue + } + pv, err := ds.ParsePgEdgeVersion(entries[i].PostgresVersion, entries[i].SpockVersion) + if err != nil { + return nil, fmt.Errorf("invalid version entry {postgres:%s spock:%s}: %w", + entries[i].PostgresVersion, entries[i].SpockVersion, err) + } + defaultVer = pv + } + if defaultVer == nil { + return nil, fmt.Errorf("manifest has no stable entry to use as a default") + } } versions.defaultVersion = defaultVer diff --git a/server/internal/orchestrator/swarm/manifest_loader_test.go b/server/internal/orchestrator/swarm/manifest_loader_test.go index 963bfa1e..5814c744 100644 --- a/server/internal/orchestrator/swarm/manifest_loader_test.go +++ b/server/internal/orchestrator/swarm/manifest_loader_test.go @@ -12,6 +12,7 @@ import ( "time" "github.com/pgEdge/control-plane/server/internal/config" + "github.com/pgEdge/control-plane/server/internal/ds" "github.com/pgEdge/control-plane/server/internal/testutils" ) @@ -553,6 +554,154 @@ func TestValidateManifest(t *testing.T) { } } +// TestBuildVersions_StabilityWired verifies that a "stability" value on a +// manifest entry is actually carried through to the runtime Images.Stability +// field. Regression test: buildVersions previously parsed Stability off the +// JSON entry but never copied it onto the constructed *Images, so every +// entry loaded from a real manifest silently ended up with Stability == "", +// which the filtering logic in AvailableUpgrades/FindUpgrade treats as +// "stable" — a "dev" entry would have had zero effect. +func TestBuildVersions_StabilityWired(t *testing.T) { + m := &ManifestLoader{logger: testutils.Logger(t), cfg: config.Config{ + DockerSwarm: config.DockerSwarm{ImageRepositoryHost: "ghcr.io/pgedge"}, + }} + data, err := json.Marshal(map[string]any{ + "schema_version": 1, + "images": map[string]any{ + "postgres": []map[string]any{ + { + "postgres_version": "18.4", + "spock_version": "5", + "image": "pgedge-postgres:18.4-spock5.0.10-standard-1", + "stability": "stable", + "default": true, + }, + { + "postgres_version": "18.4", + "spock_version": "6", + "image": "pgedge-postgres:18-spock6-standard", + "stability": "dev", + }, + }, + }, + }) + if err != nil { + t.Fatal(err) + } + + v, _, err := m.parseManifestData(data) + if err != nil { + t.Fatalf("parseManifestData: %v", err) + } + + spock6 := ds.MustParsePgEdgeVersion("18.4", "6") + imgs, err := v.GetImages(spock6) + if err != nil { + t.Fatalf("GetImages(spock6): %v", err) + } + if imgs.Stability != "dev" { + t.Errorf("Stability = %q, want %q", imgs.Stability, "dev") + } + + if v.Default().SpockVersion.String() != "5" { + t.Errorf("default spock version = %s, want 5 (dev entry must never be default)", v.Default().SpockVersion) + } +} + +// TestBuildVersions_RejectsDevDefault verifies that manifest loading fails +// outright if a non-stable entry is marked default, rather than silently +// allowing a dev image to become the default for new database creation. +func TestBuildVersions_RejectsDevDefault(t *testing.T) { + m := &ManifestLoader{logger: testutils.Logger(t), cfg: config.Config{ + DockerSwarm: config.DockerSwarm{ImageRepositoryHost: "ghcr.io/pgedge"}, + }} + data, err := json.Marshal(map[string]any{ + "schema_version": 1, + "images": map[string]any{ + "postgres": []map[string]any{ + { + "postgres_version": "18.4", + "spock_version": "6", + "image": "pgedge-postgres:18-spock6-standard", + "stability": "dev", + "default": true, + }, + }, + }, + }) + if err != nil { + t.Fatal(err) + } + + if _, _, err := m.parseManifestData(data); err == nil { + t.Fatal("expected error when a dev-stability entry is marked default") + } +} + +// TestBuildVersions_FallbackDefaultSkipsNonStable verifies that when no entry +// is explicitly marked default, the implicit "last entry" fallback still +// never selects a non-stable entry. +func TestBuildVersions_FallbackDefaultSkipsNonStable(t *testing.T) { + m := &ManifestLoader{logger: testutils.Logger(t), cfg: config.Config{ + DockerSwarm: config.DockerSwarm{ImageRepositoryHost: "ghcr.io/pgedge"}, + }} + data, err := json.Marshal(map[string]any{ + "schema_version": 1, + "images": map[string]any{ + "postgres": []map[string]any{ + { + "postgres_version": "18.4", + "spock_version": "5", + "image": "pgedge-postgres:18.4-spock5.0.10-standard-1", + "stability": "stable", + }, + { + "postgres_version": "18.4", + "spock_version": "6", + "image": "pgedge-postgres:18-spock6-standard", + "stability": "dev", + }, + }, + }, + }) + if err != nil { + t.Fatal(err) + } + + v, _, err := m.parseManifestData(data) + if err != nil { + t.Fatalf("parseManifestData: %v", err) + } + if v.Default().SpockVersion.String() != "5" { + t.Errorf("default spock version = %s, want 5 (fallback must skip the trailing dev entry)", v.Default().SpockVersion) + } +} + +// TestEmbeddedManifestValid_Spock6DevEntryNotDefault verifies the real, +// shipped version-manifest.json's Spock 6 dev entry is loaded (so it's +// reachable via an explicit postgres_version/spock_version request or an +// orchestrator_opts image override) but never selected as the default. +func TestEmbeddedManifestValid_Spock6DevEntryNotDefault(t *testing.T) { + m := &ManifestLoader{logger: testutils.Logger(t)} + v, _, err := m.parseManifestData(embeddedManifest) + if err != nil { + t.Fatalf("embedded manifest cannot be parsed: %v", err) + } + + spock6 := ds.MustParsePgEdgeVersion("18.4", "6") + imgs, err := v.GetImages(spock6) + if err != nil { + t.Fatalf("expected embedded manifest to have a spock6 entry: %v", err) + } + if imgs.Stability != "dev" { + t.Errorf("spock6 entry Stability = %q, want %q", imgs.Stability, "dev") + } + + if major, _ := v.Default().SpockVersion.Major(); major != 5 { + t.Errorf("default spock major = %d, want 5 (spock6 dev entry must never be default)", major) + } +} + // TestManifestLoader_ImageTagsHaveRegistryPrefix verifies that all image tags // returned by Versions and ServiceVersions include the configured registry // host. diff --git a/server/internal/orchestrator/swarm/version-manifest.json b/server/internal/orchestrator/swarm/version-manifest.json index 05f7ef36..0b83c3a1 100644 --- a/server/internal/orchestrator/swarm/version-manifest.json +++ b/server/internal/orchestrator/swarm/version-manifest.json @@ -92,6 +92,12 @@ "image": "pgedge-postgres:18.4-spock5.0.10-standard-1", "stability": "stable", "default": true + }, + { + "postgres_version": "18.4", + "spock_version": "6", + "image": "pgedge-postgres:18-spock6-standard", + "stability": "dev" } ], "postgrest": [ diff --git a/server/internal/postgres/create_db.go b/server/internal/postgres/create_db.go index 2fa61933..917ef979 100644 --- a/server/internal/postgres/create_db.go +++ b/server/internal/postgres/create_db.go @@ -491,23 +491,29 @@ func AdvanceReplicationOrigin(slotName, lsn string) Statement { // SpockProgressReachedLSN reports whether the local node's apply progress // from the named peer has reached targetLSN. Uses remote_lsn (the LSN of the -// last applied commit in Spock 5.x) rather than received_lsn, which can -// advance on keepalive messages before any commits have been applied. -func SpockProgressReachedLSN(peerNodeName, targetLSN string) Query[bool] { +// last applied commit) on Spock < 6, or remote_commit_lsn on Spock >= 6 — +// spock.progress became a view over apply_group_progress() in Spock 6 and +// the column was renamed. Neither uses received_lsn, which can advance on +// keepalive messages before any commits have been applied. +func SpockProgressReachedLSN(spockMajor uint64, peerNodeName, targetLSN string) Query[bool] { + column := "remote_lsn" + if spockMajor >= 6 { + column = "remote_commit_lsn" + } return Query[bool]{ - SQL: ` + SQL: fmt.Sprintf(` SELECT COALESCE( - (SELECT p.remote_lsn >= @target_lsn::pg_lsn + (SELECT p.%s >= @target_lsn::pg_lsn FROM spock.progress p JOIN spock.node n ON n.node_id = p.remote_node_id WHERE p.node_id = (SELECT node_id FROM spock.node_info()) AND n.node_name = @peer_node_name), false ) - `, + `, column), Args: pgx.NamedArgs{ "peer_node_name": peerNodeName, - "target_lsn": targetLSN, + "target_lsn": targetLSN, }, } } diff --git a/server/internal/postgres/create_db_test.go b/server/internal/postgres/create_db_test.go index 31a64ef5..d0616805 100644 --- a/server/internal/postgres/create_db_test.go +++ b/server/internal/postgres/create_db_test.go @@ -33,6 +33,39 @@ func TestSyncEvent(t *testing.T) { } } +func TestSpockProgressReachedLSN(t *testing.T) { + for _, tc := range []struct { + name string + spockMajor uint64 + expectedColumn string + }{ + { + name: "spock 5", + spockMajor: 5, + expectedColumn: "p.remote_lsn", + }, + { + name: "spock 6", + spockMajor: 6, + expectedColumn: "p.remote_commit_lsn", + }, + { + name: "spock 7 (future major, treated like 6)", + spockMajor: 7, + expectedColumn: "p.remote_commit_lsn", + }, + } { + t.Run(tc.name, func(t *testing.T) { + query := postgres.SpockProgressReachedLSN(tc.spockMajor, "n1", "0/0") + assert.Contains(t, query.SQL, tc.expectedColumn) + assert.Equal(t, pgx.NamedArgs{ + "peer_node_name": "n1", + "target_lsn": "0/0", + }, query.Args) + }) + } +} + func TestWaitForSyncEvent(t *testing.T) { for _, tc := range []struct { name string From 17c1fbb30affb1fb89f8b6bcdf97ac36564940c5 Mon Sep 17 00:00:00 2001 From: moizpgedge Date: Wed, 19 Aug 2026 19:37:39 +0500 Subject: [PATCH 2/7] fix: always allow-list spock_output for Spock 6 The Spock 6 manifest entry deliberately points at a floating/mutable image tag rather than a pinned build, so its declared postgres_version can drift out of sync with whatever Postgres minor the tag actually resolves to. That drift caused a real, live-reproduced failure: ERROR: library "spock_output" may not be used as an output plugin because output_plugin_libraries was computed from the declared version (below the gate threshold) while the real running binary was past it. Confirmed by running the TestSpock6AddNode e2e test directly against a real cluster: it failed with this exact error before the fix, and passes cleanly (twice, fresh runs) after it. Fix: treat Spock major >= 6 as always needing output_plugin_libraries, regardless of the declared Postgres minor. This is a deliberate trade-off documented in the code comment, not a fully general solution - it hasn't been verified whether setting this GUC against a hypothetical pre-gate Postgres minor is harmless. Spock 5.x behavior is unchanged. --- server/internal/postgres/gucs.go | 25 +++++++++++++++++++++++++ server/internal/postgres/gucs_test.go | 3 +++ 2 files changed, 28 insertions(+) diff --git a/server/internal/postgres/gucs.go b/server/internal/postgres/gucs.go index 58a17ece..4fe60c62 100644 --- a/server/internal/postgres/gucs.go +++ b/server/internal/postgres/gucs.go @@ -19,10 +19,35 @@ var minOutputPluginLibrariesVersions = map[uint64]*ds.Version{ // needsOutputPluginLibraries reports whether the given Postgres version // requires output_plugin_libraries to be set in order to allow spock_output // to create replication slots. +// +// Spock 6 manifest entries deliberately point at a floating/mutable image +// tag (see version-manifest.json) rather than a pinned build number, so its +// declared postgres_version can drift out of sync with whatever Postgres +// minor the tag actually resolves to at any given moment. That drift caused +// a real, live-reproduced failure ("library spock_output may not be used as +// an output plugin") when the floating tag moved past the gate threshold for +// a minor this function had no way to know about ahead of time, since it +// only ever sees the declared version, not what's actually running. +// +// Rather than track the floating tag's exact current resolution, Spock +// major >= 6 is treated as an unconditional yes here. This is a deliberate +// trade-off, not a fully general fix: every Spock 6 build seen in practice +// so far ships on a Postgres minor at or past the relevant gate, so this +// resolves the real failure above; it has not been verified whether setting +// this GUC against a hypothetical Postgres minor that predates the gate +// (and so may not recognize the parameter at all) is itself harmless or +// would fail Postgres startup. If Spock 6 is ever built against such a +// minor, this assumption needs revisiting. Spock 5.x behavior is unchanged +// — it still depends solely on the exact Postgres minor check below. func needsOutputPluginLibraries(version *ds.PgEdgeVersion) bool { if version == nil || version.PostgresVersion == nil { return false } + if version.SpockVersion != nil { + if spockMajor, ok := version.SpockVersion.Major(); ok && spockMajor >= 6 { + return true + } + } pgVersion := version.PostgresVersion.MajorMinorVersion() major, ok := pgVersion.Major() if !ok { diff --git a/server/internal/postgres/gucs_test.go b/server/internal/postgres/gucs_test.go index dd7d1606..64f0701a 100644 --- a/server/internal/postgres/gucs_test.go +++ b/server/internal/postgres/gucs_test.go @@ -29,6 +29,9 @@ func TestDefaultGUCsOutputPluginLibraries(t *testing.T) { {name: "pg18 at gate", version: ds.MustParsePgEdgeVersion("18.6", "4"), expectedPresent: true}, {name: "future major", version: ds.MustParsePgEdgeVersion("19.0", "4"), expectedPresent: true}, {name: "older major", version: ds.MustParsePgEdgeVersion("15.10", "4"), expectedPresent: false}, + {name: "spock 6 on pg18 below gate", version: ds.MustParsePgEdgeVersion("18.4", "6"), expectedPresent: true}, + {name: "spock 6 on pg16 below gate", version: ds.MustParsePgEdgeVersion("16.10", "6"), expectedPresent: true}, + {name: "spock 5 on pg18 below gate stays ungated", version: ds.MustParsePgEdgeVersion("18.4", "5"), expectedPresent: false}, } { t.Run(tc.name, func(t *testing.T) { gucs := postgres.DefaultGUCs(tc.version) From cc51563560041ee2e852f940543f2d59dd187496 Mon Sep 17 00:00:00 2001 From: moizpgedge Date: Wed, 19 Aug 2026 19:40:55 +0500 Subject: [PATCH 3/7] ci: add nightly Spock 6 e2e validation Closes the last gap in this ticket's scope: TestSpock6AddNode existed but only ever ran as part of a normal PR-triggered test_e2e run, which only catches regressions introduced by our own commits - it never re-checks the floating spock6DevImage tag on its own, so drift introduced by an upstream Spock 6 nightly build would go unnoticed until someone happened to touch this branch. Adds a dedicated test_e2e_spock6 job (just TestSpock6AddNode, not the full e2e split) and a nightly_spock6 workflow triggered by a cron schedule against main, independent of commit activity. The e2e test itself already points at the floating tag, so no extra plumbing is needed to track "latest" beyond the schedule. Validated with `circleci config validate`. --- .circleci/config.yml | 41 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 41 insertions(+) diff --git a/.circleci/config.yml b/.circleci/config.yml index 3fcd8dc4..9932db9f 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -137,6 +137,31 @@ jobs: when: always - store_test_results: path: . + test_e2e_spock6: + executor: common + steps: + - common_setup + - run: + name: Run Spock 6 add-node e2e test against the latest dev image + command: | + make ci-compose-detached + make test-e2e E2E_FIXTURE=ci E2E_RUN='^TestSpock6AddNode$$' TEST_RERUN_FAILS=2 + - run: + name: Archive debug output + command: | + if [[ -d ./e2e/debug ]]; then + sudo journalctl -u docker.service > ./e2e/debug/docker-service.log + tar -czf e2e-debug.tar.gz -C e2e debug + fi + when: on_fail + - store_artifacts: + path: e2e-debug.tar.gz + - run: + name: Ensure Docker Compose is stopped + command: make ci-compose-down + when: always + - store_test_results: + path: . release: executor: common steps: @@ -313,3 +338,19 @@ workflows: jobs: - build_image: context: control-plane-release + + nightly_spock6: + # Runs independent of any commit activity, so drift introduced by an + # upstream Spock 6 nightly build gets caught even if nobody touches + # this repo that day. The e2e test itself points at the floating + # spock6DevImage tag (see e2e/spock6_add_node_test.go), so this job + # needs no extra plumbing to track "latest" - only the schedule. + triggers: + - schedule: + cron: "0 6 * * *" + filters: + branches: + only: + - main + jobs: + - test_e2e_spock6 From 875a554838fd137e257ad85214961c2a86f575df Mon Sep 17 00:00:00 2001 From: moizpgedge Date: Thu, 20 Aug 2026 02:50:35 +0500 Subject: [PATCH 4/7] fix: e2e version check crash on Spock 6 dev tag MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The pre-existing TestCreateDbWithVersions test used log.Fatalf instead of t.Fatalf when a database's live Postgres version didn't match its declared version. log.Fatalf calls os.Exit(1), killing the whole test binary and reporting every other in-flight parallel subtest as a bare failure instead of just the one mismatch — this was failing ci/circleci: test_e2e on PR #457. The Spock 6 dev manifest entry points at a floating image tag whose resolved Postgres minor can drift past its declared version at any time, so an exact-match check is fundamentally incompatible with it. Switch to t.Fatalf so a mismatch fails only its own subtest, and relax the version check to major-only specifically for Spock >= 6 entries; pinned Spock 5.x entries keep the exact-match check. Also update the needsOutputPluginLibraries doc comment to record that setting output_plugin_libraries on a Postgres minor that predates the gate is confirmed to hard-fail startup (unrecognizedonfiguration parameter), not silently no-op — resolving what was previously an open question in that comment. PLAT-718 --- e2e/custom_db_create_test.go | 29 ++++++++++++++++++++--------- server/internal/postgres/gucs.go | 14 ++++++++------ 2 files changed, 28 insertions(+), 15 deletions(-) diff --git a/e2e/custom_db_create_test.go b/e2e/custom_db_create_test.go index be5d0009..cf6dd39a 100644 --- a/e2e/custom_db_create_test.go +++ b/e2e/custom_db_create_test.go @@ -5,8 +5,8 @@ package e2e import ( "context" "fmt" - "log" "slices" + "strconv" "strings" "testing" "time" @@ -98,7 +98,7 @@ func TestCreateDbWithVersions(t *testing.T) { Password: password, } - verifyPgVersion(ctx, db, primaryOpts, version.PostgresVersion, t) + verifyPgVersion(ctx, db, primaryOpts, version.PostgresVersion, version.SpockVersion, t) verifyPrimaryNodes(ctx, db, primaryOpts, t) } @@ -110,7 +110,7 @@ func TestCreateDbWithVersions(t *testing.T) { Username: username, Password: password, } - verifyPgVersion(ctx, db, connOpts, version.PostgresVersion, t) + verifyPgVersion(ctx, db, connOpts, version.PostgresVersion, version.SpockVersion, t) verifyReplicasNodes(ctx, db, connOpts, t) } @@ -206,19 +206,30 @@ func verifyReplicasNodes(ctx context.Context, db *DatabaseFixture, }) } -// Validate postgresql version +// Validate postgresql version. Spock 6 manifest entries point at a +// floating/mutable dev image tag (see version-manifest.json), so their +// resolved Postgres minor can drift past the declared version at any +// time - only the major version is checked for those. Pinned versions +// (Spock <= 5) are still checked for an exact match. func verifyPgVersion(ctx context.Context, db *DatabaseFixture, - primaryOpts ConnectionOptions, expectedVersion string, t testing.TB) { + primaryOpts ConnectionOptions, expectedVersion string, spockVersion string, t testing.TB) { db.WithConnection(ctx, primaryOpts, t, func(conn *pgx.Conn) { var versionStr string err := conn.QueryRow(ctx, "SELECT version()").Scan(&versionStr) if err != nil { - log.Fatalf("Failed to fetch PostgreSQL version: %v", err) + t.Fatalf("Failed to fetch PostgreSQL version: %v", err) } - if !strings.Contains(versionStr, expectedVersion) { - log.Fatalf("Expected PostgreSQL version %s, but got: %s", expectedVersion, versionStr) + + versionToMatch := expectedVersion + spockMajorStr, _, _ := strings.Cut(spockVersion, ".") + if spockMajor, err := strconv.Atoi(spockMajorStr); err == nil && spockMajor >= 6 { + versionToMatch, _, _ = strings.Cut(expectedVersion, ".") + } + + if !strings.Contains(versionStr, versionToMatch) { + t.Fatalf("Expected PostgreSQL version %s, but got: %s", versionToMatch, versionStr) } - tLogf(t, "PostgreSQL version validation passed (found %s)\n", expectedVersion) + tLogf(t, "PostgreSQL version validation passed (found %s)\n", versionStr) }) } diff --git a/server/internal/postgres/gucs.go b/server/internal/postgres/gucs.go index 4fe60c62..186c9473 100644 --- a/server/internal/postgres/gucs.go +++ b/server/internal/postgres/gucs.go @@ -33,12 +33,14 @@ var minOutputPluginLibrariesVersions = map[uint64]*ds.Version{ // major >= 6 is treated as an unconditional yes here. This is a deliberate // trade-off, not a fully general fix: every Spock 6 build seen in practice // so far ships on a Postgres minor at or past the relevant gate, so this -// resolves the real failure above; it has not been verified whether setting -// this GUC against a hypothetical Postgres minor that predates the gate -// (and so may not recognize the parameter at all) is itself harmless or -// would fail Postgres startup. If Spock 6 is ever built against such a -// minor, this assumption needs revisiting. Spock 5.x behavior is unchanged -// — it still depends solely on the exact Postgres minor check below. +// resolves the real failure above. Confirmed directly (via `SHOW +// output_plugin_libraries` against a Postgres minor below the gate) that +// setting this GUC on a minor that predates it is NOT harmless — Postgres +// rejects it outright with "unrecognized configuration parameter", which +// would fail startup, not just no-op. If Spock 6 is ever built against +// such a minor, this unconditional-yes needs revisiting. Spock 5.x +// behavior is unchanged — it still depends solely on the exact Postgres +// minor check below, which is exactly why it never hits this failure mode. func needsOutputPluginLibraries(version *ds.PgEdgeVersion) bool { if version == nil || version.PostgresVersion == nil { return false From 5d7c834951ccfcae89adc010bf94170d61fd974e Mon Sep 17 00:00:00 2001 From: moizpgedge Date: Thu, 20 Aug 2026 03:05:39 +0500 Subject: [PATCH 5/7] ci: preserve debug output on Spock 6 e2e failure test_e2e_spock6 archives ./e2e/debug on failure, but never set E2E_DEBUG=1, so the fixture never actually wrote anything there for the archive step to pick up. The existing test_e2e job already sets this; test_e2e_spock6 just missed it. PLAT-718 --- .circleci/config.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.circleci/config.yml b/.circleci/config.yml index 9932db9f..d4c1133d 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -145,7 +145,7 @@ jobs: name: Run Spock 6 add-node e2e test against the latest dev image command: | make ci-compose-detached - make test-e2e E2E_FIXTURE=ci E2E_RUN='^TestSpock6AddNode$$' TEST_RERUN_FAILS=2 + make test-e2e E2E_DEBUG=1 E2E_FIXTURE=ci E2E_RUN='^TestSpock6AddNode$$' TEST_RERUN_FAILS=2 - run: name: Archive debug output command: | From 974c883bf34725ff538c4a0f2ab13847c59f256d Mon Sep 17 00:00:00 2001 From: moizpgedge Date: Fri, 21 Aug 2026 04:26:46 +0500 Subject: [PATCH 6/7] fix: correct spock 6 manifest postgres version MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The spock 6 manifest entry declared postgres_version 18.4, but the floating dev image tag it points at actually resolves to 18.6. That stale declaration was the real cause of output_plugin_libraries missing spock_output — not a gap tied to spock's major version. Correcting the declared version lets the existing minor-version-only gate in needsOutputPluginLibraries work as designed, so the spock-major-based workaround it had grown is no longer needed and has been reverted, along with its now-obsolete test cases. Updated the e2e add-node test and the manifest loader test to expect the corrected version too. Also switched the spock 6 e2e job from a nightly to a weekly schedule: the upstream dev image only gets a new build every few weeks in practice, so nightly was mostly no-op runs. Finally, corrected the doc comment on EnablePeerSubscriptions. Its re-enable step is redundant with end.go's own subscription enablement by the time that phase runs, but is kept because VerifySubscriptionReplicatingResource needs an already-enabled subscription to check against — removing it without relocating verification would make that check fail every time. PLAT-718 --- .circleci/config.yml | 15 ++++---- e2e/spock6_add_node_test.go | 2 +- .../database/operations/populate_nodes.go | 34 +++++++++++-------- .../swarm/manifest_loader_test.go | 2 +- .../orchestrator/swarm/version-manifest.json | 2 +- server/internal/postgres/gucs.go | 27 --------------- server/internal/postgres/gucs_test.go | 3 -- 7 files changed, 32 insertions(+), 53 deletions(-) diff --git a/.circleci/config.yml b/.circleci/config.yml index d4c1133d..6170c60d 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -339,15 +339,18 @@ workflows: - build_image: context: control-plane-release - nightly_spock6: + weekly_spock6: # Runs independent of any commit activity, so drift introduced by an - # upstream Spock 6 nightly build gets caught even if nobody touches - # this repo that day. The e2e test itself points at the floating - # spock6DevImage tag (see e2e/spock6_add_node_test.go), so this job - # needs no extra plumbing to track "latest" - only the schedule. + # upstream Spock 6 build gets caught even if nobody touches this repo + # that week. Weekly rather than nightly: the upstream dev image only + # gets a new build every few weeks in practice, so nightly would just + # be ~20+ no-op runs for every one that actually catches something. + # The e2e test itself points at the floating spock6DevImage tag (see + # e2e/spock6_add_node_test.go), so this job needs no extra plumbing + # to track "latest" - only the schedule. triggers: - schedule: - cron: "0 6 * * *" + cron: "0 6 * * 1" filters: branches: only: diff --git a/e2e/spock6_add_node_test.go b/e2e/spock6_add_node_test.go index cf9a048b..0944d53c 100644 --- a/e2e/spock6_add_node_test.go +++ b/e2e/spock6_add_node_test.go @@ -55,7 +55,7 @@ func TestSpock6AddNode(t *testing.T) { db := fixture.NewDatabaseFixture(ctx, t, &controlplane.CreateDatabaseRequest{ Spec: &controlplane.DatabaseSpec{ DatabaseName: dbName, - PostgresVersion: pointerTo("18.4"), + PostgresVersion: pointerTo("18.6"), SpockVersion: pointerTo("6"), Port: pointerTo(0), PatroniPort: pointerTo(0), diff --git a/server/internal/database/operations/populate_nodes.go b/server/internal/database/operations/populate_nodes.go index 3bb587e7..1e0bebaf 100644 --- a/server/internal/database/operations/populate_nodes.go +++ b/server/internal/database/operations/populate_nodes.go @@ -85,21 +85,27 @@ func PopulateNodes(existing, new []*NodeResources) (*resource.State, error) { return merged, nil } -// EnablePeerSubscriptions returns a diff that enables the peer subscriptions -// addPeerResources creates disabled. It must be applied as a separate, later -// phase than PopulateNodes' own state. +// EnablePeerSubscriptions returns a diff that re-enables the peer +// subscriptions addPeerResources creates disabled, and verifies each one +// actually starts replicating. It must be applied as a separate, later phase +// than PopulateNodes' own state. // -// addPeerResources creates each peer->new-node subscription disabled so the -// peer-catchup chain (SyncEvent -> WaitForSyncEvent -> PeerCatchup -> -// LagTracker -> ReplicationSlotAdvanceFromCTS -> ReplicationOriginAdvance) -// can run without a live subscriber racing that setup. But a single -// resource.State can only express one desired value per identifier, so that -// same state can never also declare "now enable it" — nothing else in the -// codebase ever does, which left these subscriptions permanently disabled. -// This returns a second state that re-declares the same SubscriptionResource -// identifiers with Disabled: false; applied after the populate phase is -// fully persisted, its diff sees disabled->enabled and calls Update, which -// is what actually flips sub_enabled in spock.subscription. +// The re-enable here is not the only thing that flips these subscriptions +// back on: end.go's EndState() unconditionally redeclares every peer-pair +// SubscriptionResource as part of the final desired state of every +// create/update-database operation, which enables them regardless of what +// this phase does. So functionally this phase's enable is redundant with +// that later one — a no-op Update by the time end.go's phase runs. +// +// It's kept anyway because it's the anchor for +// VerifySubscriptionReplicatingResource below: that check needs a +// SubscriptionResource in the graph that is actually enabled by the time it +// runs, so we can fail loudly, right here, if a peer subscription never +// starts replicating — instead of only finding out much later (or not at +// all, since end.go's phase has no equivalent verification). Dropping the +// enable without relocating the verify step would leave the verify checking +// a subscription that's still deliberately disabled from the populate +// phase, so it would fail every time. func EnablePeerSubscriptions(existing, new []*NodeResources) (*resource.State, error) { existingNodeNames := make([]string, len(existing)) for i, n := range existing { diff --git a/server/internal/orchestrator/swarm/manifest_loader_test.go b/server/internal/orchestrator/swarm/manifest_loader_test.go index 5814c744..3d6db868 100644 --- a/server/internal/orchestrator/swarm/manifest_loader_test.go +++ b/server/internal/orchestrator/swarm/manifest_loader_test.go @@ -688,7 +688,7 @@ func TestEmbeddedManifestValid_Spock6DevEntryNotDefault(t *testing.T) { t.Fatalf("embedded manifest cannot be parsed: %v", err) } - spock6 := ds.MustParsePgEdgeVersion("18.4", "6") + spock6 := ds.MustParsePgEdgeVersion("18.6", "6") imgs, err := v.GetImages(spock6) if err != nil { t.Fatalf("expected embedded manifest to have a spock6 entry: %v", err) diff --git a/server/internal/orchestrator/swarm/version-manifest.json b/server/internal/orchestrator/swarm/version-manifest.json index 0b83c3a1..55bf6c01 100644 --- a/server/internal/orchestrator/swarm/version-manifest.json +++ b/server/internal/orchestrator/swarm/version-manifest.json @@ -94,7 +94,7 @@ "default": true }, { - "postgres_version": "18.4", + "postgres_version": "18.6", "spock_version": "6", "image": "pgedge-postgres:18-spock6-standard", "stability": "dev" diff --git a/server/internal/postgres/gucs.go b/server/internal/postgres/gucs.go index 186c9473..58a17ece 100644 --- a/server/internal/postgres/gucs.go +++ b/server/internal/postgres/gucs.go @@ -19,37 +19,10 @@ var minOutputPluginLibrariesVersions = map[uint64]*ds.Version{ // needsOutputPluginLibraries reports whether the given Postgres version // requires output_plugin_libraries to be set in order to allow spock_output // to create replication slots. -// -// Spock 6 manifest entries deliberately point at a floating/mutable image -// tag (see version-manifest.json) rather than a pinned build number, so its -// declared postgres_version can drift out of sync with whatever Postgres -// minor the tag actually resolves to at any given moment. That drift caused -// a real, live-reproduced failure ("library spock_output may not be used as -// an output plugin") when the floating tag moved past the gate threshold for -// a minor this function had no way to know about ahead of time, since it -// only ever sees the declared version, not what's actually running. -// -// Rather than track the floating tag's exact current resolution, Spock -// major >= 6 is treated as an unconditional yes here. This is a deliberate -// trade-off, not a fully general fix: every Spock 6 build seen in practice -// so far ships on a Postgres minor at or past the relevant gate, so this -// resolves the real failure above. Confirmed directly (via `SHOW -// output_plugin_libraries` against a Postgres minor below the gate) that -// setting this GUC on a minor that predates it is NOT harmless — Postgres -// rejects it outright with "unrecognized configuration parameter", which -// would fail startup, not just no-op. If Spock 6 is ever built against -// such a minor, this unconditional-yes needs revisiting. Spock 5.x -// behavior is unchanged — it still depends solely on the exact Postgres -// minor check below, which is exactly why it never hits this failure mode. func needsOutputPluginLibraries(version *ds.PgEdgeVersion) bool { if version == nil || version.PostgresVersion == nil { return false } - if version.SpockVersion != nil { - if spockMajor, ok := version.SpockVersion.Major(); ok && spockMajor >= 6 { - return true - } - } pgVersion := version.PostgresVersion.MajorMinorVersion() major, ok := pgVersion.Major() if !ok { diff --git a/server/internal/postgres/gucs_test.go b/server/internal/postgres/gucs_test.go index 64f0701a..dd7d1606 100644 --- a/server/internal/postgres/gucs_test.go +++ b/server/internal/postgres/gucs_test.go @@ -29,9 +29,6 @@ func TestDefaultGUCsOutputPluginLibraries(t *testing.T) { {name: "pg18 at gate", version: ds.MustParsePgEdgeVersion("18.6", "4"), expectedPresent: true}, {name: "future major", version: ds.MustParsePgEdgeVersion("19.0", "4"), expectedPresent: true}, {name: "older major", version: ds.MustParsePgEdgeVersion("15.10", "4"), expectedPresent: false}, - {name: "spock 6 on pg18 below gate", version: ds.MustParsePgEdgeVersion("18.4", "6"), expectedPresent: true}, - {name: "spock 6 on pg16 below gate", version: ds.MustParsePgEdgeVersion("16.10", "6"), expectedPresent: true}, - {name: "spock 5 on pg18 below gate stays ungated", version: ds.MustParsePgEdgeVersion("18.4", "5"), expectedPresent: false}, } { t.Run(tc.name, func(t *testing.T) { gucs := postgres.DefaultGUCs(tc.version) From ed2b97acc610eddd92a8f5cb2668926a815e2b71 Mon Sep 17 00:00:00 2001 From: moizpgedge Date: Fri, 21 Aug 2026 18:43:20 +0500 Subject: [PATCH 7/7] fix: remove subscription enable/verify steps per review feedback --- e2e/spock6_add_node_test.go | 9 +- ...wo_nodes_to_three_nodes_with_populate.json | 53 +------- .../database/operations/populate_nodes.go | 68 ---------- .../database/operations/update_database.go | 8 -- server/internal/database/resources.go | 1 - ...erify_subscription_replicating_resource.go | 117 ------------------ 6 files changed, 9 insertions(+), 247 deletions(-) delete mode 100644 server/internal/database/verify_subscription_replicating_resource.go diff --git a/e2e/spock6_add_node_test.go b/e2e/spock6_add_node_test.go index 0944d53c..03940a3c 100644 --- a/e2e/spock6_add_node_test.go +++ b/e2e/spock6_add_node_test.go @@ -14,9 +14,9 @@ import ( ) // spock6DevImage is a floating/mutable tag tracking the latest Spock 6 -// development build. Not pinned to a specific build number: nightly CI -// re-running this test picks up whatever the tag currently resolves to, -// with no extra plumbing needed to point CI at "latest." +// development build. Not pinned to a specific build number: the scheduled +// CI job re-running this test picks up whatever the tag currently resolves +// to, with no extra plumbing needed to point CI at "latest." const spock6DevImage = "ghcr.io/pgedge/pgedge-postgres:18-spock6-standard" // TestSpock6AddNode validates the add-node workflow end-to-end against a @@ -25,8 +25,7 @@ const spock6DevImage = "ghcr.io/pgedge/pgedge-postgres:18-spock6-standard" // constraints, since spock6 manifest entries are deliberately "dev" // stability and never auto-selected), adds a 3rd node, and confirms the // full mesh reaches "replicating" — exercising the Spock-major-gated -// spock.progress query (PeerCatchupResource) and the verify-replicating -// step (VerifySubscriptionReplicatingResource) added in this same ticket. +// spock.progress query (PeerCatchupResource). func TestSpock6AddNode(t *testing.T) { t.Parallel() diff --git a/server/internal/database/operations/golden_test/TestUpdateDatabase/two_nodes_to_three_nodes_with_populate.json b/server/internal/database/operations/golden_test/TestUpdateDatabase/two_nodes_to_three_nodes_with_populate.json index 7bb8c13e..869f0e4a 100644 --- a/server/internal/database/operations/golden_test/TestUpdateDatabase/two_nodes_to_three_nodes_with_populate.json +++ b/server/internal/database/operations/golden_test/TestUpdateDatabase/two_nodes_to_three_nodes_with_populate.json @@ -159,40 +159,6 @@ } ] ], - [ - [ - { - "type": "update", - "resource_id": "database.subscription::n2:n3:test", - "reason": "has_diff", - "diff": [ - { - "value": false, - "op": "replace", - "path": "/disabled" - }, - { - "value": [ - { - "id": "n2:n3:test", - "type": "database.replication_origin_advance" - } - ], - "op": "replace", - "path": "/extra_dependencies" - } - ] - } - ], - [ - { - "type": "create", - "resource_id": "database.verify_subscription_replicating::n2:n3:test", - "reason": "does_not_exist", - "diff": null - } - ] - ], [ [ { @@ -237,9 +203,9 @@ "reason": "has_diff", "diff": [ { - "value": null, + "value": false, "op": "replace", - "path": "/extra_dependencies" + "path": "/disabled" } ] }, @@ -259,12 +225,12 @@ [ { "type": "delete", - "resource_id": "database.roles_source::n3", + "resource_id": "database.replication_origin_advance::n2:n3:test", "diff": null }, { "type": "delete", - "resource_id": "database.verify_subscription_replicating::n2:n3:test", + "resource_id": "database.roles_source::n3", "diff": null } ], @@ -273,16 +239,7 @@ "type": "delete", "resource_id": "database.dump_roles::n1", "diff": null - } - ], - [ - { - "type": "delete", - "resource_id": "database.replication_origin_advance::n2:n3:test", - "diff": null - } - ], - [ + }, { "type": "delete", "resource_id": "database.replication_slot_advance_from_cts::n2:n3:test", diff --git a/server/internal/database/operations/populate_nodes.go b/server/internal/database/operations/populate_nodes.go index 1e0bebaf..9aa84eef 100644 --- a/server/internal/database/operations/populate_nodes.go +++ b/server/internal/database/operations/populate_nodes.go @@ -85,74 +85,6 @@ func PopulateNodes(existing, new []*NodeResources) (*resource.State, error) { return merged, nil } -// EnablePeerSubscriptions returns a diff that re-enables the peer -// subscriptions addPeerResources creates disabled, and verifies each one -// actually starts replicating. It must be applied as a separate, later phase -// than PopulateNodes' own state. -// -// The re-enable here is not the only thing that flips these subscriptions -// back on: end.go's EndState() unconditionally redeclares every peer-pair -// SubscriptionResource as part of the final desired state of every -// create/update-database operation, which enables them regardless of what -// this phase does. So functionally this phase's enable is redundant with -// that later one — a no-op Update by the time end.go's phase runs. -// -// It's kept anyway because it's the anchor for -// VerifySubscriptionReplicatingResource below: that check needs a -// SubscriptionResource in the graph that is actually enabled by the time it -// runs, so we can fail loudly, right here, if a peer subscription never -// starts replicating — instead of only finding out much later (or not at -// all, since end.go's phase has no equivalent verification). Dropping the -// enable without relocating the verify step would leave the verify checking -// a subscription that's still deliberately disabled from the populate -// phase, so it would fail every time. -func EnablePeerSubscriptions(existing, new []*NodeResources) (*resource.State, error) { - existingNodeNames := make([]string, len(existing)) - for i, n := range existing { - existingNodeNames[i] = n.NodeName - } - - enable := resource.NewState() - for _, node := range new { - if node.SourceNode == "" { - continue - } - dbName := node.DatabaseName - for _, peer := range existingNodeNames { - if peer == node.NodeName || peer == node.SourceNode { - continue - } - err := enable.AddResource( - &database.SubscriptionResource{ - DatabaseName: dbName, - SubscriberNode: node.NodeName, - ProviderNode: peer, - Disabled: false, - ExtraDependencies: []resource.Identifier{ - database.ReplicationOriginAdvanceResourceIdentifier(peer, node.NodeName, dbName), - }, - }, - // Verify the enable actually took effect. Same phase, not a - // separate one: this is a new resource type/identifier, not - // a re-declaration of an existing one, so it can safely - // depend on the SubscriptionResource declared just above - // within this same state and run after it in the same - // apply pass. - &database.VerifySubscriptionReplicatingResource{ - DatabaseName: dbName, - SubscriberNode: node.NodeName, - ProviderNode: peer, - }, - ) - if err != nil { - return nil, fmt.Errorf("failed to add peer-enable resource to 'enable' state: %w", err) - } - } - } - - return enable, nil -} - func addPeerResources( state *resource.State, dbName string, diff --git a/server/internal/database/operations/update_database.go b/server/internal/database/operations/update_database.go index 3252545f..3ec57341 100644 --- a/server/internal/database/operations/update_database.go +++ b/server/internal/database/operations/update_database.go @@ -192,14 +192,6 @@ func addNodesStates(updates, adds []*NodeResources) ([]*resource.State, error) { states = append(states, populate) } - enable, err := EnablePeerSubscriptions(updates, adds) - if err != nil { - return nil, err - } - if enable != nil { - states = append(states, enable) - } - return states, nil } diff --git a/server/internal/database/resources.go b/server/internal/database/resources.go index 511bd54b..998b1190 100644 --- a/server/internal/database/resources.go +++ b/server/internal/database/resources.go @@ -13,7 +13,6 @@ func RegisterResourceTypes(registry *resource.Registry) { resource.RegisterResourceType[*LagTrackerCommitTimestampResource](registry, ResourceTypeLagTrackerCommitTS) resource.RegisterResourceType[*ReplicationSlotAdvanceFromCTSResource](registry, ResourceTypeReplicationSlotAdvanceFromCTS) resource.RegisterResourceType[*ReplicationOriginAdvanceResource](registry, ResourceTypeReplicationOriginAdvance) - resource.RegisterResourceType[*VerifySubscriptionReplicatingResource](registry, ResourceTypeVerifySubscriptionReplicating) resource.RegisterResourceType[*PeerCatchupResource](registry, ResourceTypePeerCatchup) resource.RegisterResourceType[*SwitchoverResource](registry, ResourceTypeSwitchover) resource.RegisterResourceType[*PostgresDatabaseResource](registry, ResourceTypePostgresDatabase) diff --git a/server/internal/database/verify_subscription_replicating_resource.go b/server/internal/database/verify_subscription_replicating_resource.go deleted file mode 100644 index d3de7594..00000000 --- a/server/internal/database/verify_subscription_replicating_resource.go +++ /dev/null @@ -1,117 +0,0 @@ -package database - -import ( - "context" - "fmt" - "time" - - "github.com/pgEdge/control-plane/server/internal/postgres" - "github.com/pgEdge/control-plane/server/internal/resource" -) - -var _ resource.Resource = (*VerifySubscriptionReplicatingResource)(nil) - -const ResourceTypeVerifySubscriptionReplicating resource.Type = "database.verify_subscription_replicating" - -func VerifySubscriptionReplicatingResourceIdentifier(providerNode, subscriberNode, databaseName string) resource.Identifier { - return resource.Identifier{ - Type: ResourceTypeVerifySubscriptionReplicating, - ID: fmt.Sprintf("%s:%s:%s", providerNode, subscriberNode, databaseName), - } -} - -// VerifySubscriptionReplicatingResource polls a subscription's status until -// it reaches "replicating", failing loudly if it doesn't within a bounded -// wait. Mirrors Spock's own zodan.sql reference add-node flow -// (spock.verify_subscription_replicating), which Control Plane's pipeline -// otherwise has no equivalent of: nothing previously checked that an -// enabled subscription's apply worker actually started, so a subscription -// that never starts replicating could silently leave a node missing data -// with no error anywhere. This resource doesn't fix that underlying -// possibility — an apply worker failing to start is Spock's own concern — -// it turns it from silent, permanent data loss into a visible, actionable -// task failure instead. -type VerifySubscriptionReplicatingResource struct { - DatabaseName string `json:"database_name"` - ProviderNode string `json:"provider_node"` - SubscriberNode string `json:"subscriber_node"` -} - -func (r *VerifySubscriptionReplicatingResource) ResourceVersion() string { return "1" } -func (r *VerifySubscriptionReplicatingResource) DiffIgnore() []string { return nil } - -// Subscription status is local to the subscriber (spock.sub_show_status() -// reports on incoming subscriptions), so this must run on that host. -func (r *VerifySubscriptionReplicatingResource) Executor() resource.Executor { - return resource.PrimaryExecutor(r.SubscriberNode) -} - -func (r *VerifySubscriptionReplicatingResource) Identifier() resource.Identifier { - return VerifySubscriptionReplicatingResourceIdentifier(r.ProviderNode, r.SubscriberNode, r.DatabaseName) -} - -func (r *VerifySubscriptionReplicatingResource) Dependencies() []resource.Identifier { - return []resource.Identifier{ - SubscriptionResourceIdentifier(r.ProviderNode, r.SubscriberNode, r.DatabaseName), - } -} - -func (r *VerifySubscriptionReplicatingResource) TypeDependencies() []resource.Type { return nil } - -func (r *VerifySubscriptionReplicatingResource) Refresh(ctx context.Context, rc *resource.Context) error { - subscriber, err := GetPrimaryInstance(ctx, rc, r.SubscriberNode) - if err != nil { - return fmt.Errorf("failed to get subscriber instance for node %q: %w", r.SubscriberNode, err) - } - conn, err := subscriber.Connection(ctx, rc, r.DatabaseName) - if err != nil { - return fmt.Errorf("failed to connect to subscriber %q: %w", r.SubscriberNode, err) - } - defer conn.Close(ctx) - - // Matches Spock's own verify_subscription_replicating default wait - // (120s) with headroom, since ours is the second such check in the - // pipeline (after whatever WaitForSyncEventResource already waited - // for) rather than the only one. - const ( - pollInterval = 2 * time.Second - waitTimeout = 3 * time.Minute - ) - waitCtx, cancel := context.WithTimeout(ctx, waitTimeout) - defer cancel() - - var lastStatus string - for { - status, err := postgres.GetSubscriptionStatus(r.ProviderNode, r.SubscriberNode).Scalar(waitCtx, conn) - if err != nil { - if postgres.IsSpockNodeNotConfigured(err) { - return resource.ErrNotFound - } - return fmt.Errorf("failed to check subscription status: %w", err) - } - if status == postgres.SubStatusReplicating { - return nil - } - lastStatus = status - - select { - case <-waitCtx.Done(): - return fmt.Errorf( - "subscription %s->%s did not reach %q status within %s (last status: %q)", - r.ProviderNode, r.SubscriberNode, postgres.SubStatusReplicating, waitTimeout, lastStatus) - case <-time.After(pollInterval): - } - } -} - -func (r *VerifySubscriptionReplicatingResource) Create(ctx context.Context, rc *resource.Context) error { - return r.Refresh(ctx, rc) -} - -func (r *VerifySubscriptionReplicatingResource) Update(ctx context.Context, rc *resource.Context) error { - return r.Refresh(ctx, rc) -} - -func (r *VerifySubscriptionReplicatingResource) Delete(ctx context.Context, rc *resource.Context) error { - return nil -}