executor: pay down technical debt from PE review

Three improvements identified in the Principal Engineer pass but
deliberately deferred:

1. Test fidelity. Split executePlan into newPlanExecutor (constructs
   the executor seeded from observed state) and (*planExecutor).run
   (walks the DAG). Production callers go through executePlan
   unchanged. TestExecutePlanRemoveContainerDropsFromCache now uses
   newPlanExecutor + run, exercising the same errgroup, done-channel
   and group-tracker wiring as production instead of a hand-rolled
   loop over executeNode.

2. //nolint:unused chain. The three preserved helpers
   (reconciler.prompt, planRecreateVolume, servicesUsingVolume) each
   carried a separate "kept for future" comment. Consolidate the
   rationale on the reconciler.prompt field doc and point the helper
   nolint directives there, so a future cleanup is a single grep.

3. Concurrency test. Add TestExecutePlanConcurrentRemovesCacheCoherence
   which builds N independent Stop→Remove chains in one plan; the
   errgroup fans them out across goroutines that all hit
   containersByService under the mutex. Passes under -race. Failure
   would expose a missing or incorrect lock.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Signed-off-by: Guillaume Lours <glours@users.noreply.github.com>

pkg/compose/executor.go

@@ -75,16 +75,28 @@ func (pc *reconciliationContext) get(nodeID int) operationResult {
 // while respecting dependency edges. It emits progress events and handles
 // group-based event aggregation for composite operations like recreate.
 func (s *composeService) executePlan(ctx context.Context, project *types.Project, observed *ObservedState, plan *Plan) error {
-	if plan.IsEmpty() {
-		return nil
-	}
+	return s.newPlanExecutor(project, observed).run(ctx, plan)
+}
 
-	exec := &planExecutor{
+// newPlanExecutor constructs a planExecutor seeded from the observed state.
+// Split out from executePlan so tests can inspect the executor's live state
+// (e.g. the containersByService cache) after running a plan.
+func (s *composeService) newPlanExecutor(project *types.Project, observed *ObservedState) *planExecutor {
+	return &planExecutor{
 		compose:             s,
 		project:             project,
 		pctx:                &reconciliationContext{results: map[int]operationResult{}},
 		containersByService: observed.containersByService(),
 	}
+}
+
+// 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.
+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))
@@ -94,6 +106,7 @@ func (s *composeService) executePlan(ctx context.Context, project *types.Project
 
 	// 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 {
@@ -108,15 +121,15 @@ func (s *composeService) executePlan(ctx context.Context, project *types.Project
 			}
 
 			// Emit group start event if this is the first node of a group
-			groups.onNodeStart(node, s.events)
+			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, s.events)
+				groups.onNodeDone(node, events)
 			} else if ctx.Err() == nil {
-				groups.onNodeError(node, s.events, err)
+				groups.onNodeError(node, events, err)
 			}
 
 			close(done[node.ID])

pkg/compose/executor_test.go

