From a6d309ea4ae0aca1a71c1e13d6206f8f35c83624 Mon Sep 17 00:00:00 2001 From: Guillaume Lours Date: Thu, 3 Sep 2026 17:13:25 +0200 Subject: [PATCH 1/2] fix: honor --parallel across all bulk engine-call fan-outs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit errgroup.SetLimit(0) means "allow zero goroutines", not "unlimited", and maxConcurrency's Go zero-value is 0 — only NewComposeService sets it to -1 explicitly. Any composeService{} literal built without it (common in tests) silently deadlocked at every call site that called SetLimit unconditionally. Separately, --parallel/COMPOSE_PARALLEL_LIMIT was only ever wired into pull, push, and the dependency-graph traversal, despite the docs promising a generic bound on "concurrent engine calls". Every other bulk operation (kill, pause, down, the up/create plan executor, logs, ps, top, wait, restart, remove, images, model pulls, watch) launched one goroutine per container/image/DAG-node with no cap at all. Fix both via a shared newLimitedErrgroup helper applied at every fan-out site, threading maxConcurrency through forEachContainerConcurrent and ImagePruner, which had no access to composeService. Add a regression test for the zero-value case. service_containers.go's waitDependencies is intentionally left unguarded: it's a per-dependency ticker poll, not a burst of engine calls. Signed-off-by: Guillaume Lours --- pkg/compose/compose.go | 16 ++++++++++- pkg/compose/compose_test.go | 54 +++++++++++++++++++++++++++++++++++++ pkg/compose/containers.go | 5 ++-- pkg/compose/down.go | 9 +++---- pkg/compose/executor.go | 9 +++++-- pkg/compose/image_pruner.go | 18 ++++++------- pkg/compose/images.go | 5 ++-- pkg/compose/kill.go | 2 +- pkg/compose/logs.go | 2 +- pkg/compose/model.go | 3 +-- pkg/compose/pause.go | 4 +-- pkg/compose/ps.go | 3 +-- pkg/compose/pull.go | 6 ++--- pkg/compose/push.go | 4 +-- pkg/compose/remove.go | 3 +-- pkg/compose/restart.go | 3 +-- pkg/compose/top.go | 3 +-- pkg/compose/wait.go | 3 +-- pkg/compose/watch.go | 4 +-- pkg/compose/watch_test.go | 3 +-- 20 files changed, 109 insertions(+), 50 deletions(-) create mode 100644 pkg/compose/compose_test.go diff --git a/pkg/compose/compose.go b/pkg/compose/compose.go index ee797274f85..bbf46e05c56 100644 --- a/pkg/compose/compose.go +++ b/pkg/compose/compose.go @@ -34,6 +34,7 @@ import ( "github.com/moby/moby/api/types/swarm" "github.com/moby/moby/client" "github.com/sirupsen/logrus" + "golang.org/x/sync/errgroup" "github.com/docker/compose/v5/pkg/api" "github.com/docker/compose/v5/pkg/dryrun" @@ -149,7 +150,8 @@ func WithPrompt(prompt Prompt) Option { } } -// WithMaxConcurrency defines upper limit for concurrent operations against engine API +// WithMaxConcurrency defines upper limit for concurrent operations against +// engine API. A value <= 0 means unlimited. func WithMaxConcurrency(maxConcurrency int) Option { return func(s *composeService) error { s.maxConcurrency = maxConcurrency @@ -157,6 +159,18 @@ func WithMaxConcurrency(maxConcurrency int) Option { } } +// newLimitedErrgroup returns an errgroup.Group bounded to maxConcurrency +// concurrent goroutines. maxConcurrency<=0 (including the Go zero-value) +// leaves it unlimited, since errgroup.SetLimit(0) means "allow zero +// goroutines", not "unlimited". +func newLimitedErrgroup(ctx context.Context, maxConcurrency int) (*errgroup.Group, context.Context) { + eg, ctx := errgroup.WithContext(ctx) + if maxConcurrency > 0 { + eg.SetLimit(maxConcurrency) + } + return eg, ctx +} + // WithDryRun configure Compose to run without actually applying changes func WithDryRun(s *composeService) error { s.dryRun = true diff --git a/pkg/compose/compose_test.go b/pkg/compose/compose_test.go new file mode 100644 index 00000000000..71eec918f15 --- /dev/null +++ b/pkg/compose/compose_test.go @@ -0,0 +1,54 @@ +/* + Copyright 2026 Docker Compose CLI authors + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +package compose + +import ( + "fmt" + "testing" + "time" + + "gotest.tools/v3/assert" +) + +// TestNewLimitedErrgroup_NonPositiveIsUnlimited guards against the bug this +// helper exists to fix: errgroup.SetLimit(0) means "allow zero goroutines", +// not "unlimited". A composeService{} built without going through +// NewComposeService has maxConcurrency's Go zero-value (0), so an +// unconditional SetLimit(maxConcurrency) at any call site would silently +// hang forever instead of running. +func TestNewLimitedErrgroup_NonPositiveIsUnlimited(t *testing.T) { + for _, maxConcurrency := range []int{0, -1} { + t.Run(fmt.Sprintf("maxConcurrency=%d", maxConcurrency), func(t *testing.T) { + eg, _ := newLimitedErrgroup(t.Context(), maxConcurrency) + + done := make(chan struct{}) + go func() { + for range 5 { + eg.Go(func() error { return nil }) + } + close(done) + }() + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("eg.Go blocked: maxConcurrency <= 0 must mean unlimited, not SetLimit(0) (zero goroutines allowed)") + } + assert.NilError(t, eg.Wait()) + }) + } +} diff --git a/pkg/compose/containers.go b/pkg/compose/containers.go index c6ce6474fb2..fa22363af75 100644 --- a/pkg/compose/containers.go +++ b/pkg/compose/containers.go @@ -26,7 +26,6 @@ import ( "github.com/compose-spec/compose-go/v2/types" "github.com/moby/moby/api/types/container" "github.com/moby/moby/client" - "golang.org/x/sync/errgroup" "github.com/docker/compose/v5/pkg/api" ) @@ -186,8 +185,8 @@ func (containers Containers) filter(predicates ...containerPredicate) Containers } // forEachContainerConcurrent runs fn for every container concurrently and waits for all goroutines. -func forEachContainerConcurrent(ctx context.Context, containers Containers, fn func(context.Context, container.Summary) error) error { - eg, ctx := errgroup.WithContext(ctx) +func forEachContainerConcurrent(ctx context.Context, maxConcurrency int, containers Containers, fn func(context.Context, container.Summary) error) error { + eg, ctx := newLimitedErrgroup(ctx, maxConcurrency) for _, ctr := range containers { eg.Go(func() error { return fn(ctx, ctr) diff --git a/pkg/compose/down.go b/pkg/compose/down.go index 6e408ef8725..1b21f0ea23c 100644 --- a/pkg/compose/down.go +++ b/pkg/compose/down.go @@ -28,7 +28,6 @@ import ( "github.com/moby/moby/api/types/image" "github.com/moby/moby/client" "github.com/sirupsen/logrus" - "golang.org/x/sync/errgroup" "github.com/docker/compose/v5/pkg/api" "github.com/docker/compose/v5/pkg/utils" @@ -132,7 +131,7 @@ func (s *composeService) down(ctx context.Context, projectName string, options a logrus.Warnf("Warning: No resource found to remove for project %q.", projectName) } - eg, ctx := errgroup.WithContext(ctx) + eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency) for _, op := range ops { eg.Go(op) } @@ -155,7 +154,7 @@ func (s *composeService) ensureVolumesDown(ctx context.Context, project *types.P } func (s *composeService) ensureImagesDown(ctx context.Context, project *types.Project, options api.DownOptions) ([]downOp, error) { - imagePruner := NewImagePruner(s.apiClient(), project) + imagePruner := NewImagePruner(s.apiClient(), project, s.maxConcurrency) pruneOpts := ImagePruneOptions{ Mode: ImagePruneMode(options.Images), RemoveOrphans: options.RemoveOrphans, @@ -336,7 +335,7 @@ func (s *composeService) stopContainer(ctx context.Context, service *types.Servi } func (s *composeService) stopContainers(ctx context.Context, serv *types.ServiceConfig, containers []containerType.Summary, timeout *time.Duration, listener api.ContainerEventListener) error { - eg, ctx := errgroup.WithContext(ctx) + eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency) for _, ctr := range containers { eg.Go(func() error { return s.stopContainer(ctx, serv, ctr, timeout, listener) @@ -346,7 +345,7 @@ func (s *composeService) stopContainers(ctx context.Context, serv *types.Service } func (s *composeService) removeContainers(ctx context.Context, containers []containerType.Summary, service *types.ServiceConfig, timeout *time.Duration, volumes bool) error { - eg, ctx := errgroup.WithContext(ctx) + eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency) for _, ctr := range containers { eg.Go(func() error { return s.stopAndRemoveContainer(ctx, ctr, service, timeout, volumes) diff --git a/pkg/compose/executor.go b/pkg/compose/executor.go index c23b3f7b212..8adca90d0f2 100644 --- a/pkg/compose/executor.go +++ b/pkg/compose/executor.go @@ -22,7 +22,6 @@ import ( "sync" "github.com/compose-spec/compose-go/v2/types" - "golang.org/x/sync/errgroup" ) // planExecutor executes a reconciliation Plan by walking the DAG and performing @@ -102,7 +101,13 @@ func (exec *planExecutor) run(ctx context.Context, plan *Plan) error { groups := exec.buildGroupTracker(plan) events := exec.compose.events - eg, ctx := errgroup.WithContext(ctx) + // Each node's goroutine occupies its concurrency slot for the entire wait + // below, not just its own work, so a small maxConcurrency can serialize + // more than a caller might expect on a wide/shallow DAG. Forward progress + // is still guaranteed: plan.Nodes is topologically sorted, so a node's + // dependencies were always already dispatched to eg.Go by the time this + // loop reaches it. + eg, ctx := newLimitedErrgroup(ctx, exec.compose.maxConcurrency) for _, node := range plan.Nodes { eg.Go(func() error { // Wait for all dependencies diff --git a/pkg/compose/image_pruner.go b/pkg/compose/image_pruner.go index 98def7c4489..43e15021075 100644 --- a/pkg/compose/image_pruner.go +++ b/pkg/compose/image_pruner.go @@ -28,7 +28,6 @@ import ( "github.com/moby/moby/api/types/image" "github.com/moby/moby/client" "github.com/sirupsen/logrus" - "golang.org/x/sync/errgroup" "github.com/docker/compose/v5/pkg/api" ) @@ -59,15 +58,17 @@ type ImagePruneOptions struct { // ImagePruner handles image removal during Compose `down` operations. type ImagePruner struct { - client client.ImageAPIClient - project *types.Project + client client.ImageAPIClient + project *types.Project + maxConcurrency int } // NewImagePruner creates an ImagePruner object for a project. -func NewImagePruner(imageClient client.ImageAPIClient, project *types.Project) *ImagePruner { +func NewImagePruner(imageClient client.ImageAPIClient, project *types.Project, maxConcurrency int) *ImagePruner { return &ImagePruner{ - client: imageClient, - project: project, + client: imageClient, + project: project, + maxConcurrency: maxConcurrency, } } @@ -173,8 +174,7 @@ func (s *composeService) removeDanglingImages(ctx context.Context, projectName s var mu sync.Mutex var removed []string - eg, ctx := errgroup.WithContext(ctx) - eg.SetLimit(s.maxConcurrency) + eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency) for _, img := range res.Items { if keep(img) { continue @@ -225,7 +225,7 @@ func (p *ImagePruner) filterImagesByExistence(ctx context.Context, imageNames [] var mu sync.Mutex var ret []string - eg, ctx := errgroup.WithContext(ctx) + eg, ctx := newLimitedErrgroup(ctx, p.maxConcurrency) for _, img := range imageNames { eg.Go(func() error { _, err := p.client.ImageInspect(ctx, img) diff --git a/pkg/compose/images.go b/pkg/compose/images.go index 96e83972b7f..b20bb355395 100644 --- a/pkg/compose/images.go +++ b/pkg/compose/images.go @@ -35,7 +35,6 @@ import ( godigest "github.com/opencontainers/go-digest" specs "github.com/opencontainers/image-spec/specs-go/v1" "github.com/sirupsen/logrus" - "golang.org/x/sync/errgroup" "github.com/docker/compose/v5/pkg/api" ) @@ -71,7 +70,7 @@ func (s *composeService) Images(ctx context.Context, projectName string, options summary := map[string]api.ImageSummary{} var mux sync.Mutex - eg, ctx := errgroup.WithContext(ctx) + eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency) for _, ctr := range containers { eg.Go(func() error { img, err := s.containerImageSummary(ctx, ctr, withPlatform) @@ -164,7 +163,7 @@ func (s *composeService) inspectLocalImages(ctx context.Context, repoTags []stri } inspections := map[string]client.ImageInspectResult{} l := sync.Mutex{} - eg, ctx := errgroup.WithContext(ctx) + eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency) for _, repoTag := range repoTags { eg.Go(func() error { inspect, err := s.apiClient().ImageInspect(ctx, repoTag, opts...) diff --git a/pkg/compose/kill.go b/pkg/compose/kill.go index c6caffc230f..b149c1103ea 100644 --- a/pkg/compose/kill.go +++ b/pkg/compose/kill.go @@ -56,7 +56,7 @@ func (s *composeService) kill(ctx context.Context, projectName string, options a return api.ErrNoResources } - return forEachContainerConcurrent(ctx, containers, func(ctx context.Context, ctr container.Summary) error { + return forEachContainerConcurrent(ctx, s.maxConcurrency, containers, func(ctx context.Context, ctr container.Summary) error { eventName := getContainerProgressName(ctr) s.events.On(newEvent(eventName, api.Working, api.StatusKilling)) _, err := s.apiClient().ContainerKill(ctx, ctr.ID, client.ContainerKillOptions{ diff --git a/pkg/compose/logs.go b/pkg/compose/logs.go index 5bacaf76be3..15a8b7e09c7 100644 --- a/pkg/compose/logs.go +++ b/pkg/compose/logs.go @@ -42,7 +42,7 @@ func (s *composeService) Logs( return err } - eg, ctx := errgroup.WithContext(ctx) + eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency) for _, ctr := range containers { eg.Go(func() error { return s.logContainer(ctx, consumer, ctr, options) diff --git a/pkg/compose/model.go b/pkg/compose/model.go index 928ff718064..2544d74d69d 100644 --- a/pkg/compose/model.go +++ b/pkg/compose/model.go @@ -32,7 +32,6 @@ import ( "github.com/docker/cli/cli-plugins/manager" "github.com/moby/moby/client/pkg/versions" "github.com/spf13/cobra" - "golang.org/x/sync/errgroup" "github.com/docker/compose/v5/pkg/api" ) @@ -49,7 +48,7 @@ func (s *composeService) ensureModels(ctx context.Context, project *types.Projec defer mdlAPI.Close() availableModels, err := mdlAPI.ListModels(ctx) - eg, ctx := errgroup.WithContext(ctx) + eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency) eg.Go(func() error { return mdlAPI.SetModelVariables(ctx, project) }) diff --git a/pkg/compose/pause.go b/pkg/compose/pause.go index 83d9937411f..5e7c3a75777 100644 --- a/pkg/compose/pause.go +++ b/pkg/compose/pause.go @@ -42,7 +42,7 @@ func (s *composeService) pause(ctx context.Context, projectName string, options containers = containers.filter(isService(options.Project.ServiceNames()...)) } - return forEachContainerConcurrent(ctx, containers, func(ctx context.Context, ctr container.Summary) error { + return forEachContainerConcurrent(ctx, s.maxConcurrency, containers, func(ctx context.Context, ctr container.Summary) error { _, err := s.apiClient().ContainerPause(ctx, ctr.ID, client.ContainerPauseOptions{}) if err == nil { s.events.On(newEvent(getContainerProgressName(ctr), api.Done, "Paused")) @@ -67,7 +67,7 @@ func (s *composeService) unPause(ctx context.Context, projectName string, option containers = containers.filter(isService(options.Project.ServiceNames()...)) } - return forEachContainerConcurrent(ctx, containers, func(ctx context.Context, ctr container.Summary) error { + return forEachContainerConcurrent(ctx, s.maxConcurrency, containers, func(ctx context.Context, ctr container.Summary) error { _, err := s.apiClient().ContainerUnpause(ctx, ctr.ID, client.ContainerUnpauseOptions{}) if err == nil { s.events.On(newEvent(getContainerProgressName(ctr), api.Done, "Unpaused")) diff --git a/pkg/compose/ps.go b/pkg/compose/ps.go index 7bd078dfbae..e6c78673b9c 100644 --- a/pkg/compose/ps.go +++ b/pkg/compose/ps.go @@ -23,7 +23,6 @@ import ( "github.com/moby/moby/api/types/container" "github.com/moby/moby/client" - "golang.org/x/sync/errgroup" "github.com/docker/compose/v5/pkg/api" ) @@ -43,7 +42,7 @@ func (s *composeService) Ps(ctx context.Context, projectName string, options api containers = containers.filter(isService(options.Services...)) } summary := make([]api.ContainerSummary, len(containers)) - eg, ctx := errgroup.WithContext(ctx) + eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency) for i, ctr := range containers { eg.Go(func() error { var err error diff --git a/pkg/compose/pull.go b/pkg/compose/pull.go index 4cfc3fcc3f0..0733cf14d04 100644 --- a/pkg/compose/pull.go +++ b/pkg/compose/pull.go @@ -69,8 +69,7 @@ func (s *composeService) pull(ctx context.Context, project *types.Project, opts return err } - eg, ctx := errgroup.WithContext(ctx) - eg.SetLimit(s.maxConcurrency) + eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency) p := &imagePuller{ composeService: s, @@ -429,8 +428,7 @@ func (s *composeService) pullRequiredImages(ctx context.Context, project *types. // the errgroup context is canceled as soon as Wait returns; the post-pull // resolution below needs the caller's context - eg, pullCtx := errgroup.WithContext(ctx) - eg.SetLimit(s.maxConcurrency) + eg, pullCtx := newLimitedErrgroup(ctx, s.maxConcurrency) pulled := map[string]bool{} var mutex sync.Mutex for name, service := range needPull { diff --git a/pkg/compose/push.go b/pkg/compose/push.go index 494c2c79d1f..6a904cc9127 100644 --- a/pkg/compose/push.go +++ b/pkg/compose/push.go @@ -29,7 +29,6 @@ import ( "github.com/docker/go-units" "github.com/moby/moby/api/types/jsonstream" "github.com/moby/moby/client" - "golang.org/x/sync/errgroup" "github.com/docker/compose/v5/internal/registry" "github.com/docker/compose/v5/pkg/api" @@ -45,8 +44,7 @@ func (s *composeService) Push(ctx context.Context, project *types.Project, optio } func (s *composeService) push(ctx context.Context, project *types.Project, options api.PushOptions) error { - eg, ctx := errgroup.WithContext(ctx) - eg.SetLimit(s.maxConcurrency) + eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency) for _, service := range project.Services { if service.Build == nil || service.Image == "" { diff --git a/pkg/compose/remove.go b/pkg/compose/remove.go index b24f0cfb33b..98151f4313a 100644 --- a/pkg/compose/remove.go +++ b/pkg/compose/remove.go @@ -22,7 +22,6 @@ import ( "strings" "github.com/moby/moby/client" - "golang.org/x/sync/errgroup" "github.com/docker/compose/v5/pkg/api" ) @@ -98,7 +97,7 @@ func (s *composeService) Remove(ctx context.Context, projectName string, options } func (s *composeService) remove(ctx context.Context, containers Containers, options api.RemoveOptions) error { - eg, ctx := errgroup.WithContext(ctx) + eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency) for _, ctr := range containers { eg.Go(func() error { eventName := getContainerProgressName(ctr) diff --git a/pkg/compose/restart.go b/pkg/compose/restart.go index 461a0257c52..b40b58e97e5 100644 --- a/pkg/compose/restart.go +++ b/pkg/compose/restart.go @@ -23,7 +23,6 @@ import ( "github.com/compose-spec/compose-go/v2/types" "github.com/moby/moby/api/types/container" "github.com/moby/moby/client" - "golang.org/x/sync/errgroup" "github.com/docker/compose/v5/pkg/api" "github.com/docker/compose/v5/pkg/utils" @@ -53,7 +52,7 @@ func (s *composeService) restart(ctx context.Context, projectName string, option return err } - eg, ctx := errgroup.WithContext(ctx) + eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency) for _, ctr := range containers.filter(isService(service)) { eg.Go(func() error { return s.restartContainer(ctx, project.Services[service], ctr, options) diff --git a/pkg/compose/top.go b/pkg/compose/top.go index 9e736d6e2d1..a01566c7cd1 100644 --- a/pkg/compose/top.go +++ b/pkg/compose/top.go @@ -21,7 +21,6 @@ import ( "strings" "github.com/moby/moby/client" - "golang.org/x/sync/errgroup" "github.com/docker/compose/v5/pkg/api" ) @@ -37,7 +36,7 @@ func (s *composeService) Top(ctx context.Context, projectName string, services [ containers = containers.filter(isService(services...)) } summary := make([]api.ContainerProcSummary, len(containers)) - eg, ctx := errgroup.WithContext(ctx) + eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency) for i, ctr := range containers { eg.Go(func() error { topContent, err := s.apiClient().ContainerTop(ctx, ctr.ID, client.ContainerTopOptions{ diff --git a/pkg/compose/wait.go b/pkg/compose/wait.go index 29848786b6f..45d3a8f5469 100644 --- a/pkg/compose/wait.go +++ b/pkg/compose/wait.go @@ -21,7 +21,6 @@ import ( "fmt" "github.com/moby/moby/client" - "golang.org/x/sync/errgroup" "github.com/docker/compose/v5/pkg/api" ) @@ -35,7 +34,7 @@ func (s *composeService) Wait(ctx context.Context, projectName string, options a return 0, fmt.Errorf("no containers for project %q", projectName) } - eg, waitCtx := errgroup.WithContext(ctx) + eg, waitCtx := newLimitedErrgroup(ctx, s.maxConcurrency) var statusCode int64 for _, ctr := range containers { eg.Go(func() error { diff --git a/pkg/compose/watch.go b/pkg/compose/watch.go index 0c6b6f8b1d9..610cd94bd9b 100644 --- a/pkg/compose/watch.go +++ b/pkg/compose/watch.go @@ -197,7 +197,7 @@ func (s *composeService) watch(ctx context.Context, project *types.Project, opti if err != nil { return nil, err } - eg, ctx := errgroup.WithContext(ctx) + eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency) var ( rules []watchRule @@ -654,7 +654,7 @@ func (s *composeService) handleWatchBatch(ctx context.Context, project *types.Pr fmt.Sprintf("service(s) %q restarted", services)) } - eg, ctx := errgroup.WithContext(ctx) + eg, ctx := newLimitedErrgroup(ctx, s.maxConcurrency) for service, rulesToExec := range exec { slices.Sort(rulesToExec) for _, i := range slices.Compact(rulesToExec) { diff --git a/pkg/compose/watch_test.go b/pkg/compose/watch_test.go index e217d627e8e..3f5e252a3ff 100644 --- a/pkg/compose/watch_test.go +++ b/pkg/compose/watch_test.go @@ -119,8 +119,7 @@ func TestWatch_Sync(t *testing.T) { syncer := newFakeSyncer() go func() { service := composeService{ - dockerCli: cli, - maxConcurrency: -1, + dockerCli: cli, } rules, err := getWatchRules(&types.DevelopConfig{ Watch: []types.Trigger{ From 79a03e7db609cd17a3ee9321679ec733c3d6f4a0 Mon Sep 17 00:00:00 2001 From: Guillaume Lours Date: Wed, 16 Sep 2026 19:38:13 +0200 Subject: [PATCH 2/2] 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 --- pkg/compose/compose.go | 29 ++++++ pkg/compose/dependencies.go | 8 +- pkg/compose/executor_test.go | 63 +++++++++++- pkg/compose/logs.go | 50 ++++++++-- pkg/compose/logs_test.go | 187 +++++++++++++++++++++++++++++++++++ pkg/compose/restart.go | 15 ++- pkg/compose/restart_test.go | 64 ++++++++++++ pkg/compose/up.go | 2 +- 8 files changed, 405 insertions(+), 13 deletions(-) diff --git a/pkg/compose/compose.go b/pkg/compose/compose.go index bbf46e05c56..fe8e6570e40 100644 --- a/pkg/compose/compose.go +++ b/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 diff --git a/pkg/compose/dependencies.go b/pkg/compose/dependencies.go index a502b4645cf..b1c0a231f8d 100644 --- a/pkg/compose/dependencies.go +++ b/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 } diff --git a/pkg/compose/executor_test.go b/pkg/compose/executor_test.go index 9e711c22f3e..44880385505 100644 --- a/pkg/compose/executor_test.go +++ b/pkg/compose/executor_test.go @@ -21,6 +21,7 @@ import ( "errors" "strconv" "testing" + "time" "github.com/compose-spec/compose-go/v2/types" "github.com/moby/moby/api/types/container" @@ -40,14 +41,14 @@ 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) apiClient := mocks.NewMockAPIClient(mockCtrl) cli.EXPECT().Client().Return(apiClient).AnyTimes() - svc, err := NewComposeService(cli, WithEventProcessor(noopEventProcessor{})) + svc, err := NewComposeService(cli, append([]Option{WithEventProcessor(noopEventProcessor{})}, opts...)...) assert.NilError(t, err) return svc.(*composeService), apiClient } @@ -287,6 +288,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 diff --git a/pkg/compose/logs.go b/pkg/compose/logs.go index 15a8b7e09c7..95381890920 100644 --- a/pkg/compose/logs.go +++ b/pkg/compose/logs.go @@ -26,6 +26,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" @@ -42,10 +43,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) }) } @@ -59,7 +73,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) @@ -93,12 +107,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 { +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 + } res, err := s.apiClient().ContainerInspect(ctx, ctr.ID, client.ContainerInspectOptions{}) if err != nil { + releaseSlot(limiter) return err } - err = s.doLogContainer(ctx, consumer, getContainerNameWithoutProject(ctr), res.Container, options) + 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 @@ -109,18 +127,28 @@ 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 { return func(event api.ContainerEvent) { if event.Type != api.ContainerEventStarted { return } 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 } - 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: res.Container.State.StartedAt, Until: options.Until, @@ -136,7 +164,12 @@ func (s *composeService) followStartedContainersLogs(ctx context.Context, eg *er } } -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, @@ -146,6 +179,7 @@ func (s *composeService) doLogContainer(ctx context.Context, consumer api.LogCon Tail: options.Tail, Timestamps: options.Timestamps, }) + releaseSlot(limiter) if err != nil { return err } diff --git a/pkg/compose/logs_test.go b/pkg/compose/logs_test.go index b0499f15606..c216aeda219 100644 --- a/pkg/compose/logs_test.go +++ b/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 diff --git a/pkg/compose/restart.go b/pkg/compose/restart.go index b40b58e97e5..300d29d8d11 100644 --- a/pkg/compose/restart.go +++ b/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) }) } diff --git a/pkg/compose/restart_test.go b/pkg/compose/restart_test.go index d35dba7ca9c..f25ba5d6741 100644 --- a/pkg/compose/restart_test.go +++ b/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 @@ -150,3 +154,63 @@ func TestRestartContainer_Order(t *testing.T) { "Container prj-web-1: Started", }) } + +// 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") +} diff --git a/pkg/compose/up.go b/pkg/compose/up.go index c08075ec988..2a02b80b0de 100644 --- a/pkg/compose/up.go +++ b/pkg/compose/up.go @@ -366,7 +366,7 @@ func (u *upSession) streamContainerLogs(event api.ContainerEvent) error { return err } - 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: res.Container.State.StartedAt, })