fix(core): support immediate backup from a replica cluster

core/internal/client/sendwal/receiver.go

@@ -31,6 +31,7 @@ import (
 	"github.com/cloudnative-pg/machinery/pkg/log"
 	"github.com/cloudnative-pg/machinery/pkg/types"
 	"github.com/jackc/pglogrepl"
+	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgconn"
 	"github.com/jackc/pgx/v5/pgproto3"
 	"go.opentelemetry.io/otel/attribute"
@@ -202,19 +203,63 @@ func (s *Process) getReplicationStartPointFromClient(
 		return slotResult.RestartLSN, nil
 	}
 
-	// If nor the Klio server nor the replication slot are set,
-	// we use the XLOG flush position, taking care of
-	// starting streaming from the beginning of the WAL file.
+	// Neither the Klio server nor the replication slot have a resume point.
+	// This usually happens when we are running against this PostgreSQL instance
+	// for the first time.
 	//
-	// This usually happens when we are running against this
-	// PostgreSQL instance for the first time.
+	// We start from the redo point of the latest checkpoint (on a standby, the
+	// latest restartpoint) rather than from the current flush position. That
+	// redo point is the earliest LSN a later pg_backup_start on this instance
+	// can report as a backup start, so the WAL a backup needs is always within
+	// what we archive to tier1. This matters on a standby, where pg_backup_start
+	// reports the last restartpoint, which lags the flush position: streaming
+	// from the flush position would leave the segments in between permanently
+	// out of tier1.
+	redoStart, err := s.getCheckpointRedoStartLSN(ctx, segmentSize)
+	if err != nil {
+		contextLogger.Info(
+			"Could not read the checkpoint redo LSN, falling back to the current flush position",
+			"err", err.Error(),
+			"xlogFlushPos", xlogFlushPos,
+			"segmentSize", segmentSize,
+		)
+
+		return getStartWALLSN(xlogFlushPos, segmentSize), nil
+	}
+
 	contextLogger.Debug(
-		"Current flush LSN",
+		"Checkpoint redo LSN",
+		"redoStart", redoStart,
 		"xlogFlushPos", xlogFlushPos,
 		"segmentSize", segmentSize,
 	)
 
-	return getStartWALLSN(xlogFlushPos, segmentSize), nil
+	return redoStart, nil
+}
+
+// getCheckpointRedoStartLSN returns the start of the WAL file that contains the
+// redo point of the latest checkpoint (or restartpoint, on a standby). It opens
+// a regular (non-replication) connection because pg_control_checkpoint() cannot
+// be queried on the physical replication connection used for streaming.
+func (s *Process) getCheckpointRedoStartLSN(
+	ctx context.Context,
+	segmentSize uint64,
+) (pglogrepl.LSN, error) {
+	conn, err := pgx.Connect(ctx, s.config.Source.StandardDSN)
+	if err != nil {
+		return 0, fmt.Errorf("while connecting to PostgreSQL: %w", err)
+	}
+	defer func() {
+		_ = conn.Close(ctx)
+	}()
+
+	var redoLSN uint64
+	row := conn.QueryRow(ctx, "SELECT redo_lsn - '0/0' FROM pg_control_checkpoint()")
+	if err := row.Scan(&redoLSN); err != nil {
+		return 0, fmt.Errorf("while reading the checkpoint redo LSN: %w", err)
+	}
+
+	return getStartWALLSN(pglogrepl.LSN(redoLSN), segmentSize), nil
 }
 
 type walCoordinate struct {

core/internal/repository/wals.go

@@ -132,3 +132,66 @@ func (c *Connection) GetLatestWALFileForCluster(
 
 	return lastWal, nil
 }
+
+// GetEarliestWALFileForCluster gets the earliest archived WAL for a certain
+// cluster, or an empty string when the archive is empty. Because the WAL stream
+// only ever appends segments going forward, no segment older than this one will
+// ever be archived.
+//
+//nolint:cyclop
+func (c *Connection) GetEarliestWALFileForCluster(
+	ctx context.Context,
+	clusterName string,
+) (string, error) {
+	logger := log.FromContext(ctx)
+
+	readClusterDir, err := afero.ReadDir(c.fs, clusterName)
+	if err != nil {
+		if errors.Is(err, os.ErrNotExist) {
+			return "", nil
+		}
+
+		logger.Error(
+			err,
+			"while reading cluster directory",
+			"clusterName", clusterName,
+		)
+
+		return "", fmt.Errorf("while reading cluster directory: %w", err)
+	}
+
+	var earliestWalDirectoryName string
+	for _, entry := range readClusterDir {
+		if !entry.IsDir() {
+			continue
+		}
+
+		if earliestWalDirectoryName == "" || strings.Compare(entry.Name(), earliestWalDirectoryName) == -1 {
+			earliestWalDirectoryName = entry.Name()
+		}
+	}
+
+	if earliestWalDirectoryName == "" {
+		return "", nil
+	}
+
+	earliestWalDirectoryName = path.Join(clusterName, earliestWalDirectoryName)
+	readWalDirectory, err := afero.ReadDir(c.fs, earliestWalDirectoryName)
+	if err != nil {
+		logger.Error(err, "while reading directory", "earliestWalDirectoryName", earliestWalDirectoryName)
+		return "", fmt.Errorf("while reading WAL directory: %w", err)
+	}
+
+	var earliestWal string
+	for _, entry := range readWalDirectory {
+		if entry.IsDir() {
+			continue
+		}
+
+		if earliestWal == "" || strings.Compare(entry.Name(), earliestWal) == -1 {
+			earliestWal = entry.Name()
+		}
+	}
+
+	return earliestWal, nil
+}

core/internal/repository/wals_test.go

@@ -116,3 +116,65 @@ func TestGetLatestWALFileForCluster(t *testing.T) {
 	require.NoError(t, err)
 	assert.Empty(t, latestWal)
 }
+
+func TestGetEarliestWALFileForCluster(t *testing.T) {
+	opts := Options{
+		FS:       afero.NewMemMapFs(),
+		Password: "test-password",
+	}
+	require.NoError(t, Initialize(opts))
+
+	conn, err := Open(opts)
+	require.NoError(t, err)
+	require.NotNil(t, conn)
+	defer conn.Close()
+
+	tests := []struct {
+		name        string
+		clusterName string
+		createDir   bool
+		walNames    []string
+		expected    string
+	}{
+		{
+			name:        "non-existent cluster",
+			clusterName: "non-existent-cluster",
+			expected:    "",
+		},
+		{
+			name:        "several WAL files returns the smallest",
+			clusterName: "test-cluster",
+			createDir:   true,
+			walNames: []string{
+				"00000001000000000000000A",
+				"00000001000000000000000B",
+				"00000001000000000000000C",
+			},
+			expected: "00000001000000000000000A",
+		},
+		{
+			name:        "empty cluster directory",
+			clusterName: "empty-cluster",
+			createDir:   true,
+			expected:    "",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			if tc.createDir {
+				walDir := path.Join(tc.clusterName, "0000000100000000")
+				require.NoError(t, opts.FS.MkdirAll(walDir, 0o750))
+				for _, walName := range tc.walNames {
+					file, err := opts.FS.Create(path.Join(walDir, walName))
+					require.NoError(t, err)
+					require.NoError(t, file.Close())
+				}
+			}
+
+			earliestWal, err := conn.GetEarliestWALFileForCluster(context.Background(), tc.clusterName)
+			require.NoError(t, err)
+			assert.Equal(t, tc.expected, earliestWal)
+		})
+	}
+}

