diff --git a/cmd/display/tty.go b/cmd/display/tty.go index 447ecd61e7..9f379c13d4 100644 --- a/cmd/display/tty.go +++ b/cmd/display/tty.go @@ -42,19 +42,36 @@ func Full(out io.Writer, info io.Writer, detached bool) api.EventProcessor { out: out, info: info, tasks: map[string]*task{}, - done: make(chan bool), + done: newDoneSignal(), mtx: &sync.Mutex{}, detached: detached, } } +// doneSignal is a channel that's safe to close more than once: a shared bus +// can go through more than one sequential Start/Done cycle in its lifetime +// (e.g. `docker compose rm --stop` runs a "stop" cycle before its own +// "remove" cycle), and a bare channel can only ever be closed once. +type doneSignal struct { + ch chan bool + once sync.Once +} + +func newDoneSignal() *doneSignal { + return &doneSignal{ch: make(chan bool)} +} + +func (d *doneSignal) close() { + d.once.Do(func() { close(d.ch) }) +} + type ttyWriter struct { out io.Writer ids []string // tasks ids ordered as first event appeared tasks map[string]*task repeated bool numLines int - done chan bool + done *doneSignal // (re)created by Start for each Start/Done cycle mtx *sync.Mutex dryRun bool // FIXME(ndeloof) (re)implement support for dry-run operation string @@ -152,18 +169,36 @@ func (t *task) Completed() bool { } func (w *ttyWriter) Start(ctx context.Context, operation string) { - w.ticker = time.NewTicker(100 * time.Millisecond) + w.mtx.Lock() + // If a previous cycle is still open (Start called again before its + // matching Done, e.g. nested Start/Done misuse), close it out first so + // its render goroutine exits instead of leaking until ctx is done. + if w.done != nil { + w.done.close() + } + if w.ticker != nil { + w.ticker.Stop() + } + done := newDoneSignal() + ticker := time.NewTicker(100 * time.Millisecond) + w.done = done + w.ticker = ticker w.operation = operation + w.mtx.Unlock() + + // done and ticker are this cycle's own, captured locally: a later Start + // (a new Start/Done cycle on the same writer) reassigns w.done/w.ticker, + // but that never affects this goroutine's own wait. go func() { for { select { case <-ctx.Done(): // interrupted - w.ticker.Stop() + ticker.Stop() return - case <-w.done: + case <-done.ch: return - case <-w.ticker.C: + case <-ticker.C: w.print() } } @@ -172,11 +207,19 @@ func (w *ttyWriter) Start(ctx context.Context, operation string) { func (w *ttyWriter) Done(operation string, success bool) { w.print() - w.done <- true + w.mtx.Lock() + done := w.done + ticker := w.ticker + w.mtx.Unlock() + + // close never blocks, unlike a send, so this returns even if the render + // goroutine already exited via ctx.Done(). + done.close() + w.mtx.Lock() defer w.mtx.Unlock() - if w.ticker != nil { - w.ticker.Stop() + if ticker != nil { + ticker.Stop() } w.operation = "" } diff --git a/cmd/display/tty_test.go b/cmd/display/tty_test.go index c6d6165c31..3e6b870d6b 100644 --- a/cmd/display/tty_test.go +++ b/cmd/display/tty_test.go @@ -20,12 +20,14 @@ import ( "bytes" "context" "fmt" + "io" "strings" "sync" "testing" "time" "unicode/utf8" + "go.uber.org/goleak" "gotest.tools/v3/assert" "github.com/docker/compose/v5/pkg/api" @@ -37,7 +39,7 @@ func newTestWriter() (*ttyWriter, *bytes.Buffer) { out: &buf, info: &buf, tasks: map[string]*task{}, - done: make(chan bool), + done: newDoneSignal(), mtx: &sync.Mutex{}, operation: "pull", } @@ -549,6 +551,15 @@ func TestLenAnsi(t *testing.T) { } } +// TestDoneBeforeStartDoesNotPanic guards against a nil-pointer panic if Done +// is ever called without a prior Start: Full must initialize done, since +// ttyWriter is exposed only through the public api.EventProcessor interface +// and a caller isn't guaranteed to call Start first. +func TestDoneBeforeStartDoesNotPanic(t *testing.T) { + w := Full(io.Discard, io.Discard, false) + w.Done("op", false) +} + func TestDoneDeadlockFix(t *testing.T) { w, _ := newTestWriter() addTask(w, "test-task", "Working", "details", api.Working) @@ -569,6 +580,91 @@ func TestDoneDeadlockFix(t *testing.T) { } } +// TestDoneAfterContextCancelDoesNotHang is the regression test for +// docker/compose#14114: a single SIGTERM/SIGINT cancels the command context +// (AdaptCmd). The render goroutine started by Start then exits through its +// ctx.Done() branch and never receives from the done channel. Done must +// still return. +func TestDoneAfterContextCancelDoesNotHang(t *testing.T) { + w, _ := newTestWriter() + ctx, cancel := context.WithCancel(t.Context()) + w.Start(ctx, "down") + cancel() + // Let the render goroutine observe the cancellation and exit. + time.Sleep(100 * time.Millisecond) + + finished := make(chan struct{}) + go func() { + w.Done("down", false) + close(finished) + }() + select { + case <-finished: + case <-time.After(2 * time.Second): + t.Fatal("ttyWriter.Done blocked forever after context cancellation") + } +} + +// TestNestedStartDoneDoesNotPanic guards against a panic if Start/Done are +// ever nested (a Start called again before the matching Done): closing a +// channel more than once panics unless each cycle's own doneSignal (with its +// own sync.Once) is used, instead of a single shared channel. It also +// guards against a goroutine leak: the outer cycle's render goroutine must +// not be left waiting on a doneSignal nobody closes once Start replaces it. +func TestNestedStartDoneDoesNotPanic(t *testing.T) { + w, _ := newTestWriter() + ctx := t.Context() + + w.Start(ctx, "outer") + w.Start(ctx, "inner") + w.Done("inner", false) + w.Done("outer", false) + + // give the outer cycle's render goroutine a chance to observe the + // second Start closing its signal and exit before goleak checks. + time.Sleep(100 * time.Millisecond) + goleak.VerifyNone(t) +} + +// TestSequentialStartDoneEachGetFreshChannel is the regression test for the +// bus going through more than one Start/Done cycle in its lifetime, as +// happens with `docker compose rm --stop` (a "stop" cycle, then its own +// "remove" cycle): each Start must allocate a fresh doneSignal, independent +// of any earlier cycle's already-closed one, or the second cycle's render +// goroutine would see it as immediately closed and exit without ever +// ticking. +func TestSequentialStartDoneEachGetFreshChannel(t *testing.T) { + w, _ := newTestWriter() + ctx := t.Context() + + w.Start(ctx, "first") + first := w.done + w.Done("first", false) + select { + case <-first.ch: + default: + t.Fatal("expected the first cycle's done channel to be closed after its Done") + } + + w.Start(ctx, "second") + second := w.done + if second == first { + t.Fatal("expected Start to allocate a fresh done signal for the new cycle") + } + select { + case <-second.ch: + t.Fatal("expected the second cycle's done channel to still be open before its own Done") + default: + } + + w.Done("second", false) + select { + case <-second.ch: + default: + t.Fatal("expected the second cycle's done channel to be closed after its own Done") + } +} + // TestAdjustLineWidth_WideProgressForcesSizeInfoDrop is the unit-level // regression test for docker/compose#13595. When progress contains the // " X.XMB / Y.YMB" size suffix and the bar makes beforeStatus large enough diff --git a/pkg/compose/publish.go b/pkg/compose/publish.go index c1590089a5..f0fb5c2313 100644 --- a/pkg/compose/publish.go +++ b/pkg/compose/publish.go @@ -64,7 +64,11 @@ func (s *composeService) publish(ctx context.Context, project *types.Project, re if !accept { return api.ErrCanceled } - err = s.Push(ctx, project, api.PushOptions{IgnoreFailures: true, ImageMandatory: true}) + // unexported push, not the public Push: publish already runs inside its + // own "publish" Start/Done bracket, and Push would open a second one on + // the same shared bus. Quiet:true would also suppress push progress + // reporting, not just the bracket, so it's not an option here. + err = s.push(ctx, project, api.PushOptions{IgnoreFailures: true, ImageMandatory: true}) if err != nil { return err } diff --git a/pkg/compose/remove.go b/pkg/compose/remove.go index bdf6400a9e..cae6085054 100644 --- a/pkg/compose/remove.go +++ b/pkg/compose/remove.go @@ -31,6 +31,8 @@ func (s *composeService) Remove(ctx context.Context, projectName string, options projectName = strings.ToLower(projectName) if options.Stop { + // Stop's own "stop" Start/Done cycle runs sequentially, fully + // closed, before "remove"'s below — ttyWriter supports that. err := s.Stop(ctx, projectName, api.StopOptions{ Services: options.Services, Project: options.Project, diff --git a/pkg/compose/run.go b/pkg/compose/run.go index 076989cbec..3b3308493d 100644 --- a/pkg/compose/run.go +++ b/pkg/compose/run.go @@ -23,6 +23,7 @@ import ( "os" "os/signal" "slices" + "strings" "github.com/compose-spec/compose-go/v2/types" "github.com/docker/cli/cli" @@ -279,7 +280,11 @@ func (s *composeService) resolveRunServiceReferences(ctx context.Context, projec func (s *composeService) startDependencies(ctx context.Context, project *types.Project, options api.RunOptions) error { project = project.WithServicesDisabled(options.Service) - err := s.Create(ctx, project, api.CreateOptions{ + // calls the unexported create/start, not the public Create/Start: this + // already runs inside the "run" operation's Start/Done bracket (see + // prepareRun), and the public variants would open a second, nested one + // on the same shared bus. + err := s.create(ctx, project, api.CreateOptions{ Build: options.Build, IgnoreOrphans: options.IgnoreOrphans, RemoveOrphans: options.RemoveOrphans, @@ -290,9 +295,9 @@ func (s *composeService) startDependencies(ctx context.Context, project *types.P } if len(project.Services) > 0 { - return s.Start(ctx, project.Name, api.StartOptions{ + return s.start(ctx, strings.ToLower(project.Name), api.StartOptions{ Project: project, - }) + }, nil) } return nil }