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,74 @@ 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). 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
+	}
+	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

@@ -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
@@ -157,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())
 	}
@@ -178,6 +188,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)
@@ -254,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 {
@@ -321,7 +351,10 @@ 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 {
+			return
+		}
+		if event.Type != api.ContainerEventExited {
 			return
 		}
 		if u.options.Start.OnExit == api.CascadeFail && event.ExitCode == 0 {
@@ -330,10 +363,52 @@ 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()
 	}
 }
 
+// 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{
+			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 {
@@ -348,27 +423,47 @@ 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) {
+		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)
+		// 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 {
-			u.appendErr(u.streamContainerLogs(event))
+			defer u.logStreams.Done()
+			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/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