From 4edd039ed082316d2f347f960f9159021259ac01 Mon Sep 17 00:00:00 2001 From: Guillaume Lours Date: Thu, 28 May 2026 15:04:00 +0200 Subject: [PATCH] executor: pay down technical debt from PE review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) Signed-off-by: Guillaume Lours --- pkg/compose/executor.go | 27 +++++++++--- pkg/compose/executor_test.go | 84 +++++++++++++++++++++++++++++++----- pkg/compose/reconcile.go | 29 +++++++------ 3 files changed, 110 insertions(+), 30 deletions(-) diff --git a/pkg/compose/executor.go b/pkg/compose/executor.go index 89808f20c..d02b1038f 100644 --- a/pkg/compose/executor.go +++ b/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]) diff --git a/pkg/compose/executor_test.go b/pkg/compose/executor_test.go index 2d2a66529..5f4ab8804 100644 --- a/pkg/compose/executor_test.go +++ b/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{} diff --git a/pkg/compose/reconcile.go b/pkg/compose/reconcile.go index 8e19822be..9a024d983 100644 --- a/pkg/compose/reconcile.go +++ b/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) {