fix: relay standby backups to tier2 without waiting for the last WAL

core/cmd/backup/run.go

@@ -158,6 +158,7 @@ func runBackup(cmd *cobra.Command, _ []string) error {
 			EndWal:                 metadata.EndWAL,
 			SegmentSize:            metadata.SegmentSize,
 			SendToTier2:            tier2,
+			EnqueueWithoutWals:     !waitWALs,
 			Tier2RetentionPolicy:   marshalTier2RetentionPolicy(cmd.Context(), &configuration),
 			Tier2CompressionPolicy: marshalTier2CompressionPolicy(cmd.Context(), &configuration),
 		})
@@ -167,14 +168,21 @@ func runBackup(cmd *cobra.Command, _ []string) error {
 				backupfailure.RepositoryError.ExitCode)
 		}
 
-		if waitWALs && len(result.GetMissingWalFiles()) > 0 {
+		if len(result.GetMissingWalFiles()) > 0 {
+			if waitWALs {
+				contextLogger.Info(
+					"Detected missing WAL files, waiting for 5 seconds",
+					"missingWALFiles", result.GetMissingWalFiles(),
+				)
+				time.Sleep(5 * time.Second)
+
+				continue
+			}
+
 			contextLogger.Info(
-				"Detected missing WAL files, waiting for 5 seconds",
+				"Detected missing WAL files, not waiting for them",
 				"missingWALFiles", result.GetMissingWalFiles(),
 			)
-			time.Sleep(5 * time.Second)
-
-			continue
 		}
 
 		if result.GetTier2Schedule() {

core/internal/server/walserver/backup.go

@@ -45,7 +45,11 @@ func (w *Implementation) CloseBackup(
 		return nil, err
 	}
 
-	if len(missingWALFiles) > 0 {
+	// By default a client calls CloseBackup again until no WAL is missing, so
+	// the task is deferred to that call. A client may not wait for the WALs to be archived,
+	// and can set EnqueueWithoutWals to true in the request
+	// to force the backup to be enqueued even if some WALs are missing.
+	if len(missingWALFiles) > 0 && !request.GetEnqueueWithoutWals() {
 		return &grpc.CloseBackupResult{
 			Tier2Schedule:   false,
 			MissingWalFiles: missingWALFiles,
@@ -67,7 +71,7 @@ func (w *Implementation) CloseBackup(
 
 	return &grpc.CloseBackupResult{
 		Tier2Schedule:   w.queue != nil && request.GetSendToTier2(),
-		MissingWalFiles: nil,
+		MissingWalFiles: missingWALFiles,
 	}, nil
 }
 

core/internal/server/walserver/backup_test.go

@@ -0,0 +1,153 @@
+/*
+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"
+	"time"
+
+	"github.com/nats-io/nats-server/v2/server"
+	"github.com/nats-io/nats.go"
+	"github.com/spf13/afero"
+	"github.com/stretchr/testify/require"
+
+	"github.com/cloudnative-pg/klio/core/internal/grpc"
+	"github.com/cloudnative-pg/klio/core/internal/queue"
+	"github.com/cloudnative-pg/klio/core/internal/repository"
+)
+
+const (
+	testClusterName = "cluster-example"
+	testSegmentSize = 16 * 1024 * 1024
+	testStartWAL    = "000000010000000000000001"
+	testEndWAL      = "000000010000000000000002"
+)
+
+// newCloseBackupServer builds a WAL server whose repository holds only the
+// backup's first WAL segment, so the last one is always reported missing,
+// and whose queue is backed by an embedded NATS server. It returns the
+// server and the queue connection to consume the enqueued tasks from.
+func newCloseBackupServer(t *testing.T) (*Implementation, *queue.Conn) {
+	t.Helper()
+
+	ns, err := server.NewServer(&server.Options{
+		Host:      "127.0.0.1",
+		Port:      -1,
+		JetStream: true,
+		StoreDir:  t.TempDir(),
+	})
+	require.NoError(t, err)
+	go ns.Start()
+	require.True(t, ns.ReadyForConnections(4*time.Second), "NATS server not ready")
+	t.Cleanup(ns.Shutdown)
+
+	nc, err := nats.Connect(ns.ClientURL())
+	require.NoError(t, err)
+	t.Cleanup(nc.Close)
+
+	q, err := queue.New(context.Background(), nc)
+	require.NoError(t, err)
+
+	fs := afero.NewMemMapFs()
+	repoOpts := repository.Options{FS: fs, Password: "test-password"}
+	require.NoError(t, repository.Initialize(repoOpts))
+	conn, err := repository.Open(repoOpts)
+	require.NoError(t, err)
+	t.Cleanup(conn.Close)
+
+	// Only the first segment has been archived.
+	walPath := path.Join(testClusterName, testStartWAL[:16], testStartWAL)
+	require.NoError(t, afero.WriteFile(fs, walPath, []byte("wal"), 0o600))
+
+	return New(Options{Connection: conn, Queue: q}), q
+}
+
+// receiveBackupTask consumes one backup task from the queue, or returns nil
+// when none arrives within the timeout.
+func receiveBackupTask(t *testing.T, q *queue.Conn, timeout time.Duration) *queue.BackupTask {
+	t.Helper()
+
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+
+	received := make(chan *queue.BackupTask, 1)
+	go func() {
+		_ = q.ConsumeBackupReceivedMessages(ctx, func(_ context.Context, task *queue.BackupTask) error {
+			received <- task
+			cancel()
+
+			return nil
+		})
+	}()
+
+	select {
+	case task := <-received:
+		return task
+	case <-ctx.Done():
+		return nil
+	}
+}
+
+func newCloseBackupRequest(enqueueWithoutWALs bool) *grpc.CloseBackupRequest {
+	return &grpc.CloseBackupRequest{
+		ClusterName:        testClusterName,
+		BackupName:         "backup-1",
+		Timeline:           1,
+		StartWal:           testStartWAL,
+		EndWal:             testEndWAL,
+		SegmentSize:        testSegmentSize,
+		SendToTier2:        true,
+		EnqueueWithoutWals: enqueueWithoutWALs,
+	}
+}
+
+// TestCloseBackupMissingWALsWithoutWaitEnqueuesTask covers a backup taken
+// with a client that does not wait for the last WAL to be archived, so the post-backup
+// task must be enqueued on this single call or the backup is never relayed.
+func TestCloseBackupMissingWALsWithoutWaitEnqueuesTask(t *testing.T) {
+	impl, q := newCloseBackupServer(t)
+
+	result, err := impl.CloseBackup(context.Background(), newCloseBackupRequest(true))
+	require.NoError(t, err)
+	require.Equal(t, []string{testEndWAL}, result.GetMissingWalFiles())
+	require.True(t, result.GetTier2Schedule())
+
+	task := receiveBackupTask(t, q, 5*time.Second)
+	require.NotNil(t, task, "the backup task must be enqueued even if WALs are missing")
+	require.Equal(t, testClusterName, task.ClusterName)
+	require.True(t, task.SendToTier2)
+}
+
+// TestCloseBackupMissingWALsWithWaitDefersTask covers a backup taken with a client
+// that waits for the last WAL to be archived: the client retries CloseBackup until no WAL is missing, so the
+// task must not be enqueued before then.
+func TestCloseBackupMissingWALsWithWaitDefersTask(t *testing.T) {
+	impl, q := newCloseBackupServer(t)
+
+	result, err := impl.CloseBackup(context.Background(), newCloseBackupRequest(false))
+	require.NoError(t, err)
+	require.Equal(t, []string{testEndWAL}, result.GetMissingWalFiles())
+	require.False(t, result.GetTier2Schedule())
+
+	require.Nil(t, receiveBackupTask(t, q, time.Second),
+		"no backup task must be enqueued while the client is still waiting for WALs")
+}

core/proto/klio_wal.proto

@@ -164,6 +164,12 @@ message CloseBackupRequest {
 
   // When present, set the tier2 compression policy to the specified JSON-serialized policy.
   string tier2_compression_policy = 10;
+
+  // Enqueue the post-backup task even if WAL files are still missing. A
+  // client that leaves this unset calls CloseBackup again until no WAL file
+  // is missing, and the server defers the task until that call. A client can
+  // request an immediate enqueue of the post-backup task by setting this to true.
+  bool enqueue_without_wals = 11;
 }
 
 // This is sent by the WAL server in response to a CloseBackupRequest

documentation/web/docs/developer/_protocol.md

@@ -352,6 +352,7 @@ been completed.
 | send_to_tier2 | [bool](#bool) |  | Require this backup to be sent to tier2. |
 | tier2_retention_policy | [string](#string) |  | When present, set the tier2 retention policy to the specified JSON-serialized policy. |
 | tier2_compression_policy | [string](#string) |  | When present, set the tier2 compression policy to the specified JSON-serialized policy. |
+| enqueue_without_wals | [bool](#bool) |  | Enqueue the post-backup task even if WAL files are still missing. A client that leaves this unset calls CloseBackup again until no WAL file is missing, and the server defers the task until that call. A client can request an immediate enqueue of the post-backup task by setting this to true. |
 
 
 

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

@@ -121,8 +121,9 @@ The E2E tests are located in `operator/test/e2e/` and include:
    - `RecoverReplicaCluster`: replica cluster creation from backup
 - **`tablespace_recovery_test.go`** - Recovery preserving PostgreSQL
   tablespaces (`RecoverClusterWithTablespaces`)
-- **`tier2_recovery_test.go`** - Recovery from tier2 S3 storage
-  (`RecoverClusterFromTier2`)
+- **`tier2_recovery_test.go`** - Recovery from tier2 S3 storage of a
+  backup taken on a standby of a two-instance cluster, which must reach
+  tier2 without waiting for its last WAL (`RecoverClusterFromTier2`)
 - **`tier2_pitr_test.go`** - Point-in-time recovery from tier2 storage
   (`RecoverClusterFromTier2Pitr`)
 - **`tier2_recovery_common_test.go`** - Shared tier2 recovery helpers used

operator/test/e2e/tier2_recovery_test.go

@@ -40,6 +40,7 @@ import (
 	machineryConditions "github.com/cloudnative-pg/klio/operator/test/machinery/pkg/conditions"
 	machineryFeatures "github.com/cloudnative-pg/klio/operator/test/machinery/pkg/features"
 	"github.com/cloudnative-pg/klio/operator/test/machinery/pkg/namespaces"
+	"github.com/cloudnative-pg/klio/operator/test/machinery/pkg/postgres"
 )
 
 // tier2RecoveryScenario contains all resources needed for tier2 recovery testing.
@@ -168,6 +169,18 @@ func (s *tier2RecoveryScenario) Teardown(
 	return ctx
 }
 
+// switchWALOnSource forces a WAL switch on the source primary after the
+// backup. A backup taken on a standby cannot switch the WAL itself, so
+// without this the segment holding its end LSN would never be completed on
+// an idle cluster, and the recovery would wait for it forever.
+func (s *tier2RecoveryScenario) switchWALOnSource(
+	ctx context.Context,
+	_ *cnpgv1.Cluster,
+	r *resources.Resources,
+) error {
+	return postgres.CheckpointAndSwitchWal(ctx, r, &s.sourcePrimaryPod)
+}
+
 // deployRecoveryServer creates the second Klio Server after tier2 replication.
 func (s *tier2RecoveryScenario) deployRecoveryServer(
 	ctx context.Context,
@@ -220,21 +233,27 @@ func NewTier2RecoveryFeatureConfig(
 	}
 
 	recoveryConfig := machineryFeatures.RecoveryFeatureConfig{
-		Name:                  name,
-		Setup:                 scenario.Setup,
-		Teardown:              scenario.Teardown,
-		SourcePrimaryPod:      &scenario.sourcePrimaryPod,
-		Backup:                res.Backup,
-		RecoveryCluster:       res.RecoveryCluster,
-		MutateRecoveryCluster: []machineryFeatures.RecoveryClusterMutateFunc{scenario.deployRecoveryServer},
-		BackupTimeout:         5 * time.Minute,
+		Name:             name,
+		Setup:            scenario.Setup,
+		Teardown:         scenario.Teardown,
+		SourcePrimaryPod: &scenario.sourcePrimaryPod,
+		Backup:           res.Backup,
+		RecoveryCluster:  res.RecoveryCluster,
+		MutateRecoveryCluster: []machineryFeatures.RecoveryClusterMutateFunc{
+			scenario.switchWALOnSource,
+			scenario.deployRecoveryServer,
+		},
+		BackupTimeout: 5 * time.Minute,
 	}
 
 	return recoveryConfig
 }
 
 // RecoverClusterFromTier2 returns a RecoveryFeature for tier2 recovery testing.
+// The source cluster has two instances so that the backup, with the default
+// prefer-standby target, is taken on the standby: such a backup does not wait
+// for its last WAL segment and must still be relayed to tier2.
 func RecoverClusterFromTier2(namespace string) *machineryFeatures.RecoveryFeature {
 	return machineryFeatures.NewRecoveryFeature(
-		NewTier2RecoveryFeatureConfig("RecoverClusterFromTier2", 1, namespace))
+		NewTier2RecoveryFeatureConfig("RecoverClusterFromTier2", 2, namespace))
 }