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

TestAttachRestart and TestUpDependenciesNotStopped each flaked on an
assertion checked exactly once, right after an unrelated signal on a
different channel to the daemon:

- TestAttachRestart counted "failing-1  | world" immediately after a
  WaitForCondition on "failing-1 exited with code 1" reached 3. The
  exited status comes from the /events stream; the log line comes
  from a separate /logs?follow=1 connection (followStartedContainers
  in pkg/compose/up.go). 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.

- TestUpDependenciesNotStopped called RequireServiceState right after
  seeing "hello app" in the attached log stream. That line reaching
  the test is no guarantee the daemon-reported container state (read
  via a fresh `compose ps` in a brand new process) has caught up yet.

Neither is a compose bug: `ps`/ContainerList and the log stream are
independent channels with no cross-channel ordering guarantee from
the engine, and compose does not cache container state. Both tests
now poll for the actual condition instead of using a log line as a
proxy for it: TestAttachRestart gets a second, bounded
WaitForCondition on the log count, and TestUpDependenciesNotStopped
uses a new RequireEventuallyServiceState (pkg/e2e/assert.go) that
polls `compose ps` instead of checking it once.

Signed-off-by: Nicolas De Loof <[email protected]>

pkg/e2e/assert.go

@@ -20,9 +20,11 @@ import (
 	"encoding/json"
 	"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 +43,33 @@ 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()
+	var last map[string]any
+	check := func(poll.LogT) poll.Result {
+		psRes := cli.RunDockerComposeCmdNoCheck(t, "ps", "--all", "--format=json", service)
+		var serviceState map[string]any
+		if err := json.Unmarshal([]byte(psRes.Stdout()), &serviceState); err != nil {
+			return poll.Continue("invalid `compose ps` JSON: command output: %s", psRes.Combined())
+		}
+		last = serviceState
+		current, _ := serviceState["State"].(string)
+		if !strings.EqualFold(current, state) {
+			return poll.Continue("service %q not in state %q yet (got %q)", service, state, current)
+		}
+		return poll.Success()
+	}
+	poll.WaitOn(t, check, poll.WithDelay(250*time.Millisecond), poll.WithTimeout(15*time.Second))
+
+	assert.Assert(t, is.Equal(service, last["Service"]), "Found ps output for unexpected service")
+}

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