fix(e2e): cap scenario concurrency and narrow the transient classifier

.github/workflows/build-cli.yaml

@@ -88,3 +88,19 @@ jobs:
         # the product. Serial execution removes that contention; the longer
         # 60m timeout covers the resulting slower wall-clock.
         run: go test -v -parallel 1 -timeout 60m ./...
+
+      # On a retry-exhausted scenario the harness writes the last attempt's raw
+      # act stdout/stderr to e2e/_artifacts/<scenario>-attempt<N>.log so the
+      # stack-trace origin survives the CI log retention window. Upload it on
+      # every run (always()) so the evidence is recoverable whether the job
+      # failed or a flake was absorbed. The directory may not exist when no
+      # scenario exhausted its retries; if-no-files-found: ignore keeps that a
+      # clean no-op rather than a warning.
+      - name: Upload e2e crash evidence
+        if: always()
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: e2e-crash-evidence
+          path: e2e/_artifacts/
+          if-no-files-found: ignore
+          retention-days: 14

e2e/harness/act.go

@@ -389,6 +389,17 @@ func (a *ActRunner) RunWorkflowFromRepo(ctx context.Context, opts RunOpts) (*Ext
 // workflow (e.g. a missing orchestrate.yaml). Without this, such a run
 // masqueraded as Conclusion="success" with 0 jobs. A missing workflow showing
 // up as a green-but-empty scenario (#25).
+//
+// Transient classification is DELIBERATELY narrow. A non-zero act exit is only
+// tagged transient (ExecError true, retryable) when the raw output carries a
+// NAMED host-saturation signature (see detectInfraSaturation): docker
+// address-pool exhaustion (#125), disk-full, OOM, a gitea connection
+// reset/refused (#121), or "Cannot connect to the Docker daemon". Those are
+// genuine, externally-imposed transients a clean-slate retry can clear. Every
+// OTHER non-zero exit with no job failure and no crash signature is now a
+// DETERMINISTIC failure: it fails on the first attempt and surfaces a bug
+// instead of being silently retried away. This closes the masking bucket that
+// let self-inflicted contention burn the whole retry budget.
 func normalizeWorkflowResult(result *ExtendedWorkflowResult, workflowPath string, exitCode int) {
 	if exitCode != 0 {
 		// Classify BEFORE forcing the conclusion to "failure": hasJobFailure
@@ -414,10 +425,16 @@ func normalizeWorkflowResult(result *ExtendedWorkflowResult, workflowPath string
 			result.Error = "workflow execution failed"
 		default:
 			// A non-zero exit with no parsed job failure and no crash signature
-			// is a benign act/docker transport or exec hiccup: tag it transient
-			// so the scenario runner may retry from a clean slate.
-			result.ExecError = true
-			result.Error = "workflow execution failed"
+			// is transient ONLY when a named host-saturation signature is in the
+			// raw output; otherwise it is a deterministic real failure that must
+			// surface (not be masked by a retry).
+			if saturated, reason := detectInfraSaturation(infraLogs(result)); saturated {
+				result.ExecError = true
+				result.Error = fmt.Sprintf("infra saturation: %s", reason)
+			} else {
+				result.ExecError = false
+				result.Error = "workflow execution failed"
+			}
 		}
 		// A non-zero exit is always a failure; set the conclusion after
 		// classification so hasJobFailure above saw the pre-exit conclusion.

e2e/harness/concurrency.go

@@ -0,0 +1,101 @@
+package harness
+
+import (
+	"context"
+	"os"
+	"runtime"
+	"strconv"
+	"sync"
+)
+
+// maxConcurrentEnv is the environment variable that overrides the computed
+// scenario-concurrency cap. A positive integer pins the cap exactly (clamped to
+// at least 1); any other value (unset, empty, zero, negative, non-numeric) falls
+// back to the host-derived default. This lets a constrained or oversized runner
+// tune the cap without a code change, e.g. E2E_MAX_CONCURRENT=1 to force fully
+// serial heavy-scenario setup.
+const maxConcurrentEnv = "E2E_MAX_CONCURRENT"
+
+// scenarioConcurrency is a process-wide semaphore that bounds how many scenarios
+// stand up their docker/gitea/act stack CONCURRENTLY, independent of
+// `go test -parallel`. Each scenario stands up its own docker network + gitea
+// container + act container, and act spawns nested job containers per workflow
+// run; the heaviest (hotfix-family) scenarios hold many concurrent network
+// endpoints, containers, and memory. Run enough of them at once and they
+// oversubscribe the docker address pool (#125), host RAM, or gitea (#121) - the
+// self-inflicted contention that the old five-attempt retry budget masked.
+// `go test -parallel` alone cannot bound this because it counts test functions,
+// not the heavyweight infra each one provisions. This semaphore caps the real
+// resource pressure so heavy scenarios serialize against each other under load
+// and infra-saturation transients stop happening in the first place.
+var scenarioConcurrency = make(chan struct{}, computeScenarioConcurrency())
+
+// computeScenarioConcurrency derives the concurrency cap. An E2E_MAX_CONCURRENT
+// override (positive integer) wins. Otherwise the cap is host-derived and
+// deliberately conservative, because each admitted scenario provisions a full
+// docker/gitea/act stack plus nested job containers:
+//
+//   - From CPUs: floor(NumCPU/2), so a 4-core runner admits 2 and an 8-core
+//     runner admits 4. Halving leaves headroom for the docker daemon, gitea, and
+//     act's own orchestration rather than saturating every core with job
+//     containers.
+//   - From memory: floor(totalGiB/3), budgeting ~3 GiB per concurrent stack
+//     (gitea + act + nested job containers). On an 8 GiB runner that is 2.
+//   - The cap is the MIN of the two (the binding resource), floored at 1 so the
+//     suite always makes progress even on a tiny host. Total memory is read
+//     best-effort; when it is unknown the CPU bound alone applies.
+//
+// On the diagnosis's 4-core / ~8 GiB reference runner this yields min(2, 2) = 2,
+// which keeps the heavy hotfix scenarios from oversubscribing docker/gitea.
+func computeScenarioConcurrency() int {
+	if n, ok := parsePositiveInt(os.Getenv(maxConcurrentEnv)); ok {
+		return n
+	}
+
+	limit := runtime.NumCPU() / 2
+	if limit < 1 {
+		limit = 1
+	}
+
+	if gib := totalMemoryGiB(); gib > 0 {
+		memLimit := gib / 3
+		if memLimit < 1 {
+			memLimit = 1
+		}
+		if memLimit < limit {
+			limit = memLimit
+		}
+	}
+
+	return limit
+}
+
+// parsePositiveInt parses s as a positive integer, returning (n, true) only for
+// a strictly-positive value. Empty, non-numeric, zero, and negative inputs
+// return (0, false) so the caller falls back to the host-derived default.
+func parsePositiveInt(s string) (int, bool) {
+	if s == "" {
+		return 0, false
+	}
+	n, err := strconv.Atoi(s)
+	if err != nil || n < 1 {
+		return 0, false
+	}
+	return n, true
+}
+
+// acquireScenarioSlot blocks until a scenario-concurrency slot is free (or ctx
+// is cancelled), then returns a release func the caller must invoke (via defer)
+// to return the slot. Acquiring before SetupInfra and releasing after Cleanup
+// bounds how many docker/gitea/act stacks exist at once. On ctx cancellation it
+// returns a non-nil error and a no-op release, so a cancelled caller neither
+// blocks nor double-releases.
+func acquireScenarioSlot(ctx context.Context) (release func(), err error) {
+	select {
+	case scenarioConcurrency <- struct{}{}:
+		var once sync.Once
+		return func() { once.Do(func() { <-scenarioConcurrency }) }, nil
+	case <-ctx.Done():
+		return func() {}, ctx.Err()
+	}
+}

e2e/harness/concurrency_test.go

@@ -0,0 +1,142 @@
+package harness
+
+import (
+	"context"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestParsePositiveInt(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name   string
+		in     string
+		wantN  int
+		wantOK bool
+	}{
+		{name: "empty falls back", in: "", wantN: 0, wantOK: false},
+		{name: "zero falls back", in: "0", wantN: 0, wantOK: false},
+		{name: "negative falls back", in: "-3", wantN: 0, wantOK: false},
+		{name: "non-numeric falls back", in: "abc", wantN: 0, wantOK: false},
+		{name: "one is honored", in: "1", wantN: 1, wantOK: true},
+		{name: "larger value is honored", in: "6", wantN: 6, wantOK: true},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			n, ok := parsePositiveInt(tt.in)
+			if n != tt.wantN || ok != tt.wantOK {
+				t.Fatalf("parsePositiveInt(%q) = (%d,%v), want (%d,%v)", tt.in, n, ok, tt.wantN, tt.wantOK)
+			}
+		})
+	}
+}
+
+// TestComputeScenarioConcurrency_EnvOverride pins the cap via E2E_MAX_CONCURRENT
+// and confirms a positive override wins over the host-derived default while an
+// invalid one falls back to a sane (>=1) host value.
+func TestComputeScenarioConcurrency_EnvOverride(t *testing.T) {
+	t.Setenv(maxConcurrentEnv, "3")
+	if got := computeScenarioConcurrency(); got != 3 {
+		t.Fatalf("computeScenarioConcurrency() with override=3 = %d, want 3", got)
+	}
+
+	t.Setenv(maxConcurrentEnv, "bogus")
+	if got := computeScenarioConcurrency(); got < 1 {
+		t.Fatalf("computeScenarioConcurrency() with invalid override = %d, want >= 1", got)
+	}
+}
+
+// TestComputeScenarioConcurrency_HostDefault confirms the default is always at
+// least 1 (the suite must make progress on any host) when no override is set.
+func TestComputeScenarioConcurrency_HostDefault(t *testing.T) {
+	t.Setenv(maxConcurrentEnv, "")
+	if got := computeScenarioConcurrency(); got < 1 {
+		t.Fatalf("computeScenarioConcurrency() host default = %d, want >= 1", got)
+	}
+}
+
+// TestAcquireScenarioSlot_BoundsConcurrency proves the semaphore never admits
+// more than its capacity at once: with a cap of 2 and many contending
+// goroutines, the observed in-flight count never exceeds 2.
+func TestAcquireScenarioSlot_BoundsConcurrency(t *testing.T) {
+	const capLimit = 2
+	// Swap the package semaphore for a capacity-controlled one for this test,
+	// then restore it so other tests see the real cap.
+	orig := scenarioConcurrency
+	scenarioConcurrency = make(chan struct{}, capLimit)
+	t.Cleanup(func() { scenarioConcurrency = orig })
+
+	var (
+		inFlight int32
+		maxSeen  int32
+		wg       sync.WaitGroup
+	)
+	for i := 0; i < 20; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			release, err := acquireScenarioSlot(context.Background())
+			if err != nil {
+				t.Errorf("acquireScenarioSlot returned error: %v", err)
+				return
+			}
+			defer release()
+
+			cur := atomic.AddInt32(&inFlight, 1)
+			for {
+				prev := atomic.LoadInt32(&maxSeen)
+				if cur <= prev || atomic.CompareAndSwapInt32(&maxSeen, prev, cur) {
+					break
+				}
+			}
+			time.Sleep(2 * time.Millisecond)
+			atomic.AddInt32(&inFlight, -1)
+		}()
+	}
+	wg.Wait()
+
+	if maxSeen > capLimit {
+		t.Fatalf("max concurrent slots = %d, want <= %d", maxSeen, capLimit)
+	}
+	// All slots must have been returned: a fresh acquire must not block.
+	release, err := acquireScenarioSlot(context.Background())
+	if err != nil {
+		t.Fatalf("post-run acquire errored: %v", err)
+	}
+	release()
+}
+
+// TestAcquireScenarioSlot_ContextCancelled confirms a caller cancelled while the
+// semaphore is full gets a non-nil error and a no-op release (never blocks,
+// never frees a slot it does not hold).
+func TestAcquireScenarioSlot_ContextCancelled(t *testing.T) {
+	orig := scenarioConcurrency
+	scenarioConcurrency = make(chan struct{}, 1)
+	t.Cleanup(func() { scenarioConcurrency = orig })
+
+	// Fill the only slot.
+	hold, err := acquireScenarioSlot(context.Background())
+	if err != nil {
+		t.Fatalf("initial acquire errored: %v", err)
+	}
+	defer hold()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	release, err := acquireScenarioSlot(ctx)
+	if err == nil {
+		t.Fatalf("acquireScenarioSlot with cancelled ctx returned nil error")
+	}
+	// The returned release must be a safe no-op (must not free the held slot).
+	release()
+	select {
+	case scenarioConcurrency <- struct{}{}:
+		t.Fatalf("no-op release freed a slot it did not own (semaphore under-counted)")
+	default:
+	}
+}

