fix: share --parallel budget across down's ops and up's log-attach

down's outer ops dispatch bounded only how many ops ran concurrently,
separate from the semaphore already shared by image removal -- a
single network plus --rmi could spend up to 2x the configured
budget. Network and volume removal now share that same semaphore.

up's post-attach log streaming called ContainerInspect directly and
passed no limiter to doLogContainer, unlike `compose logs --follow`'s
equivalent path, so it never shared --parallel's budget either.

The plan executor also held a node's slot for its whole dependency
wait instead of just its own work, and internal/sync/tar.go's file
sync used a hardcoded cap of 16 instead of --parallel.

Signed-off-by: Guillaume Lours <[email protected]>
Co-Authored-By: Claude Sonnet 5 <[email protected]>

internal/sync/tar.go

@@ -53,14 +53,19 @@ type Tar struct {
 	client LowLevelClient
 
 	projectName string
+	// maxConcurrency bounds concurrent per-container sync calls. <=0
+	// (including the Go zero-value) leaves it unlimited, since
+	// errgroup.SetLimit(0) means "allow zero goroutines", not "unlimited".
+	maxConcurrency int
 }
 
 var _ Syncer = &Tar{}
 
-func NewTar(projectName string, client LowLevelClient) *Tar {
+func NewTar(projectName string, client LowLevelClient, maxConcurrency int) *Tar {
 	return &Tar{
-		projectName: projectName,
-		client:      client,
+		projectName:    projectName,
+		client:         client,
+		maxConcurrency: maxConcurrency,
 	}
 }
 
@@ -93,7 +98,9 @@ func (t *Tar) Sync(ctx context.Context, service string, paths []*PathMapping) er
 		errs  = make([]error, 0, len(containers)*2) // max 2 errs per container
 	)
 