@@ -18,6 +18,7 @@ package compose
 
 import (
 	"context"
+	"fmt"
 	"testing"
 
 	"github.com/compose-spec/compose-go/v2/types"
@@ -135,6 +136,10 @@ func emptyObservedState(project string) *ObservedState {
 // is removed, subsequent service-reference resolution does not see its stale ID.
 // Without this guarantee, a recreate followed by a dependent's create can pick
 // up the just-removed container, depending on the canonical-name sort order.
+//
+// Goes through newPlanExecutor + run (i.e. the same code path executePlan
+// uses in production) so the test exercises the errgroup, done-channel
+// wiring and group tracker — not a hand-rolled loop over executeNode.
 func TestExecutePlanRemoveContainerDropsFromCache(t *testing.T) {
 	svc, apiClient := newTestService(t)
 
@@ -175,22 +180,79 @@ func TestExecutePlanRemoveContainerDropsFromCache(t *testing.T) {
 		Container:  &oldCtr,
 	}, "", stopNode)
 
-	// Seed and run the plan via the same code path as production.
-	// After completion, the cache must no longer contain old-id.
-	exec := &planExecutor{
-		compose:             svc,
-		project:             &types.Project{Name: "test"},
-		pctx:                &reconciliationContext{results: map[int]operationResult{}},
-		containersByService: observed.containersByService(),
-	}
-	for _, node := range plan.Nodes {
-		assert.NilError(t, exec.executeNode(t.Context(), node))
-	}
+	exec := svc.newPlanExecutor(&types.Project{Name: "test"}, observed)
+	assert.NilError(t, exec.run(t.Context(), plan))
 
 	assert.Equal(t, len(exec.containersByService["web"]), 0,
 		"removed container should be dropped from the live view")
 }
 
+// TestExecutePlanConcurrentRemovesCacheCoherence stresses the cache mutex by
+// scheduling N independent Stop+Remove pairs that the DAG lets the errgroup
+// run concurrently. After the plan completes the cache must be empty, with no
+// duplicates and no surviving entries — failure under -race would indicate a
+// missing or incorrect lock around containersByService.
+func TestExecutePlanConcurrentRemovesCacheCoherence(t *testing.T) {
+	svc, apiClient := newTestService(t)
+
+	const replicas = 5
+	ctrs := make([]container.Summary, replicas)
+	for i := range ctrs {
+		ctrs[i] = container.Summary{
+			ID:    fmt.Sprintf("c%d", i),
+			Names: []string{fmt.Sprintf("/test-web-%d", i+1)},
+			Labels: map[string]string{
+				api.ServiceLabel:         "web",
+				api.ContainerNumberLabel: fmt.Sprintf("%d", i+1),
+			},
+		}
+	}
+
+	// Each container gets exactly one Stop and one Remove. gomock matches by
+	// any order across calls, so concurrent execution is fine.
+	for i := range ctrs {
+		apiClient.EXPECT().ContainerStop(gomock.Any(), ctrs[i].ID, gomock.Any()).
+			Return(client.ContainerStopResult{}, nil)
+		apiClient.EXPECT().ContainerRemove(gomock.Any(), ctrs[i].ID, gomock.Any()).
+			Return(client.ContainerRemoveResult{}, nil)
+	}
+
+	webContainers := make([]ObservedContainer, replicas)
+	for i := range ctrs {
+		webContainers[i] = ObservedContainer{ID: ctrs[i].ID, Summary: ctrs[i]}
+	}
+	observed := &ObservedState{
+		ProjectName: "test",
+		Containers:  map[string][]ObservedContainer{"web": webContainers},
+		Networks:    map[string]ObservedNetwork{},
+		Volumes:     map[string]ObservedVolume{},
+	}
+
+	// Build N independent Stop→Remove chains. The errgroup will fan them out
+	// across goroutines that all hammer containersByService under the mutex.
+	plan := &Plan{}
+	for i := range ctrs {
+		stop := plan.addNode(Operation{
+			Type:       OpStopContainer,
+			ResourceID: fmt.Sprintf("service:web:%d", i+1),
+			Cause:      "scale down",
+			Container:  &ctrs[i],
+		}, "")
+		plan.addNode(Operation{
+			Type:       OpRemoveContainer,
+			ResourceID: fmt.Sprintf("service:web:%d", i+1),
+			Cause:      "scale down",
+			Container:  &ctrs[i],
+		}, "", stop)
+	}
+
+	exec := svc.newPlanExecutor(&types.Project{Name: "test"}, observed)
+	assert.NilError(t, exec.run(t.Context(), plan))
+
+	assert.Equal(t, len(exec.containersByService["web"]), 0,
+		"all removed containers should be dropped from the live view")
+}
+
 // notFoundError implements the errdefs.ErrNotFound interface for test mocks.
 type notFoundError struct{}
 

pkg/compose/reconcile.go

@@ -60,10 +60,21 @@ type reconciler struct {
 	project  *types.Project
 	observed *ObservedState
 	options  ReconcileOptions
-	// prompt is wired through for future use: when divergence detection for
-	// volumes/networks migrates fully into the reconciler (today it lives in
-	// ensureProjectVolumes/ensureNetworks), prompts will fire from here. Kept
-	// available now so call sites do not have to change later.
+	// Seam-consolidation infrastructure.
+	//
+	// Today, divergence detection and recreation for volumes/networks live in
+	// ensureProjectVolumes/ensureNetworks (called before reconcile). The plan
+	// is to migrate that responsibility into the reconciler. The hooks below
+	// are kept so the migration can land in one commit instead of touching
+	// every caller:
+	//
+	//   - prompt (this field)               — user interaction
+	//   - planRecreateVolume (below)        — the volume recreate sequence
+	//   - servicesUsingVolume (below)       — its only caller today
+	//
+	// When the migration lands, remove all three together if it ends up
+	// shaped differently. The //nolint:unused markers on the helpers point
+	// here for context.
 	prompt Prompt
 	plan   *Plan
 
@@ -236,11 +247,7 @@ func (r *reconciler) planCreateVolume(key string, vol *types.VolumeConfig) *Plan
 // Containers must be removed (not just stopped) because Docker does not allow
 // removing a volume that is referenced by any container, even a stopped one.
 //
-// Currently unused: divergence detection and recreation live in
-// ensureProjectVolumes (see create.go:1626). Kept in place so the reconciler
-// can take over that responsibility when the seam is consolidated.
-//
-//nolint:unused
+//nolint:unused // see reconciler.prompt field doc — seam consolidation.
 func (r *reconciler) planRecreateVolume(key string, vol *types.VolumeConfig) {
 	observed := r.observed.Volumes[key]
 	affectedServices := r.servicesUsingVolume(key)
@@ -306,9 +313,7 @@ func (r *reconciler) servicesUsingNetwork(networkKey string) []string {
 // servicesUsingVolume returns the names of services that mount the given
 // compose volume key, sorted for deterministic plan output.
 //
-// Currently used only by planRecreateVolume (also unused — see its doc).
-//
-//nolint:unused
+//nolint:unused // see reconciler.prompt field doc — seam consolidation.
 func (r *reconciler) servicesUsingVolume(volumeKey string) []string {
 	var names []string
 	for _, key := range sortedKeys(r.project.Services) {