fix: close remaining --parallel gaps on down/stop/start

down and stop rebuilt their per-container limiter on every service
visit, so independent services could each spend a full budget
concurrently -- the same leak already fixed on restart. start's
InDependencyOrder call had no concurrency option, leaving every
`up`'s container-start phase fully unbounded. wait.go goes back to
unbounded on purpose: ContainerWait blocks per container
indefinitely, so capping it would serialize waits meant to run
concurrently.

Also extracts the shared per-container fan-out into one helper reused
by restart/down/stop, and dedupes the concurrency-guard test fixtures.

Signed-off-by: Guillaume Lours <[email protected]>

pkg/compose/compose.go

@@ -200,6 +200,27 @@ func releaseSlot(limiter *semaphore.Weighted) {
 	}
 }
 
+// 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/containers.go

@@ -209,7 +209,11 @@ func (containers Containers) filter(predicates ...containerPredicate) Containers
 	return filtered
 }
 
-// forEachContainerConcurrent runs fn for every container concurrently and waits for all goroutines.
+// 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 {

pkg/compose/down.go

@@ -30,6 +30,7 @@ import (
 	"github.com/moby/moby/api/types/image"
 	"github.com/moby/moby/client"
 	"github.com/sirupsen/logrus"
+	"golang.org/x/sync/semaphore"
 
 	"github.com/docker/compose/v5/pkg/api"
 	"github.com/docker/compose/v5/pkg/utils"
@@ -96,8 +97,15 @@ func (s *composeService) down(ctx context.Context, projectName string, options a
 		resourceToRemove = true
 	}
 
+	// shared by every service so the dependency-order fan-out and the
+	// per-service container fan-out combined never exceed maxConcurrency
+	// concurrent container removals — a per-service bound alone allows as
+	// many independent services to run at once as the graph permits, each
+	// with its own maxConcurrency budget (same fix as restart.go)
+	limiter := newOptionalLimiter(s.maxConcurrency)
+
 	err = InReverseDependencyOrder(ctx, project, func(c context.Context, service string) error {
-		return s.downService(ctx, project, containers, options, service)
+		return s.downService(ctx, project, containers, options, service, limiter)
 	}, WithRootNodesAndDown(options.Services))
 	if err != nil {
 		return err
@@ -105,7 +113,7 @@ func (s *composeService) down(ctx context.Context, projectName string, options a
 
 	orphans := containers.filter(isOrphaned(project))
 	if options.RemoveOrphans && len(orphans) > 0 {
-		err := s.removeContainers(ctx, orphans, nil, options.Timeout, false)
+		err := s.removeContainers(ctx, orphans, nil, options.Timeout, false, limiter)
 		if err != nil {
 			return err
 		}
@@ -424,24 +432,30 @@ func (s *composeService) stopContainer(ctx context.Context, service *types.Servi
 	return nil
 }
 
-func (s *composeService) stopContainers(ctx context.Context, serv *types.ServiceConfig, containers []containerType.Summary, timeout *time.Duration, listener api.ContainerEventListener) error {
+func (s *composeService) stopContainers(
-	eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency)
+	ctx context.Context,
-	for _, ctr := range containers {
+	serv *types.ServiceConfig,
-		eg.Go(func() error {
+	containers []containerType.Summary,
-			return s.stopContainer(ctx, serv, ctr, timeout, listener)
+	timeout *time.Duration,
-		})
+	listener api.ContainerEventListener,
-	}
+	limiter *semaphore.Weighted,
-	return eg.Wait()
+) error {
+	return forEachContainerWithLimiter(ctx, limiter, containers, func(ctx context.Context, ctr containerType.Summary) error {
+		return s.stopContainer(ctx, serv, ctr, timeout, listener)
+	})
 }
 
-func (s *composeService) removeContainers(ctx context.Context, containers []containerType.Summary, service *types.ServiceConfig, timeout *time.Duration, volumes bool) error {
+func (s *composeService) removeContainers(
-	eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency)
+	ctx context.Context,
-	for _, ctr := range containers {
+	containers []containerType.Summary,
-		eg.Go(func() error {
+	service *types.ServiceConfig,
-			return s.stopAndRemoveContainer(ctx, ctr, service, timeout, volumes)
+	timeout *time.Duration,
-		})
+	volumes bool,
-	}
+	limiter *semaphore.Weighted,
-	return eg.Wait()
+) error {
+	return forEachContainerWithLimiter(ctx, limiter, containers, func(ctx context.Context, ctr containerType.Summary) error {
+		return s.stopAndRemoveContainer(ctx, ctr, service, timeout, volumes)
+	})
 }
 
 func (s *composeService) stopAndRemoveContainer(ctx context.Context, ctr containerType.Summary, service *types.ServiceConfig, timeout *time.Duration, volumes bool) error {
@@ -471,10 +485,10 @@ func (s *composeService) stopAndRemoveContainer(ctx context.Context, ctr contain
 // own project containers — the relay deployed when it published endpoints —
 // and the plugin only removes the provider's own resource, so the containers
 // go first, mirroring up, which provisions the resource before the relay.
-func (s *composeService) downService(ctx context.Context, project *types.Project, containers Containers, options api.DownOptions, service string) error {
+func (s *composeService) downService(ctx context.Context, project *types.Project, containers Containers, options api.DownOptions, service string, limiter *semaphore.Weighted) error {
 	serv := project.Services[service]
 	serviceContainers := containers.filter(isService(service))
-	if err := s.removeContainers(ctx, serviceContainers, &serv, options.Timeout, options.Volumes); err != nil {
+	if err := s.removeContainers(ctx, serviceContainers, &serv, options.Timeout, options.Volumes, limiter); err != nil {
 		return err
 	}
 	if serv.Provider != nil {

pkg/compose/down_test.go

@@ -23,6 +23,7 @@ import (
 	"os"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/compose-spec/compose-go/v2/types"
 	"github.com/containerd/errdefs"
@@ -99,6 +100,41 @@ func TestDown(t *testing.T) {
 	assert.NilError(t, err)
 }
 
+// TestDown_ConcurrencyIsBoundedAcrossServices guards against the same
+// per-service budget leak fixed on restart.go and stop.go:
+// InReverseDependencyOrder dispatches independent services concurrently, so
+// a limiter created fresh inside removeContainers for each service visit
+// would let each one spend its own maxConcurrency budget at the same time.
+func TestDown_ConcurrencyIsBoundedAcrossServices(t *testing.T) {
+	svc, apiClient := newTestService(t, WithMaxConcurrency(1))
+
+	const numServices = 4
+	project, containers := nIndependentServiceContainers(numServices)
+
+	apiClient.EXPECT().ContainerList(gomock.Any(), gomock.Any()).
+		Return(client.ContainerListResult{Items: containers}, nil)
+	apiClient.EXPECT().ContainerList(gomock.Any(), gomock.Any()).
+		Return(client.ContainerListResult{}, nil) // removePreStartHookContainers lookup
+
+	apiClient.EXPECT().ContainerStop(gomock.Any(), gomock.Any(), gomock.Any()).
+		Return(client.ContainerStopResult{}, nil).
+		Times(numServices)
+
+	tracker := &peakConcurrencyTracker{}
+	apiClient.EXPECT().ContainerRemove(gomock.Any(), gomock.Any(), gomock.Any()).
+		DoAndReturn(func(context.Context, string, client.ContainerRemoveOptions) (client.ContainerRemoveResult, error) {
+			tracker.enter()
+			time.Sleep(20 * time.Millisecond) // widen the window for a concurrency violation to show up
+			tracker.leave()
+			return client.ContainerRemoveResult{}, nil
+		}).
+		Times(numServices)
+
+	err := svc.down(t.Context(), "prj", compose.DownOptions{Project: project})
+	assert.NilError(t, err)
+	assert.Equal(t, tracker.Peak(), 1, "down must never run more than maxConcurrency ContainerRemove calls at once")
+}
+
 func TestDownWithGivenServices(t *testing.T) {
 	mockCtrl := gomock.NewController(t)
 	defer mockCtrl.Finish()

pkg/compose/executor_test.go

@@ -19,7 +19,9 @@ package compose
 import (
 	"context"
 	"errors"
+	"fmt"
 	"strconv"
+	"sync"
 	"testing"
 	"time"
 
@@ -42,6 +44,51 @@ func (noopEventProcessor) Start(_ context.Context, _ string) {}
 func (noopEventProcessor) On(_ ...api.Resource)              {}
 func (noopEventProcessor) Done(_ string, _ bool)             {}
 
+// peakConcurrencyTracker records the highest number of overlapping
+// enter()/leave() pairs seen, used to assert a --parallel bound was honored.
+type peakConcurrencyTracker struct {
+	mu      sync.Mutex
+	current int
+	peak    int
+}
+
+func (t *peakConcurrencyTracker) enter() {
+	t.mu.Lock()
+	t.current++
+	if t.current > t.peak {
+		t.peak = t.current
+	}
+	t.mu.Unlock()
+}
+
+func (t *peakConcurrencyTracker) leave() {
+	t.mu.Lock()
+	t.current--
+	t.mu.Unlock()
+}
+
+func (t *peakConcurrencyTracker) Peak() int {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	return t.peak
+}
+
+// nIndependentServiceContainers builds n services with no depends_on between
+// them — so InDependencyOrder/InReverseDependencyOrder dispatches them
+// concurrently — plus one non-oneoff container per service, used to assert a
+// limiter shared across service visits (see TestDown/TestStop
+// ConcurrencyIsBoundedAcrossServices).
+func nIndependentServiceContainers(n int) (*types.Project, []container.Summary) {
+	project := &types.Project{Name: "prj", Services: types.Services{}}
+	var containers []container.Summary
+	for i := range n {
+		name := fmt.Sprintf("svc%d", i)
+		project.Services[name] = types.ServiceConfig{Name: name}
+		containers = append(containers, testContainer(name, fmt.Sprintf("c%d", i), false))
+	}
+	return project, containers
+}
+
 func newTestService(t *testing.T, opts ...Option) (*composeService, *mocks.MockAPIClient) {
 	t.Helper()
 	mockCtrl := gomock.NewController(t)

pkg/compose/logs.go

@@ -107,18 +107,29 @@ func (s *composeService) selectLogsContainers(ctx context.Context, projectName s
 	return containers, nil
 }
 
+// inspectWithSlot acquires limiter's slot, then inspects the container,
+// releasing the slot on error since no caller reaches doLogContainer (which
+// owns the slot from here on) in that case.
+func (s *composeService) inspectWithSlot(ctx context.Context, limiter *semaphore.Weighted, id string) (container.InspectResponse, error) {
+	if err := acquireSlot(ctx, limiter); err != nil {
+		return container.InspectResponse{}, err
+	}
+	res, err := s.apiClient().ContainerInspect(ctx, id, client.ContainerInspectOptions{})
+	if err != nil {
+		releaseSlot(limiter)
+		return container.InspectResponse{}, err
+	}
+	return res.Container, nil
+}
+
 // logContainer streams a container's logs, warning when its logging driver
 // doesn't support reading logs
 func (s *composeService) logContainer(ctx context.Context, limiter *semaphore.Weighted, consumer api.LogConsumer, ctr container.Summary, options api.LogOptions) error {
-	if err := acquireSlot(ctx, limiter); err != nil {
+	res, err := s.inspectWithSlot(ctx, limiter, ctr.ID)
-		return err
-	}
-	res, err := s.apiClient().ContainerInspect(ctx, ctr.ID, client.ContainerInspectOptions{})
 	if err != nil {
-		releaseSlot(limiter)
 		return err
 	}
-	err = s.doLogContainer(ctx, limiter, consumer, getContainerNameWithoutProject(ctr), res.Container, options)
+	err = s.doLogContainer(ctx, limiter, consumer, getContainerNameWithoutProject(ctr), res, options)
 	if errdefs.IsNotImplemented(err) {
 		logrus.Warnf("Can't retrieve logs for %q: %s", getCanonicalContainerName(ctr), err.Error())
 		return nil
@@ -148,19 +159,15 @@ func (s *composeService) followStartedContainersLogs(
 		// would drop the whole run.
 		since := runEnds.Since(event.ID)
 		eg.Go(func() error {
-			if err := acquireSlot(ctx, limiter); err != nil {
+			res, err := s.inspectWithSlot(ctx, limiter, event.ID)
-				return err
-			}
-			res, err := s.apiClient().ContainerInspect(ctx, event.ID, client.ContainerInspectOptions{})
 			if err != nil {
-				releaseSlot(limiter)
 				return err
 			}
 			if since == "" {
-				since = logsSinceLastRun(res.Container)
+				since = logsSinceLastRun(res)
 			}
 
-			err = s.doLogContainer(ctx, limiter, consumer, event.Source, res.Container, api.LogOptions{
+			err = s.doLogContainer(ctx, limiter, consumer, event.Source, res, api.LogOptions{
 				Follow:     options.Follow,
 				Since:      since,
 				Until:      options.Until,

pkg/compose/logs_test.go

@@ -344,27 +344,14 @@ func TestComposeService_Logs_FollowLimitsConcurrentStreamOpens(t *testing.T) {
 			}, nil)
 	}
 
-	var (
+	tracker := &peakConcurrencyTracker{}
-		mu      sync.Mutex
-		current int
-		peak    int
-	)
 	writers := make(chan *io.PipeWriter, len(ids))
 	for _, id := range ids {
 		api.EXPECT().ContainerLogs(anyCancellableContext(), id, gomock.Any()).
 			DoAndReturn(func(context.Context, string, client.ContainerLogsOptions) (io.ReadCloser, error) {
-				mu.Lock()
+				tracker.enter()
-				current++
-				if current > peak {
-					peak = current
-				}
-				mu.Unlock()
-
 				time.Sleep(20 * time.Millisecond) // widen the window for a concurrency violation to show up
-
+				tracker.leave()
-				mu.Lock()
-				current--
-				mu.Unlock()
 
 				r, w := io.Pipe()
 				writers <- w
@@ -403,7 +390,7 @@ func TestComposeService_Logs_FollowLimitsConcurrentStreamOpens(t *testing.T) {
 		_ = w.Close()
 	}
 	assert.NilError(t, <-done)
-	assert.Equal(t, peak, 1, "opening follow-mode log streams must be bounded by maxConcurrency")
+	assert.Equal(t, tracker.Peak(), 1, "opening follow-mode log streams must be bounded by maxConcurrency")
 }
 
 type testLogConsumer struct {

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"
@@ -61,17 +60,9 @@ func (s *composeService) restart(ctx context.Context, projectName string, option
 			return err
 		}
 
-		eg, ctx := errgroup.WithContext(ctx)
+		return forEachContainerWithLimiter(ctx, limiter, containers.filter(isService(service)), func(ctx context.Context, ctr container.Summary) error {
-		for _, ctr := range containers.filter(isService(service)) {
+			return s.restartContainer(ctx, project.Services[service], ctr, options)
-			eg.Go(func() error {
+		})
-				if err := acquireSlot(ctx, limiter); err != nil {
-					return err
-				}
-				defer releaseSlot(limiter)
-				return s.restartContainer(ctx, project.Services[service], ctr, options)
-			})
-		}
-		return eg.Wait()
 	})
 }
 

pkg/compose/restart_test.go

@@ -22,7 +22,6 @@ import (
 	"context"
 	"fmt"
 	"net"
-	"sync"
 	"testing"
 	"time"
 
@@ -33,7 +32,6 @@ import (
 	"gotest.tools/v3/assert"
 
 	"github.com/docker/compose/v5/pkg/api"
-	"github.com/docker/compose/v5/pkg/mocks"
 )
 
 // These tests characterize the restart path before the lifecycle engines
@@ -182,19 +180,11 @@ func TestRestartContainerSkipsHooksForRelay(t *testing.T) {
 // spawning up to maxConcurrency ContainerRestart calls — up to N times the
 // documented --parallel bound. The limiter must be shared across services.
 func TestRestart_ConcurrencyIsBoundedAcrossServices(t *testing.T) {
-	mockCtrl := gomock.NewController(t)
+	svc, apiClient := newTestService(t, WithMaxConcurrency(1))
-	t.Cleanup(mockCtrl.Finish)
-	apiClient := mocks.NewMockAPIClient(mockCtrl)
-	cli := mocks.NewMockCli(mockCtrl)
-	cli.EXPECT().Client().Return(apiClient).AnyTimes()
 	apiClient.EXPECT().Ping(gomock.Any(), client.PingOptions{NegotiateAPIVersion: true}).
 		Return(client.PingResult{APIVersion: "1.44"}, nil).AnyTimes()
 	apiClient.EXPECT().ClientVersion().Return("1.44").AnyTimes()
 
-	svcIface, err := NewComposeService(cli, WithMaxConcurrency(1))
-	assert.NilError(t, err)
-	svc := svcIface.(*composeService)
-
 	const numServices = 4
 	project := &types.Project{Name: "prj", Services: types.Services{}}
 	var containers Containers
@@ -207,30 +197,17 @@ func TestRestart_ConcurrencyIsBoundedAcrossServices(t *testing.T) {
 	apiClient.EXPECT().ContainerList(gomock.Any(), gomock.Any()).
 		Return(client.ContainerListResult{Items: containers}, nil)
 
-	var (
+	tracker := &peakConcurrencyTracker{}
-		mu      sync.Mutex
-		current int
-		peak    int
-	)
 	apiClient.EXPECT().ContainerRestart(gomock.Any(), gomock.Any(), gomock.Any()).
 		DoAndReturn(func(context.Context, string, client.ContainerRestartOptions) (client.ContainerRestartResult, error) {
-			mu.Lock()
+			tracker.enter()
-			current++
-			if current > peak {
-				peak = current
-			}
-			mu.Unlock()
-
 			time.Sleep(20 * time.Millisecond) // widen the window for a concurrency violation to show up
-
+			tracker.leave()
-			mu.Lock()
-			current--
-			mu.Unlock()
 			return client.ContainerRestartResult{}, nil
 		}).
 		Times(numServices)
 
-	err = svc.restart(t.Context(), "prj", api.RestartOptions{Project: project})
+	err := svc.restart(t.Context(), "prj", api.RestartOptions{Project: project})
 	assert.NilError(t, err)
-	assert.Equal(t, peak, 1, "restart must never run more than maxConcurrency ContainerRestart calls at once")
+	assert.Equal(t, tracker.Peak(), 1, "restart must never run more than maxConcurrency ContainerRestart calls at once")
 }

pkg/compose/start.go

@@ -61,6 +61,10 @@ func (s *composeService) start(ctx context.Context, projectName string, options
 		return err
 	}
 
+	// the node-level bound is a correct proxy for --parallel here, unlike
+	// restart/down/stop: startService starts a service's containers in a
+	// plain sequential loop, so one engine-call burst per node is all this
+	// traversal ever dispatches concurrently.
 	err = InDependencyOrder(ctx, project, func(c context.Context, name string) error {
 		service, err := project.GetService(name)
 		if err != nil {
@@ -68,6 +72,8 @@ func (s *composeService) start(ctx context.Context, projectName string, options
 		}
 
 		return s.startService(ctx, project, service, containers, listener, options.WaitTimeout)
+	}, func(traversal *graphTraversal) {
+		traversal.maxConcurrency = s.maxConcurrency
 	})
 	if err != nil {
 		return err

pkg/compose/start_test.go

@@ -20,10 +20,12 @@ package compose
 
 import (
 	"context"
+	"fmt"
 	"net"
 	"strconv"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/compose-spec/compose-go/v2/types"
 	"github.com/docker/cli/cli/config/configfile"
@@ -117,6 +119,53 @@ func serviceContainer(service string, num int, state container.ContainerState) c
 	}
 }
 
+// TestStart_ConcurrencyIsBoundedAcrossServices guards against the same
+// per-service budget leak fixed on restart.go/down.go/stop.go:
+// InDependencyOrder dispatches independent services concurrently, and
+// startService's own per-container loop is sequential, so a missing
+// node-level maxConcurrency option left the dispatch itself unbounded.
+// ContainerStart is serialized process-wide by startMx regardless of this
+// bound, so the injected-secret copy (which isn't) is used as the observable
+// signal instead.
+func TestStart_ConcurrencyIsBoundedAcrossServices(t *testing.T) {
+	svc, apiClient := newTestService(t, WithMaxConcurrency(1))
+
+	const numServices = 4
+	project := &types.Project{Name: "prj", Services: types.Services{}, Secrets: types.Secrets{}}
+	var containers []container.Summary
+	for i := range numServices {
+		name := fmt.Sprintf("svc%d", i)
+		secretName := fmt.Sprintf("secret%d", i)
+		project.Secrets[secretName] = types.SecretConfig{Name: secretName, Content: "shh"}
+		project.Services[name] = types.ServiceConfig{
+			Name:          name,
+			ContainerSpec: types.ContainerSpec{Secrets: []types.ServiceSecretConfig{{Source: secretName}}},
+		}
+		containers = append(containers, testContainer(name, fmt.Sprintf("c%d", i), false))
+	}
+
+	apiClient.EXPECT().ContainerList(gomock.Any(), gomock.Any()).
+		Return(client.ContainerListResult{Items: containers}, nil)
+
+	tracker := &peakConcurrencyTracker{}
+	apiClient.EXPECT().CopyToContainer(gomock.Any(), gomock.Any(), gomock.Any()).
+		DoAndReturn(func(context.Context, string, client.CopyToContainerOptions) (client.CopyToContainerResult, error) {
+			tracker.enter()
+			time.Sleep(20 * time.Millisecond) // widen the window for a concurrency violation to show up
+			tracker.leave()
+			return client.CopyToContainerResult{}, nil
+		}).
+		Times(numServices)
+
+	apiClient.EXPECT().ContainerStart(gomock.Any(), gomock.Any(), gomock.Any()).
+		Return(client.ContainerStartResult{}, nil).
+		Times(numServices)
+
+	err := svc.start(t.Context(), "prj", api.StartOptions{Project: project}, nil)
+	assert.NilError(t, err)
+	assert.Equal(t, tracker.Peak(), 1, "start must never dispatch more than maxConcurrency services concurrently")
+}
+
 func TestStartService_AlreadyRunningIsSilent(t *testing.T) {
 	svc, _, rec := newStartTestService(t)
 

pkg/compose/stop.go

@@ -48,6 +48,13 @@ func (s *composeService) stop(ctx context.Context, projectName string, options a
 		options.Services = project.ServiceNames()
 	}
 
+	// shared by every service so the dependency-order fan-out and the
+	// per-service container fan-out combined never exceed maxConcurrency
+	// concurrent container stops — a per-service bound alone allows as many
+	// independent services to run at once as the graph permits, each with
+	// its own maxConcurrency budget (same fix as restart.go)
+	limiter := newOptionalLimiter(s.maxConcurrency)
+
 	return InReverseDependencyOrder(ctx, project, func(c context.Context, service string) error {
 		if !slices.Contains(options.Services, service) {
 			return nil
@@ -57,7 +64,7 @@ func (s *composeService) stop(ctx context.Context, projectName string, options a
 		// deployed when it published endpoints — and the plugin's own stop
 		// hook (if any) only concerns the provider's resource, so the
 		// container is stopped the same way as for any other service.
-		if err := s.stopContainers(ctx, &serv, containers.filter(isService(service)).filter(isNotOneOff), options.Timeout, event); err != nil {
+		if err := s.stopContainers(ctx, &serv, containers.filter(isService(service)).filter(isNotOneOff), options.Timeout, event, limiter); err != nil {
 			return err
 		}
 		if serv.Provider != nil {

pkg/compose/stop_test.go

@@ -17,6 +17,7 @@
 package compose
 
 import (
+	"context"
 	"strings"
 	"testing"
 	"time"
@@ -66,3 +67,32 @@ func TestStopTimeout(t *testing.T) {
 	})
 	assert.NilError(t, err)
 }
+
+// TestStop_ConcurrencyIsBoundedAcrossServices guards against the same
+// per-service budget leak fixed on restart.go and down.go: InReverseDependencyOrder
+// dispatches independent services concurrently, so a limiter created fresh
+// inside stopContainers for each service visit would let each one spend its
+// own maxConcurrency budget at the same time.
+func TestStop_ConcurrencyIsBoundedAcrossServices(t *testing.T) {
+	svc, apiClient := newTestService(t, WithMaxConcurrency(1))
+
+	const numServices = 4
+	project, containers := nIndependentServiceContainers(numServices)
+
+	apiClient.EXPECT().ContainerList(gomock.Any(), gomock.Any()).
+		Return(client.ContainerListResult{Items: containers}, nil)
+
+	tracker := &peakConcurrencyTracker{}
+	apiClient.EXPECT().ContainerStop(gomock.Any(), gomock.Any(), gomock.Any()).
+		DoAndReturn(func(context.Context, string, client.ContainerStopOptions) (client.ContainerStopResult, error) {
+			tracker.enter()
+			time.Sleep(20 * time.Millisecond) // widen the window for a concurrency violation to show up
+			tracker.leave()
+			return client.ContainerStopResult{}, nil
+		}).
+		Times(numServices)
+
+	err := svc.stop(t.Context(), "prj", compose.StopOptions{Project: project}, nil)
+	assert.NilError(t, err)
+	assert.Equal(t, tracker.Peak(), 1, "stop must never run more than maxConcurrency ContainerStop calls at once")
+}

pkg/compose/wait.go

@@ -22,6 +22,7 @@ import (
 	"fmt"
 
 	"github.com/moby/moby/client"
+	"golang.org/x/sync/errgroup"
 
 	"github.com/docker/compose/v5/pkg/api"
 )
@@ -54,7 +55,10 @@ func (s *composeService) Wait(ctx context.Context, projectName string, options a
 		return 0, fmt.Errorf("no containers for project %q", projectName)
 	}
 
-	eg, waitCtx := newLimitedErrgroup(ctx, s.maxConcurrency)
+	// ContainerWait blocks until the container exits, so it must not be
+	// bound by --parallel: capping concurrency here would serialize waits
+	// that are meant to run together (same rationale as waitDependencies).
+	eg, waitCtx := errgroup.WithContext(ctx)
 	var statusCode int64
 	for _, ctr := range containers {
 		eg.Go(func() error {

pkg/compose/watch.go

@@ -197,7 +197,9 @@ func (s *composeService) watch(ctx context.Context, project *types.Project, opti
 	if err != nil {
 		return nil, err
 	}
-	eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency)
+	// exactly one goroutine is ever submitted to eg (below); the real
+	// --parallel bound for watch is handleWatchBatch's per-batch fan-out.
+	eg, ctx := errgroup.WithContext(ctx)
 
 	var (
 		rules []watchRule