Identify the backed-up volume by CSI volume handle in the in-place restore pre-flight check (#10530)

The pre-flight check that verifies the existing PVC is still bound to the
backed-up volume compared PV names. A block data mover restore of a file
system volume recreates the PV under a new name (the volumeMode field is
immutable), so a second in-place restore of the same workload failed the
check even though the PVC was bound to the very same volume.

Record the CSI volume handle in the backup volume info (PVInfo) and
compare handles when both the backup and the bound PV record one; the
PV name remains the fallback for non-CSI volumes and for backups taken
before the handle was recorded. The handle reaches the PVC CSI RIA
through the same carrier annotation mechanism as the source size.

Signed-off-by: chlins <chlins.zhang@gmail.com>

changelogs/unreleased/10530-chlins

@@ -0,0 +1 @@
+Identify the backed-up volume by CSI volume handle in the in-place restore pre-flight check

internal/volume/volumes_information.go

@@ -384,6 +384,21 @@ type PVInfo struct {
 
 	// The PV's labels should be kept after recreation.
 	Labels map[string]string `json:"labels"`
+
+	// VolumeHandle is the CSI volume handle of the PV, identifying the underlying
+	// volume independently of the PV name. Empty for non-CSI volumes.
+	VolumeHandle string `json:"volumeHandle,omitempty"`
+}
+
+func newPVInfo(pv *corev1api.PersistentVolume) *PVInfo {
+	info := &PVInfo{
+		ReclaimPolicy: string(pv.Spec.PersistentVolumeReclaimPolicy),
+		Labels:        pv.Labels,
+	}
+	if pv.Spec.CSI != nil {
+		info.VolumeHandle = pv.Spec.CSI.VolumeHandle
+	}
+	return info
 }
 
 // BackupVolumesInformation contains the information needs by generating
@@ -460,10 +475,7 @@ func (v *BackupVolumesInformation) generateVolumeInfoForSkippedPV() {
 				SnapshotDataMoved: false,
 				Skipped:           true,
 				SkippedReason:     skippedReason,
-				PVInfo: &PVInfo{
-					ReclaimPolicy: string(pvcPVInfo.PV.Spec.PersistentVolumeReclaimPolicy),
-					Labels:        pvcPVInfo.PV.Labels,
-				},
+				PVInfo:            newPVInfo(&pvcPVInfo.PV),
 			}
 			tmpVolumeInfos = append(tmpVolumeInfos, volumeInfo)
 		} else {
@@ -496,10 +508,7 @@ func (v *BackupVolumesInformation) generateVolumeInfoForVeleroNativeSnapshot() {
 				// although NativeSnapshot doesn't check whether the snapshot creation result.
 				Result:             volumeResult,
 				NativeSnapshotInfo: newNativeSnapshotInfo(nativeSnapshot),
-				PVInfo: &PVInfo{
-					ReclaimPolicy: string(pvcPVInfo.PV.Spec.PersistentVolumeReclaimPolicy),
-					Labels:        pvcPVInfo.PV.Labels,
-				},
+				PVInfo:             newPVInfo(&pvcPVInfo.PV),
 			}
 			tmpVolumeInfos = append(tmpVolumeInfos, volumeInfo)
 		} else {
@@ -591,10 +600,7 @@ func (v *BackupVolumesInformation) generateVolumeInfoForCSIVolumeSnapshot() {
 					ReadyToUse:                volumeSnapshot.Status.ReadyToUse,
 					VolumeGroupSnapshotHandle: volumeGroupSnapshotHandle,
 				},
-				PVInfo: &PVInfo{
-					ReclaimPolicy: string(pvcPVInfo.PV.Spec.PersistentVolumeReclaimPolicy),
-					Labels:        pvcPVInfo.PV.Labels,
-				},
+				PVInfo: newPVInfo(&pvcPVInfo.PV),
 			}
 
 			if volumeSnapshot.Status.CreationTime != nil {
@@ -648,10 +654,7 @@ func (v *BackupVolumesInformation) generateVolumeInfoFromPVB() {
 				volumeInfo.PVCName = pvcPVInfo.PVCName
 				volumeInfo.PVCNamespace = pvcPVInfo.PVCNamespace
 				volumeInfo.PVName = pvcPVInfo.PV.Name
-				volumeInfo.PVInfo = &PVInfo{
-					ReclaimPolicy: string(pvcPVInfo.PV.Spec.PersistentVolumeReclaimPolicy),
-					Labels:        pvcPVInfo.PV.Labels,
-				}
+				volumeInfo.PVInfo = newPVInfo(&pvcPVInfo.PV)
 			} else {
 				v.logger.Warnf("Cannot find info for PVC %s/%s", pvb.Spec.Pod.Namespace, pvcName)
 				continue
@@ -760,10 +763,7 @@ func (v *BackupVolumesInformation) generateVolumeInfoFromDataUpload() {
 					Size:           dataUpload.Status.Progress.TotalBytes,
 					SnapshotHandle: dataUpload.Status.SnapshotID,
 				},
-				PVInfo: &PVInfo{
-					ReclaimPolicy: string(pvcPVInfo.PV.Spec.PersistentVolumeReclaimPolicy),
-					Labels:        pvcPVInfo.PV.Labels,
-				},
+				PVInfo: newPVInfo(&pvcPVInfo.PV),
 			}
 
 			if dataUpload.Spec.ParentSnapshot == veleroshared.ParentSnapshotNone {

internal/volume/volumes_information_test.go

@@ -1685,3 +1685,13 @@ func TestNewPodVolumeInfoFromPVB(t *testing.T) {
 		})
 	}
 }
