Merge 8c2cf048ec into 806f47b483

Neither task depends on the other's output: task ci runs lint,
tests, e2e, and docs, while task publish builds and pushes the
multi-arch images. Running them as sequential steps in one job
only adds their durations. Splitting publish into its own job
lets it run alongside ci instead, and lets each job carry only
the permissions it actually needs: ci never writes to the
registry, and neither task touches repository contents, so only
packages: write on the publish job remains.

QEMU is only needed for publish, since it's the only job that
builds non-native (arm64) platforms; ci's e2e-ephemeral image
build is amd64-only. The disk cleanup steps aren't needed for
publish either: it only builds two lean multi-arch images, well
within the free disk space available even on the default
ubuntu-latest runner.

Note that publishing testing images no longer waits on ci
passing, since gating it away would remove the parallelism this
is meant to gain.

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

.github/workflows/ci.yml

@@ -6,10 +6,6 @@ on:
 
 permissions: read-all
 
-concurrency:
-  group: ${{ github.workflow }}-${{ github.ref }}
-  cancel-in-progress: ${{ github.event_name == 'pull_request' }}
-
 jobs:
   ci:
     runs-on: ${{ vars.CI_RUNNERS || 'ubuntu-latest' }}

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: f7187c307f2ba79290efbfff5ce436b956162c53
+      DAGGER_SPELLCHECK_SHA: 7431c24907de3c6b1516114c8990d4f8d23d92cd
     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: f7187c307f2ba79290efbfff5ce436b956162c53
+      DAGGER_COMMITLINT_SHA: 7431c24907de3c6b1516114c8990d4f8d23d92cd
     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: f7187c307f2ba79290efbfff5ce436b956162c53
+      DAGGER_UNCOMMITTED_SHA: 7431c24907de3c6b1516114c8990d4f8d23d92cd
     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: f7187c307f2ba79290efbfff5ce436b956162c53
+      DAGGER_CRDGENREF_SHA: 7431c24907de3c6b1516114c8990d4f8d23d92cd
       # renovate: datasource=go depName=github.com/elastic/crd-ref-docs
       CRDREFDOCS_VERSION: v0.3.0
     cmds:
@@ -383,7 +383,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: f7187c307f2ba79290efbfff5ce436b956162c53
+      DAGGER_CONTROLLER_GEN_SHA: 7431c24907de3c6b1516114c8990d4f8d23d92cd
     cmds:
       - >
         GITHUB_REF= dagger -s call -m github.com/cloudnative-pg/daggerverse/controller-gen@${DAGGER_CONTROLLER_GEN_SHA}

internal/cnpgi/common/wal.go

@@ -27,7 +27,6 @@ import (
 	"path"
 	"time"
 
-	barmanapi "github.com/cloudnative-pg/barman-cloud/pkg/api"
 	"github.com/cloudnative-pg/barman-cloud/pkg/archiver"
 	barmanCommand "github.com/cloudnative-pg/barman-cloud/pkg/command"
 	barmanCredentials "github.com/cloudnative-pg/barman-cloud/pkg/credentials"
@@ -250,11 +249,9 @@ func (w WALServiceImplementation) Restore(
 		"Restoring WAL file",
 		"objectStore", objectStore.Name,
 		"serverName", serverName,
-		"walName", walName,
-		"mode", request.GetMode())
+		"walName", walName)
 	return &wal.WALRestoreResult{}, w.restoreFromBarmanObjectStore(
-		ctx, configuration.Cluster, &objectStore, serverName, walName, destinationPath,
-		request.GetMode() == wal.WALRestoreRequest_MODE_REWIND)
+		ctx, configuration.Cluster, &objectStore, serverName, walName, destinationPath)
 }
 
 // resolveRestoreObjectStore selects the object store and server name to use when
