test: tighten WAL restore assertions and fix a misleading comment

Assert the actual exit code on the end-of-wal-stream fast-fail path
instead of just HaveOccurred(), which would pass for any error. Also
correct a comment claiming ExecuteInContainer folds command output
into the returned error; it only carries the exit code, so forging a
WAL segment fails with no indication of the underlying aws CLI error.
Drop an unnecessary var declaration and note the real bound on the
forged segment numbering.

Signed-off-by: Marco Nenciarini <marco.nenciarini@enterprisedb.com>

Taskfile.yml

@@ -46,7 +46,7 @@ tasks:
       - wordlist-ordered
     env:
       # renovate: datasource=git-refs depName=spellcheck lookupName=https://github.com/cloudnative-pg/daggerverse currentValue=main
-      DAGGER_SPELLCHECK_SHA: 679d5056358d3cd838ec226333decff673602937
+      DAGGER_SPELLCHECK_SHA: 7a53f0b7f36c786389a6f72714d5875b9a63e9c1
     cmds:
       - >
         GITHUB_REF= dagger -s call -m github.com/cloudnative-pg/daggerverse/spellcheck@${DAGGER_SPELLCHECK_SHA}
@@ -60,7 +60,7 @@ tasks:
     desc: Check for conventional commits
     env:
       # renovate: datasource=git-refs depName=commitlint lookupName=https://github.com/cloudnative-pg/daggerverse currentValue=main
-      DAGGER_COMMITLINT_SHA: 679d5056358d3cd838ec226333decff673602937
+      DAGGER_COMMITLINT_SHA: 7a53f0b7f36c786389a6f72714d5875b9a63e9c1
     cmds:
       - >
         GITHUB_REF= dagger -s call -m github.com/cloudnative-pg/daggerverse/commitlint@${DAGGER_COMMITLINT_SHA}
@@ -74,7 +74,7 @@ tasks:
       - wordlist-ordered
     env:
       # renovate: datasource=git-refs depName=uncommitted lookupName=https://github.com/cloudnative-pg/daggerverse currentValue=main
-      DAGGER_UNCOMMITTED_SHA: 679d5056358d3cd838ec226333decff673602937
+      DAGGER_UNCOMMITTED_SHA: 7a53f0b7f36c786389a6f72714d5875b9a63e9c1
     cmds:
       - GITHUB_REF= dagger -s call -m github.com/cloudnative-pg/daggerverse/uncommitted@${DAGGER_UNCOMMITTED_SHA} check-uncommitted --source . stdout
     sources:
@@ -86,7 +86,7 @@ tasks:
       - controller-gen
     env:
       # renovate: datasource=git-refs depName=crd-gen-refs lookupName=https://github.com/cloudnative-pg/daggerverse currentValue=main
-      DAGGER_CRDGENREF_SHA: 679d5056358d3cd838ec226333decff673602937
+      DAGGER_CRDGENREF_SHA: 7a53f0b7f36c786389a6f72714d5875b9a63e9c1
       # renovate: datasource=go depName=github.com/elastic/crd-ref-docs
       CRDREFDOCS_VERSION: v0.3.0
     cmds:
@@ -381,7 +381,7 @@ tasks:
     run: once
     env:
       # renovate: datasource=git-refs depName=controller-gen lookupName=https://github.com/cloudnative-pg/daggerverse currentValue=main
-      DAGGER_CONTROLLER_GEN_SHA: 679d5056358d3cd838ec226333decff673602937
+      DAGGER_CONTROLLER_GEN_SHA: 7a53f0b7f36c786389a6f72714d5875b9a63e9c1
     cmds:
       - >
         GITHUB_REF= dagger -s call -m github.com/cloudnative-pg/daggerverse/controller-gen@${DAGGER_CONTROLLER_GEN_SHA}

containers/Dockerfile.sidecar

@@ -55,7 +55,7 @@ COPY containers/sidecar-requirements.txt .
 
 # Create virtualenv and install dependencies
 RUN python3 -m venv /venv && \
-    /venv/bin/pip install --upgrade pip setuptools wheel && \
+    /venv/bin/pip install --upgrade pip wheel && \
     /venv/bin/pip install --no-cache-dir -r sidecar-requirements.txt
 
 # Download and extract runtime library packages and their dependencies

containers/sidecar-requirements.in

@@ -1,3 +1,2 @@
 barman[azure,cloud,google,snappy,zstandard,lz4]==3.19.1
-setuptools==82.0.1
 zipp>=3.19.1 # not directly required, pinned by Snyk to avoid a vulnerability

containers/sidecar-requirements.txt

@@ -22,19 +22,19 @@ barman==3.19.1 \
     --hash=sha256:0a6a9e1babf97687732d8b2a3eb79ea95d55246a5257b9433865cb6e755221c0 \
     --hash=sha256:2f71c4a1f1ba53f694cbdf838bb9906d8ba02b97d1fd3041196e8999bec7a1ee
     # via -r sidecar-requirements.in
