fix(logs): restart re-attach loses a fast run's output — found by deflaking e2e

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,

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")
+}

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

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))
 }

pkg/e2e/compose_test.go

@@ -157,7 +157,19 @@ 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.
+	// 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\ndaemon log view:\n%s\n",
+				res.Combined(), daemonView)
+	}, time.Minute, time.Second)
 }
 
 func TestInitContainer(t *testing.T) {

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))
 }