Align volume ownership semantics

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Copilot-Session: adc97cfc-61a5-473f-b0a2-e1b981651796

Author: danegsta

api/v2/physical_container_volume_types.go

@@ -73,11 +73,14 @@ type PhysicalContainerVolumeSpec struct {
 	// VolumeName is the runtime name to use when creating a new volume. Required when volumeID is omitted.
 	VolumeName string `json:"volumeName,omitempty"`
 
-	// Persistent keeps the runtime volume in place when this resource is deleted.
-	// By default the runtime volume is removed, including when this resource only tracks a
-	// volume it did not create.
+	// Persistent keeps a runtime volume created by this resource in place when the resource is deleted.
+	// Existing runtime volumes referenced by volumeID are always retained.
 	Persistent bool `json:"persistent,omitempty"`
 
+	// ReplaceExisting removes an existing runtime volume with volumeName before creating a new one.
+	// Replacement waits while the existing volume is in use and never removes attached containers.
+	ReplaceExisting bool `json:"replaceExisting,omitempty"`
+
 	// Labels contains labels to apply to a newly-created runtime volume.
 	// +listType=map
 	// +listMapKey=key
@@ -185,6 +188,9 @@ func (pv *PhysicalContainerVolume) Validate(ctx context.Context) field.ErrorList
 	errorList = append(errorList, commonapi.ValidateAnnotationsSize(pv.Annotations, field.NewPath("metadata", "annotations"))...)
 
 	if pv.Spec.VolumeID != "" {
+		if pv.Spec.Persistent {
+			errorList = append(errorList, field.Forbidden(specPath.Child("persistent"), "persistent cannot be set when volumeID is set"))
+		}
 		if strings.TrimSpace(pv.Spec.VolumeID) != pv.Spec.VolumeID {
 			errorList = append(errorList, field.Invalid(specPath.Child("volumeID"), pv.Spec.VolumeID, "volumeID must not have leading or trailing whitespace"))
 		}
@@ -194,6 +200,9 @@ func (pv *PhysicalContainerVolume) Validate(ctx context.Context) field.ErrorList
 		if len(pv.Spec.Labels) > 0 {
 			errorList = append(errorList, field.Forbidden(specPath.Child("labels"), "labels cannot be set when volumeID is set"))
 		}
+		if pv.Spec.ReplaceExisting {
+			errorList = append(errorList, field.Forbidden(specPath.Child("replaceExisting"), "replaceExisting cannot be set when volumeID is set"))
+		}
 		return errorList
 	}
 

api/v2/physical_container_volume_types_test.go

@@ -38,6 +38,16 @@ func TestPhysicalContainerVolumeValidate(t *testing.T) {
 				Spec:       PhysicalContainerVolumeSpec{VolumeID: "test-runtime-volume"},
 			},
 		},
