From 04e6991a2434c46e491d0dc9f43690361a2619cf Mon Sep 17 00:00:00 2001 From: "renovate[bot]" <29139614+renovate[bot]@users.noreply.github.com> Date: Mon, 20 Jul 2026 14:14:52 +0000 Subject: [PATCH 1/3] fix(deps): update module github.com/cloudnative-pg/cnpg-i to v0.6.0 | datasource | package | from | to | | ---------- | -------------------------------- | ------ | ------ | | go | github.com/cloudnative-pg/cnpg-i | v0.5.0 | v0.6.0 | Signed-off-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com> --- go.mod | 2 +- go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index 5a95b0e..d3e56db 100644 --- a/go.mod +++ b/go.mod @@ -7,7 +7,7 @@ require ( github.com/cloudnative-pg/api v1.30.0 github.com/cloudnative-pg/barman-cloud v0.5.2-0.20260709152604-43158c204df1 github.com/cloudnative-pg/cloudnative-pg v1.30.0 - github.com/cloudnative-pg/cnpg-i v0.5.0 + github.com/cloudnative-pg/cnpg-i v0.6.0 github.com/cloudnative-pg/cnpg-i-machinery v0.4.2 github.com/cloudnative-pg/machinery v0.5.0 github.com/onsi/ginkgo/v2 v2.32.0 diff --git a/go.sum b/go.sum index ada804d..d64259e 100644 --- a/go.sum +++ b/go.sum @@ -24,8 +24,8 @@ github.com/cloudnative-pg/barman-cloud v0.5.2-0.20260709152604-43158c204df1 h1:+ github.com/cloudnative-pg/barman-cloud v0.5.2-0.20260709152604-43158c204df1/go.mod h1:ZQLkdpk44FW5/BGWzABTOEcV9qPbwC+rbdscg2I8mBI= github.com/cloudnative-pg/cloudnative-pg v1.30.0 h1:fnhVq44xXx97MNiuvJsPrX1vSjYbgdyBK5MSGfdHdp0= github.com/cloudnative-pg/cloudnative-pg v1.30.0/go.mod h1:QkolwBOWZ+GvAiJt6KpDSymwkpf0K19/p4Q6MQlTM8U= -github.com/cloudnative-pg/cnpg-i v0.5.0 h1:/TOzpNT6cwNgrpftTtrnLKdoHgMwd+88vZgXjlVgXeE= -github.com/cloudnative-pg/cnpg-i v0.5.0/go.mod h1:7Gh4+UzhBpGhr4DreB1GN9wGYfvxwXCXZUyVt3zE/3I= +github.com/cloudnative-pg/cnpg-i v0.6.0 h1:LA//DLkFOLIjU0ASOpFkydZhGir9IAIDfgSsTTX9IpU= +github.com/cloudnative-pg/cnpg-i v0.6.0/go.mod h1:4kcpLAj+feMTnh0TVadXRASdVnEhBytNSm/BvktLgmI= github.com/cloudnative-pg/cnpg-i-machinery v0.4.2 h1:0reS9MtyLYINHXQ/MfxJ9jp39hhBf8e3Qdj+T5Nsq6I= github.com/cloudnative-pg/cnpg-i-machinery v0.4.2/go.mod h1:gvrKabgxXq0zGthXGucemDdsxakLEQDMxn43M4HLW30= github.com/cloudnative-pg/machinery v0.5.0 h1:hhTnkzn+AiN3NmbjCQ6RXj5rfqV3K6arzq6kdXAzcnQ= From 51ec351614a3e9a6cde194645428916afe0e8191 Mon Sep 17 00:00:00 2001 From: Armando Ruocco Date: Wed, 15 Jul 2026 15:20:26 +0200 Subject: [PATCH 2/3] fix(walrestore): serve pg_rewind without prefetching and flag machinery The operator now tells WAL restore plugins when a restore request is made on behalf of pg_rewind (cloudnative-pg/cnpg-i#351). pg_rewind walks the timeline backwards, fetches every WAL file it needs exactly once, and treats any restore failure as fatal, so both optimizations meant for an instance in recovery must stay off: prefetching the following segments is wasted work that ends in an archive miss, and the end-of-wal-stream flag recorded by such a miss makes a later invocation fail on a segment that is available in the archive, aborting the whole rewind. The cnpg-i dependency points to a pseudo-version of that pull request and will be moved to the next tagged release once it is available. Ref: cloudnative-pg/cloudnative-pg#11200 Signed-off-by: Armando Ruocco --- internal/cnpgi/common/wal.go | 55 +++++++++++++++++++++++++------ internal/cnpgi/common/wal_test.go | 45 +++++++++++++++++++++++++ 2 files changed, 90 insertions(+), 10 deletions(-) diff --git a/internal/cnpgi/common/wal.go b/internal/cnpgi/common/wal.go index bb7d3e0..8a196cc 100644 --- a/internal/cnpgi/common/wal.go +++ b/internal/cnpgi/common/wal.go @@ -27,6 +27,7 @@ import ( "path" "time" + barmanapi "github.com/cloudnative-pg/barman-cloud/pkg/api" "github.com/cloudnative-pg/barman-cloud/pkg/archiver" barmanCommand "github.com/cloudnative-pg/barman-cloud/pkg/command" barmanCredentials "github.com/cloudnative-pg/barman-cloud/pkg/credentials" @@ -249,9 +250,11 @@ func (w WALServiceImplementation) Restore( "Restoring WAL file", "objectStore", objectStore.Name, "serverName", serverName, - "walName", walName) + "walName", walName, + "mode", request.GetMode()) return &wal.WALRestoreResult{}, w.restoreFromBarmanObjectStore( - ctx, configuration.Cluster, &objectStore, serverName, walName, destinationPath) + ctx, configuration.Cluster, &objectStore, serverName, walName, destinationPath, + request.GetMode() == wal.WALRestoreRequest_MODE_REWIND) } // resolveRestoreObjectStore selects the object store and server name to use when @@ -288,6 +291,7 @@ func (w WALServiceImplementation) restoreFromBarmanObjectStore( serverName string, walName string, destinationPath string, + rewindMode bool, ) error { contextLogger := log.FromContext(ctx) startTime := time.Now() @@ -331,8 +335,10 @@ func (w WALServiceImplementation) restoreFromBarmanObjectStore( return nil } - // We skip this step if streaming connection is not available - if isStreamingAvailable(cluster, w.InstanceName) { + // Step 2: return error if the end-of-wal-stream flag is set. + // We skip this step if the flag machinery does not apply to this invocation + useEndOfWALStreamFlag := shouldUseEndOfWALStreamFlag(cluster, w.InstanceName, rewindMode) + if useEndOfWALStreamFlag { if err := checkEndOfWALStreamFlag(walRestorer); err != nil { return err } @@ -340,10 +346,7 @@ func (w WALServiceImplementation) restoreFromBarmanObjectStore( // Step 3: gather the WAL files names to restore. If the required file isn't a regular WAL, we download it directly. var walFilesList []string - maxParallel := 1 - if barmanConfiguration.Wal != nil && barmanConfiguration.Wal.MaxParallel > 1 { - maxParallel = barmanConfiguration.Wal.MaxParallel - } + maxParallel := maxWALFilesPerInvocation(barmanConfiguration, rewindMode) if IsWALFile(walName) { // If this is a regular WAL file, we try to prefetch if walFilesList, err = gatherWALFilesToRestore(walName, maxParallel); err != nil { @@ -365,9 +368,9 @@ func (w WALServiceImplementation) restoreFromBarmanObjectStore( return classifyWALRestoreError(walStatus[0].WalName, walStatus[0].Err) } - // We skip this step if streaming connection is not available + // We skip this step if the flag machinery does not apply to this invocation endOfWALStream := isEndOfWALStream(walStatus) - if isStreamingAvailable(cluster, w.InstanceName) && endOfWALStream { + if useEndOfWALStreamFlag && endOfWALStream { contextLogger.Info( "Set end-of-wal-stream flag as one of the WAL files to be prefetched was not found") @@ -415,6 +418,38 @@ func (w WALServiceImplementation) SetFirstRequired( panic("implement me") } +// maxWALFilesPerInvocation returns how many WAL files a single restore +// invocation is allowed to fetch, the requested one included. Prefetching is +// disabled when restoring on behalf of pg_rewind, which walks the WAL +// backward: the following segments would never be requested, and past the end +// of the timeline they do not even exist +func maxWALFilesPerInvocation(barmanConfiguration *barmanapi.BarmanObjectStoreConfiguration, rewindMode bool) int { + if rewindMode { + return 1 + } + + if barmanConfiguration.Wal != nil && barmanConfiguration.Wal.MaxParallel > 1 { + return barmanConfiguration.Wal.MaxParallel + } + + return 1 +} + +// shouldUseEndOfWALStreamFlag returns true when the end-of-wal-stream flag +// machinery applies to the current invocation. The flag makes the following +// invocation fail, so that PostgreSQL stops polling the WAL archive and +// switches to streaming replication. It does not apply when no streaming +// connection is available, nor when restoring on behalf of pg_rewind: +// pg_rewind cannot fall back to streaming replication, and a stale flag would +// make it abort on a segment that is available in the archive +func shouldUseEndOfWALStreamFlag(cluster *cnpgv1.Cluster, podName string, rewindMode bool) bool { + if rewindMode { + return false + } + + return isStreamingAvailable(cluster, podName) +} + // isStreamingAvailable checks if this pod can replicate via streaming connection. func isStreamingAvailable(cluster *cnpgv1.Cluster, podName string) bool { if cluster == nil { diff --git a/internal/cnpgi/common/wal_test.go b/internal/cnpgi/common/wal_test.go index 92d836c..e5306d7 100644 --- a/internal/cnpgi/common/wal_test.go +++ b/internal/cnpgi/common/wal_test.go @@ -20,6 +20,7 @@ SPDX-License-Identifier: Apache-2.0 package common import ( + barmanapi "github.com/cloudnative-pg/barman-cloud/pkg/api" cnpgv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" @@ -96,3 +97,47 @@ var _ = Describe("resolveRestoreObjectStore", func() { "cluster-server", "cluster-store"), ) }) + +var _ = Describe("maxWALFilesPerInvocation", func() { + configWithMaxParallel := func(maxParallel int) *barmanapi.BarmanObjectStoreConfiguration { + return &barmanapi.BarmanObjectStoreConfiguration{ + Wal: &barmanapi.WalBackupConfiguration{MaxParallel: maxParallel}, + } + } + + DescribeTable( + "computes how many WAL files a single invocation may fetch", + func(cfg *barmanapi.BarmanObjectStoreConfiguration, rewindMode bool, want int) { + Expect(maxWALFilesPerInvocation(cfg, rewindMode)).To(Equal(want)) + }, + + Entry("no WAL configuration", &barmanapi.BarmanObjectStoreConfiguration{}, false, 1), + Entry("parallel restore configured", configWithMaxParallel(8), false, 8), + + // pg_rewind walks the timeline backwards: prefetching must stay off no + // matter what the object store configuration asks for + Entry("rewind mode overrides the configured parallelism", configWithMaxParallel(8), true, 1), + ) +}) + +var _ = Describe("shouldUseEndOfWALStreamFlag", func() { + clusterWithPrimary := func(currentPrimary string) *cnpgv1.Cluster { + return &cnpgv1.Cluster{ + Status: cnpgv1.ClusterStatus{CurrentPrimary: currentPrimary}, + } + } + + DescribeTable( + "decides whether the end-of-wal-stream flag machinery applies", + func(cluster *cnpgv1.Cluster, podName string, rewindMode bool, want bool) { + Expect(shouldUseEndOfWALStreamFlag(cluster, podName, rewindMode)).To(Equal(want)) + }, + + Entry("replica with streaming available", clusterWithPrimary("cluster-1"), "cluster-2", false, true), + Entry("primary cannot stream from anyone", clusterWithPrimary("cluster-1"), "cluster-1", false, false), + + // pg_rewind cannot fall back to streaming replication: the flag machinery + // must stay off even where a standby would use it + Entry("rewind mode wins over streaming availability", clusterWithPrimary("cluster-1"), "cluster-2", true, false), + ) +}) From 5428a338763d2b045d81610f46e45b299e82d966 Mon Sep 17 00:00:00 2001 From: Marco Nenciarini Date: Fri, 17 Jul 2026 11:56:23 +0200 Subject: [PATCH 3/3] fix: clear stale end-of-wal-stream flag before pg_rewind restore The previous commit stopped pg_rewind restores from checking the end-of-wal-stream flag, but left an existing flag on disk untouched. A flag set by a normal-recovery invocation before this pod was demoted would then resurface once the rewind finished and wrongly abort the following normal-recovery invocation. Clear the flag unconditionally in rewind mode, before the spool short-circuit, so a request for a WAL file already staged in the spool cannot skip the clear. Signed-off-by: Marco Nenciarini --- internal/cnpgi/common/wal.go | 30 ++++++++++++++++++++++++++ internal/cnpgi/common/wal_test.go | 36 +++++++++++++++++++++++++++++++ 2 files changed, 66 insertions(+) diff --git a/internal/cnpgi/common/wal.go b/internal/cnpgi/common/wal.go index 8a196cc..1d4fd30 100644 --- a/internal/cnpgi/common/wal.go +++ b/internal/cnpgi/common/wal.go @@ -323,6 +323,20 @@ func (w WALServiceImplementation) restoreFromBarmanObjectStore( return fmt.Errorf("while creating the restorer: %w", err) } + // A flag left over from a normal-recovery invocation that ran before this pod + // was demoted must not survive into a pg_rewind restore: the flag machinery + // does not apply while restoring on behalf of pg_rewind (see + // shouldUseEndOfWALStreamFlag below), so it is never checked here, but left + // untouched it would resurface and wrongly abort the first normal-recovery + // invocation that runs once the rewind is done. This runs unconditionally, + // before the spool short-circuit in Step 1, so a request for a WAL file that + // happens to already be staged in the spool cannot skip the clear. + if rewindMode { + if err := clearEndOfWALStreamFlag(walRestorer); err != nil { + return err + } + } + // Step 1: check if this WAL file is not already in the spool var wasInSpool bool if wasInSpool, err = walRestorer.RestoreFromSpool(walName, destinationPath); err != nil { @@ -526,6 +540,22 @@ func checkEndOfWALStreamFlag(walRestorer *barmanRestorer.WALRestorer) error { return nil } +// clearEndOfWALStreamFlag removes the end-of-wal-stream flag, if present, without +// treating it as an error. It is used instead of checkEndOfWALStreamFlag when +// restoring on behalf of pg_rewind, which must not abort on a flag it did not +// set itself. +func clearEndOfWALStreamFlag(walRestorer *barmanRestorer.WALRestorer) error { + contain, err := walRestorer.IsEndOfWALStream() + if err != nil { + return err + } + + if contain { + return walRestorer.ResetEndOfWalStream() + } + return nil +} + // isEndOfWALStream returns true if one of the downloads has returned // a file-not-found error. func isEndOfWALStream(results []barmanRestorer.Result) bool { diff --git a/internal/cnpgi/common/wal_test.go b/internal/cnpgi/common/wal_test.go index e5306d7..1529800 100644 --- a/internal/cnpgi/common/wal_test.go +++ b/internal/cnpgi/common/wal_test.go @@ -20,7 +20,10 @@ SPDX-License-Identifier: Apache-2.0 package common import ( + "context" + barmanapi "github.com/cloudnative-pg/barman-cloud/pkg/api" + barmanRestorer "github.com/cloudnative-pg/barman-cloud/pkg/restorer" cnpgv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" @@ -141,3 +144,36 @@ var _ = Describe("shouldUseEndOfWALStreamFlag", func() { Entry("rewind mode wins over streaming availability", clusterWithPrimary("cluster-1"), "cluster-2", true, false), ) }) + +var _ = Describe("clearEndOfWALStreamFlag", func() { + newRestorer := func() *barmanRestorer.WALRestorer { + restorer, err := barmanRestorer.New(context.Background(), nil, GinkgoT().TempDir()) + Expect(err).ToNot(HaveOccurred()) + return restorer + } + + It("is a no-op when the flag is not set", func() { + restorer := newRestorer() + + Expect(clearEndOfWALStreamFlag(restorer)).To(Succeed()) + + isEOS, err := restorer.IsEndOfWALStream() + Expect(err).ToNot(HaveOccurred()) + Expect(isEOS).To(BeFalse()) + }) + + // Regression guard: a flag left over from a normal-recovery invocation that + // ran before this pod was demoted must not survive a pg_rewind restore, or + // it would resurface and wrongly abort the first normal-recovery invocation + // that runs once the rewind is done. + It("removes a pre-existing flag without returning an error", func() { + restorer := newRestorer() + Expect(restorer.SetEndOfWALStream()).To(Succeed()) + + Expect(clearEndOfWALStreamFlag(restorer)).To(Succeed()) + + isEOS, err := restorer.IsEndOfWALStream() + Expect(err).ToNot(HaveOccurred()) + Expect(isEOS).To(BeFalse()) + }) +})