-	eg.SetLimit(16) // arbitrary limit, adjust to taste :D
+	if t.maxConcurrency > 0 {
+		eg.SetLimit(t.maxConcurrency)
+	}
 	for i := range containers {
 		containerID := containers[i].ID
 		tarReader := tarArchive(pathsToCopy, keepImpliedDirectories)

internal/sync/tar_test.go

@@ -22,7 +22,9 @@ import (
 	"os"
 	"path/filepath"
 	"runtime"
+	"sync"
 	"testing"
+	"time"
 
 	"github.com/moby/moby/api/types/container"
 	"gotest.tools/v3/assert"
@@ -31,6 +33,7 @@ import (
 
 // fakeLowLevelClient records calls made to it for test assertions.
 type fakeLowLevelClient struct {
+	mu         sync.Mutex
 	containers []container.Summary
 	execCmds   [][]string
 	untarCount int
@@ -38,6 +41,9 @@ type fakeLowLevelClient struct {
 	untarErrs []error
 	// untarHeaders[i] holds the headers the i-th Untar call received, by entry name
 	untarHeaders []map[string]tar.Header
+	// onUntar, when set, is invoked at the start of every Untar call --
+	// used to observe how many run concurrently.
+	onUntar func()
 }
 
 func (f *fakeLowLevelClient) ContainersForService(_ context.Context, _ string, _ string) ([]container.Summary, error) {
@@ -50,6 +56,9 @@ func (f *fakeLowLevelClient) Exec(_ context.Context, _ string, cmd []string, _ i
 }
 
 func (f *fakeLowLevelClient) Untar(_ context.Context, _ string, reader io.ReadCloser) error {
+	if f.onUntar != nil {
+		f.onUntar()
+	}
 	headers := map[string]tar.Header{}
 	tr := tar.NewReader(reader)
 	for {
@@ -62,6 +71,9 @@ func (f *fakeLowLevelClient) Untar(_ context.Context, _ string, reader io.ReadCl
 		}
 		headers[header.Name] = *header
 	}
+
+	f.mu.Lock()
+	defer f.mu.Unlock()
 	f.untarHeaders = append(f.untarHeaders, headers)
 
 	f.untarCount++
@@ -79,7 +91,7 @@ func TestSync_ExistingPath(t *testing.T) {
 	client := &fakeLowLevelClient{
 		containers: []container.Summary{{ID: "ctr1"}},
 	}
-	syncer := NewTar("proj", client)
+	syncer := NewTar("proj", client, 0)
 
 	err := syncer.Sync(t.Context(), "svc", []*PathMapping{
 		{HostPath: existingFile, ContainerPath: "/app/exists.txt"},
@@ -94,7 +106,7 @@ func TestSync_NonExistentPath(t *testing.T) {
 	client := &fakeLowLevelClient{
 		containers: []container.Summary{{ID: "ctr1"}},
 	}
-	syncer := NewTar("proj", client)
+	syncer := NewTar("proj", client, 0)
 
 	err := syncer.Sync(t.Context(), "svc", []*PathMapping{
 		{HostPath: "/no/such/file", ContainerPath: "/app/gone.txt"},
@@ -128,7 +140,7 @@ func TestSync_StatPermissionError(t *testing.T) {
 	client := &fakeLowLevelClient{
 		containers: []container.Summary{{ID: "ctr1"}},
 	}
-	syncer := NewTar("proj", client)
+	syncer := NewTar("proj", client, 0)
 
 	err := syncer.Sync(t.Context(), "svc", []*PathMapping{
 		{HostPath: targetFile, ContainerPath: "/app/secret.txt"},
@@ -140,6 +152,88 @@ func TestSync_StatPermissionError(t *testing.T) {
 	assert.Equal(t, len(client.execCmds), 0, "should not attempt delete on stat error")
 }
 
+// syncConcurrencyTracker records the highest number of overlapping
+// enter()/leave() pairs seen, used to assert Sync's concurrency bound.
+type syncConcurrencyTracker struct {
+	mu      sync.Mutex
+	current int
+	peak    int
+}
+
+func (t *syncConcurrencyTracker) enter() {
+	t.mu.Lock()
+	t.current++
+	if t.current > t.peak {
+		t.peak = t.current
+	}
+	t.mu.Unlock()
+}
+
+func (t *syncConcurrencyTracker) leave() {
+	t.mu.Lock()
+	t.current--
+	t.mu.Unlock()
+}
+
+func (t *syncConcurrencyTracker) Peak() int {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	return t.peak
+}
+
+// TestSync_ConcurrencyIsBounded guards internal/sync/tar.go's own fan-out:
+// it used to hardcode eg.SetLimit(16) regardless of --parallel, the same
+// class of gap fixed everywhere else in this repo for bulk engine calls.
+func TestSync_ConcurrencyIsBounded(t *testing.T) {
+	tracker := &syncConcurrencyTracker{}
+	client := &fakeLowLevelClient{
+		containers: []container.Summary{{ID: "ctr1"}, {ID: "ctr2"}, {ID: "ctr3"}},
+		onUntar: func() {
+			tracker.enter()
+			time.Sleep(20 * time.Millisecond) // widen the window for a concurrency violation to show up
+			tracker.leave()
+		},
+	}
+	syncer := NewTar("proj", client, 1)
+
+	tmpDir := t.TempDir()
+	existingFile := filepath.Join(tmpDir, "exists.txt")
+	assert.NilError(t, os.WriteFile(existingFile, []byte("data"), 0o644))
+
+	err := syncer.Sync(t.Context(), "svc", []*PathMapping{
+		{HostPath: existingFile, ContainerPath: "/app/exists.txt"},
+	})
+	assert.NilError(t, err)
+	assert.Equal(t, tracker.Peak(), 1, "Sync must never run more than maxConcurrency Untar calls at once")
+}
+
+// TestSync_ZeroMaxConcurrencyIsUnlimited guards the errgroup.SetLimit(0)
+// footgun: 0 is maxConcurrency's Go zero-value, and SetLimit(0) means "allow
+// zero goroutines", not "unlimited" -- which would deadlock Sync forever.
+func TestSync_ZeroMaxConcurrencyIsUnlimited(t *testing.T) {
+	client := &fakeLowLevelClient{
+		containers: []container.Summary{{ID: "ctr1"}, {ID: "ctr2"}},
+	}
+	syncer := NewTar("proj", client, 0)
+
+	tmpDir := t.TempDir()
+	existingFile := filepath.Join(tmpDir, "exists.txt")
+	assert.NilError(t, os.WriteFile(existingFile, []byte("data"), 0o644))
+
+	done := make(chan error, 1)
+	go func() {
+		done <- syncer.Sync(t.Context(), "svc", []*PathMapping{
+			{HostPath: existingFile, ContainerPath: "/app/exists.txt"},
+		})
+	}()
+	select {
+	case err := <-done:
+		assert.NilError(t, err)
+	case <-time.After(2 * time.Second):
+		t.Fatal("Sync deadlocked with maxConcurrency=0")
+	}
+}
+
 func TestSync_MixedPaths(t *testing.T) {
 	tmpDir := t.TempDir()
 	existingFile := filepath.Join(tmpDir, "keep.txt")
@@ -148,7 +242,7 @@ func TestSync_MixedPaths(t *testing.T) {
 	client := &fakeLowLevelClient{
 		containers: []container.Summary{{ID: "ctr1"}},
 	}
-	syncer := NewTar("proj", client)
+	syncer := NewTar("proj", client, 0)
 
 	err := syncer.Sync(t.Context(), "svc", []*PathMapping{
 		{HostPath: existingFile, ContainerPath: "/app/keep.txt"},
@@ -170,7 +264,7 @@ func TestSync_PopulatedDirectoryKeepsItsHeader(t *testing.T) {
 	client := &fakeLowLevelClient{
 		containers: []container.Summary{{ID: "ctr1"}},
 	}
-	syncer := NewTar("proj", client)
+	syncer := NewTar("proj", client, 0)
 
 	err := syncer.Sync(t.Context(), "svc", []*PathMapping{
 		{HostPath: filepath.Join(tmpDir, "sub"), ContainerPath: "/app/sub"},
@@ -198,7 +292,7 @@ func TestSync_RetriesWithoutImpliedDirectoriesWhenCopyFails(t *testing.T) {
 		containers: []container.Summary{{ID: "ctr1"}},
 		untarErrs:  []error{errors.New(`cannot overwrite non-directory "/app/sub" with directory "/"`)},
 	}
-	syncer := NewTar("proj", client)
+	syncer := NewTar("proj", client, 0)
 
 	err := syncer.Sync(t.Context(), "svc", []*PathMapping{
 		{HostPath: filepath.Join(tmpDir, "sub"), ContainerPath: "/app/sub"},
@@ -224,7 +318,7 @@ func TestSync_ReportsTheCopyErrorWhenTheRetryFailsToo(t *testing.T) {
 			errors.New("no such container"),
 		},
 	}
-	syncer := NewTar("proj", client)
+	syncer := NewTar("proj", client, 0)
 
 	err := syncer.Sync(t.Context(), "svc", []*PathMapping{
 		{HostPath: filepath.Join(tmpDir, "sub"), ContainerPath: "/app/sub"},
@@ -243,7 +337,7 @@ func TestSync_DoesNotRetryACopyThatFailedForAnotherReason(t *testing.T) {
 		containers: []container.Summary{{ID: "ctr1"}},
 		untarErrs:  []error{errors.New("no such container")},
 	}
-	syncer := NewTar("proj", client)
+	syncer := NewTar("proj", client, 0)
 
 	err := syncer.Sync(t.Context(), "svc", []*PathMapping{
 		{HostPath: filepath.Join(tmpDir, "sub"), ContainerPath: "/app/sub"},

pkg/compose/compose.go

@@ -212,22 +212,22 @@ func panicSafeReleaseSlot(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
+// forEachWithLimiter runs fn concurrently for each item, 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, or one per sibling down op), 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 {
+func forEachWithLimiter[T any](ctx context.Context, limiter *semaphore.Weighted, items []T, fn func(context.Context, T) error) error {
 	eg, ctx := errgroup.WithContext(ctx)
-	for _, ctr := range containers {
+	for _, item := range items {
 		eg.Go(func() error {
 			if err := acquireSlot(ctx, limiter); err != nil {
 				return err
 			}
 			defer releaseSlot(limiter)
-			return fn(ctx, ctr)
+			return fn(ctx, item)
 		})
 	}
 	return eg.Wait()

pkg/compose/containers.go

@@ -210,7 +210,7 @@ func (containers Containers) filter(predicates ...containerPredicate) Containers
 }
 
 // forEachContainerConcurrent runs fn for every container concurrently and
-// waits for all goroutines. Use forEachContainerWithLimiter (compose.go)
+// waits for all goroutines. Use forEachWithLimiter (compose.go)
 // instead when the concurrency budget must be shared across several
 // concurrently-dispatched calls, e.g. one per service visited by
 // InDependencyOrder.

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/errgroup"
 	"golang.org/x/sync/semaphore"
 
 	"github.com/docker/compose/v5/pkg/api"
@@ -120,28 +121,35 @@ func (s *composeService) down(ctx context.Context, projectName string, options a
 		return err
 	}
 
-	ops := s.ensureNetworksDown(ctx, project)
+	ops := s.ensureNetworksDown(ctx, project, limiter)
 
 	if options.Images != "" {
-		ops = append(ops, s.ensureImagesDown(ctx, project, options)...)
+		ops = append(ops, s.ensureImagesDown(ctx, project, options, limiter)...)
 	}
 
 	if options.Volumes {
-		ops = append(ops, s.ensureVolumesDown(ctx, project)...)
+		ops = append(ops, s.ensureVolumesDown(ctx, project, limiter)...)
 	}
 
 	if !resourceToRemove && len(ops) == 0 {
 		logrus.Warnf("Warning: No resource found to remove for project %q.", projectName)
 	}
 
-	eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency)
+	// ops dispatch itself is intentionally unbounded: each op already gates
+	// its own engine call(s) on the shared limiter (network/volume ops
+	// directly, image ops internally via their own forEachWithLimiter call),
+	// so bounding this dispatch too on the same limiter would have an op
+	// hold a slot for its own fan-out to acquire from -- deadlocking once
+	// enough ops are in flight to exhaust it (see
+	// TestDown_NetworkAndImageRemovalShareConcurrencyBudget).
+	eg, ctx := errgroup.WithContext(ctx)
 	for _, op := range ops {
 		eg.Go(op)
 	}
 	return eg.Wait()
 }
 
-func (s *composeService) ensureVolumesDown(ctx context.Context, project *types.Project) []downOp {
+func (s *composeService) ensureVolumesDown(ctx context.Context, project *types.Project, limiter *semaphore.Weighted) []downOp {
 	var ops []downOp
 	for _, vol := range project.Volumes {
 		if vol.External {
@@ -149,6 +157,10 @@ func (s *composeService) ensureVolumesDown(ctx context.Context, project *types.P
 		}
 		volumeName := vol.Name
 		ops = append(ops, func() error {
+			if err := acquireSlot(ctx, limiter); err != nil {
+				return err
+			}
+			defer releaseSlot(limiter)
 			return s.removeVolume(ctx, volumeName)
 		})
 	}
@@ -156,7 +168,7 @@ func (s *composeService) ensureVolumesDown(ctx context.Context, project *types.P
 	return ops
 }
 
-func (s *composeService) ensureImagesDown(ctx context.Context, project *types.Project, options api.DownOptions) []downOp {
+func (s *composeService) ensureImagesDown(ctx context.Context, project *types.Project, options api.DownOptions, limiter *semaphore.Weighted) []downOp {
 	pruneOpts := ImagePruneOptions{
 		Mode:          ImagePruneMode(options.Images),
 		RemoveOrphans: options.RemoveOrphans,
@@ -165,7 +177,7 @@ func (s *composeService) ensureImagesDown(ctx context.Context, project *types.Pr
 	var ops []downOp
 	if pruneOpts.Mode != ImagePruneNone {
 		ops = append(ops, func() error {
-			return s.removeTaggedImagesOp(ctx, project, pruneOpts)
+			return s.removeTaggedImagesOp(ctx, project, pruneOpts, limiter)
 		})
 
 		// a dangling image from a service no longer in the project must be
@@ -176,7 +188,7 @@ func (s *composeService) ensureImagesDown(ctx context.Context, project *types.Pr
 		}
 		projectName := project.Name
 		ops = append(ops, func() error {
-			return s.removeDanglingImagesOp(ctx, projectName, keep)
+			return s.removeDanglingImagesOp(ctx, projectName, keep, limiter)
 		})
 	}
 	return ops
@@ -189,8 +201,12 @@ func (s *composeService) ensureImagesDown(ctx context.Context, project *types.Pr
 // sibling down ops (network, volumes, dangling images), since — unlike
 // this op — they've already been scheduled onto the same errgroup by the
 // time `ensureImagesDown` used to fail synchronously.
-func (s *composeService) removeTaggedImagesOp(ctx context.Context, project *types.Project, pruneOpts ImagePruneOptions) error {
-	images, err := NewImagePruner(s.apiClient(), project, s.maxConcurrency).ImagesToPrune(ctx, pruneOpts)
+//
+// limiter is shared with the rest of down's ops (see down()'s comment) so
+// this op's fan-out doesn't spend its own full maxConcurrency budget on top
+// of removeDanglingImagesOp's, which runs concurrently with it.
+func (s *composeService) removeTaggedImagesOp(ctx context.Context, project *types.Project, pruneOpts ImagePruneOptions, limiter *semaphore.Weighted) error {
+	images, err := NewImagePruner(s.apiClient(), project, limiter).ImagesToPrune(ctx, pruneOpts)
 	if err != nil {
 		s.events.On(errorEvent("Tagged images", err.Error()))
 		return err
@@ -198,22 +214,21 @@ func (s *composeService) removeTaggedImagesOp(ctx context.Context, project *type
 
 	var mu sync.Mutex
 	var errs []error
-	eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency)
-	for i := range images {
-		img := images[i]
-		eg.Go(func() error {
-			if err := s.removeResource("Image "+img, func() error {
-				_, err := s.apiClient().ImageRemove(ctx, img, client.ImageRemoveOptions{})
-				return err
-			}); err != nil {
-				mu.Lock()
-				errs = append(errs, err)
-				mu.Unlock()
-			}
-			return nil
-		})
-	}
-	_ = eg.Wait() // errgroup is only used for fan-out here; goroutines never return an error
+	// forEachWithLimiter's own error handling isn't used here since a single
+	// image failure must not cancel the shared errgroup ctx (see
+	// TestRemoveTaggedImagesOp_ContinuesAfterOneImageFails) -- every failure
+	// is collected instead, so fn itself always returns nil.
+	_ = forEachWithLimiter(ctx, limiter, images, func(ctx context.Context, img string) error {
+		if err := s.removeResource("Image "+img, func() error {
+			_, err := s.apiClient().ImageRemove(ctx, img, client.ImageRemoveOptions{})
+			return err
+		}); err != nil {
+			mu.Lock()
+			errs = append(errs, err)
+			mu.Unlock()
+		}
+		return nil
+	})
 	return errors.Join(errs...)
 }
 
@@ -237,7 +252,7 @@ func (s *composeService) removeTaggedImagesOp(ctx context.Context, project *type
 // removed either: stillInUse (a real, not-yet-achieved goal) is checked
 // separately, the same distinction removeResource already makes per tagged
 // image (an already-gone image simply isn't counted in either bucket).
-func (s *composeService) removeDanglingImagesOp(ctx context.Context, projectName string, keep func(image.Summary) bool) error {
+func (s *composeService) removeDanglingImagesOp(ctx context.Context, projectName string, keep func(image.Summary) bool, limiter *semaphore.Weighted) error {
 	eventID := "Dangling images"
 	eligible, err := s.eligibleDanglingImages(ctx, projectName, keep)
 	if err != nil {
@@ -249,7 +264,7 @@ func (s *composeService) removeDanglingImagesOp(ctx context.Context, projectName
 	}
 
 	s.events.On(removingEvent(eventID))
-	removed, stillInUse, err := s.removeImages(ctx, eligible)
+	removed, stillInUse, err := s.removeImages(ctx, eligible, limiter)
 	var stillInUseNote string
 	if len(stillInUse) > 0 {
 		stillInUseNote = fmt.Sprintf("%d image(s) still in use", len(stillInUse))
@@ -281,7 +296,7 @@ func (s *composeService) removeDanglingImagesOp(ctx context.Context, projectName
 	return err
 }
 
-func (s *composeService) ensureNetworksDown(ctx context.Context, project *types.Project) []downOp {
+func (s *composeService) ensureNetworksDown(ctx context.Context, project *types.Project, limiter *semaphore.Weighted) []downOp {
 	var ops []downOp
 	for key, n := range project.Networks {
 		if n.External {
@@ -291,6 +306,10 @@ func (s *composeService) ensureNetworksDown(ctx context.Context, project *types.
 		networkKey := key
 		idOrName := n.Name
 		ops = append(ops, func() error {
+			if err := acquireSlot(ctx, limiter); err != nil {
+				return err
+			}
+			defer releaseSlot(limiter)
 			return s.removeNetwork(ctx, networkKey, project.Name, idOrName)
 		})
 	}
@@ -437,7 +456,7 @@ func (s *composeService) stopContainers(
 	listener api.ContainerEventListener,
 	limiter *semaphore.Weighted,
 ) error {
-	return forEachContainerWithLimiter(ctx, limiter, containers, func(ctx context.Context, ctr containerType.Summary) error {
+	return forEachWithLimiter(ctx, limiter, containers, func(ctx context.Context, ctr containerType.Summary) error {
 		return s.stopContainer(ctx, serv, ctr, timeout, listener)
 	})
 }
@@ -450,7 +469,7 @@ func (s *composeService) removeContainers(
 	volumes bool,
 	limiter *semaphore.Weighted,
 ) error {
-	return forEachContainerWithLimiter(ctx, limiter, containers, func(ctx context.Context, ctr containerType.Summary) error {
+	return forEachWithLimiter(ctx, limiter, containers, func(ctx context.Context, ctr containerType.Summary) error {
 		return s.stopAndRemoveContainer(ctx, ctr, service, timeout, volumes)
 	})
 }

pkg/compose/down_test.go

@@ -21,6 +21,7 @@ import (
 	"errors"
 	"fmt"
 	"os"
+	"sort"
 	"strings"
 	"testing"
 	"time"
@@ -135,6 +136,113 @@ func TestDown_ConcurrencyIsBoundedAcrossServices(t *testing.T) {
 	assert.Equal(t, tracker.Peak(), 1, "down must never run more than maxConcurrency ContainerRemove calls at once")
 }
 
+// TestDown_ImagePruningSharesConcurrencyBudgetAcrossOps guards against a
+// regression flagged in review: removeTaggedImagesOp and
+// removeDanglingImagesOp each opened their own bounded errgroup sized to the
+// full maxConcurrency budget, but run concurrently as two independent down
+// ops -- so combined image-removal calls could reach 2x maxConcurrency
+// instead of sharing one budget, the same class of leak
+// TestDown_ConcurrencyIsBoundedAcrossServices guards for container removal.
+func TestDown_ImagePruningSharesConcurrencyBudgetAcrossOps(t *testing.T) {
+	svc, apiClient := newTestService(t, WithMaxConcurrency(2))
+
+	project := &types.Project{Name: "prj"}
+
+	apiClient.EXPECT().ContainerList(gomock.Any(), gomock.Any()).
+		Return(client.ContainerListResult{}, nil)
+	apiClient.EXPECT().ContainerList(gomock.Any(), gomock.Any()).
+		Return(client.ContainerListResult{}, nil) // removePreStartHookContainers lookup
+
+	apiClient.EXPECT().ImageList(gomock.Any(), client.ImageListOptions{
+		Filters: projectFilter("prj").Add("dangling", "false"),
+	}).Return(client.ImageListResult{Items: []image.Summary{
+		{ID: "sha256:tag1", RepoTags: []string{"repo/tag1:latest"}},
+		{ID: "sha256:tag2", RepoTags: []string{"repo/tag2:latest"}},
+	}}, nil)
+	apiClient.EXPECT().ImageList(gomock.Any(), client.ImageListOptions{
+		Filters: projectFilter("prj").Add("dangling", "true"),
+	}).Return(client.ImageListResult{Items: []image.Summary{
+		{ID: "sha256:dangling1"},
+		{ID: "sha256:dangling2"},
+	}}, nil)
+
+	tracker := &peakConcurrencyTracker{}
+	apiClient.EXPECT().ImageRemove(gomock.Any(), gomock.Any(), gomock.Any()).
+		DoAndReturn(func(context.Context, string, client.ImageRemoveOptions) (client.ImageRemoveResult, error) {
+			tracker.enter()
+			time.Sleep(20 * time.Millisecond) // widen the window for a concurrency violation to show up
+			tracker.leave()
+			return client.ImageRemoveResult{}, nil
+		}).
+		Times(4)
+
+	err := svc.down(t.Context(), "prj", compose.DownOptions{Project: project, Images: "local", RemoveOrphans: true})
+	assert.NilError(t, err)
+	assert.Assert(t, tracker.Peak() <= 2, "tagged- and dangling-image removal must share the same budget, got peak %d", tracker.Peak())
+}
+
+// TestDown_NetworkAndImageRemovalShareConcurrencyBudget guards against a
+// regression flagged in review: the outer ops dispatch bounded only the
+// *count* of concurrently-running down ops, independent of the shared
+// limiter each op's own engine calls are gated on -- so a single ordinary
+// project network plus --rmi could spend up to 2x maxConcurrency, the same
+// class of leak TestDown_ImagePruningSharesConcurrencyBudgetAcrossOps guards
+// between the two image ops.
+func TestDown_NetworkAndImageRemovalShareConcurrencyBudget(t *testing.T) {
+	svc, apiClient := newTestService(t, WithMaxConcurrency(2))
+
+	project := &types.Project{
+		Name: "prj",
+		Networks: types.Networks{
+			"default": {Name: "prj_default"},
+		},
+	}
+
+	apiClient.EXPECT().ContainerList(gomock.Any(), gomock.Any()).
+		Return(client.ContainerListResult{}, nil)
+	apiClient.EXPECT().ContainerList(gomock.Any(), gomock.Any()).
+		Return(client.ContainerListResult{}, nil) // removePreStartHookContainers lookup
+
+	apiClient.EXPECT().ImageList(gomock.Any(), client.ImageListOptions{
+		Filters: projectFilter("prj").Add("dangling", "false"),
+	}).Return(client.ImageListResult{Items: []image.Summary{
+		{ID: "sha256:tag1", RepoTags: []string{"repo/tag1:latest"}},
+		{ID: "sha256:tag2", RepoTags: []string{"repo/tag2:latest"}},
+	}}, nil)
+	apiClient.EXPECT().ImageList(gomock.Any(), client.ImageListOptions{
+		Filters: projectFilter("prj").Add("dangling", "true"),
+	}).Return(client.ImageListResult{}, nil)
+
+	tracker := &peakConcurrencyTracker{}
+	apiClient.EXPECT().ImageRemove(gomock.Any(), gomock.Any(), gomock.Any()).
+		DoAndReturn(func(context.Context, string, client.ImageRemoveOptions) (client.ImageRemoveResult, error) {
+			tracker.enter()
+			time.Sleep(20 * time.Millisecond) // widen the window for a concurrency violation to show up
+			tracker.leave()
+			return client.ImageRemoveResult{}, nil
+		}).
+		Times(2)
+
+	apiClient.EXPECT().NetworkList(gomock.Any(), client.NetworkListOptions{
+		Filters: projectFilter("prj").Add("label", networkFilter("default")),
+	}).Return(client.NetworkListResult{Items: []network.Summary{
+		{Network: network.Network{ID: "net1", Name: "prj_default"}},
+	}}, nil)
+	apiClient.EXPECT().NetworkInspect(gomock.Any(), "net1", gomock.Any()).
+		DoAndReturn(func(context.Context, string, client.NetworkInspectOptions) (client.NetworkInspectResult, error) {
+			tracker.enter()
+			time.Sleep(20 * time.Millisecond)
+			tracker.leave()
+			return client.NetworkInspectResult{Network: network.Inspect{Network: network.Network{ID: "net1"}}}, nil
+		})
+	apiClient.EXPECT().NetworkRemove(gomock.Any(), "net1", gomock.Any()).
+		Return(client.NetworkRemoveResult{}, nil)
+
+	err := svc.down(t.Context(), "prj", compose.DownOptions{Project: project, Images: "local", RemoveOrphans: true})
+	assert.NilError(t, err)
+	assert.Assert(t, tracker.Peak() <= 2, "network- and image-removal ops must share the same concurrency budget, got peak %d", tracker.Peak())
+}
+
 func TestDownWithGivenServices(t *testing.T) {
 	mockCtrl := gomock.NewController(t)
 	defer mockCtrl.Finish()
@@ -534,7 +642,7 @@ func TestEnsureImagesDown_TaggedImageListingFailureStaysVisibleWithoutAbortingDo
 		Filters: projectFilter("prj").Add("dangling", "true"),
 	}).Return(client.ImageListResult{}, nil)
 
-	ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true})
+	ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}, nil)
 	assert.Equal(t, len(ops), 2) // tagged-images op + dangling-images op
 
 	opErr := ops[0]()
@@ -595,19 +703,25 @@ func TestRemoveTaggedImagesOp_ContinuesAfterOneImageFails(t *testing.T) {
 	apiClient.EXPECT().ImageRemove(liveCtx{}, "repo/c:latest", client.ImageRemoveOptions{}).
 		Return(client.ImageRemoveResult{}, nil)
 
-	opErr := svc.removeTaggedImagesOp(t.Context(), project, ImagePruneOptions{Mode: ImagePruneLocal, RemoveOrphans: true})
+	opErr := svc.removeTaggedImagesOp(t.Context(), project, ImagePruneOptions{Mode: ImagePruneLocal, RemoveOrphans: true}, newOptionalLimiter(1))
 	assert.ErrorContains(t, opErr, daemonErr.Error())
 
+	// dispatch order across images is no longer deterministic now that the
+	// concurrency bound comes from a shared limiter acquired inside each
+	// goroutine (see removeImages) rather than errgroup.SetLimit blocking the
+	// submitting loop itself -- only the per-image Removing->Removed sequence
+	// (enforced by removeResource) still holds, so compare as a set.
 	events := make([]string, len(rec.resources))
 	for i, e := range rec.resources {
 		events[i] = e.ID + ": " + e.Text
 	}
+	sort.Strings(events)
 	assert.DeepEqual(t, events, []string{
 		"Image repo/a:latest: Removing",
-		"Image repo/b:latest: Removing",
 		"Image repo/b:latest: Removed",
-		"Image repo/c:latest: Removing",
+		"Image repo/b:latest: Removing",
 		"Image repo/c:latest: Removed",
+		"Image repo/c:latest: Removing",
 	})
 }
 
@@ -641,7 +755,7 @@ func TestEnsureImagesDown_ReportsDanglingImagesAsOneGroupedEvent(t *testing.T) {
 
 	// RemoveOrphans:true bypasses the per-service keep check so this test
 	// stays focused on the single-grouped-event behavior.
-	ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true})
+	ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}, nil)
 	for _, op := range ops {
 		assert.NilError(t, op())
 	}
@@ -690,7 +804,7 @@ func TestEnsureImagesDown_SparesDanglingImagesOfOrphanedServices(t *testing.T) {
 	apiClient.EXPECT().ImageRemove(gomock.Any(), "sha256:web-dangling", client.ImageRemoveOptions{}).
 		Return(client.ImageRemoveResult{}, nil)
 
-	ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local"})
+	ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local"}, nil)
 	for _, op := range ops {
 		assert.NilError(t, op())
 	}
@@ -725,7 +839,7 @@ func TestEnsureImagesDown_RemoveOrphansAlsoTakesDanglingImages(t *testing.T) {
 	apiClient.EXPECT().ImageRemove(gomock.Any(), "sha256:orphan-dangling", client.ImageRemoveOptions{}).
 		Return(client.ImageRemoveResult{}, nil)
 
-	ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true})
+	ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}, nil)
 	for _, op := range ops {
 		assert.NilError(t, op())
 	}
@@ -762,7 +876,7 @@ func TestEnsureImagesDown_NoDanglingImagesToRemove(t *testing.T) {
 		Filters: projectFilter("prj").Add("dangling", "true"),
 	}).Return(client.ImageListResult{}, nil)
 
-	ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true})
+	ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}, nil)
 	assert.Equal(t, len(ops), 2) // tagged-images op + dangling-images op
 	for _, op := range ops {
 		assert.NilError(t, op())
@@ -785,7 +899,7 @@ func TestEnsureImagesDown_AllDanglingImagesSparedStaysSilent(t *testing.T) {
 	}}, nil)
 	// no ImageRemove expectation — a call for the spared image fails the test
 
-	ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local"})
+	ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local"}, nil)
 	assert.Equal(t, len(ops), 2) // tagged-images op + dangling-images op
 	for _, op := range ops {
 		assert.NilError(t, op())
@@ -812,7 +926,7 @@ func TestEnsureImagesDown_DanglingListFailureStaysVisibleWithoutAbortingDown(t *
 	// ensureImagesDown itself must not fail: the listing error is only
 	// surfaced when the returned op actually runs, same as every other
 	// down op, so it can't block ops collected/scheduled around it.
-	ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true})
+	ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}, nil)
 	assert.Equal(t, len(ops), 2) // tagged-images op + dangling-images op
 
 	assert.NilError(t, ops[0]()) // tagged-images op: nothing to prune, trivially succeeds
@@ -848,7 +962,7 @@ func TestEnsureImagesDown_PartialRemovalFailureStaysVisibleAlongsideRemoved(t *t
 	apiClient.EXPECT().ImageRemove(gomock.Any(), "sha256:fails", client.ImageRemoveOptions{}).
 		Return(client.ImageRemoveResult{}, errdefs.ErrPermissionDenied.WithMessage("permission denied"))
 
-	ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true})
+	ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}, nil)
 	assert.Equal(t, len(ops), 2) // tagged-images op + dangling-images op
 
 	assert.NilError(t, ops[0]()) // tagged-images op: nothing to prune, trivially succeeds
@@ -879,7 +993,7 @@ func TestEnsureImagesDown_TotalRemovalFailureReportsPlainError(t *testing.T) {
 	apiClient.EXPECT().ImageRemove(gomock.Any(), "sha256:fails", client.ImageRemoveOptions{}).
 		Return(client.ImageRemoveResult{}, errdefs.ErrPermissionDenied.WithMessage("permission denied"))
 
-	ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true})
+	ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}, nil)
 	assert.Equal(t, len(ops), 2) // tagged-images op + dangling-images op
 
 	assert.NilError(t, ops[0]()) // tagged-images op: nothing to prune, trivially succeeds
@@ -915,7 +1029,7 @@ func TestEnsureImagesDown_TotalFailureAlsoMentionsStillInUse(t *testing.T) {
 	apiClient.EXPECT().ImageRemove(gomock.Any(), "sha256:fails", client.ImageRemoveOptions{}).
 		Return(client.ImageRemoveResult{}, errdefs.ErrPermissionDenied.WithMessage("permission denied"))
 
-	ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true})
+	ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}, nil)
 	assert.Equal(t, len(ops), 2) // tagged-images op + dangling-images op
 
 	assert.NilError(t, ops[0]()) // tagged-images op: nothing to prune, trivially succeeds
@@ -949,7 +1063,7 @@ func TestEnsureImagesDown_AllEligibleStillInUseDoesNotClaimRemoved(t *testing.T)
 	apiClient.EXPECT().ImageRemove(gomock.Any(), "sha256:in-use", client.ImageRemoveOptions{}).
 		Return(client.ImageRemoveResult{}, errdefs.ErrConflict.WithMessage("image is being used by a container"))
 
-	ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true})
+	ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}, nil)
 	assert.Equal(t, len(ops), 2) // tagged-images op + dangling-images op
 
 	assert.NilError(t, ops[0]()) // tagged-images op: nothing to prune, trivially succeeds
@@ -983,7 +1097,7 @@ func TestEnsureImagesDown_MixedRemovedAndStillInUseKeepsBothVisible(t *testing.T
 	apiClient.EXPECT().ImageRemove(gomock.Any(), "sha256:in-use", client.ImageRemoveOptions{}).
 		Return(client.ImageRemoveResult{}, errdefs.ErrConflict.WithMessage("image is being used by a container"))
 
-	ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true})
+	ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}, nil)
 	assert.Equal(t, len(ops), 2) // tagged-images op + dangling-images op
 
 	assert.NilError(t, ops[0]()) // tagged-images op: nothing to prune, trivially succeeds
@@ -1020,7 +1134,7 @@ func TestEnsureImagesDown_PartialFailureAlsoMentionsStillInUse(t *testing.T) {
 	apiClient.EXPECT().ImageRemove(gomock.Any(), "sha256:fails", client.ImageRemoveOptions{}).
 		Return(client.ImageRemoveResult{}, errdefs.ErrPermissionDenied.WithMessage("permission denied"))
 
-	ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true})
+	ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}, nil)
 	assert.Equal(t, len(ops), 2) // tagged-images op + dangling-images op
 
 	assert.NilError(t, ops[0]()) // tagged-images op: nothing to prune, trivially succeeds
@@ -1052,7 +1166,7 @@ func TestEnsureImagesDown_AllEligibleAlreadyGoneReportsNoResourceFound(t *testin
 	apiClient.EXPECT().ImageRemove(gomock.Any(), "sha256:already-gone", client.ImageRemoveOptions{}).
 		Return(client.ImageRemoveResult{}, errdefs.ErrNotFound.WithMessage("already removed"))
 
-	ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true})
+	ops := svc.ensureImagesDown(t.Context(), project, compose.DownOptions{Images: "local", RemoveOrphans: true}, nil)
 	assert.Equal(t, len(ops), 2) // tagged-images op + dangling-images op
 
 	assert.NilError(t, ops[0]()) // tagged-images op: nothing to prune, trivially succeeds

pkg/compose/executor.go

@@ -22,6 +22,7 @@ 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
@@ -101,13 +102,14 @@ func (exec *planExecutor) run(ctx context.Context, plan *Plan) error {
 	groups := exec.buildGroupTracker(plan)
 	events := exec.compose.events
 
-	// 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)
+	// Every node's goroutine is dispatched unconditionally, so one waiting on
+	// a dependency never occupies a concurrency slot -- only the actual
+	// executeNode call does, via the semaphore below. This also means
+	// deadlock-freedom no longer depends on plan.Nodes staying topologically
+	// sorted: a goroutine blocked on <-done[dep.ID] holds no slot for a
+	// still-pending dependency to be starved on.
+	eg, ctx := errgroup.WithContext(ctx)
+	limiter := newOptionalLimiter(exec.compose.maxConcurrency)
 	for _, node := range plan.Nodes {
 		eg.Go(func() error {
 			// Wait for all dependencies
@@ -119,6 +121,11 @@ func (exec *planExecutor) run(ctx context.Context, plan *Plan) error {
 				}
 			}
 
+			if err := acquireSlot(ctx, limiter); err != nil {
+				return err
+			}
+			defer releaseSlot(limiter)
+
 			// Emit group start event if this is the first node of a group
 			groups.onNodeStart(node, events)
 

pkg/compose/executor_test.go

@@ -370,17 +370,14 @@ 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.
+// TestExecutePlanRespectsMaxConcurrencyAcrossDependencyChain guards that a
+// multi-level dependency chain still executes serially under
+// maxConcurrency=1 and completes rather than deadlocking. run() dispatches
+// every node's goroutine unconditionally and only acquires a concurrency
+// slot around the actual executeNode call (see run()'s comment), so this no
+// longer depends on plan.Nodes staying topologically sorted -- a goroutine
+// blocked on <-done[dep.ID] holds no slot for its dependency to starve on,
+// however plan.Nodes is ordered.
 func TestExecutePlanRespectsMaxConcurrencyAcrossDependencyChain(t *testing.T) {
 	svc, apiClient := newTestService(t, WithMaxConcurrency(1))
 
@@ -424,7 +421,91 @@ func TestExecutePlanRespectsMaxConcurrencyAcrossDependencyChain(t *testing.T) {
 	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")
+		t.Fatal("run() deadlocked on a serial dependency chain under maxConcurrency=1")
+	}
+}
+
+// TestExecutePlanIndependentNodeNotSerializedBehindADependencyWait guards
+// that a node with no unmet dependencies can run concurrently with an
+// unrelated node that's still waiting on its own dependency, instead of
+// being starved behind it. Before run() acquired its concurrency slot only
+// around executeNode, a waiting node's goroutine held its slot for the whole
+// wait, so on a wide/shallow DAG an independent, ready node could be blocked
+// from even dispatching until the waiter's chain finished -- the two
+// ContainerStop calls below could never overlap under the old behavior,
+// since the independent node's slot only freed once the dependency chain's
+// own goroutine (holding the other slot for its entire wait) had returned.
+func TestExecutePlanIndependentNodeNotSerializedBehindADependencyWait(t *testing.T) {
+	svc, apiClient := newTestService(t, WithMaxConcurrency(2))
+
+	slow := container.Summary{ID: "c-slow", Names: []string{"/test-web-1"}, Labels: map[string]string{
+		api.ServiceLabel: "web", api.ContainerNumberLabel: "1",
+	}}
+	waiter := container.Summary{ID: "c-waiter", Names: []string{"/test-web-2"}, Labels: map[string]string{
+		api.ServiceLabel: "web", api.ContainerNumberLabel: "2",
+	}}
+	independent := container.Summary{ID: "c-independent", Names: []string{"/test-app-1"}, Labels: map[string]string{
+		api.ServiceLabel: "app", api.ContainerNumberLabel: "1",
+	}}
+
+	// slow blocks mid-call until the test releases it, so the assertion below
+	// is a deterministic happens-before check, not a timing race.
+	slowStarted := make(chan struct{})
+	slowRelease := make(chan struct{})
+	apiClient.EXPECT().ContainerStop(gomock.Any(), slow.ID, gomock.Any()).
+		DoAndReturn(func(context.Context, string, client.ContainerStopOptions) (client.ContainerStopResult, error) {
+			close(slowStarted)
+			<-slowRelease
+			return client.ContainerStopResult{}, nil
+		})
+	apiClient.EXPECT().ContainerStop(gomock.Any(), waiter.ID, gomock.Any()).
+		Return(client.ContainerStopResult{}, nil)
+
+	independentStarted := make(chan struct{})
+	apiClient.EXPECT().ContainerStop(gomock.Any(), independent.ID, gomock.Any()).
+		DoAndReturn(func(context.Context, string, client.ContainerStopOptions) (client.ContainerStopResult, error) {
+			close(independentStarted)
+			return client.ContainerStopResult{}, nil
+		})
+
+	// slow -> waiter (waiter depends on slow); independent has no deps and
+	// shares no resource with either.
+	plan := &Plan{}
+	slowNode := plan.addNode(Operation{
+		Type: OpStopContainer, ResourceID: "service:web:1", Cause: "chain", Container: &slow,
+	}, "")
+	plan.addNode(Operation{
+		Type: OpStopContainer, ResourceID: "service:web:2", Cause: "chain", Container: &waiter,
+	}, "", slowNode)
+	plan.addNode(Operation{
+		Type: OpStopContainer, ResourceID: "service:app:1", Cause: "unrelated", Container: &independent,
+	}, "")
+
+	exec := svc.newPlanExecutor(&types.Project{Name: "test"}, emptyObservedState("test"))
+
+	done := make(chan error, 1)
+	go func() { done <- exec.run(t.Context(), plan) }()
+
+	select {
+	case <-slowStarted:
+	case <-time.After(2 * time.Second):
+		t.Fatal("slow node never started")
+	}
+
+	select {
+	case <-independentStarted:
+		// the independent node ran while slow is still blocked -- it wasn't
+		// serialized behind the dependency chain's wait.
+	case <-time.After(2 * time.Second):
+		t.Fatal("independent node never started while the slow node (holding the other slot) was still running -- it was serialized behind the dependency chain")
+	}
+	close(slowRelease)
+
+	select {
+	case err := <-done:
+		assert.NilError(t, err)
+	case <-time.After(2 * time.Second):
+		t.Fatal("run() did not complete after releasing the slow node")
 	}
 }
 

pkg/compose/image_pruner.go

@@ -29,6 +29,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"
 )
@@ -59,17 +60,22 @@ type ImagePruneOptions struct {
 
 // ImagePruner handles image removal during Compose `down` operations.
 type ImagePruner struct {
-	client         client.ImageAPIClient
-	project        *types.Project
-	maxConcurrency int
+	client  client.ImageAPIClient
+	project *types.Project
+	// limiter bounds concurrent engine calls, shared with the rest of the
+	// down op set so tagged- and dangling-image removal (which run as two
+	// independent, concurrently-dispatched ops) don't each spend their own
+	// full budget on top of one another.
+	limiter *semaphore.Weighted
 }
 
-// NewImagePruner creates an ImagePruner object for a project.
-func NewImagePruner(imageClient client.ImageAPIClient, project *types.Project, maxConcurrency int) *ImagePruner {
+// NewImagePruner creates an ImagePruner object for a project. limiter may be
+// nil, meaning unbounded.
+func NewImagePruner(imageClient client.ImageAPIClient, project *types.Project, limiter *semaphore.Weighted) *ImagePruner {
 	return &ImagePruner{
-		client:         imageClient,
-		project:        project,
-		maxConcurrency: maxConcurrency,
+		client:  imageClient,
+		project: project,
+		limiter: limiter,
 	}
 }
 
@@ -175,37 +181,36 @@ func (p *ImagePruner) labeledLocalImages(ctx context.Context) ([]image.Summary,
 // removal, so it's simply not counted against either bucket. Any other
 // removal error is joined into err so callers reporting failures to the
 // user keep the actual daemon error instead of just an image ID.
-func (s *composeService) removeImages(ctx context.Context, images []image.Summary) (removed, stillInUse []string, err error) {
+func (s *composeService) removeImages(ctx context.Context, images []image.Summary, limiter *semaphore.Weighted) (removed, stillInUse []string, err error) {
 	var mu sync.Mutex
 	var errs []error
-	eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency)
-	for _, img := range images {
-		eg.Go(func() error {
-			if _, err := s.apiClient().ImageRemove(ctx, img.ID, client.ImageRemoveOptions{}); err != nil {
-				mu.Lock()
-				defer mu.Unlock()
-				imgStillInUse, alreadyGone := classifyRemovalError(err)
-				switch {
-				case alreadyGone:
-					// e.g. removed concurrently by something else
-					logrus.Debugf("dangling image %s already removed: %v", img.ID, err)
-				case imgStillInUse:
-					// the same benign skip the tagged-image path reports via
-					// removeResource's conflict branch
-					logrus.Debugf("dangling image %s still in use: %v", img.ID, err)
-					stillInUse = append(stillInUse, img.ID)
-				default:
-					errs = append(errs, fmt.Errorf("image %s: %w", img.ID, err))
-				}
-				return nil
-			}
+	// as in removeTaggedImagesOp, a single image failure must not cancel the
+	// shared errgroup ctx, so fn always returns nil and collects into the
+	// buckets below instead.
+	_ = forEachWithLimiter(ctx, limiter, images, func(ctx context.Context, img image.Summary) error {
+		if _, err := s.apiClient().ImageRemove(ctx, img.ID, client.ImageRemoveOptions{}); err != nil {
 			mu.Lock()
 			defer mu.Unlock()
-			removed = append(removed, img.ID)
+			imgStillInUse, alreadyGone := classifyRemovalError(err)
+			switch {
+			case alreadyGone:
+				// e.g. removed concurrently by something else
+				logrus.Debugf("dangling image %s already removed: %v", img.ID, err)
+			case imgStillInUse:
+				// the same benign skip the tagged-image path reports via
+				// removeResource's conflict branch
+				logrus.Debugf("dangling image %s still in use: %v", img.ID, err)
+				stillInUse = append(stillInUse, img.ID)
+			default:
+				errs = append(errs, fmt.Errorf("image %s: %w", img.ID, err))
+			}
 			return nil
-		})
-	}
-	_ = eg.Wait() // errgroup is only used for fan-out here; goroutines never return an error
+		}
+		mu.Lock()
+		defer mu.Unlock()
+		removed = append(removed, img.ID)
+		return nil
+	})
 	return removed, stillInUse, errors.Join(errs...)
 }
 
@@ -238,7 +243,7 @@ func (s *composeService) removeDanglingImages(ctx context.Context, projectName s
 	if err != nil {
 		return nil, err
 	}
-	removed, _, err := s.removeImages(ctx, eligible)
+	removed, _, err := s.removeImages(ctx, eligible, newOptionalLimiter(s.maxConcurrency))
 	return removed, err
 }
 
@@ -273,23 +278,19 @@ func (p *ImagePruner) filterImagesByExistence(ctx context.Context, imageNames []
 	var mu sync.Mutex
 	var ret []string
 
-	eg, ctx := newLimitedErrgroup(ctx, p.maxConcurrency)
-	for _, img := range imageNames {
-		eg.Go(func() error {
-			_, err := p.client.ImageInspect(ctx, img)
-			if errdefs.IsNotFound(err) {
-				// err on the side of caution: only skip if we successfully
-				// queried the API and got back a definitive "not exists"
-				return nil
-			}
-			mu.Lock()
-			defer mu.Unlock()
-			ret = append(ret, img)
+	err := forEachWithLimiter(ctx, p.limiter, imageNames, func(ctx context.Context, img string) error {
+		_, err := p.client.ImageInspect(ctx, img)
+		if errdefs.IsNotFound(err) {
+			// err on the side of caution: only skip if we successfully
+			// queried the API and got back a definitive "not exists"
 			return nil
-		})
-	}
-
-	if err := eg.Wait(); err != nil {
+		}
+		mu.Lock()
+		defer mu.Unlock()
+		ret = append(ret, img)
+		return nil
+	})
+	if err != nil {
 		return nil, err
 	}
 

pkg/compose/image_pruner_test.go

@@ -101,7 +101,7 @@ func TestRemoveImages_BucketsStillInUseSeparatelyFromRemoved(t *testing.T) {
 	apiClient.EXPECT().ImageRemove(gomock.Any(), "sha256:fails", client.ImageRemoveOptions{}).
 		Return(client.ImageRemoveResult{}, errdefs.ErrPermissionDenied.WithMessage("permission denied"))
 
-	removed, stillInUse, err := svc.removeImages(t.Context(), images)
+	removed, stillInUse, err := svc.removeImages(t.Context(), images, nil)
 	assert.DeepEqual(t, removed, []string{"sha256:removed"})
 	assert.DeepEqual(t, stillInUse, []string{"sha256:in-use"})
 	assert.ErrorContains(t, err, "sha256:fails")

pkg/compose/restart.go

@@ -60,7 +60,7 @@ func (s *composeService) restart(ctx context.Context, projectName string, option
 			return err
 		}
 
-		return forEachContainerWithLimiter(ctx, limiter, containers.filter(isService(service)), func(ctx context.Context, ctr container.Summary) error {
+		return forEachWithLimiter(ctx, limiter, containers.filter(isService(service)), func(ctx context.Context, ctr container.Summary) error {
 			return s.restartContainer(ctx, project.Services[service], ctr, options)
 		})
 	})

pkg/compose/up.go

@@ -32,9 +32,9 @@ import (
 	"github.com/containerd/errdefs"
 	"github.com/docker/cli/cli"
 	"github.com/eiannone/keyboard"
-	"github.com/moby/moby/client"
 	"github.com/sirupsen/logrus"
 	"golang.org/x/sync/errgroup"
+	"golang.org/x/sync/semaphore"
 
 	"github.com/docker/compose/v5/cmd/formatter"
 	"github.com/docker/compose/v5/internal/desktop"
@@ -82,6 +82,10 @@ type upSession struct {
 	menu       *formatter.LogKeyboard
 	globalCtx  context.Context
 	cancel     context.CancelFunc
+	// logOpenLimiter bounds concurrent log-attach opens from
+	// followStartedContainers, the same footgun --parallel closes for
+	// `compose logs --follow` (see doLogContainer).
+	logOpenLimiter *semaphore.Weighted
 
 	signalChan   chan os.Signal
 	isTerminated atomic.Bool
@@ -136,6 +140,7 @@ func (s *composeService) runInteractiveUp(ctx context.Context, project *types.Pr
 		menu:           navigationMenu,
 		globalCtx:      globalCtx,
 		cancel:         cancel,
+		logOpenLimiter: newOptionalLimiter(s.maxConcurrency),
 		signalChan:     signalChan,
 	}
 
@@ -453,15 +458,15 @@ func (u *upSession) followStartedContainers(attached []string) api.ContainerEven
 }
 
 func (u *upSession) streamContainerLogs(event api.ContainerEvent, since string) error {
-	res, err := u.apiClient().ContainerInspect(u.globalCtx, event.ID, client.ContainerInspectOptions{})
+	ctr, err := u.inspectWithSlot(u.globalCtx, u.logOpenLimiter, event.ID)
 	if err != nil {
 		return err
 	}
 	if since == "" {
-		since = logsSinceLastRun(res.Container)
+		since = logsSinceLastRun(ctr)
 	}
 
-	err = u.doLogContainer(u.globalCtx, nil, u.options.Start.Attach, event.Source, res.Container, api.LogOptions{
+	err = u.doLogContainer(u.globalCtx, u.logOpenLimiter, u.options.Start.Attach, event.Source, ctr, api.LogOptions{
 		Follow: true,
 		Since:  since,
 	})

pkg/compose/up_test.go

@@ -20,8 +20,15 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
+	"strings"
 	"testing"
+	"time"
 
+	containerType "github.com/moby/moby/api/types/container"
+	"github.com/moby/moby/client"
+	"go.uber.org/mock/gomock"
+	"golang.org/x/sync/errgroup"
 	"gotest.tools/v3/assert"
 
 	"github.com/docker/compose/v5/pkg/api"
@@ -179,3 +186,45 @@ func TestAppendErrKeepsCancellationBeforeShutdown(t *testing.T) {
 	u.appendErr(context.Canceled)
 	assert.Equal(t, len(u.errs), 1)
 }
+
+// TestStreamContainerLogs_ConcurrencyIsBounded guards a regression flagged
+// in review: streamContainerLogs (attaches logs for containers that start
+// after up's initial attach, e.g. via depends_on/restart/watch rebuilds)
+// used to call ContainerInspect directly and pass nil to doLogContainer, so
+// unlike `compose logs --follow` it never shared --parallel's budget
+// bounding concurrent log-open calls.
+func TestStreamContainerLogs_ConcurrencyIsBounded(t *testing.T) {
+	svc, apiClient := newTestService(t, WithMaxConcurrency(1))
+
+	tracker := &peakConcurrencyTracker{}
+	for _, id := range []string{"c1", "c2"} {
+		apiClient.EXPECT().ContainerInspect(gomock.Any(), id, gomock.Any()).
+			DoAndReturn(func(context.Context, string, client.ContainerInspectOptions) (client.ContainerInspectResult, error) {
+				tracker.enter()
+				time.Sleep(20 * time.Millisecond) // widen the window for a concurrency violation to show up
+				tracker.leave()
+				return client.ContainerInspectResult{
+					Container: containerType.InspectResponse{
+						ID:     id,
+						Config: &containerType.Config{Tty: false},
+						State:  &containerType.State{},
+					},
+				}, nil
+			})
+		apiClient.EXPECT().ContainerLogs(gomock.Any(), id, gomock.Any()).
+			Return(io.NopCloser(strings.NewReader("")), nil)
+	}
+
+	u := &upSession{
+		composeService: svc,
+		globalCtx:      t.Context(),
+		logOpenLimiter: newOptionalLimiter(1),
+		options:        api.UpOptions{Start: api.StartOptions{Attach: &testLogConsumer{}}},
+	}
+
+	var eg errgroup.Group
+	eg.Go(func() error { return u.streamContainerLogs(api.ContainerEvent{ID: "c1", Service: "web"}, "") })
+	eg.Go(func() error { return u.streamContainerLogs(api.ContainerEvent{ID: "c2", Service: "web"}, "") })
+	assert.NilError(t, eg.Wait())
+	assert.Equal(t, tracker.Peak(), 1, "streamContainerLogs must share --parallel's budget for concurrent log-attach opens")
+}

pkg/compose/watch.go

@@ -129,7 +129,7 @@ func (s *composeService) getSyncImplementation(project *types.Project) (sync.Syn
 		return nil, errors.New("no available sync implementation")
 	}
 
-	return sync.NewTar(project.Name, tarDockerClient{s: s}), nil
+	return sync.NewTar(project.Name, tarDockerClient{s: s}, s.maxConcurrency), nil
 }
 
 func (s *composeService) Watch(ctx context.Context, project *types.Project, options api.WatchOptions) error {