@@ -291,7 +288,6 @@ func (w WALServiceImplementation) restoreFromBarmanObjectStore(
 	serverName string,
 	walName string,
 	destinationPath string,
-	rewindMode bool,
 ) error {
 	contextLogger := log.FromContext(ctx)
 	startTime := time.Now()
@@ -323,20 +319,6 @@ func (w WALServiceImplementation) restoreFromBarmanObjectStore(
 		return fmt.Errorf("while creating the restorer: %w", err)
 	}
 
-	// A flag left over from a normal-recovery invocation that ran before this pod
-	// was demoted must not survive into a pg_rewind restore: the flag machinery
-	// does not apply while restoring on behalf of pg_rewind (see
-	// shouldUseEndOfWALStreamFlag below), so it is never checked here, but left
-	// untouched it would resurface and wrongly abort the first normal-recovery
-	// invocation that runs once the rewind is done. This runs unconditionally,
-	// before the spool short-circuit in Step 1, so a request for a WAL file that
-	// happens to already be staged in the spool cannot skip the clear.
-	if rewindMode {
-		if err := clearEndOfWALStreamFlag(walRestorer); err != nil {
-			return err
-		}
-	}
-
 	// Step 1: check if this WAL file is not already in the spool
 	var wasInSpool bool
 	if wasInSpool, err = walRestorer.RestoreFromSpool(walName, destinationPath); err != nil {
@@ -349,10 +331,8 @@ func (w WALServiceImplementation) restoreFromBarmanObjectStore(
 		return nil
 	}
 
-	// Step 2: return error if the end-of-wal-stream flag is set.
-	// We skip this step if the flag machinery does not apply to this invocation
-	useEndOfWALStreamFlag := shouldUseEndOfWALStreamFlag(cluster, w.InstanceName, rewindMode)
-	if useEndOfWALStreamFlag {
+	// We skip this step if streaming connection is not available
+	if isStreamingAvailable(cluster, w.InstanceName) {
 		if err := checkEndOfWALStreamFlag(walRestorer); err != nil {
 			return err
 		}
@@ -360,7 +340,10 @@ func (w WALServiceImplementation) restoreFromBarmanObjectStore(
 
 	// Step 3: gather the WAL files names to restore. If the required file isn't a regular WAL, we download it directly.
 	var walFilesList []string
-	maxParallel := maxWALFilesPerInvocation(barmanConfiguration, rewindMode)
+	maxParallel := 1
+	if barmanConfiguration.Wal != nil && barmanConfiguration.Wal.MaxParallel > 1 {
+		maxParallel = barmanConfiguration.Wal.MaxParallel
+	}
 	if IsWALFile(walName) {
 		// If this is a regular WAL file, we try to prefetch
 		if walFilesList, err = gatherWALFilesToRestore(walName, maxParallel); err != nil {
@@ -382,9 +365,9 @@ func (w WALServiceImplementation) restoreFromBarmanObjectStore(
 		return classifyWALRestoreError(walStatus[0].WalName, walStatus[0].Err)
 	}
 
-	// We skip this step if the flag machinery does not apply to this invocation
+	// We skip this step if streaming connection is not available
 	endOfWALStream := isEndOfWALStream(walStatus)
-	if useEndOfWALStreamFlag && endOfWALStream {
+	if isStreamingAvailable(cluster, w.InstanceName) && endOfWALStream {
 		contextLogger.Info(
 			"Set end-of-wal-stream flag as one of the WAL files to be prefetched was not found")
 
@@ -432,38 +415,6 @@ func (w WALServiceImplementation) SetFirstRequired(
 	panic("implement me")
 }
 
-// maxWALFilesPerInvocation returns how many WAL files a single restore
-// invocation is allowed to fetch, the requested one included. Prefetching is
-// disabled when restoring on behalf of pg_rewind, which walks the WAL
-// backward: the following segments would never be requested, and past the end
-// of the timeline they do not even exist
-func maxWALFilesPerInvocation(barmanConfiguration *barmanapi.BarmanObjectStoreConfiguration, rewindMode bool) int {
-	if rewindMode {
-		return 1
-	}
-
-	if barmanConfiguration.Wal != nil && barmanConfiguration.Wal.MaxParallel > 1 {
-		return barmanConfiguration.Wal.MaxParallel
-	}
-
-	return 1
-}
-
-// shouldUseEndOfWALStreamFlag returns true when the end-of-wal-stream flag
-// machinery applies to the current invocation. The flag makes the following
-// invocation fail, so that PostgreSQL stops polling the WAL archive and
-// switches to streaming replication. It does not apply when no streaming
-// connection is available, nor when restoring on behalf of pg_rewind:
-// pg_rewind cannot fall back to streaming replication, and a stale flag would
-// make it abort on a segment that is available in the archive
-func shouldUseEndOfWALStreamFlag(cluster *cnpgv1.Cluster, podName string, rewindMode bool) bool {
-	if rewindMode {
-		return false
-	}
-
-	return isStreamingAvailable(cluster, podName)
-}
-
 // isStreamingAvailable checks if this pod can replicate via streaming connection.
 func isStreamingAvailable(cluster *cnpgv1.Cluster, podName string) bool {
 	if cluster == nil {
@@ -540,22 +491,6 @@ func checkEndOfWALStreamFlag(walRestorer *barmanRestorer.WALRestorer) error {
 	return nil
 }
 
-// clearEndOfWALStreamFlag removes the end-of-wal-stream flag, if present, without
-// treating it as an error. It is used instead of checkEndOfWALStreamFlag when
-// restoring on behalf of pg_rewind, which must not abort on a flag it did not
-// set itself.
-func clearEndOfWALStreamFlag(walRestorer *barmanRestorer.WALRestorer) error {
-	contain, err := walRestorer.IsEndOfWALStream()
-	if err != nil {
-		return err
-	}
-
-	if contain {
-		return walRestorer.ResetEndOfWalStream()
-	}
-	return nil
-}
-
 // isEndOfWALStream returns true if one of the downloads has returned
 // a file-not-found error.
 func isEndOfWALStream(results []barmanRestorer.Result) bool {

internal/cnpgi/common/wal_test.go

@@ -20,10 +20,6 @@ SPDX-License-Identifier: Apache-2.0
 package common
 
 import (
-	"context"
-
-	barmanapi "github.com/cloudnative-pg/barman-cloud/pkg/api"
-	barmanRestorer "github.com/cloudnative-pg/barman-cloud/pkg/restorer"
 	cnpgv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
@@ -100,80 +96,3 @@ var _ = Describe("resolveRestoreObjectStore", func() {
 			"cluster-server", "cluster-store"),
 	)
 })
-
-var _ = Describe("maxWALFilesPerInvocation", func() {
-	configWithMaxParallel := func(maxParallel int) *barmanapi.BarmanObjectStoreConfiguration {
-		return &barmanapi.BarmanObjectStoreConfiguration{
-			Wal: &barmanapi.WalBackupConfiguration{MaxParallel: maxParallel},
-		}
-	}
-
-	DescribeTable(
-		"computes how many WAL files a single invocation may fetch",
-		func(cfg *barmanapi.BarmanObjectStoreConfiguration, rewindMode bool, want int) {
-			Expect(maxWALFilesPerInvocation(cfg, rewindMode)).To(Equal(want))
-		},
-
-		Entry("no WAL configuration", &barmanapi.BarmanObjectStoreConfiguration{}, false, 1),
-		Entry("parallel restore configured", configWithMaxParallel(8), false, 8),
-
-		// pg_rewind walks the timeline backwards: prefetching must stay off no
-		// matter what the object store configuration asks for
-		Entry("rewind mode overrides the configured parallelism", configWithMaxParallel(8), true, 1),
-	)
-})
-
-var _ = Describe("shouldUseEndOfWALStreamFlag", func() {
-	clusterWithPrimary := func(currentPrimary string) *cnpgv1.Cluster {
-		return &cnpgv1.Cluster{
-			Status: cnpgv1.ClusterStatus{CurrentPrimary: currentPrimary},
-		}
-	}
-
-	DescribeTable(
-		"decides whether the end-of-wal-stream flag machinery applies",
-		func(cluster *cnpgv1.Cluster, podName string, rewindMode bool, want bool) {
-			Expect(shouldUseEndOfWALStreamFlag(cluster, podName, rewindMode)).To(Equal(want))
-		},
-
-		Entry("replica with streaming available", clusterWithPrimary("cluster-1"), "cluster-2", false, true),
-		Entry("primary cannot stream from anyone", clusterWithPrimary("cluster-1"), "cluster-1", false, false),
-
-		// pg_rewind cannot fall back to streaming replication: the flag machinery
-		// must stay off even where a standby would use it
-		Entry("rewind mode wins over streaming availability", clusterWithPrimary("cluster-1"), "cluster-2", true, false),
-	)
-})
-
-var _ = Describe("clearEndOfWALStreamFlag", func() {
-	newRestorer := func() *barmanRestorer.WALRestorer {
-		restorer, err := barmanRestorer.New(context.Background(), nil, GinkgoT().TempDir())
-		Expect(err).ToNot(HaveOccurred())
-		return restorer
-	}
-
-	It("is a no-op when the flag is not set", func() {
-		restorer := newRestorer()
-
-		Expect(clearEndOfWALStreamFlag(restorer)).To(Succeed())
-
-		isEOS, err := restorer.IsEndOfWALStream()
-		Expect(err).ToNot(HaveOccurred())
-		Expect(isEOS).To(BeFalse())
-	})
-
-	// Regression guard: a flag left over from a normal-recovery invocation that
-	// ran before this pod was demoted must not survive a pg_rewind restore, or
-	// it would resurface and wrongly abort the first normal-recovery invocation
-	// that runs once the rewind is done.
-	It("removes a pre-existing flag without returning an error", func() {
-		restorer := newRestorer()
-		Expect(restorer.SetEndOfWALStream()).To(Succeed())
-
-		Expect(clearEndOfWALStreamFlag(restorer)).To(Succeed())
-
-		isEOS, err := restorer.IsEndOfWALStream()
-		Expect(err).ToNot(HaveOccurred())
-		Expect(isEOS).To(BeFalse())
-	})
-})

test/e2e/internal/tests/walrestore/fixtures.go

@@ -121,8 +121,8 @@ func newS3ClientDeployment(namespace string) *appsv1.Deployment {
 						{
 							Name: s3ClientName,
 							// renovate: datasource=docker depName=amazon/aws-cli versioning=docker
-							// Version: 2.36.3
-							Image:   "docker.io/amazon/aws-cli@sha256:bdd02067a00c354684086071b475955c54caa7bd88b851aac99a51326fe19652",
+							// Version: 2.36.2
+							Image:   "docker.io/amazon/aws-cli@sha256:964336bffb17b82d2e84a2526b0672e70a2c881544e1a281acaca1d9aa41b536",
 							Command: []string{"sleep", "infinity"},
 							Env: []corev1.EnvVar{
 								{