Repository: cloudnative-pg/cloudnative-pg

Issue:
[Bug]: Backups considered "failed" while underlying VolumeSnapshots succeed
### Is there an existing issue already for this bug?

- [x] I have searched for an existing issue, and could not find anything. I believe this is a new bug.

### I have read the troubleshooting guide

- [x] I have read the troubleshooting guide and I think this is a new bug.

### I am running a supported version of CloudNativePG

- [x] I have read the troubleshooting guide and I think this is a new bug.

### Contact Details

jeffm@gisual.com

### Version

trunk (main)

### What version of Kubernetes are you using?

1.29

### What is your Kubernetes environment?

Cloud: Azure AKS

### How did you install the operator?

YAML manifest

### What happened?

Sometimes a backup is marked as failed even though the underlying volume snapshots succeed.

I ran the following commands to create a volume snapshot backup:

**One against a standby (failed):**
`kubectl cnpg backup -n db --backup-name manual-volume-snapshot-standby2 --immediate-checkpoint true --method volumeSnapshot  --backup-target prefer-standby production-pg-common`

**One against the primary (succeeded):**
`kubectl cnpg backup -n db --backup-name manual-volume-snapshot-primary2 --immediate-checkpoint true --method volumeSnapshot  --backup-target primary production-pg-common`

