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,88 @@ 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)
+	}
+}
+
+// panicSafeReleaseSlot is deferred by callers that release a slot earlier
+// than function exit on the success path (see inspectWithSlot and
+// doLogContainer) to avoid leaking it if the guarded call panics before
+// reaching that point. It is a no-op unless the deferring goroutine is
+// unwinding from a panic, in which case it releases the slot and re-panics.
+func panicSafeReleaseSlot(limiter *semaphore.Weighted) {
+	if p := recover(); p != nil {
+		releaseSlot(limiter)
+		panic(p)
+	}
+}
+
+// forEachContainerWithLimiter runs fn concurrently for each container,
+// bounded by limiter. Unlike newLimitedErrgroup, limiter is built by the
+// caller and can be shared across several concurrently-dispatched calls
+// (e.g. one per service visited by InDependencyOrder), so the combined
+// concurrency across all of them never exceeds the configured budget.
+// Use forEachContainerConcurrent (containers.go) instead when the call is
+// standalone and doesn't need to share its budget with any other call.
+func forEachContainerWithLimiter(ctx context.Context, limiter *semaphore.Weighted, containers []container.Summary, fn func(context.Context, container.Summary) error) error {
+	eg, ctx := errgroup.WithContext(ctx)
+	for _, ctr := range containers {
+		eg.Go(func() error {
+			if err := acquireSlot(ctx, limiter); err != nil {
+				return err
+			}
+			defer releaseSlot(limiter)
+			return fn(ctx, ctr)
+		})
+	}
+	return eg.Wait()
+}
+
 // 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"
 )
@@ -185,9 +184,13 @@ func (containers Containers) filter(predicates ...containerPredicate) Containers
 	return filtered
 }
 
-// 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)
+// forEachContainerConcurrent runs fn for every container concurrently and
+// waits for all goroutines. Use forEachContainerWithLimiter (compose.go)
+// instead when the concurrency budget must be shared across several
+// concurrently-dispatched calls, e.g. one per service visited by
+// InDependencyOrder.
+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/cp.go

@@ -29,7 +29,6 @@ import (
 	"github.com/moby/go-archive"
 	"github.com/moby/moby/api/types/container"
 	"github.com/moby/moby/client"
-	"golang.org/x/sync/errgroup"
 
 	"github.com/docker/compose/v5/pkg/api"
 )
@@ -79,37 +78,31 @@ func (s *composeService) copy(ctx context.Context, projectName string, options a
 		return err
 	}
 
