From a8ad922c8851bde6a564b0dee25ca5986467886d Mon Sep 17 00:00:00 2001 From: Guillaume Lours Date: Mon, 24 Aug 2026 12:53:35 +0200 Subject: [PATCH] fix(display): stop ttyWriter.Done from hanging after context cancel Fixes #14114: Done() blocked forever once the render goroutine had already exited via ctx.Done() (e.g. after SIGTERM), since it sent on an unbuffered channel with nobody left to receive. Closing the channel instead of sending never blocks, but a bare close() panics the moment ttyWriter runs a second Start/Done cycle on the same shared bus - which already happens today (run's nested create/start, publish's nested push, rm --stop's sequential stop then remove). Start now hands out a fresh, once-only-closable signal each cycle instead of a single shared channel, and run/publish stop opening a redundant nested bracket around calls already inside one. Signed-off-by: Guillaume Lours --- cmd/display/tty.go | 61 +++++++++++++++++++++---- cmd/display/tty_test.go | 98 ++++++++++++++++++++++++++++++++++++++++- pkg/compose/publish.go | 6 ++- pkg/compose/remove.go | 2 + pkg/compose/run.go | 11 +++-- 5 files changed, 164 insertions(+), 14 deletions(-) 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 }