From bf955430cb58987bbbe3cd2925ea06700cf85609 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Niccol=C3=B2=20Fei?= Date: Thu, 9 Jul 2026 10:36:09 +0200 Subject: [PATCH 1/3] fix: reduce startupProbe periodSeconds without losing failure tolerance (#992) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The startup probe for the injected plugin-barman-cloud sidecar previously left `periodSeconds` unset, so the API server defaulted it to 10s. Because the sidecar is a native init container that gates the main postgres container on reaching `Started`, this added roughly one full period to every pod's startup, even though the probe itself (a local unix-socket health check) normally succeeds in milliseconds. `periodSeconds` is now 1s, so the probe reports success almost immediately in the common case. To avoid trading away failure tolerance for that faster common case, `failureThreshold` is raised to 30 and `timeoutSeconds` is lowered to 5s: a unix-socket call essentially never times out under mere load, so hitting the timeout means the sidecar is genuinely unresponsive rather than just slow, and it's fine to give that rare case more attempts before restarting the container. Closes #991 Signed-off-by: Niccolò Fei Signed-off-by: Marco Nenciarini Co-authored-by: Marco Nenciarini --- internal/cnpgi/operator/lifecycle.go | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/internal/cnpgi/operator/lifecycle.go b/internal/cnpgi/operator/lifecycle.go index 776b301..8bb26d8 100644 --- a/internal/cnpgi/operator/lifecycle.go +++ b/internal/cnpgi/operator/lifecycle.go @@ -403,8 +403,9 @@ func reconcilePodSpec( envs = append(envs, config.env...) baseProbe := &corev1.Probe{ - FailureThreshold: 10, - TimeoutSeconds: 10, + PeriodSeconds: 1, + FailureThreshold: 30, + TimeoutSeconds: 5, ProbeHandler: corev1.ProbeHandler{ Exec: &corev1.ExecAction{ Command: []string{"/manager", "healthcheck", "unix"}, From b0e45381473de42f77f15c44a846478585e35ea2 Mon Sep 17 00:00:00 2001 From: Marco Nenciarini Date: Thu, 9 Jul 2026 11:00:04 +0200 Subject: [PATCH 2/3] chore(renovate): stop ignoring test/e2e/internal/tests/** (#996) ignorePaths still had '**/tests/**' left over from the recommended preset, even though a comment right above documents removing '**/test/**' specifically to let renovate scan test/e2e for emulator image dependencies. Every e2e Ginkgo package lives under test/e2e/internal/tests/**, so the plural pattern was silently excluding all of them; any image pinned by digest in those packages (e.g. minio, aws-cli) was never going to get a bump PR. No other tracked path in the repo matches '**/tests/**' once node_modules is excluded by its own rule, so this narrows nothing else. Signed-off-by: Marco Nenciarini --- renovate.json5 | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/renovate.json5 b/renovate.json5 index b5b8a32..132ff19 100644 --- a/renovate.json5 +++ b/renovate.json5 @@ -12,14 +12,14 @@ rebaseWhen: 'never', prConcurrentLimit: 5, // Override default ignorePaths to scan test/e2e for emulator image dependencies - // Removed: '**/test/**' + // Removed: '**/test/**', '**/tests/**' (this repo's e2e Ginkgo packages live + // under test/e2e/internal/tests/**, which the plural pattern was excluding) ignorePaths: [ '**/node_modules/**', '**/bower_components/**', '**/vendor/**', '**/examples/**', '**/__tests__/**', - '**/tests/**', '**/__fixtures__/**', ], lockFileMaintenance: { From 21aadbfd11a9346095931f2fbf7d754cdeb59ec9 Mon Sep 17 00:00:00 2001 From: Armando Ruocco Date: Thu, 9 Jul 2026 11:26:08 +0200 Subject: [PATCH 3/3] test(e2e): cover parallel WAL restore via the plugin (#978) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Recreate the parallel WAL-restore coverage in the plugin repo: a 2-instance cluster archiving to minio with wal.maxParallel=3, forged WAL segments on the object store, and assertions on the plugin's prefetch/spool/end-of-wal-stream state machine driven through `/controller/manager wal-restore` on the standby. Part of cloudnative-pg/cloudnative-pg#10954. Signed-off-by: Armando Ruocco Signed-off-by: Niccolò Fei Signed-off-by: Marco Nenciarini Co-authored-by: Niccolò Fei Co-authored-by: Marco Nenciarini --- test/e2e/e2e_suite_test.go | 1 + test/e2e/internal/tests/walrestore/doc.go | 24 ++ .../e2e/internal/tests/walrestore/fixtures.go | 178 ++++++++++ .../internal/tests/walrestore/walrestore.go | 312 ++++++++++++++++++ 4 files changed, 515 insertions(+) create mode 100644 test/e2e/internal/tests/walrestore/doc.go create mode 100644 test/e2e/internal/tests/walrestore/fixtures.go create mode 100644 test/e2e/internal/tests/walrestore/walrestore.go diff --git a/test/e2e/e2e_suite_test.go b/test/e2e/e2e_suite_test.go index bbf7b90..0f2bfdf 100644 --- a/test/e2e/e2e_suite_test.go +++ b/test/e2e/e2e_suite_test.go @@ -39,6 +39,7 @@ import ( _ "github.com/cloudnative-pg/plugin-barman-cloud/test/e2e/internal/tests/backup" _ "github.com/cloudnative-pg/plugin-barman-cloud/test/e2e/internal/tests/credentialrotation" _ "github.com/cloudnative-pg/plugin-barman-cloud/test/e2e/internal/tests/replicacluster" + _ "github.com/cloudnative-pg/plugin-barman-cloud/test/e2e/internal/tests/walrestore" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" diff --git a/test/e2e/internal/tests/walrestore/doc.go b/test/e2e/internal/tests/walrestore/doc.go new file mode 100644 index 0000000..00fbafb --- /dev/null +++ b/test/e2e/internal/tests/walrestore/doc.go @@ -0,0 +1,24 @@ +/* +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 walrestore contains the end-to-end test for the parallel WAL restore +// behaviour of the Barman Cloud Plugin: prefetching upcoming segments into the +// spool directory, serving later requests from the spool, and tracking the +// end-of-wal-stream sentinel. +package walrestore diff --git a/test/e2e/internal/tests/walrestore/fixtures.go b/test/e2e/internal/tests/walrestore/fixtures.go new file mode 100644 index 0000000..e417dd9 --- /dev/null +++ b/test/e2e/internal/tests/walrestore/fixtures.go @@ -0,0 +1,178 @@ +/* +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 walrestore + +import ( + cloudnativepgv1 "github.com/cloudnative-pg/api/pkg/api/v1" + barmanapi "github.com/cloudnative-pg/barman-cloud/pkg/api" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/utils/ptr" + + pluginBarmanCloudV1 "github.com/cloudnative-pg/plugin-barman-cloud/api/v1" + "github.com/cloudnative-pg/plugin-barman-cloud/test/e2e/internal/objectstore" +) + +const ( + minioName = "minio" + objectStoreName = "source" + clusterName = "source" + s3ClientName = "s3-client" + storageSize = "1Gi" + // walMaxParallel is the prefetch parallelism under test: for a regular WAL + // request the plugin restores the requested segment and prefetches the next + // ones, up to this many segments in total. + walMaxParallel = 3 +) + +// newObjectStoreResources returns the minio server Deployment/Service/Secret/PVC. +func newObjectStoreResources(namespace string) *objectstore.Resources { + return objectstore.NewMinioObjectStoreResources(namespace, minioName) +} + +// newObjectStore returns a minio-backed ObjectStore configured with the WAL +// prefetch parallelism (maxParallel) under test. Archiving with gzip makes the +// archived segments carry the ".gz" suffix that forged segments are copied from. +func newObjectStore(namespace string) *pluginBarmanCloudV1.ObjectStore { + store := objectstore.NewMinioObjectStore(namespace, objectStoreName, minioName) + store.Spec.Configuration.Wal = &barmanapi.WalBackupConfiguration{ + MaxParallel: walMaxParallel, + Compression: barmanapi.CompressionTypeGzip, + } + return store +} + +// newCluster returns a 2-instance cluster that uses the plugin as its WAL +// archiver, so the standby drives WAL restore (and its prefetch/spool/ +// end-of-wal-stream state machine) through the plugin. +func newCluster(namespace string) *cloudnativepgv1.Cluster { + return &cloudnativepgv1.Cluster{ + TypeMeta: metav1.TypeMeta{ + Kind: "Cluster", + APIVersion: "postgresql.cnpg.io/v1", + }, + ObjectMeta: metav1.ObjectMeta{ + Name: clusterName, + Namespace: namespace, + }, + Spec: cloudnativepgv1.ClusterSpec{ + Instances: 2, + ImagePullPolicy: corev1.PullAlways, + Plugins: []cloudnativepgv1.PluginConfiguration{ + { + Name: "barman-cloud.cloudnative-pg.io", + Parameters: map[string]string{ + "barmanObjectName": objectStoreName, + }, + IsWALArchiver: ptr.To(true), + }, + }, + PostgresConfiguration: cloudnativepgv1.PostgresConfiguration{ + Parameters: map[string]string{ + "log_min_messages": "DEBUG4", + }, + }, + StorageConfiguration: cloudnativepgv1.StorageConfiguration{ + Size: storageSize, + }, + }, + } +} + +// newS3ClientDeployment returns a deployment running the AWS CLI configured to +// talk to the in-namespace minio service. The test execs `aws s3` commands in +// it to forge WAL segments on the object store and to check their presence. +func newS3ClientDeployment(namespace string) *appsv1.Deployment { + labels := map[string]string{"app": s3ClientName} + return &appsv1.Deployment{ + TypeMeta: metav1.TypeMeta{ + Kind: "Deployment", + APIVersion: "apps/v1", + }, + ObjectMeta: metav1.ObjectMeta{ + Name: s3ClientName, + Namespace: namespace, + }, + Spec: appsv1.DeploymentSpec{ + Replicas: ptr.To(int32(1)), + Selector: &metav1.LabelSelector{MatchLabels: labels}, + Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{Labels: labels}, + Spec: corev1.PodSpec{ + Containers: []corev1.Container{ + { + Name: s3ClientName, + // renovate: datasource=docker depName=amazon/aws-cli versioning=docker + // Version: 2.35.11 + Image: "docker.io/amazon/aws-cli@sha256:749bfaf91d690b9a1768083822d620f96c19defdf9ca2dc227eb3695281fda5b", + Command: []string{"sleep", "infinity"}, + Env: []corev1.EnvVar{ + { + Name: "AWS_ENDPOINT_URL", + Value: "http://" + minioName + ":9000", + }, + { + Name: "AWS_ACCESS_KEY_ID", + ValueFrom: &corev1.EnvVarSource{ + SecretKeyRef: &corev1.SecretKeySelector{ + LocalObjectReference: corev1.LocalObjectReference{Name: minioName}, + Key: "ACCESS_KEY_ID", + }, + }, + }, + { + Name: "AWS_SECRET_ACCESS_KEY", + ValueFrom: &corev1.EnvVarSource{ + SecretKeyRef: &corev1.SecretKeySelector{ + LocalObjectReference: corev1.LocalObjectReference{Name: minioName}, + Key: "ACCESS_SECRET_KEY", + }, + }, + }, + { + Name: "AWS_DEFAULT_REGION", + Value: "us-east-1", + }, + // The CRC-based default checksums introduced in AWS + // CLI 2.23 are not supported by every S3-compatible + // object store, minio included. + { + Name: "AWS_REQUEST_CHECKSUM_CALCULATION", + Value: "when_required", + }, + { + Name: "AWS_RESPONSE_CHECKSUM_VALIDATION", + Value: "when_required", + }, + }, + SecurityContext: &corev1.SecurityContext{ + AllowPrivilegeEscalation: ptr.To(false), + SeccompProfile: &corev1.SeccompProfile{ + Type: corev1.SeccompProfileTypeRuntimeDefault, + }, + }, + }, + }, + }, + }, + }, + } +} diff --git a/test/e2e/internal/tests/walrestore/walrestore.go b/test/e2e/internal/tests/walrestore/walrestore.go new file mode 100644 index 0000000..287748f --- /dev/null +++ b/test/e2e/internal/tests/walrestore/walrestore.go @@ -0,0 +1,312 @@ +/* +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 walrestore + +import ( + "context" + "errors" + "fmt" + "strconv" + "strings" + "time" + + corev1 "k8s.io/api/core/v1" + apitypes "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" + executil "k8s.io/client-go/util/exec" + "sigs.k8s.io/controller-runtime/pkg/client" + + internalClient "github.com/cloudnative-pg/plugin-barman-cloud/test/e2e/internal/client" + internalCluster "github.com/cloudnative-pg/plugin-barman-cloud/test/e2e/internal/cluster" + "github.com/cloudnative-pg/plugin-barman-cloud/test/e2e/internal/command" + "github.com/cloudnative-pg/plugin-barman-cloud/test/e2e/internal/deployment" + nmsp "github.com/cloudnative-pg/plugin-barman-cloud/test/e2e/internal/namespace" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +const ( + // spoolDirectory is where the plugin sidecar prefetches WAL segments and + // records the end-of-wal-stream sentinel. It must match the SPOOL_DIRECTORY + // the operator injects on the sidecar (internal/cnpgi/operator/lifecycle.go), + // which lives on the CloudNativePG scratch-data volume shared with the + // postgres container, so the path is visible from the postgres container too. + spoolDirectory = "/controller/wal-restore-spool" + // pgWalPath is PgDataPath + "/pg_wal" in the CloudNativePG operand image. + pgWalPath = "/var/lib/postgresql/data/pgdata/pg_wal" + // endOfWALStreamFlag is the sentinel file the barman-cloud restorer writes in + // the spool to record that the archive ran out of segments. + endOfWALStreamFlag = "end-of-wal-stream" + // managerExecutable is the CloudNativePG instance manager. Its wal-restore + // subcommand delegates to the plugin when the cluster uses it as WAL archiver. + managerExecutable = "/controller/manager" + // postgresContainer is the container we exec into on instance pods. + postgresContainer = "postgres" + // walLogDir is the WALs subdirectory (timeline + log id) the forged segments + // live under; a freshly bootstrapped, idle cluster stays within it. + walLogDir = "0000000100000000" + // bucket is the destination bucket of the minio ObjectStore. + bucket = "backups" +) + +// walFile returns the name of the n-th forged WAL segment (segment 0xF0+n on +// timeline 1, log 0, for n <= 15). The high segment number keeps it out of the +// range an idle PostgreSQL would archive on its own. +func walFile(n int) string { + return fmt.Sprintf("0000000100000000%08X", 0xF0+n) +} + +// walObjectURI returns the s3 URI of a wals object (by file name) in the store. +func walObjectURI(name string) string { + return fmt.Sprintf("s3://%s/%s/wals/%s/%s", bucket, clusterName, walLogDir, name) +} + +// execInPod runs a command in the given container and returns stdout, stderr +// and the error (non-nil for a non-zero exit code). +func execInPod( + ctx context.Context, + clientSet *kubernetes.Clientset, + cfg *rest.Config, + namespace, pod, container string, + args ...string, +) (string, string, error) { + return command.ExecuteInContainer( + ctx, + *clientSet, + cfg, + command.ContainerLocator{ + NamespaceName: namespace, + PodName: pod, + ContainerName: container, + }, + nil, + args, + ) +} + +// This test drives the plugin's parallel WAL restore directly: it invokes the +// instance-manager wal-restore command on the standby (which delegates to the +// plugin) and asserts the prefetch/spool/end-of-wal-stream state machine with +// maxParallel = 3. To control the archive deterministically, it forges WAL +// segments on the object store by copying a real archived segment under new +// names. +var _ = Describe("Parallel WAL restore", func() { + var ( + namespace *corev1.Namespace + cl client.Client + clientSet *kubernetes.Clientset + cfg *rest.Config + ) + + BeforeEach(func(ctx SpecContext) { + var err error + cl, _, err = internalClient.NewClient() + Expect(err).NotTo(HaveOccurred()) + clientSet, cfg, err = internalClient.NewClientSet() + Expect(err).NotTo(HaveOccurred()) + namespace, err = nmsp.CreateUniqueNamespace(ctx, cl, "wal-restore-parallel") + Expect(err).NotTo(HaveOccurred()) + }) + + AfterEach(func(ctx SpecContext) { + Expect(cl.Delete(ctx, namespace)).To(Succeed()) + }) + + It("prefetches segments, serves them from the spool and tracks end-of-wal-stream", + func(ctx SpecContext) { + ns := namespace.Name + + By("creating the object store backing resources") + Expect(newObjectStoreResources(ns).Create(ctx, cl)).To(Succeed()) + + By("creating the ObjectStore with WAL maxParallel") + Expect(cl.Create(ctx, newObjectStore(ns))).To(Succeed()) + + By("deploying the S3 client used to forge and inspect WAL segments") + Expect(cl.Create(ctx, newS3ClientDeployment(ns))).To(Succeed()) + + By("creating the cluster using the plugin as WAL archiver") + cluster := newCluster(ns) + Expect(cl.Create(ctx, cluster)).To(Succeed()) + + By("waiting for the cluster to become ready") + Eventually(func(g Gomega) { + g.Expect(cl.Get(ctx, + apitypes.NamespacedName{Name: clusterName, Namespace: ns}, + cluster)).To(Succeed()) + g.Expect(internalCluster.IsReady(*cluster)).To(BeTrue()) + }).WithTimeout(10 * time.Minute).WithPolling(10 * time.Second).Should(Succeed()) + + By("waiting for the S3 client to become ready") + Eventually(func(g Gomega) { + ready, err := deployment.IsReady(ctx, cl, + apitypes.NamespacedName{Name: s3ClientName, Namespace: ns}) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(ready).To(BeTrue()) + }).WithTimeout(2 * time.Minute).WithPolling(5 * time.Second).Should(Succeed()) + + primary := cluster.Status.CurrentPrimary + Expect(primary).NotTo(BeEmpty()) + standby := clusterName + "-2" + + var s3ClientPods corev1.PodList + Expect(cl.List(ctx, &s3ClientPods, + client.InNamespace(ns), + client.MatchingLabels{"app": s3ClientName})).To(Succeed()) + Expect(s3ClientPods.Items).NotTo(BeEmpty()) + s3Client := s3ClientPods.Items[0].Name + + // Operations scoped to the fixed pods/clients, kept as closures so the + // step assertions below read like the original state-machine table. + restore := func(name string) error { + _, _, err := execInPod(ctx, clientSet, cfg, ns, standby, postgresContainer, + managerExecutable, "wal-restore", name, pgWalPath+"/"+name) + return err + } + existsIn := func(dir, name string) bool { + _, _, err := execInPod(ctx, clientSet, cfg, ns, standby, postgresContainer, + "test", "-f", dir+"/"+name) + return err == nil + } + flagSet := func() bool { return existsIn(spoolDirectory, endOfWALStreamFlag) } + spoolSegments := func() int { + out, _, _ := execInPod(ctx, clientSet, cfg, ns, standby, postgresContainer, + "sh", "-c", + "ls -1 "+spoolDirectory+" 2>/dev/null | grep -Ec '^[0-9A-F]{24}$' || true") + n, _ := strconv.Atoi(strings.TrimSpace(out)) + return n + } + purgeSpool := func() { + _, _, _ = execInPod(ctx, clientSet, cfg, ns, standby, postgresContainer, + "sh", "-c", "rm -f "+spoolDirectory+"/* 2>/dev/null; true") + } + forge := func(src, dst string) { + // ExecuteInContainer drops stdout/stderr on a non-zero exit, so on + // failure this only reports the exit code, not the aws CLI's error text. + _, _, err := execInPod(ctx, clientSet, cfg, ns, s3Client, s3ClientName, + "aws", "s3", "cp", walObjectURI(src), walObjectURI(dst)) + Expect(err).NotTo(HaveOccurred(), "forging %s -> %s", src, dst) + } + objectExists := func(name string) bool { + out, _, err := execInPod(ctx, clientSet, cfg, ns, s3Client, s3ClientName, + "aws", "s3", "ls", walObjectURI(name)) + return err == nil && strings.TrimSpace(out) != "" + } + + By("archiving a real WAL on the primary and learning its name") + _, _, err := execInPod(ctx, clientSet, cfg, ns, primary, postgresContainer, + "psql", "-tAc", "CHECKPOINT") + Expect(err).NotTo(HaveOccurred(), "CHECKPOINT on the primary failed") + out, _, err := execInPod(ctx, clientSet, cfg, ns, primary, postgresContainer, + "psql", "-tAc", "SELECT pg_walfile_name(pg_switch_wal())") + Expect(err).NotTo(HaveOccurred(), "switching WAL on the primary failed") + latestWAL := strings.TrimSpace(out) + Expect(latestWAL).To(HavePrefix(walLogDir), + "the freshly bootstrapped cluster should still be on the first WAL log") + + By("waiting for the archived WAL to land on the object store") + Eventually(func() bool { + return objectExists(latestWAL + ".gz") + }).WithTimeout(2 * time.Minute).WithPolling(5 * time.Second).Should(BeTrue()) + + By("forging WAL segments #1 to #5 from the archived WAL") + for n := 1; n <= 5; n++ { + forge(latestWAL+".gz", walFile(n)) + } + + By("ensuring the spool directory is empty on the standby") + purgeSpool() + + // #1: served fresh; #2 and #3 prefetched into the spool; flag unset. + By("requesting WAL #1: #1 restored, #2 and #3 prefetched") + Expect(restore(walFile(1))).To(Succeed()) + Eventually(func(g Gomega) { + g.Expect(existsIn(pgWalPath, walFile(1))).To(BeTrue(), "#1 in pg_wal") + g.Expect(existsIn(spoolDirectory, walFile(2))).To(BeTrue(), "#2 in spool") + g.Expect(existsIn(spoolDirectory, walFile(3))).To(BeTrue(), "#3 in spool") + g.Expect(flagSet()).To(BeFalse(), "end-of-wal-stream unset") + }).WithTimeout(time.Minute).WithPolling(2 * time.Second).Should(Succeed()) + + // #2: served from the spool; #3 stays prefetched; no new prefetch; flag unset. + By("requesting WAL #2: served from the spool, #3 still prefetched") + Expect(restore(walFile(2))).To(Succeed()) + Eventually(func(g Gomega) { + g.Expect(existsIn(pgWalPath, walFile(2))).To(BeTrue(), "#2 in pg_wal") + g.Expect(existsIn(spoolDirectory, walFile(3))).To(BeTrue(), "#3 in spool") + g.Expect(flagSet()).To(BeFalse(), "end-of-wal-stream unset") + }).WithTimeout(time.Minute).WithPolling(2 * time.Second).Should(Succeed()) + + // #3: served from the spool; spool now empty; flag unset. + By("requesting WAL #3: served from the spool, spool now empty") + Expect(restore(walFile(3))).To(Succeed()) + Eventually(func(g Gomega) { + g.Expect(existsIn(pgWalPath, walFile(3))).To(BeTrue(), "#3 in pg_wal") + g.Expect(spoolSegments()).To(Equal(0), "no WAL segments in spool") + g.Expect(flagSet()).To(BeFalse(), "end-of-wal-stream unset") + }).WithTimeout(time.Minute).WithPolling(2 * time.Second).Should(Succeed()) + + // #4: served fresh; #5 prefetched; #6 absent so end-of-wal-stream is set. + By("requesting WAL #4: #4 restored, #5 prefetched, end-of-wal-stream set") + Expect(restore(walFile(4))).To(Succeed()) + Eventually(func(g Gomega) { + g.Expect(existsIn(pgWalPath, walFile(4))).To(BeTrue(), "#4 in pg_wal") + g.Expect(existsIn(spoolDirectory, walFile(5))).To(BeTrue(), "#5 in spool") + g.Expect(flagSet()).To(BeTrue(), "end-of-wal-stream set (#6 absent)") + }).WithTimeout(time.Minute).WithPolling(2 * time.Second).Should(Succeed()) + + By("forging WAL segment #6 on the object store") + forge(latestWAL+".gz", walFile(6)) + + // #5: served from the spool; flag untouched (served before it is checked). + By("requesting WAL #5: served from the spool, end-of-wal-stream still set") + Expect(restore(walFile(5))).To(Succeed()) + Eventually(func(g Gomega) { + g.Expect(existsIn(pgWalPath, walFile(5))).To(BeTrue(), "#5 in pg_wal") + g.Expect(spoolSegments()).To(Equal(0), "no WAL segments in spool") + g.Expect(flagSet()).To(BeTrue(), "end-of-wal-stream still set") + }).WithTimeout(time.Minute).WithPolling(2 * time.Second).Should(Succeed()) + + // #6 (first): flag is set, so the request fails fast (exit 1) and the + // flag is consumed, leaving an empty spool. + By("requesting WAL #6: fails fast on the end-of-wal-stream flag, spool cleared") + restoreErr := restore(walFile(6)) + var exitErr executil.CodeExitError + Expect(errors.As(restoreErr, &exitErr)).To(BeTrue(), + "expected a CodeExitError, got %T: %v", restoreErr, restoreErr) + Expect(exitErr.ExitStatus()).To(Equal(1), "exit code should be 1") + Eventually(func(g Gomega) { + g.Expect(existsIn(pgWalPath, walFile(6))).To(BeFalse(), "#6 not restored") + g.Expect(spoolSegments()).To(Equal(0), "no WAL segments in spool") + g.Expect(flagSet()).To(BeFalse(), "end-of-wal-stream consumed") + }).WithTimeout(time.Minute).WithPolling(2 * time.Second).Should(Succeed()) + + // #6 (second): now present, restored; #7 and #8 absent so the flag is + // set again. + By("requesting WAL #6 again: #6 restored, end-of-wal-stream set again") + Expect(restore(walFile(6))).To(Succeed()) + Eventually(func(g Gomega) { + g.Expect(existsIn(pgWalPath, walFile(6))).To(BeTrue(), "#6 in pg_wal") + g.Expect(spoolSegments()).To(Equal(0), "no WAL segments in spool") + g.Expect(flagSet()).To(BeTrue(), "end-of-wal-stream set (#7/#8 absent)") + }).WithTimeout(time.Minute).WithPolling(2 * time.Second).Should(Succeed()) + }) +})