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,59 @@ 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.
+	// Failing rather than falling back to the flush position: the fallback
+	// reinstates the gap permanently, since once the slot and the server hold a
+	// resume point past it, no later run comes back to it.
+	redoStart, err := s.getCheckpointRedoStartLSN(ctx, segmentSize)
+	if err != nil {
+		return 0, err
+	}
+
 	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,79 @@ func (c *Connection) GetLatestWALFileForCluster(
 
 	return lastWal, nil
 }
+
+// GetEarliestWALFileForCluster gets the earliest archived WAL segment for a
+// certain cluster, or an empty string when the archive holds none.
+//
+// Only complete WAL segments are considered. The archive also stores the
+// in-flight `.partial` file, backup labels and history files, and a name such
+// as "000000010000000000000005.partial" would otherwise be reported as older
+// than the very segment "000000010000000000000005" that is being written into
+// it.
+//
+// This is the earliest WAL that currently survives in the archive, not the
+// earliest one ever archived: the retention removes older segments, and
+// `klio reset-lsn` can leave a gap behind.
+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)
+	}
+
+	// afero.ReadDir sorts its result by name, and a WAL directory sorts in the
+	// same order as the segments it holds. The earliest directories may hold no
+	// complete segment at all: the retention skips files carrying an extension,
+	// so an orphan `.partial` keeps a directory alive. The scan therefore
+	// continues until a directory yields a segment.
+	for _, entry := range readClusterDir {
+		if !entry.IsDir() {
+			continue
+		}
+
+		earliestWal, err := c.getEarliestWALFileInDirectory(ctx, path.Join(clusterName, entry.Name()))
+		if err != nil {
+			return "", err
+		}
+
+		if earliestWal != "" {
+			return earliestWal, nil
+		}
+	}
+
+	return "", nil
+}
+
+// getEarliestWALFileInDirectory gets the earliest complete WAL segment held by
+// the passed WAL archive directory, or an empty string when it holds none.
+func (c *Connection) getEarliestWALFileInDirectory(ctx context.Context, directory string) (string, error) {
+	readWalDirectory, err := afero.ReadDir(c.fs, directory)
+	if err != nil {
+		log.FromContext(ctx).Error(err, "while reading directory", "directory", directory)
+		return "", fmt.Errorf("while reading WAL directory: %w", err)
+	}
+
+	for _, entry := range readWalDirectory {
+		if entry.IsDir() || len(entry.Name()) != expectedWalFileNameLength {
+			continue
+		}
+
+		return entry.Name(), nil
+	}
+
+	return "", nil
+}

core/internal/repository/wals_test.go

@@ -116,3 +116,94 @@ 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
+		// walDirs maps each WAL archive directory to the files it holds. A
+		// directory with no files is still created.
+		walDirs  map[string][]string
+		expected string
+	}{
+		{
+			name:        "non-existent cluster",
+			clusterName: "non-existent-cluster",
+			expected:    "",
+		},
+		{
+			name:        "several WAL files returns the smallest",
+			clusterName: "test-cluster",
+			walDirs: map[string][]string{
+				"0000000100000000": {
+					"00000001000000000000000A",
+					"00000001000000000000000B",
+					"00000001000000000000000C",
+				},
+			},
+			expected: "00000001000000000000000A",
+		},
+		{
+			name:        "empty cluster directory",
+			clusterName: "empty-cluster",
+			walDirs:     map[string][]string{"0000000100000000": {}},
+			expected:    "",
+		},
+		{
+			name:        "in-flight partial is not a segment",
+			clusterName: "partial-only-cluster",
+			walDirs: map[string][]string{
+				"0000000100000000": {"000000010000000000000005.partial"},
+			},
+			expected: "",
+		},
+		{
+			name:        "backup label is not a segment",
+			clusterName: "label-only-cluster",
+			walDirs: map[string][]string{
+				"0000000100000000": {"000000010000000000000004.00000028.backup"},
+			},
+			expected: "",
+		},
+		{
+			name:        "scan continues past a directory holding no segment",
+			clusterName: "partial-then-segments-cluster",
+			walDirs: map[string][]string{
+				"0000000100000000": {"000000010000000000000005.partial"},
+				"0000000100000001": {
+					"000000010000000100000002",
+					"000000010000000100000003",
+				},
+			},
+			expected: "000000010000000100000002",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			for walDir, walNames := range tc.walDirs {
+				require.NoError(t, opts.FS.MkdirAll(path.Join(tc.clusterName, walDir), 0o750))
+				for _, walName := range walNames {
+					file, err := opts.FS.Create(path.Join(tc.clusterName, 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

@@ -32,20 +32,47 @@ import (
 	"github.com/cloudnative-pg/klio/core/internal/grpc"
 	"github.com/cloudnative-pg/klio/core/internal/kopia"
 	"github.com/cloudnative-pg/klio/core/internal/queue"
+	"github.com/cloudnative-pg/klio/core/internal/repository"
 )
 
 // CloseBackup implements the CloseBackup GRPC call.
 func (w *Implementation) CloseBackup(
 	ctx context.Context,
 	request *grpc.CloseBackupRequest,
 ) (*grpc.CloseBackupResult, error) {
+	if err := repository.ValidatePathComponent(request.GetClusterName()); err != nil {
+		return nil, status.Errorf(codes.InvalidArgument, "invalid cluster name: %v", err.Error())
+	}
+
 	// Step 1: verify if the WALs have been archived
 	missingWALFiles, err := w.checkWALFiles(request)
 	if err != nil {
 		return nil, err
 	}
 
 	if len(missingWALFiles) > 0 {
+		// If a required WAL predates the earliest segment the archive holds, it
+		// can never be archived: this cluster started streaming from a later
+		// point, and nothing will go back to fill the gap. Fail the backup
+		// instead of letting the client wait for a WAL that will never arrive.
+		//
+		// checkWALFiles walks a single timeline by ascending position, so the
+		// missing list is already sorted and only its first entry can be the
+		// oldest required segment.
+		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 != "" && missingWALFiles[0] < earliestWAL {
+			return nil, status.Errorf(
+				codes.FailedPrecondition,
+				"backup requires WAL %q which predates the earliest archived WAL %q and can never be "+
+					"archived: the backup ran on an instance whose last checkpoint precedes the point "+
+					"the WAL stream started from. Retry the backup targeting the primary, or wait for a "+
+					"checkpoint to be replayed on this instance",
+				missingWALFiles[0], earliestWAL)
+		}
+
 		return &grpc.CloseBackupResult{
 			Tier2Schedule:   false,
 			MissingWalFiles: missingWALFiles,