fix: honor --parallel across all bulk engine-call fan-outs

pkg/compose/compose.go

@@ -34,6 +34,8 @@ import (
 	"github.com/moby/moby/api/types/swarm"
 	"github.com/moby/moby/client"
 	"github.com/sirupsen/logrus"
+	"golang.org/x/sync/errgroup"
+	"golang.org/x/sync/semaphore"
 
 	"github.com/docker/compose/v5/pkg/api"
 	"github.com/docker/compose/v5/pkg/dryrun"
@@ -149,14 +151,55 @@ func WithPrompt(prompt Prompt) Option {
 	}
 }
 
-// WithMaxConcurrency defines upper limit for concurrent operations against engine API
+// WithMaxConcurrency defines upper limit for concurrent operations against
+// engine API. A value <= 0 means unlimited.
 func WithMaxConcurrency(maxConcurrency int) Option {
 	return func(s *composeService) error {
 		s.maxConcurrency = maxConcurrency
 		return nil
 	}
 }
 
+// newLimitedErrgroup returns an errgroup.Group bounded to maxConcurrency
+// concurrent goroutines. maxConcurrency<=0 (including the Go zero-value)
+// leaves it unlimited, since errgroup.SetLimit(0) means "allow zero
+// goroutines", not "unlimited".
+func newLimitedErrgroup(ctx context.Context, maxConcurrency int) (*errgroup.Group, context.Context) {
+	eg, ctx := errgroup.WithContext(ctx)
+	if maxConcurrency > 0 {
+		eg.SetLimit(maxConcurrency)
+	}
+	return eg, ctx
+}
+
+// newOptionalLimiter returns a semaphore bounding concurrency to
+// maxConcurrency, or nil when maxConcurrency<=0 (unlimited). Use it, with
+// acquireSlot/releaseSlot, to gate only part of a goroutine's work — e.g. an
+// indefinite stream's opening call, not the stream itself — where
+// newLimitedErrgroup's whole-goroutine bound doesn't apply.
+func newOptionalLimiter(maxConcurrency int) *semaphore.Weighted {
+	if maxConcurrency <= 0 {
+		return nil
+	}
+	return semaphore.NewWeighted(int64(maxConcurrency))
+}
+
+// acquireSlot acquires a slot from limiter, or is a no-op when limiter is nil.
+func acquireSlot(ctx context.Context, limiter *semaphore.Weighted) error {
+	if limiter == nil {
+		return nil
+	}
+	return limiter.Acquire(ctx, 1)
+}
+
+// releaseSlot releases a slot acquired via acquireSlot, or is a no-op when
+// limiter is nil.
+func releaseSlot(limiter *semaphore.Weighted) {
+	if limiter != nil {
+		limiter.Release(1)
+	}
+}
+
 // WithDryRun configure Compose to run without actually applying changes
 func WithDryRun(s *composeService) error {
 	s.dryRun = true

pkg/compose/compose_test.go

@@ -0,0 +1,54 @@
+/*
+   Copyright 2026 Docker Compose CLI authors
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package compose
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"gotest.tools/v3/assert"
+)
+
+// TestNewLimitedErrgroup_NonPositiveIsUnlimited guards against the bug this
+// helper exists to fix: errgroup.SetLimit(0) means "allow zero goroutines",
+// not "unlimited". A composeService{} built without going through
+// NewComposeService has maxConcurrency's Go zero-value (0), so an
+// unconditional SetLimit(maxConcurrency) at any call site would silently
+// hang forever instead of running.
+func TestNewLimitedErrgroup_NonPositiveIsUnlimited(t *testing.T) {
+	for _, maxConcurrency := range []int{0, -1} {
+		t.Run(fmt.Sprintf("maxConcurrency=%d", maxConcurrency), func(t *testing.T) {
+			eg, _ := newLimitedErrgroup(t.Context(), maxConcurrency)
+
+			done := make(chan struct{})
+			go func() {
+				for range 5 {
+					eg.Go(func() error { return nil })
+				}
+				close(done)
+			}()
+
+			select {
+			case <-done:
+			case <-time.After(2 * time.Second):
+				t.Fatal("eg.Go blocked: maxConcurrency <= 0 must mean unlimited, not SetLimit(0) (zero goroutines allowed)")
+			}
+			assert.NilError(t, eg.Wait())
+		})
+	}
+}

pkg/compose/containers.go

@@ -26,7 +26,6 @@ import (
 	"github.com/compose-spec/compose-go/v2/types"
 	"github.com/moby/moby/api/types/container"
 	"github.com/moby/moby/client"
-	"golang.org/x/sync/errgroup"
 
 	"github.com/docker/compose/v5/pkg/api"
 )
@@ -186,8 +185,8 @@ func (containers Containers) filter(predicates ...containerPredicate) Containers
 }
 
 // forEachContainerConcurrent runs fn for every container concurrently and waits for all goroutines.
-func forEachContainerConcurrent(ctx context.Context, containers Containers, fn func(context.Context, container.Summary) error) error {
-	eg, ctx := errgroup.WithContext(ctx)
+func forEachContainerConcurrent(ctx context.Context, maxConcurrency int, containers Containers, fn func(context.Context, container.Summary) error) error {
+	eg, ctx := newLimitedErrgroup(ctx, maxConcurrency)
 	for _, ctr := range containers {
 		eg.Go(func() error {
 			return fn(ctx, ctr)

pkg/compose/dependencies.go

@@ -49,7 +49,13 @@ type graphTraversal struct {
 	targetServiceStatus         ServiceStatus
 	adjacentServiceStatusToSkip ServiceStatus
 
-	visitorFn      func(context.Context, string) error
+	visitorFn func(context.Context, string) error
+	// maxConcurrency bounds concurrent node (service) visits, not concurrent
+	// engine calls: it's only a correct proxy for --parallel when visitorFn
+	// makes exactly one engine call per node (e.g. build_classic.go). A
+	// visitor that fans out multiple engine calls per node — like restart's,
+	// one per container — needs its own call-level limiter shared across
+	// nodes instead (see restart.go), or this bound is too coarse to help.
 	maxConcurrency int
 }
 

pkg/compose/down.go

@@ -28,7 +28,6 @@ import (
 	"github.com/moby/moby/api/types/image"
 	"github.com/moby/moby/client"
 	"github.com/sirupsen/logrus"
-	"golang.org/x/sync/errgroup"
 
 	"github.com/docker/compose/v5/pkg/api"
 	"github.com/docker/compose/v5/pkg/utils"
@@ -132,7 +131,7 @@ func (s *composeService) down(ctx context.Context, projectName string, options a
 		logrus.Warnf("Warning: No resource found to remove for project %q.", projectName)
 	}
 
-	eg, ctx := errgroup.WithContext(ctx)
+	eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency)
 	for _, op := range ops {
 		eg.Go(op)
 	}
@@ -155,7 +154,7 @@ func (s *composeService) ensureVolumesDown(ctx context.Context, project *types.P
 }
 
 func (s *composeService) ensureImagesDown(ctx context.Context, project *types.Project, options api.DownOptions) ([]downOp, error) {
-	imagePruner := NewImagePruner(s.apiClient(), project)
+	imagePruner := NewImagePruner(s.apiClient(), project, s.maxConcurrency)
 	pruneOpts := ImagePruneOptions{
 		Mode:          ImagePruneMode(options.Images),
 		RemoveOrphans: options.RemoveOrphans,
@@ -336,7 +335,7 @@ func (s *composeService) stopContainer(ctx context.Context, service *types.Servi
 }
 
 func (s *composeService) stopContainers(ctx context.Context, serv *types.ServiceConfig, containers []containerType.Summary, timeout *time.Duration, listener api.ContainerEventListener) error {
-	eg, ctx := errgroup.WithContext(ctx)
+	eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency)
 	for _, ctr := range containers {
 		eg.Go(func() error {
 			return s.stopContainer(ctx, serv, ctr, timeout, listener)
@@ -346,7 +345,7 @@ func (s *composeService) stopContainers(ctx context.Context, serv *types.Service
 }
 
 func (s *composeService) removeContainers(ctx context.Context, containers []containerType.Summary, service *types.ServiceConfig, timeout *time.Duration, volumes bool) error {
-	eg, ctx := errgroup.WithContext(ctx)
+	eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency)
 	for _, ctr := range containers {
 		eg.Go(func() error {
 			return s.stopAndRemoveContainer(ctx, ctr, service, timeout, volumes)

pkg/compose/executor.go

@@ -22,7 +22,6 @@ import (
 	"sync"
 
 	"github.com/compose-spec/compose-go/v2/types"
-	"golang.org/x/sync/errgroup"
 )
 
 // planExecutor executes a reconciliation Plan by walking the DAG and performing
@@ -102,7 +101,13 @@ func (exec *planExecutor) run(ctx context.Context, plan *Plan) error {
 	groups := exec.buildGroupTracker(plan)
 	events := exec.compose.events
 
-	eg, ctx := errgroup.WithContext(ctx)
+	// Each node's goroutine occupies its concurrency slot for the entire wait
+	// below, not just its own work, so a small maxConcurrency can serialize
+	// more than a caller might expect on a wide/shallow DAG. Forward progress
+	// is still guaranteed: plan.Nodes is topologically sorted, so a node's
+	// dependencies were always already dispatched to eg.Go by the time this
+	// loop reaches it.
+	eg, ctx := newLimitedErrgroup(ctx, exec.compose.maxConcurrency)
 	for _, node := range plan.Nodes {
 		eg.Go(func() error {
 			// Wait for all dependencies

pkg/compose/executor_test.go

@@ -21,6 +21,7 @@ import (
 	"errors"
 	"strconv"
 	"testing"
+	"time"
 
 	"github.com/compose-spec/compose-go/v2/types"
 	"github.com/moby/moby/api/types/container"
@@ -40,14 +41,14 @@ func (noopEventProcessor) Start(_ context.Context, _ string) {}
 func (noopEventProcessor) On(_ ...api.Resource)              {}
 func (noopEventProcessor) Done(_ string, _ bool)             {}
 
-func newTestService(t *testing.T) (*composeService, *mocks.MockAPIClient) {
+func newTestService(t *testing.T, opts ...Option) (*composeService, *mocks.MockAPIClient) {
 	t.Helper()
 	mockCtrl := gomock.NewController(t)
 	cli := mocks.NewMockCli(mockCtrl)
 	apiClient := mocks.NewMockAPIClient(mockCtrl)
 	cli.EXPECT().Client().Return(apiClient).AnyTimes()
 
-	svc, err := NewComposeService(cli, WithEventProcessor(noopEventProcessor{}))
+	svc, err := NewComposeService(cli, append([]Option{WithEventProcessor(noopEventProcessor{})}, opts...)...)
 	assert.NilError(t, err)
 	return svc.(*composeService), apiClient
 }
@@ -287,6 +288,64 @@ func TestExecutePlanConcurrentRemovesCacheCoherence(t *testing.T) {
 		"all removed containers should be dropped from the live view")
 }
 
+// TestExecutePlanRespectsMaxConcurrencyAcrossDependencyChain guards the
+// invariant run() relies on to stay deadlock-free once maxConcurrency bounds
+// the errgroup (see the comment on newLimitedErrgroup's call in run()):
+// plan.Nodes must stay topologically sorted, so a node's dependencies are
+// always already dispatched to eg.Go by the time the dispatch loop reaches
+// a dependent. With maxConcurrency=1, a multi-level dependency chain forces
+// strictly serial execution; if that invariant were ever broken (a node
+// added before a dependency it references), the dispatch loop would stall
+// forever waiting for a slot held by a goroutine itself waiting on a
+// not-yet-dispatched dependency — so this test would hang instead of
+// completing.
+func TestExecutePlanRespectsMaxConcurrencyAcrossDependencyChain(t *testing.T) {
+	svc, apiClient := newTestService(t, WithMaxConcurrency(1))
+
+	const depth = 3
+	ctrs := make([]container.Summary, depth)
+	for i := range ctrs {
+		ctrs[i] = container.Summary{
+			ID:    "c" + strconv.Itoa(i),
+			Names: []string{"/test-web-" + strconv.Itoa(i+1)},
+			Labels: map[string]string{
+				api.ServiceLabel:         "web",
+				api.ContainerNumberLabel: strconv.Itoa(i + 1),
+			},
+		}
+		apiClient.EXPECT().ContainerStop(gomock.Any(), ctrs[i].ID, gomock.Any()).
+			Return(client.ContainerStopResult{}, nil)
+	}
+
+	// A -> B -> C: each node depends on the previous one.
+	plan := &Plan{}
+	var last *PlanNode
+	for i := range ctrs {
+		var deps []*PlanNode
+		if last != nil {
+			deps = []*PlanNode{last}
+		}
+		last = plan.addNode(Operation{
+			Type:       OpStopContainer,
+			ResourceID: "service:web:" + strconv.Itoa(i+1),
+			Cause:      "chain",
+			Container:  &ctrs[i],
+		}, "", deps...)
+	}
+
+	exec := svc.newPlanExecutor(&types.Project{Name: "test"}, emptyObservedState("test"))
+
+	done := make(chan error, 1)
+	go func() { done <- exec.run(t.Context(), plan) }()
+
+	select {
+	case err := <-done:
+		assert.NilError(t, err)
+	case <-time.After(5 * time.Second):
+		t.Fatal("run() deadlocked: a bounded errgroup requires plan.Nodes to stay topologically sorted")
+	}
+}
+
 // TestExecutePlanRecreateVolume drives the destructive core of a volume
 // recreation — stop container → remove container → remove volume → create
 // volume — end to end through the executor, asserting each Docker API call

pkg/compose/image_pruner.go

@@ -28,7 +28,6 @@ import (
 	"github.com/moby/moby/api/types/image"
 	"github.com/moby/moby/client"
 	"github.com/sirupsen/logrus"
-	"golang.org/x/sync/errgroup"
 
 	"github.com/docker/compose/v5/pkg/api"
 )
@@ -59,15 +58,17 @@ type ImagePruneOptions struct {
 
 // ImagePruner handles image removal during Compose `down` operations.
 type ImagePruner struct {
-	client  client.ImageAPIClient
-	project *types.Project
+	client         client.ImageAPIClient
+	project        *types.Project
+	maxConcurrency int
 }
 
 // NewImagePruner creates an ImagePruner object for a project.
-func NewImagePruner(imageClient client.ImageAPIClient, project *types.Project) *ImagePruner {
+func NewImagePruner(imageClient client.ImageAPIClient, project *types.Project, maxConcurrency int) *ImagePruner {
 	return &ImagePruner{
-		client:  imageClient,
-		project: project,
+		client:         imageClient,
+		project:        project,
+		maxConcurrency: maxConcurrency,
 	}
 }
 
@@ -173,8 +174,7 @@ func (s *composeService) removeDanglingImages(ctx context.Context, projectName s
 
 	var mu sync.Mutex
 	var removed []string
-	eg, ctx := errgroup.WithContext(ctx)
-	eg.SetLimit(s.maxConcurrency)
+	eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency)
 	for _, img := range res.Items {
 		if keep(img) {
 			continue
@@ -225,7 +225,7 @@ func (p *ImagePruner) filterImagesByExistence(ctx context.Context, imageNames []
 	var mu sync.Mutex
 	var ret []string
 
-	eg, ctx := errgroup.WithContext(ctx)
+	eg, ctx := newLimitedErrgroup(ctx, p.maxConcurrency)
 	for _, img := range imageNames {
 		eg.Go(func() error {
 			_, err := p.client.ImageInspect(ctx, img)

pkg/compose/images.go

@@ -35,7 +35,6 @@ import (
 	godigest "github.com/opencontainers/go-digest"
 	specs "github.com/opencontainers/image-spec/specs-go/v1"
 	"github.com/sirupsen/logrus"
-	"golang.org/x/sync/errgroup"
 
 	"github.com/docker/compose/v5/pkg/api"
 )
@@ -71,7 +70,7 @@ func (s *composeService) Images(ctx context.Context, projectName string, options
 
 	summary := map[string]api.ImageSummary{}
 	var mux sync.Mutex
-	eg, ctx := errgroup.WithContext(ctx)
+	eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency)
 	for _, ctr := range containers {
 		eg.Go(func() error {
 			img, err := s.containerImageSummary(ctx, ctr, withPlatform)
@@ -164,7 +163,7 @@ func (s *composeService) inspectLocalImages(ctx context.Context, repoTags []stri
 	}
 	inspections := map[string]client.ImageInspectResult{}
 	l := sync.Mutex{}
-	eg, ctx := errgroup.WithContext(ctx)
+	eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency)
 	for _, repoTag := range repoTags {
 		eg.Go(func() error {
 			inspect, err := s.apiClient().ImageInspect(ctx, repoTag, opts...)

pkg/compose/kill.go

@@ -56,7 +56,7 @@ func (s *composeService) kill(ctx context.Context, projectName string, options a
 		return api.ErrNoResources
 	}
 
-	return forEachContainerConcurrent(ctx, containers, func(ctx context.Context, ctr container.Summary) error {
+	return forEachContainerConcurrent(ctx, s.maxConcurrency, containers, func(ctx context.Context, ctr container.Summary) error {
 		eventName := getContainerProgressName(ctr)
 		s.events.On(newEvent(eventName, api.Working, api.StatusKilling))
 		_, err := s.apiClient().ContainerKill(ctx, ctr.ID, client.ContainerKillOptions{