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

pkg/compose/compose.go

@@ -35,6 +35,7 @@ import (
 	"github.com/moby/moby/api/types/swarm"
 	"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/dryrun"
@@ -151,14 +152,27 @@ 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
+}
+
 // 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/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"
@@ -138,7 +137,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)
 	}
@@ -161,7 +160,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,
@@ -342,7 +341,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)
@@ -352,7 +351,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/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{

pkg/compose/logs.go

@@ -42,7 +42,7 @@ func (s *composeService) Logs(
 		return err
 	}
 
-	eg, ctx := errgroup.WithContext(ctx)
+	eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency)
 	for _, ctr := range containers {
 		eg.Go(func() error {
 			return s.logContainer(ctx, consumer, ctr, options)

pkg/compose/model.go

@@ -31,7 +31,6 @@ import (
 	"github.com/docker/cli/cli-plugins/manager"
 	"github.com/moby/moby/client/pkg/versions"
 	"github.com/spf13/cobra"
-	"golang.org/x/sync/errgroup"
 
 	"github.com/docker/compose/v5/pkg/api"
 )
@@ -48,7 +47,7 @@ func (s *composeService) ensureModels(ctx context.Context, project *types.Projec
 	defer mdlAPI.Close()
 	availableModels, err := mdlAPI.ListModels(ctx)
 
-	eg, ctx := errgroup.WithContext(ctx)
+	eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency)
 	eg.Go(func() error {
 		return mdlAPI.SetModelVariables(ctx, project)
 	})

pkg/compose/pause.go

@@ -42,7 +42,7 @@ func (s *composeService) pause(ctx context.Context, projectName string, options
 		containers = containers.filter(isService(options.Project.ServiceNames()...))
 	}
 
-	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 {
 		_, err := s.apiClient().ContainerPause(ctx, ctr.ID, client.ContainerPauseOptions{})
 		if err == nil {
 			s.events.On(newEvent(getContainerProgressName(ctr), api.Done, "Paused"))
@@ -67,7 +67,7 @@ func (s *composeService) unPause(ctx context.Context, projectName string, option
 		containers = containers.filter(isService(options.Project.ServiceNames()...))
 	}
 
-	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 {
 		_, err := s.apiClient().ContainerUnpause(ctx, ctr.ID, client.ContainerUnpauseOptions{})
 		if err == nil {
 			s.events.On(newEvent(getContainerProgressName(ctr), api.Done, "Unpaused"))

pkg/compose/ps.go

@@ -23,7 +23,6 @@ import (
 
 	"github.com/moby/moby/api/types/container"
 	"github.com/moby/moby/client"
-	"golang.org/x/sync/errgroup"
 
 	"github.com/docker/compose/v5/pkg/api"
 )
@@ -43,7 +42,7 @@ func (s *composeService) Ps(ctx context.Context, projectName string, options api
 		containers = containers.filter(isService(options.Services...))
 	}
 	summary := make([]api.ContainerSummary, len(containers))
-	eg, ctx := errgroup.WithContext(ctx)
+	eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency)
 	for i, ctr := range containers {
 		eg.Go(func() error {
 			var err error

pkg/compose/pull.go

@@ -69,8 +69,7 @@ func (s *composeService) pull(ctx context.Context, project *types.Project, opts
 		return err
 	}
 
-	eg, ctx := errgroup.WithContext(ctx)
-	eg.SetLimit(s.maxConcurrency)
+	eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency)
 
 	p := &imagePuller{
 		composeService: s,
@@ -429,8 +428,7 @@ func (s *composeService) pullRequiredImages(ctx context.Context, project *types.
 
 	// the errgroup context is canceled as soon as Wait returns; the post-pull
 	// resolution below needs the caller's context
-	eg, pullCtx := errgroup.WithContext(ctx)
-	eg.SetLimit(s.maxConcurrency)
+	eg, pullCtx := newLimitedErrgroup(ctx, s.maxConcurrency)
 	pulled := map[string]bool{}
 	var mutex sync.Mutex
 	for name, service := range needPull {

pkg/compose/push.go

@@ -29,7 +29,6 @@ import (
 	"github.com/docker/go-units"
 	"github.com/moby/moby/api/types/jsonstream"
 	"github.com/moby/moby/client"
-	"golang.org/x/sync/errgroup"
 
 	"github.com/docker/compose/v5/internal/registry"
 	"github.com/docker/compose/v5/pkg/api"
@@ -45,8 +44,7 @@ func (s *composeService) Push(ctx context.Context, project *types.Project, optio
 }
 
 func (s *composeService) push(ctx context.Context, project *types.Project, options api.PushOptions) error {
-	eg, ctx := errgroup.WithContext(ctx)
-	eg.SetLimit(s.maxConcurrency)
+	eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency)
 
 	for _, service := range project.Services {
 		if service.Build == nil || service.Image == "" {

pkg/compose/remove.go

@@ -22,7 +22,6 @@ import (
 	"strings"
 
 	"github.com/moby/moby/client"
-	"golang.org/x/sync/errgroup"
 
 	"github.com/docker/compose/v5/pkg/api"
 )
@@ -98,7 +97,7 @@ func (s *composeService) Remove(ctx context.Context, projectName string, options
 }
 
 func (s *composeService) remove(ctx context.Context, containers Containers, options api.RemoveOptions) error {
-	eg, ctx := errgroup.WithContext(ctx)
+	eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency)
 	for _, ctr := range containers {
 		eg.Go(func() error {
 			eventName := getContainerProgressName(ctr)

pkg/compose/restart.go

@@ -23,7 +23,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"
 	"github.com/docker/compose/v5/pkg/utils"
@@ -53,7 +52,7 @@ func (s *composeService) restart(ctx context.Context, projectName string, option
 			return err
 		}
 
-		eg, ctx := errgroup.WithContext(ctx)
+		eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency)
 		for _, ctr := range containers.filter(isService(service)) {
 			eg.Go(func() error {
 				return s.restartContainer(ctx, project.Services[service], ctr, options)