fix: validate restore volume binding lifecycle

controllers/dataprotection/volumepopulator_controller.go

@@ -173,7 +173,7 @@ func (r *VolumePopulatorReconciler) syncPVC(reqCtx intctrlutil.RequestCtx, pvc *
 		return nil
 	}
 	if !pvc.DeletionTimestamp.IsZero() {
-		return r.Cleanup(reqCtx, pvc)
+		return r.cleanupDeletingPVC(reqCtx, pvc)
 	}
 	var restoreCtx *pvcRestoreContext
 	if pvc.Spec.DataSourceRef.Kind == dptypes.RestoreKind {
@@ -188,10 +188,191 @@ func (r *VolumePopulatorReconciler) syncPVC(reqCtx intctrlutil.RequestCtx, pvc *
 	if pvc.Spec.VolumeName == "" {
 		return r.dispatchUnboundPVC(reqCtx, pvc, restoreCtx)
 	}
-	if err = r.completeBoundPVCIfNeeded(reqCtx, pvc, restoreCtx); err != nil {
+	if err = r.validateBoundPVCForCompletion(reqCtx, pvc, restoreCtx); err != nil {
 		return err
 	}
-	return r.Cleanup(reqCtx, pvc)
+	return r.completeBoundPVCIfNeeded(reqCtx, pvc, restoreCtx)
+}
+
+type boundPVCArtifacts struct {
+	targetPV         *corev1.PersistentVolume
+	populatePVC      *corev1.PersistentVolumeClaim
+	populatePV       *corev1.PersistentVolume
+	executionRestore *dpv1alpha1.Restore
+}
+
+func (a *boundPVCArtifacts) describe(pvc *corev1.PersistentVolumeClaim) string {
+	claimRefDescription := func(claimRef *corev1.ObjectReference) string {
+		if claimRef == nil {
+			return "<nil>"
+		}
+		return fmt.Sprintf("%s/%s uid=%s", claimRef.Namespace, claimRef.Name, claimRef.UID)
+	}
+	targetPVDescription := fmt.Sprintf("target PV %q not found", pvc.Spec.VolumeName)
+	if a.targetPV != nil {
+		targetPVDescription = fmt.Sprintf("target PV %q claimRef=%s %s=%q", a.targetPV.Name,
+			claimRefDescription(a.targetPV.Spec.ClaimRef), AnnPopulateFrom, a.targetPV.Annotations[AnnPopulateFrom])
+	}
+	populatePVCName := getPopulatePVCName(pvc.UID)
+	populatePVCDescription := fmt.Sprintf("helper PVC %s/%s not found", pvc.Namespace, populatePVCName)
+	if a.populatePVC != nil {
+		populatePVCDescription = fmt.Sprintf("helper PVC %s/%s volumeName=%q deleting=%t", a.populatePVC.Namespace,
+			a.populatePVC.Name, a.populatePVC.Spec.VolumeName, !a.populatePVC.DeletionTimestamp.IsZero())
+	}
+	populatePVDescription := "helper PV not found"
+	if a.populatePV != nil {
+		populatePVDescription = fmt.Sprintf("helper PV %q claimRef=%s %s=%q", a.populatePV.Name,
+			claimRefDescription(a.populatePV.Spec.ClaimRef), AnnPopulateFrom, a.populatePV.Annotations[AnnPopulateFrom])
+	}
+	restoreDescription := fmt.Sprintf("execution Restore %s/%s not found", pvc.Namespace, populatePVCName)
+	if a.executionRestore != nil {
+		restoreDescription = fmt.Sprintf("execution Restore %s/%s phase=%q", a.executionRestore.Namespace,
+			a.executionRestore.Name, a.executionRestore.Status.Phase)
+	}
+	return strings.Join([]string{
+		fmt.Sprintf("target PVC %s/%s volumeName=%q", pvc.Namespace, pvc.Name, pvc.Spec.VolumeName),
+		targetPVDescription,
+		populatePVCDescription,
+		populatePVDescription,
+		restoreDescription,
+	}, "; ")
+}
+
+func (r *VolumePopulatorReconciler) inspectBoundPVCArtifacts(reqCtx intctrlutil.RequestCtx,
+	pvc *corev1.PersistentVolumeClaim,
+	restoreCtx *pvcRestoreContext) (*boundPVCArtifacts, error) {
+	artifacts := &boundPVCArtifacts{}
+	populatePVCName := getPopulatePVCName(pvc.UID)
+	inspectionError := func(operation string, err error) error {
+		return intctrlutil.NewRequeueError(reconcileInterval, fmt.Sprintf(
+			"transient error while %s; will retry without releasing restore artifacts: %v; %s",
+			operation, err, artifacts.describe(pvc)))
+	}
+	if restoreCtx.mode == pvcRestoreModeRestoreData {
+		executionRestore := &dpv1alpha1.Restore{}
+		if err := r.Client.Get(reqCtx.Ctx, types.NamespacedName{Namespace: pvc.Namespace, Name: populatePVCName}, executionRestore); err != nil {
+			if !apierrors.IsNotFound(err) {
+				return nil, inspectionError(fmt.Sprintf("getting execution Restore %s/%s", pvc.Namespace, populatePVCName), err)
+			}
+		} else {
+			artifacts.executionRestore = executionRestore
+		}
+	}
+
+	populatePVC := &corev1.PersistentVolumeClaim{}
+	if err := r.Client.Get(reqCtx.Ctx, types.NamespacedName{Namespace: pvc.Namespace, Name: populatePVCName}, populatePVC); err != nil {
+		if !apierrors.IsNotFound(err) {
+			return nil, inspectionError(fmt.Sprintf("getting helper PVC %s/%s", pvc.Namespace, populatePVCName), err)
+		}
+	} else {
+		artifacts.populatePVC = populatePVC
+	}
+
+	targetPV := &corev1.PersistentVolume{}
+	if err := r.Client.Get(reqCtx.Ctx, types.NamespacedName{Name: pvc.Spec.VolumeName}, targetPV); err != nil {
+		if !apierrors.IsNotFound(err) {
+			return nil, inspectionError(fmt.Sprintf("getting target PV %q for target PVC %s/%s",
+				pvc.Spec.VolumeName, pvc.Namespace, pvc.Name), err)
+		}
+	} else {
+		artifacts.targetPV = targetPV
+	}
+
+	if artifacts.populatePVC != nil && artifacts.populatePVC.Spec.VolumeName != "" {
+		if artifacts.targetPV != nil && artifacts.populatePVC.Spec.VolumeName == artifacts.targetPV.Name {
+			artifacts.populatePV = artifacts.targetPV
+		} else {
+			populatePV := &corev1.PersistentVolume{}
+			if err := r.Client.Get(reqCtx.Ctx, types.NamespacedName{Name: artifacts.populatePVC.Spec.VolumeName}, populatePV); err != nil {
+				if !apierrors.IsNotFound(err) {
+					return nil, inspectionError(fmt.Sprintf("getting helper PV %q for helper PVC %s/%s",
+						artifacts.populatePVC.Spec.VolumeName, artifacts.populatePVC.Namespace, artifacts.populatePVC.Name), err)
+				}
+			} else {
+				artifacts.populatePV = populatePV
+			}
+		}
+	}
+	return artifacts, nil
+}
+
+// validateBoundPVCForCompletion proves that a bound target PVC was bound by this
+// populator and that its prepareData Restore has finished. volumeName alone is
+// insufficient: another provisioner may have bound an empty PV first.
+func (r *VolumePopulatorReconciler) validateBoundPVCForCompletion(reqCtx intctrlutil.RequestCtx,
+	pvc *corev1.PersistentVolumeClaim,
+	restoreCtx *pvcRestoreContext) error {
+	artifacts, err := r.inspectBoundPVCArtifacts(reqCtx, pvc, restoreCtx)
+	if err != nil {
+		return err
+	}
+	details := artifacts.describe(pvc)
+
+	if restoreCtx.mode == pvcRestoreModeRestoreData {
+		if artifacts.executionRestore == nil {
+			return intctrlutil.NewFatalError("bound target PVC has no execution Restore; refusing to treat restore as completed: " + details)
+		}
+		switch artifacts.executionRestore.Status.Phase {
+		case dpv1alpha1.RestorePhaseCompleted:
+			// Continue with binding provenance validation.
+		case dpv1alpha1.RestorePhaseFailed:
+			return intctrlutil.NewFatalError("execution Restore failed; preserving restore artifacts: " + details)
+		default:
+			message := "waiting for execution Restore before accepting target PVC binding; preserving restore artifacts: " + details
+			if err := r.UpdatePVCConditions(reqCtx, pvc, ReasonPopulatingProcessing, message); err != nil {
+				return err
+			}
+			return intctrlutil.NewRequeueError(reconcileInterval, message)
+		}
+	}
+
+	if artifacts.targetPV == nil {
+		return r.waitForBoundPVCObservation(reqCtx, pvc,
+			"target PV to become observable before validating restore binding", details)
+	}
+	if !pvClaimRefMatchesPVC(artifacts.targetPV.Spec.ClaimRef, pvc) {
+		if helperPVCBindingMayBeOutOfOrder(artifacts) {
+			return r.waitForBoundPVCObservation(reqCtx, pvc,
+				"target PV ClaimRef cache to reflect the completed helper-to-target rebind", details)
+		}
+		return intctrlutil.NewFatalError("target PV ClaimRef does not match target PVC; preserving restore artifacts: " + details)
+	}
+	expectedSource := pvc.Spec.DataSourceRef.Name
+	if artifacts.targetPV.Annotations[AnnPopulateFrom] != expectedSource {
+		if helperPVCBindingMayBeOutOfOrder(artifacts) {
+			return r.waitForBoundPVCObservation(reqCtx, pvc,
+				fmt.Sprintf("target PV annotation %s cache to reflect the completed helper-to-target rebind", AnnPopulateFrom), details)
+		}
+		return intctrlutil.NewFatalError(fmt.Sprintf(
+			"target PV %s annotation %s does not identify restore source %q; preserving restore artifacts: %s",
+			artifacts.targetPV.Name, AnnPopulateFrom, expectedSource, details))
+	}
+	if artifacts.populatePVC != nil {
+		if artifacts.populatePVC.Spec.VolumeName == "" {
+			return r.waitForBoundPVCObservation(reqCtx, pvc,
+				"helper PVC binding state to catch up before releasing restore artifacts", details)
+		}
+		if artifacts.populatePVC.Spec.VolumeName != artifacts.targetPV.Name {
+			return intctrlutil.NewFatalError("helper PVC and target PVC refer to different PVs; preserving restore artifacts: " + details)
+		}
+	}
+	return nil
+}
+
+func helperPVCBindingMayBeOutOfOrder(artifacts *boundPVCArtifacts) bool {
+	return artifacts.populatePVC != nil && artifacts.targetPV != nil &&
+		(artifacts.populatePVC.Spec.VolumeName == "" || artifacts.populatePVC.Spec.VolumeName == artifacts.targetPV.Name)
+}
+
+func (r *VolumePopulatorReconciler) waitForBoundPVCObservation(reqCtx intctrlutil.RequestCtx,
+	pvc *corev1.PersistentVolumeClaim,
+	observation,
+	details string) error {
+	message := fmt.Sprintf("waiting for %s; preserving restore artifacts: %s", observation, details)
+	if err := r.UpdatePVCConditions(reqCtx, pvc, ReasonPopulatingProcessing, message); err != nil {
+		return err
+	}
+	return intctrlutil.NewRequeueError(reconcileInterval, message)
 }
 
 // dispatchUnboundPVC routes an unbound PVC to either Populate or ProvisionOnly.
@@ -1014,7 +1195,7 @@ func (r *VolumePopulatorReconciler) completeBoundPVCIfNeeded(reqCtx intctrlutil.
 		// actions may need the workload pod to start, which cannot happen while
 		// the populate PVC still owns the restored PV or while the target PVC is
 		// still marked as being populated.
-		if err := r.Cleanup(reqCtx, pvc); err != nil {
+		if err := r.releasePopulateResources(reqCtx, pvc); err != nil {
 			return err
 		}
 		reason := ReasonPopulatingSucceed
@@ -1082,7 +1263,7 @@ func (r *VolumePopulatorReconciler) waitForSerialPredecessors(reqCtx intctrlutil
 		if cond != nil && cond.Status == corev1.ConditionFalse {
 			return intctrlutil.NewFatalError(fmt.Sprintf("previous restore PVC %s/%s failed: %s", item.Namespace, item.Name, cond.Message))
 		}
-		if item.Spec.VolumeName != "" {
+		if pvcPopulateReleased(item) {
 			continue
 		}
 		if err = r.UpdatePVCConditions(reqCtx, pvc, ReasonPopulatingProcessing,
@@ -1387,7 +1568,7 @@ func (r *VolumePopulatorReconciler) allRestorePVCsForComponentBound(reqCtx intct
 		if cond != nil && cond.Status == corev1.ConditionFalse {
 			return false, intctrlutil.NewFatalError(fmt.Sprintf("restore PVC %s/%s failed: %s", item.Namespace, item.Name, cond.Message))
 		}
-		if item.Spec.VolumeName == "" {
+		if !pvcPopulateReleased(item) {
 			return false, nil
 		}
 	}
@@ -1405,7 +1586,7 @@ func (r *VolumePopulatorReconciler) allRestorePVCsForClusterBound(reqCtx intctrl
 		if cond != nil && cond.Status == corev1.ConditionFalse {
 			return false, intctrlutil.NewFatalError(fmt.Sprintf("restore PVC %s/%s failed: %s", item.Namespace, item.Name, cond.Message))
 		}
-		if item.Spec.VolumeName == "" {
+		if !pvcPopulateReleased(item) {
 			return false, nil
 		}
 	}
@@ -1457,9 +1638,17 @@ func findPVCConditionByType(pvc *corev1.PersistentVolumeClaim, conditionType str
 }
 
 func pvcPopulateReleased(pvc *corev1.PersistentVolumeClaim) bool {
-	cond := findPVCConditionByType(pvc, string(PersistentVolumeClaimPopulating))
-	return cond != nil && cond.Status == corev1.ConditionTrue &&
-		(cond.Reason == ReasonPopulatingSucceed || cond.Reason == ReasonPopulatingProvisioned)
+	if pvc == nil {
+		return false
+	}
+	for i := range pvc.Status.Conditions {
+		condition := &pvc.Status.Conditions[i]
+		if condition.Type != PersistentVolumeClaimPopulating || condition.Status != corev1.ConditionTrue {
+			continue
+		}
+		return condition.Reason == ReasonPopulatingSucceed || condition.Reason == ReasonPopulatingProvisioned
+	}
+	return false
 }
 
 func (r *VolumePopulatorReconciler) listRestorePVCsForComponent(reqCtx intctrlutil.RequestCtx, pvc *corev1.PersistentVolumeClaim) ([]corev1.PersistentVolumeClaim, error) {
@@ -1613,7 +1802,37 @@ func postReadyRestoreName(componentUID types.UID) string {
 	return constant.ShortenKubeName(fmt.Sprintf("restore-%s-post-ready", componentUID), constant.KubeNameMaxLength)
 }
 
-func (r *VolumePopulatorReconciler) Cleanup(reqCtx intctrlutil.RequestCtx, pvc *corev1.PersistentVolumeClaim) error {
+// cleanupDeletingPVC stops the per-PVC execution Restore before releasing its
+// helper PVC and the target finalizer. This lets the Restore controller finish
+// deleting its Jobs before the target PVC disappears.
+func (r *VolumePopulatorReconciler) cleanupDeletingPVC(reqCtx intctrlutil.RequestCtx, pvc *corev1.PersistentVolumeClaim) error {
+	executionRestore := &dpv1alpha1.Restore{}
+	key := types.NamespacedName{Namespace: pvc.Namespace, Name: getPopulatePVCName(pvc.UID)}
+	if err := r.Client.Get(reqCtx.Ctx, key, executionRestore); err != nil {
+		if !apierrors.IsNotFound(err) {
+			return err
+		}
+	} else if hasOwnerReference(executionRestore.OwnerReferences, pvc.UID) {
+		if executionRestore.DeletionTimestamp.IsZero() {
+			if err := r.Client.Delete(reqCtx.Ctx, executionRestore); err != nil && !apierrors.IsNotFound(err) {
+				return err
+			}
+		}
+		return intctrlutil.NewRequeueError(reconcileInterval,
+			fmt.Sprintf("waiting for execution Restore %s/%s to be deleted before cleaning target PVC %s/%s",
+				executionRestore.Namespace, executionRestore.Name, pvc.Namespace, pvc.Name))
+	} else {
+		return intctrlutil.NewFatalError(fmt.Sprintf(
+			"execution Restore %s/%s is not owned by deleting target PVC %s/%s uid=%s; refusing to release helper resources",
+			executionRestore.Namespace, executionRestore.Name, pvc.Namespace, pvc.Name, pvc.UID))
+	}
+	return r.releasePopulateResources(reqCtx, pvc)
+}
+
+// releasePopulateResources is used only after binding provenance and Restore
+// completion have been verified, or after cleanupDeletingPVC has stopped the
+// execution Restore.
+func (r *VolumePopulatorReconciler) releasePopulateResources(reqCtx intctrlutil.RequestCtx, pvc *corev1.PersistentVolumeClaim) error {
 	populatePVC := &corev1.PersistentVolumeClaim{}
 	if err := r.Client.Get(reqCtx.Ctx, types.NamespacedName{Name: getPopulatePVCName(pvc.UID),
 		Namespace: pvc.Namespace}, populatePVC); err != nil {
@@ -1665,7 +1884,8 @@ func (r *VolumePopulatorReconciler) waitForPVCSelectedNode(reqCtx intctrlutil.Re
 		if storageClass.VolumeBindingMode != nil && storagev1.VolumeBindingWaitForFirstConsumer == *storageClass.VolumeBindingMode {
 			nodeName = pvc.Annotations[AnnSelectedNode]
 			if nodeName == "" {
-				// Wait for the PVC to get a node name before continuing
+				// Wait for the workload Pod to select a node. The target PVC remains
+				// unbound, so the Pod cannot start before population completes.
 				return true, nodeName, nil
 			}
 		}