From 42a7b35f4238f94445d3e793f2a9e2bfaaf5592c Mon Sep 17 00:00:00 2001 From: Nicolas De Loof Date: Thu, 27 Aug 2026 12:12:14 +0200 Subject: [PATCH 1/3] 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 | 66 +++++++++++++++++++++++++++++--------- pkg/e2e/compose_test.go | 10 +++++- pkg/e2e/compose_up_test.go | 2 +- 3 files changed, 61 insertions(+), 17 deletions(-) diff --git a/pkg/e2e/assert.go b/pkg/e2e/assert.go index a5b973dd15..c71f7d07b3 100644 --- a/pkg/e2e/assert.go +++ b/pkg/e2e/assert.go @@ -18,26 +18,62 @@ package e2e import ( "encoding/json" + "fmt" "strings" "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)) } diff --git a/pkg/e2e/compose_test.go b/pkg/e2e/compose_test.go index 59d6cdd481..c88742adde 100644 --- a/pkg/e2e/compose_test.go +++ b/pkg/e2e/compose_test.go @@ -157,7 +157,15 @@ func TestAttachRestart(t *testing.T) { debug) }, 4*time.Minute, 2*time.Second) - assert.Equal(t, strings.Count(res.Stdout(), "failing-1 | world"), 3, res.Combined()) + // The exit notice comes from the events monitor while the log line comes + // from the logs stream compose re-attaches after each restart: two + // asynchronous channels, so the third "world" may land shortly after the + // third exit notice — wait for it rather than asserting a snapshot. + c.WaitForCondition(t, func() (bool, string) { + return strings.Count(res.Stdout(), + "failing-1 | world") == 3, fmt.Sprintf("'failing-1 | world' not found 3 times in : \n%s\n", + res.Combined()) + }, time.Minute, time.Second) } func TestInitContainer(t *testing.T) { 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 5b654baccdeba0f96e31d6b0df005de4430e1f2a Mon Sep 17 00:00:00 2001 From: Nicolas De Loof Date: Thu, 27 Aug 2026 12:24:50 +0200 Subject: [PATCH 2/3] 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 0beb363735..2c5e87869c 100644 --- a/pkg/compose/up.go +++ b/pkg/compose/up.go @@ -338,26 +338,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 27306093db76d895e459402d501df784ef44c958 Mon Sep 17 00:00:00 2001 From: Nicolas De Loof Date: Thu, 27 Aug 2026 12:42:51 +0200 Subject: [PATCH 3/3] 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 | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/pkg/e2e/compose_test.go b/pkg/e2e/compose_test.go index c88742adde..69e92b0427 100644 --- a/pkg/e2e/compose_test.go +++ b/pkg/e2e/compose_test.go @@ -161,10 +161,14 @@ func TestAttachRestart(t *testing.T) { // from the logs stream compose re-attaches after each restart: two // asynchronous channels, so the third "world" may land shortly after the // third exit notice — wait for it rather than asserting a snapshot. + // 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) { + 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", - res.Combined()) + "failing-1 | world") == 3, fmt.Sprintf("'failing-1 | world' not found 3 times in : \n%s\ndaemon log view:\n%s\n", + res.Combined(), daemonView) }, time.Minute, time.Second) }