fix(logs): anchor restart re-attach on the previous run's end, not the new run's start

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 <[email protected]>

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

@@ -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