+
+func TestNewPVInfo(t *testing.T) {
+	csiPV := builder.ForPersistentVolume("pv-1").ReclaimPolicy(corev1api.PersistentVolumeReclaimRetain).
+		ObjectMeta(builder.WithLabels("k", "v")).Result()
+	csiPV.Spec.CSI = &corev1api.CSIPersistentVolumeSource{Driver: "fake.csi", VolumeHandle: "vol-1"}
+	require.Equal(t, &PVInfo{ReclaimPolicy: "Retain", Labels: map[string]string{"k": "v"}, VolumeHandle: "vol-1"}, newPVInfo(csiPV))
+
+	localPV := builder.ForPersistentVolume("pv-2").ReclaimPolicy(corev1api.PersistentVolumeReclaimDelete).Result()
+	require.Equal(t, &PVInfo{ReclaimPolicy: "Delete"}, newPVInfo(localPV))
+}

pkg/apis/velero/v1/labels_annotations.go

@@ -192,6 +192,14 @@ const (
 	// run the in-place restore capacity pre-flight check without access to the volume info.
 	// The annotation is always stripped by the restore engine; it never lands on the cluster.
 	InplaceRestoreSourceSizeAnnotation = "restore.velero.io/inplace-restore-source-size"
+
+	// InplaceRestoreVolumeHandleAnnotation is a Velero-internal carrier annotation set by the
+	// restore engine on a PVC item before RestoreItemActions run. It carries the CSI volume
+	// handle of the PV the PVC was bound to at backup time, recorded in the backup volume info,
+	// so the PVC CSI RestoreItemAction can verify the existing PVC is still bound to the
+	// backed-up volume. The annotation is always stripped by the restore engine; it never lands
+	// on the cluster.
+	InplaceRestoreVolumeHandleAnnotation = "restore.velero.io/inplace-restore-volume-handle"
 	// SkippedNoCSIPVAnnotation - Velero checks this annotation on processed PVC to
 	// find out if the snapshot was skipped b/c the PV is not provisioned via CSI
 	SkippedNoCSIPVAnnotation = "backup.velero.io/skipped-no-csi-pv"

pkg/podvolume/restorer.go

@@ -190,7 +190,11 @@ func (r *restorer) RestorePodVolumes(data RestoreData, tracker *volume.RestoreVo
 		// restore's PodVolumeRestores complete.
 		if data.Restore.IsVolumeDataInplaceRestore() && pvc != nil {
 			pvName := backedUpPVName(data.BackupVolumeInfos, data.SourceNamespace, pvc.Name)
-			if err := inplace.CheckPVCBoundToBackedUpPV(pvc, pvName, data.SourceNamespace); err != nil {
+			var volumeHandle string
+			if info := data.BackupVolumeInfos[pvName].PVInfo; info != nil {
+				volumeHandle = info.VolumeHandle
+			}
+			if err := inplace.CheckPVCBoundToBackedUpVolume(r.ctx, r.crClient, pvc, pvName, volumeHandle, data.SourceNamespace); err != nil {
 				errs = append(errs, err)
 				continue
 			}

pkg/podvolume/restorer_test.go

@@ -367,6 +367,7 @@ func TestRestorePodVolumes(t *testing.T) {
 			inplace: true,
 			kubeClientObj: []runtime.Object{
 				createNodeAgentDaemonset(),
+				createPVObj(1, false),
 				createPVCObj(1),
 				func() *corev1api.Pod {
 					pod := builder.ForPod("fake-ns", "other-pod").
@@ -398,6 +399,7 @@ func TestRestorePodVolumes(t *testing.T) {
 			inplace: true,
 			kubeClientObj: []runtime.Object{
 				createNodeAgentDaemonset(),
+				createPVObj(1, false),
 				createPVCObj(1),
 				createGatedPodObj("old-restore-uid", 1),
 			},
@@ -423,6 +425,7 @@ func TestRestorePodVolumes(t *testing.T) {
 			inplace: true,
 			kubeClientObj: []runtime.Object{
 				createNodeAgentDaemonset(),
+				createPVObj(1, false),
 				createPVCObj(1),
 			},
 			ctlClientObj: []runtime.Object{
@@ -440,6 +443,72 @@ func TestRestorePodVolumes(t *testing.T) {
 				},
 			},
 		},
+		{
+			// The PV was recreated under a new name by a previous in-place restore
+			// (block data mover on a file system volume) but is the same CSI volume.
+			name: "in-place restore proceeds when the PVC is bound to the backed-up volume under a recreated PV name",
+			pvbs: []*velerov1api.PodVolumeBackup{
+				createPVBObj(true, true, 1, "kopia"),
+			},
+			inplace: true,
+			kubeClientObj: []runtime.Object{
+				createNodeAgentDaemonset(),
+				createNodeObj(),
+				func() *corev1api.PersistentVolume {
+					pv := createPVObj(1, false)
+					pv.Spec.CSI = &corev1api.CSIPersistentVolumeSource{Driver: "fake.csi", VolumeHandle: "vol-1"}
+					return pv
+				}(),
+				createPVCObj(1),
+				createGatedPodObj("fake-restore-uid", 1),
+				createNodeAgentPodObj(true),
+			},
+			ctlClientObj: []runtime.Object{
+				createBackupRepoObj(),
+			},
+			restoredPod:     createPodObj(true, true, true, 1),
+			sourceNamespace: "fake-ns",
+			bsl:             "fake-bsl",
+			volumeInfos: map[string]volume.BackupVolumeInfo{
+				"backed-up-pv": {PVCNamespace: "fake-ns", PVCName: "fake-pvc-1", PVInfo: &volume.PVInfo{VolumeHandle: "vol-1"}},
+			},
+			runtimeScheme: scheme,
+			retPVRs: []*velerov1api.PodVolumeRestore{
+				completedPVR,
+			},
+		},
+		{
+			name: "in-place restore blocked when the PVC is bound to a different CSI volume",
+			pvbs: []*velerov1api.PodVolumeBackup{
+				createPVBObj(true, true, 1, "kopia"),
+			},
+			inplace: true,
+			kubeClientObj: []runtime.Object{
+				createNodeAgentDaemonset(),
+				func() *corev1api.PersistentVolume {
+					pv := createPVObj(1, false)
+					pv.Spec.CSI = &corev1api.CSIPersistentVolumeSource{Driver: "fake.csi", VolumeHandle: "vol-other"}
+					return pv
+				}(),
+				createPVCObj(1),
+			},
+			ctlClientObj: []runtime.Object{
+				createBackupRepoObj(),
+			},
+			restoredPod:     createPodObj(true, true, true, 1),
+			sourceNamespace: "fake-ns",
+			bsl:             "fake-bsl",
+			volumeInfos: map[string]volume.BackupVolumeInfo{
+				"fake-pv-1": {PVCNamespace: "fake-ns", PVCName: "fake-pvc-1", PVInfo: &volume.PVInfo{VolumeHandle: "vol-1"}},
+			},
+			runtimeScheme: scheme,
+			errs: []expectError{
+				{
+					err:        "in-place restore pre-flight check failed, skipping volume data restore: PVC fake-ns/fake-pvc-1 is bound to volume vol-other (PV fake-pv-1), but was bound to volume vol-1 (PV fake-pv-1) at backup time",
+					prefixOnly: true,
+				},
+			},
+		},
 		{
 			name: "in-place restore blocked when the PVC is too small for the source volume",
 			pvbs: []*velerov1api.PodVolumeBackup{
@@ -448,6 +517,7 @@ func TestRestorePodVolumes(t *testing.T) {
 			inplace: true,
 			kubeClientObj: []runtime.Object{
 				createNodeAgentDaemonset(),
+				createPVObj(1, false),
 				func() *corev1api.PersistentVolumeClaim {
 					pvc := createPVCObj(1)
 					pvc.Status.Capacity = corev1api.ResourceList{corev1api.ResourceStorage: resource.MustParse("100Mi")}
@@ -479,6 +549,7 @@ func TestRestorePodVolumes(t *testing.T) {
 			inplace: true,
 			kubeClientObj: []runtime.Object{
 				createNodeAgentDaemonset(),
+				createPVObj(1, false),
 				createNodeObj(),
 				createPVCObj(1),
 				createGatedPodObj("fake-restore-uid", 1),

pkg/restore/actions/csi/pvc_action.go

@@ -247,7 +247,7 @@ func (p *pvcRestoreItemAction) executeWithDataMove(logger *logrus.Entry, input *
 	restoreType := input.Restore.Spec.ExistingVolumeDataPolicy
 	if pvcExists {
 		// Pre-flight checks must pass before any side effect on the existing PVC/PV.
-		if err := inplace.CheckPVCBoundToBackedUpPV(existingPVC, pvcFromBackup.Spec.VolumeName, pvcFromBackup.Namespace); err != nil {
+		if err := inplace.CheckPVCBoundToBackedUpVolume(ctx, p.crClient, existingPVC, pvcFromBackup.Spec.VolumeName, pvc.Annotations[velerov1api.InplaceRestoreVolumeHandleAnnotation], pvcFromBackup.Namespace); err != nil {
 			return nil, errors.WithStack(err)
 		}
 		if err := inplace.CheckPVCCapacity(existingPVC, sourceSizeFromCarrier(pvc)); err != nil {

pkg/restore/actions/csi/pvc_action_test.go

@@ -843,12 +843,14 @@ func TestExecuteInplaceRestorePreflight(t *testing.T) {
 	}
 
 	tests := []struct {
-		name           string
-		pod            *corev1api.Pod
-		backedUpPVName string
-		sourceSize     string // carried on the PVC item by the restore engine
-		pvcCapacity    string
-		expectBlock    string
+		name             string
+		pod              *corev1api.Pod
+		backedUpPVName   string
+		backedUpHandle   string // carried on the PVC item by the restore engine
+		existingPVHandle string // CSI volume handle of the PV bound to the existing PVC
+		sourceSize       string // carried on the PVC item by the restore engine
+		pvcCapacity      string
+		expectBlock      string
 	}{
 		{
 			name:           "checks pass, restore proceeds",
@@ -865,6 +867,22 @@ func TestExecuteInplaceRestorePreflight(t *testing.T) {
 			backedUpPVName: "backupPV",
 			expectBlock:    "was bound to PV backupPV at backup time",
 		},
+		{
+			// The PV was recreated under a new name by a previous in-place
+			// restore (block data mover on a file system volume) but is the
+			// same CSI volume.
+			name:             "PVC bound to the backed-up volume under a recreated PV name proceeds",
+			backedUpPVName:   "backupPV",
+			backedUpHandle:   "vol-1",
+			existingPVHandle: "vol-1",
+		},
+		{
+			name:             "PVC bound to a different CSI volume under the same PV name blocks the restore",
+			backedUpPVName:   "testPV",
+			backedUpHandle:   "vol-1",
+			existingPVHandle: "vol-other",
+			expectBlock:      "is bound to volume vol-other (PV testPV), but was bound to volume vol-1 (PV testPV) at backup time",
+		},
 		{
 			// Backed-up PV unknown so the same-volume skip does not apply.
 			name:        "PVC smaller than the source volume blocks the restore",
@@ -887,6 +905,9 @@ func TestExecuteInplaceRestorePreflight(t *testing.T) {
 				existingPVC.Status.Capacity = corev1api.ResourceList{corev1api.ResourceStorage: resource.MustParse(tc.pvcCapacity)}
 			}
 			existingPV := builder.ForPersistentVolume("testPV").Result()
+			if tc.existingPVHandle != "" {
+				existingPV.Spec.CSI = &corev1api.CSIPersistentVolumeSource{Driver: "fake.csi", VolumeHandle: tc.existingPVHandle}
+			}
 			backup := builder.ForBackup("velero", "testBackup").SnapshotMoveData(true).Result()
 			restore := builder.ForRestore("velero", "testRestore").Backup("testBackup").
 				ObjectMeta(builder.WithUID("uid")).ExistingVolumeDataPolicy("full").Result()
@@ -920,6 +941,9 @@ func TestExecuteInplaceRestorePreflight(t *testing.T) {
 			if tc.sourceSize != "" {
 				item.Annotations[velerov1api.InplaceRestoreSourceSizeAnnotation] = tc.sourceSize
 			}
+			if tc.backedUpHandle != "" {
+				item.Annotations[velerov1api.InplaceRestoreVolumeHandleAnnotation] = tc.backedUpHandle
+			}
 			pvcMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(item)
 			require.NoError(t, err)
 			pvcFromBackupMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(pvcFromBackup)

pkg/restore/inplace/preflight.go

@@ -135,15 +135,25 @@ func gatedByThisRestore(pod *corev1api.Pod, restoreUID types.UID) bool {
 	return true
 }
 
-// CheckPVCBoundToBackedUpPV verifies the existing PVC is still bound to the
-// same PV it was bound to at backup time. An in-place restore onto a
-// different volume is unsafe: an incremental (CBT) restore computes deltas
-// against a different volume lineage, and even a full restore would patch and
-// write into a volume unrelated to the backup. The PV comparison is skipped
-// when the PVC is restored into a different namespace, where it is necessarily
-// bound to a different PV (the documented cross-namespace clone-and-restore
-// workflow), and when the backed-up PV name is unknown.
-func CheckPVCBoundToBackedUpPV(existingPVC *corev1api.PersistentVolumeClaim, backedUpPVName, sourceNamespace string) error {
+// CheckPVCBoundToBackedUpVolume verifies the existing PVC is still bound to the
+// volume that was backed up. An in-place restore onto a different volume is
+// unsafe: an incremental (CBT) restore computes deltas against a different
+// volume lineage, and even a full restore would patch and write into a volume
+// unrelated to the backup.
+//
+// The volume is identified by its CSI volume handle when the backup recorded
+// one: Velero itself recreates the PV under a new name during a block data
+// mover restore of a file system volume, so the PV name alone is not a stable
+// identity. Non-CSI volumes fall back to the PV name. The comparison is
+// skipped when the PVC is restored into a different namespace, where it is
+// necessarily bound to a different volume (the documented cross-namespace
+// clone-and-restore workflow), and when the backed-up volume is unknown.
+func CheckPVCBoundToBackedUpVolume(
+	ctx context.Context,
+	cli crclient.Client,
+	existingPVC *corev1api.PersistentVolumeClaim,
+	backedUpPVName, backedUpVolumeHandle, sourceNamespace string,
+) error {
 	if existingPVC == nil {
 		return errors.New("existing PVC cannot be nil")
 	}
@@ -151,11 +161,29 @@ func CheckPVCBoundToBackedUpPV(existingPVC *corev1api.PersistentVolumeClaim, bac
 		return errors.Errorf("in-place restore pre-flight check failed, skipping volume data restore: PVC %s/%s is not bound (phase %s)",
 			existingPVC.Namespace, existingPVC.Name, existingPVC.Status.Phase)
 	}
-	if existingPVC.Namespace != sourceNamespace || backedUpPVName == "" || existingPVC.Spec.VolumeName == backedUpPVName {
+	if existingPVC.Namespace != sourceNamespace || backedUpPVName == "" {
 		return nil
 	}
-	return errors.Errorf("in-place restore pre-flight check failed, skipping volume data restore: PVC %s/%s is bound to PV %s, but was bound to PV %s at backup time",
-		existingPVC.Namespace, existingPVC.Name, existingPVC.Spec.VolumeName, backedUpPVName)
+
+	if backedUpVolumeHandle != "" {
+		pv := new(corev1api.PersistentVolume)
+		if err := cli.Get(ctx, crclient.ObjectKey{Name: existingPVC.Spec.VolumeName}, pv); err != nil {
+			return errors.Wrapf(err, "failed to get PV %s bound to PVC %s/%s", existingPVC.Spec.VolumeName, existingPVC.Namespace, existingPVC.Name)
+		}
+		if pv.Spec.CSI != nil && pv.Spec.CSI.VolumeHandle != "" {
+			if pv.Spec.CSI.VolumeHandle != backedUpVolumeHandle {
+				return errors.Errorf("in-place restore pre-flight check failed, skipping volume data restore: PVC %s/%s is bound to volume %s (PV %s), but was bound to volume %s (PV %s) at backup time",
+					existingPVC.Namespace, existingPVC.Name, pv.Spec.CSI.VolumeHandle, pv.Name, backedUpVolumeHandle, backedUpPVName)
+			}
+			return nil
+		}
+	}
+
+	if existingPVC.Spec.VolumeName != backedUpPVName {
+		return errors.Errorf("in-place restore pre-flight check failed, skipping volume data restore: PVC %s/%s is bound to PV %s, but was bound to PV %s at backup time",
+			existingPVC.Namespace, existingPVC.Name, existingPVC.Spec.VolumeName, backedUpPVName)
+	}
+	return nil
 }
 
 // CheckPVCCapacity verifies the existing PVC is large enough to hold the

pkg/restore/inplace/preflight_test.go

@@ -216,7 +216,7 @@ func TestCheckPVCNotInUse(t *testing.T) {
 	}
 }
 
-func TestCheckPVCBoundToBackedUpPV(t *testing.T) {
+func TestCheckPVCBoundToBackedUpVolume(t *testing.T) {
 	pvc := func(namespace, pvName string, phase corev1api.PersistentVolumeClaimPhase) *corev1api.PersistentVolumeClaim {
 		return &corev1api.PersistentVolumeClaim{
 			ObjectMeta: metav1.ObjectMeta{Name: "pvc-1", Namespace: namespace},
@@ -224,57 +224,117 @@ func TestCheckPVCBoundToBackedUpPV(t *testing.T) {
 			Status:     corev1api.PersistentVolumeClaimStatus{Phase: phase},
 		}
 	}
+	csiPV := func(name, handle string) *corev1api.PersistentVolume {
+		return &corev1api.PersistentVolume{
+			ObjectMeta: metav1.ObjectMeta{Name: name},
+			Spec: corev1api.PersistentVolumeSpec{PersistentVolumeSource: corev1api.PersistentVolumeSource{
+				CSI: &corev1api.CSIPersistentVolumeSource{VolumeHandle: handle},
+			}},
+		}
+	}
+	localPV := func(name string) *corev1api.PersistentVolume {
+		return &corev1api.PersistentVolume{ObjectMeta: metav1.ObjectMeta{Name: name}}
+	}
 
 	tests := []struct {
 		name           string
 		existingPVC    *corev1api.PersistentVolumeClaim
+		existingPV     *corev1api.PersistentVolume
 		backedUpPVName string
+		backedUpHandle string
 		expectError    string
 	}{
 		{
 			name:           "nil existing PVC returns error",
-			existingPVC:    nil,
 			backedUpPVName: "pv-1",
 			expectError:    "existing PVC cannot be nil",
 		},
 		{
-			name:           "bound to the backed-up PV, check passes",
+			name:           "same volume handle under the same PV name, check passes",
 			existingPVC:    pvc("default", "pv-1", corev1api.ClaimBound),
+			existingPV:     csiPV("pv-1", "vol-1"),
+			backedUpPVName: "pv-1",
+			backedUpHandle: "vol-1",
+		},
+		{
+			name:           "same volume handle under a recreated PV name, check passes",
+			existingPVC:    pvc("default", "pv-recreated", corev1api.ClaimBound),
+			existingPV:     csiPV("pv-recreated", "vol-1"),
+			backedUpPVName: "pv-1",
+			backedUpHandle: "vol-1",
+		},
+		{
+			name:           "different volume handle under the same PV name, check fails",
+			existingPVC:    pvc("default", "pv-1", corev1api.ClaimBound),
+			existingPV:     csiPV("pv-1", "vol-other"),
+			backedUpPVName: "pv-1",
+			backedUpHandle: "vol-1",
+			expectError:    "is bound to volume vol-other (PV pv-1), but was bound to volume vol-1 (PV pv-1) at backup time",
+		},
+		{
+			name:           "bound PV not found, check fails",
+			existingPVC:    pvc("default", "pv-gone", corev1api.ClaimBound),
+			backedUpPVName: "pv-1",
+			backedUpHandle: "vol-1",
+			expectError:    "failed to get PV pv-gone bound to PVC default/pvc-1",
+		},
+		{
+			name:           "non-CSI volume with the same PV name, check passes",
+			existingPVC:    pvc("default", "pv-1", corev1api.ClaimBound),
+			existingPV:     localPV("pv-1"),
 			backedUpPVName: "pv-1",
 		},
 		{
-			name:           "bound to a different PV, check fails",
+			name:           "non-CSI volume with a different PV name, check fails",
 			existingPVC:    pvc("default", "pv-other", corev1api.ClaimBound),
+			existingPV:     localPV("pv-other"),
 			backedUpPVName: "pv-1",
 			expectError:    "is bound to PV pv-other, but was bound to PV pv-1 at backup time",
 		},
+		{
+			name:           "backup without volume handle falls back to the PV name",
+			existingPVC:    pvc("default", "pv-other", corev1api.ClaimBound),
+			existingPV:     csiPV("pv-other", "vol-1"),
+			backedUpPVName: "pv-1",
+			expectError:    "is bound to PV pv-other, but was bound to PV pv-1 at backup time",
+		},
+		{
+			name:           "existing PV without volume handle falls back to the PV name",
+			existingPVC:    pvc("default", "pv-other", corev1api.ClaimBound),
+			existingPV:     localPV("pv-other"),
+			backedUpPVName: "pv-1",
+			backedUpHandle: "vol-1",
+			expectError:    "is bound to PV pv-other, but was bound to PV pv-1 at backup time",
+		},
 		{
 			name:           "PVC not bound, check fails",
 			existingPVC:    pvc("default", "", corev1api.ClaimPending),
 			backedUpPVName: "pv-1",
+			backedUpHandle: "vol-1",
 			expectError:    "is not bound (phase Pending)",
 		},
 		{
-			name:           "different PV in a different namespace, check passes",
+			name:           "different volume in a different namespace, check passes",
 			existingPVC:    pvc("mapped-ns", "pv-other", corev1api.ClaimBound),
+			existingPV:     csiPV("pv-other", "vol-other"),
 			backedUpPVName: "pv-1",
+			backedUpHandle: "vol-1",
 		},
 		{
-			name:           "backed-up PV name unknown, check passes",
-			existingPVC:    pvc("default", "pv-other", corev1api.ClaimBound),
-			backedUpPVName: "",
-		},
-		{
-			name:           "different namespace but PVC not bound, check still fails",
-			existingPVC:    pvc("mapped-ns", "", corev1api.ClaimLost),
-			backedUpPVName: "pv-1",
-			expectError:    "is not bound (phase Lost)",
+			name:        "backed-up volume unknown, check passes",
+			existingPVC: pvc("default", "pv-other", corev1api.ClaimBound),
+			existingPV:  csiPV("pv-other", "vol-other"),
 		},
 	}
 
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
-			err := CheckPVCBoundToBackedUpPV(tc.existingPVC, tc.backedUpPVName, "default")
+			objs := []runtime.Object{}
+			if tc.existingPV != nil {
+				objs = append(objs, tc.existingPV)
+			}
+			cli := velerotest.NewFakeControllerRuntimeClient(t, objs...)
+			err := CheckPVCBoundToBackedUpVolume(t.Context(), cli, tc.existingPVC, tc.backedUpPVName, tc.backedUpHandle, "default")
 			if tc.expectError == "" {
 				require.NoError(t, err)
 				return

pkg/restore/restore.go

@@ -1644,16 +1644,22 @@ func (ctx *restoreContext) restoreItem(obj *unstructured.Unstructured, groupReso
 	// newly provisioned PVC to a stale node.
 	stripInplaceRestoreCarrierAnnotations(obj)
 
-	// Carry the source volume size from the backup volume info to the PVC CSI RIA, which has no
-	// access to the volume info, so it can run the in-place restore capacity pre-flight check.
+	// Carry backup volume info the PVC CSI RIA needs for the in-place restore pre-flight
+	// checks but has no access to: the source volume size and the backed-up volume handle.
 	if groupResource == kuberesource.PersistentVolumeClaims {
 		pvName, _, _ := unstructured.NestedString(obj.Object, "spec", "volumeName")
-		if sourceSize := ctx.backupVolumeInfoMap[pvName].SourceSize(); sourceSize > 0 {
-			annotations := obj.GetAnnotations()
-			if annotations == nil {
-				annotations = map[string]string{}
-			}
+		volumeInfo := ctx.backupVolumeInfoMap[pvName]
+		annotations := obj.GetAnnotations()
+		if annotations == nil {
+			annotations = map[string]string{}
+		}
+		if sourceSize := volumeInfo.SourceSize(); sourceSize > 0 {
 			annotations[velerov1api.InplaceRestoreSourceSizeAnnotation] = strconv.FormatInt(sourceSize, 10)
+		}
+		if volumeInfo.PVInfo != nil && volumeInfo.PVInfo.VolumeHandle != "" {
+			annotations[velerov1api.InplaceRestoreVolumeHandleAnnotation] = volumeInfo.PVInfo.VolumeHandle
+		}
+		if len(annotations) > 0 {
 			obj.SetAnnotations(annotations)
 		}
 	}
@@ -2526,6 +2532,7 @@ func resetMetadataAndStatus(obj *unstructured.Unstructured) (*unstructured.Unstr
 var inplaceRestoreCarrierAnnotations = []string{
 	velerov1api.InplaceRestoreSelectedNodeAnnotation,
 	velerov1api.InplaceRestoreSourceSizeAnnotation,
+	velerov1api.InplaceRestoreVolumeHandleAnnotation,
 }
 
 func stripInplaceRestoreCarrierAnnotations(obj metav1.Object) {

pkg/restore/restore_test.go

@@ -5299,6 +5299,33 @@ func TestRestoreInplaceSourceSizeCarrierAnnotation(t *testing.T) {
 		assert.NotContains(t, got.GetAnnotations(), velerov1api.InplaceRestoreSourceSizeAnnotation)
 	})
 
+	t.Run("volume handle from volume info is carried to RIAs and stripped from the cluster object", func(t *testing.T) {
+		h := newHarness(t)
+		h.AddItems(t, test.PVCs())
+		var seen string
+		capture := &pluggableAction{
+			executeFunc: func(input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
+				item := input.Item.(*unstructured.Unstructured)
+				seen = item.GetAnnotations()[velerov1api.InplaceRestoreVolumeHandleAnnotation]
+				return &velero.RestoreItemActionExecuteOutput{UpdatedItem: item}, nil
+			},
+		}
+
+		warnings, errs := h.restorer.Restore(
+			newRequest(t, h, map[string]volume.BackupVolumeInfo{
+				"pv-1": {PVCNamespace: "ns-1", PVCName: "pvc-1", PVInfo: &volume.PVInfo{VolumeHandle: "vol-1"}},
+			}),
+			[]riav2.RestoreItemAction{capture},
+			nil,
+		)
+		assertEmptyResults(t, warnings, errs)
+		assert.Equal(t, "vol-1", seen)
+
+		got, err := h.DynamicClient.Resource(test.PVCs().GVR()).Namespace("ns-1").Get(t.Context(), "pvc-1", metav1.GetOptions{})
+		require.NoError(t, err)
+		assert.NotContains(t, got.GetAnnotations(), velerov1api.InplaceRestoreVolumeHandleAnnotation)
+	})
+
 	t.Run("no carrier when the volume info has no source size", func(t *testing.T) {
 		h := newHarness(t)
 		h.AddItems(t, test.PVCs())