From ea8807943346ffe3d03000bf611ff738ee3f444a Mon Sep 17 00:00:00 2001 From: Guillaume Lours Date: Wed, 16 Sep 2026 19:38:13 +0200 Subject: [PATCH] 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 | 54 ++++++++-- 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, 407 insertions(+), 15 deletions(-) diff --git a/pkg/compose/compose.go b/pkg/compose/compose.go index 979a37263..09596a933 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 a502b4645..b1c0a231f 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 ab825fc38..63b9ed64e 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/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 diff --git a/pkg/compose/logs.go b/pkg/compose/logs.go index f5b3f74ab..0fc281716 100644 --- a/pkg/compose/logs.go +++ b/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 } diff --git a/pkg/compose/logs_test.go b/pkg/compose/logs_test.go index 27a9b859a..b4b4fb5c4 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 920d911b2..5373d06af 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 9c04f1147..e7904d1fb 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 @@ -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") +} diff --git a/pkg/compose/up.go b/pkg/compose/up.go index 8aed08360..93a2e80d1 100644 --- a/pkg/compose/up.go +++ b/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, })