+		{
+			name: "valid replacement volume",
+			volume: PhysicalContainerVolume{
+				ObjectMeta: metav1.ObjectMeta{Name: "test-volume", Namespace: "test-namespace"},
+				Spec: PhysicalContainerVolumeSpec{
+					VolumeName:      "test-runtime-volume",
+					ReplaceExisting: true,
+				},
+			},
+		},
 		{
 			name: "missing namespace",
 			volume: PhysicalContainerVolume{
@@ -61,6 +71,17 @@ func TestPhysicalContainerVolumeValidate(t *testing.T) {
 			},
 			expectedError: "spec.volumeName",
 		},
+		{
+			name: "persistent with tracked volume",
+			volume: PhysicalContainerVolume{
+				ObjectMeta: metav1.ObjectMeta{Name: "test-volume", Namespace: "test-namespace"},
+				Spec: PhysicalContainerVolumeSpec{
+					VolumeID:   "test-runtime-volume",
+					Persistent: true,
+				},
+			},
+			expectedError: "spec.persistent",
+		},
 		{
 			name: "volume name with tracked volume",
 			volume: PhysicalContainerVolume{
@@ -72,6 +93,17 @@ func TestPhysicalContainerVolumeValidate(t *testing.T) {
 			},
 			expectedError: "spec.volumeName",
 		},
+		{
+			name: "replace existing with tracked volume",
+			volume: PhysicalContainerVolume{
+				ObjectMeta: metav1.ObjectMeta{Name: "test-volume", Namespace: "test-namespace"},
+				Spec: PhysicalContainerVolumeSpec{
+					VolumeID:        "test-runtime-volume",
+					ReplaceExisting: true,
+				},
+			},
+			expectedError: "spec.replaceExisting",
+		},
 		{
 			name: "labels with tracked volume",
 			volume: PhysicalContainerVolume{

controllers/physical_container_volume_controller.go

@@ -276,6 +276,22 @@ func (r *PhysicalContainerVolumeReconciler) createPhysicalContainerVolume(
 	data *physicalContainerVolumeData,
 	log logr.Logger,
 ) {
+	if volume.Spec.ReplaceExisting {
+		replaced, replaceErr := r.replacePhysicalContainerVolume(ctx, volume, data, log)
+		if replaceErr != nil {
+			log.Error(replaceErr, "Failed to replace existing runtime volume", "VolumeName", volume.Spec.VolumeName)
+			data.conditionReason = apiv2.PhysicalContainerVolumeReasonReconciliationFailed
+			data.failureMessage = fmt.Sprintf("Failed to replace existing runtime volume: %v", replaceErr)
+			data.retryAfter = time.Now().Add(delayDurations[LongDelay].Duration)
+			r.queuePhysicalContainerVolumeDataResult(volume, stateKey, data)
+			return
+		}
+		if !replaced {
+			r.queuePhysicalContainerVolumeDataResult(volume, stateKey, data)
+			return
+		}
+	}
+
 	createErr := r.orchestrator.CreateVolume(ctx, containers.CreateVolumeOptions{
 		Name:   volume.Spec.VolumeName,
 		Labels: physicalContainerVolumeCreationLabels(volume, log),
@@ -284,6 +300,43 @@ func (r *PhysicalContainerVolumeReconciler) createPhysicalContainerVolume(
 	r.queuePhysicalContainerVolumeDataResult(volume, stateKey, data)
 }
 
+func (r *PhysicalContainerVolumeReconciler) replacePhysicalContainerVolume(
+	ctx context.Context,
+	volume *apiv2.PhysicalContainerVolume,
+	data *physicalContainerVolumeData,
+	log logr.Logger,
+) (bool, error) {
+	inspectedVolume, inspectErr := inspectPhysicalContainerVolume(ctx, r.orchestrator, volume.Spec.VolumeName)
+	if errors.Is(inspectErr, containers.ErrNotFound) {
+		return true, nil
+	}
+	if inspectErr != nil {
+		return false, fmt.Errorf("inspect runtime volume %q: %w", volume.Spec.VolumeName, inspectErr)
+	}
+	if inspectedVolume.Name == "" {
+		return false, fmt.Errorf("inspect runtime volume %q returned an empty name", volume.Spec.VolumeName)
+	}
+	if physicalContainerVolumeBelongsToResource(inspectedVolume, volume) {
+		data.conditionReason = apiv2.PhysicalContainerVolumeReasonCreated
+		data.volumeID = inspectedVolume.Name
+		data.failureMessage = ""
+		data.retryAfter = time.Time{}
+		log.V(1).Info("Adopted runtime volume created by an earlier attempt", "VolumeID", inspectedVolume.Name)
+		return false, nil
+	}
+
+	if !r.removeRuntimeVolume(ctx, inspectedVolume.Name, log) {
+		return false, fmt.Errorf("remove runtime volume %q", inspectedVolume.Name)
+	}
+
+	log.V(1).Info(
+		"Removed existing runtime volume before replacement",
+		"VolumeID", inspectedVolume.Name,
+		"VolumeName", inspectedVolume.Name,
+	)
+	return true, nil
+}
+
 func (r *PhysicalContainerVolumeReconciler) applyPhysicalContainerVolumeCreateResult(
 	ctx context.Context,
 	volume *apiv2.PhysicalContainerVolume,
@@ -293,6 +346,7 @@ func (r *PhysicalContainerVolumeReconciler) applyPhysicalContainerVolumeCreateRe
 ) {
 	if createErr != nil {
 		log.Error(createErr, "Failed to create runtime volume", "VolumeName", volume.Spec.VolumeName)
+		data.failureMessage = fmt.Sprintf("Failed to create runtime volume: %v", createErr)
 	}
 
 	inspectedVolume, inspectErr := inspectPhysicalContainerVolume(ctx, r.orchestrator, volume.Spec.VolumeName)
@@ -304,9 +358,17 @@ func (r *PhysicalContainerVolumeReconciler) applyPhysicalContainerVolumeCreateRe
 		return
 	}
 	if inspectErr == nil {
-		data.conditionReason = apiv2.PhysicalContainerVolumeReasonCreateFailed
-		data.failureMessage = fmt.Sprintf("Runtime volume name %q is already in use.", volume.Spec.VolumeName)
-		data.retryAfter = time.Time{}
+		if volume.Spec.ReplaceExisting {
+			data.conditionReason = apiv2.PhysicalContainerVolumeReasonReconciliationFailed
+			if data.failureMessage == "" {
+				data.failureMessage = fmt.Sprintf("Runtime volume name %q was claimed during replacement.", volume.Spec.VolumeName)
+			}
+			data.retryAfter = time.Now().Add(delayDurations[LongDelay].Duration)
+		} else {
+			data.conditionReason = apiv2.PhysicalContainerVolumeReasonCreateFailed
+			data.failureMessage = fmt.Sprintf("Runtime volume name %q is already in use.", volume.Spec.VolumeName)
+			data.retryAfter = time.Time{}
+		}
 		return
 	}
 
@@ -384,18 +446,29 @@ func handlePhysicalContainerVolumeRecoverableCreateFailed(
 
 	inspectedVolume, inspectErr := inspectPhysicalContainerVolume(ctx, reconciler.orchestrator, volume.Spec.VolumeName)
 	if inspectErr == nil {
-		if !physicalContainerVolumeBelongsToResource(inspectedVolume, volume) {
+		belongsToResource := physicalContainerVolumeBelongsToResource(inspectedVolume, volume)
+		if !belongsToResource && !volume.Spec.ReplaceExisting {
 			data.conditionReason = apiv2.PhysicalContainerVolumeReasonCreateFailed
 			data.failureMessage = fmt.Sprintf("Runtime volume name %q is already in use.", volume.Spec.VolumeName)
 			data.retryAfter = time.Time{}
-		} else {
-			data.conditionReason = apiv2.PhysicalContainerVolumeReasonCreated
-			data.volumeID = inspectedVolume.Name
-			data.failureMessage = ""
-			data.retryAfter = time.Time{}
+			stateKey, _ := reconciler.volumeData.BorrowByNamespacedName(volume.NamespacedName())
+			if reconciler.volumeData.Update(volume.NamespacedName(), stateKey, data) {
+				return data.applyTo(volume)
+			}
+			return additionalReconciliationNeeded
+		}
+		if !belongsToResource {
+			log.V(1).Info("Retrying runtime volume replacement", "VolumeID", inspectedVolume.Name, "VolumeName", inspectedVolume.Name)
+			return reconciler.schedulePhysicalContainerVolumeCreate(volume, log)
 		}
+
+		data.conditionReason = apiv2.PhysicalContainerVolumeReasonCreated
+		data.volumeID = inspectedVolume.Name
+		data.failureMessage = ""
+		data.retryAfter = time.Time{}
 		stateKey, _ := reconciler.volumeData.BorrowByNamespacedName(volume.NamespacedName())
 		if reconciler.volumeData.Update(volume.NamespacedName(), stateKey, data) {
+			log.V(1).Info("Adopted runtime volume created by an earlier attempt", "VolumeID", inspectedVolume.Name)
 			return data.applyTo(volume) | applyReadyPhysicalContainerVolumeStatus(volume, inspectedVolume)
 		}
 		return additionalReconciliationNeeded
@@ -444,7 +517,7 @@ func (r *PhysicalContainerVolumeReconciler) handleDeletionRequest(ctx context.Co
 	if volumeID == "" {
 		volumeID = volume.Spec.VolumeID
 	}
-	if !volume.Spec.Persistent &&
+	if volume.Spec.VolumeID == "" && !volume.Spec.Persistent &&
 		volumeID == "" && data != nil &&
 		data.conditionReason == apiv2.PhysicalContainerVolumeReasonReconciliationFailed {
 		inspectedVolume, inspectErr := inspectPhysicalContainerVolume(ctx, r.orchestrator, volume.Spec.VolumeName)
@@ -456,7 +529,7 @@ func (r *PhysicalContainerVolumeReconciler) handleDeletionRequest(ctx context.Co
 		}
 	}
 
-	if !volume.Spec.Persistent && volumeID != "" && !r.removeRuntimeVolume(ctx, volumeID, log) {
+	if volume.Spec.VolumeID == "" && !volume.Spec.Persistent && volumeID != "" && !r.removeRuntimeVolume(ctx, volumeID, log) {
 		return additionalReconciliationNeeded
 	}
 

docs/v2-resource-plan.md

@@ -72,7 +72,7 @@ This document tracks the intended direction for DCP V2 resources. The current V2
 - `PhysicalContainerImage` provides source image pull and build workflows.
 - `PhysicalContainer` creates or tracks one runtime container, reports runtime status and port mappings, and references a same-namespace `PhysicalContainerImage`.
 - `PhysicalContainerNetwork` creates or references one runtime container network and reports its observed identity, driver, and address allocations. Networks referenced by runtime ID are always retained. Created networks are retained when `persistent` is true; otherwise deletion enumerates running and stopped attachments, forcibly disconnects each container without removing it, and then removes the network. Name collisions are terminal unless `replaceExisting` is true, in which case the controller safely removes the specifically resolved network before creating its replacement. Runtime adapters classify their own built-in, non-removable networks, and replacement rejects them before disconnecting any attachments.
-- `PhysicalContainerVolume` creates or tracks one runtime container volume and reports its observed name, driver, scope, mount point, and creation time. Unless persistent, deletion removes the volume after it is no longer referenced by a container. It deliberately does not use force removal: Docker still rejects in-use volumes, while Podman force removal deletes the containers using the volume. Created volumes carry creator and persistence labels, and startup harvesting removes abandoned non-persistent volumes after abandoned containers.
+- `PhysicalContainerVolume` creates or references one runtime container volume and reports its observed name, driver, scope, mount point, and creation time. Volumes referenced by runtime ID are always retained. Created volumes are retained when `persistent` is true; otherwise deletion removes them after they are no longer referenced by a container. Name collisions are terminal unless `replaceExisting` is true, in which case the controller safely removes the specifically resolved volume before creating its replacement. Removal deliberately does not use force: Docker still rejects in-use volumes, while Podman force removal deletes the containers using the volume. Created volumes carry creator and persistence labels, and startup harvesting removes abandoned non-persistent volumes after abandoned containers.
 - The physical resources use in-memory progress data, standardized `Ready` conditions, and queued work where side effects can block.
 
 ## Follow-up roadmap

internal/testutil/ctrlutil/test_container_orchestrator.go

@@ -83,6 +83,9 @@ type TestContainerOrchestrator struct {
 	createVolumeCalls       map[string]int
 	createVolumeBlocks      map[string]chan struct{}
 	createVolumePostErrors  map[string][]error
+	removeVolumeCalls       map[string]int
+	removeVolumeErrors      map[string][]error
+	removeVolumePostErrors  map[string][]error
 	inspectVolumeCalls      map[string]int
 	pullImageCalls          map[string]int
 	pullImageBlocks         map[string]chan struct{}
@@ -212,6 +215,9 @@ func NewTestContainerOrchestrator(
 		createVolumeCalls:       map[string]int{},
 		createVolumeBlocks:      map[string]chan struct{}{},
 		createVolumePostErrors:  map[string][]error{},
+		removeVolumeCalls:       map[string]int{},
+		removeVolumeErrors:      map[string][]error{},
+		removeVolumePostErrors:  map[string][]error{},
 		inspectVolumeCalls:      map[string]int{},
 		pullImageCalls:          map[string]int{},
 		pullImageBlocks:         map[string]chan struct{}{},
@@ -790,6 +796,11 @@ func (to *TestContainerOrchestrator) CreateVolume(ctx context.Context, options c
 }
 
 func (to *TestContainerOrchestrator) RemoveVolumes(ctx context.Context, options containers.RemoveVolumesOptions) ([]string, error) {
+	removeErr, postRemoveErr := to.recordRemoveVolumesOperation(options.Volumes)
+	if removeErr != nil {
+		return nil, removeErr
+	}
+
 	to.mutex.Lock()
 	defer to.mutex.Unlock()
 
@@ -838,7 +849,7 @@ func (to *TestContainerOrchestrator) RemoveVolumes(ctx context.Context, options
 		err = errors.Join(err, errors.Join(containers.ErrIncomplete, fmt.Errorf("not all volumes were removed, expected %d but got %d", len(options.Volumes), len(removed))))
 	}
 
-	return removed, err
+	return removed, errors.Join(err, postRemoveErr)
 }
 
 func (to *TestContainerOrchestrator) InspectVolumes(ctx context.Context, options containers.InspectVolumesOptions) ([]containers.InspectedVolume, error) {
@@ -1343,6 +1354,36 @@ func (to *TestContainerOrchestrator) InspectVolumeCallCount(volume string) int {
 	return to.inspectVolumeCalls[volume]
 }
 
+func (to *TestContainerOrchestrator) FailNextRemoveVolume(name string, removeErr error) {
+	if removeErr == nil {
+		removeErr = errors.New("simulated volume removal failure")
+	}
+
+	to.operationMutex.Lock()
+	defer to.operationMutex.Unlock()
+
+	to.removeVolumeErrors[name] = append(to.removeVolumeErrors[name], removeErr)
+}
+
+// FailNextRemoveVolumeAfterRemoval simulates a runtime removal whose result is uncertain to the caller.
+func (to *TestContainerOrchestrator) FailNextRemoveVolumeAfterRemoval(name string, removeErr error) {
+	if removeErr == nil {
+		removeErr = errors.New("simulated lost volume removal response")
+	}
+
+	to.operationMutex.Lock()
+	defer to.operationMutex.Unlock()
+
+	to.removeVolumePostErrors[name] = append(to.removeVolumePostErrors[name], removeErr)
+}
+
+func (to *TestContainerOrchestrator) RemoveVolumeCallCount(name string) int {
+	to.operationMutex.Lock()
+	defer to.operationMutex.Unlock()
+
+	return to.removeVolumeCalls[name]
+}
+
 // FailNextCreateNetworkAfterCreation simulates a runtime create whose result is uncertain to the caller.
 func (to *TestContainerOrchestrator) FailNextCreateNetworkAfterCreation(name string, createErr error) {
 	if createErr == nil {
@@ -1492,6 +1533,33 @@ func (to *TestContainerOrchestrator) recordInspectVolumesOperation(volumes []str
 	}
 }
 
+func (to *TestContainerOrchestrator) recordRemoveVolumesOperation(volumes []string) (error, error) {
+	to.operationMutex.Lock()
+	defer to.operationMutex.Unlock()
+
+	var removeErr error
+	var postRemoveErr error
+	for _, volume := range volumes {
+		to.removeVolumeCalls[volume]++
+		if len(to.removeVolumeErrors[volume]) > 0 {
+			removeErr = errors.Join(removeErr, to.removeVolumeErrors[volume][0])
+			to.removeVolumeErrors[volume] = to.removeVolumeErrors[volume][1:]
+			if len(to.removeVolumeErrors[volume]) == 0 {
+				delete(to.removeVolumeErrors, volume)
+			}
+		}
+		if len(to.removeVolumePostErrors[volume]) > 0 {
+			postRemoveErr = errors.Join(postRemoveErr, to.removeVolumePostErrors[volume][0])
+			to.removeVolumePostErrors[volume] = to.removeVolumePostErrors[volume][1:]
+			if len(to.removeVolumePostErrors[volume]) == 0 {
+				delete(to.removeVolumePostErrors, volume)
+			}
+		}
+	}
+
+	return removeErr, postRemoveErr
+}
+
 func (to *TestContainerOrchestrator) takeCreateNetworkPostError(name string) error {
 	to.operationMutex.Lock()
 	defer to.operationMutex.Unlock()