-	g := errgroup.Group{}
-	for _, cont := range containers {
-		ctr := cont
-		g.Go(func() error {
-			name := getCanonicalContainerName(ctr)
-			var msg string
-			if direction == fromService {
-				msg = fmt.Sprintf("%s:%s to %s", name, srcPath, dstPath)
-			} else {
-				msg = fmt.Sprintf("%s to %s:%s", srcPath, name, dstPath)
-			}
-			s.events.On(api.Resource{
-				ID:      name,
-				Text:    api.StatusCopying,
-				Details: msg,
-				Status:  api.Working,
-			})
-			if err := copyFunc(ctx, ctr.ID, srcPath, dstPath, options); err != nil {
-				return err
-			}
-			s.events.On(api.Resource{
-				ID:      name,
-				Text:    api.StatusCopied,
-				Details: msg,
-				Status:  api.Done,
-			})
-			return nil
+	return forEachContainerConcurrent(ctx, s.maxConcurrency, containers, func(ctx context.Context, ctr container.Summary) error {
+		name := getCanonicalContainerName(ctr)
+		var msg string
+		if direction == fromService {
+			msg = fmt.Sprintf("%s:%s to %s", name, srcPath, dstPath)
+		} else {
+			msg = fmt.Sprintf("%s to %s:%s", srcPath, name, dstPath)
+		}
+		s.events.On(api.Resource{
+			ID:      name,
+			Text:    api.StatusCopying,
+			Details: msg,
+			Status:  api.Working,
 		})
-	}
-
-	return g.Wait()
+		if err := copyFunc(ctx, ctr.ID, srcPath, dstPath, options); err != nil {
+			return err
+		}
+		s.events.On(api.Resource{
+			ID:      name,
+			Text:    api.StatusCopied,
+			Details: msg,
+			Status:  api.Done,
+		})
+		return nil
+	})
 }
 
 func (s *composeService) listContainersTargetedForCopy(ctx context.Context, projectName string, options api.CopyOptions, direction copyDirection, serviceName string) (Containers, error) {

pkg/compose/cp_test.go

@@ -0,0 +1,75 @@
+/*
+   Copyright 2020 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 (
+	"context"
+	"io"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/moby/moby/api/types/container"
+	"github.com/moby/moby/client"
+	"go.uber.org/mock/gomock"
+	"gotest.tools/v3/assert"
+
+	compose "github.com/docker/compose/v5/pkg/api"
+)
+
+// TestCopyToService_ConcurrencyIsBoundedAcrossContainers guards against the
+// same unbounded-fan-out bug fixed everywhere else in this package: `compose
+// cp` copying to a service fans out one goroutine per matched container via
+// a bare errgroup.Group, ignoring --parallel entirely.
+func TestCopyToService_ConcurrencyIsBoundedAcrossContainers(t *testing.T) {
+	svc, apiClient := newTestService(t, WithMaxConcurrency(1))
+
+	srcFile := filepath.Join(t.TempDir(), "src.txt")
+	assert.NilError(t, os.WriteFile(srcFile, []byte("hello"), 0o644))
+
+	const numContainers = 4
+	var containers []container.Summary
+	for i := range numContainers {
+		containers = append(containers, testContainer("myservice", string(rune('a'+i)), false))
+	}
+
+	apiClient.EXPECT().ContainerList(gomock.Any(), gomock.Any()).
+		Return(client.ContainerListResult{Items: containers}, nil)
+
+	apiClient.EXPECT().ContainerStatPath(gomock.Any(), gomock.Any(), gomock.Any()).
+		Return(client.ContainerStatPathResult{Stat: container.PathStat{Mode: os.ModeDir}}, nil).
+		Times(numContainers)
+
+	tracker := &peakConcurrencyTracker{}
+	apiClient.EXPECT().CopyToContainer(gomock.Any(), gomock.Any(), gomock.Any()).
+		DoAndReturn(func(_ context.Context, _ string, opts client.CopyToContainerOptions) (client.CopyToContainerResult, error) {
+			tracker.enter()
+			time.Sleep(20 * time.Millisecond) // widen the window for a concurrency violation to show up
+			tracker.leave()
+			_, err := io.Copy(io.Discard, opts.Content)
+			return client.CopyToContainerResult{}, err
+		}).
+		Times(numContainers)
+
+	err := svc.copy(t.Context(), "prj", compose.CopyOptions{
+		Source:      srcFile,
+		Destination: "myservice:/dest",
+	})
+	assert.NilError(t, err)
+	assert.Equal(t, tracker.Peak(), 1, "cp must never run more than maxConcurrency CopyToContainer calls at once")
+}

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
 }
 
@@ -135,10 +141,11 @@ func (t *graphTraversal) visit(ctx context.Context, g *Graph) error {
 		return nil
 	}
 
-	eg, ctx := errgroup.WithContext(ctx)
+	limit := 0
 	if t.maxConcurrency > 0 {
-		eg.SetLimit(t.maxConcurrency + 1)
+		limit = t.maxConcurrency + 1
 	}
+	eg, ctx := newLimitedErrgroup(ctx, limit)
 	nodeCh := make(chan *Vertex, expect)
 	defer close(nodeCh)
 	// nodeCh need to allow n=expect writers while reader goroutine could have returner after ctx.Done