fix(walrestore): use timeline-aware barman-cloud end-of-WAL markers

internal/cnpgi/common/wal.go

@@ -361,7 +361,7 @@ func (w WALServiceImplementation) restoreFromBarmanObjectStore(
 	// 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 {
+		if err := walRestorer.ClearEndOfWALStream(); err != nil {
 			return err
 		}
 	}
@@ -379,10 +379,14 @@ func (w WALServiceImplementation) restoreFromBarmanObjectStore(
 	}
 
 	// Step 2: return error if the end-of-wal-stream flag is set.
+	// A restore_command failure tells PostgreSQL the WAL is not in the archive,
+	// so it can switch to streaming. Prefetch sets the flag when it exhausts the
+	// archive for a timeline, and one failure is returned per flag.
+	//
 	// 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 {
+		if err := checkEndOfWALStreamFlag(walRestorer, walName); err != nil {
 			return err
 		}
 	}
@@ -419,15 +423,15 @@ func (w WALServiceImplementation) restoreFromBarmanObjectStore(
 	}
 
 	// We skip this step if the flag machinery does not apply to this invocation
-	endOfWALStream := isEndOfWALStream(walStatus)
-	if useEndOfWALStreamFlag && endOfWALStream {
-		contextLogger.Info(
-			"Set end-of-wal-stream flag as one of the WAL files to be prefetched was not found")
-
-		err = walRestorer.SetEndOfWALStream() //nolint:staticcheck // remove when #993 merges
+	if useEndOfWALStreamFlag {
+		endOfWALStream, err := walRestorer.SetEndOfWALStreamFromResults(walStatus)
 		if err != nil {
 			return err
 		}
+		if endOfWALStream {
+			contextLogger.Info(
+				"Set end-of-wal-stream flag as one of the WAL files to be prefetched was not found")
+		}
 	}
 
 	successfulWalRestore := 0
@@ -558,48 +562,15 @@ func gatherWALFilesToRestore(walName string, parallel int) (walList []string, er
 	return walList, err
 }
 
-// checkEndOfWALStreamFlag returns ErrEndOfWALStreamReached if the flag is set in the restorer.
-func checkEndOfWALStreamFlag(walRestorer *barmanRestorer.WALRestorer) error {
-	contain, err := walRestorer.IsEndOfWALStream() //nolint:staticcheck // remove when #993 merges
+// checkEndOfWALStreamFlag returns ErrEndOfWALStreamReached if the flag is set for the
+// timeline of the requested WAL, and removes the flag so it fails only once.
+func checkEndOfWALStreamFlag(walRestorer *barmanRestorer.WALRestorer, walName string) error {
+	contains, err := walRestorer.ConsumeEndOfWALStreamForWAL(walName)
 	if err != nil {
 		return err
 	}
-
-	if contain {
-		err := walRestorer.ResetEndOfWalStream() //nolint:staticcheck // remove when #993 merges
-		if err != nil {
-			return err
-		}
-
+	if contains {
 		return ErrEndOfWALStreamReached
 	}
 	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() //nolint:staticcheck // remove when #993 merges
-	if err != nil {
-		return err
-	}
-
-	if contain {
-		return walRestorer.ResetEndOfWalStream() //nolint:staticcheck // remove when #993 merges
-	}
-	return nil
-}
-
-// isEndOfWALStream returns true if one of the downloads has returned
-// a file-not-found error.
-func isEndOfWALStream(results []barmanRestorer.Result) bool {
-	for _, result := range results {
-		if errors.Is(result.Err, barmanRestorer.ErrWALNotFound) {
-			return true
-		}
-	}
-
-	return false
-}

internal/cnpgi/common/wal_test.go

@@ -30,6 +30,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/ptr"
 
+	barmancloudv1 "github.com/cloudnative-pg/plugin-barman-cloud/api/v1"
 	"github.com/cloudnative-pg/plugin-barman-cloud/internal/cnpgi/metadata"
 	"github.com/cloudnative-pg/plugin-barman-cloud/internal/cnpgi/operator/config"
 
@@ -150,36 +151,75 @@ var _ = Describe("shouldUseEndOfWALStreamFlag", func() {
 	)
 })
 
-var _ = Describe("clearEndOfWALStreamFlag", func() {
-	newRestorer := func() *barmanRestorer.WALRestorer {
-		restorer, err := barmanRestorer.New(context.Background(), nil, GinkgoT().TempDir())
+var _ = Describe("checkEndOfWALStreamFlag", func() {
+	var spoolDirectory string
+	var walRestorer *barmanRestorer.WALRestorer
+
+	BeforeEach(func(ctx SpecContext) {
+		spoolDirectory = GinkgoT().TempDir()
+		var err error
+		walRestorer, err = barmanRestorer.New(ctx, nil, spoolDirectory)
 		Expect(err).ToNot(HaveOccurred())
-		return restorer
-	}
 
-	It("is a no-op when the flag is not set", func() {
-		restorer := newRestorer()
+		_, err = walRestorer.SetEndOfWALStreamFromResults([]barmanRestorer.Result{
+			{WalName: "000000010000000000000002", Err: barmanRestorer.ErrWALNotFound},
+		})
+		Expect(err).ToNot(HaveOccurred())
+	})
+
+	It("ignores the flag of another timeline", func() {
+		Expect(checkEndOfWALStreamFlag(walRestorer, "000000020000000000000003")).To(Succeed())
+		Expect(filepath.Join(spoolDirectory, "end-of-wal-stream-00000001")).To(BeAnExistingFile())
+	})
 
-		Expect(clearEndOfWALStreamFlag(restorer)).To(Succeed())
+	It("fails once and removes the flag of the requested timeline", func() {
+		Expect(checkEndOfWALStreamFlag(walRestorer, "000000010000000000000003")).
+			To(MatchError(ErrEndOfWALStreamReached))
+		Expect(filepath.Join(spoolDirectory, "end-of-wal-stream-00000001")).ToNot(BeAnExistingFile())
 
-		isEOS, err := restorer.IsEndOfWALStream()
-		Expect(err).ToNot(HaveOccurred())
-		Expect(isEOS).To(BeFalse())
+		Expect(checkEndOfWALStreamFlag(walRestorer, "000000010000000000000003")).To(Succeed())
 	})
+})
 
-	// 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())
+var _ = Describe("restoreFromBarmanObjectStore in rewind mode", func() {
+	// A flag left over from a normal-recovery invocation that ran before this pod
+	// was demoted must not survive a pg_rewind restore, whatever its timeline,
+	// even when the requested WAL is already in the spool and no download runs.
+	It("clears every end-of-wal-stream flag before serving from the spool", func() {
+		const walName = "000000020000000000000005"
+		spoolDirectory := GinkgoT().TempDir()
+		for _, name := range []string{
+			"end-of-wal-stream",
+			"end-of-wal-stream-00000001",
+			"end-of-wal-stream-00000002",
+			walName,
+		} {
+			Expect(os.WriteFile(filepath.Join(spoolDirectory, name), nil, 0o600)).To(Succeed())
+		}
+		destinationPath := filepath.Join(GinkgoT().TempDir(), walName)
+
+		w := WALServiceImplementation{SpoolDirectory: spoolDirectory, InstanceName: "cluster-1"}
+		objectStore := &barmancloudv1.ObjectStore{
+			ObjectMeta: metav1.ObjectMeta{Name: "store", Namespace: "test-ns"},
+			Spec: barmancloudv1.ObjectStoreSpec{
+				Configuration: barmanapi.BarmanObjectStoreConfiguration{
+					DestinationPath: "s3://bucket/",
+					BarmanCredentials: barmanapi.BarmanCredentials{
+						AWS: &barmanapi.S3Credentials{InheritFromIAMRole: true},
+					},
+				},
+			},
+		}
+		cluster := &cnpgv1.Cluster{Status: cnpgv1.ClusterStatus{CurrentPrimary: "cluster-1"}}
 
-		Expect(clearEndOfWALStreamFlag(restorer)).To(Succeed())
+		Expect(w.restoreFromBarmanObjectStore(
+			context.Background(), cluster, objectStore, "cluster", walName, destinationPath, true,
+		)).To(Succeed())
 
-		isEOS, err := restorer.IsEndOfWALStream()
+		Expect(destinationPath).To(BeAnExistingFile())
+		entries, err := os.ReadDir(spoolDirectory)
 		Expect(err).ToNot(HaveOccurred())
-		Expect(isEOS).To(BeFalse())
+		Expect(entries).To(BeEmpty())
 	})
 })
 

test/e2e/internal/tests/walrestore/walrestore.go

@@ -54,15 +54,19 @@ const (
 	// pgWalPath is PgDataPath + "/pg_wal" in the CloudNativePG operand image.
 	pgWalPath = "/var/lib/postgresql/data/pgdata/pg_wal"
 	// endOfWALStreamFlag is the sentinel file the barman-cloud restorer writes in
-	// the spool to record that the archive ran out of segments.
-	endOfWALStreamFlag = "end-of-wal-stream"
+	// the spool to record that the archive ran out of segments. Markers are scoped
+	// by timeline; the forged wals are on timeline 1.
+	endOfWALStreamFlag = "end-of-wal-stream-00000001"
+	// A forged timeline 2 segment (child of timeline 1) and its marker.
+	childWALFile            = "0000000200000000000000F7"
+	childEndOfWALStreamFlag = "end-of-wal-stream-00000002"
 	// managerExecutable is the CloudNativePG instance manager. Its wal-restore
 	// subcommand delegates to the plugin when the cluster uses it as WAL archiver.
 	managerExecutable = "/controller/manager"
 	// postgresContainer is the container we exec into on instance pods.
 	postgresContainer = "postgres"
-	// walLogDir is the WALs subdirectory (timeline + log id) the forged segments
-	// live under; a freshly bootstrapped, idle cluster stays within it.
+	// walLogDir is the timeline and log id of the WAL files a freshly
+	// bootstrapped, idle cluster archives.
 	walLogDir = "0000000100000000"
 	// bucket is the destination bucket of the S3 ObjectStore.
 	bucket = "backups"
@@ -75,9 +79,10 @@ func walFile(n int) string {
 	return fmt.Sprintf("0000000100000000%08X", 0xF0+n)
 }
 
-// walObjectURI returns the s3 URI of a wals object (by file name) in the store.
+// walObjectURI returns the s3 URI of a wals object (by file name) in the store,
+// which barman keeps in a subdirectory named after its timeline and log id.
 func walObjectURI(name string) string {
-	return fmt.Sprintf("s3://%s/%s/wals/%s/%s", bucket, clusterName, walLogDir, name)
+	return fmt.Sprintf("s3://%s/%s/wals/%s/%s", bucket, clusterName, name[:16], name)
 }
 
 // execInPod runs a command in the given container and returns stdout, stderr
@@ -308,5 +313,17 @@ var _ = Describe("Parallel WAL restore", func() {
 				g.Expect(spoolSegments()).To(Equal(0), "no WAL segments in spool")
 				g.Expect(flagSet()).To(BeTrue(), "end-of-wal-stream set (#7/#8 absent)")
 			}).WithTimeout(time.Minute).WithPolling(2 * time.Second).Should(Succeed())
+
+			// The parent-timeline flag must not fail a child-timeline request
+			// (cloudnative-pg/barman-cloud#276).
+			By("requesting a child-timeline WAL: restored, the parent-timeline flag untouched")
+			forge(latestWAL+".gz", childWALFile)
+			Expect(restore(childWALFile)).To(Succeed())
+			Eventually(func(g Gomega) {
+				g.Expect(existsIn(pgWalPath, childWALFile)).To(BeTrue(), "child-timeline WAL in pg_wal")
+				g.Expect(flagSet()).To(BeTrue(), "parent-timeline end-of-wal-stream untouched")
+				g.Expect(existsIn(spoolDirectory, childEndOfWALStreamFlag)).To(BeTrue(),
+					"child-timeline end-of-wal-stream set (its next segments absent)")
+			}).WithTimeout(time.Minute).WithPolling(2 * time.Second).Should(Succeed())
 		})
 })