From dd54f584082321e2f31cb99e0ae6cde7a8df7439 Mon Sep 17 00:00:00 2001 From: Gabriele Quaresima Date: Tue, 1 Sep 2026 17:25:13 +0200 Subject: [PATCH 1/4] fix(core): support immediate backup from a replica cluster On a freshly-created replica cluster, an immediate backup could hang forever. With no prior archive to resume from, the WAL streamer of the designated primary started from the current flush position, while pg_backup_start on the underlying standby reports the last replayed restartpoint, which lags behind. The WAL segments between the two were never archived to tier1, so the backup waited for WAL files that would never arrive. Start WAL streaming from the redo point of the latest checkpoint (the latest restartpoint on a standby) instead of the current flush position. That redo point is the earliest LSN a later pg_backup_start on the same instance can report as a backup start, so tier1 always covers the WAL a backup needs. As a safety net, fail the backup with a terminal error when a required WAL predates the earliest archived segment, and can therefore never be archived, instead of letting the client wait indefinitely. Add an e2e scenario that takes an immediate backup from a replica cluster and asserts it completes. Assisted-by: Claude Signed-off-by: Gabriele Quaresima --- core/internal/client/sendwal/receiver.go | 59 +++- core/internal/repository/wals.go | 63 ++++ core/internal/repository/wals_test.go | 62 ++++ core/internal/server/walserver/backup.go | 20 ++ core/internal/server/walserver/backup_test.go | 121 ++++++++ .../web/docs/developer/running-e2e-tests.md | 4 + .../e2e/backup_from_replica_cluster_test.go | 280 ++++++++++++++++++ operator/test/e2e/main_test.go | 1 + 8 files changed, 603 insertions(+), 7 deletions(-) create mode 100644 core/internal/server/walserver/backup_test.go create mode 100644 operator/test/e2e/backup_from_replica_cluster_test.go diff --git a/core/internal/client/sendwal/receiver.go b/core/internal/client/sendwal/receiver.go index 245d5a68..3d5e8932 100644 --- a/core/internal/client/sendwal/receiver.go +++ b/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 { diff --git a/core/internal/repository/wals.go b/core/internal/repository/wals.go index 10997f1c..bc8d29f0 100644 --- a/core/internal/repository/wals.go +++ b/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 +} diff --git a/core/internal/repository/wals_test.go b/core/internal/repository/wals_test.go index 7d765c31..60fe8884 100644 --- a/core/internal/repository/wals_test.go +++ b/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) + }) + } +} diff --git a/core/internal/server/walserver/backup.go b/core/internal/server/walserver/backup.go index 76645ae1..b5c9487c 100644 --- a/core/internal/server/walserver/backup.go +++ b/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, diff --git a/core/internal/server/walserver/backup_test.go b/core/internal/server/walserver/backup_test.go new file mode 100644 index 00000000..220d64ed --- /dev/null +++ b/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()) +} diff --git a/documentation/web/docs/developer/running-e2e-tests.md b/documentation/web/docs/developer/running-e2e-tests.md index 5c173203..a1e83b9e 100644 --- a/documentation/web/docs/developer/running-e2e-tests.md +++ b/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 diff --git a/operator/test/e2e/backup_from_replica_cluster_test.go b/operator/test/e2e/backup_from_replica_cluster_test.go new file mode 100644 index 00000000..aba11731 --- /dev/null +++ b/operator/test/e2e/backup_from_replica_cluster_test.go @@ -0,0 +1,280 @@ +/* +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 e2e + +import ( + "context" + "testing" + "time" + + certmanagerv1 "github.com/cert-manager/cert-manager/pkg/apis/certmanager/v1" + cnpgv1 "github.com/cloudnative-pg/api/pkg/api/v1" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/e2e-framework/klient/k8s/resources" + "sigs.k8s.io/e2e-framework/klient/wait" + "sigs.k8s.io/e2e-framework/pkg/envconf" + "sigs.k8s.io/e2e-framework/pkg/types" + + kliov1alpha1 "github.com/cloudnative-pg/klio/operator/api/v1alpha1" + "github.com/cloudnative-pg/klio/operator/internal/klioconfig" + machineryConditions "github.com/cloudnative-pg/klio/operator/test/machinery/pkg/conditions" + "github.com/cloudnative-pg/klio/operator/test/machinery/pkg/postgres" + "github.com/cloudnative-pg/klio/operator/test/utils/templates/certificates" + "github.com/cloudnative-pg/klio/operator/test/utils/templates/cnpg" + "github.com/cloudnative-pg/klio/operator/test/utils/templates/klio" + "github.com/cloudnative-pg/klio/operator/test/utils/templates/secrets" +) + +// ReplicaClusterBackupFeature verifies that an immediate backup taken from a +// freshly-created replica cluster completes. On a replica cluster the WAL +// streamer of the designated primary starts archiving from the current flush +// position, while pg_backup_start on the underlying standby reports the older +// last-restartpoint LSN: the WAL segments in between must still end up in tier1 +// or the backup waits for WAL files that never arrive. +type ReplicaClusterBackupFeature struct { + scenario *commonBackupRestoreScenario + + // sourceBackup is the base backup of the source cluster the replica + // bootstraps from. + sourceBackup *cnpgv1.Backup + // replicaCluster is the replica cluster archiving to its own tier1. + replicaCluster *cnpgv1.Cluster + // replicaUserCertificate authenticates the replica cluster against the + // Klio server under its own cluster name. + replicaUserCertificate *certmanagerv1.Certificate + // replicaPluginConfiguration wires the replica cluster to its own tier1. + replicaPluginConfiguration *kliov1alpha1.PluginConfiguration + // replicaBackup is the immediate backup taken from the replica cluster. + replicaBackup *cnpgv1.Backup + + sourceBackupTimeout time.Duration + recoveryTimeout time.Duration + // replicaBackupTimeout bounds the wait for the replica backup so a + // never-arriving WAL fails the test instead of hanging. + replicaBackupTimeout time.Duration + checkInterval time.Duration +} + +// BackupFromReplicaCluster builds the "immediate backup from a replica cluster" +// feature: it backs up a source cluster, bootstraps a replica cluster that +// streams from it and archives to its own tier1, then takes an immediate backup +// of the replica cluster and asserts it completes. +func BackupFromReplicaCluster(namespace string) *ReplicaClusterBackupFeature { + const ( + sourceClusterName = "test-cluster-source" + replicaClusterName = "test-cluster-replica" + sourceExternalName = "source-cluster" + ) + + namespaceObj := &corev1.Namespace{ + ObjectMeta: metav1.ObjectMeta{Name: namespace}, + } + + issuer := certificates.GetSelfSignedIssuerObject("selfsigned-issuer", namespace) + certificate := certificates.GetCertificateObject("test", namespace, []string{klioServerName}, issuer) + + caCertificate := certificates.GetCACertificateObject("test-ca", namespace, issuer) + caIssuer := certificates.GetCAIssuerObject("test-ca-issuer", namespace, caCertificate.Spec.SecretName) + + sourceCluster := cnpg.GetCnpgClusterObject(sourceClusterName, namespace, 1, + "klio-plugin-configuration", + cnpg.ClusterTemplateOptions{StorageClass: testCfg.StorageClass}) + // Switch WAL frequently so the segments required by a backup are archived + // promptly, and keep enough WAL around for the replica streamer to resume + // from an older position. + sourceCluster.Spec.PostgresConfiguration.Parameters = map[string]string{ + "archive_timeout": "30s", + "wal_keep_size": "512MB", + } + + sourceUserCertificate := certificates.GetUserCertificateObject( + "klio-user", namespace, "klio-user@"+sourceClusterName, caIssuer) + sourcePluginConfiguration := klio.GetPluginConfigurationObject( + "klio-plugin-configuration", + namespace, + klio.PluginConfigurationTemplateOptions{ + ServerCertificate: certificate, + ClientCertificate: sourceUserCertificate, + ClusterName: sourceClusterName, + }, + ) + // The replica reads the source's tier1 (same server, same cluster name) to + // bootstrap and to stream from it. + sourceExternalPluginConfiguration := sourcePluginConfiguration.DeepCopy() + sourceExternalPluginConfiguration.Name = "klio-plugin-configuration-source" + + ageSecrets := secrets.GetKlioAgeEncryptionSecrets("encryption", namespace, "testencryptionpassword123") + klioServer := klio.GetServerObject( + klioServerName, + namespace, + klio.ServerTemplateOptions{ + Image: testCfg.ServerImage, + StorageClass: testCfg.StorageClass, + TLSSecretName: certificate.Spec.SecretName, + ClientCASecretName: caCertificate.Spec.SecretName, + Encryption: klio.EncryptionOptions{ + EncryptionKeySecretName: ageSecrets.EncryptionKeySecret.Name, + EncryptionKeyFileName: "encryption-key.age", + IdentitySecretName: ageSecrets.IdentitySecret.Name, + IdentityFileName: "identity.txt", + }, + }, + ) + + sourceBackup := cnpg.GetCnpgBackupObject("test-backup-source", namespace, + cnpgv1.BackupTargetPrimary, sourceCluster) + + // The replica cluster archives to its own tier1 (its own cluster name and + // client certificate). + replicaUserCertificate := certificates.GetUserCertificateObject( + "klio-user-replica", namespace, "klio-user@"+replicaClusterName, caIssuer) + replicaPluginConfiguration := klio.GetPluginConfigurationObject( + "klio-plugin-configuration-replica", + namespace, + klio.PluginConfigurationTemplateOptions{ + ServerCertificate: certificate, + ClientCertificate: replicaUserCertificate, + ClusterName: replicaClusterName, + }, + ) + + replicaCluster := sourceCluster.DeepCopy() + replicaCluster.Name = replicaClusterName + replicaCluster.Spec.Plugins[0].Parameters[klioconfig.PluginConfigurationRefParam] = replicaPluginConfiguration.Name + replicaCluster.Spec.Bootstrap = &cnpgv1.BootstrapConfiguration{ + Recovery: &cnpgv1.BootstrapRecovery{ + Source: sourceExternalName, + }, + } + replicaCluster.Spec.ReplicaCluster = &cnpgv1.ReplicaClusterConfiguration{ + Source: sourceExternalName, + Enabled: new(true), + } + replicaCluster.Spec.ExternalClusters = []cnpgv1.ExternalCluster{{ + Name: sourceExternalName, + PluginConfiguration: &cnpgv1.PluginConfiguration{ + Name: "klio.cnpg.io", + Enabled: new(true), + Parameters: map[string]string{ + klioconfig.PluginConfigurationRefParam: sourceExternalPluginConfiguration.Name, + }, + }, + }} + + replicaBackup := cnpg.GetCnpgBackupObject("test-backup-replica", namespace, + cnpgv1.BackupTargetPrimary, replicaCluster) + + scenario := &commonBackupRestoreScenario{ + namespace: namespaceObj, + cnpgCluster: sourceCluster, + userCertificate: sourceUserCertificate, + encryptionSecret: ageSecrets.EncryptionKeySecret, + identitySecret: ageSecrets.IdentitySecret, + issuer: issuer, + caIssuer: caIssuer, + caCertificate: caCertificate, + certificate: certificate, + klioServer: klioServer, + klioPluginConfigurationSource: sourcePluginConfiguration, + klioPluginConfigurationRecovery: sourceExternalPluginConfiguration, + name: "BackupFromReplicaCluster", + } + + return &ReplicaClusterBackupFeature{ + scenario: scenario, + sourceBackup: sourceBackup, + replicaCluster: replicaCluster, + replicaUserCertificate: replicaUserCertificate, + replicaPluginConfiguration: replicaPluginConfiguration, + replicaBackup: replicaBackup, + sourceBackupTimeout: 2 * time.Minute, + recoveryTimeout: 5 * time.Minute, + replicaBackupTimeout: 3 * time.Minute, + checkInterval: 10 * time.Second, + } +} + +// Name returns the feature name. +func (f *ReplicaClusterBackupFeature) Name() string { + return f.scenario.name +} + +// Setup creates the source cluster, the Klio server and the source-side plugin +// configurations, and waits for them to be ready. +func (f *ReplicaClusterBackupFeature) Setup() types.StepFunc { + return f.scenario.Setup +} + +// Run backs up the source cluster, bootstraps the replica cluster, then takes an +// immediate backup of the replica cluster and asserts it completes. +func (f *ReplicaClusterBackupFeature) Run() types.StepFunc { + return func(ctx context.Context, t *testing.T, cfg *envconf.Config) context.Context { + t.Helper() + t.Log("Running backup-from-replica-cluster feature test") + r, err := resources.New(cfg.Client().RESTConfig()) + require.NoError(t, err, "failed to create resources client") + + // Take a base backup of the source cluster so the replica can bootstrap. + require.NoError(t, r.Create(ctx, f.sourceBackup), "failed to create source backup") + require.NoError(t, wait.For( + machineryConditions.BackupIsCompleted(r, f.sourceBackup), + wait.WithTimeout(f.sourceBackupTimeout), + wait.WithInterval(f.checkInterval), + ), "source backup not completed") + + // Advance the source WAL (without a checkpoint) so the replica, once + // bootstrapped, replays past its last restartpoint: this is the state in + // which the streamer starts ahead of what pg_backup_start reports. + _, err = postgres.ExecPostgresQuery(ctx, r, &f.scenario.sourcePrimaryPod, "postgres", + "CREATE TABLE numbers AS SELECT generate_series(1, 1000) AS x; "+ + "SELECT pg_switch_wal(); SELECT pg_switch_wal();") + require.NoError(t, err, "failed to advance source WAL") + + // Create the replica-side archiving resources and the replica cluster. + require.NoError(t, r.Create(ctx, f.replicaUserCertificate), + "failed to create replica user certificate") + require.NoError(t, r.Create(ctx, f.replicaPluginConfiguration), + "failed to create replica plugin configuration") + require.NoError(t, r.Create(ctx, f.replicaCluster), "failed to create replica cluster") + require.NoError(t, wait.For( + machineryConditions.ClusterIsReady(r, f.replicaCluster), + wait.WithTimeout(f.recoveryTimeout), + wait.WithInterval(f.checkInterval), + ), "replica cluster not ready") + + // The immediate backup of the freshly-created replica cluster must + // complete: before the fix it loops forever on missing WAL files. + require.NoError(t, r.Create(ctx, f.replicaBackup), "failed to create replica backup") + require.NoError(t, wait.For( + machineryConditions.BackupIsCompleted(r, f.replicaBackup), + wait.WithTimeout(f.replicaBackupTimeout), + wait.WithInterval(f.checkInterval), + ), "replica cluster backup not completed") + + return ctx + } +} + +// Teardown removes the resources created for the feature. +func (f *ReplicaClusterBackupFeature) Teardown() types.StepFunc { + return f.scenario.Teardown +} diff --git a/operator/test/e2e/main_test.go b/operator/test/e2e/main_test.go index b0b41e72..28958988 100644 --- a/operator/test/e2e/main_test.go +++ b/operator/test/e2e/main_test.go @@ -50,6 +50,7 @@ func TestMain(m *testing.M) { runner.RegisterFeature(BackupFromPrimary(envconf.RandomName("backup-from-primary", 32))) runner.RegisterFeature(BackupFromStandby(envconf.RandomName("backup-from-standby", 32))) + runner.RegisterFeature(BackupFromReplicaCluster(envconf.RandomName("backup-from-replica-cluster", 32))) runner.RegisterFeature(Tier1ServerSideMaintenance(envconf.RandomName("tier1-maintenance", 32))) runner.RegisterFeature(RecoverClusterFromBackupID(envconf.RandomName("recovery-from-backup-id", 32))) runner.RegisterFeature(RecoverClusterFromLatestBackup(envconf.RandomName("recovery-from-latest-backup", 32))) From bab292cbb339b73da34c6ecc7be2c30e12846a21 Mon Sep 17 00:00:00 2001 From: Armando Ruocco Date: Thu, 3 Sep 2026 15:13:30 +0200 Subject: [PATCH 2/4] fix(wals): consider only real segments as the earliest archived WAL GetEarliestWALFileForCluster returned any file found in the earliest WAL directory, including the `.partial` file the writer creates for the segment it is currently receiving, backup labels and history files. Since "000000010000000000000005" sorts before "000000010000000000000005.partial", the segment being streamed right now was reported as older than the earliest archived one, and CloseBackup declared it un-archivable. Restrict the scan to complete 24-character segment names. A directory can now hold no segment at all - the WAL retention skips files carrying an extension, so an orphan `.partial` keeps a directory alive - so the scan continues to the next directory instead of giving up, which would silently disable the caller's check. The doc comment claimed the WAL stream only ever appends segments going forward. Both the in-flight `.partial` and `klio reset-lsn` falsify it, and that claim is what justified skipping the filter. Signed-off-by: Armando Ruocco --- core/internal/repository/wals.go | 53 +++++++++++++++---------- core/internal/repository/wals_test.go | 57 ++++++++++++++++++++------- 2 files changed, 76 insertions(+), 34 deletions(-) diff --git a/core/internal/repository/wals.go b/core/internal/repository/wals.go index bc8d29f0..f3c01db5 100644 --- a/core/internal/repository/wals.go +++ b/core/internal/repository/wals.go @@ -133,12 +133,18 @@ 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. +// GetEarliestWALFileForCluster gets the earliest archived WAL segment for a +// certain cluster, or an empty string when the archive holds none. // -//nolint:cyclop +// 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, @@ -160,38 +166,45 @@ func (c *Connection) GetEarliestWALFileForCluster( return "", fmt.Errorf("while reading cluster directory: %w", err) } - var earliestWalDirectoryName string + // 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 } - if earliestWalDirectoryName == "" || strings.Compare(entry.Name(), earliestWalDirectoryName) == -1 { - earliestWalDirectoryName = entry.Name() + earliestWal, err := c.getEarliestWALFileInDirectory(ctx, path.Join(clusterName, entry.Name())) + if err != nil { + return "", err } - } - if earliestWalDirectoryName == "" { - return "", nil + if earliestWal != "" { + return earliestWal, nil + } } - earliestWalDirectoryName = path.Join(clusterName, earliestWalDirectoryName) - readWalDirectory, err := afero.ReadDir(c.fs, earliestWalDirectoryName) + 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 { - logger.Error(err, "while reading directory", "earliestWalDirectoryName", earliestWalDirectoryName) + log.FromContext(ctx).Error(err, "while reading directory", "directory", directory) return "", fmt.Errorf("while reading WAL directory: %w", err) } - var earliestWal string for _, entry := range readWalDirectory { - if entry.IsDir() { + if entry.IsDir() || len(entry.Name()) != expectedWalFileNameLength { continue } - if earliestWal == "" || strings.Compare(entry.Name(), earliestWal) == -1 { - earliestWal = entry.Name() - } + return entry.Name(), nil } - return earliestWal, nil + return "", nil } diff --git a/core/internal/repository/wals_test.go b/core/internal/repository/wals_test.go index 60fe8884..4b89d076 100644 --- a/core/internal/repository/wals_test.go +++ b/core/internal/repository/wals_test.go @@ -132,9 +132,10 @@ func TestGetEarliestWALFileForCluster(t *testing.T) { tests := []struct { name string clusterName string - createDir bool - walNames []string - expected 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", @@ -144,29 +145,57 @@ func TestGetEarliestWALFileForCluster(t *testing.T) { { name: "several WAL files returns the smallest", clusterName: "test-cluster", - createDir: true, - walNames: []string{ - "00000001000000000000000A", - "00000001000000000000000B", - "00000001000000000000000C", + walDirs: map[string][]string{ + "0000000100000000": { + "00000001000000000000000A", + "00000001000000000000000B", + "00000001000000000000000C", + }, }, expected: "00000001000000000000000A", }, { name: "empty cluster directory", clusterName: "empty-cluster", - createDir: true, + 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) { - 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)) + 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()) } From 90b01e2538a6bf77c6987332db9d42f208cddab2 Mon Sep 17 00:00:00 2001 From: Armando Ruocco Date: Thu, 3 Sep 2026 15:13:30 +0200 Subject: [PATCH 3/4] fix(walserver): validate the cluster name and simplify the WAL gap check The missing WAL list is built by walking a single timeline by ascending position, so it is already sorted and only its first entry can be the oldest required segment: the loop over the whole list was dead weight. CloseBackup was the only WAL server RPC not validating the cluster name it uses to build repository paths. Containment does not depend on it, but every other RPC validates, and this one should not be the exception. The message now also tells the operator what to do about the one state that legitimately reaches it: a backup taken on an instance whose last restartpoint precedes the point the WAL stream started from. Signed-off-by: Armando Ruocco --- core/internal/server/walserver/backup.go | 35 +++++++++++-------- core/internal/server/walserver/backup_test.go | 25 +++++++++++++ 2 files changed, 46 insertions(+), 14 deletions(-) diff --git a/core/internal/server/walserver/backup.go b/core/internal/server/walserver/backup.go index b5c9487c..24084bdf 100644 --- a/core/internal/server/walserver/backup.go +++ b/core/internal/server/walserver/backup.go @@ -32,6 +32,7 @@ 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. @@ -39,6 +40,10 @@ 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 { @@ -46,24 +51,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. + // 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 != "" { - 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) - } - } + 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{ diff --git a/core/internal/server/walserver/backup_test.go b/core/internal/server/walserver/backup_test.go index 220d64ed..5b56bab7 100644 --- a/core/internal/server/walserver/backup_test.go +++ b/core/internal/server/walserver/backup_test.go @@ -119,3 +119,28 @@ func TestCloseBackupWaitsForRecentMissingWAL(t *testing.T) { require.NotNil(t, result) assert.Equal(t, []string{"000000010000000000000006"}, result.GetMissingWalFiles()) } + +// TestCloseBackupWaitsForWALStillBeingStreamed verifies that the in-flight +// `.partial` file the WAL writer creates for the segment it is receiving does +// not make that same segment look permanently un-archivable. This is the state +// a freshly created cluster is in when its first backup closes. +func TestCloseBackupWaitsForWALStillBeingStreamed(t *testing.T) { + const clusterName = "test-cluster" + + // Nothing is archived yet: segment 05 is still being received. + impl := newTestImplementation(t, clusterName, []string{ + "000000010000000000000005.partial", + }) + + result, err := impl.CloseBackup(context.Background(), &grpc.CloseBackupRequest{ + ClusterName: clusterName, + Timeline: 1, + StartWal: "000000010000000000000005", + EndWal: "000000010000000000000005", + SegmentSize: closeBackupSegmentSize, + }) + + require.NoError(t, err) + require.NotNil(t, result) + assert.Equal(t, []string{"000000010000000000000005"}, result.GetMissingWalFiles()) +} From 33c3cb0edee351cca34280ce24099d13e2903a5c Mon Sep 17 00:00:00 2001 From: Armando Ruocco Date: Thu, 3 Sep 2026 15:13:30 +0200 Subject: [PATCH 4/4] fix(sendwal): do not fall back to the flush position Falling back to the current flush position when the checkpoint redo LSN cannot be read reinstates the very gap this start point avoids, and it does so permanently: once the replication slot and the Klio server hold a resume point past the gap, no later run goes back to fill it. Returning the error restarts the sidecar, which retries with a fresh redo point, like every other failure in this path. Signed-off-by: Armando Ruocco --- core/internal/client/sendwal/receiver.go | 12 ++++-------- 1 file changed, 4 insertions(+), 8 deletions(-) diff --git a/core/internal/client/sendwal/receiver.go b/core/internal/client/sendwal/receiver.go index 3d5e8932..5a6fe9c5 100644 --- a/core/internal/client/sendwal/receiver.go +++ b/core/internal/client/sendwal/receiver.go @@ -215,16 +215,12 @@ func (s *Process) getReplicationStartPointFromClient( // 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 { - 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 + return 0, err } contextLogger.Debug(