fix: stop --parallel starving log-follow's monitor or restart's cap

Logs --follow reused the bounded errgroup for both the indefinite log
streams and the monitor goroutine, so once maxConcurrency streams
opened the monitor never started and further services never showed
logs. restart gave each service its own budget, so ready services
could multiply maxConcurrency instead of sharing it.

Bound both with a semaphore that gates only the connect/restart call,
never the indefinite work that follows, sharing one instance across
restart's services. Adds regression tests, plus one guarding the
topological node order a bounded errgroup now relies on in executor.

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

pkg/compose/compose.go

@@ -35,6 +35,7 @@ import (
 	"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"
@@ -171,6 +172,34 @@ func newLimitedErrgroup(ctx context.Context, maxConcurrency int) (*errgroup.Grou
 	return eg, ctx
 }
 
+// newOptionalLimiter returns a semaphore bounding concurrency to
+// maxConcurrency, or nil when maxConcurrency<=0 (unlimited). Use it, with
+// acquireSlot/releaseSlot, to gate only part of a goroutine's work — e.g. an
+// indefinite stream's opening call, not the stream itself — where
+// newLimitedErrgroup's whole-goroutine bound doesn't apply.
+func newOptionalLimiter(maxConcurrency int) *semaphore.Weighted {
+	if maxConcurrency <= 0 {
+		return nil
+	}
+	return semaphore.NewWeighted(int64(maxConcurrency))
+}
+
+// acquireSlot acquires a slot from limiter, or is a no-op when limiter is nil.
+func acquireSlot(ctx context.Context, limiter *semaphore.Weighted) error {
+	if limiter == nil {
+		return nil
+	}
+	return limiter.Acquire(ctx, 1)
+}
+
+// releaseSlot releases a slot acquired via acquireSlot, or is a no-op when
+// limiter is nil.
+func releaseSlot(limiter *semaphore.Weighted) {
+	if limiter != nil {
+		limiter.Release(1)
+	}
+}
+
 // WithDryRun configure Compose to run without actually applying changes
 func WithDryRun(s *composeService) error {
 	s.dryRun = true

pkg/compose/dependencies.go

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

pkg/compose/executor_test.go

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

pkg/compose/logs.go

@@ -28,6 +28,7 @@ import (
 	"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/utils"
@@ -44,10 +45,23 @@ func (s *composeService) Logs(
 		return err
 	}
 
-	eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency)
+	var eg *errgroup.Group
+	// limiter bounds how many containers are connecting (ContainerInspect +
+	// opening ContainerLogs) at once, in follow mode only: the streams
+	// themselves run indefinitely once opened, so gating the whole call
+	// (like the non-follow case does) would pin every slot forever and
+	// starve the monitor and any later container, the same reason
+	// waitDependencies is excluded from the concurrency cap.
+	var limiter *semaphore.Weighted
+	if options.Follow {
+		eg, ctx = errgroup.WithContext(ctx)
+		limiter = newOptionalLimiter(s.maxConcurrency)
+	} else {
+		eg, ctx = newLimitedErrgroup(ctx, s.maxConcurrency)
+	}
 	for _, ctr := range containers {
 		eg.Go(func() error {
-			return s.logContainer(ctx, consumer, ctr, options)
+			return s.logContainer(ctx, limiter, consumer, ctr, options)
 		})
 	}
 
@@ -61,7 +75,7 @@ func (s *composeService) Logs(
 			monitor.withServices(options.Project.ServiceNames())
 		}
 		monitor.withListener(printer.HandleEvent)
-		monitor.withListener(s.followStartedContainersLogs(ctx, eg, consumer, options))
+		monitor.withListener(s.followStartedContainersLogs(ctx, eg, limiter, consumer, options))
 		eg.Go(func() error {
 			// pass ctx so monitor will immediately stop on SIGINT
 			return monitor.Start(ctx)
@@ -95,12 +109,16 @@ func (s *composeService) selectLogsContainers(ctx context.Context, projectName s
 
 // logContainer streams a container's logs, warning when its logging driver
 // doesn't support reading logs
-func (s *composeService) logContainer(ctx context.Context, consumer api.LogConsumer, ctr container.Summary, options api.LogOptions) error {
-	res, err := s.apiClient().ContainerInspect(ctx, ctr.ID, client.ContainerInspectOptions{})
-	if err != nil {
+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 {
 		return err
 	}
-	err = s.doLogContainer(ctx, consumer, getContainerNameWithoutProject(ctr), res.Container, options)
+	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)
 	if errdefs.IsNotImplemented(err) {
 		logrus.Warnf("Can't retrieve logs for %q: %s", getCanonicalContainerName(ctr), err.Error())
 		return nil
@@ -111,7 +129,13 @@ func (s *composeService) logContainer(ctx context.Context, consumer api.LogConsu
 // followStartedContainersLogs streams the logs of containers (re)started
 // while following, ignoring those whose logging driver doesn't support
 // reading logs
-func (s *composeService) followStartedContainersLogs(ctx context.Context, eg *errgroup.Group, consumer api.LogConsumer, options api.LogOptions) api.ContainerEventListener {
+func (s *composeService) followStartedContainersLogs(
+	ctx context.Context,
+	eg *errgroup.Group,
+	limiter *semaphore.Weighted,
+	consumer api.LogConsumer,
+	options api.LogOptions,
+) api.ContainerEventListener {
 	runEnds := newRunEndTracker()
 	return func(event api.ContainerEvent) {
 		runEnds.Observe(event)
@@ -124,15 +148,19 @@ func (s *composeService) followStartedContainersLogs(ctx context.Context, eg *er
 		// would drop the whole run.
 		since := runEnds.Since(event.ID)
 		eg.Go(func() error {
+			if err := acquireSlot(ctx, limiter); err != nil {
+				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)
 			}
 
-			err = s.doLogContainer(ctx, consumer, event.Source, res.Container, api.LogOptions{
+			err = s.doLogContainer(ctx, limiter, consumer, event.Source, res.Container, api.LogOptions{
 				Follow:     options.Follow,
 				Since:      since,
 				Until:      options.Until,
@@ -216,7 +244,12 @@ func logsSinceLastRun(ctr container.InspectResponse) string {
 	return finished
 }
 
-func (s *composeService) doLogContainer(ctx context.Context, consumer api.LogConsumer, name string, ctr container.InspectResponse, options api.LogOptions) error {
+// doLogContainer opens the container's log stream and copies it to consumer
+// until it ends. The caller must have already acquired limiter's slot (see
+// acquireSlot); it is released here right after ContainerLogs returns, so a
+// long-lived --follow stream never keeps blocking new connections or the
+// monitor.
+func (s *composeService) doLogContainer(ctx context.Context, limiter *semaphore.Weighted, consumer api.LogConsumer, name string, ctr container.InspectResponse, options api.LogOptions) error {
 	r, err := s.apiClient().ContainerLogs(ctx, ctr.ID, client.ContainerLogsOptions{
 		ShowStdout: true,
 		ShowStderr: true,
@@ -226,6 +259,7 @@ func (s *composeService) doLogContainer(ctx context.Context, consumer api.LogCon
 		Tail:       options.Tail,
 		Timestamps: options.Timestamps,
 	})
+	releaseSlot(limiter)
 	if err != nil {
 		return err
 	}

pkg/compose/logs_test.go

@@ -18,16 +18,19 @@ package compose
 
 import (
 	"bytes"
+	"context"
 	"encoding/binary"
 	"errors"
 	"io"
 	"strings"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/compose-spec/compose-go/v2/types"
 	"github.com/moby/moby/api/pkg/stdcopy"
 	containerType "github.com/moby/moby/api/types/container"
+	"github.com/moby/moby/api/types/events"
 	"github.com/moby/moby/client"
 	"go.uber.org/mock/gomock"
 	"gotest.tools/v3/assert"
@@ -219,6 +222,190 @@ func TestComposeService_Logs_ServiceFiltering(t *testing.T) {
 	assert.Assert(t, is.DeepEqual([]string{"hello c4"}, consumer.LogsForContainer("c4")))
 }
 
+// TestComposeService_Logs_FollowDoesNotStarveMonitor guards against a
+// regression where bounding the errgroup used for `--follow` log streams
+// (which never return) starved the monitor goroutine, submitted after them
+// to the same group: once maxConcurrency streams were open, the monitor
+// never started and later containers never appeared.
+func TestComposeService_Logs_FollowDoesNotStarveMonitor(t *testing.T) {
+	mockCtrl := gomock.NewController(t)
+	defer mockCtrl.Finish()
+
+	api, cli := prepareMocks(mockCtrl)
+	tested, err := NewComposeService(cli, WithMaxConcurrency(1))
+	assert.NilError(t, err)
+
+	name := strings.ToLower(testProject)
+
+	api.EXPECT().ContainerList(gomock.Any(), gomock.Any()).
+		Return(client.ContainerListResult{
+			Items: []containerType.Summary{
+				testContainer("service", "c1", false),
+				testContainer("service", "c2", false),
+			},
+		}, nil).
+		Times(2) // selectLogsContainers, then the monitor's initialContainers
+
+	writers := make(map[string]*io.PipeWriter)
+	// opened only fires once a container's stream is actually established
+	// (post-semaphore, mid-copy): closing writers before that would race
+	// with the semaphore-gated open itself, closing a pipe nothing reads yet.
+	opened := make(chan struct{}, 2)
+	for _, id := range []string{"c1", "c2"} {
+		r, w := io.Pipe()
+		writers[id] = w
+		t.Cleanup(func() { _ = r.Close() })
+
+		api.EXPECT().
+			ContainerInspect(anyCancellableContext(), id, gomock.Any()).
+			Return(client.ContainerInspectResult{
+				Container: containerType.InspectResponse{
+					ID:     id,
+					Config: &containerType.Config{Tty: false},
+				},
+			}, nil)
+		// never closed on its own: simulates a `--follow` stream that stays
+		// open for the container's lifetime
+		api.EXPECT().ContainerLogs(anyCancellableContext(), id, gomock.Any()).
+			DoAndReturn(func(context.Context, string, client.ContainerLogsOptions) (io.ReadCloser, error) {
+				opened <- struct{}{}
+				return r, nil
+			})
+	}
+
+	started := make(chan struct{})
+	api.EXPECT().Events(gomock.Any(), gomock.Any()).
+		DoAndReturn(func(_ context.Context, _ client.EventsListOptions) client.EventsResult {
+			close(started)
+			return client.EventsResult{Messages: make(chan events.Message), Err: make(chan error)}
+		})
+
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+
+	done := make(chan error, 1)
+	go func() {
+		done <- tested.Logs(ctx, name, &testLogConsumer{}, compose.LogOptions{Follow: true})
+	}()
+
+	select {
+	case <-started:
+	case <-time.After(5 * time.Second):
+		t.Fatal("monitor never started: follow-mode log streams starved the bounded errgroup")
+	}
+	for range 2 {
+		select {
+		case <-opened:
+		case <-time.After(5 * time.Second):
+			t.Fatal("not every container's log stream opened")
+		}
+	}
+
+	cancel()
+	for _, w := range writers {
+		_ = w.Close()
+	}
+	assert.NilError(t, <-done)
+}
+
+// TestComposeService_Logs_FollowLimitsConcurrentStreamOpens guards against a
+// regression where removing the concurrency bound entirely for `--follow`
+// (to stop it starving the monitor, see
+// TestComposeService_Logs_FollowDoesNotStarveMonitor) let an unbounded
+// number of ContainerLogs calls open at once. Opening a stream must still
+// respect maxConcurrency; only the (indefinite) copy that follows is exempt.
+func TestComposeService_Logs_FollowLimitsConcurrentStreamOpens(t *testing.T) {
+	mockCtrl := gomock.NewController(t)
+	defer mockCtrl.Finish()
+
+	api, cli := prepareMocks(mockCtrl)
+	tested, err := NewComposeService(cli, WithMaxConcurrency(1))
+	assert.NilError(t, err)
+
+	name := strings.ToLower(testProject)
+
+	ids := []string{"c1", "c2", "c3"}
+	var containerItems []containerType.Summary
+	for _, id := range ids {
+		containerItems = append(containerItems, testContainer("service", id, false))
+	}
+	api.EXPECT().ContainerList(gomock.Any(), gomock.Any()).
+		Return(client.ContainerListResult{Items: containerItems}, nil).
+		Times(2) // selectLogsContainers, then the monitor's initialContainers
+
+	for _, id := range ids {
+		api.EXPECT().
+			ContainerInspect(anyCancellableContext(), id, gomock.Any()).
+			Return(client.ContainerInspectResult{
+				Container: containerType.InspectResponse{
+					ID:     id,
+					Config: &containerType.Config{Tty: false},
+				},
+			}, nil)
+	}
+
+	var (
+		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()
+				current++
+				if current > peak {
+					peak = current
+				}
+				mu.Unlock()
+
+				time.Sleep(20 * time.Millisecond) // widen the window for a concurrency violation to show up
+
+				mu.Lock()
+				current--
+				mu.Unlock()
+
+				r, w := io.Pipe()
+				writers <- w
+				return r, nil
+			})
+	}
+
+	started := make(chan struct{})
+	api.EXPECT().Events(gomock.Any(), gomock.Any()).
+		DoAndReturn(func(_ context.Context, _ client.EventsListOptions) client.EventsResult {
+			close(started)
+			return client.EventsResult{Messages: make(chan events.Message), Err: make(chan error)}
+		})
+
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+
+	done := make(chan error, 1)
+	go func() {
+		done <- tested.Logs(ctx, name, &testLogConsumer{}, compose.LogOptions{Follow: true})
+	}()
+
+	<-started
+	var openedWriters []*io.PipeWriter
+	for range ids {
+		select {
+		case w := <-writers:
+			openedWriters = append(openedWriters, w)
+		case <-time.After(5 * time.Second):
+			t.Fatal("not all follow-mode streams opened: maxConcurrency must bound the open burst, not the whole call")
+		}
+	}
+
+	cancel()
+	for _, w := range openedWriters {
+		_ = w.Close()
+	}
+	assert.NilError(t, <-done)
+	assert.Equal(t, peak, 1, "opening follow-mode log streams must be bounded by maxConcurrency")
+}
+
 type testLogConsumer struct {
 	mu sync.Mutex
 	// logs is keyed by container ID; values are log lines

pkg/compose/restart.go

@@ -23,6 +23,7 @@ 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"
@@ -45,6 +46,14 @@ func (s *composeService) restart(ctx context.Context, projectName string, option
 		return err
 	}
 
+	// shared by every service so the dependency-order fan-out and the
+	// per-service container fan-out combined never exceed maxConcurrency
+	// concurrent per-container restarts (pre_stop hook, ContainerRestart,
+	// post_start hook) — a per-service bound alone allows as many
+	// independent services to run at once as the graph permits, each with
+	// its own maxConcurrency budget
+	limiter := newOptionalLimiter(s.maxConcurrency)
+
 	return InDependencyOrder(ctx, project, func(c context.Context, service string) error {
 		config := project.Services[service]
 		err := s.waitDependencies(ctx, project, service, config.DependsOn, containers, 0)
@@ -52,9 +61,13 @@ func (s *composeService) restart(ctx context.Context, projectName string, option
 			return err
 		}
 
-		eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency)
+		eg, ctx := errgroup.WithContext(ctx)
 		for _, ctr := range containers.filter(isService(service)) {
 			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)
 			})
 		}

pkg/compose/restart_test.go

@@ -20,8 +20,11 @@ package compose
 
 import (
 	"context"
+	"fmt"
 	"net"
+	"sync"
 	"testing"
+	"time"
 
 	"github.com/compose-spec/compose-go/v2/types"
 	"github.com/moby/moby/api/types/container"
@@ -30,6 +33,7 @@ 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
@@ -170,3 +174,63 @@ func TestRestartContainerSkipsHooksForRelay(t *testing.T) {
 	err := svc.restartContainer(t.Context(), service, relay, api.RestartOptions{})
 	assert.NilError(t, err)
 }
+
+// TestRestart_ConcurrencyIsBoundedAcrossServices guards against a regression
+// (PR #14177 review) where each service's per-container fan-out got its own
+// maxConcurrency budget: with several independent services ready at once,
+// the dependency-order traversal could run all of them in parallel, each
+// 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)
+	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
+	for i := range numServices {
+		name := fmt.Sprintf("svc%d", i)
+		project.Services[name] = types.ServiceConfig{Name: name}
+		containers = append(containers, serviceContainer(name, 1, container.StateRunning))
+	}
+
+	apiClient.EXPECT().ContainerList(gomock.Any(), gomock.Any()).
+		Return(client.ContainerListResult{Items: containers}, nil)
+
+	var (
+		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()
+			current++
+			if current > peak {
+				peak = current
+			}
+			mu.Unlock()
+
+			time.Sleep(20 * time.Millisecond) // widen the window for a concurrency violation to show up
+
+			mu.Lock()
+			current--
+			mu.Unlock()
+			return client.ContainerRestartResult{}, nil
+		}).
+		Times(numServices)
+
+	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")
+}

pkg/compose/up.go

@@ -461,7 +461,7 @@ func (u *upSession) streamContainerLogs(event api.ContainerEvent, since string)
 		since = logsSinceLastRun(res.Container)
 	}
 
-	err = u.doLogContainer(u.globalCtx, u.options.Start.Attach, event.Source, res.Container, api.LogOptions{
+	err = u.doLogContainer(u.globalCtx, nil, u.options.Start.Attach, event.Source, res.Container, api.LogOptions{
 		Follow: true,
 		Since:  since,
 	})