Merge branch 'main' into fix/backup-name-validation

.github/copilot-instructions.md

@@ -60,6 +60,19 @@ branches.
   space-delimited: `/backport release-1.17 release-1.18`.  The label causes the
   backport to run automatically when the PR merges.
 - **After merge:** the same comment immediately creates the backport PR.
+- **Shorthand:** a bare version like `/backport 1.17` is automatically expanded
+  to `release-1.17`; this works generically for any `X.Y` version.
+- **Changelog filename:** the cherry-picked commit(s) carry over the source
+  PR's `changelogs/unreleased/<source_pr>-<user>` file. The workflow
+  automatically renames it to `<backport_pr>-<user>` on the backport branch so
+  `hack/changelog-check.sh` passes and release notes cite the correct PR.
+- **Changelog-not-required:** if the source PR is labeled
+  `kind/changelog-not-required`, that label is copied to the backport PR so
+  it isn't flagged as missing a changelog.
+- **DCO signoff:** every commit on a backport branch is re-signed with the
+  bot's `Signed-off-by` trailer (`git rebase --signoff`), including
+  cherry-picked commits from the original author, so the DCO check always
+  passes on backport PRs.
 - Only repository **owners, members, and collaborators** may trigger these commands.
 
 ## General coding guidelines

.github/workflows/backport.yml

@@ -16,6 +16,26 @@ name: Backport merged pull request
 # In both cases multiple target branches can be space-delimited in a comment:
 #   /backport release-1.17 release-1.18
 #
+# As a shorthand, a bare release version (e.g. `1.17`) is automatically
+# expanded to the corresponding `release-1.17` branch, so `/backport 1.17`
+# and `/backport release-1.17` are equivalent. This works generically for
+# any `X.Y` version, e.g. `/backport 1.18 1.19`.
+#
+# The cherry-picked commit(s) carry over the original PR's changelog file
+# (changelogs/unreleased/<source_pr>-<user>), which no longer matches the
+# backport PR's own number. After the backport PR is created, its changelog
+# file is automatically renamed to <backport_pr>-<user> so that
+# hack/changelog-check.sh passes and release notes cite the correct PR.
+#
+# If the source PR is labeled `kind/changelog-not-required` (i.e. it has no
+# changelog file), that label is copied to the backport PR so it isn't
+# flagged as missing a changelog either.
+#
+# Every commit on a backport branch (the cherry-picked commit(s), even from
+# the original author, plus the changelog rename commit) is re-signed with
+# the bot's Signed-off-by trailer via `git rebase --signoff`, so the DCO
+# check always passes regardless of whether the original commit had one.
+#
 # See: https://github.com/velero-io/velero/issues/9603
 
 on:
@@ -89,6 +109,10 @@ jobs:
           fi
 
           for branch in $branches; do
+            # Shorthand: a bare version like "1.17" expands to "release-1.17".
+            if [[ "$branch" =~ ^[0-9]+\.[0-9]+$ ]]; then
+              branch="release-${branch}"
+            fi
             label="backport ${branch}"
             echo "Applying label: '${label}'"
             # Create the label if it does not exist yet (idempotent).
@@ -141,18 +165,31 @@ jobs:
           # (may be empty, falls back to labels).
           line=$(printf '%s' "$COMMENT_BODY" | head -n1 | tr -d '\r')
           branches=$(printf '%s' "$line" | sed -E 's#^/(backport|cherrypick)[[:space:]]*##')
-          echo "branches=${branches}" >> "$GITHUB_OUTPUT"
+
+          normalized=""
+          for branch in $branches; do
+            # Shorthand: a bare version like "1.17" expands to "release-1.17".
+            if [[ "$branch" =~ ^[0-9]+\.[0-9]+$ ]]; then
+              branch="release-${branch}"
+            fi
+            normalized="${normalized}${normalized:+ }${branch}"
+          done
+          echo "branches=${normalized}" >> "$GITHUB_OUTPUT"
 
       - uses: actions/checkout@v7
         with:
           fetch-depth: 0
 
       - name: Create backport pull requests
+        id: backport
         # Pin to commit SHA: workflow has contents/pull-requests write.
         uses: korthout/backport-action@2e830a1d0b8269505846ddd407a70876913ad1f8 # v4.6.0
         with:
           # Labels like `backport release-1.17` select the target branch.
           label_pattern: '^backport ([^ ]+)$'
