From 04fda3a6c86af7f65004de92c724db08309e40f0 Mon Sep 17 00:00:00 2001 From: Gabriele Quaresima Date: Tue, 1 Sep 2026 17:25:13 +0200 Subject: [PATCH 1/6] 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 245d5a687..3d5e89323 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 10997f1c7..bc8d29f07 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 7d765c317..60fe88848 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 fefe4d85a..d5f6a3465 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 000000000..220d64edd --- /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 b63f1565a..9c7edc40f 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 000000000..aba117312 --- /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 b05680ea5..b5a80ff22 100644 --- a/operator/test/e2e/main_test.go +++ b/operator/test/e2e/main_test.go @@ -51,6 +51,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 39148a297b436b4bb1a72f6abdc0232dbb5b7e03 Mon Sep 17 00:00:00 2001 From: Armando Ruocco Date: Thu, 3 Sep 2026 15:13:30 +0200 Subject: [PATCH 2/6] 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 bc8d29f07..f3c01db5e 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 60fe88848..4b89d0769 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 83ece02cad5a0f7a525af4406b39f17ab5ed5d20 Mon Sep 17 00:00:00 2001 From: Armando Ruocco Date: Thu, 3 Sep 2026 15:13:30 +0200 Subject: [PATCH 3/6] 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 d5f6a3465..7ba3c74a2 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 220d64edd..5b56bab71 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 91c625df2336986eeb667456a67082c06c7b56a9 Mon Sep 17 00:00:00 2001 From: Armando Ruocco Date: Thu, 3 Sep 2026 15:13:30 +0200 Subject: [PATCH 4/6] 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 3d5e89323..5a6fe9c5e 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( From 98fe1907f610eb80137c1bb2bc981401599f4037 Mon Sep 17 00:00:00 2001 From: Francesco Canovai Date: Thu, 10 Sep 2026 14:12:27 +0200 Subject: [PATCH 5/6] feat(sendwal): use RESERVE_WAL keyword when creating the replication slot Signed-off-by: Francesco Canovai --- core/internal/client/sendwal/receiver.go | 77 ++------- core/internal/repository/wals.go | 76 --------- core/internal/repository/wals_test.go | 91 ----------- core/internal/server/walserver/backup.go | 27 ---- core/internal/server/walserver/backup_test.go | 146 ------------------ 5 files changed, 14 insertions(+), 403 deletions(-) delete mode 100644 core/internal/server/walserver/backup_test.go diff --git a/core/internal/client/sendwal/receiver.go b/core/internal/client/sendwal/receiver.go index 5a6fe9c5e..cda72122f 100644 --- a/core/internal/client/sendwal/receiver.go +++ b/core/internal/client/sendwal/receiver.go @@ -31,7 +31,6 @@ 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" @@ -158,6 +157,10 @@ func (s *Process) Start(ctx context.Context) error { "systemID", identifyData.SystemID, ) + if err := s.ensureReplicationSlotExists(ctx, conn); err != nil { + return err + } + // Negotiate the starting point with the server point, err := s.getReplicationStartPoint(ctx, conn, identifyData, walSegmentSize) if err != nil { @@ -174,10 +177,6 @@ func (s *Process) Start(ctx context.Context) error { contextLogger.Debug("Some timeline history files could not be processed", "innerErr", histErr.Error()) } - if err := s.ensureReplicationSlotExists(ctx, conn); err != nil { - return err - } - return s.startReplication(ctx, conn, point, walSegmentSize) } @@ -203,59 +202,16 @@ func (s *Process) getReplicationStartPointFromClient( return slotResult.RestartLSN, nil } - // 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. - // - // 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 - } - + // 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. contextLogger.Debug( - "Checkpoint redo LSN", - "redoStart", redoStart, + "Current flush LSN", "xlogFlushPos", xlogFlushPos, "segmentSize", segmentSize, ) - 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 + return getStartWALLSN(xlogFlushPos, segmentSize), nil } type walCoordinate struct { @@ -349,16 +305,11 @@ func (s *Process) ensureReplicationSlotExists( return nil } - replicationSlotResult, err := pglogrepl.CreateReplicationSlot( - ctx, - conn, - s.config.Source.Slot, - "", // output plugin name: this is meaningful only for logical replication - pglogrepl.CreateReplicationSlotOptions{ - Temporary: false, - Mode: pglogrepl.PhysicalReplication, - }, - ) + // pglogrepl.CreateReplicationSlotOptions has no RESERVE_WAL field, so the + // command is built and parsed manually here. + sql := fmt.Sprintf("CREATE_REPLICATION_SLOT %s PHYSICAL RESERVE_WAL", s.config.Source.Slot) + + replicationSlotResult, err := pglogrepl.ParseCreateReplicationSlot(conn.Exec(ctx, sql)) if err != nil { return fmt.Errorf("while creating temporary replication slot: %w", err) } diff --git a/core/internal/repository/wals.go b/core/internal/repository/wals.go index f3c01db5e..10997f1c7 100644 --- a/core/internal/repository/wals.go +++ b/core/internal/repository/wals.go @@ -132,79 +132,3 @@ 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 -} diff --git a/core/internal/repository/wals_test.go b/core/internal/repository/wals_test.go index 4b89d0769..7d765c317 100644 --- a/core/internal/repository/wals_test.go +++ b/core/internal/repository/wals_test.go @@ -116,94 +116,3 @@ 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) - }) - } -} diff --git a/core/internal/server/walserver/backup.go b/core/internal/server/walserver/backup.go index 7ba3c74a2..fefe4d85a 100644 --- a/core/internal/server/walserver/backup.go +++ b/core/internal/server/walserver/backup.go @@ -32,7 +32,6 @@ 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. @@ -40,10 +39,6 @@ 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 { @@ -51,28 +46,6 @@ func (w *Implementation) CloseBackup( } 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, diff --git a/core/internal/server/walserver/backup_test.go b/core/internal/server/walserver/backup_test.go deleted file mode 100644 index 5b56bab71..000000000 --- a/core/internal/server/walserver/backup_test.go +++ /dev/null @@ -1,146 +0,0 @@ -/* -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()) -} - -// 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 7aaec1ead420be3f1c881b39f360b80d35492fc4 Mon Sep 17 00:00:00 2001 From: Francesco Canovai Date: Thu, 10 Sep 2026 19:16:10 +0200 Subject: [PATCH 6/6] docs: better comments Signed-off-by: Francesco Canovai --- core/internal/client/sendwal/receiver.go | 8 ++++++++ operator/test/e2e/backup_from_replica_cluster_test.go | 11 +++-------- 2 files changed, 11 insertions(+), 8 deletions(-) diff --git a/core/internal/client/sendwal/receiver.go b/core/internal/client/sendwal/receiver.go index cda72122f..4dcf4da74 100644 --- a/core/internal/client/sendwal/receiver.go +++ b/core/internal/client/sendwal/receiver.go @@ -157,6 +157,9 @@ func (s *Process) Start(ctx context.Context) error { "systemID", identifyData.SystemID, ) + // The slot must exist, with its WAL reserved, before we read its restart + // LSN as the start point below: otherwise the WAL between here and the + // slot's eventual creation could be recycled before we ever read it. if err := s.ensureReplicationSlotExists(ctx, conn); err != nil { return err } @@ -288,6 +291,11 @@ func getStartWALLSN(xlogFlushPos pglogrepl.LSN, segmentSize uint64) pglogrepl.LS return pglogrepl.LSN(uint64(xlogFlushPos) & ^(segmentSize - 1)) } +// ensureReplicationSlotExists creates the Klio physical replication slot if +// it does not exist yet, using RESERVE_WAL so its restart LSN, and the WAL +// from it, are reserved immediately at creation instead of at the first +// replication connection. Without it, WAL between slot creation and that +// first connection is free to be recycled before Klio ever streams it. func (s *Process) ensureReplicationSlotExists( ctx context.Context, conn *pgconn.PgConn, diff --git a/operator/test/e2e/backup_from_replica_cluster_test.go b/operator/test/e2e/backup_from_replica_cluster_test.go index aba117312..a4f415be7 100644 --- a/operator/test/e2e/backup_from_replica_cluster_test.go +++ b/operator/test/e2e/backup_from_replica_cluster_test.go @@ -45,11 +45,7 @@ import ( ) // 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. +// freshly-created replica cluster completes. type ReplicaClusterBackupFeature struct { scenario *commonBackupRestoreScenario @@ -242,8 +238,7 @@ func (f *ReplicaClusterBackupFeature) Run() types.StepFunc { ), "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. + // bootstrapped, has the final WAL and isn't stuck waiting for it. _, 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();") @@ -262,7 +257,7 @@ func (f *ReplicaClusterBackupFeature) Run() types.StepFunc { ), "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. + // complete. require.NoError(t, r.Create(ctx, f.replicaBackup), "failed to create replica backup") require.NoError(t, wait.For( machineryConditions.BackupIsCompleted(r, f.replicaBackup),