From f25415eeb4bd698c5dc10d280844ca2e7691f819 Mon Sep 17 00:00:00 2001
From: Jeremy Schneider <schneider@ardentperf.com>
Date: Mon, 6 Jul 2026 09:10:06 -0700
Subject: [PATCH] fix(walrestore): use timeline-aware barman-cloud end-of-WAL
 markers

Update `wal-restore` to use the new `barman-cloud` restorer APIs,
resolving a bug that can cause replicas to enter a broken state that
CNPG cant recover from. Local marker naming, WAL timeline parsing, and
restore-result filtering policy lives in
`github.com/cloudnative-pg/barman-cloud/pkg/restorer`. depends on
https://github.com/cloudnative-pg/barman-cloud/pull/277

Signed-off-by: Jeremy Schneider <schneider@ardentperf.com>
---
 internal/cnpgi/common/wal.go | 46 +++++++++++++-----------------------
 1 file changed, 16 insertions(+), 30 deletions(-)

diff --git a/internal/cnpgi/common/wal.go b/internal/cnpgi/common/wal.go
index bb7d3e0..fc974cd 100644
--- a/internal/cnpgi/common/wal.go
+++ b/internal/cnpgi/common/wal.go
@@ -21,7 +21,6 @@ package common
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"os"
 	"path"
@@ -331,9 +330,14 @@ func (w WALServiceImplementation) restoreFromBarmanObjectStore(
 		return nil
 	}
 
+	// Step 2: return error if the end-of-wal-stream flag is set.
+	// PostgreSQL treats a restore_command failure as "WAL not available from
+	// archive" and can then try streaming via primary_conninfo. When prefetch has
+	// already observed archive exhaustion for this timeline, return one failure to
+	// let PostgreSQL switch sources instead of repeatedly probing the archive.
 	// We skip this step if streaming connection is not available
 	if isStreamingAvailable(cluster, w.InstanceName) {
-		if err := checkEndOfWALStreamFlag(walRestorer); err != nil {
+		if err := checkEndOfWALStreamFlag(walRestorer, walName); err != nil {
 			return err
 		}
 	}
@@ -366,15 +370,15 @@ func (w WALServiceImplementation) restoreFromBarmanObjectStore(
 	}
 
 	// We skip this step if streaming connection is not available
-	endOfWALStream := isEndOfWALStream(walStatus)
-	if isStreamingAvailable(cluster, w.InstanceName) && endOfWALStream {
-		contextLogger.Info(
-			"Set end-of-wal-stream flag as one of the WAL files to be prefetched was not found")
-
-		err = walRestorer.SetEndOfWALStream()
+	if isStreamingAvailable(cluster, w.InstanceName) {
+		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
@@ -473,32 +477,14 @@ 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()
+// checkEndOfWALStreamFlag returns ErrEndOfWALStreamReached if the flag is set for the requested WAL timeline.
+func checkEndOfWALStreamFlag(walRestorer *barmanRestorer.WALRestorer, walName string) error {
+	contains, err := walRestorer.ConsumeEndOfWALStreamForWAL(walName)
 	if err != nil {
 		return err
 	}
-
-	if contain {
-		err := walRestorer.ResetEndOfWalStream()
-		if err != nil {
-			return err
-		}
-
+	if contains {
 		return ErrEndOfWALStreamReached
 	}
 	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
-}