fix(ci): go-test-ran ignores unrelated skips; TestRunsWait stable under load

free/ci/cmd/runs_wait.go

@@ -9,7 +9,9 @@ package main
 // Outputs: one v1 envelope (--json) or a table; exit 0, 10, 1 (unknown run,
 // usage) or 2 (infrastructure).
 // Constraints: read class; the wait never ends later than the invoke deadline
-// minus 10 s, so a document call is not killed first; each poll recovers after
+// minus 10 s, so a document call is not killed first; the first read of a
+// local run is allowed to finish (firstReadGrace) even when the budget is
+// shorter, so a real run reports its real state; each poll recovers after
 // dead coordinators (idempotent), so a killed run turns `lost` (terminal, exit
 // 0) instead of staying running; remote runs are asked through the registered
 // RemoteResolvers, which read --coordinator as NSELF_CI_COORDINATOR.
@@ -39,6 +41,10 @@ const (
 	deadlineMargin = 10 * time.Second
 	// exitStillRunning is the exit status of a wait that timed out.
 	exitStillRunning = 10
+	// firstReadGrace is the longest the first read of a local run may take when the budget is shorter
+	// (store open, dead-coordinator recovery, the read itself). It is half the deadline margin, so the
+	// wait still ends before the invoker's deadline, and it bounds a held SQLite lock.
+	firstReadGrace = deadlineMargin / 2
 )
 
 // waitPoll is the store poll interval (a var for tests).
