fix some design issues

Signed-off-by: Nicolas De Loof <nicolas.deloof@gmail.com>

Author: ndeloof

pkg/compose/plan_executor.go

@@ -27,6 +27,7 @@ import (
 	"sync"
 
 	"github.com/compose-spec/compose-go/v2/types"
+	"github.com/containerd/errdefs"
 	containerType "github.com/moby/moby/api/types/container"
 	"github.com/moby/moby/client"
 	"go.opentelemetry.io/otel"
@@ -43,15 +44,11 @@ import (
 type executionState struct {
 	mu         sync.Mutex
 	containers map[string]Containers // service name -> containers created/updated
-	networks   map[string]string     // network key -> ID
-	volumes    map[string]string     // volume key -> ID
 }
 
 func newExecutionState() *executionState {
 	return &executionState{
 		containers: make(map[string]Containers),
-		networks:   make(map[string]string),
-		volumes:    make(map[string]string),
 	}
 }
 
@@ -79,18 +76,6 @@ func (es *executionState) getContainers(serviceName string) Containers {
 	return slices.Clone(es.containers[serviceName])
 }
 
-func (es *executionState) setNetworkID(key, id string) {
-	es.mu.Lock()
-	defer es.mu.Unlock()
-	es.networks[key] = id
-}
-
-func (es *executionState) setVolumeID(key, id string) {
-	es.mu.Lock()
-	defer es.mu.Unlock()
-	es.volumes[key] = id
-}
-
 // resolveServiceReferences replaces service references in a ServiceConfig with
 // actual container IDs from the execution state. This mirrors the logic in
 // convergence.resolveServiceReferences but uses executionState instead.
@@ -157,6 +142,13 @@ func (s *composeService) ExecutePlan(ctx context.Context, project *types.Project
 		return nil
 	}
 
+	// Validate the plan has no dependency cycles before executing.
+	// Without this check, a cycle would cause the executor to hang
+	// indefinitely waiting for operations that can never be scheduled.
+	if _, err := topologicalSort(plan); err != nil {
+		return err
+	}
+
 	// Pre-populate execution state with existing containers so that
 	// resolveServiceReferences can find containers for services not
 	// included in the plan (e.g. --no-deps scenarios).
@@ -172,7 +164,15 @@ func (s *composeService) ExecutePlan(ctx context.Context, project *types.Project
 	expect := len(plan.Operations)
 	eg, ctx := errgroup.WithContext(ctx)
 	opCh := make(chan *Operation, expect)
-	defer close(opCh)
+
+	// sendDone sends a completed operation to the consumer goroutine,
+	// respecting context cancellation to avoid blocking or panicking.
+	sendDone := func(op *Operation) {
+		select {
+		case opCh <- op:
+		case <-ctx.Done():
+		}
+	}
 
 	// Consumer goroutine: waits for completed ops and enqueues newly-ready dependents
 	eg.Go(func() error {
@@ -195,7 +195,7 @@ func (s *composeService) ExecutePlan(ctx context.Context, project *types.Project
 							if err := s.executeOperation(ctx, project, depOp, state); err != nil {
 								return err
 							}
-							opCh <- depOp
+							sendDone(depOp)
 							return nil
 						})
 					}
@@ -210,7 +210,7 @@ func (s *composeService) ExecutePlan(ctx context.Context, project *types.Project
 			if err := s.executeOperation(ctx, project, op, state); err != nil {
 				return err
 			}
-			opCh <- op
+			sendDone(op)
 			return nil
 		})
 	}
@@ -245,15 +245,15 @@ func (s *composeService) executeOperation(ctx context.Context, project *types.Pr
 func (s *composeService) dispatchOperation(ctx context.Context, project *types.Project, op *Operation, state *executionState) error {
 	switch op.Type {
 	case OpCreateNetwork:
-		return s.executePlanCreateNetwork(ctx, project, op, state)
+		return s.executePlanCreateNetwork(ctx, project, op)
 	case OpRemoveNetwork:
 		return s.executePlanRemoveNetwork(ctx, project, op)
 	case OpDisconnectNetwork:
 		return s.executePlanDisconnectNetwork(ctx, op)
 	case OpConnectNetwork:
 		return s.executePlanConnectNetwork(ctx, op)
 	case OpCreateVolume:
-		return s.executePlanCreateVolume(ctx, project, op, state)
+		return s.executePlanCreateVolume(ctx, project, op)
 	case OpRemoveVolume:
 		return s.executePlanRemoveVolume(ctx, op)
 	case OpCreateContainer:
@@ -273,13 +273,9 @@ func (s *composeService) dispatchOperation(ctx context.Context, project *types.P
 	}
 }
 
-func (s *composeService) executePlanCreateNetwork(ctx context.Context, project *types.Project, op *Operation, state *executionState) error {
-	id, err := s.ensureNetwork(ctx, project, op.NetworkOp.NetworkKey, op.NetworkOp.Desired)
-	if err != nil {
-		return err
-	}
-	state.setNetworkID(op.NetworkOp.NetworkKey, id)
-	return nil
+func (s *composeService) executePlanCreateNetwork(ctx context.Context, project *types.Project, op *Operation) error {
+	_, err := s.ensureNetwork(ctx, project, op.NetworkOp.NetworkKey, op.NetworkOp.Desired)
+	return err
 }
 
 func (s *composeService) executePlanRemoveNetwork(ctx context.Context, project *types.Project, op *Operation) error {
@@ -301,17 +297,13 @@ func (s *composeService) executePlanConnectNetwork(ctx context.Context, op *Oper
 	return err
 }
 
-func (s *composeService) executePlanCreateVolume(ctx context.Context, project *types.Project, op *Operation, state *executionState) error {
+func (s *composeService) executePlanCreateVolume(ctx context.Context, project *types.Project, op *Operation) error {
 	volume := *op.VolumeOp.Desired
 	volume.CustomLabels = volume.CustomLabels.Add(api.VolumeLabel, op.VolumeOp.VolumeKey)
 	volume.CustomLabels = volume.CustomLabels.Add(api.ProjectLabel, project.Name)
 	volume.CustomLabels = volume.CustomLabels.Add(api.VersionLabel, api.ComposeVersion)
-	id, err := s.ensureVolume(ctx, op.VolumeOp.VolumeKey, volume, project)
-	if err != nil {
-		return err
-	}
-	state.setVolumeID(op.VolumeOp.VolumeKey, id)
-	return nil
+	_, err := s.ensureVolume(ctx, op.VolumeOp.VolumeKey, volume, project)
+	return err
 }
 
 func (s *composeService) executePlanRemoveVolume(ctx context.Context, op *Operation) error {
@@ -410,8 +402,18 @@ func (s *composeService) executePlanStopContainer(ctx context.Context, op *Opera
 }
 
 func (s *composeService) executePlanRemoveContainer(ctx context.Context, op *Operation) error {
-	service := op.ContainerOp.Service
-	return s.stopAndRemoveContainer(ctx, *op.ContainerOp.Existing, &service, op.ContainerOp.Timeout, false)
+	ctr := *op.ContainerOp.Existing
+	eventName := getContainerProgressName(ctr)
+	s.events.On(removingEvent(eventName))
+	_, err := s.apiClient().ContainerRemove(ctx, ctr.ID, client.ContainerRemoveOptions{
+		Force: true,
+	})
+	if err != nil && !errdefs.IsNotFound(err) && !errdefs.IsConflict(err) {
+		s.events.On(errorEvent(eventName, "Error while Removing"))
+		return err
+	}
+	s.events.On(removedEvent(eventName))
+	return nil
 }
 
 func (s *composeService) executePlanRunPlugin(ctx context.Context, project *types.Project, op *Operation) error {

pkg/compose/reconcile.go

@@ -125,11 +125,11 @@ type ContainerOperation struct {
 	Existing        *container.Summary
 	Inherit         bool
 	Timeout         *time.Duration
+	NetworkRecreate bool // true when this op was created for a network recreation
 }
 
 // RenameOperation holds details for renaming a container.
 type RenameOperation struct {
-	ContainerID string
 	CurrentName string
 	NewName     string
 }
@@ -303,6 +303,14 @@ func Reconcile(project *types.Project, observed *ObservedState, opts ReconcileOp
 		return nil, err
 	}
 
+	// Step 3b - Clean up stale network disconnect/connect operations.
+	// When a container is being recreated (has a rename op in the plan),
+	// any disconnect/connect ops created by reconcileNetworks use the old
+	// container ID which will be stale after recreation. The new container
+	// will be connected to the recreated network automatically via
+	// buildDependencyEdges (create-container depends on create-network).
+	pruneStaleNetworkOpsForRecreatedContainers(plan)
+
 	// Step 4 - Orphans
 	if opts.RemoveOrphans {
 		reconcileOrphans(observed.Orphans, plan, opts)
@@ -446,9 +454,10 @@ func reconcileNetworks(project *types.Project, observed *ObservedState, plan *Re
 					ServiceName: ctr.Labels[api.ServiceLabel],
 					Resource:    ctrName,
 					ContainerOp: &ContainerOperation{
-						ContainerName: ctrName,
-						Existing:      &ctr,
-						Timeout:       opts.Timeout,
+						ContainerName:   ctrName,
+						Existing:        &ctr,
+						Timeout:         opts.Timeout,
+						NetworkRecreate: true,
 					},
 					Reason: fmt.Sprintf("network %q is being recreated", n.Name),
 				}
@@ -1228,25 +1237,132 @@ func addCascadingRestarts(project *types.Project, observed *ObservedState, plan
 				}
 
 				startID := fmt.Sprintf("start-container:%s", ctrName)
-				plan.Operations[startID] = &Operation{
-					ID:          startID,
-					Type:        OpStartContainer,
-					ServiceName: service.Name,
-					Resource:    ctrName,
-					ContainerOp: &ContainerOperation{
-						Service:       service,
-						ContainerName: ctrName,
-						Existing:      &ctr,
-					},
-					DependsOn: []string{stopID},
-					Reason:    fmt.Sprintf("restart after dependency %q recreated", depName),
+				if existingStart, exists := plan.Operations[startID]; exists {
+					// A start operation already exists (e.g. from network recreation).
+					// Add our stop as a dependency instead of overwriting.
+					if !slices.Contains(existingStart.DependsOn, stopID) {
+						existingStart.DependsOn = append(existingStart.DependsOn, stopID)
+					}
+				} else {
+					plan.Operations[startID] = &Operation{
+						ID:          startID,
+						Type:        OpStartContainer,
+						ServiceName: service.Name,
+						Resource:    ctrName,
+						ContainerOp: &ContainerOperation{
+							Service:       service,
+							ContainerName: ctrName,
+							Existing:      &ctr,
+						},
+						DependsOn: []string{stopID},
+						Reason:    fmt.Sprintf("restart after dependency %q recreated", depName),
+					}
 				}
 			}
 			break // Only need to process once per service
 		}
 	}
 }
 
+// pruneStaleNetworkOpsForRecreatedContainers removes disconnect/connect/start
+// operations that were created by reconcileNetworks for containers that are
+// also being recreated. When a container is recreated, the old container ID
+// used in connect/disconnect ops becomes stale. The new container will
+// automatically connect to the new network because buildDependencyEdges
+// ensures create-container depends on create-network.
+func pruneStaleNetworkOpsForRecreatedContainers(plan *ReconciliationPlan) {
+	recreatedContainers := collectRecreatedContainerNames(plan)
+	if len(recreatedContainers) == 0 {
+		return
+	}
+
+	toDelete := findStaleNetworkOps(plan, recreatedContainers)
+	deletePlanOps(plan, toDelete)
+}
+
+// collectRecreatedContainerNames returns the final names of containers that
+// have a recreate chain (identified by rename operations).
+func collectRecreatedContainerNames(plan *ReconciliationPlan) map[string]bool {
+	result := map[string]bool{}
+	for _, op := range plan.Operations {
+		if op.Type == OpRenameContainer && op.RenameOp != nil {
+			result[op.RenameOp.NewName] = true
+		}
+	}
+	return result
+}
+
+// findStaleNetworkOps identifies disconnect/connect/start/stop operations that
+// reference containers being recreated. These ops use stale container IDs.
+func findStaleNetworkOps(plan *ReconciliationPlan, recreated map[string]bool) []string {
+	var toDelete []string
+	for id, op := range plan.Operations {
+		switch op.Type {
+		case OpDisconnectNetwork, OpConnectNetwork:
+			if isNetworkOpForRecreatedContainer(op, recreated) {
+				toDelete = append(toDelete, id)
+			}
+		case OpStartContainer:
+			if isNetworkStartForRecreatedContainer(plan, op, recreated) {
+				toDelete = append(toDelete, id)
+			}
+		case OpStopContainer:
+			if isNetworkStopForRecreatedContainer(op, recreated) {
+				toDelete = append(toDelete, id)
+			}
+		}
+	}
+	return toDelete
+}
+
+func isNetworkOpForRecreatedContainer(op *Operation, recreated map[string]bool) bool {
+	if op.ContainerNetworkOp == nil {
+		return false
+	}
+	parts := strings.Fields(op.Resource)
+	return len(parts) > 0 && recreated[parts[0]]
+}
+
+func isNetworkStartForRecreatedContainer(plan *ReconciliationPlan, op *Operation, recreated map[string]bool) bool {
+	if op.ContainerOp == nil || !recreated[op.ContainerOp.ContainerName] {
+		return false
+	}
+	// Only remove start ops created by network recreation (depend on a connect op),
+	// not the start from the recreate chain itself (depends on a rename op).
+	for _, depID := range op.DependsOn {
+		if dep, ok := plan.Operations[depID]; ok && dep.Type == OpConnectNetwork {
+			return true
+		}
+	}
+	return false
+}
+
+func isNetworkStopForRecreatedContainer(op *Operation, recreated map[string]bool) bool {
+	if op.ContainerOp == nil || !recreated[op.ContainerOp.ContainerName] {
+		return false
+	}
+	// Only remove stop ops from network recreation, not from the recreate chain.
+	return op.ContainerOp.NetworkRecreate
+}
+
+// deletePlanOps removes operations by ID and cleans up DependsOn references.
+func deletePlanOps(plan *ReconciliationPlan, ids []string) {
+	deleted := map[string]bool{}
+	for _, id := range ids {
+		deleted[id] = true
+		delete(plan.Operations, id)
+	}
+	for _, op := range plan.Operations {
+		var cleaned []string
+		for _, depID := range op.DependsOn {
+			if !deleted[depID] {
+				cleaned = append(cleaned, depID)
+			}
+		}
+		op.DependsOn = cleaned
+	}
+}
+
 // serviceUsesRecreatedVolume checks if a service mounts any volume that is
 // being recreated, and returns a reason string if so.
 func serviceUsesRecreatedVolume(service types.ServiceConfig, recreatedVolumes map[string]bool) string {