Method run

pkg/compose/executor.go:90–135

run walks the plan DAG, executing nodes in parallel where possible while respecting dependency edges. Emits progress events and handles group-based event aggregation for composite operations like recreate.

(ctx context.Context, plan *Plan)

// respecting dependency edges. Emits progress events and handles group-based
// event aggregation for composite operations like recreate.
func (exec *planExecutor) run(ctx context.Context, plan *Plan) error {
	if plan.IsEmpty() {
		return nil
	}

	// Build a done-channel per node so dependents can wait
	done := make(map[int]chan struct{}, len(plan.Nodes))
	for _, node := range plan.Nodes {
		done[node.ID] = make(chan struct{})
	}

	// Track group event state: first node emits Working, last emits Done
	groups := exec.buildGroupTracker(plan)
	events := exec.compose.events

	eg, ctx := errgroup.WithContext(ctx)
	for _, node := range plan.Nodes {
		eg.Go(func() error {
			// Wait for all dependencies
			for _, dep := range node.DependsOn {
				select {
				case <-done[dep.ID]:
				case <-ctx.Done():
					return ctx.Err()
				}
			}

			// Emit group start event if this is the first node of a group
			groups.onNodeStart(node, events)

			err := exec.executeNode(ctx, node)

			if err == nil {
				// Emit group done event if this is the last node of a group
				groups.onNodeDone(node, events)
			} else if ctx.Err() == nil {
				groups.onNodeError(node, events, err)
			}

			close(done[node.ID])
			return err
		})
	}

	return eg.Wait()
}

// executeNode dispatches a single plan node to the appropriate API call.
func (exec *planExecutor) executeNode(ctx context.Context, node *PlanNode) error {