**Operator trace-level logs:**
[cnpg_operator_logs_filtered.log](https://github.com/user-attachments/files/18993935/cnpg_operator_logs_filtered.log)

**Additional cluster/volume/snapshot state before and after the failed backup:**
[standy-cnpg-failed-volumesnapshot-succeeded.zip](https://github.com/user-attachments/files/18993926/standy-cnpg-failed-volumesnapshot-succeeded.zip)

The instance logs didn't contain anything notable other than a single log entry for closing a stale backup connection (no other mentions of backups, however, they're included in the .zip for completeness. Unfortunately, I didn't capture the primary's instance logs in this run).

### Cluster resource
Refer to provided .zip

### Relevant log output
Refer to provided .zip or .log.

### Code of Conduct

- [x] I agree to follow this project's Code of Conduct

Hints from issue comments:
@jmealo: @leonardoce Let me know if we need anymore logs. These operator logs include both one successful and one failed backup. I'll help however I can.@jmealo: To summarize a discussion on Slack between @leonardoce and I:

# Summary of Discussion on Volume Snapshot Backup Failures
After investigating the issue where backups are considered "failed" while the underlying VolumeSnapshots succeed, we've identified several key points:

## Current Behavior

CNPG considers every error (except known retryable ones) as non-retryable
When a non-retryable error is discovered, CNPG marks the backup as failed and closes the PostgreSQL connection
In this case, a timeout during snapshot creation is being treated as a non-retryable error
Even though the Kubernetes CSI controller retries the snapshot creation and it eventually succeeds, CNPG has already marked the backup as failed

## Technical Details
The backup process sequence is:

1. `pg_backup_start()` in a connection (connection should remain open)
2. Take snapshots
3. `pg_backup_stop()` in the same connection to get the backup label
4. Use the backup label to annotate the snapshot
5. Close the connection

## Key Problems

- Without the backup labels, snapshots are "useless" for restore purposes
- The Kubernetes CSI snapshot controller ([`kubernetes-csi/external-snapshotter`](https://github.com/kubernetes-csi/external-snapshotter)) will retry snapshot creation, but CNPG isn't coordinating with this behavior
- When CNPG encounters the timeout error, it marks the backup as failed immediately instead of waiting for retries to complete
- If we simply retry all errors, we can't run a new backup until the current one completes (or fails definitively)

## Potential Solutions Discussed

- Declare timeout errors as retryable
- Add a configurable deadline for backup attempts ([See TODO comment](https://github.com/cloudnative-pg/cloudnative-pg/blob/34ab236e0a442e7bd30b401257824926660d1ae3/pkg/reconciler/backup/volumesnapshot/resources.go#L71-L74))
- Consider options for handling backup conflicts:
  - Use a different standby if available
  - Cancel existing backup
  - Do not schedule new backup while one is in progress
  - Schedule backup to run after current one completes

## Open Questions

- ⭐ Is considering context deadline exceeded/timeouts as retryable the only change required to fix this issue?
  - If not, and we need to be more selective, What errors are truly non-retryable? 
- How to properly handle the case where snapshots exist but aren't marked as "Ready to use"
- How to determine when to definitively fail a backup vs. waiting for retries to complete
- How to balance keeping PostgreSQL connections open (which creates stale processes) with the need to complete the backup workflow properly

This issue appears to require changes to how CNPG determines which errors are retryable and how it monitors snapshot status after creation has been initiated.

Solution patch:
diff --git a/api/v1/backup_funcs.go b/api/v1/backup_funcs.go
index e7ec0411..e6264fd3 100644
--- a/api/v1/backup_funcs.go
+++ b/api/v1/backup_funcs.go
@@ -19,7 +19,9 @@ package v1
 import (
 	"context"
 	"sort"
+	"strconv"
 	"strings"
+	"time"
 
 	volumesnapshot "github.com/kubernetes-csi/external-snapshotter/client/v8/apis/volumesnapshot/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -95,6 +97,23 @@ func (backupStatus *BackupStatus) GetOnline() bool {
 	return *backupStatus.Online
 }
 
+// GetVolumeSnapshotDeadline returns the volume snapshot deadline in minutes.
+func (backup *Backup) GetVolumeSnapshotDeadline() time.Duration {
+	const defaultValue = 10
+
+	value := backup.Annotations[utils.BackupVolumeSnapshotDeadlineAnnotationName]
+	if value == "" {
+		return defaultValue * time.Minute
+	}
+
+	minutes, err := strconv.Atoi(value)
+	if err != nil {
+		return defaultValue * time.Minute
+	}
+
+	return time.Duration(minutes) * time.Minute
+}
+
 // IsCompletedVolumeSnapshot checks if a backup is completed using the volume snapshot method.
 // It returns true if the backup's method is BackupMethodVolumeSnapshot and its status phase is BackupPhaseCompleted.
 // Otherwise, it returns false.
diff --git a/api/v1/scheduledbackup_funcs.go b/api/v1/scheduledbackup_funcs.go
index 6acf6c88..770683ce 100644
--- a/api/v1/scheduledbackup_funcs.go
+++ b/api/v1/scheduledbackup_funcs.go
@@ -78,5 +78,14 @@ func (scheduledBackup *ScheduledBackup) CreateBackup(name string) *Backup {
 		},
 	}
 	utils.InheritAnnotations(&backup.ObjectMeta, scheduledBackup.Annotations, nil, configuration.Current)
+
+	if backup.Annotations == nil {
+		backup.Annotations = make(map[string]string)
+	}
+
+	if v := scheduledBackup.Annotations[utils.BackupVolumeSnapshotDeadlineAnnotationName]; v != "" {
+		backup.Annotations[utils.BackupVolumeSnapshotDeadlineAnnotationName] = v
+	}
+
 	return &backup
 }
diff --git a/docs/src/backup_volumesnapshot.md b/docs/src/backup_volumesnapshot.md
index a927b01f..964081b9 100644
--- a/docs/src/backup_volumesnapshot.md
+++ b/docs/src/backup_volumesnapshot.md
@@ -242,7 +242,95 @@ referenced in the `.spec.backup.volumeSnapshot.className` option.
 Please refer to the [Kubernetes documentation on Volume Snapshot Classes](https://kubernetes.io/docs/concepts/storage/volume-snapshot-classes/)
 for details on this standard behavior.
 
-## Example
+## Backup Volume Snapshot Deadlines
+
+CloudNativePG supports backups using the volume snapshot method. In some
+environments, volume snapshots may encounter temporary issues that can be
+retried.
+
+The `backup.cnpg.io/volumeSnapshotDeadline` annotation defines how long
+CloudNativePG should continue retrying recoverable errors before marking the
+backup as failed.
+
+You can add the `backup.cnpg.io/volumeSnapshotDeadline` annotation to both
+`Backup` and `ScheduledBackup` resources. For `ScheduledBackup` resources, this
+annotation is automatically inherited by any `Backup` resources created from
+the schedule.
+
+If not specified, the default retry deadline is **10 minutes**.
+
+### Error Handling
+
+When a retryable error occurs during a volume snapshot operation:
+
+1. CloudNativePG records the time of the first error.
+2. The system retries the operation every **10 seconds**.
+3. If the error persists beyond the specified deadline (or the default 10
+   minutes), the backup is marked as **failed**.
+
+### Retryable Errors
+
+CloudNativePG treats the following types of errors as retryable:
+
+- **Server timeout errors** (HTTP 408, 429, 500, 502, 503, 504)
+- **Conflicts** (optimistic locking errors)
+- **Internal errors**
+- **Context deadline exceeded errors**
+- **Timeout errors from the CSI snapshot controller**
+
+### Examples
+
+You can add the annotation to a `ScheduledBackup` resource as follows:
+
+```yaml
+apiVersion: postgresql.cnpg.io/v1
+kind: ScheduledBackup
+metadata:
+  name: daily-backup-schedule
+  annotations:
+    backup.cnpg.io/volumeSnapshotDeadline: "20"
+spec:
+  schedule: "0 0 * * *"
+  backupOwnerReference: self
+  method: volumeSnapshot
+  # other configuration...
+```
+
+When you define a `ScheduledBackup` with the annotation, any `Backup` resources
+created from this schedule automatically inherit the specified timeout value.
+
+In the following example, all backups created from the schedule will have a
+30-minute timeout for retrying recoverable snapshot errors.
+
+```yaml
+apiVersion: postgresql.cnpg.io/v1
+kind: ScheduledBackup
+metadata:
+  name: weekly-backup
+  annotations:
+    backup.cnpg.io/volumeSnapshotDeadline: "30"
+spec:
+  schedule: "0 0 * * 0"  # Weekly backup on Sunday
+  method: volumeSnapshot
+  cluster:
+    name: my-postgresql-cluster
+```
+
+Alternatively, you can add the annotation directly to a `Backup` Resource:
+
+```yaml
+apiVersion: postgresql.cnpg.io/v1
+kind: Backup
+metadata:
+  name: my-backup
+  annotations:
+    backup.cnpg.io/volumeSnapshotDeadline: "15"
+spec:
+  method: volumeSnapshot
+  # other backup configuration...
+```
+
+## Example of Volume Snapshot Backup
 
 The following example shows how to configure volume snapshot base backups on an
 EKS cluster on AWS using the `ebs-sc` storage class and the `csi-aws-vsc`
diff --git a/docs/src/labels_annotations.md b/docs/src/labels_annotations.md
index 677ee563..b4d3a0a6 100644
--- a/docs/src/labels_annotations.md
+++ b/docs/src/labels_annotations.md
@@ -227,6 +227,11 @@ CloudNativePG manages the following predefined annotations:
 `cnpg.io/snapshotEndTime`
 :   The time a snapshot was marked as ready to use.
 
+`cnpg.io/volumeSnapshotDeadline`
+:   Applied to `Backup` and `ScheduledBackup` resources, allows you to control
+    how long the operator should retry recoverable errors before considering the
+    volume snapshot backup failed. In minutes, defaulting to 10.
+
 `kubectl.kubernetes.io/restartedAt`
 :   When available, the time of last requested restart of a Postgres cluster.
 
diff --git a/internal/controller/backup_controller.go b/internal/controller/backup_controller.go
index 5d588b62..a9582701 100644
--- a/internal/controller/backup_controller.go
+++ b/internal/controller/backup_controller.go
@@ -456,10 +456,6 @@ func (r *BackupReconciler) reconcileSnapshotBackup(
 		Build()
 
 	res, err := reconciler.Reconcile(ctx, cluster, backup, targetPod, pvcs)
-	if isErrorRetryable(err) {
-		contextLogger.Error(err, "detected retryable error while executing snapshot backup, retrying...")
-		return &ctrl.Result{RequeueAfter: 5 * time.Second}, nil
-	}
 	if err != nil {
 		// Volume Snapshot errors are not retryable, we need to set this backup as failed
 		// and un-fence the Pod
@@ -569,11 +565,6 @@ func updateClusterWithSnapshotsBackupTimes(
 	return nil
 }
 
-// isErrorRetryable detects is an error is retryable or not
-func isErrorRetryable(err error) bool {
-	return apierrs.IsServerTimeout(err) || apierrs.IsConflict(err) || apierrs.IsInternalError(err)
-}
-
 // getBackupTargetPod returns the pod that should run the backup according to the current
 // cluster's target policy
 func (r *BackupReconciler) getBackupTargetPod(ctx context.Context,
diff --git a/internal/webhook/v1/backup_webhook.go b/internal/webhook/v1/backup_webhook.go
index 87ed87d1..681e40dc 100644
--- a/internal/webhook/v1/backup_webhook.go
+++ b/internal/webhook/v1/backup_webhook.go
@@ -19,6 +19,7 @@ package v1
 import (
 	"context"
 	"fmt"
+	"strconv"
 
 	"github.com/cloudnative-pg/machinery/pkg/log"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -168,5 +169,16 @@ func (v *BackupCustomValidator) validate(r *apiv1.Backup) field.ErrorList {
 		))
 	}
 
+	if value := r.Annotations[utils.BackupVolumeSnapshotDeadlineAnnotationName]; value != "" {
+		_, err := strconv.Atoi(value)
+		if err != nil {
+			result = append(result, field.Invalid(
+				field.NewPath("metadata", "annotations", utils.BackupVolumeSnapshotDeadlineAnnotationName),
+				value,
+				"must be an integer",
+			))
+		}
+	}
+
 	return result
 }
diff --git a/pkg/reconciler/backup/volumesnapshot/errors.go b/pkg/reconciler/backup/volumesnapshot/errors.go
index e5d8b115..e5c9d4c4 100644
--- a/pkg/reconciler/backup/volumesnapshot/errors.go
+++ b/pkg/reconciler/backup/volumesnapshot/errors.go
@@ -17,9 +17,13 @@ limitations under the License.
 package volumesnapshot
 
 import (
+	"context"
+	"errors"
 	"regexp"
 	"strconv"
 	"strings"
+
+	apierrs "k8s.io/apimachinery/pkg/api/errors"
 )
 
 var (
@@ -27,15 +31,32 @@ var (
 	httpStatusCodeRegex  = regexp.MustCompile(`HTTPStatusCode:\s(\d{3})`)
 )
 
-// isRetriableErrorMessage detects if a certain error message belongs
-// to a retriable error or not. This is obviously an heuristic but
-// unfortunately we don't have that information exposed in the
-// Kubernetes VolumeSnapshot API and the CSI driver haven't that too.
-func isRetriableErrorMessage(msg string) bool {
+// isErrorRetryable detects is an error is retryable or not.
+//
+// Important: this function is intended for detecting errors that
+// occur during communication between the operator and the Kubernetes
+// API server, as well as between the operator and the instance
+// manager.
+// It is not designed to check errors raised by the CSI driver and
+// exposed by the CSI snapshotter sidecar.
+func isNetworkErrorRetryable(err error) bool {
+	return apierrs.IsServerTimeout(err) || apierrs.IsConflict(err) || apierrs.IsInternalError(err) ||
+		errors.Is(err, context.DeadlineExceeded)
+}
+
+// isCSIErrorMessageRetriable detects if a certain error message
+// raised by the CSI driver corresponds to a retriable error or
+// not.
+//
+// It relies on heuristics, as this information is not available in
+// the Kubernetes VolumeSnapshot API, and the CSI driver does not
+// expose it either.
+func isCSIErrorMessageRetriable(msg string) bool {
 	isRetryableFuncs := []func(string) bool{
 		isExplicitlyRetriableError,
 		isRetryableHTTPError,
 		isConflictError,
+		isContextDeadlineExceededError,
 	}
 
 	for _, isRetryableFunc := range isRetryableFuncs {
@@ -47,6 +68,12 @@ func isRetriableErrorMessage(msg string) bool {
 	return false
 }
 
+// isContextDeadlineExceededError detects context deadline exceeded errors
+// These are timeouts that may be retried by the Kubernetes CSI controller
+func isContextDeadlineExceededError(msg string) bool {
+	return strings.Contains(msg, "deadline exceeded") || strings.Contains(msg, "timed out")
+}
+
 // isConflictError detects optimistic locking errors
 func isConflictError(msg string) bool {
 	// Obviously this is a heuristic, but unfortunately we don't have
diff --git a/pkg/reconciler/backup/volumesnapshot/reconciler.go b/pkg/reconciler/backup/volumesnapshot/reconciler.go
index 730fa8f2..2ff568ed 100644
--- a/pkg/reconciler/backup/volumesnapshot/reconciler.go
+++ b/pkg/reconciler/backup/volumesnapshot/reconciler.go
@@ -155,6 +155,24 @@ func (se *Reconciler) Reconcile(
 	backup *apiv1.Backup,
 	targetPod *corev1.Pod,
 	pvcs []corev1.PersistentVolumeClaim,
+) (*ctrl.Result, error) {
+	contextLogger := log.FromContext(ctx).WithName("volumesnapshot_reconciler")
+
+	res, err := se.internalReconcile(ctx, cluster, backup, targetPod, pvcs)
+	if isNetworkErrorRetryable(err) {
+		contextLogger.Error(err, "detected retryable error while executing snapshot backup, retrying...")
+		return &ctrl.Result{RequeueAfter: 5 * time.Second}, nil
+	}
+
+	return res, err
+}
+
+func (se *Reconciler) internalReconcile(
+	ctx context.Context,
+	cluster *apiv1.Cluster,
+	backup *apiv1.Backup,
+	targetPod *corev1.Pod,
+	pvcs []corev1.PersistentVolumeClaim,
 ) (*ctrl.Result, error) {
 	if cluster.Spec.Backup == nil || cluster.Spec.Backup.VolumeSnapshot == nil {
 		return nil, fmt.Errorf("cannot execute a VolumeSnapshot on a cluster without configuration")
@@ -190,7 +208,7 @@ func (se *Reconciler) Reconcile(
 	}
 
 	// Step 3: wait for snapshots to be provisioned
-	if res, err := se.waitSnapshotToBeProvisionedStep(ctx, volumeSnapshots); res != nil || err != nil {
+	if res, err := se.waitSnapshotToBeProvisionedStep(ctx, backup, volumeSnapshots); res != nil || err != nil {
 		return res, err
 	}
 
@@ -207,7 +225,7 @@ func (se *Reconciler) Reconcile(
 	}
 
 	// Step 5: wait for snapshots to be ready to use
-	if res, err := se.waitSnapshotToBeReadyStep(ctx, volumeSnapshots); res != nil || err != nil {
+	if res, err := se.waitSnapshotToBeReadyStep(ctx, backup, volumeSnapshots); res != nil || err != nil {
 		return res, err
 	}
 
@@ -385,10 +403,11 @@ func (se *Reconciler) createSnapshotPVCGroupStep(
 // waitSnapshotToBeProvisionedStep waits for every PVC snapshot to be claimed
 func (se *Reconciler) waitSnapshotToBeProvisionedStep(
 	ctx context.Context,
+	backup *apiv1.Backup,
 	snapshots []storagesnapshotv1.VolumeSnapshot,
 ) (*ctrl.Result, error) {
 	for i := range snapshots {
-		if res, err := se.waitSnapshotToBeProvisionedAndAnnotate(ctx, &snapshots[i]); res != nil || err != nil {
+		if res, err := se.waitSnapshotToBeProvisionedAndAnnotate(ctx, backup, &snapshots[i]); res != nil || err != nil {
 			return res, err
 		}
 	}
@@ -399,10 +418,11 @@ func (se *Reconciler) waitSnapshotToBeProvisionedStep(
 // waitSnapshotToBeReadyStep waits for every PVC snapshot to be ready to use
 func (se *Reconciler) waitSnapshotToBeReadyStep(
 	ctx context.Context,
+	backup *apiv1.Backup,
 	snapshots []storagesnapshotv1.VolumeSnapshot,
 ) (*ctrl.Result, error) {
 	for i := range snapshots {
-		if res, err := se.waitSnapshotToBeReady(ctx, &snapshots[i]); res != nil || err != nil {
+		if res, err := se.waitSnapshotToBeReady(ctx, backup, &snapshots[i]); res != nil || err != nil {
 			return res, err
 		}
 	}
@@ -501,20 +521,14 @@ func transferLabelsToAnnotations(labels map[string]string, annotations map[strin
 // SnapshotStartTimeAnnotationName and SnapshotEndTimeAnnotationName.
 func (se *Reconciler) waitSnapshotToBeProvisionedAndAnnotate(
 	ctx context.Context,
+	backup *apiv1.Backup,
 	snapshot *storagesnapshotv1.VolumeSnapshot,
 ) (*ctrl.Result, error) {
 	contextLogger := log.FromContext(ctx)
 
 	info := parseVolumeSnapshotInfo(snapshot)
 	if info.error != nil {
-		if info.error.isRetryable() {
-			contextLogger.Error(info.error,
-				"Retryable snapshot provisioning error, trying again",
-				"volumeSnapshotName", snapshot.Name)
-			return &ctrl.Result{RequeueAfter: 10 * time.Second}, nil
-		}
-
-		return nil, info.error
+		return se.handleSnapshotErrors(ctx, backup, info.error)
 	}
 	if !info.provisioned {
 		contextLogger.Info(
@@ -544,20 +558,14 @@ func (se *Reconciler) waitSnapshotToBeProvisionedAndAnnotate(
 // SnapshotStartTimeAnnotationName and SnapshotEndTimeAnnotationName.
 func (se *Reconciler) waitSnapshotToBeReady(
 	ctx context.Context,
+	backup *apiv1.Backup,
 	snapshot *storagesnapshotv1.VolumeSnapshot,
 ) (*ctrl.Result, error) {
 	contextLogger := log.FromContext(ctx)
 
 	info := parseVolumeSnapshotInfo(snapshot)
 	if info.error != nil {
-		if info.error.isRetryable() {
-			contextLogger.Error(info.error,
-				"Retryable snapshot provisioning error, trying again",
-				"volumeSnapshotName", snapshot.Name)
-			return &ctrl.Result{RequeueAfter: 10 * time.Second}, nil
-		}
-
-		return nil, info.error
+		return se.handleSnapshotErrors(ctx, backup, info.error)
 	}
 	if !info.ready {
 		contextLogger.Info(
@@ -570,3 +578,89 @@ func (se *Reconciler) waitSnapshotToBeReady(
 
 	return nil, nil
 }
+
+func (se *Reconciler) handleSnapshotErrors(
+	ctx context.Context,
+	backup *apiv1.Backup,
+	snapshotErr *volumeSnapshotError,
+) (*ctrl.Result, error) {
+	contextLogger := log.FromContext(ctx).
+		WithName("handle_snapshot_errors")
+
+	if !snapshotErr.isRetryable() {
+		return nil, snapshotErr
+	}
+
+	if err := addDeadlineStatus(ctx, se.cli, backup); err != nil {
+		return nil, fmt.Errorf("while adding deadline status: %w", err)
+	}
+
+	exceeded, err := isDeadlineExceeded(backup)
+	if err != nil {
+		return nil, fmt.Errorf("while checking if deadline was exceeded: %w", err)
+	}
+	if exceeded {
+		return nil, fmt.Errorf("deadline exceeded for error %w", snapshotErr)
+	}
+
+	contextLogger.Error(snapshotErr,
+		"Retryable snapshot provisioning error, trying again",
+	)
+	return &ctrl.Result{RequeueAfter: 10 * time.Second}, nil
+}
+
+func isDeadlineExceeded(backup *apiv1.Backup) (bool, error) {
+	if backup.Status.PluginMetadata[pluginName] == "" {
+		return false, fmt.Errorf("no plugin metadata found in backup status")
+	}
+
+	data, err := unmarshalMetadata(backup.Status.PluginMetadata[pluginName])
+	if err != nil {
+		return false, fmt.Errorf("while unmarshalling plugin metadata: %w", err)
+	}
+
+	// if the deadline have passed since firstFailureTime we need to consider the deadline exceeded
+	deadline := int64(backup.GetVolumeSnapshotDeadline().Seconds())
+	return time.Now().Unix()-data.VolumeSnapshotFirstDetectedFailure > deadline, nil
+}
+
+type metadata struct {
+	// VolumeSnapshotFirstDetectedFailure is UNIX the timestamp when the first volume snapshot failure was detected
+	VolumeSnapshotFirstDetectedFailure int64 `json:"volumeSnapshotFirstFailure,omitempty"`
+}
+
+func unmarshalMetadata(rawData string) (*metadata, error) {
+	var data metadata
+	if err := json.Unmarshal([]byte(rawData), &data); err != nil {
+		return nil, fmt.Errorf("while unmarshalling metadata: %w", err)
+	}
+
+	if data.VolumeSnapshotFirstDetectedFailure == 0 {
+		return nil, fmt.Errorf("no volumeSnapshotFirstFailure found in plugin metadata: %s", pluginName)
+	}
+
+	return &data, nil
+}
+
+func addDeadlineStatus(ctx context.Context, cli client.Client, backup *apiv1.Backup) error {
+	if value, ok := backup.Status.PluginMetadata[pluginName]; ok {
+		if _, err := unmarshalMetadata(value); err == nil {
+			return nil
+		}
+	}
+
+	data := &metadata{VolumeSnapshotFirstDetectedFailure: time.Now().Unix()}
+	rawData, err := json.Marshal(data)
+	if err != nil {
+		return err
+	}
+
+	if backup.Status.PluginMetadata == nil {
+		backup.Status.PluginMetadata = map[string]string{}
+	}
+
+	origBackup := backup.DeepCopy()
+	backup.Status.PluginMetadata[pluginName] = string(rawData)
+
+	return cli.Status().Patch(ctx, backup, client.MergeFrom(origBackup))
+}
diff --git a/pkg/reconciler/backup/volumesnapshot/resources.go b/pkg/reconciler/backup/volumesnapshot/resources.go
index eaee26cf..71794502 100644
--- a/pkg/reconciler/backup/volumesnapshot/resources.go
+++ b/pkg/reconciler/backup/volumesnapshot/resources.go
@@ -26,6 +26,8 @@ import (
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/utils"
 )
 
+const pluginName = "cnpg_volumesnapshot"
+
 // volumeSnapshotInfo host information about a volume snapshot
 type volumeSnapshotInfo struct {
 	// error contains the raised error when the volume snapshot terminated
@@ -68,16 +70,15 @@ func (err volumeSnapshotError) Error() string {
 // IsRetryable returns true if the external snapshotter controller
 // will retry taking the snapshot
 func (err volumeSnapshotError) isRetryable() bool {
-	// TODO: instead of blindingly retry on matching errors, we
-	// should enhance our CRD with a configurable deadline. After
-	// the deadline have been met on err.InternalError.CreatedAt
-	// the backup can be marked as failed
+	// The Kubernetes CSI driver/controller will automatically retry snapshot creation
+	// for certain errors, including timeouts. We use pattern matching to identify
+	// these retryable errors and handle them appropriately.
 
 	if err.InternalError.Message == nil {
 		return false
 	}
 
-	return isRetriableErrorMessage(*err.InternalError.Message)
+	return isCSIErrorMessageRetriable(*err.InternalError.Message)
 }
 
 // slice represents a slice of []storagesnapshotv1.VolumeSnapshot
diff --git a/pkg/utils/labels_annotations.go b/pkg/utils/labels_annotations.go
index e8970cd5..32d63496 100644
--- a/pkg/utils/labels_annotations.go
+++ b/pkg/utils/labels_annotations.go
@@ -209,6 +209,10 @@ const (
 	// BackupTablespaceMapFileAnnotationName is the name of the annotation where the `tablespace_map` file is kept
 	BackupTablespaceMapFileAnnotationName = MetadataNamespace + "/backupTablespaceMapFile"
 
+	// BackupVolumeSnapshotDeadlineAnnotationName is the annotation for the snapshot backup failure deadline in minutes.
+	// It is only applied to snapshot retryable errors
+	BackupVolumeSnapshotDeadlineAnnotationName = MetadataNamespace + "/volumeSnapshotDeadline"
+
 	// SnapshotStartTimeAnnotationName is the name of the annotation where a snapshot's start time is kept
 	SnapshotStartTimeAnnotationName = MetadataNamespace + "/snapshotStartTime"
 



Test patch:
diff --git a/api/v1/scheduledbackup_funcs_test.go b/api/v1/scheduledbackup_funcs_test.go
index 150192d8..ca957482 100644
--- a/api/v1/scheduledbackup_funcs_test.go
+++ b/api/v1/scheduledbackup_funcs_test.go
@@ -17,7 +17,10 @@ limitations under the License.
 package v1
 
 import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
 	"github.com/cloudnative-pg/cloudnative-pg/internal/configuration"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/utils"
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
@@ -28,7 +31,11 @@ var _ = Describe("Scheduled backup", func() {
 	backupName := "test"
 
 	BeforeEach(func() {
-		scheduledBackup = &ScheduledBackup{}
+		scheduledBackup = &ScheduledBackup{
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: make(map[string]string),
+			},
+		}
 	})
 
 	It("properly creates a backup with no annotations", func() {
@@ -38,6 +45,14 @@ var _ = Describe("Scheduled backup", func() {
 		Expect(backup.Annotations).To(BeEmpty())
 	})
 
+	It("should always inherit volumeSnapshotDeadline while creating a backup", func() {
+		scheduledBackup.Annotations[utils.BackupVolumeSnapshotDeadlineAnnotationName] = "20"
+		backup := scheduledBackup.CreateBackup("test")
+		Expect(backup).ToNot(BeNil())
+		Expect(backup.ObjectMeta.Name).To(BeEquivalentTo(backupName))
+		Expect(backup.Annotations[utils.BackupVolumeSnapshotDeadlineAnnotationName]).To(BeEquivalentTo("20"))
+	})
+
 	It("properly creates a backup with annotations", func() {
 		annotations := make(map[string]string, 1)
 		annotations["test"] = "annotations"
diff --git a/internal/webhook/v1/backup_webhook_test.go b/internal/webhook/v1/backup_webhook_test.go
index 2ac2fbf8..60c73f9e 100644
--- a/internal/webhook/v1/backup_webhook_test.go
+++ b/internal/webhook/v1/backup_webhook_test.go
@@ -17,6 +17,7 @@ limitations under the License.
 package v1
 
 import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/ptr"
 
 	apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
@@ -78,4 +79,40 @@ var _ = Describe("Backup webhook validate", func() {
 		Expect(result).To(HaveLen(1))
 		Expect(result[0].Field).To(Equal("spec.onlineConfiguration"))
 	})
+
+	It("returns error if BackupVolumeSnapshotDeadlineAnnotationName is not an integer", func() {
+		backup := &apiv1.Backup{
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: map[string]string{
+					utils.BackupVolumeSnapshotDeadlineAnnotationName: "not-an-integer",
+				},
+			},
+		}
+		result := v.validate(backup)
+		Expect(result).To(HaveLen(1))
+		Expect(result[0].Field).To(Equal("metadata.annotations." + utils.BackupVolumeSnapshotDeadlineAnnotationName))
+		Expect(result[0].Error()).To(ContainSubstring("must be an integer"))
+	})
+
+	It("does not return error if BackupVolumeSnapshotDeadlineAnnotationName is an integer", func() {
+		backup := &apiv1.Backup{
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: map[string]string{
+					utils.BackupVolumeSnapshotDeadlineAnnotationName: "123",
+				},
+			},
+		}
+		result := v.validate(backup)
+		Expect(result).To(BeEmpty())
+	})
+
+	It("does not return error if BackupVolumeSnapshotDeadlineAnnotationName is not set", func() {
+		backup := &apiv1.Backup{
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: map[string]string{},
+			},
+		}
+		result := v.validate(backup)
+		Expect(result).To(BeEmpty())
+	})
 })
diff --git a/pkg/reconciler/backup/volumesnapshot/errors_test.go b/pkg/reconciler/backup/volumesnapshot/errors_test.go
index 1652fd06..81536f93 100644
--- a/pkg/reconciler/backup/volumesnapshot/errors_test.go
+++ b/pkg/reconciler/backup/volumesnapshot/errors_test.go
@@ -17,6 +17,13 @@ limitations under the License.
 package volumesnapshot
 
 import (
+	"context"
+	"errors"
+	"fmt"
+
+	apierrs "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 )
@@ -25,7 +32,7 @@ var _ = Describe("Retriable error messages", func() {
 	DescribeTable(
 		"Retriable error messages",
 		func(msg string, isRetriable bool) {
-			Expect(isRetriableErrorMessage(msg)).To(Equal(isRetriable))
+			Expect(isCSIErrorMessageRetriable(msg)).To(Equal(isRetriable))
 		},
 		Entry("conflict", "Hey, the object has been modified!", true),
 		Entry("non-retriable error", "VolumeSnapshotClass not found", false),
@@ -33,5 +40,60 @@ var _ = Describe("Retriable error messages", func() {
 		Entry("explicitly non-retriable error", "Retriable: false because my pod is working", false),
 		Entry("error code 502 - retriable", "RetryAfter: 0s, HTTPStatusCode: 502, RawError: Internal Server Error", true),
 		Entry("error code 404 - non retriable", "RetryAfter: 0s, HTTPStatusCode: 404, RawError: Not found", false),
+		Entry("context deadline exceeded - retriable", "context deadline exceeded waiting for snapshot creation", true),
+		Entry("deadline exceeded - retriable", "deadline exceeded during Azure snapshot creation", true),
+		Entry("timed out - retriable", "operation timed out for csi-disk-handler", true),
 	)
+
+	Describe("isContextDeadlineExceededError", func() {
+		It("detects 'context deadline exceeded' error messages", func() {
+			Expect(isContextDeadlineExceededError("context deadline exceeded")).To(BeTrue())
+		})
+
+		It("detects 'deadline exceeded' error messages", func() {
+			Expect(isContextDeadlineExceededError("deadline exceeded")).To(BeTrue())
+		})
+
+		It("detects 'timed out' error messages", func() {
+			Expect(isContextDeadlineExceededError("operation timed out")).To(BeTrue())
+		})
+
+		It("rejects non-timeout error messages", func() {
+			Expect(isContextDeadlineExceededError("not found")).To(BeFalse())
+			Expect(isContextDeadlineExceededError("permission denied")).To(BeFalse())
+			Expect(isContextDeadlineExceededError("invalid input")).To(BeFalse())
+		})
+	})
+})
+
+var _ = Describe("isNetworkErrorRetryable", func() {
+	It("recognizes server timeout errors", func() {
+		err := apierrs.NewServerTimeout(schema.GroupResource{}, "test", 1)
+		Expect(isNetworkErrorRetryable(err)).To(BeTrue())
+	})
+
+	It("recognizes conflict errors", func() {
+		err := apierrs.NewConflict(schema.GroupResource{}, "test", nil)
+		Expect(isNetworkErrorRetryable(err)).To(BeTrue())
+	})
+
+	It("recognizes internal errors", func() {
+		err := apierrs.NewInternalError(fmt.Errorf("test error"))
+		Expect(isNetworkErrorRetryable(err)).To(BeTrue())
+	})
+
+	It("recognizes context deadline exceeded errors", func() {
+		err := context.DeadlineExceeded
+		Expect(isNetworkErrorRetryable(err)).To(BeTrue())
+	})
+
+	It("does not retry on not found errors", func() {
+		err := apierrs.NewNotFound(schema.GroupResource{}, "test")
+		Expect(isNetworkErrorRetryable(err)).To(BeFalse())
+	})
+
+	It("does not retry on random errors", func() {
+		err := errors.New("random error")
+		Expect(isNetworkErrorRetryable(err)).To(BeFalse())
+	})
 })
diff --git a/pkg/reconciler/backup/volumesnapshot/reconciler_test.go b/pkg/reconciler/backup/volumesnapshot/reconciler_test.go
index 8ce7adc3..e3950ea9 100644
--- a/pkg/reconciler/backup/volumesnapshot/reconciler_test.go
+++ b/pkg/reconciler/backup/volumesnapshot/reconciler_test.go
@@ -18,6 +18,7 @@ package volumesnapshot
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"time"
 
@@ -451,3 +452,137 @@ var _ = Describe("annotateSnapshotsWithBackupData", func() {
 		}
 	})
 })
+
+var _ = Describe("addDeadlineStatus", func() {
+	var (
+		ctx    context.Context
+		backup *apiv1.Backup
+		cli    k8client.Client
+	)
+
+	BeforeEach(func() {
+		ctx = context.TODO()
+		backup = &apiv1.Backup{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: "test-namespace",
+				Name:      "test-backup",
+			},
+			Status: apiv1.BackupStatus{
+				PluginMetadata: make(map[string]string),
+			},
+		}
+		cli = fake.NewClientBuilder().WithScheme(scheme.BuildWithAllKnownScheme()).
+			WithObjects(backup).
+			WithStatusSubresource(&apiv1.Backup{}).
+			Build()
+	})
+
+	It("should add deadline status if not present", func() {
+		err := addDeadlineStatus(ctx, cli, backup)
+		Expect(err).ToNot(HaveOccurred())
+
+		var updatedBackup apiv1.Backup
+		err = cli.Get(ctx, types.NamespacedName{Name: backup.Name, Namespace: backup.Namespace}, &updatedBackup)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(updatedBackup.Status.PluginMetadata).To(HaveKey(pluginName))
+		Expect(updatedBackup.Status.PluginMetadata[pluginName]).ToNot(BeEmpty())
+		Expect(updatedBackup.Status.PluginMetadata[pluginName]).To(MatchRegexp(`{"volumeSnapshotFirstFailure":\d+}`))
+	})
+
+	It("should not modify deadline status if already present", func() {
+		backup.Status.PluginMetadata[pluginName] = `{"volumeSnapshotFirstFailure": 1234567890}`
+		err := cli.Status().Update(ctx, backup)
+		Expect(err).ToNot(HaveOccurred())
+
+		err = addDeadlineStatus(ctx, cli, backup)
+		Expect(err).ToNot(HaveOccurred())
+
+		var updatedBackup apiv1.Backup
+		err = cli.Get(ctx, types.NamespacedName{Name: backup.Name, Namespace: backup.Namespace}, &updatedBackup)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(updatedBackup.Status.PluginMetadata[pluginName]).To(Equal(`{"volumeSnapshotFirstFailure": 1234567890}`))
+	})
+})
+
+var _ = Describe("isDeadlineExceeded", func() {
+	var backup *apiv1.Backup
+
+	BeforeEach(func() {
+		backup = &apiv1.Backup{
+			Status: apiv1.BackupStatus{
+				PluginMetadata: make(map[string]string),
+			},
+		}
+	})
+
+	It("should return an error if plugin metadata is empty", func() {
+		_, err := isDeadlineExceeded(backup)
+		Expect(err).To(HaveOccurred())
+	})
+
+	It("should return error if unmarshalling fails", func() {
+		backup.Status.PluginMetadata[pluginName] = "invalid-json"
+		exceeded, err := isDeadlineExceeded(backup)
+		Expect(err).To(HaveOccurred())
+		Expect(exceeded).To(BeFalse())
+	})
+
+	It("should return error if no volumeSnapshotFirstFailure found in plugin metadata", func() {
+		backup.Status.PluginMetadata[pluginName] = `{}`
+		exceeded, err := isDeadlineExceeded(backup)
+		Expect(err).To(HaveOccurred())
+		Expect(exceeded).To(BeFalse())
+	})
+
+	It("should return false if deadline has not exceeded", func() {
+		data := metadata{VolumeSnapshotFirstDetectedFailure: time.Now().Unix()}
+		rawData, _ := json.Marshal(data)
+		backup.Status.PluginMetadata[pluginName] = string(rawData)
+		backup.Annotations = map[string]string{utils.BackupVolumeSnapshotDeadlineAnnotationName: "10"}
+
+		exceeded, err := isDeadlineExceeded(backup)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(exceeded).To(BeFalse())
+	})
+
+	It("should return true if deadline has exceeded", func() {
+		data := metadata{VolumeSnapshotFirstDetectedFailure: time.Now().Add(-20 * time.Minute).Unix()}
+		rawData, _ := json.Marshal(data)
+		backup.Status.PluginMetadata[pluginName] = string(rawData)
+		backup.Annotations = map[string]string{utils.BackupVolumeSnapshotDeadlineAnnotationName: "10"}
+
+		exceeded, err := isDeadlineExceeded(backup)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(exceeded).To(BeTrue())
+	})
+})
+
+var _ = Describe("unmarshalMetadata", func() {
+	It("should unmarshal valid metadata correctly", func() {
+		rawData := `{"volumeSnapshotFirstFailure": 1234567890}`
+		data, err := unmarshalMetadata(rawData)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(data.VolumeSnapshotFirstDetectedFailure).To(Equal(int64(1234567890)))
+	})
+
+	It("should return an error if rawData is invalid JSON", func() {
+		rawData := `invalid-json`
+		data, err := unmarshalMetadata(rawData)
+		Expect(err).To(HaveOccurred())
+		Expect(data).To(BeNil())
+	})
+
+	It("should return an error if volumeSnapshotFirstFailure is missing", func() {
+		rawData := `{}`
+		data, err := unmarshalMetadata(rawData)
+		Expect(err).To(HaveOccurred())
+		Expect(data).To(BeNil())
+	})
+
+	It("should return an error if volumeSnapshotFirstFailure is zero", func() {
+		rawData := `{"volumeSnapshotFirstFailure": 0}`
+		data, err := unmarshalMetadata(rawData)
+		Expect(err).To(HaveOccurred())
+		Expect(data).To(BeNil())
+	})
+})