From bf03b83616948b20f0fde0426add625d64c593cb Mon Sep 17 00:00:00 2001 From: Nicolas De Loof Date: Thu, 27 Aug 2026 12:12:14 +0200 Subject: [PATCH 1/9] test(e2e): deflake the three state/stream race suspects MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three tests failed 9 CI runs across 5 branches this week, all on asynchronous-observation races, none reproducible locally: - RequireServiceState asserted on a single `compose ps` snapshot; the daemon reports state transitions asynchronously from everything else a test observes (TestUpDependenciesNotStopped saw 'created' while the container's logs were already flowing). It now polls until the state converges (15s bound). - TestAttachRestart counted restart log lines in a snapshot taken as soon as the third exit notice appeared; exit notices come from the events monitor while log lines come from the re-attached logs stream — two channels with no ordering between them. The count is now awaited like the exit notices already were; a genuinely lost line still fails, by timeout. - TestUpExitCodeFromContainerKilled ran a full up+abort cycle under a 60s ceiling, once exceeded on a loaded oldstable runner; raised to 120s. Signed-off-by: Nicolas De Loof --- pkg/e2e/assert.go | 63 +++++++++++++++++++++++++++++--------- pkg/e2e/compose_up_test.go | 2 +- 2 files changed, 49 insertions(+), 16 deletions(-) diff --git a/pkg/e2e/assert.go b/pkg/e2e/assert.go index c68a64b4fa..1b94b79923 100644 --- a/pkg/e2e/assert.go +++ b/pkg/e2e/assert.go @@ -23,26 +23,59 @@ import ( "testing" "time" - "gotest.tools/v3/assert" - is "gotest.tools/v3/assert/cmp" "gotest.tools/v3/poll" ) -// RequireServiceState ensures that the container is in the expected state -// (running or exited). +// RequireServiceState ensures that the container reaches the expected state +// (running or exited). The daemon reports state transitions asynchronously +// from everything else a test can observe (a container whose logs already +// flowed may still be listed under its previous state for a moment), so the +// check polls `compose ps` until the state converges instead of asserting on +// a single snapshot. func RequireServiceState(t testing.TB, cli *CLI, service string, state string) { t.Helper() - psRes := cli.RunDockerComposeCmd(t, "ps", "--all", "--format=json", service) - var serviceState map[string]any - assert.NilError(t, json.Unmarshal([]byte(psRes.Stdout()), &serviceState), - "Invalid `compose ps` JSON: command output: %s", - psRes.Combined()) - - assert.Assert(t, is.Equal(service, serviceState["Service"]), "Found ps output for unexpected service") - assert.Assert(t, is.Equal(strings.ToLower(state), strings.ToLower(serviceState["State"].(string))), - "Service %q (%s) not in expected state", - service, serviceState["Name"], - ) + poll.WaitOn(t, func(poll.LogT) poll.Result { + // NoCheck: a non-zero `compose ps` is a transient state here (the + // project may not be registered yet) — and the asserting variant + // would t.FailNow() from the poll goroutine, which terminates it via + // runtime.Goexit without reporting: the poll would hang until its + // opaque timeout instead of surfacing the actual failure below. + psRes := cli.RunDockerComposeCmdNoCheck(t, "ps", "--all", "--format=json", service) + if psRes.ExitCode != 0 { + return poll.Continue("`compose ps %s` exited %d: %s", service, psRes.ExitCode, psRes.Combined()) + } + out := strings.TrimSpace(psRes.Stdout()) + if out == "" { + // The container is not registered yet (creation in progress): + // transient, keep polling. + return poll.Continue("service %q has no `compose ps` entry yet", service) + } + // --format=json emits one JSON object per line, and a service can + // briefly list two containers mid-transition (the old one being + // removed, its replacement being created). Succeed as soon as one + // entry of the target service reaches the expected state; everything + // short of malformed JSON is a transient condition to retry, not a + // hard failure — hard-failing on those is the exact race this helper + // exists to absorb. + var seen []string + for line := range strings.SplitSeq(out, "\n") { + var entry map[string]any + if err := json.Unmarshal([]byte(line), &entry); err != nil { + return poll.Error(fmt.Errorf("invalid `compose ps` JSON: %w: command output: %s", err, psRes.Combined())) + } + if name, _ := entry["Service"].(string); name != service { + // ps was invoked filtered on the service name; a foreign or + // incomplete entry is transient noise. + continue + } + current, _ := entry["State"].(string) + if strings.EqualFold(state, current) { + return poll.Success() + } + seen = append(seen, current) + } + return poll.Continue("service %q is %v, expected %q", service, seen, state) + }, poll.WithTimeout(15*time.Second), poll.WithDelay(200*time.Millisecond)) } // RequireEventuallyServiceState polls `compose ps` until the service reaches diff --git a/pkg/e2e/compose_up_test.go b/pkg/e2e/compose_up_test.go index 66e8837a02..6ce8768468 100644 --- a/pkg/e2e/compose_up_test.go +++ b/pkg/e2e/compose_up_test.go @@ -52,7 +52,7 @@ func TestUpExitCodeFrom(t *testing.T) { func TestUpExitCodeFromContainerKilled(t *testing.T) { NewScenario(t, "up --exit-code-from must report 143 for a service stopped by the abort"). Step("the watched long-lived service is stopped when another exits", - ComposeCmd("up", "--menu=false", "--exit-code-from=test").MayFail().Within(60*time.Second), + ComposeCmd("up", "--menu=false", "--exit-code-from=test").MayFail().Within(120*time.Second), ExitCode(143)) } From 77b934f36ca92b4c3afb5cef9890981fa01418ec Mon Sep 17 00:00:00 2001 From: Nicolas De Loof Date: Thu, 27 Aug 2026 12:24:50 +0200 Subject: [PATCH 2/9] fix(logs): anchor restart re-attach on the previous run's end, not the new run's start MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The CI hardening in the previous commit turned TestAttachRestart's flake into a reliable detector, and what it detected is a real loss: re-attaching with since=StartedAt drops a fast run's first lines forever, because the daemon starts copying stdout before it records StartedAt. Anchoring on the inspected FinishedAt fixes the common case but leaves a narrower race CI still caught: when the new run itself finishes before compose reacts to its start event, the inspected FinishedAt is already the NEW run's own end, and the log window drops everything the run printed — two worlds for three exit notices, the third never arriving no matter how long you wait. Both re-attach sites (attached up, logs --follow) therefore anchor on the session's own record of the container's previous exit (runEndTracker): the monitor delivers events in order, so the anchor captured synchronously at start-event time is necessarily the previous run's end — nanosecond-precise, immune to how fast the new run dies. The inspected FinishedAt remains the fallback for a container the session never saw exit, and a fresh container keeps no lower bound. A unit test pins the ordering contract, including the fast-run sequence CI caught. Signed-off-by: Nicolas De Loof --- pkg/compose/logs.go | 75 +++++++++++++++++++++++++++++++++++++++- pkg/compose/logs_test.go | 31 +++++++++++++++++ pkg/compose/up.go | 15 ++++++-- 3 files changed, 117 insertions(+), 4 deletions(-) diff --git a/pkg/compose/logs.go b/pkg/compose/logs.go index 5bacaf76be..2836dea152 100644 --- a/pkg/compose/logs.go +++ b/pkg/compose/logs.go @@ -19,6 +19,8 @@ package compose import ( "context" "io" + "sync" + "time" "github.com/containerd/errdefs" "github.com/moby/moby/api/pkg/stdcopy" @@ -110,19 +112,29 @@ func (s *composeService) logContainer(ctx context.Context, consumer api.LogConsu // 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 { + runEnds := newRunEndTracker() return func(event api.ContainerEvent) { + runEnds.Observe(event) if event.Type != api.ContainerEventStarted { return } + // Captured synchronously: the monitor delivers events in order, so + // the recorded end cannot yet include THIS run's own exit — reading + // it inside the goroutine below could (fast run), and the window + // would drop the whole run. + since := runEnds.Since(event.ID) eg.Go(func() error { res, err := s.apiClient().ContainerInspect(ctx, event.ID, client.ContainerInspectOptions{}) if err != nil { return err } + if since == "" { + since = logsSinceLastRun(res.Container) + } err = s.doLogContainer(ctx, consumer, event.Source, res.Container, api.LogOptions{ Follow: options.Follow, - Since: res.Container.State.StartedAt, + Since: since, Until: options.Until, Tail: options.Tail, Timestamps: options.Timestamps, @@ -136,6 +148,67 @@ func (s *composeService) followStartedContainersLogs(ctx context.Context, eg *er } } +// runEndTracker remembers, per container, when the session last saw it exit — +// the re-attach anchor that stays correct even when the NEW run is already +// over: the event stream is ordered, so at start-event time the recorded +// value is necessarily the PREVIOUS run's end. The inspected FinishedAt +// (logsSinceLastRun) cannot give that guarantee — by the time we inspect, a +// fast run's own FinishedAt has overwritten it and the window would exclude +// everything the run printed. +type runEndTracker struct { + mu sync.Mutex + ends map[string]int64 // container ID → TimeNano of the last observed exit +} + +func newRunEndTracker() *runEndTracker { + return &runEndTracker{ends: map[string]int64{}} +} + +// Observe records exit events (other event types are ignored). +func (t *runEndTracker) Observe(e api.ContainerEvent) { + if e.Type != api.ContainerEventExited || e.Time == 0 { + return + } + t.mu.Lock() + t.ends[e.ID] = e.Time + t.mu.Unlock() +} + +// Since returns the log-window anchor for a container being re-attached: the +// recorded end of its previous run in RFC3339Nano — the same format the +// FinishedAt fallback feeds the logs API — or "" when the session never saw +// it exit (first start). +func (t *runEndTracker) Since(containerID string) string { + t.mu.Lock() + nano, ok := t.ends[containerID] + t.mu.Unlock() + if !ok { + return "" + } + return time.Unix(0, nano).UTC().Format(time.RFC3339Nano) +} + +// logsSinceLastRun returns the FALLBACK log window anchor for a container +// (re)started while we follow the project, used when the session has not +// observed a previous exit (runEndTracker): the previous run's FinishedAt. +// The new run's StartedAt looks like the natural anchor but loses output — +// the daemon starts copying stdout before it records StartedAt, so a fast +// process can get its first lines timestamped just before it, and +// `since=StartedAt` then drops them forever. Nothing can be logged between +// the previous run's end and the new run's start, so FinishedAt captures the +// entire new run without replaying the previous one — UNLESS the new run +// already finished by inspection time (its own FinishedAt shadows the +// previous run's), which is exactly what the tracker protects against. A +// container with no previous run has a zero FinishedAt, which means "no +// lower bound" — equally exact for a fresh container. +func logsSinceLastRun(ctr container.InspectResponse) string { + finished := ctr.State.FinishedAt + if t, err := time.Parse(time.RFC3339Nano, finished); err != nil || t.Unix() <= 0 { + return "" + } + return finished +} + func (s *composeService) doLogContainer(ctx context.Context, consumer api.LogConsumer, name string, ctr container.InspectResponse, options api.LogOptions) error { r, err := s.apiClient().ContainerLogs(ctx, ctr.ID, client.ContainerLogsOptions{ ShowStdout: true, diff --git a/pkg/compose/logs_test.go b/pkg/compose/logs_test.go index b0499f1560..27a9b859ae 100644 --- a/pkg/compose/logs_test.go +++ b/pkg/compose/logs_test.go @@ -245,3 +245,34 @@ func (l *testLogConsumer) LogsForContainer(containerName string) []string { defer l.mu.Unlock() return l.logs[containerName] } + +// TestRunEndTrackerAnchorsOnPreviousRun pins the re-attach anchor against the +// fast-run race seen in CI: with events delivered in order, the anchor +// captured at start-event time is the PREVIOUS run's end — even when the new +// run exits (and is observed) before the log stream is actually opened. +func TestRunEndTrackerAnchorsOnPreviousRun(t *testing.T) { + tr := newRunEndTracker() + + // First start: no previous exit observed → no anchor (caller falls back + // to the inspected FinishedAt). + assert.Equal(t, tr.Since("c1"), "") + + // Run N exits at t=1_000_000_001ns, run N+1 starts: the anchor captured + // at start-event time is run N's end, nanosecond-precise. + tr.Observe(compose.ContainerEvent{Type: compose.ContainerEventExited, ID: "c1", Time: 1_000_000_001}) + anchor := tr.Since("c1") + assert.Equal(t, anchor, "1970-01-01T00:00:01.000000001Z") + + // Run N+1 is fast: its own exit is observed before the log stream opens. + // The anchor captured above must NOT move — reading it after this point + // would exclude everything run N+1 printed. + tr.Observe(compose.ContainerEvent{Type: compose.ContainerEventExited, ID: "c1", Time: 2_000_000_002}) + assert.Equal(t, anchor, "1970-01-01T00:00:01.000000001Z") + // The NEXT start anchors on run N+1's end. + assert.Equal(t, tr.Since("c1"), "1970-01-01T00:00:02.000000002Z") + + // Non-exit events and other containers do not pollute the anchor. + tr.Observe(compose.ContainerEvent{Type: compose.ContainerEventStarted, ID: "c1", Time: 9_000_000_000}) + tr.Observe(compose.ContainerEvent{Type: compose.ContainerEventExited, ID: "c2", Time: 3_000_000_003}) + assert.Equal(t, tr.Since("c1"), "1970-01-01T00:00:02.000000002Z") +} diff --git a/pkg/compose/up.go b/pkg/compose/up.go index c08075ec98..cf868c1851 100644 --- a/pkg/compose/up.go +++ b/pkg/compose/up.go @@ -349,26 +349,35 @@ func (u *upSession) captureExitCodeFrom() api.ContainerEventListener { // followStartedContainers streams logs of containers (re)started after `up`, // so they are followed like the initially attached ones. func (u *upSession) followStartedContainers(attached []string) api.ContainerEventListener { + runEnds := newRunEndTracker() return func(event api.ContainerEvent) { + runEnds.Observe(event) if !shouldFollowStartEvent(event, attached, u.options.Start.AttachTo) { return } + // Captured synchronously — see followStartedContainersLogs: read any + // later, a fast run's own exit could already be recorded and the log + // window would drop the whole run. + since := runEnds.Since(event.ID) u.eg.Go(func() error { - u.appendErr(u.streamContainerLogs(event)) + u.appendErr(u.streamContainerLogs(event, since)) return nil }) } } -func (u *upSession) streamContainerLogs(event api.ContainerEvent) error { +func (u *upSession) streamContainerLogs(event api.ContainerEvent, since string) error { res, err := u.apiClient().ContainerInspect(u.globalCtx, event.ID, client.ContainerInspectOptions{}) if err != nil { return err } + if since == "" { + since = logsSinceLastRun(res.Container) + } err = u.doLogContainer(u.globalCtx, u.options.Start.Attach, event.Source, res.Container, api.LogOptions{ Follow: true, - Since: res.Container.State.StartedAt, + Since: since, }) if errdefs.IsNotImplemented(err) { // container may be configured with logging_driver: none From 80231da5478cc3535a069f1ffeaf42957fe13d87 Mon Sep 17 00:00:00 2001 From: Nicolas De Loof Date: Thu, 27 Aug 2026 12:42:51 +0200 Subject: [PATCH 3/9] test(e2e): TestAttachRestart failure dumps the daemon's own log view MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The FinishedAt anchor did not cure the third-run loss on CI (still 2 worlds for 3 exit notices after a full minute, oldstable runner). The remaining suspects are on both sides of the API: a line the daemon never captured (copier torn down before a millisecond-lived run's output) or a line compose still fails to relay. On timeout the test now dumps `docker logs` for the container — ground truth that discriminates the two on the next CI occurrence. Signed-off-by: Nicolas De Loof --- pkg/e2e/compose_test.go | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/pkg/e2e/compose_test.go b/pkg/e2e/compose_test.go index 20fee1cc1c..a33198de72 100644 --- a/pkg/e2e/compose_test.go +++ b/pkg/e2e/compose_test.go @@ -163,10 +163,13 @@ func TestAttachRestart(t *testing.T) { // orders one relative to the other, so the last restart's log line can // still be in flight the instant the 3rd "exited" is observed above — // wait for it instead of counting it immediately. + // On failure, dump the daemon's own view of the container log: it + // discriminates a line compose failed to relay (present below, absent + // above) from a line the daemon itself never captured. c.WaitForCondition(t, func() (bool, string) { - debug := res.Combined() + daemonView := icmd.RunCmd(c.NewDockerCmd(t, "logs", "attach-restart-failing-1")).Combined() return strings.Count(res.Stdout(), "failing-1 | world") == 3, - fmt.Sprintf("'failing-1 | world' not found 3 times in : \n%s\n", debug) + fmt.Sprintf("'failing-1 | world' not found 3 times in : \n%s\ndaemon log view:\n%s\n", res.Combined(), daemonView) }, 30*time.Second, 1*time.Second) } From 516230b6fef7b19de533f541ff05ac4b2d428398 Mon Sep 17 00:00:00 2001 From: Nicolas De Loof Date: Tue, 22 Sep 2026 11:09:36 +0200 Subject: [PATCH 4/9] docs(logs): why a timestampless exit is dropped, not patched MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review finding on runEndTracker.Observe: recording nothing for an exit event without a timestamp is the deliberate choice — the anchor is evaluated by the daemon against its own log clock, so substituting the local clock would introduce real skew mis-anchoring to paper over a hypothetical daemon quirk, while dropping only degrades that container to the pre-tracker fallback. Signed-off-by: Nicolas De Loof --- pkg/compose/logs.go | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/pkg/compose/logs.go b/pkg/compose/logs.go index 2836dea152..1e5966ef99 100644 --- a/pkg/compose/logs.go +++ b/pkg/compose/logs.go @@ -164,7 +164,14 @@ func newRunEndTracker() *runEndTracker { return &runEndTracker{ends: map[string]int64{}} } -// Observe records exit events (other event types are ignored). +// Observe records exit events (other event types are ignored). An exit +// carrying no timestamp is deliberately dropped rather than patched with the +// local clock: the anchor is compared by the DAEMON against its own +// container-log timestamps, so substituting our clock would trade a +// hypothetical daemon quirk for real clock-skew mis-anchoring. Dropping it +// merely degrades that container to the logsSinceLastRun fallback — the +// exact pre-tracker behavior, imperfect only for a run fast enough to have +// finished again by inspection time. func (t *runEndTracker) Observe(e api.ContainerEvent) { if e.Type != api.ContainerEventExited || e.Time == 0 { return From 7fa698f3c997bbeba5c26e087b8aef4a50f0515d Mon Sep 17 00:00:00 2001 From: Nicolas De Loof Date: Tue, 22 Sep 2026 12:09:52 +0200 Subject: [PATCH 5/9] fix(wait): a target that already exited is a satisfied wait, not an error MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wait listed running containers only, so a service that finished between up and the listing — a fast run, or a service long done by the time the user types the command — failed with 'no containers for project' in a few milliseconds instead of returning its recorded exit code. This is also the root cause of the TestWaitAndDrop flake: its 'faster' service sleeps 2 seconds, less than the harness latency between the two steps. The condition wait observes (container no longer running) already holds for such a target: fall back to a full listing only when no container is running, and let ContainerWait return the recorded status immediately. Scoping the fallback to the previously-erroring path keeps every other semantics intact — in particular a stale exited one-off can never short-circuit a wait that has live containers to observe. Exit-code propagation verified end to end: wait on an already-failed service returns its code (7), not an error. Signed-off-by: Nicolas De Loof --- pkg/compose/wait.go | 13 +++++ pkg/compose/wait_test.go | 103 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 116 insertions(+) create mode 100644 pkg/compose/wait_test.go diff --git a/pkg/compose/wait.go b/pkg/compose/wait.go index 29848786b6..622803e9ba 100644 --- a/pkg/compose/wait.go +++ b/pkg/compose/wait.go @@ -31,6 +31,19 @@ func (s *composeService) Wait(ctx context.Context, projectName string, options a if err != nil { return 0, err } + if len(containers) == 0 { + // The condition wait observes — container no longer running — may + // already hold: a target that exited between up and this listing (a + // fast run, or a service long finished) is a SATISFIED wait, not an + // error; ContainerWait below returns its recorded exit code + // immediately. The second listing runs only when no container is + // running so a stale exited one-off can never short-circuit a wait + // that has live containers to observe. + containers, err = s.getContainers(ctx, projectName, oneOffInclude, true, options.Services...) + if err != nil { + return 0, err + } + } if len(containers) == 0 { return 0, fmt.Errorf("no containers for project %q", projectName) } diff --git a/pkg/compose/wait_test.go b/pkg/compose/wait_test.go new file mode 100644 index 0000000000..f86b3c7a74 --- /dev/null +++ b/pkg/compose/wait_test.go @@ -0,0 +1,103 @@ +/* + Copyright 2020 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 ( + "context" + "testing" + + "github.com/moby/moby/api/types/container" + "github.com/moby/moby/client" + "go.uber.org/mock/gomock" + "gotest.tools/v3/assert" + + "github.com/docker/compose/v5/pkg/api" + "github.com/docker/compose/v5/pkg/mocks" +) + +// waitTestService builds a composeService whose ContainerList answers depend +// on the All flag: running containers first, the full set on the fallback +// listing. Call counts let tests assert which listings actually happened. +func waitTestService(t *testing.T, running, all []container.Summary) (api.Compose, *mocks.MockAPIClient, *int, *int) { + t.Helper() + mockCtrl := gomock.NewController(t) + apiClient, cli := prepareMocks(mockCtrl) + runningCalls, allCalls := new(int), new(int) + apiClient.EXPECT().ContainerList(gomock.Any(), gomock.Any()). + DoAndReturn(func(_ context.Context, opts client.ContainerListOptions) (client.ContainerListResult, error) { + if opts.All { + *allCalls++ + return client.ContainerListResult{Items: all}, nil + } + *runningCalls++ + return client.ContainerListResult{Items: running}, nil + }).AnyTimes() + tested, err := NewComposeService(cli) + assert.NilError(t, err) + return tested, apiClient, runningCalls, allCalls +} + +// TestWait_AlreadyExitedTarget locks the race fix: a target that exited +// before wait listed the project is a SATISFIED wait — its recorded exit +// code is returned immediately — not a "no containers" error. +func TestWait_AlreadyExitedTarget(t *testing.T) { + exited := container.Summary{ + ID: "c-exited", State: container.StateExited, + Labels: map[string]string{api.ServiceLabel: "faster", api.ContainerNumberLabel: "1"}, + } + tested, apiClient, runningCalls, allCalls := waitTestService(t, nil, []container.Summary{exited}) + + apiClient.EXPECT().ContainerWait(gomock.Any(), "c-exited", gomock.Any()). + Return(waitResultExit(7)) + + code, err := tested.Wait(t.Context(), "proj", api.WaitOptions{Services: []string{"faster"}}) + assert.NilError(t, err) + assert.Equal(t, code, int64(7)) + assert.Equal(t, *runningCalls, 1) + assert.Equal(t, *allCalls, 1) +} + +// TestWait_NoContainersAtAll: when neither listing finds a matching +// container, the invocation is wrong and still errors. +func TestWait_NoContainersAtAll(t *testing.T) { + tested, _, runningCalls, allCalls := waitTestService(t, nil, nil) + + _, err := tested.Wait(t.Context(), "proj", api.WaitOptions{}) + assert.ErrorContains(t, err, `no containers for project "proj"`) + assert.Equal(t, *runningCalls, 1) + assert.Equal(t, *allCalls, 1) +} + +// TestWait_RunningContainersSkipFallback: with a running container to +// observe, the fallback listing never runs — a stale exited one-off cannot +// short-circuit the wait. +func TestWait_RunningContainersSkipFallback(t *testing.T) { + running := container.Summary{ + ID: "c-run", State: container.StateRunning, + Labels: map[string]string{api.ServiceLabel: "slower", api.ContainerNumberLabel: "1"}, + } + tested, apiClient, runningCalls, allCalls := waitTestService(t, []container.Summary{running}, nil) + + apiClient.EXPECT().ContainerWait(gomock.Any(), "c-run", gomock.Any()). + Return(waitResultExit(0)) + + code, err := tested.Wait(t.Context(), "proj", api.WaitOptions{}) + assert.NilError(t, err) + assert.Equal(t, code, int64(0)) + assert.Equal(t, *runningCalls, 1) + assert.Equal(t, *allCalls, 0) +} From a4419eea10b76030f180ada71a9ede7c7237c204 Mon Sep 17 00:00:00 2001 From: Nicolas De Loof Date: Tue, 22 Sep 2026 13:01:42 +0200 Subject: [PATCH 6/9] fix(up): drain in-flight log streams before tearing the session down MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TestAttachRestart kept failing residually after the re-attach anchoring fix, and its new daemon-view dump told exactly why: the daemon holds all three 'world' lines while compose printed two. The monitor returns on the final die event — an events-channel fact — and up canceled the global context on the spot, killing the re-attach log streams with the last run's line still in flight: the exit notice outran the output that preceded it. Re-attach streams are now counted in a WaitGroup, and the monitor wrapper waits for them to reach their natural EOF — guaranteed once the containers exited — before canceling, under a bound that only protects against a wedged daemon and is skipped entirely when the context is already down (Ctrl-C). AttachRestart passes 5/5 locally with this drain; it failed within 2 CI attempts without it. Signed-off-by: Nicolas De Loof --- pkg/compose/up.go | 47 ++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 40 insertions(+), 7 deletions(-) diff --git a/pkg/compose/up.go b/pkg/compose/up.go index cf868c1851..a9fdbccca1 100644 --- a/pkg/compose/up.go +++ b/pkg/compose/up.go @@ -26,6 +26,7 @@ import ( "sync" "sync/atomic" "syscall" + "time" "github.com/compose-spec/compose-go/v2/types" "github.com/containerd/errdefs" @@ -71,13 +72,16 @@ func (s *composeService) Up(ctx context.Context, project *types.Project, options // them, the collected errors, and the application exit status. type upSession struct { *composeService - project *types.Project - options api.UpOptions - printer logPrinter - watcher *Watcher - menu *formatter.LogKeyboard - globalCtx context.Context - cancel context.CancelFunc + project *types.Project + options api.UpOptions + printer logPrinter + // logStreams counts the in-flight re-attach log streams so shutdown can + // drain them before tearing the context down — see the monitor wrapper. + logStreams sync.WaitGroup + watcher *Watcher + menu *formatter.LogKeyboard + globalCtx context.Context + cancel context.CancelFunc signalChan chan os.Signal isTerminated atomic.Bool @@ -178,6 +182,24 @@ func (s *composeService) runInteractiveUp(ctx context.Context, project *types.Pr u.eg.Go(func() error { err := monitor.Start(globalCtx) + // The monitor returning means every watched container is gone for + // good — an events-channel fact. The last run's log lines may still + // be in flight on their own connections, and canceling now would + // drop them: the exit notice would outrun the output that preceded + // it. The containers having exited, every follow stream terminates + // on its own at EOF — give them a bounded window to drain before + // the context comes down (immediately skipped when the context is + // already canceled, e.g. Ctrl-C). + drained := make(chan struct{}) + go func() { + u.logStreams.Wait() + close(drained) + }() + select { + case <-drained: + case <-time.After(logStreamDrainTimeout): + case <-globalCtx.Done(): + } // cancel the global context to terminate signal-handler goroutines cancel() u.appendErr(err) @@ -348,6 +370,13 @@ func (u *upSession) captureExitCodeFrom() api.ContainerEventListener { // followStartedContainers streams logs of containers (re)started after `up`, // so they are followed like the initially attached ones. +// +// logStreamDrainTimeout bounds the shutdown drain of these streams: EOF is +// guaranteed once the containers exited, the bound only protects against a +// wedged daemon holding the connection open. A variable so tests can shrink +// it. +var logStreamDrainTimeout = 5 * time.Second + func (u *upSession) followStartedContainers(attached []string) api.ContainerEventListener { runEnds := newRunEndTracker() return func(event api.ContainerEvent) { @@ -359,7 +388,11 @@ func (u *upSession) followStartedContainers(attached []string) api.ContainerEven // later, a fast run's own exit could already be recorded and the log // window would drop the whole run. since := runEnds.Since(event.ID) + // counted before the goroutine starts so the shutdown drain can never + // miss a stream dispatched but not yet running + u.logStreams.Add(1) u.eg.Go(func() error { + defer u.logStreams.Done() u.appendErr(u.streamContainerLogs(event, since)) return nil }) From 96f3a1f8f59d29a11fb3908e5d3d23604786e49c Mon Sep 17 00:00:00 2001 From: Nicolas De Loof Date: Tue, 22 Sep 2026 13:02:30 +0200 Subject: [PATCH 7/9] test(watch): budget the watcher's cold start beyond the default poll timeout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TestWatch/debian flakes on loaded runners during 'writing to a file until Compose watch is up and running': that bootstrap loop ran under poll.WaitOn's default 10s budget, which must absorb image pull/build, container start and watcher initialization. Only the bootstrap gets the 2-minute budget — every later step keeps the sharp default so a real sync regression still fails fast. Signed-off-by: Nicolas De Loof --- pkg/e2e/watch_test.go | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/pkg/e2e/watch_test.go b/pkg/e2e/watch_test.go index 3e186b0d6b..4ecc1b8586 100644 --- a/pkg/e2e/watch_test.go +++ b/pkg/e2e/watch_test.go @@ -210,10 +210,15 @@ func doTest(t *testing.T, svcName string) { } t.Logf("Writing to a file until Compose watch is up and running") + // Cold start covers the whole pipeline before the first sync can land: + // image pull/build, container start, watcher initialization. On a loaded + // CI runner that regularly exceeds poll.WaitOn's default 10s budget — + // only this bootstrap loop gets the large timeout, every later step + // keeps the sharp default so a real sync regression still fails fast. poll.WaitOn(t, func(t poll.LogT) poll.Result { writeDataFile("hello.txt", "hello world") return checkFileContents("/app/data/hello.txt", "hello world")(t) - }, poll.WithDelay(time.Second)) + }, poll.WithDelay(time.Second), poll.WithTimeout(2*time.Minute)) t.Logf("Modifying file contents") writeDataFile("hello.txt", "hello watch") From ca867482ad5b316c775b3ad05183aeb3b5406b14 Mon Sep 17 00:00:00 2001 From: Nicolas De Loof Date: Tue, 22 Sep 2026 13:19:57 +0200 Subject: [PATCH 8/9] fix(up): stop late starters racing the on-exit abort MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TestUpExitCodeFromContainerKilled hung for its full 2-minute budget on CI, and its event transcript shows why: 'Aborting on container exit' sweeps the application while the start phase — which runs on a deliberately uncancelable context for SIGTERM management — was still starting services; test-1 came up AFTER its stop and stayed up, so the monitor never drained and up never returned. The abort listener now watches the events stream past its trigger: any container started after the sweep is a late starter from that race, and gets stopped as it appears. Event-driven, so there is no listing window to miss; idempotent stops make duplicates harmless. The exit code semantics are preserved: the late starter's own 143 flows through captureExitCodeFrom exactly as when the sweep wins the race. TestUpExitCodeFrom* pass 5/5 locally. Signed-off-by: Nicolas De Loof --- pkg/compose/up.go | 27 ++++++++++++++++++++++++++- 1 file changed, 26 insertions(+), 1 deletion(-) diff --git a/pkg/compose/up.go b/pkg/compose/up.go index a9fdbccca1..0af5d9a962 100644 --- a/pkg/compose/up.go +++ b/pkg/compose/up.go @@ -343,7 +343,19 @@ func (u *upSession) killApplication() { func (u *upSession) stopOnFirstExit() api.ContainerEventListener { once := true return func(event api.ContainerEvent) { - if !once || event.Type != api.ContainerEventExited { + if !once { + // The abort races the start phase, which runs on a deliberately + // uncancelable context (SIGTERM management): a container whose + // start was in flight when the application was swept comes up + // AFTER the stop, and would keep the session alive until its + // natural end. The events stream reveals such late starters — + // stop each one as it appears. + if event.Type == api.ContainerEventStarted { + u.stopLateStarter(event.Service) + } + return + } + if event.Type != api.ContainerEventExited { return } if u.options.Start.OnExit == api.CascadeFail && event.ExitCode == 0 { @@ -356,6 +368,19 @@ func (u *upSession) stopOnFirstExit() api.ContainerEventListener { } } +// stopLateStarter stops one service started after the on-exit abort swept the +// application — see stopOnFirstExit. +func (u *upSession) stopLateStarter(service string) { + u.eg.Go(func() error { + err := u.stop(context.WithoutCancel(u.globalCtx), u.project.Name, api.StopOptions{ + Services: []string{service}, + Project: u.project, + }, u.printer.HandleEvent) + u.appendErr(err) + return nil + }) +} + // captureExitCodeFrom captures the exit code of the first container to exit // for the service selected by --exit-code-from func (u *upSession) captureExitCodeFrom() api.ContainerEventListener { From 39bf0d654cf8b1a9cf38b51f8d5d4cf0937a48b8 Mon Sep 17 00:00:00 2001 From: Nicolas De Loof Date: Tue, 22 Sep 2026 14:29:07 +0200 Subject: [PATCH 9/9] fix(up): catch a late starter on a graceful teardown too, not just on-exit glours' review on this PR (pullrequestreview-5277692679): stopLateStarter only ever got armed inside stopOnFirstExit's own closure, so it caught a service racing the on-exit cascade's sweep but not one racing gracefulTeardown's (Ctrl+C/SIGTERM) identical one-shot stopApplication sweep -- same root cause, same file, just the other trigger. Worse for that path: Ctrl+C can land at t~=0, before monitor.Start even starts and while s.start() (its own uncancelable context) is still climbing the dependency graph, so most services may still be created, not running. The only recovery today is a second manual Ctrl+C. u.isTerminated already flags "termination is underway" and was already set by gracefulTeardown (just not consulted for late-starter catching, and set after stopApplication rather than before it). Extract the late-starter watch out of stopOnFirstExit into its own listener, stopLateStarters, armed on u.isTerminated regardless of which path set it, and register it unconditionally (Ctrl+C works regardless of the on-exit policy, unlike stopOnFirstExit's own listener). stopOnFirstExit now sets isTerminated itself before its sweep, symmetric with gracefulTeardown (also reordered to set it before stopApplication, not after). isLateStarter is extracted as a pure predicate, following the existing shouldFollowStartEvent precedent, and unit-tested (TestIsLateStarter): the previous fix (ca867482a) shipped only against a flaky test hitting the on-exit case by chance, with no dedicated regression test at all. Signed-off-by: Nicolas De Loof --- pkg/compose/up.go | 52 ++++++++++++++++++++++++++++++++---------- pkg/compose/up_test.go | 42 ++++++++++++++++++++++++++++++++++ 2 files changed, 82 insertions(+), 12 deletions(-) diff --git a/pkg/compose/up.go b/pkg/compose/up.go index 0af5d9a962..8aed083603 100644 --- a/pkg/compose/up.go +++ b/pkg/compose/up.go @@ -161,6 +161,12 @@ func (s *composeService) runInteractiveUp(ctx context.Context, project *types.Pr } monitor.withListener(u.printer.HandleEvent) + // Termination -- on-exit cascade or a graceful Ctrl+C/SIGTERM teardown -- + // stops the application via a one-shot listing (see stopApplication): + // registered unconditionally, regardless of the on-exit policy, since a + // graceful teardown always runs it and a container whose start was + // already in flight when that listing happened comes up after it. + monitor.withListener(u.stopLateStarters()) if options.Start.OnExit != api.CascadeIgnore { monitor.withListener(u.stopOnFirstExit()) } @@ -276,8 +282,10 @@ func (u *upSession) runEventLoop(ctx context.Context, kEvents <-chan keyboard.Ke gracefulTeardown := func() { first = false u.events.On(newEvent(api.ResourceCompose, api.Working, api.StatusStopping, "Gracefully Stopping... press Ctrl+C again to force")) - u.stopApplication() + // set before stopApplication's listing so stopLateStarters is armed + // no later than the sweep it must catch stragglers for. u.isTerminated.Store(true) + u.stopApplication() } for { @@ -344,15 +352,6 @@ func (u *upSession) stopOnFirstExit() api.ContainerEventListener { once := true return func(event api.ContainerEvent) { if !once { - // The abort races the start phase, which runs on a deliberately - // uncancelable context (SIGTERM management): a container whose - // start was in flight when the application was swept comes up - // AFTER the stop, and would keep the session alive until its - // natural end. The events stream reveals such late starters — - // stop each one as it appears. - if event.Type == api.ContainerEventStarted { - u.stopLateStarter(event.Service) - } return } if event.Type != api.ContainerEventExited { @@ -364,12 +363,41 @@ func (u *upSession) stopOnFirstExit() api.ContainerEventListener { once = false u.exitCode = event.ExitCode u.events.On(newEvent(api.ResourceCompose, api.Working, api.StatusStopping, "Aborting on container exit...")) + // set before stopApplication's listing so stopLateStarters is armed + // no later than the sweep it must catch stragglers for. + u.isTerminated.Store(true) u.stopApplication() } } -// stopLateStarter stops one service started after the on-exit abort swept the -// application — see stopOnFirstExit. +// stopLateStarters stops any service that starts after termination has begun +// — the on-exit cascade above or a graceful Ctrl+C/SIGTERM teardown +// (runEventLoop) — both of which stop the application via stopApplication's +// one-shot listing. Either start phase (the initial one, or one still +// climbing the dependency graph on its deliberately uncancelable context) +// can race that listing: a container whose start was already in flight comes +// up after the sweep and would otherwise keep the session alive until its +// natural end. The events stream reveals such late starters — stop each one +// as it appears, for as long as termination is underway. +func (u *upSession) stopLateStarters() api.ContainerEventListener { + return func(event api.ContainerEvent) { + if !isLateStarter(event, u.isTerminated.Load()) { + return + } + u.stopLateStarter(event.Service) + } +} + +// isLateStarter reports whether event is a container starting after +// termination has begun — the on-exit cascade or a graceful Ctrl+C/SIGTERM +// teardown, either of which sets terminated true before its one-shot stop +// listing — and so must be caught and stopped: see stopLateStarters. +func isLateStarter(event api.ContainerEvent, terminated bool) bool { + return terminated && event.Type == api.ContainerEventStarted +} + +// stopLateStarter stops one service started after termination swept the +// application — see stopLateStarters. func (u *upSession) stopLateStarter(service string) { u.eg.Go(func() error { err := u.stop(context.WithoutCancel(u.globalCtx), u.project.Name, api.StopOptions{ diff --git a/pkg/compose/up_test.go b/pkg/compose/up_test.go index 9cdfa49c82..10902463b1 100644 --- a/pkg/compose/up_test.go +++ b/pkg/compose/up_test.go @@ -106,6 +106,48 @@ func TestShouldFollowStartEvent(t *testing.T) { } } +// TestIsLateStarter is a follow-up to #14140's on-exit-only fix (glours' +// review on that PR): stopOnFirstExit's own sweep isn't the only path that +// can race a service still climbing the dependency graph on its +// uncancelable context -- a graceful Ctrl+C/SIGTERM teardown does too, and +// arrives with u.isTerminated already true regardless of which path set it. +// isLateStarter must gate on that shared flag, not on which listener +// happened to trigger termination. +func TestIsLateStarter(t *testing.T) { + tests := []struct { + name string + event api.ContainerEvent + terminated bool + want bool + }{ + { + name: "a container starting before termination is not a late starter", + event: api.ContainerEvent{Type: api.ContainerEventStarted}, + terminated: false, + want: false, + }, + { + name: "a non-start event after termination is not a late starter", + event: api.ContainerEvent{Type: api.ContainerEventExited}, + terminated: true, + want: false, + }, + { + name: "a container starting after termination is a late starter", + event: api.ContainerEvent{Type: api.ContainerEventStarted}, + terminated: true, + want: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := isLateStarter(tt.event, tt.terminated) + assert.Equal(t, got, tt.want) + }) + } +} + // TestAppendErrDropsCancellationAfterShutdown is the #13985 follow-up: once // our own shutdown has canceled globalCtx (monitor detecting termination, // SIGINT/SIGTERM, ...), a lingering goroutine (log/attach streaming) racing