-boto3==1.43.30 \
+boto3==1.43.36 \
-    --hash=sha256:6b1ee360f363a457f67a8f5702f522043d8a32d67a97c362ad12075d8b5b531e \
+    --hash=sha256:42942dde254673abcbc9e6e60017c88341a4f49d99d24e1f2e290fb38138c26f \
-    --hash=sha256:89e982463d94773136ccf69be77cccd54ff1ce351a6aadd1d3437fcb693681b5
+    --hash=sha256:587d7ee92a12e440ad12b0e7f11f3358f0c4d65b19f64726efc94aaf194aff28
     # via barman
-botocore==1.43.30 \
+botocore==1.43.36 \
-    --hash=sha256:19ed560cb35ae43bf010d37da429a553c07063bf7efea0f2cb53be8a78d3e3d5 \
+    --hash=sha256:3c65fdc39ed01d8dfde1e961b34038aed03c459f8ddf80717a12ac006475e49d \
-    --hash=sha256:26b1dded84d89b396180916f56900bd2ab1c0d545a66d1d2c3eeb40f772935b2
+    --hash=sha256:4cae47d1b2d426316b85a0087d9e69e048f13bc003b5177d74639fe9dfd28205
     # via
     #   boto3
     #   s3transfer
-certifi==2026.5.20 \
+certifi==2026.6.17 \
-    --hash=sha256:3c52e209ba0a4ad7aebe60436a4ab349c39e1e602e8c134221e546902ad25897 \
+    --hash=sha256:024c88eeec92ca068db80f02b8b07c9cef7b9fe261d1d535abfd5abd6f6af432 \
-    --hash=sha256:69dea482ab64caa7b9f6aba1c6bf48bb6a5448d1c0f1b17ab42ad8c763a5344d
+    --hash=sha256:2227dcbaafe0d2f59279d1762ddddc37783ed4354594f194ffc31d20f41fc3db
     # via requests
 cffi==2.0.0 \
     --hash=sha256:00bdf7acc5f795150faa6957054fbbca2439db2f775ce831222b66f192f03beb \
@@ -654,9 +654,9 @@ requests==2.34.2 \
     #   google-api-core
     #   google-cloud-storage
     #   msal
-s3transfer==0.18.0 \
+s3transfer==0.19.0 \
-    --hash=sha256:239c13b09e65ad0346e1be7348b8a202dcad44ac7ea7c6eb858fc881dce739b6 \
+    --hash=sha256:777cc2415536f1debadb5c2ef7779275d0fc0fe0e042411cdd6caebeb2685262 \
-    --hash=sha256:3760b8b7ec1315da54048b2d626276732bee4300d054d492d4e1d43e20d4ecbd
+    --hash=sha256:ce436931687addc4c1712d52d40b32f53e88315723f107ffa20ba82b05a0f685
     # via boto3
 six==1.17.0 \
     --hash=sha256:4721f391ed90541fddacab5acf947aa0d3dc7d27b2e1e8eda2be8970586c3274 \
@@ -780,9 +780,3 @@ zstandard==0.25.0 \
     --hash=sha256:fd7a5004eb1980d3cefe26b2685bcb0b17989901a70a1040d1ac86f1d898c551 \
     --hash=sha256:ffef5a74088f1e09947aecf91011136665152e0b4b359c42be3373897fb39b01
     # via barman
-
-# The following packages are considered to be unsafe in a requirements file:
-setuptools==82.0.1 \
-    --hash=sha256:7d872682c5d01cfde07da7bccc7b65469d3dca203318515ada1de5eda35efbf9 \
-    --hash=sha256:a59e362652f08dcd477c78bb6e7bd9d80a7995bc73ce773050228a348ce2e5bb
-    # via -r sidecar-requirements.in

internal/cnpgi/operator/lifecycle.go

@@ -403,8 +403,9 @@ func reconcilePodSpec(
 	envs = append(envs, config.env...)
 
 	baseProbe := &corev1.Probe{
-		FailureThreshold: 10,
+		PeriodSeconds:    1,
-		TimeoutSeconds:   10,
+		FailureThreshold: 30,
+		TimeoutSeconds:   5,
 		ProbeHandler: corev1.ProbeHandler{
 			Exec: &corev1.ExecAction{
 				Command: []string{"/manager", "healthcheck", "unix"},

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: {

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"

test/e2e/internal/certmanager/certmanager.go

@@ -58,7 +58,7 @@ func WithIgnoreExistingResources(ignore bool) InstallOption {
 // DefaultVersion is the default version of cert-manager to install.
 //
 // renovate: datasource=github-releases depName=cert-manager/cert-manager
-const DefaultVersion = "v1.20.2"
+const DefaultVersion = "v1.20.3"
 
 // Install installs cert-manager using kubectl.
 func Install(ctx context.Context, cl client.Client, opts ...InstallOption) error {

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

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,
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+}

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())
+		})
+})