core/internal/server/walserver/backup.go

@@ -46,6 +46,26 @@ func (w *Implementation) CloseBackup(
 	}
 
 	if len(missingWALFiles) > 0 {
+		// If a required WAL predates the earliest segment the archive will ever
+		// hold, it can never be archived: the stream only appends segments going
+		// forward. Fail the backup instead of letting the client wait for a WAL
+		// that will never arrive.
+		earliestWAL, err := w.conn.GetEarliestWALFileForCluster(ctx, request.GetClusterName())
+		if err != nil {
+			return nil, status.Errorf(codes.Internal, "while reading earliest archived WAL: %v", err.Error())
+		}
+		if earliestWAL != "" {
+			for _, missing := range missingWALFiles {
+				if missing < earliestWAL {
+					return nil, status.Errorf(
+						codes.FailedPrecondition,
+						"backup requires WAL %q which predates the earliest archived WAL %q "+
+							"and can never be archived",
+						missing, earliestWAL)
+				}
+			}
+		}
+
 		return &grpc.CloseBackupResult{
 			Tier2Schedule:   false,
 			MissingWalFiles: missingWALFiles,

core/internal/server/walserver/backup_test.go

@@ -0,0 +1,121 @@
+/*
+Copyright © contributors to CloudNativePG, established as
+CloudNativePG a Series of LF Projects, LLC.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package walserver
+
+import (
+	"context"
+	"path"
+	"testing"
+
+	"github.com/spf13/afero"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	"github.com/cloudnative-pg/klio/core/internal/grpc"
+	"github.com/cloudnative-pg/klio/core/internal/repository"
+)
+
+const closeBackupSegmentSize = 16 * 1024 * 1024
+
+// newTestImplementation returns a WAL server backed by an in-memory repository
+// pre-populated with the given WAL files for a single cluster.
+func newTestImplementation(t *testing.T, clusterName string, walFiles []string) *Implementation {
+	t.Helper()
+
+	opts := repository.Options{
+		FS:       afero.NewMemMapFs(),
+		Password: "test-password",
+	}
+	require.NoError(t, repository.Initialize(opts))
+
+	conn, err := repository.Open(opts)
+	require.NoError(t, err)
+	t.Cleanup(conn.Close)
+
+	for _, walName := range walFiles {
+		walDir := path.Join(clusterName, walName[0:16])
+		require.NoError(t, opts.FS.MkdirAll(walDir, 0o750))
+		file, err := opts.FS.Create(path.Join(walDir, walName))
+		require.NoError(t, err)
+		require.NoError(t, file.Close())
+	}
+
+	return New(Options{Connection: conn})
+}
+
+// TestCloseBackupFailsOnPermanentlyMissingWAL verifies that CloseBackup returns
+// a terminal error when a required WAL predates the earliest archived WAL and
+// can therefore never be archived.
+func TestCloseBackupFailsOnPermanentlyMissingWAL(t *testing.T) {
+	const clusterName = "test-cluster"
+
+	// The archive starts at segment 05: segments 03 and 04 required by the
+	// backup will never appear.
+	impl := newTestImplementation(t, clusterName, []string{
+		"000000010000000000000005",
+		"000000010000000000000006",
+		"000000010000000000000007",
+	})
+
+	result, err := impl.CloseBackup(context.Background(), &grpc.CloseBackupRequest{
+		ClusterName: clusterName,
+		Timeline:    1,
+		StartWal:    "000000010000000000000003",
+		EndWal:      "000000010000000000000007",
+		SegmentSize: closeBackupSegmentSize,
+	})
+
+	require.Error(t, err)
+	require.Nil(t, result)
+
+	s, ok := status.FromError(err)
+	require.True(t, ok, "expected a gRPC status error")
+	assert.Equal(t, codes.FailedPrecondition, s.Code())
+}
+
+// TestCloseBackupWaitsForRecentMissingWAL verifies that CloseBackup keeps
+// reporting a not-yet-archived WAL as missing (so the client waits) when that
+// WAL does not predate the earliest archived WAL.
+func TestCloseBackupWaitsForRecentMissingWAL(t *testing.T) {
+	const clusterName = "test-cluster"
+
+	// Segment 06 is not archived yet, but it does not predate the earliest
+	// archived WAL (03): it can still arrive.
+	impl := newTestImplementation(t, clusterName, []string{
+		"000000010000000000000003",
+		"000000010000000000000004",
+		"000000010000000000000005",
+		"000000010000000000000007",
+	})
+
+	result, err := impl.CloseBackup(context.Background(), &grpc.CloseBackupRequest{
+		ClusterName: clusterName,
+		Timeline:    1,
+		StartWal:    "000000010000000000000003",
+		EndWal:      "000000010000000000000007",
+		SegmentSize: closeBackupSegmentSize,
+	})
+
+	require.NoError(t, err)
+	require.NotNil(t, result)
+	assert.Equal(t, []string{"000000010000000000000006"}, result.GetMissingWalFiles())
+}

documentation/web/docs/developer/running-e2e-tests.md

@@ -106,6 +106,10 @@ The E2E tests are located in `operator/test/e2e/` and include:
    - `BackupFromPrimary`: backup from a single-instance cluster
    - `BackupFromStandby`: backup from a standby in a multi-instance
     cluster
+- **`backup_from_replica_cluster_test.go`** - Immediate backup from a
+  freshly-created replica cluster: verifies the backup completes even
+  when the WAL streamer and `pg_backup_start` disagree on the starting
+  WAL (`BackupFromReplicaCluster`)
 - **`maintenance_test.go`** - Server-side post-backup maintenance on a
   tier1-only deployment: verifies the backup queue consumer applies
   tier1 WAL retention after a backup even when tier2 is not configured