@@ -100,11 +106,16 @@ func runWait(args []string) int {
 	// still gets a short first look), so a stalled call cannot outlive the invoke deadline.
 	wctx, cancel := context.WithDeadline(ctx, started.Add(cmp.Or(budget, 500*time.Millisecond)))
 	defer cancel()
-	src, err := openRunSource(wctx, pos[0], f.coordinator)
+	// The first local read gets at least firstReadGrace: a budget that ends before it would report the
+	// run as "unknown" although it exists. Later polls keep the budget.
+	gctx, gcancel := context.WithDeadline(ctx, started.Add(max(budget, firstReadGrace)))
+	defer gcancel()
+	src, err := openRunSourceGrace(wctx, gctx, pos[0], f.coordinator)
 	if err != nil {
 		return waitFail(ctx, wctx, err, detach.View{RunID: pos[0]}, f.json, started)
 	}
 	defer src.Close()
+	src.firstRead = gctx
 	last := detach.View{RunID: pos[0], State: "unknown"}
 	for {
 		v, err := src.View(wctx)
@@ -161,13 +172,15 @@ func printWait(asJSON bool, v detach.View, d cirun.RunView, terminal bool, waite
 // runSource reads one run's state: from a RemoteResolver when the run lives on
 // another machine, else from the local store.
 type runSource struct {
-	resolver cirun.RemoteResolver
-	ref      string
-	remote   bool
-	st       *store.Store
-	row      store.PipelineRow
-	found    bool
-	last     time.Time
+	resolver  cirun.RemoteResolver
+	ref       string
+	remote    bool
+	st        *store.Store
+	row       store.PipelineRow
+	found     bool
+	read      bool            // a local read has finished
+	firstRead context.Context // when set, bounds the first local read instead of the caller's context
+	last      time.Time
 	// Detail is the RunView of the latest View call.
 	Detail cirun.RunView
 }
@@ -176,6 +189,12 @@ type runSource struct {
 const recoverEvery = 2 * time.Second
 
 func openRunSource(ctx context.Context, ref, coordinator string) (*runSource, error) {
+	return openRunSourceGrace(ctx, ctx, ref, coordinator)
+}
+
+// openRunSourceGrace is openRunSource with a separate context for opening the local store, which
+// bounds remote lookups by ctx and the store open by storeCtx.
+func openRunSourceGrace(ctx, storeCtx context.Context, ref, coordinator string) (*runSource, error) {
 	if coordinator != "" {
 		if err := os.Setenv(detach.EnvCoordinator, coordinator); err != nil {
 			return nil, err
@@ -193,7 +212,7 @@ func openRunSource(ctx context.Context, ref, coordinator string) (*runSource, er
 		}
 	}
 	// recovery after dead coordinators runs in View, under the caller's deadline
-	st, err := openStoreCtx(ctx, runsStorePath())
+	st, err := openStoreCtx(storeCtx, runsStorePath())
 	if err != nil {
 		return nil, err
 	}
@@ -223,6 +242,9 @@ func (s *runSource) View(ctx context.Context) (detach.View, error) {
 		}
 		return detach.View{}, fmt.Errorf("run %q is no longer known to its coordinator", s.ref)
 	}
+	if !s.read && s.firstRead != nil {
+		ctx = s.firstRead
+	}
 	if time.Since(s.last) >= recoverEvery {
 		s.last = time.Now()
 		_ = cirun.Recover(ctx, s.st, "", os.Getenv("NSELF_CI_HOME"))
@@ -243,6 +265,7 @@ func (s *runSource) View(ctx context.Context) (detach.View, error) {
 		return detach.View{}, err
 	}
 	s.Detail = v
+	s.read = true
 	return toView(v), nil
 }
 

free/ci/cmd/runs_wait_test.go

@@ -226,7 +226,7 @@ func TestRunsWait(t *testing.T) {
 	t.Run("a stalled resolver cannot outlive the budget", func(t *testing.T) {
 		runsHome(t)
 		old := cirun.Registered.Remote
-		cirun.Registered.Remote = append(append([]cirun.RemoteResolver(nil), old...), stalledRemote{})
+		cirun.Registered.Remote = append(append([]cirun.RemoteResolver(nil), old...), &stalledRemote{})
 		t.Cleanup(func() { cirun.Registered.Remote = old })
 		code, w, took := waitJSON(t, "remote-stall", "--timeout", "200ms", "--json")
 		if code != 10 || w.Data.Terminal || took > 700*time.Millisecond {
@@ -263,17 +263,16 @@ func (f *fakeRemote) Resolve(_ context.Context, ref string) (cirun.RunView, bool
 	return f.views[i], true, nil
 }
 
-// stalledRemote answers the first look at once, then blocks until its context ends.
-type stalledRemote struct{}
+// stalledRemote answers the first look at once, then blocks until its context ends. The look count
+// belongs to the value, so every test (and every -count repeat) starts with a first look that answers.
+type stalledRemote struct{ looks int }
 
-var stalls int
-
-func (stalledRemote) Resolve(ctx context.Context, ref string) (cirun.RunView, bool, error) {
+func (r *stalledRemote) Resolve(ctx context.Context, ref string) (cirun.RunView, bool, error) {
 	if ref != "remote-stall" {
 		return cirun.RunView{}, false, nil
 	}
-	stalls++
-	if stalls == 1 {
+	r.looks++
+	if r.looks == 1 {
 		return cirun.RunView{ID: ref, Status: "running"}, true, nil
 	}
 	<-ctx.Done()

scripts/ci/go-test-ran.py

@@ -1,5 +1,11 @@
 #!/usr/bin/env python3
-"""Require actual passing go test -json events for every named test prefix."""
+"""Require actual passing go test -json events for every named test prefix.
+
+Fails on: a non-JSON line, any fail event (test or package), a skipped test whose
+name starts with a named prefix, and a named prefix with no passing test.
+Skips of tests that were not named (and package-level skips such as "no test
+files") are ignored, because a package run always carries some of them.
+"""
 import json
 import sys
 
@@ -15,7 +21,9 @@ def check(stream, prefixes):
             continue
         action = event.get("Action")
         name = event.get("Test", "")
-        if action in {"skip", "fail"}:
+        if action == "fail":
+            bad = True
+        if action == "skip" and name and any(name.startswith(p) for p in prefixes):
             bad = True
         if action == "pass" and name:
             for prefix in passed:

scripts/ci/tests/test_go_test_ran.py

@@ -0,0 +1,87 @@
+"""Tests for scripts/ci/go-test-ran.py.
+
+The script reads `go test -json` events on stdin and exits 0 only when every
+named test prefix has a passing test. Unrelated skips must not fail it (a
+package run such as ./cmd/... always carries some); a failure anywhere, or a
+named test that was skipped, must.
+Runs under unittest (no third-party packages) and under pytest.
+"""
+import json
+import os
+import subprocess
+import sys
+import unittest
+
+SCRIPT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..",
+                      "go-test-ran.py")
+
+
+def ev(action, test=None, package="example.com/p"):
+    event = {"Action": action, "Package": package}
+    if test:
+        event["Test"] = test
+    return json.dumps(event)
+
+
+def run(lines, *prefixes):
+    proc = subprocess.run([sys.executable, SCRIPT, *prefixes],
+                          input="\n".join(lines).encode(),
+                          stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    return proc.returncode
+
+
+class GoTestRan(unittest.TestCase):
+    def test_named_tests_pass(self):
+        lines = [ev("run", "TestA"), ev("pass", "TestA"), ev("pass")]
+        self.assertEqual(run(lines, "TestA"), 0)
+
+    def test_unrelated_skips_do_not_fail(self):
+        lines = [ev("pass", "TestA"), ev("skip", "TestOther"),
+                 ev("skip", "TestOther2/sub"), ev("skip", package="example.com/empty"),
+                 ev("pass")]
+        self.assertEqual(run(lines, "TestA"), 0)
+
+    def test_named_test_skipped_fails(self):
+        lines = [ev("skip", "TestA"), ev("pass", "TestB")]
+        self.assertEqual(run(lines, "TestA", "TestB"), 1)
+
+    def test_named_subtest_skipped_fails(self):
+        lines = [ev("pass", "TestA/one"), ev("skip", "TestA/two"), ev("pass", "TestA")]
+        self.assertEqual(run(lines, "TestA"), 1)
+
+    def test_named_test_never_ran_fails(self):
+        lines = [ev("pass", "TestB"), ev("pass")]
+        self.assertEqual(run(lines, "TestA"), 1)
+
+    def test_every_prefix_needs_a_pass(self):
+        lines = [ev("pass", "TestA")]
+        self.assertEqual(run(lines, "TestA", "TestB"), 1)
+
+    def test_prefix_matches_a_passing_subtest(self):
+        self.assertEqual(run([ev("pass", "TestLinuxLeak/found")], "TestLinux"), 0)
+
+    def test_failed_named_test_fails(self):
+        lines = [ev("pass", "TestA/one"), ev("fail", "TestA")]
+        self.assertEqual(run(lines, "TestA"), 1)
+
+    def test_failed_unrelated_test_fails(self):
+        lines = [ev("pass", "TestA"), ev("fail", "TestOther")]
+        self.assertEqual(run(lines, "TestA"), 1)
+
+    def test_failed_package_fails(self):
+        lines = [ev("pass", "TestA"), ev("fail")]
+        self.assertEqual(run(lines, "TestA"), 1)
+
+    def test_non_json_line_fails(self):
+        lines = [ev("pass", "TestA"), "# example.com/p [build failed]"]
+        self.assertEqual(run(lines, "TestA"), 1)
+
+    def test_empty_input_fails(self):
+        self.assertEqual(run([], "TestA"), 1)
+
+    def test_no_prefix_is_a_usage_error(self):
+        self.assertNotEqual(run([ev("pass", "TestA")]), 0)
+
+
+if __name__ == "__main__":
+    unittest.main()