e2e/harness/crash_test.go

@@ -241,12 +241,16 @@ func TestNormalizeWorkflowResult_CrashIsNotTransient(t *testing.T) {
 	}
 }
 
-// TestNormalizeWorkflowResult_ActOnlyDumpIsTransient is the sibling guarantee:
-// a non-zero act exit whose logs carry only an act/docker dump (no cascade
-// frame) is NOT flagged Crashed by the parser, so it falls through to the
-// transient ExecError path and the scenario runner retries it as an infra
-// flake.
-func TestNormalizeWorkflowResult_ActOnlyDumpIsTransient(t *testing.T) {
+// TestNormalizeWorkflowResult_ActOnlyDumpIsDeterministic is the sibling
+// guarantee to the crash test, updated for the narrowed classifier: a non-zero
+// act exit whose logs carry only an act/docker dump (no cascade frame) is still
+// correctly NOT flagged Crashed by the parser - the #146/#147 crash-frame logic
+// is intact - but it is NO LONGER auto-retried. With no cascade crash frame and
+// no named infra-saturation signature, it is a deterministic failure that must
+// surface on the first attempt rather than being masked by the old catch-all
+// transient bucket. An infra signature in the SAME shape of dump is what makes
+// it transient again (covered by the infra subtest below).
+func TestNormalizeWorkflowResult_ActOnlyDumpIsDeterministic(t *testing.T) {
 	t.Parallel()
 
 	for _, tt := range []struct {
@@ -265,18 +269,40 @@ func TestNormalizeWorkflowResult_ActOnlyDumpIsTransient(t *testing.T) {
 			if err != nil {
 				t.Fatalf("ParseActOutput returned error: %v", err)
 			}
+			if result.Crashed {
+				t.Fatalf("an act-only dump must NOT be flagged Crashed (no cascade frame)")
+			}
 			normalizeWorkflowResult(result, ".github/workflows/orchestrate.yaml", 2)
 
 			if result.Conclusion != "failure" {
 				t.Fatalf("Conclusion = %q, want failure", result.Conclusion)
 			}
-			if !result.ExecError {
-				t.Fatalf("an act-only dump must be tagged transient (ExecError) so it is retried")
+			if result.ExecError {
+				t.Fatalf("an act-only dump with no infra signature must be deterministic (not retried)")
 			}
 			failErr := workflowFailureError("orchestrate", result)
-			if !IsTransientWorkflowError(failErr) {
-				t.Fatalf("an act-only dump failure must be classified transient: %v", failErr)
+			if IsTransientWorkflowError(failErr) {
+				t.Fatalf("an act-only dump with no infra signature must NOT be classified transient: %v", failErr)
 			}
 		})
 	}
+
+	// The same act-only dump shape, but carrying a named infra-saturation
+	// signature, IS a genuine retryable transient - proving the gate is the
+	// signature, not the dump.
+	t.Run("act dump with infra signature is transient", func(t *testing.T) {
+		t.Parallel()
+		logs := bareActGoroutineDump + "\nError response from daemon: all predefined address pools have been fully subnetted\n"
+		result, err := ParseActOutput(logs)
+		if err != nil {
+			t.Fatalf("ParseActOutput returned error: %v", err)
+		}
+		normalizeWorkflowResult(result, ".github/workflows/orchestrate.yaml", 2)
+		if !result.ExecError {
+			t.Fatalf("an act dump carrying an infra signature must be transient (ExecError)")
+		}
+		if !IsTransientWorkflowError(workflowFailureError("orchestrate", result)) {
+			t.Fatalf("an act dump carrying an infra signature must be classified transient")
+		}
+	})
 }

e2e/harness/memory_darwin.go

@@ -0,0 +1,26 @@
+//go:build darwin
+
+package harness
+
+import (
+	"os/exec"
+	"strconv"
+	"strings"
+)
+
+// totalMemoryGiB returns the host's total physical memory in whole GiB on
+// darwin, read best-effort via `sysctl -n hw.memsize` (bytes). It returns 0 when
+// the value cannot be read or parsed so the caller falls back to the CPU-only
+// concurrency bound. Reading via sysctl avoids promoting a memory-introspection
+// dependency to direct for what is a best-effort sizing hint.
+func totalMemoryGiB() int {
+	out, err := exec.Command("sysctl", "-n", "hw.memsize").Output()
+	if err != nil {
+		return 0
+	}
+	bytes, err := strconv.ParseUint(strings.TrimSpace(string(out)), 10, 64)
+	if err != nil {
+		return 0
+	}
+	return int(bytes / (1 << 30))
+}