+          # Carry over `kind/changelog-not-required` from the source PR so
+          # backport PRs of changelog-exempt changes aren't flagged as missing one.
+          copy_labels_pattern: '^kind/changelog-not-required$'
           # Prefer draft PRs with conflict markers over failing the job silently.
           experimental: |
             {
@@ -161,3 +198,63 @@ jobs:
           # Empty when triggered by merge labels; set when `/backport` or `/cherrypick` includes branches.
           target_branches: ${{ steps.parse.outputs.branches }}
           github_token: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Rename changelog file(s) and ensure DCO signoff
+        # The cherry-picked commit(s) still carry the source PR's changelog
+        # filename (e.g. changelogs/unreleased/9795-kaovilai), which no
+        # longer matches the new backport PR's number. Rename it on each
+        # created backport branch so hack/changelog-check.sh passes and the
+        # release notes cite the correct PR.
+        #
+        # Also ensure every commit on the backport branch passes the DCO
+        # check by re-signing it with the bot's Signed-off-by trailer via
+        # `git rebase --signoff`. This covers the cherry-picked commits
+        # (even when the original author's commit had no trailer) as well
+        # as the changelog rename commit added above; it preserves any
+        # existing Signed-off-by trailers rather than replacing them.
+        if: steps.backport.outputs.created_pull_numbers != ''
+        env:
+          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          REPO: ${{ github.repository }}
+          SOURCE_PR_NUMBER: ${{ github.event.pull_request.number || github.event.issue.number }}
+          CREATED_PR_NUMBERS: ${{ steps.backport.outputs.created_pull_numbers }}
+        run: |
+          git config user.name "github-actions[bot]"
+          git config user.email "github-actions[bot]@users.noreply.github.com"
+
+          shopt -s nullglob
+          for new_pr in $CREATED_PR_NUMBERS; do
+            if [ "$new_pr" = "$SOURCE_PR_NUMBER" ]; then
+              continue
+            fi
+
+            branch=$(gh pr view "$new_pr" --repo "$REPO" --json headRefName -q .headRefName)
+            base_branch=$(gh pr view "$new_pr" --repo "$REPO" --json baseRefName -q .baseRefName)
+            git fetch origin "$branch" "$base_branch"
+            git checkout -B "$branch" "origin/${branch}"
+
+            files=(changelogs/unreleased/"${SOURCE_PR_NUMBER}"-*)
+            if [ ${#files[@]} -gt 0 ]; then
+              for old_file in "${files[@]}"; do
+                suffix=$(basename "$old_file" | sed -E "s/^${SOURCE_PR_NUMBER}-//")
+                new_file="changelogs/unreleased/${new_pr}-${suffix}"
+                if [ "$old_file" != "$new_file" ]; then
+                  git mv "$old_file" "$new_file"
+                fi
+              done
+              if ! git diff --cached --quiet; then
+                git commit -m "Rename changelog to match backport PR #${new_pr}"
+              fi
+            else
+              echo "No changelog file for PR ${SOURCE_PR_NUMBER} found on ${branch}; skipping rename."
+            fi
+
+            # Add the bot's Signed-off-by trailer to every commit ahead of
+            # the target branch (cherry-picked commits + the rename commit).
+            if ! git rebase --signoff "origin/${base_branch}"; then
+              echo "::error::git rebase --signoff failed for PR #${new_pr} on branch ${branch}; aborting rebase, branch left unchanged." >&2
+              git rebase --abort
+              exit 1
+            fi
+            git push --force-with-lease origin "HEAD:${branch}"
+          done

.github/workflows/e2e-test-kind.yaml

@@ -94,12 +94,14 @@ jobs:
         id: set-matrix
         # everything excluding older tags. limits needs to be high enough to cover all latest versions
         # and test labels
-        # grep -E "v[1-9]\.(2[5-9]|[3-9][0-9])" filters for v1.25 to v9.99
+        # grep -E "^v[1-9]\.(2[5-9]|[3-9][0-9])\.[0-9]+$" filters for well-formed v1.25.x to v9.99.x
+        # GA releases only, so a pre-release tag like v1.37.0-rc.1 can't reach the
+        # awk step below and be misparsed as a patch release (e.g. "1.37.1")
         # and removes older patches of the same minor version
         # awk -F. '{if(!a[$1"."$2]++)print $1"."$2"."$NF}'
         run: |
           echo "matrix={\
-            \"k8s\":$(wget -q -O - "https://hub.docker.com/v2/namespaces/kindest/repositories/node/tags?page_size=50" | grep -o '"name": *"[^"]*' | grep -o '[^"]*$' | grep -v -E "alpha|beta" | grep -E "v[1-9]\.(2[5-9]|[3-9][0-9])" | awk -F. '{if(!a[$1"."$2]++)print $1"."$2"."$NF}' | sort -r | sed s/v//g | jq -R -c -s 'split("\n")[:-1]'),\
+            \"k8s\":$(wget -q -O - "https://hub.docker.com/v2/namespaces/kindest/repositories/node/tags?page_size=50" | grep -o '"name": *"[^"]*' | grep -o '[^"]*$' | grep -E "^v[1-9]\.(2[5-9]|[3-9][0-9])\.[0-9]+$" | awk -F. '{if(!a[$1"."$2]++)print $1"."$2"."$NF}' | sort -r | sed s/v//g | jq -R -c -s 'split("\n")[:-1]'),\
             \"labels\":[\
               \"Basic && (ClusterResource || NodePort || StorageClass)\", \
               \"ResourceFiltering && !FSBackup\", \
@@ -140,7 +142,7 @@ jobs:
       - name: Install MinIO
         run: |
           docker run -d --rm -p 9000:9000 -e "MINIO_ROOT_USER=minio" -e "MINIO_ROOT_PASSWORD=minio123" -e "MINIO_DEFAULT_BUCKETS=bucket,additional-bucket" bitnami/minio:local
-      - uses: helm/kind-action@v1
+      - uses: helm/kind-action@7a97ed793754775518f9db3a8151ee7461dc9c31 # v1 + fix: add curl retry flags (https://github.com/helm/kind-action/pull/165)
         with:
           cluster_name: "kind"
           version: "v0.32.0"

.github/workflows/nightly-trivy-scan.yml

@@ -31,6 +31,6 @@ jobs:
           output: 'trivy-results.sarif'
 
       - name: Upload Trivy scan results to GitHub Security tab
-        uses: github/codeql-action/upload-sarif@v4.37.6
+        uses: github/codeql-action/upload-sarif@v4.37.7
         with:
           sarif_file: 'trivy-results.sarif'

.github/workflows/pr-codespell.yml

@@ -14,8 +14,7 @@ jobs:
     - name: Codespell
       uses: codespell-project/actions-codespell@master
       with:
-        # ignore the config/.../crd.go file as it's generated binary data that is edited elsewhere.
-        skip: .git,*.png,*.jpg,*.woff,*.ttf,*.gif,*.ico,./config/crd/v1beta1/crds/crds.go,./config/crd/v1/crds/crds.go,./config/crd/v2alpha1/crds/crds.go,./go.sum,./LICENSE
+        skip: .git,*.png,*.jpg,*.woff,*.ttf,*.gif,*.ico,./go.sum,./LICENSE
         ignore_words_list: iam,aks,ist,bridget,ue,shouldnot,atleast,notin,sme,optin,sie
         check_filenames: true
         check_hidden: true

README.md

@@ -1,7 +1,9 @@
 ![100]
 
 [![Build Status][1]][2] [![CII Best Practices](https://bestpractices.coreinfrastructure.org/projects/3811/badge)](https://bestpractices.coreinfrastructure.org/projects/3811)
-![GitHub release (latest SemVer)](https://img.shields.io/github/v/release/velero-io/velero)
+[![GitHub release (latest SemVer)](https://img.shields.io/github/v/release/velero-io/velero)](https://github.com/velero-io/velero/releases)
+[![GitHub stars](https://img.shields.io/github/stars/velero-io/velero)](https://github.com/velero-io/velero/stargazers)
+[![Docker Pulls](https://img.shields.io/docker/pulls/velero/velero.svg)](https://hub.docker.com/r/velero/velero)
 
 ## Overview
 

changelogs/unreleased/10185-Lyndon-Li

@@ -0,0 +1 @@
+Support full backup for file system data mover and pod volume backup

changelogs/unreleased/10229-Ralthos

@@ -0,0 +1 @@
+Add printer columns for DownloadRequest and ServerStatusRequest so kubectl shows their target, status and server version

changelogs/unreleased/10242-beep-boopp

@@ -0,0 +1 @@
+Fix schedule reconciler aliasing &c.skipImmediately into Schedule specs, corrupting the server-wide --schedule-skip-immediately default after the first reconcile

changelogs/unreleased/10245-Ralthos

@@ -0,0 +1 @@
+Document that the DownloadRequest Processed phase means a URL has been signed, and that it does not imply the target object exists

changelogs/unreleased/10247-opbot-xd

@@ -0,0 +1 @@
+Refactor: Replace context.TODO() with properly plumbed contexts in CSI backup actions and utility functions to enable proper cancellation of in-flight API requests.

changelogs/unreleased/10252-Ralthos

@@ -0,0 +1 @@
+Skip signing a download URL when no artifacts can exist yet, and set a Failed phase with the reason so callers stop waiting

changelogs/unreleased/10254-Lyndon-Li

@@ -0,0 +1 @@
+Ignore credentialFile filled into BSL by users to avoid unexpected credentials used by Velero

changelogs/unreleased/10255-Lyndon-Li

@@ -0,0 +1 @@
+Use thread safe map for cancel recorder

changelogs/unreleased/10256-Lyndon-Li

@@ -0,0 +1 @@
+Clarify the security context to Velero server

changelogs/unreleased/10258-Lyndon-Li

@@ -0,0 +1 @@
+Cap the unzip of metadata download to avoid OOM kill

changelogs/unreleased/10259-Jay2006sawant

@@ -0,0 +1 @@
+Trim spaces around resource names in --ordered-resources so comma-separated lists with spaces still match

changelogs/unreleased/10260-Lyndon-Li

@@ -0,0 +1 @@
+Add cap for backup data extraction

changelogs/unreleased/10269-blackpiglet

@@ -0,0 +1 @@
+Remove PVC and PV inclusion check during creating PVR.

changelogs/unreleased/10270-Lyndon-Li

@@ -0,0 +1 @@
+Cap the metadata decompression in object store

changelogs/unreleased/10279-harshitsaini17

@@ -0,0 +1 @@
+Use the well-known label constants exported by k8s.io/api/core/v1 instead of hardcoded label strings for kubernetes.io/hostname, kubernetes.io/os, kubernetes.io/arch, topology.kubernetes.io/zone and failure-domain.beta.kubernetes.io/zone

changelogs/unreleased/10280-nitishmalang

@@ -0,0 +1 @@
+Bound WaitRestoreExecHook polling with resourceTimeout to avoid an infinite wait when restore exec hooks never complete.

changelogs/unreleased/10283-opbot-xd

@@ -0,0 +1 @@
+test: add verification for skippedPVTracker in backup tests

changelogs/unreleased/10292-samay43

@@ -0,0 +1 @@
+Fix nil pointer dereference in EnsureDeleteVS and EnsureDeleteVSC when the API call times out before the object is retrieved

changelogs/unreleased/10293-samay43

@@ -0,0 +1 @@
+Fix nil pointer dereference in EnsureDeletePVC, EnsureDeletePV and EnsureDeletePod when the API call times out before the object is retrieved

changelogs/unreleased/10308-kaovilai

@@ -0,0 +1 @@
+Fix block data mover cancellation being reported as a backup failure

changelogs/unreleased/10312-samay43

@@ -0,0 +1 @@
+Assert expected errors from the test case rather than the returned error in pkg/util/csi tests

changelogs/unreleased/10315-opbot-xd

@@ -0,0 +1 @@
+Testing: Implement missing unit tests for pkg/backup/snapshots.go

changelogs/unreleased/10317-opbot-xd

@@ -0,0 +1 @@
+Cleanup: Remove deprecated --wait flag from velero uninstall

changelogs/unreleased/10322-Lyndon-Li

@@ -0,0 +1 @@
+Fix issue #10321, when data mover pod is evicted get the message from the data mover pod instead of the terminal message

changelogs/unreleased/10329-lubronzhan

@@ -0,0 +1 @@
+Replace generated `config/crd/{v1,v2alpha1}/crds/crds.go` with `go:embed` of the CRD YAML bases, removing the codegen step and its CI drift check

changelogs/unreleased/10333-reasonerjt

@@ -0,0 +1 @@
+Enforce namespace of the "musthave" resources in restore

changelogs/unreleased/10338-blackpiglet

@@ -0,0 +1 @@
+Avoid duplicated InitContainer names generated in velero install CLI.

changelogs/unreleased/10339-shubham-pampattiwar

@@ -0,0 +1 @@
+Add readWriteOncePod backupPVC config to enable mount-level SELinux labeling

changelogs/unreleased/10342-shubham-pampattiwar

@@ -0,0 +1 @@
+Fix issue #10341, avoid mutating the cached node-agent LoadAffinity so the OS node selector term is not appended repeatedly to data mover pods

changelogs/unreleased/10343-chlins

@@ -0,0 +1 @@
+Only sync finished backups from object storage

changelogs/unreleased/10344-Lyndon-Li

@@ -0,0 +1 @@
+Fix repo connection contest of the two repositories with the same storage type

changelogs/unreleased/10346-reasonerjt

@@ -0,0 +1 @@
+Double check the label for backup when deleting VSC- #10346

changelogs/unreleased/10352-samay43

@@ -0,0 +1 @@
+Fix nil pointer dereference in WaitUntilVSCHandleIsReady when a VolumeSnapshotContent error has no message

changelogs/unreleased/10357-samay43

@@ -0,0 +1 @@
+translate parent snapshot "auto" to an empty parent snapshot in both data mover micro services

changelogs/unreleased/10359-lubronzhan

@@ -0,0 +1 @@
+Fix e2e kind test matrix generation misparsing kindest/node pre-release tags (e.g. v1.37.0-rc.1) as bogus patch versions

changelogs/unreleased/10370-samay43

@@ -0,0 +1 @@
+fix log format string mismatches that produce wrong or mangled output

changelogs/unreleased/10371-samay43

@@ -0,0 +1 @@
+prevent panic when the restore hook init container command annotation is empty

changelogs/unreleased/9795-kaovilai

@@ -0,0 +1 @@
+Skip DeleteSnapshot when ProviderSnapshotID is empty

changelogs/unreleased/9920-shubham-pampattiwar

@@ -0,0 +1 @@
+Support copying namespace-scoped secrets and configmaps for backup and restore PVC provisioning to enable datamover backup/restore of encrypted CSI volumes

config/crd/v1/bases/velero.io_downloadrequests.yaml

@@ -16,7 +16,23 @@ spec:
     singular: downloadrequest
   scope: Namespaced
   versions:
-  - name: v1
+  - additionalPrinterColumns:
+    - description: The type of file to download
+      jsonPath: .spec.target.kind
+      name: Target Kind
+      type: string
+    - description: The name of the resource the file is associated with
+      jsonPath: .spec.target.name
+      name: Target Name
+      type: string
+    - description: The status of the download request
+      jsonPath: .status.phase
+      name: Status
+      type: string
+    - jsonPath: .metadata.creationTimestamp
+      name: Age
+      type: date
+    name: v1
     schema:
       openAPIV3Schema:
         description: |-
@@ -79,8 +95,9 @@ spec:
             description: DownloadRequestStatus is the current status of a DownloadRequest.
             properties:
               downloadURL:
-                description: DownloadURL contains the pre-signed URL for the target
-                  file.
+                description: |-
+                  DownloadURL contains the pre-signed URL for the target file. It is signed for a fixed
+                  lifetime and expires at Expiration, so it should be used promptly and not cached.
                 type: string
               expiration:
                 description: Expiration is when this DownloadRequest expires and can
@@ -88,13 +105,24 @@ spec:
                 format: date-time
                 nullable: true
                 type: string
+              message:
+                description: Message explains a Failed phase. It is empty in every
+                  other phase.
+                type: string
               phase:
-                description: Phase is the current state of the DownloadRequest.
+                description: |-
+                  Phase is the current state of the DownloadRequest. Processed means a URL has been
+                  signed into DownloadURL. It does not mean the target object exists in object storage,
+                  so a request whose target never produced a file still reaches Processed and the URL
+                  returns 404. Callers should check that the backup or restore is in a phase that
+                  produces the target before relying on the download.
                 enum:
                 - New
                 - Processed
+                - Failed
                 type: string
             type: object
         type: object
     served: true
     storage: true
+    subresources: {}

config/crd/v1/bases/velero.io_podvolumebackups.yaml

@@ -96,6 +96,13 @@ spec:
                 description: Node is the name of the node that the Pod is running
                   on.
                 type: string
+              parentSnapshot:
+                description: |-
+                  ParentSnapshot specifies the parent snapshot that current backup is based on.
+                  If its value is "" or "auto", the data mover finds the recent backup of the same volume as parent.
+                  If its value is "none", the data mover will do a full backup
+                  If its value is a specific snapshotID, the data mover finds the specific snapshot as parent.
+                type: string
               pod:
                 description: Pod is a reference to the pod containing the volume to
                   be backed up.

config/crd/v1/bases/velero.io_serverstatusrequests.yaml

@@ -16,7 +16,23 @@ spec:
     singular: serverstatusrequest
   scope: Namespaced
   versions:
-  - name: v1
+  - additionalPrinterColumns:
+    - description: The status of the server status request
+      jsonPath: .status.phase
+      name: Status
+      type: string
+    - description: The Velero server version
+      jsonPath: .status.serverVersion
+      name: Server Version
+      type: string
+    - description: The time the request was processed by the controller
+      jsonPath: .status.processedTimestamp
+      name: Processed
+      type: date
+    - jsonPath: .metadata.creationTimestamp
+      name: Age
+      type: date
+    name: v1
     schema:
       openAPIV3Schema:
         description: |-
@@ -82,3 +98,4 @@ spec:
         type: object
     served: true
     storage: true
+    subresources: {}

config/crd/v1/crds.go

@@ -0,0 +1,58 @@
+/*
+Copyright the Velero contributors.
+
+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.
+*/
+
+// Package crds embeds the controller-tools generated CRD manifests from
+// ./bases into the binary via go:embed.
+package crds
+
+import (
+	"embed"
+
+	apiextinstall "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/install"
+	apiextv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	"k8s.io/client-go/kubernetes/scheme"
+)
+
+//go:embed bases/*.yaml
+var basesFS embed.FS
+
+var CRDs = crds()
+
+func crds() []*apiextv1.CustomResourceDefinition {
+	apiextinstall.Install(scheme.Scheme)
+	decode := scheme.Codecs.UniversalDeserializer().Decode
+
+	entries, err := basesFS.ReadDir("bases")
+	if err != nil {
+		panic(err)
+	}
+
+	objs := make([]*apiextv1.CustomResourceDefinition, 0, len(entries))
+	for _, entry := range entries {
+		data, err := basesFS.ReadFile("bases/" + entry.Name())
+		if err != nil {
+			panic(err)
+		}
+
+		obj, _, err := decode(data, nil, nil)
+		if err != nil {
+			panic(err)
+		}
+		objs = append(objs, obj.(*apiextv1.CustomResourceDefinition))
+	}
+
+	return objs
+}

config/crd/v1/crds/doc.go

@@ -1,4 +0,0 @@
-// Package crds embeds the controller-tools generated CRD manifests
-package crds
-
-//go:generate go run ../../../../hack/crd-gen/v1/main.go

config/crd/v2alpha1/crds.go

@@ -0,0 +1,58 @@
+/*
+Copyright the Velero contributors.
+
+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.
+*/
+
+// Package crds embeds the controller-tools generated CRD manifests from
+// ./bases into the binary via go:embed.
+package crds
+
+import (
+	"embed"
+
+	apiextinstall "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/install"
+	apiextv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	"k8s.io/client-go/kubernetes/scheme"
+)
+
+//go:embed bases/*.yaml
+var basesFS embed.FS
+
+var CRDs = crds()
+
+func crds() []*apiextv1.CustomResourceDefinition {
+	apiextinstall.Install(scheme.Scheme)
+	decode := scheme.Codecs.UniversalDeserializer().Decode
+
+	entries, err := basesFS.ReadDir("bases")
+	if err != nil {
+		panic(err)
+	}
+
+	objs := make([]*apiextv1.CustomResourceDefinition, 0, len(entries))
+	for _, entry := range entries {
+		data, err := basesFS.ReadFile("bases/" + entry.Name())
+		if err != nil {
+			panic(err)
+		}
+
+		obj, _, err := decode(data, nil, nil)
+		if err != nil {
+			panic(err)
+		}
+		objs = append(objs, obj.(*apiextv1.CustomResourceDefinition))
+	}
+
+	return objs
+}

config/crd/v2alpha1/crds/doc.go

@@ -1,4 +0,0 @@
-// Package crds embeds the controller-tools generated CRD manifests
-package crds
-
-//go:generate go run ../../../../hack/crd-gen/v1/main.go

config/rbac/role.yaml

@@ -4,6 +4,16 @@ kind: ClusterRole
 metadata:
   name: velero-perms
 rules:
+- apiGroups:
+  - ""
+  resources:
+  - configmaps
+  - secrets
+  verbs:
+  - create
+  - delete
+  - get
+  - list
 - apiGroups:
   - ""
   resources:

go.mod

@@ -50,7 +50,7 @@ require (
 	golang.org/x/text v0.37.0
 	google.golang.org/api v0.283.0
 	google.golang.org/grpc v1.82.1
-	google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af
+	google.golang.org/protobuf v1.36.12
 	k8s.io/api v0.36.0
 	k8s.io/apiextensions-apiserver v0.36.0
 	k8s.io/apimachinery v0.36.0

go.sum

@@ -566,8 +566,8 @@ google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa h1:
 google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8=
 google.golang.org/grpc v1.82.1 h1:NnAxzGRA0677vCa4BUkOAnO5+FfQqVl9iUXeD0IqcGE=
 google.golang.org/grpc v1.82.1/go.mod h1:yzTZ1TB1Z3SG+LIYaI+WiE8D5+PZ3ArnrSp8zF3+/ZA=
-google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af h1:+5/Sw3GsDNlEmu7TfklWKPdQ0Ykja5VEmq2i817+jbI=
-google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
+google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc=
+google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=

hack/crd-gen/v1/main.go

@@ -1,134 +0,0 @@
-/*
-Copyright the Velero contributors.
-
-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.
-*/
-
-// This code embeds the CRD manifests in ../bases in ../crds/crds.go
-
-package main
-
-import (
-	"bytes"
-	"compress/gzip"
-	"fmt"
-	"io"
-	"log"
-	"os"
-	"text/template"
-)
-
-// This is relative to config/crd/crds
-const goHeaderFile = "../../../../hack/boilerplate.go.txt"
-
-const tpl = `{{.GoHeader}}
-// Code generated by crds_generate.go; DO NOT EDIT.
-
-package crds
-
-import (
-	"bytes"
-	"compress/gzip"
-	"io"
-
-	apiextinstall "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/install"
-	apiextv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
-	"k8s.io/client-go/kubernetes/scheme"
-)
-
-var rawCRDs = [][]byte{
-{{- range .RawCRDs }}
-	[]byte({{ . }}),
-{{- end }}
-}
-
-var CRDs = crds()
-
-func crds() []*apiextv1.CustomResourceDefinition {
-	apiextinstall.Install(scheme.Scheme)
-	decode := scheme.Codecs.UniversalDeserializer().Decode
-	var objs []*apiextv1.CustomResourceDefinition
-	for _, crd := range rawCRDs {
-		gzr, err := gzip.NewReader(bytes.NewReader(crd))
-		if err != nil {
-			panic(err)
-		}
-		bytes, err := io.ReadAll(gzr)
-		if err != nil {
-			panic(err)
-		}
-		gzr.Close()
-
-		obj, _, err := decode(bytes, nil, nil)
-		if err != nil {
-			panic(err)
-		}
-		objs = append(objs, obj.(*apiextv1.CustomResourceDefinition))
-	}
-	return objs
-}
-`
-
-type templateData struct {
-	GoHeader string
-	RawCRDs  []string
-}
-
-func main() {
-	headerBytes, err := os.ReadFile(goHeaderFile)
-	if err != nil {
-		log.Fatalln(err)
-	}
-
-	data := templateData{
-		GoHeader: string(headerBytes),
-	}
-
-	// This is relative to config/crd/crds
-	manifests, err := os.ReadDir("../bases")
-	if err != nil {
-		log.Fatalln(err)
-	}
-
-	for _, crd := range manifests {
-		file, err := os.Open("../bases/" + crd.Name())
-		if err != nil {
-			log.Fatalln(err)
-		}
-
-		// gzip compress manifest
-		var buf bytes.Buffer
-		gzw := gzip.NewWriter(&buf)
-		if _, err := io.Copy(gzw, file); err != nil {
-			log.Fatalln(err)
-		}
-		file.Close()
-		gzw.Close()
-
-		data.RawCRDs = append(data.RawCRDs, fmt.Sprintf("%q", buf.Bytes()))
-	}
-
-	t, err := template.New("crd").Parse(tpl)
-	if err != nil {
-		log.Fatalln(err)
-	}
-
-	out, err := os.Create("crds.go")
-	if err != nil {
-		log.Fatalln(err)
-	}
-
-	if err := t.Execute(out, data); err != nil {
-		log.Fatalln(err)
-	}
-}

hack/update-3generated-crd-code.sh

@@ -55,6 +55,6 @@ controller-gen \
   paths=./pkg/controller/... \
   rbac:roleName=velero-perms
 
-go generate ./config/crd/v1/crds
-
-go generate ./config/crd/v2alpha1/crds
+# The CRD manifests above are embedded directly into the binary via
+# go:embed (see config/crd/v1/crds.go and config/crd/v2alpha1/crds.go),
+# so no further code generation step is required.

hack/verify-generated-crd-code.sh

@@ -1,29 +0,0 @@
-#!/bin/bash -e
-#
-# Copyright the Velero contributors.
-#
-# 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.
-
-HACK_DIR=$(dirname "${BASH_SOURCE}")
-
-${HACK_DIR}/update-3generated-crd-code.sh
-
-# ensure no changes to generated CRDs
-if ! git diff --exit-code config/crd/v1/crds/crds.go config/crd/v2alpha1/crds/crds.go &> /dev/null; then
-  # revert changes to state before running CRD generation to stay consistent
-  # with code-generator `--verify-only` option which discards generated changes
-  git checkout config/crd
-
-  echo "CRD verification - failed! Generated CRDs are out-of-date, please run 'make update' and 'git add' the generated file(s)."
-  exit 1
-fi

internal/delete/actions/csi/volumesnapshotcontent_action.go

@@ -86,7 +86,7 @@ func (p *volumeSnapshotContentDeleteItemAction) Execute(
 	// This handles legacy (pre-1.15) backups where the original VSC
 	// with DeletionPolicy=Retain still exists in the cluster.
 	originalVSCName := snapCont.Name
-	if cleaned := p.tryDeleteOriginalVSC(context.TODO(), originalVSCName); cleaned {
+	if cleaned := p.tryDeleteOriginalVSC(context.TODO(), originalVSCName, input.Backup.Name); cleaned {
 		p.log.Infof("Successfully deleted original VolumeSnapshotContent %s from cluster, skipping temp VSC creation", originalVSCName)
 		return nil
 	}
@@ -149,10 +149,11 @@ func (p *volumeSnapshotContentDeleteItemAction) Execute(
 // the cluster (legacy pre-1.15 backups). It patches the DeletionPolicy to
 // Delete so the CSI driver also removes the cloud snapshot, then deletes
 // the VSC object itself.
-// Returns true if the original VSC was found and deletion was initiated.
+// Returns true if the original VSC was found, carries the backup label, and deletion was initiated.
 func (p *volumeSnapshotContentDeleteItemAction) tryDeleteOriginalVSC(
 	ctx context.Context,
 	vscName string,
+	backupName string,
 ) bool {
 	existing := new(snapshotv1api.VolumeSnapshotContent)
 	if err := p.crClient.Get(ctx, crclient.ObjectKey{Name: vscName}, existing); err != nil {
@@ -164,6 +165,15 @@ func (p *volumeSnapshotContentDeleteItemAction) tryDeleteOriginalVSC(
 		return false
 	}
 
+	if !kubeutil.HasBackupLabel(&existing.ObjectMeta, backupName) {
+		p.log.Warnf(
+			"Original VolumeSnapshotContent %s in cluster does not belong to backup %s, skipping direct deletion",
+			vscName,
+			backupName,
+		)
+		return false
+	}
+
 	p.log.Debugf("Found original VolumeSnapshotContent %s in cluster (legacy backup), cleaning up directly", vscName)
 
 	// Patch DeletionPolicy to Delete so the CSI driver removes the cloud snapshot

internal/delete/actions/csi/volumesnapshotcontent_action_test.go

@@ -122,7 +122,29 @@ func TestVSCExecute(t *testing.T) {
 			backup:    builder.ForBackup("velero", "backup").Result(),
 			expectErr: false,
 			preExistingVSC: &snapshotv1api.VolumeSnapshotContent{
-				ObjectMeta: metav1.ObjectMeta{Name: "bar"},
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "bar",
+					Labels: map[string]string{
+						velerov1api.BackupNameLabel: "backup",
+					},
+				},
+				Spec: snapshotv1api.VolumeSnapshotContentSpec{
+					DeletionPolicy:    snapshotv1api.VolumeSnapshotContentRetain,
+					Driver:            "disk.csi.azure.com",
+					Source:            snapshotv1api.VolumeSnapshotContentSource{SnapshotHandle: stringPtr("snap-123")},
+					VolumeSnapshotRef: corev1api.ObjectReference{Name: "vs-1", Namespace: "default"},
+				},
+			},
+		},
+		{
+			name:      "Original VSC exists in cluster without backup label, falls through to temp VSC flow",
+			vsc:       builder.ForVolumeSnapshotContent("bar").ObjectMeta(builder.WithLabelsMap(map[string]string{velerov1api.BackupNameLabel: "backup"})).Status(&snapshotv1api.VolumeSnapshotContentStatus{SnapshotHandle: &snapshotHandleStr}).Result(),
+			backup:    builder.ForBackup("velero", "backup").Result(),
+			expectErr: false,
+			preExistingVSC: &snapshotv1api.VolumeSnapshotContent{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "bar",
+				},
 				Spec: snapshotv1api.VolumeSnapshotContentSpec{
 					DeletionPolicy:    snapshotv1api.VolumeSnapshotContentRetain,
 					Driver:            "disk.csi.azure.com",
@@ -200,22 +222,51 @@ func TestNewVolumeSnapshotContentDeleteItemAction(t *testing.T) {
 
 func TestTryDeleteOriginalVSC(t *testing.T) {
 	tests := []struct {
-		name      string
-		vscName   string
-		existing  *snapshotv1api.VolumeSnapshotContent
-		createIt  bool
-		expectRet bool
+		name       string
+		vscName    string
+		backupName string
+		existing   *snapshotv1api.VolumeSnapshotContent
+		createIt   bool
+		expectRet  bool
 	}{
 		{
-			name:      "VSC not found in cluster, returns false",
-			vscName:   "not-found",
+			name:       "VSC not found in cluster, returns false",
+			vscName:    "not-found",
+			backupName: "test-backup",
+			expectRet:  false,
+		},
+		{
+			name:       "VSC found in cluster without backup label, returns false",
+			vscName:    "unlabeled-vsc",
+			backupName: "test-backup",
+			existing: &snapshotv1api.VolumeSnapshotContent{
+				ObjectMeta: metav1.ObjectMeta{Name: "unlabeled-vsc"},
+				Spec: snapshotv1api.VolumeSnapshotContentSpec{
+					DeletionPolicy: snapshotv1api.VolumeSnapshotContentRetain,
+					Driver:         "disk.csi.azure.com",
+					Source: snapshotv1api.VolumeSnapshotContentSource{
+						SnapshotHandle: stringPtr("snap-123"),
+					},
+					VolumeSnapshotRef: corev1api.ObjectReference{
+						Name:      "vs-1",
+						Namespace: "default",
+					},
+				},
+			},
+			createIt:  true,
 			expectRet: false,
 		},
 		{
-			name:    "VSC found with Retain policy, patches and deletes",
-			vscName: "legacy-vsc",
+			name:       "VSC found with Retain policy and matching backup label, patches and deletes",
+			vscName:    "legacy-vsc",
+			backupName: "test-backup",
 			existing: &snapshotv1api.VolumeSnapshotContent{
-				ObjectMeta: metav1.ObjectMeta{Name: "legacy-vsc"},
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "legacy-vsc",
+					Labels: map[string]string{
+						velerov1api.BackupNameLabel: "test-backup",
+					},
+				},
 				Spec: snapshotv1api.VolumeSnapshotContentSpec{
 					DeletionPolicy: snapshotv1api.VolumeSnapshotContentRetain,
 					Driver:         "disk.csi.azure.com",
@@ -232,10 +283,16 @@ func TestTryDeleteOriginalVSC(t *testing.T) {
 			expectRet: true,
 		},
 		{
-			name:    "VSC found with Delete policy already, just deletes",
-			vscName: "already-delete-vsc",
+			name:       "VSC found with Delete policy and matching backup label, just deletes",
+			vscName:    "already-delete-vsc",
+			backupName: "test-backup",
 			existing: &snapshotv1api.VolumeSnapshotContent{
-				ObjectMeta: metav1.ObjectMeta{Name: "already-delete-vsc"},
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "already-delete-vsc",
+					Labels: map[string]string{
+						velerov1api.BackupNameLabel: "test-backup",
+					},
+				},
 				Spec: snapshotv1api.VolumeSnapshotContentSpec{
 					DeletionPolicy: snapshotv1api.VolumeSnapshotContentDelete,
 					Driver:         "disk.csi.azure.com",
@@ -266,7 +323,7 @@ func TestTryDeleteOriginalVSC(t *testing.T) {
 				require.NoError(t, crClient.Create(t.Context(), test.existing))
 			}
 
-			result := p.tryDeleteOriginalVSC(t.Context(), test.vscName)
+			result := p.tryDeleteOriginalVSC(t.Context(), test.vscName, test.backupName)
 			require.Equal(t, test.expectRet, result)
 
 			// If cleanup succeeded, verify the VSC is gone
@@ -289,13 +346,18 @@ func TestTryDeleteOriginalVSC(t *testing.T) {
 			log:      logrus.StandardLogger(),
 			crClient: errClient,
 		}
-		require.False(t, p.tryDeleteOriginalVSC(t.Context(), "some-vsc"))
+		require.False(t, p.tryDeleteOriginalVSC(t.Context(), "some-vsc", "test-backup"))
 	})
 
 	t.Run("Patch fails, returns false", func(t *testing.T) {
 		realClient := velerotest.NewFakeControllerRuntimeClient(t)
 		vsc := &snapshotv1api.VolumeSnapshotContent{
-			ObjectMeta: metav1.ObjectMeta{Name: "patch-fail-vsc"},
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "patch-fail-vsc",
+				Labels: map[string]string{
+					velerov1api.BackupNameLabel: "test-backup",
+				},
+			},
 			Spec: snapshotv1api.VolumeSnapshotContentSpec{
 				DeletionPolicy:    snapshotv1api.VolumeSnapshotContentRetain,
 				Driver:            "disk.csi.azure.com",
@@ -313,13 +375,18 @@ func TestTryDeleteOriginalVSC(t *testing.T) {
 			log:      logrus.StandardLogger(),
 			crClient: errClient,
 		}
-		require.False(t, p.tryDeleteOriginalVSC(t.Context(), "patch-fail-vsc"))
+		require.False(t, p.tryDeleteOriginalVSC(t.Context(), "patch-fail-vsc", "test-backup"))
 	})
 
 	t.Run("Delete fails, returns false", func(t *testing.T) {
 		realClient := velerotest.NewFakeControllerRuntimeClient(t)
 		vsc := &snapshotv1api.VolumeSnapshotContent{
-			ObjectMeta: metav1.ObjectMeta{Name: "delete-fail-vsc"},
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "delete-fail-vsc",
+				Labels: map[string]string{
+					velerov1api.BackupNameLabel: "test-backup",
+				},
+			},
 			Spec: snapshotv1api.VolumeSnapshotContentSpec{
 				DeletionPolicy:    snapshotv1api.VolumeSnapshotContentDelete,
 				Driver:            "disk.csi.azure.com",
@@ -337,7 +404,7 @@ func TestTryDeleteOriginalVSC(t *testing.T) {
 			log:      logrus.StandardLogger(),
 			crClient: errClient,
 		}
-		require.False(t, p.tryDeleteOriginalVSC(t.Context(), "delete-fail-vsc"))
+		require.False(t, p.tryDeleteOriginalVSC(t.Context(), "delete-fail-vsc", "test-backup"))
 	})
 }
 

internal/hook/item_hook_handler.go

@@ -365,6 +365,12 @@ func getPodExecHookFromAnnotations(annotations map[string]string, phase HookPhas
 
 func parseStringToCommand(commandValue string) []string {
 	var command []string
+	// An empty command means the container image's own entrypoint should be used.
+	// Callers that require a command already return early; getInitContainerFromAnnotation
+	// deliberately allows this case, so return nil rather than indexing an empty string.
+	if commandValue == "" {
+		return nil
+	}
 	// check for json array
 	if commandValue[0] == '[' {
 		if err := json.Unmarshal([]byte(commandValue), &command); err != nil {
@@ -419,7 +425,7 @@ func getInitContainerFromAnnotation(podName string, annotations map[string]strin
 		return nil
 	}
 	if command == "" {
-		log.Infof("RestoreHook init container for pod %s is using container's default entrypoint", podName, containerImage)
+		log.Infof("RestoreHook init container for pod %s is using the default entrypoint of image %s", podName, containerImage)
 	}
 	if containerName == "" {
 		uid, err := uuid.NewRandom()

internal/hook/item_hook_handler_test.go

@@ -1287,6 +1287,25 @@ func TestGetInitContainerFromAnnotations(t *testing.T) {
 				podRestoreHookInitContainerCommandAnnotationKey: "[foobarbaz",
 			},
 		},
+		{
+			name:      "should use the image's default entrypoint when the command annotation is empty",
+			expectNil: false,
+			expected:  builder.ForContainer("restore-init1", "busy-box").Result(),
+			inputAnnotations: map[string]string{
+				podRestoreHookInitContainerImageAnnotationKey:   "busy-box",
+				podRestoreHookInitContainerNameAnnotationKey:    "restore-init",
+				podRestoreHookInitContainerCommandAnnotationKey: "",
+			},
+		},
+		{
+			name:      "should use the image's default entrypoint when the command annotation is missing",
+			expectNil: false,
+			expected:  builder.ForContainer("restore-init1", "busy-box").Result(),
+			inputAnnotations: map[string]string{
+				podRestoreHookInitContainerImageAnnotationKey: "busy-box",
+				podRestoreHookInitContainerNameAnnotationKey:  "restore-init",
+			},
+		},
 	}
 
 	for _, tc := range testCases {

internal/volume/snapshotlocation.go

@@ -29,6 +29,10 @@ func UpdateVolumeSnapshotLocationWithCredentialConfig(location *velerov1api.Volu
 	if location.Spec.Config == nil {
 		location.Spec.Config = make(map[string]string)
 	}
+
+	// Delete any user-provided credentialsFile to prevent path traversal vulnerabilities
+	delete(location.Spec.Config, "credentialsFile")
+
 	// If the VSL specifies a credential, fetch its path on disk and pass to
 	// plugin via the config.
 	if location.Spec.Credential != nil && credentialStore != nil {

internal/volume/volumes_information.go

@@ -36,6 +36,7 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/features"
 	"github.com/vmware-tanzu/velero/pkg/itemoperation"
 	"github.com/vmware-tanzu/velero/pkg/kuberesource"
+	"github.com/vmware-tanzu/velero/pkg/util/stringptr"
 )
 
 type Method string
@@ -494,7 +495,8 @@ func (v *BackupVolumesInformation) generateVolumeInfoForCSIVolumeSnapshot() {
 
 			tmpVolumeInfos = append(tmpVolumeInfos, volumeInfo)
 		} else {
-			v.logger.Warnf("cannot find info for PVC %s/%s", volumeSnapshot.Namespace, volumeSnapshot.Spec.Source.PersistentVolumeClaimName)
+			v.logger.Warnf("cannot find info for PVC %s/%s", volumeSnapshot.Namespace,
+				stringptr.GetString(volumeSnapshot.Spec.Source.PersistentVolumeClaimName))
 			continue
 		}
 	}

pkg/apis/velero/shared/constants.go

@@ -17,6 +17,6 @@ limitations under the License.
 package shared
 
 const (
-	DataUploadParentSnapshotNone = "none"
-	DataUploadParentSnapshotAuto = "auto"
+	ParentSnapshotNone = "none"
+	ParentSnapshotAuto = "auto"
 )

pkg/apis/velero/v1/download_request_types.go

@@ -56,7 +56,7 @@ type DownloadTarget struct {
 }
 
 // DownloadRequestPhase represents the lifecycle phase of a DownloadRequest.
-// +kubebuilder:validation:Enum=New;Processed
+// +kubebuilder:validation:Enum=New;Processed;Failed
 type DownloadRequestPhase string
 
 const (
@@ -64,18 +64,30 @@ const (
 	// DownloadRequestController yet.
 	DownloadRequestPhaseNew DownloadRequestPhase = "New"
 
-	// DownloadRequestPhaseProcessed means the DownloadRequest has been processed by the
-	// DownloadRequestController.
+	// DownloadRequestPhaseProcessed means the DownloadRequestController has signed a URL
+	// into Status.DownloadURL. The controller signs the key by convention and does not
+	// check that the object is present, so this phase does not imply the file exists.
 	DownloadRequestPhaseProcessed DownloadRequestPhase = "Processed"
+
+	// DownloadRequestPhaseFailed means the controller will not sign a URL for this request
+	// and no retry will change that. Status.Message carries the reason. A caller waiting on
+	// Status.DownloadURL should stop when it sees this phase rather than poll until its own
+	// timeout, which would report a storage problem that is not the cause.
+	DownloadRequestPhaseFailed DownloadRequestPhase = "Failed"
 )
 
 // DownloadRequestStatus is the current status of a DownloadRequest.
 type DownloadRequestStatus struct {
-	// Phase is the current state of the DownloadRequest.
+	// Phase is the current state of the DownloadRequest. Processed means a URL has been
+	// signed into DownloadURL. It does not mean the target object exists in object storage,
+	// so a request whose target never produced a file still reaches Processed and the URL
+	// returns 404. Callers should check that the backup or restore is in a phase that
+	// produces the target before relying on the download.
 	// +optional
 	Phase DownloadRequestPhase `json:"phase,omitempty"`
 
-	// DownloadURL contains the pre-signed URL for the target file.
+	// DownloadURL contains the pre-signed URL for the target file. It is signed for a fixed
+	// lifetime and expires at Expiration, so it should be used promptly and not cached.
 	// +optional
 	DownloadURL string `json:"downloadURL,omitempty"`
 
@@ -83,6 +95,10 @@ type DownloadRequestStatus struct {
 	// +optional
 	// +nullable
 	Expiration *metav1.Time `json:"expiration,omitempty"`
+
+	// Message explains a Failed phase. It is empty in every other phase.
+	// +optional
+	Message string `json:"message,omitempty"`
 }
 
 // TODO(2.0) After converting all resources to use the runtime-controller client,
@@ -93,6 +109,10 @@ type DownloadRequestStatus struct {
 // +kubebuilder:object:generate=true
 // +kubebuilder:storageversion
 // +kubebuilder:resource:shortName=dreq
+// +kubebuilder:printcolumn:name="Target Kind",type="string",JSONPath=".spec.target.kind",description="The type of file to download"
+// +kubebuilder:printcolumn:name="Target Name",type="string",JSONPath=".spec.target.name",description="The name of the resource the file is associated with"
+// +kubebuilder:printcolumn:name="Status",type="string",JSONPath=".status.phase",description="The status of the download request"
+// +kubebuilder:printcolumn:name="Age",type="date",JSONPath=".metadata.creationTimestamp"
 
 // DownloadRequest is a request to download an artifact from backup object storage, such as a backup
 // log file.

pkg/apis/velero/v1/pod_volume_backup_types.go

@@ -61,6 +61,12 @@ type PodVolumeBackupSpec struct {
 	// Cancel indicates request to cancel the ongoing PodVolumeBackup. It can be set
 	// when the PodVolumeBackup is in InProgress phase
 	Cancel bool `json:"cancel,omitempty"`
+
+	// ParentSnapshot specifies the parent snapshot that current backup is based on.
+	// If its value is "" or "auto", the data mover finds the recent backup of the same volume as parent.
+	// If its value is "none", the data mover will do a full backup
+	// If its value is a specific snapshotID, the data mover finds the specific snapshot as parent.
+	ParentSnapshot string `json:"parentSnapshot,omitempty"`
 }
 
 // PodVolumeBackupPhase represents the lifecycle phase of a PodVolumeBackup.

pkg/apis/velero/v1/server_status_request_types.go

@@ -28,6 +28,10 @@ import (
 // +kubebuilder:resource:shortName=ssr
 // +kubebuilder:object:generate=true
 // +kubebuilder:storageversion
+// +kubebuilder:printcolumn:name="Status",type="string",JSONPath=".status.phase",description="The status of the server status request"
+// +kubebuilder:printcolumn:name="Server Version",type="string",JSONPath=".status.serverVersion",description="The Velero server version"
+// +kubebuilder:printcolumn:name="Processed",type="date",JSONPath=".status.processedTimestamp",description="The time the request was processed by the controller"
+// +kubebuilder:printcolumn:name="Age",type="date",JSONPath=".metadata.creationTimestamp"
 
 // ServerStatusRequest is a request to access current status information about
 // the Velero server.

pkg/archive/extractor.go

@@ -32,14 +32,27 @@ import (
 // Extractor unzips/extracts a backup tarball to a local
 // temp directory.
 type Extractor struct {
-	log logrus.FieldLogger
-	fs  filesystem.Interface
+	log                logrus.FieldLogger
+	fs                 filesystem.Interface
+	maxExtractionSize  int64
+	totalExtractedSize int64
+}
+
+var maxExtractionSize = int64(16) << 30
+
+// SetMaxExtractionSize sets the maximum extraction size. It is normally called at server startup.
+func SetMaxExtractionSize(size int64) {
+	if size > 0 {
+		maxExtractionSize = size
+	}
 }
 
 func NewExtractor(log logrus.FieldLogger, fs filesystem.Interface) *Extractor {
 	return &Extractor{
-		log: log,
-		fs:  fs,
+		log:                log,
+		fs:                 fs,
+		maxExtractionSize:  maxExtractionSize,
+		totalExtractedSize: 0,
 	}
 }
 
@@ -96,6 +109,15 @@ func (e *Extractor) readBackup(tarRdr *tar.Reader) (string, error) {
 			return "", err
 		}
 
+		// Enforce maximum extraction size to prevent memory/storage exhaustion and zip bombs.
+		maxSize := e.maxExtractionSize
+		e.totalExtractedSize += header.Size
+		if e.totalExtractedSize > maxSize {
+			err := fmt.Errorf("decompressed backup exceeds maximum allowed size of %d bytes", maxSize)
+			e.log.Infof("error checking extracted size: %v", err)
+			return "", err
+		}
+
 		target, err := sanitizeArchivePath(dir, header.Name)
 		if err != nil {
 			e.log.Infof("error sanitizing archive path: %s", err.Error())

pkg/archive/extractor_test.go

@@ -20,6 +20,7 @@ import (
 	"archive/tar"
 	"bytes"
 	"compress/gzip"
+	"fmt"
 	"io"
 	"os"
 	"testing"
@@ -113,6 +114,66 @@ func TestUnzipAndExtractBackupRejectsPathTraversal(t *testing.T) {
 	require.Contains(t, err.Error(), "invalid archive path")
 }
 
+func TestUnzipAndExtractBackupRejectsLargeFile(t *testing.T) {
+	SetMaxExtractionSize(1024)
+	defer SetMaxExtractionSize(16 * 1024 * 1024 * 1024)
+	ext := NewExtractor(test.NewLogger(), test.NewFakeFileSystem())
+
+	var buf bytes.Buffer
+	gzw := gzip.NewWriter(&buf)
+	tw := tar.NewWriter(gzw)
+
+	data := make([]byte, 2048) // 2KB data
+	err := tw.WriteHeader(&tar.Header{
+		Name:     "large.txt",
+		Mode:     0600,
+		Typeflag: tar.TypeReg,
+		Size:     int64(len(data)),
+	})
+	require.NoError(t, err)
+
+	_, err = tw.Write(data)
+	require.NoError(t, err)
+	require.NoError(t, tw.Close())
+	require.NoError(t, gzw.Close())
+
+	_, err = ext.UnzipAndExtractBackup(&buf)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "decompressed backup exceeds maximum allowed size")
+}
+
+func TestUnzipAndExtractBackupRejectsManySmallFiles(t *testing.T) {
+	SetMaxExtractionSize(1024)
+	defer SetMaxExtractionSize(16 * 1024 * 1024 * 1024)
+	ext := NewExtractor(test.NewLogger(), test.NewFakeFileSystem())
+
+	var buf bytes.Buffer
+	gzw := gzip.NewWriter(&buf)
+	tw := tar.NewWriter(gzw)
+
+	// Create 100 files of 20 bytes each (total 2000 bytes, exceeding the 1024 byte limit)
+	for i := 0; i < 100; i++ {
+		data := make([]byte, 20)
+		err := tw.WriteHeader(&tar.Header{
+			Name:     fmt.Sprintf("small_%d.txt", i),
+			Mode:     0600,
+			Typeflag: tar.TypeReg,
+			Size:     int64(len(data)),
+		})
+		require.NoError(t, err)
+
+		_, err = tw.Write(data)
+		require.NoError(t, err)
+	}
+
+	require.NoError(t, tw.Close())
+	require.NoError(t, gzw.Close())
+
+	_, err := ext.UnzipAndExtractBackup(&buf)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "decompressed backup exceeds maximum allowed size")
+}
+
 func createArchive(files []string, fs filesystem.Interface) (string, error) {
 	outName := "output.tar.gz"
 	out, err := fs.Create(outName)

pkg/backup/actions/csi/pvc_action.go

@@ -212,6 +212,7 @@ func (p *pvcBackupItemAction) validatePVCAndPV(
 }
 
 func (p *pvcBackupItemAction) createVolumeSnapshot(
+	ctx context.Context,
 	pvc corev1api.PersistentVolumeClaim,
 	backup *velerov1api.Backup,
 	policySnapshotClass string,
@@ -222,7 +223,7 @@ func (p *pvcBackupItemAction) createVolumeSnapshot(
 	p.log.Debugf("Fetching storage class for PV %s", *pvc.Spec.StorageClassName)
 	storageClass := new(storagev1api.StorageClass)
 	if err := p.crClient.Get(
-		context.TODO(), crclient.ObjectKey{Name: *pvc.Spec.StorageClassName},
+		ctx, crclient.ObjectKey{Name: *pvc.Spec.StorageClassName},
 		storageClass,
 	); err != nil {
 		return nil, errors.Wrap(err, "error getting storage class")
@@ -230,6 +231,7 @@ func (p *pvcBackupItemAction) createVolumeSnapshot(
 
 	p.log.Debugf("Fetching VolumeSnapshotClass for %s", storageClass.Provisioner)
 	vsClass, err := csi.GetVolumeSnapshotClass(
+		ctx,
 		storageClass.Provisioner,
 		backup,
 		&pvc,
@@ -266,7 +268,7 @@ func (p *pvcBackupItemAction) createVolumeSnapshot(
 		},
 	}
 
-	if err := p.crClient.Create(context.TODO(), vs); err != nil {
+	if err := p.crClient.Create(ctx, vs); err != nil {
 		return nil, errors.Wrapf(
 			err, "error creating volume snapshot",
 		)
@@ -295,6 +297,8 @@ func (p *pvcBackupItemAction) Execute(
 ) {
 	p.log.Info("Starting PVCBackupItemAction")
 
+	ctx := context.Background()
+
 	if valid := p.validateBackup(*backup); !valid {
 		return item, nil, "", nil, nil
 	}
@@ -319,7 +323,7 @@ func (p *pvcBackupItemAction) Execute(
 	}
 
 	// Ensure PVC-to-Pod cache is built for this namespace (lazy per-namespace caching)
-	if err := p.ensurePVCPodCacheForNamespace(context.TODO(), pvc.Namespace); err != nil {
+	if err := p.ensurePVCPodCacheForNamespace(ctx, pvc.Namespace); err != nil {
 		return nil, nil, "", nil, err
 	}
 
@@ -347,7 +351,7 @@ func (p *pvcBackupItemAction) Execute(
 	// created but never processed (the DataUpload controller runs inside node-agent),
 	// causing the backup to hang until itemOperationTimeout expires.
 	if boolptr.IsSetToTrue(backup.Spec.SnapshotMoveData) && datamover.IsBuiltInDataMover(backup.Spec.DataMover) {
-		if err := nodeagent.IsReady(context.TODO(), backup.Namespace, p.crClient, p.log); err != nil {
+		if err := nodeagent.IsReady(ctx, backup.Namespace, p.crClient, p.log); err != nil {
 			p.log.WithError(err).Error("cannot perform snapshot data movement without running node-agent pods")
 			return nil, nil, "", nil, errors.Wrap(err, "CSI PVC BIA cannot proceed: node-agent is not ready for snapshot data movement")
 		}
@@ -360,7 +364,7 @@ func (p *pvcBackupItemAction) Execute(
 		p.log.Infof("Volume policy specifies snapshotClass=%s for PVC %s/%s", policySnapshotClass, pvc.Namespace, pvc.Name)
 	}
 
-	vs, err := p.getVolumeSnapshotReference(context.TODO(), pvc, backup, policySnapshotClass)
+	vs, err := p.getVolumeSnapshotReference(ctx, pvc, backup, policySnapshotClass)
 	if err != nil {
 		return nil, nil, "", nil, err
 	}
@@ -376,7 +380,7 @@ func (p *pvcBackupItemAction) Execute(
 	if err != nil {
 		p.log.Errorf("Failed to wait for VolumeSnapshot %s/%s to become ReadyToUse within timeout %v: %s",
 			vs.Namespace, vs.Name, backup.Spec.CSISnapshotTimeout.Duration, err.Error())
-		csi.CleanupVolumeSnapshot(vs, p.crClient, p.log)
+		csi.CleanupVolumeSnapshot(ctx, vs, p.crClient, p.log)
 		return nil, nil, "", nil, errors.WithStack(err)
 	}
 
@@ -427,7 +431,7 @@ func (p *pvcBackupItemAction) Execute(
 
 			// TODO: need to use DeleteVolumeSnapshotIfAny, after data mover
 			// adopting the controller-runtime client.
-			if deleteErr := p.crClient.Delete(context.TODO(), vs); deleteErr != nil {
+			if deleteErr := p.crClient.Delete(ctx, vs); deleteErr != nil {
 				if !apierrors.IsNotFound(deleteErr) {
 					dataUploadLog.WithError(deleteErr).Error("fail to delete VolumeSnapshot")
 				}
@@ -565,7 +569,7 @@ func newDataUpload(
 	parentSnapshot := ""
 
 	if backup.Spec.BackupType == velerov1api.BackupTypeFull {
-		parentSnapshot = veleroshared.DataUploadParentSnapshotNone
+		parentSnapshot = veleroshared.ParentSnapshotNone
 	}
 
 	dataMover := backup.Spec.DataMover
@@ -841,7 +845,7 @@ func (p *pvcBackupItemAction) getVolumeSnapshotReference(
 	}
 
 	// Legacy fallback: create individual VS
-	return p.createVolumeSnapshot(pvc, backup, policySnapshotClass)
+	return p.createVolumeSnapshot(ctx, pvc, backup, policySnapshotClass)
 }
 
 func (p *pvcBackupItemAction) findExistingVSForBackup(
@@ -1230,7 +1234,7 @@ func setPVCRequestSizeToVSRestoreSize(
 	logger logrus.FieldLogger,
 ) {
 	if vsc.Status.RestoreSize != nil {
-		logger.Debugf("Patching PVC request size to fit the volumesnapshot restore size %d", vsc.Status.RestoreSize)
+		logger.Debugf("Patching PVC request size to fit the volumesnapshot restore size %d", *vsc.Status.RestoreSize)
 		restoreSize := *resource.NewQuantity(*vsc.Status.RestoreSize, resource.BinarySI)
 
 		// It is possible that the volume provider allocated a larger

pkg/backup/actions/csi/pvc_action_test.go

@@ -131,7 +131,7 @@ func TestExecute(t *testing.T) {
 			vsClass: builder.ForVolumeSnapshotClass("testVSClass").Driver("hostpath").ObjectMeta(builder.WithLabels(velerov1api.VolumeSnapshotClassSelectorLabel, "")).Result(),
 			extraObjects: []runtime.Object{
 				&corev1api.Node{
-					ObjectMeta: metav1.ObjectMeta{Name: "linux-node", Labels: map[string]string{"kubernetes.io/os": "linux"}},
+					ObjectMeta: metav1.ObjectMeta{Name: "linux-node", Labels: map[string]string{corev1api.LabelOSStable: "linux"}},
 				},
 				&appsv1api.DaemonSet{
 					ObjectMeta: metav1.ObjectMeta{Namespace: "velero", Name: "node-agent"},
@@ -186,7 +186,7 @@ func TestExecute(t *testing.T) {
 			vsClass: builder.ForVolumeSnapshotClass("tescVSClass").Driver("hostpath").ObjectMeta(builder.WithLabels(velerov1api.VolumeSnapshotClassSelectorLabel, "")).Result(),
 			extraObjects: []runtime.Object{
 				&corev1api.Node{
-					ObjectMeta: metav1.ObjectMeta{Name: "linux-node", Labels: map[string]string{"kubernetes.io/os": "linux"}},
+					ObjectMeta: metav1.ObjectMeta{Name: "linux-node", Labels: map[string]string{corev1api.LabelOSStable: "linux"}},
 				},
 				&appsv1api.DaemonSet{
 					ObjectMeta: metav1.ObjectMeta{Namespace: "velero", Name: "node-agent"},

pkg/backup/actions/csi/volumesnapshot_action.go

@@ -78,6 +78,8 @@ func (p *volumeSnapshotBackupItemAction) Execute(
 ) {
 	p.log.Infof("Executing VolumeSnapshotBackupItemAction")
 
+	ctx := context.Background()
+
 	vs := new(snapshotv1api.VolumeSnapshot)
 	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(
 		item.UnstructuredContent(), vs); err != nil {
@@ -90,7 +92,7 @@ func (p *volumeSnapshotBackupItemAction) Execute(
 			WithField("Backup", fmt.Sprintf("%s/%s", backup.Namespace, backup.Name)).
 			WithField("BackupPhase", backup.Status.Phase).Debugf("Cleaning VolumeSnapshots.")
 
-		csi.DeleteReadyVolumeSnapshot(*vs, p.crClient, p.log)
+		csi.DeleteReadyVolumeSnapshot(ctx, *vs, p.crClient, p.log)
 		return item, nil, "", nil, nil
 	}
 
@@ -115,11 +117,9 @@ func (p *volumeSnapshotBackupItemAction) Execute(
 	p.log.Infof("Getting VolumesnapshotContent for Volumesnapshot %s/%s",
 		vs.Namespace, vs.Name)
 
-	ctx := context.TODO()
-
 	vsc, err := csi.GetVSCForVS(ctx, vs, p.crClient)
 	if err != nil {
-		csi.CleanupVolumeSnapshot(vs, p.crClient, p.log)
+		csi.CleanupVolumeSnapshot(ctx, vs, p.crClient, p.log)
 		return nil, nil, "", nil, errors.WithStack(err)
 	}
 
@@ -187,7 +187,7 @@ func (p *volumeSnapshotBackupItemAction) Execute(
 		)
 
 		if vscPatchError := p.crClient.Patch(
-			context.TODO(),
+			ctx,
 			vsc,
 			crclient.MergeFrom(originVSC),
 		); vscPatchError != nil {
@@ -203,7 +203,7 @@ func (p *volumeSnapshotBackupItemAction) Execute(
 	originVS := vs.DeepCopy()
 	kubeutil.AddAnnotations(&vs.ObjectMeta, annotations)
 	if err := p.crClient.Patch(
-		context.TODO(),
+		ctx,
 		vs,
 		crclient.MergeFrom(originVS),
 	); err != nil {
@@ -269,8 +269,8 @@ func (p *volumeSnapshotBackupItemAction) Progress(
 	}
 	var err error
 	if progress.Started, err = time.Parse(time.RFC3339, operationIDParts[2]); err != nil {
-		p.log.Errorf("error parsing operation ID's StartedTime",
-			"part into time %s: %s", operationID, err.Error())
+		p.log.Errorf("error parsing operation ID's StartedTime part into time %s: %s",
+			operationID, err.Error())
 		return progress, errors.WithStack(err)
 	}
 

pkg/backup/actions/csi/volumesnapshotcontent_action.go

@@ -107,8 +107,7 @@ func (p *volumeSnapshotContentBackupItemAction) Execute(
 	}
 
 	p.log.Infof(
-		"Returning from VolumeSnapshotContentBackupItemAction",
-		"with %d additionalItems to backup",
+		"Returning from VolumeSnapshotContentBackupItemAction with %d additionalItems to backup",
 		len(additionalItems),
 	)
 	return &unstructured.Unstructured{Object: snapContMap}, additionalItems, "", nil, nil

pkg/backup/backup.go

@@ -1263,21 +1263,12 @@ func buildFinalTarball(tr *tar.Reader, tw tarWriter, updateFiles map[string]File
 				return errors.WithStack(err)
 			}
 			delete(updateFiles, header.Name)
-			// skip over file contents from old tarball
-			_, err := io.ReadAll(tr)
-			if err != nil {
-				return errors.WithStack(err)
-			}
 		} else {
 			// Add original content to new tarball, as item wasn't updated
-			oldContents, err := io.ReadAll(tr)
-			if err != nil {
-				return errors.WithStack(err)
-			}
 			if err := tw.WriteHeader(header); err != nil {
 				return errors.WithStack(err)
 			}
-			if _, err := tw.Write(oldContents); err != nil {
+			if _, err := io.Copy(tw, tr); err != nil {
 				return errors.WithStack(err)
 			}
 		}

pkg/backup/backup_test.go

@@ -2931,7 +2931,6 @@ func (*fakeVolumeSnapshotter) DeleteSnapshot(snapshotID string) error {
 // looking at the backup request's VolumeSnapshots field. This test uses the fakeVolumeSnapshotter
 // struct in place of real volume snapshotters.
 func TestBackupWithSnapshots(t *testing.T) {
-	// TODO: add more verification for skippedPVTracker
 	itemBlockPool := StartItemBlockWorkerPool(t.Context(), 1, logrus.StandardLogger())
 	defer itemBlockPool.Stop()
 	tests := []struct {
@@ -2941,6 +2940,7 @@ func TestBackupWithSnapshots(t *testing.T) {
 		apiResources      []*test.APIResource
 		snapshotterGetter volumeSnapshotterGetter
 		want              []*volume.Snapshot
+		wantSkippedPVs    []SkippedPV
 	}{
 		{
 			name: "persistent volume with no zone annotation creates a snapshot",
@@ -2977,6 +2977,7 @@ func TestBackupWithSnapshots(t *testing.T) {
 					},
 				},
 			},
+			wantSkippedPVs: []SkippedPV{},
 		},
 		{
 			name: "persistent volume with deprecated zone annotation creates a snapshot",
@@ -2991,7 +2992,7 @@ func TestBackupWithSnapshots(t *testing.T) {
 			},
 			apiResources: []*test.APIResource{
 				test.PVs(
-					builder.ForPersistentVolume("pv-1").ObjectMeta(builder.WithLabels("failure-domain.beta.kubernetes.io/zone", "zone-1")).Result(),
+					builder.ForPersistentVolume("pv-1").ObjectMeta(builder.WithLabels(corev1api.LabelFailureDomainBetaZone, "zone-1")).Result(),
 				),
 			},
 			snapshotterGetter: map[string]vsv1.VolumeSnapshotter{
@@ -3014,6 +3015,7 @@ func TestBackupWithSnapshots(t *testing.T) {
 					},
 				},
 			},
+			wantSkippedPVs: []SkippedPV{},
 		},
 		{
 			name: "persistent volume with GA zone annotation creates a snapshot",
@@ -3028,7 +3030,7 @@ func TestBackupWithSnapshots(t *testing.T) {
 			},
 			apiResources: []*test.APIResource{
 				test.PVs(
-					builder.ForPersistentVolume("pv-1").ObjectMeta(builder.WithLabels("topology.kubernetes.io/zone", "zone-1")).Result(),
+					builder.ForPersistentVolume("pv-1").ObjectMeta(builder.WithLabels(corev1api.LabelTopologyZone, "zone-1")).Result(),
 				),
 			},
 			snapshotterGetter: map[string]vsv1.VolumeSnapshotter{
@@ -3051,6 +3053,7 @@ func TestBackupWithSnapshots(t *testing.T) {
 					},
 				},
 			},
+			wantSkippedPVs: []SkippedPV{},
 		},
 		{
 			name: "persistent volume with both GA and deprecated zone annotation creates a snapshot and should use the GA",
@@ -3065,7 +3068,7 @@ func TestBackupWithSnapshots(t *testing.T) {
 			},
 			apiResources: []*test.APIResource{
 				test.PVs(
-					builder.ForPersistentVolume("pv-1").ObjectMeta(builder.WithLabelsMap(map[string]string{"failure-domain.beta.kubernetes.io/zone": "zone-1-deprecated", "topology.kubernetes.io/zone": "zone-1-ga"})).Result(),
+					builder.ForPersistentVolume("pv-1").ObjectMeta(builder.WithLabelsMap(map[string]string{corev1api.LabelFailureDomainBetaZone: "zone-1-deprecated", corev1api.LabelTopologyZone: "zone-1-ga"})).Result(),
 				),
 			},
 			snapshotterGetter: map[string]vsv1.VolumeSnapshotter{
@@ -3088,6 +3091,7 @@ func TestBackupWithSnapshots(t *testing.T) {
 					},
 				},
 			},
+			wantSkippedPVs: []SkippedPV{},
 		},
 		{
 			name: "error returned from CreateSnapshot results in a failed snapshot",
@@ -3123,6 +3127,7 @@ func TestBackupWithSnapshots(t *testing.T) {
 					},
 				},
 			},
+			wantSkippedPVs: []SkippedPV{},
 		},
 		{
 			name: "backup with SnapshotVolumes=false does not create any snapshots",
@@ -3144,6 +3149,17 @@ func TestBackupWithSnapshots(t *testing.T) {
 				"default": new(fakeVolumeSnapshotter).WithVolume("pv-1", "vol-1", "", "type-1", 100, false),
 			},
 			want: nil,
+			wantSkippedPVs: []SkippedPV{
+				{
+					Name: "pv-1",
+					Reasons: []PVSkipReason{
+						{
+							Approach: volumeSnapshotApproach,
+							Reason:   "not satisfy the criteria for VolumePolicy or the legacy snapshot way",
+						},
+					},
+				},
+			},
 		},
 		{
 			name: "backup with no volume snapshot locations does not create any snapshots",
@@ -3162,6 +3178,17 @@ func TestBackupWithSnapshots(t *testing.T) {
 				"default": new(fakeVolumeSnapshotter).WithVolume("pv-1", "vol-1", "", "type-1", 100, false),
 			},
 			want: nil,
+			wantSkippedPVs: []SkippedPV{
+				{
+					Name: "pv-1",
+					Reasons: []PVSkipReason{
+						{
+							Approach: volumeSnapshotApproach,
+							Reason:   "no applicable volumesnapshotter found",
+						},
+					},
+				},
+			},
 		},
 		{
 			name: "backup with no volume snapshotters does not create any snapshots",
@@ -3181,6 +3208,17 @@ func TestBackupWithSnapshots(t *testing.T) {
 			},
 			snapshotterGetter: map[string]vsv1.VolumeSnapshotter{},
 			want:              nil,
+			wantSkippedPVs: []SkippedPV{
+				{
+					Name: "pv-1",
+					Reasons: []PVSkipReason{
+						{
+							Approach: volumeSnapshotApproach,
+							Reason:   "no applicable volumesnapshotter found",
+						},
+					},
+				},
+			},
 		},
 		{
 			name: "unsupported persistent volume type does not create any snapshots",
@@ -3202,6 +3240,17 @@ func TestBackupWithSnapshots(t *testing.T) {
 				"default": new(fakeVolumeSnapshotter),
 			},
 			want: nil,
+			wantSkippedPVs: []SkippedPV{
+				{
+					Name: "pv-1",
+					Reasons: []PVSkipReason{
+						{
+							Approach: volumeSnapshotApproach,
+							Reason:   "no applicable volumesnapshotter found",
+						},
+					},
+				},
+			},
 		},
 		{
 			name: "when there are multiple volumes, snapshot locations, and snapshotters, volumes are matched to the right snapshotters",
@@ -3255,6 +3304,7 @@ func TestBackupWithSnapshots(t *testing.T) {
 					},
 				},
 			},
+			wantSkippedPVs: []SkippedPV{},
 		},
 	}
 
@@ -3273,6 +3323,7 @@ func TestBackupWithSnapshots(t *testing.T) {
 			require.NoError(t, err)
 
 			assert.Equal(t, tc.want, tc.req.VolumeSnapshots.Get())
+			assert.Equal(t, tc.wantSkippedPVs, tc.req.SkippedPVTracker.Summary())
 		})
 	}
 }

pkg/backup/item_backupper.go

@@ -569,9 +569,9 @@ func (ib *itemBackupper) executeActions(
 // zoneLabel is the label that stores availability-zone info
 // on PVs
 const (
-	zoneLabelDeprecated = "failure-domain.beta.kubernetes.io/zone"
+	zoneLabelDeprecated = corev1api.LabelFailureDomainBetaZone
 	// this is reused for nodeAffinity requirements
-	zoneLabel = "topology.kubernetes.io/zone"
+	zoneLabel = corev1api.LabelTopologyZone
 
 	awsEbsCsiZoneKey = "topology.ebs.csi.aws.com/zone"
 	azureCsiZoneKey  = "topology.disk.csi.azure.com/zone"

pkg/backup/item_collector.go

@@ -346,7 +346,15 @@ func getOrderedResourcesForType(
 	if !ok || len(orderStr) == 0 {
 		return nil
 	}
-	orders := strings.Split(orderStr, ",")
+	parts := strings.Split(orderStr, ",")
+	orders := make([]string, 0, len(parts))
+	for _, part := range parts {
+		name := strings.TrimSpace(part)
+		if name == "" {
+			continue
+		}
+		orders = append(orders, name)
+	}
 	return orders
 }
 

pkg/backup/item_collector_test.go

@@ -445,3 +445,24 @@ func TestGetResourceItems(t *testing.T) {
 		})
 	}
 }
+
+func TestGetOrderedResourcesForTypeTrimsSpaces(t *testing.T) {
+	// Spaces after commas are common in CLI input and should not break ordering.
+	orders := getOrderedResourcesForType(map[string]string{
+		"pods": "ns1/pod2, ns1/pod1",
+	}, "pods")
+	require.Equal(t, []string{"ns1/pod2", "ns1/pod1"}, orders)
+
+	log := logrus.StandardLogger()
+	podResources := []*kubernetesResource{
+		{namespace: "ns1", name: "pod3"},
+		{namespace: "ns1", name: "pod1"},
+		{namespace: "ns1", name: "pod2"},
+	}
+	sorted := sortResourcesByOrder(log, podResources, orders)
+	require.Equal(t, []*kubernetesResource{
+		{namespace: "ns1", name: "pod2", orderedResource: true},
+		{namespace: "ns1", name: "pod1", orderedResource: true},
+		{namespace: "ns1", name: "pod3"},
+	}, sorted)
+}

pkg/backup/snapshots_test.go

@@ -0,0 +1,241 @@
+/*
+Copyright The Velero Contributors.
+
+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.
+*/
+
+package backup
+
+import (
+	"testing"
+
+	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v8/apis/volumesnapshot/v1"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	kbclient "sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/features"
+	velerotest "github.com/vmware-tanzu/velero/pkg/test"
+	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
+)
+
+func TestGetBackupCSIResources(t *testing.T) {
+	scheme := runtime.NewScheme()
+	require.NoError(t, snapshotv1api.AddToScheme(scheme))
+	require.NoError(t, velerov1api.AddToScheme(scheme))
+
+	tests := []struct {
+		name                 string
+		backup               *velerov1api.Backup
+		csiFeatureEnabled    bool
+		existingObjects      []kbclient.Object
+		wantSnapshots        int
+		wantSnapshotContents int
+		wantSnapshotClasses  int
+	}{
+		{
+			name: "SnapshotMoveData is true, skip CSI resources",
+			backup: &velerov1api.Backup{
+				ObjectMeta: metav1.ObjectMeta{Name: "test-backup"},
+				Spec: velerov1api.BackupSpec{
+					SnapshotMoveData: boolptr.True(),
+				},
+			},
+			csiFeatureEnabled: true,
+			existingObjects: []kbclient.Object{
+				&snapshotv1api.VolumeSnapshot{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "vs-1",
+						Namespace: "ns-1",
+						Labels: map[string]string{
+							velerov1api.BackupNameLabel: "test-backup",
+						},
+					},
+				},
+				&snapshotv1api.VolumeSnapshotContent{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "vsc-1",
+						Labels: map[string]string{
+							velerov1api.BackupNameLabel: "test-backup",
+						},
+					},
+					Spec: snapshotv1api.VolumeSnapshotContentSpec{
+						VolumeSnapshotClassName: func(s string) *string { return &s }("vsc-class-1"),
+					},
+				},
+				&snapshotv1api.VolumeSnapshotClass{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "vsc-class-1",
+					},
+				},
+			},
+			wantSnapshots:        0,
+			wantSnapshotContents: 0,
+			wantSnapshotClasses:  0,
+		},
+		{
+			name: "CSIFeatureFlag is false, skip CSI resources",
+			backup: &velerov1api.Backup{
+				ObjectMeta: metav1.ObjectMeta{Name: "test-backup"},
+				Spec: velerov1api.BackupSpec{
+					SnapshotMoveData: boolptr.False(),
+				},
+			},
+			csiFeatureEnabled: false,
+			existingObjects: []kbclient.Object{
+				&snapshotv1api.VolumeSnapshot{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "vs-1",
+						Namespace: "ns-1",
+						Labels: map[string]string{
+							velerov1api.BackupNameLabel: "test-backup",
+						},
+					},
+				},
+				&snapshotv1api.VolumeSnapshotContent{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "vsc-1",
+						Labels: map[string]string{
+							velerov1api.BackupNameLabel: "test-backup",
+						},
+					},
+					Spec: snapshotv1api.VolumeSnapshotContentSpec{
+						VolumeSnapshotClassName: func(s string) *string { return &s }("vsc-class-1"),
+					},
+				},
+				&snapshotv1api.VolumeSnapshotClass{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "vsc-class-1",
+					},
+				},
+			},
+			wantSnapshots:        0,
+			wantSnapshotContents: 0,
+			wantSnapshotClasses:  0,
+		},
+		{
+			name: "CSIFeatureFlag enabled, retrieve CSI resources",
+			backup: &velerov1api.Backup{
+				ObjectMeta: metav1.ObjectMeta{Name: "test-backup"},
+				Spec: velerov1api.BackupSpec{
+					SnapshotMoveData: boolptr.False(),
+				},
+			},
+			csiFeatureEnabled: true,
+			existingObjects: []kbclient.Object{
+				&snapshotv1api.VolumeSnapshot{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "vs-1",
+						Namespace: "ns-1",
+						Labels: map[string]string{
+							velerov1api.BackupNameLabel: "test-backup",
+						},
+					},
+				},
+				&snapshotv1api.VolumeSnapshotContent{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "vsc-1",
+						Labels: map[string]string{
+							velerov1api.BackupNameLabel: "test-backup",
+						},
+					},
+					Spec: snapshotv1api.VolumeSnapshotContentSpec{
+						VolumeSnapshotClassName: func(s string) *string { return &s }("vsc-class-1"),
+					},
+				},
+				&snapshotv1api.VolumeSnapshotClass{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "vsc-class-1",
+					},
+				},
+			},
+			wantSnapshots:        1,
+			wantSnapshotContents: 1,
+			wantSnapshotClasses:  1,
+		},
+		{
+			name: "CSIFeatureFlag enabled, multiple contents referencing same class",
+			backup: &velerov1api.Backup{
+				ObjectMeta: metav1.ObjectMeta{Name: "test-backup"},
+				Spec: velerov1api.BackupSpec{
+					SnapshotMoveData: boolptr.False(),
+				},
+			},
+			csiFeatureEnabled: true,
+			existingObjects: []kbclient.Object{
+				&snapshotv1api.VolumeSnapshotContent{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "vsc-1",
+						Labels: map[string]string{
+							velerov1api.BackupNameLabel: "test-backup",
+						},
+					},
+					Spec: snapshotv1api.VolumeSnapshotContentSpec{
+						VolumeSnapshotClassName: func(s string) *string { return &s }("vsc-class-1"),
+					},
+				},
+				&snapshotv1api.VolumeSnapshotContent{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "vsc-2",
+						Labels: map[string]string{
+							velerov1api.BackupNameLabel: "test-backup",
+						},
+					},
+					Spec: snapshotv1api.VolumeSnapshotContentSpec{
+						VolumeSnapshotClassName: func(s string) *string { return &s }("vsc-class-1"),
+					},
+				},
+				&snapshotv1api.VolumeSnapshotClass{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "vsc-class-1",
+					},
+				},
+			},
+			wantSnapshots:        0,
+			wantSnapshotContents: 2,
+			wantSnapshotClasses:  1,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			defer features.NewFeatureFlagSet()
+
+			if tc.csiFeatureEnabled {
+				features.Enable(velerov1api.CSIFeatureFlag)
+			} else {
+				features.Disable(velerov1api.CSIFeatureFlag)
+			}
+
+			client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(tc.existingObjects...).Build()
+			logger := velerotest.NewLogger()
+
+			snaps, contents, classes := GetBackupCSIResources(client, client, tc.backup, logger)
+
+			assert.Len(t, snaps, tc.wantSnapshots)
+			assert.Len(t, contents, tc.wantSnapshotContents)
+			assert.Len(t, classes, tc.wantSnapshotClasses)
+
+			// If we expect CSI resources to be pulled, ensure the attempts count was updated on the backup object
+			if tc.csiFeatureEnabled && !boolptr.IsSetToTrue(tc.backup.Spec.SnapshotMoveData) {
+				assert.Equal(t, tc.wantSnapshots, tc.backup.Status.CSIVolumeSnapshotsAttempted)
+			} else {
+				assert.Equal(t, 0, tc.backup.Status.CSIVolumeSnapshotsAttempted)
+			}
+		})
+	}
+}

pkg/builder/container_builder.go

@@ -18,10 +18,12 @@ package builder
 
 import (
 	"encoding/json"
+	"fmt"
 	"strings"
 
 	corev1api "k8s.io/api/core/v1"
 	apimachineryRuntime "k8s.io/apimachinery/pkg/runtime"
+	utilrand "k8s.io/apimachinery/pkg/util/rand"
 
 	"github.com/vmware-tanzu/velero/pkg/label"
 )
@@ -42,15 +44,22 @@ func ForContainer(name, image string) *ContainerBuilder {
 }
 
 // ForPluginContainer is a helper builder specifically for plugin init containers
-func ForPluginContainer(image string, pullPolicy corev1api.PullPolicy) *ContainerBuilder {
+func ForPluginContainer(image string, pullPolicy corev1api.PullPolicy, existingContainers []corev1api.Container) *ContainerBuilder {
 	volumeMount := ForVolumeMount("plugins", "/target").Result()
-	return ForContainer(getName(image), image).PullPolicy(pullPolicy).VolumeMounts(volumeMount)
+	return ForContainer(getName(image, existingContainers), image).PullPolicy(pullPolicy).VolumeMounts(volumeMount)
 }
 
 // getName returns the 'name' component of a docker image that includes the entire string
 // except the registry name, and transforms the combined string into a DNS-1123 compatible name
 // that fits within the 63-character limit for Kubernetes container names.
-func getName(image string) string {
+// It appends a random string if there is a collision with existing container names.
+func getName(image string, existingContainers []corev1api.Container) string {
+	// Convert existingContainers to a map for O(1) collision lookups
+	existingNames := make(map[string]bool, len(existingContainers))
+	for _, c := range existingContainers {
+		existingNames[c.Name] = true
+	}
+
 	slashIndex := strings.Index(image, "/")
 	slashCount := 0
 	if slashIndex >= 0 {
@@ -88,7 +97,20 @@ func getName(image string) string {
 	name := re.Replace(image[start:end])
 
 	// Ensure the name doesn't exceed Kubernetes container name length limit
-	return label.GetValidName(name)
+	name = label.GetValidName(name)
+
+	for existingNames[name] {
+		name = re.Replace(image[start:end])
+		if len(name) > 57 {
+			// Leave 6 characters for "-xxxxx" random string
+			name = name[:57]
+			name = strings.TrimSuffix(name, "-")
+		}
+		name = fmt.Sprintf("%s-%s", name, utilrand.String(5))
+		name = label.GetValidName(name)
+	}
+
+	return name
 }
 
 // Result returns the built Container.

pkg/builder/container_builder_test.go

@@ -16,16 +16,19 @@ limitations under the License.
 package builder
 
 import (
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	corev1api "k8s.io/api/core/v1"
 )
 
 func TestGetName(t *testing.T) {
 	tests := []struct {
-		name     string
-		image    string
-		expected string
+		name               string
+		image              string
+		existingContainers []corev1api.Container
+		expected           string
 	}{
 		{
 			name:     "image name with registry hostname and tag",
@@ -92,11 +95,25 @@ func TestGetName(t *testing.T) {
 			image:    "quay.io/vmware-tanzu/velero@sha256:a75f9e8c3ced3943515f249597be389f8233e1258d289b11184796edceaa7dab",
 			expected: "vmware-tanzu-velero",
 		},
+		{
+			name:  "duplicate plugin name",
+			image: "gcr.io/my-repo/my-image:latest",
+			existingContainers: []corev1api.Container{
+				{Name: "my-repo-my-image"},
+			},
+			expected: "my-repo-my-image-", // we will check it has the prefix
+		},
 	}
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			assert.Equal(t, test.expected, getName(test.image))
+			if test.name == "duplicate plugin name" {
+				result := getName(test.image, test.existingContainers)
+				assert.True(t, strings.HasPrefix(result, test.expected), "expected prefix %s in %s", test.expected, result)
+				assert.Len(t, result, len(test.expected)+5)
+			} else {
+				assert.Equal(t, test.expected, getName(test.image, test.existingContainers))
+			}
 		})
 	}
 }
@@ -117,7 +134,7 @@ func TestGetNameWithLongPaths(t *testing.T) {
 				// Should be exactly 63 characters (truncated with hash)
 				assert.Len(t, result, 63)
 				// Should be deterministic
-				result2 := getName("arohcpsvcdev.azurecr.io/redhat-user-workloads/ocp-art-tenant/oadp-hypershift-oadp-plugin-main@sha256:adb840bf3890b4904a8cdda1a74c82cf8d96c52eba9944ac10e795335d6fd450")
+				result2 := getName("arohcpsvcdev.azurecr.io/redhat-user-workloads/ocp-art-tenant/oadp-hypershift-oadp-plugin-main@sha256:adb840bf3890b4904a8cdda1a74c82cf8d96c52eba9944ac10e795335d6fd450", nil)
 				assert.Equal(t, result, result2)
 			},
 		},
@@ -142,7 +159,7 @@ func TestGetNameWithLongPaths(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			result := getName(test.image)
+			result := getName(test.image, nil)
 			test.validate(t, result)
 		})
 	}

pkg/cmd/cli/backup/create.go

@@ -416,8 +416,19 @@ func ParseOrderedResources(orderMapStr string) (map[string]string, error) {
 			return nil, fmt.Errorf("invalid OrderedResources '%s'", entry)
 		}
 		kind := strings.TrimSpace(kv[0])
-		order := strings.TrimSpace(kv[1])
-		orderedResources[kind] = order
+		orderParts := strings.Split(kv[1], ",")
+		cleaned := make([]string, 0, len(orderParts))
+		for _, part := range orderParts {
+			name := strings.TrimSpace(part)
+			if name == "" {
+				continue
+			}
+			cleaned = append(cleaned, name)
+		}
+		if kind == "" || len(cleaned) == 0 {
+			return nil, fmt.Errorf("invalid OrderedResources '%s'", entry)
+		}
+		orderedResources[kind] = strings.Join(cleaned, ",")
 	}
 	return orderedResources, nil
 }

pkg/cmd/cli/backup/create_test.go

@@ -234,6 +234,14 @@ func TestCreateOptions_OrderedResources(t *testing.T) {
 		"persistentvolumes": "pv1,pv2",
 	}
 	assert.Equal(t, expectedMixedResources, orderedResources)
+
+	// Spaces after commas in the resource list must be trimmed.
+	orderedResources, err = ParseOrderedResources("pods=ns1/p1, ns1/p2 ; persistentvolumeclaims= ns2/pvc1,  ns2/pvc2")
+	require.NoError(t, err)
+	assert.Equal(t, map[string]string{
+		"pods":                   "ns1/p1,ns1/p2",
+		"persistentvolumeclaims": "ns2/pvc1,ns2/pvc2",
+	}, orderedResources)
 }
 
 func TestCreateCommand(t *testing.T) {

pkg/cmd/cli/plugin/add.go

@@ -111,7 +111,7 @@ func NewAddCommand(f client.Factory) *cobra.Command {
 			}
 
 			// add the plugin as an init container
-			plugin := *builder.ForPluginContainer(args[0], corev1api.PullPolicy(imagePullPolicyFlag.String())).Result()
+			plugin := *builder.ForPluginContainer(args[0], corev1api.PullPolicy(imagePullPolicyFlag.String()), veleroDeploy.Spec.Template.Spec.InitContainers).Result()
 
 			veleroDeploy.Spec.Template.Spec.InitContainers = append(veleroDeploy.Spec.Template.Spec.InitContainers, plugin)
 

pkg/cmd/cli/uninstall/uninstall.go

@@ -57,13 +57,11 @@ var resToDelete = []kbclient.ObjectList{}
 
 // uninstallOptions collects all the options for uninstalling Velero from a Kubernetes cluster.
 type uninstallOptions struct {
-	wait  bool // deprecated
 	force bool
 }
 
 // BindFlags adds command line values to the options struct.
 func (o *uninstallOptions) BindFlags(flags *pflag.FlagSet) {
-	flags.BoolVar(&o.wait, "wait", o.wait, "Wait for Velero uninstall to be ready. Optional. Deprecated.")
 	flags.BoolVar(&o.force, "force", o.force, "Forces the Velero uninstall. Optional.")
 }
 
@@ -81,10 +79,6 @@ Use '--force' to skip the prompt confirming if you want to uninstall Velero.
 		`,
 		Example: ` # velero uninstall --namespace staging`,
 		Run: func(c *cobra.Command, args []string) {
-			if o.wait {
-				fmt.Println("Warning: the \"--wait\" option is deprecated and will be removed in a future release. The uninstall command always waits for the uninstall to complete.")
-			}
-
 			// Confirm if not asked to force-skip confirmation
 			if !o.force {
 				fmt.Println("You are about to uninstall Velero.")

pkg/cmd/server/config/config.go

@@ -28,6 +28,11 @@ const (
 	defaultPodVolumeOperationTimeout  = 240 * time.Minute
 	defaultResourceTerminatingTimeout = 10 * time.Minute
 
+	// DefaultResourceTimeout is the default for --resource-timeout. It matches
+	// defaultResourceTerminatingTimeout so controller fallbacks stay aligned with
+	// server defaults (see pkg/cmd/server/config/config.go).
+	DefaultResourceTimeout = defaultResourceTerminatingTimeout
+
 	// server's client default qps and burst
 	defaultClientQPS      float32 = 100.0
 	defaultClientBurst    int     = 100
@@ -41,7 +46,7 @@ const (
 	defaultCSISnapshotTimeout   = 10 * time.Minute
 	defaultItemOperationTimeout = 4 * time.Hour
 
-	resourceTimeout = 10 * time.Minute
+	resourceTimeout = defaultResourceTerminatingTimeout
 
 	defaultMaxConcurrentK8SConnections = 30
 	defaultDisableInformerCache        = false
@@ -183,6 +188,7 @@ type Config struct {
 	ConcurrentBackups                   int
 	GlobalBackupVolumePoliciesConfigMap string
 	DefaultResourceModifierConfigMap    string
+	MaxBackupExtractionSize             int
 }
 
 func GetDefaultConfig() *Config {
@@ -289,4 +295,10 @@ func (c *Config) BindFlags(flags *pflag.FlagSet) {
 		c.DefaultResourceModifierConfigMap,
 		"The name of a ConfigMap in the Velero namespace containing default resource modifier rules applied to all restores. Ignored when a per-restore resource modifier is specified.",
 	)
+	flags.IntVar(
+		&c.MaxBackupExtractionSize,
+		"max-backup-extraction-size",
+		c.MaxBackupExtractionSize,
+		"Maximum size of a backup extraction in megabytes. If not set, default value (16GB) will be used.",
+	)
 }

pkg/cmd/server/server.go

@@ -61,6 +61,7 @@ import (
 	"github.com/vmware-tanzu/velero/internal/storage"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	velerov2alpha1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
+	"github.com/vmware-tanzu/velero/pkg/archive"
 	"github.com/vmware-tanzu/velero/pkg/backup"
 	"github.com/vmware-tanzu/velero/pkg/buildinfo"
 	"github.com/vmware-tanzu/velero/pkg/client"
@@ -935,6 +936,11 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string
 		}
 	}
 
+	if s.config.MaxBackupExtractionSize > 0 {
+		s.logger.Infof("Setting backup data extraction cap as %v MB", s.config.MaxBackupExtractionSize)
+		archive.SetMaxExtractionSize(int64(s.config.MaxBackupExtractionSize) * 1024 * 1024)
+	}
+
 	s.logger.Info("Server starting...")
 
 	if err := s.mgr.Start(s.ctx); err != nil {

pkg/cmd/util/downloadrequest/downloadrequest.go

@@ -40,6 +40,12 @@ import (
 // not found
 var ErrNotFound = errors.New("file not found")
 var ErrDownloadRequestDownloadURLTimeout = errors.New("download request download url timeout, check velero server logs for errors. backup storage location may not be available")
+var unzipLimit int64 = 1024 * 1024 * 1024 // 1GB limit
+
+// ErrDownloadRequestFailed is returned when the server refused the request and gave no
+// reason. The controller sets a message in every path that fails today, so this is a
+// fallback rather than the usual case.
+var ErrDownloadRequestFailed = errors.New("download request failed, check velero server logs for errors")
 
 func Stream(
 	ctx context.Context,
@@ -114,6 +120,16 @@ func getDownloadURL(
 			if updated.Status.DownloadURL != "" {
 				return updated.Status.DownloadURL, nil
 			}
+
+			// Failed is terminal. Waiting for a URL that will never be signed would end in
+			// ErrDownloadRequestDownloadURLTimeout, which blames the storage location for
+			// something the status already explains.
+			if updated.Status.Phase == veleroV1api.DownloadRequestPhaseFailed {
+				if updated.Status.Message != "" {
+					return "", errors.New(updated.Status.Message)
+				}
+				return "", ErrDownloadRequestFailed
+			}
 		}
 	}
 }
@@ -202,17 +218,35 @@ func download(
 		return errors.Errorf("request failed: %v", string(body))
 	}
 
-	reader := resp.Body
+	var r io.Reader = resp.Body
+	var gzipReader *gzip.Reader
 	if kind != veleroV1api.DownloadTargetKindBackupContents {
 		// need to decompress logs
-		gzipReader, err := gzip.NewReader(resp.Body)
+		var err error
+		gzipReader, err = gzip.NewReader(resp.Body)
 		if err != nil {
 			return err
 		}
 		defer gzipReader.Close()
-		reader = gzipReader
+
+		r = io.LimitReader(gzipReader, unzipLimit)
 	}
 
-	_, err = io.Copy(w, reader)
-	return err
+	_, err = io.Copy(w, r)
+	if err != nil {
+		return err
+	}
+
+	if gzipReader != nil {
+		var buf [1]byte
+		n, err := gzipReader.Read(buf[:])
+		if n > 0 || err == nil {
+			return errors.Errorf("decompressed data exceeds the limit")
+		}
+		if err != io.EOF {
+			return err
+		}
+	}
+
+	return nil
 }

pkg/cmd/util/downloadrequest/downloadrequest_test.go

@@ -463,6 +463,7 @@ func TestDownload(t *testing.T) {
 		expectedContent       string
 		expectedError         bool
 		errorType             error
+		expectedErrMsg        string
 	}{
 		{
 			name: "successful download with gzip for logs",
@@ -474,6 +475,16 @@ func TestDownload(t *testing.T) {
 			expectedContent: testContent,
 			expectedError:   false,
 		},
+		{
+			name: "error decompressed data exceeds the limit",
+			serverHandler: func(w http.ResponseWriter, r *http.Request) {
+				w.WriteHeader(http.StatusOK)
+				w.Write(compressedContent.Bytes())
+			},
+			target:         velerov1api.DownloadTargetKindBackupLog,
+			expectedError:  true,
+			expectedErrMsg: "decompressed data exceeds the limit",
+		},
 		{
 			name: "successful download without gzip for backup contents",
 			serverHandler: func(w http.ResponseWriter, r *http.Request) {
@@ -506,6 +517,12 @@ func TestDownload(t *testing.T) {
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
+			originalLimit := unzipLimit
+			if tc.expectedErrMsg == "decompressed data exceeds the limit" {
+				unzipLimit = 10
+			}
+			defer func() { unzipLimit = originalLimit }()
+
 			server := httptest.NewServer(tc.serverHandler)
 			defer server.Close()
 
@@ -525,6 +542,9 @@ func TestDownload(t *testing.T) {
 				if tc.errorType != nil {
 					assert.Equal(t, tc.errorType, err)
 				}
+				if tc.expectedErrMsg != "" {
+					assert.Contains(t, err.Error(), tc.expectedErrMsg)
+				}
 			} else {
 				require.NoError(t, err)
 				assert.Equal(t, tc.expectedContent, buf.String())

pkg/cmd/util/output/backup_printer.go

@@ -90,8 +90,12 @@ func printBackup(backup *velerov1api.Backup) []metav1.TableRow {
 	if backup.Status.Expiration != nil {
 		expiration = backup.Status.Expiration.Time
 	}
-	if expiration.IsZero() && backup.Spec.TTL.Duration > 0 {
-		expiration = backup.CreationTimestamp.Add(backup.Spec.TTL.Duration)
+	// Only estimate expiration from TTL after the backup has started. Backups
+	// stalled in New have no Status.Expiration yet; using CreationTimestamp
+	// would incorrectly show them as already expired (issue #3555).
+	if expiration.IsZero() && backup.Spec.TTL.Duration > 0 &&
+		backup.Status.StartTimestamp != nil && !backup.Status.StartTimestamp.Time.IsZero() {
+		expiration = backup.Status.StartTimestamp.Time.Add(backup.Spec.TTL.Duration)
 	}
 
 	status := string(backup.Status.Phase)

pkg/cmd/util/output/printer_timestamp_test.go

@@ -76,6 +76,26 @@ func TestPrintBackupWithoutStartTimestamp(t *testing.T) {
 	assert.Equal(t, string(velerov1api.BackupPhaseFailedValidation), rows[0].Cells[1])
 }
 
+func TestPrintBackupExpiresForStalledNewBackup(t *testing.T) {
+	created := metav1.NewTime(time.Now().Add(-20 * 24 * time.Hour))
+	backup := &velerov1api.Backup{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:              "clusterstate-20210128123759",
+			CreationTimestamp: created,
+		},
+		Spec: velerov1api.BackupSpec{
+			TTL: metav1.Duration{Duration: 10 * 24 * time.Hour},
+		},
+		Status: velerov1api.BackupStatus{
+			Phase: velerov1api.BackupPhaseNew,
+		},
+	}
+
+	rows := printBackup(backup)
+	require.Len(t, rows, 1)
+	assert.Equal(t, "n/a", rows[0].Cells[5], "stalled New backup should not show expiration in the past")
+}
+
 func TestPrintBackupWithStartTimestamp(t *testing.T) {
 	started := metav1.NewTime(time.Date(2026, 8, 8, 21, 6, 28, 0, time.UTC))
 	backup := &velerov1api.Backup{

pkg/controller/backup_deletion_controller.go

@@ -321,6 +321,10 @@ func (r *backupDeletionReconciler) Reconcile(ctx context.Context, req ctrl.Reque
 					volumeSnapshotters[snapshot.Spec.Location] = volumeSnapshotter
 				}
 
+				if snapshot.Status.ProviderSnapshotID == "" {
+					log.WithField("volumeSnapshot", snapshot.Spec.PersistentVolumeName).Warn("Skipping snapshot deletion: empty ProviderSnapshotID")
+					continue
+				}
 				if err := volumeSnapshotter.DeleteSnapshot(snapshot.Status.ProviderSnapshotID); err != nil {
 					errs = append(errs, errors.Wrapf(err, "error deleting snapshot %s", snapshot.Status.ProviderSnapshotID).Error())
 				}
@@ -531,7 +535,7 @@ func (r *backupDeletionReconciler) deleteCSIVolumeSnapshotsIfAny(ctx context.Con
 	}
 	for _, item := range vsList.Items {
 		vs := item
-		csi.CleanupVolumeSnapshot(&vs, r.Client, log)
+		csi.CleanupVolumeSnapshot(ctx, &vs, r.Client, log)
 	}
 }
 

pkg/controller/backup_deletion_controller_test.go

@@ -397,6 +397,74 @@ func TestBackupDeletionControllerReconcile(t *testing.T) {
 		// Make sure snapshot was deleted
 		assert.Equal(t, 0, td.volumeSnapshotter.SnapshotsTaken.Len())
 	})
+	t.Run("empty ProviderSnapshotID skips DeleteSnapshot call", func(t *testing.T) {
+		input := defaultTestDbr()
+
+		backup := builder.ForBackup(velerov1api.DefaultNamespace, input.Spec.BackupName).Result()
+		backup.UID = "uid"
+		backup.Spec.StorageLocation = "primary"
+
+		restore1 := builder.ForRestore(backup.Namespace, "restore-1").
+			Phase(velerov1api.RestorePhaseCompleted).
+			Backup(backup.Name).
+			Result()
+
+		location := &velerov1api.BackupStorageLocation{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: backup.Namespace,
+				Name:      "primary",
+			},
+			Spec: velerov1api.BackupStorageLocationSpec{
+				Provider: "objStoreProvider",
+				StorageType: velerov1api.StorageType{
+					ObjectStorage: &velerov1api.ObjectStorageLocation{
+						Bucket: "bucket",
+					},
+				},
+			},
+			Status: velerov1api.BackupStorageLocationStatus{
+				Phase: velerov1api.BackupStorageLocationPhaseAvailable,
+			},
+		}
+
+		snapshotLocation := &velerov1api.VolumeSnapshotLocation{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: backup.Namespace,
+				Name:      "vsl-1",
+			},
+			Spec: velerov1api.VolumeSnapshotLocationSpec{
+				Provider: "provider-1",
+			},
+		}
+		td := setupBackupDeletionControllerTest(t, input, backup, restore1, location, snapshotLocation)
+
+		snapshots := []*volume.Snapshot{
+			{
+				Spec: volume.SnapshotSpec{
+					Location:             "vsl-1",
+					PersistentVolumeName: "pv-1",
+				},
+				Status: volume.SnapshotStatus{
+					ProviderSnapshotID: "",
+				},
+			},
+		}
+
+		pluginManager := &pluginmocks.Manager{}
+		pluginManager.On("GetVolumeSnapshotter", "provider-1").Return(td.volumeSnapshotter, nil)
+		pluginManager.On("GetDeleteItemActions").Return(nil, nil)
+		pluginManager.On("CleanupClients")
+		td.controller.newPluginManager = func(logrus.FieldLogger) clientmgmt.Manager { return pluginManager }
+
+		td.backupStore.On("GetBackupVolumeSnapshots", input.Spec.BackupName).Return(snapshots, nil)
+		td.backupStore.On("GetBackupContents", input.Spec.BackupName).Return(io.NopCloser(bytes.NewReader([]byte("hello world"))), nil)
+		td.backupStore.On("DeleteBackup", input.Spec.BackupName).Return(nil)
+
+		_, err := td.controller.Reconcile(t.Context(), td.req)
+		require.NoError(t, err)
+
+		td.backupStore.AssertCalled(t, "DeleteBackup", input.Spec.BackupName)
+	})
 	t.Run("full delete, no errors, with backup name greater than 63 chars", func(t *testing.T) {
 		backup := defaultBackup().
 			ObjectMeta(

pkg/controller/backup_sync_controller.go

@@ -164,17 +164,37 @@ func (b *backupSyncReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 			continue
 		}
 
-		if backup.Status.Phase == velerov1api.BackupPhaseWaitingForPluginOperations ||
-			backup.Status.Phase == velerov1api.BackupPhaseWaitingForPluginOperationsPartiallyFailed ||
-			backup.Status.Phase == velerov1api.BackupPhaseFinalizing ||
-			backup.Status.Phase == velerov1api.BackupPhaseFinalizingPartiallyFailed {
+		// Only sync backup metadata that has reached a phase Velero itself writes to
+		// object storage. Anything else (including an empty or New phase) would be
+		// created in the cluster as a backup that still looks pending, which the backup
+		// queue controller would then pick up and run as if it were a newly requested
+		// backup.
+		switch backup.Status.Phase {
+		case velerov1api.BackupPhaseCompleted,
+			velerov1api.BackupPhasePartiallyFailed,
+			velerov1api.BackupPhaseFailed:
+			// finished backups are synced as-is
+		case velerov1api.BackupPhaseWaitingForPluginOperations,
+			velerov1api.BackupPhaseWaitingForPluginOperationsPartiallyFailed,
+			velerov1api.BackupPhaseFinalizing,
+			velerov1api.BackupPhaseFinalizingPartiallyFailed:
 			if backup.Status.Expiration == nil || backup.Status.Expiration.After(time.Now()) {
 				log.Debugf("Skipping non-expired incomplete backup %v", backup.Name)
 				continue
 			}
 			log.Debugf("%v Backup is past expiration, syncing for garbage collection", backup.Status.Phase)
 			backup.Status.Phase = velerov1api.BackupPhasePartiallyFailed
+		default:
+			log.Infof("Skipping backup %v, phase %q in the backup store is not a phase that can be synced", backup.Name, backup.Status.Phase)
+			continue
 		}
+
+		// A synced backup is a record of a backup that already ran somewhere else, not
+		// a backup to run here. Hooks are only read while a backup is being executed,
+		// so they have no consumer for a synced backup and are dropped rather than
+		// stored as an executable payload.
+		backup.Spec.Hooks = velerov1api.BackupHooks{}
+
 		backup.Namespace = b.namespace
 		backup.ResourceVersion = ""
 

pkg/controller/backup_sync_controller_test.go

@@ -204,10 +204,10 @@ var _ = Describe("Backup Sync Reconciler", func() {
 				location:  defaultLocation("ns-1"),
 				cloudBackups: []*cloudBackupData{
 					{
-						backup: builder.ForBackup("ns-1", "backup-1").Result(),
+						backup: builder.ForBackup("ns-1", "backup-1").Phase(velerov1api.BackupPhaseCompleted).Result(),
 					},
 					{
-						backup: builder.ForBackup("ns-1", "backup-2").Result(),
+						backup: builder.ForBackup("ns-1", "backup-2").Phase(velerov1api.BackupPhaseCompleted).Result(),
 					},
 				},
 			},
@@ -309,10 +309,10 @@ var _ = Describe("Backup Sync Reconciler", func() {
 				location:  defaultLocation("velero"),
 				cloudBackups: []*cloudBackupData{
 					{
-						backup: builder.ForBackup("ns-1", "backup-1").Result(),
+						backup: builder.ForBackup("ns-1", "backup-1").Phase(velerov1api.BackupPhaseCompleted).Result(),
 					},
 					{
-						backup: builder.ForBackup("ns-1", "backup-2").Result(),
+						backup: builder.ForBackup("ns-1", "backup-2").Phase(velerov1api.BackupPhaseCompleted).Result(),
 					},
 				},
 			},
@@ -322,10 +322,10 @@ var _ = Describe("Backup Sync Reconciler", func() {
 				location:  defaultLocation("ns-1"),
 				cloudBackups: []*cloudBackupData{
 					{
-						backup: builder.ForBackup("ns-1", "backup-1").Result(),
+						backup: builder.ForBackup("ns-1", "backup-1").Phase(velerov1api.BackupPhaseCompleted).Result(),
 					},
 					{
-						backup: builder.ForBackup("ns-1", "backup-2").Result(),
+						backup: builder.ForBackup("ns-1", "backup-2").Phase(velerov1api.BackupPhaseCompleted).Result(),
 					},
 				},
 				existingBackups: []*velerov1api.Backup{
@@ -341,7 +341,7 @@ var _ = Describe("Backup Sync Reconciler", func() {
 				location:  defaultLocation("ns-1"),
 				cloudBackups: []*cloudBackupData{
 					{
-						backup: builder.ForBackup("ns-1", "backup-1").Result(),
+						backup: builder.ForBackup("ns-1", "backup-1").Phase(velerov1api.BackupPhaseCompleted).Result(),
 					},
 				},
 				existingBackups: []*velerov1api.Backup{
@@ -356,10 +356,10 @@ var _ = Describe("Backup Sync Reconciler", func() {
 				location:  defaultLocation("ns-1"),
 				cloudBackups: []*cloudBackupData{
 					{
-						backup: builder.ForBackup("ns-1", "backup-1").StorageLocation("foo").ObjectMeta(builder.WithLabels(velerov1api.StorageLocationLabel, "foo")).Result(),
+						backup: builder.ForBackup("ns-1", "backup-1").StorageLocation("foo").ObjectMeta(builder.WithLabels(velerov1api.StorageLocationLabel, "foo")).Phase(velerov1api.BackupPhaseCompleted).Result(),
 					},
 					{
-						backup: builder.ForBackup("ns-1", "backup-2").Result(),
+						backup: builder.ForBackup("ns-1", "backup-2").Phase(velerov1api.BackupPhaseCompleted).Result(),
 					},
 				},
 			},
@@ -370,10 +370,10 @@ var _ = Describe("Backup Sync Reconciler", func() {
 				longLocationNameEnabled: true,
 				cloudBackups: []*cloudBackupData{
 					{
-						backup: builder.ForBackup("ns-1", "backup-1").StorageLocation("foo").ObjectMeta(builder.WithLabels(velerov1api.StorageLocationLabel, "foo")).Result(),
+						backup: builder.ForBackup("ns-1", "backup-1").StorageLocation("foo").ObjectMeta(builder.WithLabels(velerov1api.StorageLocationLabel, "foo")).Phase(velerov1api.BackupPhaseCompleted).Result(),
 					},
 					{
-						backup: builder.ForBackup("ns-1", "backup-2").Result(),
+						backup: builder.ForBackup("ns-1", "backup-2").Phase(velerov1api.BackupPhaseCompleted).Result(),
 					},
 				},
 			},
@@ -383,13 +383,13 @@ var _ = Describe("Backup Sync Reconciler", func() {
 				location:  defaultLocation("ns-1"),
 				cloudBackups: []*cloudBackupData{
 					{
-						backup: builder.ForBackup("ns-1", "backup-1").Result(),
+						backup: builder.ForBackup("ns-1", "backup-1").Phase(velerov1api.BackupPhaseCompleted).Result(),
 						podVolumeBackups: []*velerov1api.PodVolumeBackup{
 							builder.ForPodVolumeBackup("ns-1", "pvb-1").Result(),
 						},
 					},
 					{
-						backup: builder.ForBackup("ns-1", "backup-2").Result(),
+						backup: builder.ForBackup("ns-1", "backup-2").Phase(velerov1api.BackupPhaseCompleted).Result(),
 						podVolumeBackups: []*velerov1api.PodVolumeBackup{
 							builder.ForPodVolumeBackup("ns-1", "pvb-2").Result(),
 						},
@@ -402,13 +402,13 @@ var _ = Describe("Backup Sync Reconciler", func() {
 				location:  defaultLocation("ns-1"),
 				cloudBackups: []*cloudBackupData{
 					{
-						backup: builder.ForBackup("ns-1", "backup-1").Result(),
+						backup: builder.ForBackup("ns-1", "backup-1").Phase(velerov1api.BackupPhaseCompleted).Result(),
 						podVolumeBackups: []*velerov1api.PodVolumeBackup{
 							builder.ForPodVolumeBackup("ns-1", "pvb-1").Result(),
 						},
 					},
 					{
-						backup: builder.ForBackup("ns-1", "backup-2").Result(),
+						backup: builder.ForBackup("ns-1", "backup-2").Phase(velerov1api.BackupPhaseCompleted).Result(),
 						podVolumeBackups: []*velerov1api.PodVolumeBackup{
 							builder.ForPodVolumeBackup("ns-1", "pvb-3").Result(),
 						},
@@ -557,6 +557,184 @@ var _ = Describe("Backup Sync Reconciler", func() {
 		}
 	})
 
+	It("Test synced backups are never picked up by the backup queue controller", func() {
+		fakeClock := testclocks.NewFakeClock(time.Now())
+		hooks := velerov1api.BackupHooks{
+			Resources: []velerov1api.BackupResourceHookSpec{
+				{
+					Name: "hook-1",
+					PreHooks: []velerov1api.BackupResourceHook{
+						{
+							Exec: &velerov1api.ExecHook{
+								Container: "container-1",
+								Command:   []string{"/bin/sh", "-c", "echo hello"},
+							},
+						},
+					},
+				},
+			},
+		}
+
+		tests := []struct {
+			name         string
+			cloudBackup  *velerov1api.Backup
+			expectSynced bool
+			// phase expected in the cluster after the sync and queue reconciles have run.
+			// only checked when expectSynced is true.
+			expectPhase velerov1api.BackupPhase
+		}{
+			{
+				name:         "backup metadata with an empty phase is not synced",
+				cloudBackup:  builder.ForBackup("ns-1", "backup-1").Hooks(hooks).Result(),
+				expectSynced: false,
+			},
+			{
+				name:         "backup metadata in phase New is not synced",
+				cloudBackup:  builder.ForBackup("ns-1", "backup-1").Phase(velerov1api.BackupPhaseNew).Hooks(hooks).Result(),
+				expectSynced: false,
+			},
+			{
+				name:         "backup metadata in phase Queued is not synced",
+				cloudBackup:  builder.ForBackup("ns-1", "backup-1").Phase(velerov1api.BackupPhaseQueued).Hooks(hooks).Result(),
+				expectSynced: false,
+			},
+			{
+				name:         "backup metadata in phase ReadyToStart is not synced",
+				cloudBackup:  builder.ForBackup("ns-1", "backup-1").Phase(velerov1api.BackupPhaseReadyToStart).Hooks(hooks).Result(),
+				expectSynced: false,
+			},
+			{
+				name:         "backup metadata in phase InProgress is not synced",
+				cloudBackup:  builder.ForBackup("ns-1", "backup-1").Phase(velerov1api.BackupPhaseInProgress).Hooks(hooks).Result(),
+				expectSynced: false,
+			},
+			{
+				name:         "backup metadata in phase Deleting is not synced",
+				cloudBackup:  builder.ForBackup("ns-1", "backup-1").Phase(velerov1api.BackupPhaseDeleting).Hooks(hooks).Result(),
+				expectSynced: false,
+			},
+			{
+				name:         "backup metadata in phase Completed is synced and stays Completed",
+				cloudBackup:  builder.ForBackup("ns-1", "backup-1").Phase(velerov1api.BackupPhaseCompleted).Hooks(hooks).Result(),
+				expectSynced: true,
+				expectPhase:  velerov1api.BackupPhaseCompleted,
+			},
+			{
+				name:         "backup metadata in phase PartiallyFailed is synced and stays PartiallyFailed",
+				cloudBackup:  builder.ForBackup("ns-1", "backup-1").Phase(velerov1api.BackupPhasePartiallyFailed).Result(),
+				expectSynced: true,
+				expectPhase:  velerov1api.BackupPhasePartiallyFailed,
+			},
+			{
+				name:         "backup metadata in phase Failed is synced and stays Failed",
+				cloudBackup:  builder.ForBackup("ns-1", "backup-1").Phase(velerov1api.BackupPhaseFailed).Result(),
+				expectSynced: true,
+				expectPhase:  velerov1api.BackupPhaseFailed,
+			},
+			{
+				name: "non-expired backup waiting for plugin operations is not synced",
+				cloudBackup: builder.ForBackup("ns-1", "backup-1").
+					Phase(velerov1api.BackupPhaseWaitingForPluginOperations).
+					Expiration(fakeClock.Now().Add(time.Hour)).Result(),
+				expectSynced: false,
+			},
+			{
+				name: "expired backup waiting for plugin operations is synced as PartiallyFailed",
+				cloudBackup: builder.ForBackup("ns-1", "backup-1").
+					Phase(velerov1api.BackupPhaseWaitingForPluginOperations).
+					Expiration(fakeClock.Now().Add(-time.Hour)).Result(),
+				expectSynced: true,
+				expectPhase:  velerov1api.BackupPhasePartiallyFailed,
+			},
+			{
+				name: "expired backup waiting for plugin operations partially failed is synced as PartiallyFailed",
+				cloudBackup: builder.ForBackup("ns-1", "backup-1").
+					Phase(velerov1api.BackupPhaseWaitingForPluginOperationsPartiallyFailed).
+					Expiration(fakeClock.Now().Add(-time.Hour)).Result(),
+				expectSynced: true,
+				expectPhase:  velerov1api.BackupPhasePartiallyFailed,
+			},
+			{
+				name: "expired finalizing backup is synced as PartiallyFailed",
+				cloudBackup: builder.ForBackup("ns-1", "backup-1").
+					Phase(velerov1api.BackupPhaseFinalizing).
+					Expiration(fakeClock.Now().Add(-time.Hour)).Result(),
+				expectSynced: true,
+				expectPhase:  velerov1api.BackupPhasePartiallyFailed,
+			},
+			{
+				name: "expired finalizing partially failed backup is synced as PartiallyFailed",
+				cloudBackup: builder.ForBackup("ns-1", "backup-1").
+					Phase(velerov1api.BackupPhaseFinalizingPartiallyFailed).
+					Expiration(fakeClock.Now().Add(-time.Hour)).Result(),
+				expectSynced: true,
+				expectPhase:  velerov1api.BackupPhasePartiallyFailed,
+			},
+		}
+
+		queueScheme := runtime.NewScheme()
+		Expect(velerov1api.AddToScheme(queueScheme)).ShouldNot(HaveOccurred())
+
+		for _, test := range tests {
+			var (
+				client        = ctrlfake.NewClientBuilder().Build()
+				pluginManager = &pluginmocks.Manager{}
+				backupStores  = make(map[string]*persistencemocks.BackupStore)
+				location      = defaultLocation("ns-1")
+			)
+
+			pluginManager.On("CleanupClients").Return(nil)
+			syncReconciler := backupSyncReconciler{
+				client:                  client,
+				namespace:               "ns-1",
+				defaultBackupSyncPeriod: time.Second * 10,
+				newPluginManager:        func(logrus.FieldLogger) clientmgmt.Manager { return pluginManager },
+				backupStoreGetter:       NewFakeObjectBackupStoreGetter(backupStores),
+				logger:                  velerotest.NewLogger(),
+			}
+
+			Expect(client.Create(ctx, location)).ShouldNot(HaveOccurred(), test.name)
+			backupStore := &persistencemocks.BackupStore{}
+			backupStores[location.Name] = backupStore
+			backupStore.On("ListBackups").Return([]string{test.cloudBackup.Name}, nil)
+			backupStore.On("BackupExists", "bucket-1", test.cloudBackup.Name).Return(true, nil)
+			backupStore.On("GetBackupMetadata", test.cloudBackup.Name).Return(test.cloudBackup, nil)
+			backupStore.On("GetPodVolumeBackups", test.cloudBackup.Name).Return(nil, nil)
+
+			_, err := syncReconciler.Reconcile(ctx, ctrl.Request{
+				NamespacedName: types.NamespacedName{Namespace: location.Namespace, Name: location.Name},
+			})
+			Expect(err).ShouldNot(HaveOccurred(), test.name)
+
+			backupKey := types.NamespacedName{Namespace: "ns-1", Name: test.cloudBackup.Name}
+			synced := &velerov1api.Backup{}
+			err = client.Get(ctx, backupKey, synced)
+
+			if !test.expectSynced {
+				Expect(apierrors.IsNotFound(err)).To(BeTrue(), test.name)
+				continue
+			}
+			Expect(err).ShouldNot(HaveOccurred(), test.name)
+
+			// Reconcile the synced backup with the queue controller twice: the first
+			// reconcile would move a New/empty-phase backup to Queued, the second one
+			// would move it on to ReadyToStart, which is what hands it to the backup
+			// controller for execution.
+			queueReconciler := NewBackupQueueReconciler(client, queueScheme, velerotest.NewLogger(), 1, NewBackupTracker())
+			for range 2 {
+				_, err = queueReconciler.Reconcile(ctx, ctrl.Request{NamespacedName: backupKey})
+				Expect(err).ShouldNot(HaveOccurred(), test.name)
+			}
+
+			after := &velerov1api.Backup{}
+			Expect(client.Get(ctx, backupKey, after)).ShouldNot(HaveOccurred(), test.name)
+			Expect(after.Status.Phase).To(BeEquivalentTo(test.expectPhase), test.name)
+			// Hooks are dropped on sync, so the stored metadata cannot carry a payload
+			// that a later code path could execute.
+			Expect(after.Spec.Hooks.Resources).To(BeEmpty(), test.name)
+		}
+	})
+
 	It("Test deleting orphaned backups.", func() {
 		longLabelName := "the-really-long-location-name-that-is-much-more-than-63-characters"
 

pkg/controller/data_download_controller.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/cockroachdb/errors"
@@ -74,7 +75,7 @@ type DataDownloadReconciler struct {
 	podResources          corev1api.ResourceRequirements
 	preparingTimeout      time.Duration
 	metrics               *metrics.ServerMetrics
-	cancelledDataDownload map[string]time.Time
+	cancelledDataDownload sync.Map
 	dataMovePriorityClass string
 	repoConfigMgr         repository.ConfigManager
 	podLabels             map[string]string
@@ -118,7 +119,6 @@ func NewDataDownloadReconciler(
 		podResources:          podResources,
 		preparingTimeout:      preparingTimeout,
 		metrics:               metrics,
-		cancelledDataDownload: make(map[string]time.Time),
 		dataMovePriorityClass: dataMovePriorityClass,
 		repoConfigMgr:         repoConfigMgr,
 		podLabels:             podLabels,
@@ -131,6 +131,7 @@ func NewDataDownloadReconciler(
 // +kubebuilder:rbac:groups="",resources=pods,verbs=get
 // +kubebuilder:rbac:groups="",resources=persistentvolumes,verbs=get
 // +kubebuilder:rbac:groups="",resources=persistentvolumerclaims,verbs=get
+// +kubebuilder:rbac:groups="",resources=secrets;configmaps,verbs=get;list;create;delete
 
 func (r *DataDownloadReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
 	log := r.logger.WithFields(logrus.Fields{
@@ -198,7 +199,7 @@ func (r *DataDownloadReconciler) Reconcile(ctx context.Context, req ctrl.Request
 			}
 		}
 	} else {
-		delete(r.cancelledDataDownload, dd.Name)
+		r.cancelledDataDownload.Delete(dd.Name)
 
 		// put the finalizer remove action here for all cr will goes to the final status, we could check finalizer and do remove action in final status
 		// instead of intermediate state.
@@ -223,9 +224,9 @@ func (r *DataDownloadReconciler) Reconcile(ctx context.Context, req ctrl.Request
 	}
 
 	if dd.Spec.Cancel {
-		if spotted, found := r.cancelledDataDownload[dd.Name]; !found {
-			r.cancelledDataDownload[dd.Name] = r.Clock.Now()
-		} else {
+		v, loaded := r.cancelledDataDownload.LoadOrStore(dd.Name, r.Clock.Now())
+		if loaded {
+			spotted := v.(time.Time)
 			delay := cancelDelayOthers
 			if dd.Status.Phase == velerov2alpha1api.DataDownloadPhaseInProgress {
 				delay = cancelDelayInProgress
@@ -234,7 +235,7 @@ func (r *DataDownloadReconciler) Reconcile(ctx context.Context, req ctrl.Request
 			if time.Since(spotted) > delay {
 				log.Infof("Data download %s is canceled in Phase %s but not handled in rasonable time", dd.GetName(), dd.Status.Phase)
 				if r.tryCancelDataDownload(ctx, dd, "") {
-					delete(r.cancelledDataDownload, dd.Name)
+					r.cancelledDataDownload.Delete(dd.Name)
 				}
 
 				return ctrl.Result{}, nil
@@ -556,7 +557,7 @@ func (r *DataDownloadReconciler) OnDataDownloadCancelled(ctx context.Context, na
 		log.WithError(err).Error("error updating data download status")
 	} else {
 		r.metrics.RegisterDataDownloadCancel(r.nodeName)
-		delete(r.cancelledDataDownload, dd.Name)
+		r.cancelledDataDownload.Delete(dd.Name)
 	}
 }
 

pkg/controller/data_download_controller_test.go

@@ -19,9 +19,12 @@ package controller
 import (
 	"context"
 	"fmt"
+	"sync"
 	"testing"
 	"time"
 
+	clocktesting "k8s.io/utils/clock/testing"
+
 	"github.com/cockroachdb/errors"
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
@@ -507,7 +510,7 @@ func TestDataDownloadReconcile(t *testing.T) {
 			}
 
 			if test.sportTime != nil {
-				r.cancelledDataDownload[test.dd.Name] = test.sportTime.Time
+				r.cancelledDataDownload.Store(test.dd.Name, test.sportTime.Time)
 			}
 
 			if test.constrained {
@@ -624,9 +627,15 @@ func TestDataDownloadReconcile(t *testing.T) {
 			}
 
 			if test.expectCancelRecord {
-				assert.Contains(t, r.cancelledDataDownload, test.dd.Name)
+				_, ok := r.cancelledDataDownload.Load(test.dd.Name)
+				assert.True(t, ok)
 			} else {
-				assert.Empty(t, r.cancelledDataDownload)
+				empty := true
+				r.cancelledDataDownload.Range(func(key, value any) bool {
+					empty = false
+					return false
+				})
+				assert.True(t, empty)
 			}
 
 			if isDataDownloadInFinalState(&dd) || dd.Status.Phase == velerov2alpha1api.DataDownloadPhaseInProgress {
@@ -1437,3 +1446,50 @@ func TestDataDownloadSetupExposeParam(t *testing.T) {
 		})
 	}
 }
+
+type sequenceClock struct {
+	*clocktesting.FakeClock
+	mu sync.Mutex
+}
+
+func (c *sequenceClock) Now() time.Time {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.FakeClock.Step(time.Second)
+	return c.FakeClock.Now()
+}
+
+func TestDataDownloadCancelConcurrency(t *testing.T) {
+	ctx := t.Context()
+	dd := dataDownloadBuilder().Cancel(true).Phase(velerov2alpha1api.DataDownloadPhaseInProgress).Result()
+
+	r, err := initDataDownloadReconciler(t, nil)
+	require.NoError(t, err)
+
+	err = r.client.Create(ctx, dd)
+	require.NoError(t, err)
+
+	firstTime := time.Now()
+	// manually store the initial time
+	r.cancelledDataDownload.Store(dd.Name, firstTime)
+
+	// Custom clock that returns a different time each call
+	r.Clock = &sequenceClock{FakeClock: clocktesting.NewFakeClock(firstTime)}
+
+	var wg sync.WaitGroup
+	routines := 50
+	wg.Add(routines)
+
+	for i := 0; i < routines; i++ {
+		go func() {
+			defer wg.Done()
+			_, _ = r.Reconcile(ctx, ctrl.Request{NamespacedName: types.NamespacedName{Name: dd.Name, Namespace: dd.Namespace}})
+		}()
+	}
+
+	wg.Wait()
+
+	v, ok := r.cancelledDataDownload.Load(dd.Name)
+	assert.True(t, ok)
+	assert.Equal(t, firstTime, v.(time.Time), "The initially recorded timestamp should be preserved")
+}