test(e2e): deflake the three state/stream race suspects

Three tests failed 9 CI runs across 5 branches this week, all on
asynchronous-observation races, none reproducible locally:

- RequireServiceState asserted on a single `compose ps` snapshot; the
  daemon reports state transitions asynchronously from everything else
  a test observes (TestUpDependenciesNotStopped saw 'created' while
  the container's logs were already flowing). It now polls until the
  state converges (15s bound).
- TestAttachRestart counted restart log lines in a snapshot taken as
  soon as the third exit notice appeared; exit notices come from the
  events monitor while log lines come from the re-attached logs
  stream — two channels with no ordering between them. The count is
  now awaited like the exit notices already were; a genuinely lost
  line still fails, by timeout.
- TestUpExitCodeFromContainerKilled ran a full up+abort cycle under a
  60s ceiling, once exceeded on a loaded oldstable runner; raised to
  120s.

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

pkg/e2e/assert.go

@@ -23,26 +23,59 @@ import (
 	"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))
 }
 
 // RequireEventuallyServiceState polls `compose ps` until the service reaches

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