test(e2e): wait on state, not on an unsynchronized log line

pkg/e2e/assert.go

@@ -18,11 +18,14 @@ 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
@@ -41,3 +44,49 @@ func RequireServiceState(t testing.TB, cli *CLI, service string, state string) {
 		service, serviceState["Name"],
 	)
 }
+
+// RequireEventuallyServiceState polls `compose ps` until the service reaches
+// the expected state, instead of checking once.
+//
+// `compose ps` makes a fresh ContainerList call to the daemon, an entirely
+// different channel than a container's log stream: an application log line
+// already relayed to a caller is no guarantee the daemon-reported state
+// has caught up yet by the time that caller turns around and calls `ps` in a
+// new process. Callers observing readiness through a log line (or any signal
+// other than this same `ps` state) must poll here rather than check once.
+func RequireEventuallyServiceState(t testing.TB, cli *CLI, service string, state string) {
+	t.Helper()
+	check := func(poll.LogT) poll.Result {
+		psRes := cli.RunDockerComposeCmdNoCheck(t, "ps", "--all", "--format=json", service)
+		if psRes.ExitCode != 0 {
+			return poll.Continue("compose ps exited %d: %s", psRes.ExitCode, psRes.Combined())
+		}
+		// --format=json prints one JSON object per line (NDJSON), not a
+		// single object or array: a scaled service, or a transient window
+		// during recreation where the old and new containers are both
+		// listed, means more than one line for the requested service — every
+		// one of them must reach the expected state, not just the first.
+		var found bool
+		for _, line := range strings.Split(strings.TrimSpace(psRes.Stdout()), "\n") {
+			if line == "" {
+				continue
+			}
+			var entry map[string]any
+			if err := json.Unmarshal([]byte(line), &entry); err != nil {
+				return poll.Error(fmt.Errorf("invalid `compose ps` JSON line %q: %w", line, err))
+			}
+			if svc, _ := entry["Service"].(string); !strings.EqualFold(svc, service) {
+				continue
+			}
+			found = true
+			if current, _ := entry["State"].(string); !strings.EqualFold(current, state) {
+				return poll.Continue("service %q not in state %q yet (got %q): %s", service, state, current, psRes.Stdout())
+			}
+		}
+		if found {
+			return poll.Success()
+		}
+		return poll.Continue("service %q not in state %q yet: %s", service, state, psRes.Stdout())
+	}
+	poll.WaitOn(t, check, poll.WithDelay(250*time.Millisecond), poll.WithTimeout(15*time.Second))
+}

pkg/e2e/compose_test.go

@@ -157,7 +157,17 @@ 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 "exited" status above comes from the /events stream; the "world"
+	// log line comes from a separate, unsynchronized /logs?follow=1
+	// connection (pkg/compose/up.go's followStartedContainers). Nothing
+	// 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.
+	c.WaitForCondition(t, func() (bool, string) {
+		debug := res.Combined()
+		return strings.Count(res.Stdout(), "failing-1  | world") == 3,
+			fmt.Sprintf("'failing-1  | world' not found 3 times in : \n%s\n", debug)
+	}, 30*time.Second, 1*time.Second)
 }
 
 func TestInitContainer(t *testing.T) {

pkg/e2e/up_test.go

@@ -81,8 +81,13 @@ func TestUpDependenciesNotStopped(t *testing.T) {
 
 	t.Log("Waiting for containers to be in running state")
 	upOut.RequireEventuallyContains(t, "hello app")
-	RequireServiceState(t, c, "app", "running")
-	RequireServiceState(t, c, "dependency", "running")
+	// "hello app" comes from the log stream, an unsynchronized channel from
+	// the `ps` call below (a fresh ContainerList in a new process): the log
+	// line reaching this test is no guarantee the daemon-reported state has
+	// caught up yet, so poll for it instead of checking once (see
+	// RequireEventuallyServiceState).
+	RequireEventuallyServiceState(t, c, "app", "running")
+	RequireEventuallyServiceState(t, c, "dependency", "running")
 
 	t.Log("Simulating Ctrl-C")
 	assert.NilError(t, syscall.Kill(-cmd.Process.Pid, syscall.SIGINT),