issue 974: review 2: a check stopped by its key says no verdict, an error judges nothing, and an unacknowledged check goes to a second builder

cmd/mesh-controller/checks.go

@@ -583,6 +583,13 @@ const maxCheckReport = 60 << 10
 // checked says a merge check's verdict as the controller's `checked`. Nothing is recorded or
 // registered: a check builds nothing (issue 240's rule for a dry run, kept for a check).
 func checked(ctx context.Context, result link.BuildResult) {
+	// **A check stopped by its key says nothing** (novox/hq ADR 0406 §4, issue 974, review 2): it stopped because the head or
+	// the trunk commit moved on, and the newer request will judge. Said as an error it would name the live head — red
+	// statuses on the forge, a delivery refused — and written to the key it would stand in the newer request's way.
+	if result.Failed == link.SupersededByKey {
+		fmt.Printf("%s: stopped, its pull request's key moved on; the newer check will judge\n", result.ID)
+		return
+	}
 	sayChecked(ctx, checkedOf(result))
 }
 

cmd/mesh-controller/checks_test.go

@@ -1,9 +1,16 @@
 package main
 
 import (
+	"encoding/json"
 	"errors"
 	"strings"
 	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go"
+
+	"github.com/novox/mesh-controller/internal/broker"
+	"github.com/novox/mesh-controller/internal/testbus"
 
 	"github.com/novox/mesh-controller/internal/inventory"
 	"github.com/novox/mesh-controller/internal/link"
@@ -288,3 +295,47 @@ func TestARecheckedHeadsDirectVerdictNamesItsTrunk(t *testing.T) {
 		t.Fatalf("a rechecked head's direct verdict names trunk %q", c.Trunk)
 	}
 }
+
+// A check stopped by its pull request's key moving on says no verdict (issue 974, review 2): an error said for it would name the
+// live head, and the newer request is the one that judges. A check that failed for any other reason still says its error.
+func TestACheckStoppedByItsKeyIsAnnouncedAsNoVerdict(t *testing.T) {
+	js, err := broker.Dial(testbus.URL(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(js.Close)
+	if err := broker.AssertMeshStreams(js); err != nil {
+		t.Fatal(err)
+	}
+	legacy, err := js.Conn().JetStream()
+	if err != nil {
+		t.Fatal(err)
+	}
+	was := checkEvents
+	checkEvents = link.OverNATS{Conn: js.Conn(), JS: legacy}
+	t.Cleanup(func() { checkEvents = was })
+	heard := make(chan *nats.Msg, 4)
+	sub, err := js.Conn().ChanSubscribe(link.SeatEventSubject(link.MeshControllerSeat, link.KeyChecked), heard)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = sub.Unsubscribe() }()
+	asked := &link.CheckRequest{Owner: "novox", Repo: "mesh-catalog", Number: 3}
+	checked(t.Context(), link.BuildResult{ID: "stopped", Ref: "abc", Checked: asked, Failed: link.SupersededByKey})
+	checked(t.Context(), link.BuildResult{ID: "broken", Ref: "abc", Checked: asked, Failed: "the facts snapshot cannot be read"})
+	select {
+	case m := <-heard:
+		var c link.Checked
+		_ = json.Unmarshal(m.Data, &c)
+		if c.ID != "broken" || c.Verdict != "error" {
+			t.Fatalf("the first verdict said is %+v; want only the broken check's error", c)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("nothing was said")
+	}
+	select {
+	case m := <-heard:
+		t.Fatalf("a second verdict was said: %s", m.Data)
+	case <-time.After(300 * time.Millisecond):
+	}
+}

internal/link/checks_nats.go

@@ -85,7 +85,10 @@ func (k PullKey) Moved(head, trunk string) bool {
 
 // Judged says a key holds the verdict of the check of this head against this trunk commit.
 func (k PullKey) Judged(head, trunk string) bool {
-	return k.Verdict != "" && sameCommit(k.Head, head) && sameCommit(k.Trunk, trunk) && trunk != ""
+	// A verdict of a check that ran: a pass, a warning or a fail. An error is a check that could not run, and a request
+	// after it must run (issue 974, review 2).
+	judgedBy := k.Verdict == "pass" || k.Verdict == "warning" || k.Verdict == "fail"
+	return judgedBy && sameCommit(k.Head, head) && sameCommit(k.Trunk, trunk) && trunk != ""
 }
 
 // ReadPullKey reads a pull request's key from mesh-delivery's bucket: false when it has none.

internal/link/checks_nats_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 
 	"github.com/novox/mesh-controller/internal/broker"
@@ -484,3 +485,166 @@ func TestARunningCheckRenewsItsKeyAndDeletesItWhenItEnds(t *testing.T) {
 		}
 	}
 }
+
+// A key whose verdict is "error" — a check that could not run, or was stopped — judged nothing: a request for that head and
+// trunk commit must run (issue 974, review 2).
+func TestAnErrorVerdictJudgesNothing(t *testing.T) {
+	for verdict, judged := range map[string]bool{"pass": true, "warning": true, "fail": true, "error": false, "": false, "pending": false} {
+		k := PullKey{Head: "aaaaaaaaaaaa1111", Trunk: "tttttttttttt1111", Verdict: verdict}
+		if got := k.Judged("aaaaaaaaaaaa1111", "tttttttttttt1111"); got != judged {
+			t.Errorf("a key with verdict %q judged = %v; want %v", verdict, got, judged)
+		}
+	}
+}
+
+// A check stopped because the trunk moved leaves the newer request, for the head against the new trunk commit, taken and
+// run: the stop is the stopped one's failure, not the key's verdict, and the newer request is not dropped by it.
+func TestACheckStoppedByATrunkMoveLeavesTheNewerRequestTakenAndRun(t *testing.T) {
+	js := aBusWithTheBuildRole(t)
+	kv := aKeyBucket(t, js)
+	aRunningBucket(t, js, 2*time.Minute)
+	ctx, stop := context.WithCancel(t.Context())
+	defer stop()
+	asks := BuildsOn(js, TheBuildMachine)
+	outcomes := make(chan BuildResult, 8)
+	watching, err := js.Conn().Subscribe(BuildOutcome(), func(m *nats.Msg) {
+		var r BuildResult
+		if json.Unmarshal(m.Data, &r) == nil {
+			outcomes <- r
+		}
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = watching.Unsubscribe() }()
+
+	ran := make(chan string, 4)
+	m := MachineOverNATS(js, "builder-a")
+	defer m.Close()
+	go func() {
+		_ = m.Take(ctx, func(ctx context.Context, work Build) {
+			building, cancel := context.WithCancel(ctx)
+			defer cancel()
+			ran <- work.Request().ID
+			stopped := make(chan struct{}, 1)
+			end, err := work.(interface {
+				WatchPull(context.Context, func(PullKey)) (func(), error)
+			}).WatchPull(building, func(PullKey) { stopped <- struct{}{}; cancel() })
+			if err != nil {
+				t.Error(err)
+				return
+			}
+			defer end()
+			if work.Request().ID == "older" {
+				select {
+				case <-stopped:
+					_ = work.Announce(ctx, BuildResult{ID: "older", On: "builder-a", Failed: SupersededByKey})
+				case <-time.After(15 * time.Second):
+					_ = work.Announce(ctx, BuildResult{ID: "older", On: "builder-a", Failed: "never stopped"})
+				}
+			} else {
+				_ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "builder-a", Check: &CheckOutcome{Verdict: "pass"}})
+			}
+			_ = work.Done()
+		})
+	}()
+	if err := asks.Ask(ctx, aPullCheck("older", 7, "head-7-000000000", "trunk-1-000000", false)); err != nil {
+		t.Fatal(err)
+	}
+	if id := <-ran; id != "older" {
+		t.Fatalf("took %s", id)
+	}
+	time.Sleep(300 * time.Millisecond) // the watch is made
+	// The trunk moved; the head did not. mesh-delivery writes the key for the new request, and the request is asked.
+	putKey(t, kv, 7, PullKey{Head: "head-7-000000000", Trunk: "trunk-2-000000"})
+	if err := asks.Ask(ctx, aPullCheck("newer", 7, "head-7-000000000", "trunk-2-000000", false)); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case id := <-ran:
+		if id != "newer" {
+			t.Fatalf("took %s", id)
+		}
+	case <-time.After(15 * time.Second):
+		t.Fatal("the newer request, for the same head on the new trunk commit, was not taken")
+	}
+	got := map[string]string{}
+	for len(got) < 2 {
+		select {
+		case r := <-outcomes:
+			got[r.ID] = r.Failed
+		case <-time.After(10 * time.Second):
+			t.Fatalf("outcomes so far: %v", got)
+		}
+	}
+	if got["older"] != SupersededByKey || got["newer"] != "" {
+		t.Errorf("outcomes %v; want the older stopped and the newer run to a verdict", got)
+	}
+}
+
+// An unacknowledged check goes to a second builder: taken from a check worker and not acknowledged, it is handed out again
+// when the worker's wait is up.
+func TestAnUnacknowledgedCheckGoesToASecondBuilder(t *testing.T) {
+	for _, ready := range []bool{false, true} {
+		js := aBusWithTheBuildRole(t)
+		name, worker := "other", broker.ChecksOtherWorker
+		if ready {
+			name, worker = "ready", broker.ChecksReadyWorker
+		}
+		// A worker whose wait is a second, as the mesh's is a minute.
+		for _, w := range broker.CheckWorkers() {
+			if w.Name == worker {
+				w.AckWaitSeconds = 1
+				if err := js.EnsureConsumer(w); err != nil {
+					t.Fatal(err)
+				}
+			}
+		}
+		ctx, stop := context.WithCancel(t.Context())
+		if err := BuildsOn(js, TheBuildMachine).Ask(ctx, aPullCheck("dies-"+name, 7, "head-7-000000000", "trunk-1-000000", ready)); err != nil {
+			t.Fatal(err)
+		}
+		// The first builder takes it and dies: its pull is fetched, never acknowledged.
+		if _, err := fetchOne(js, worker); err != nil {
+			t.Fatal(err)
+		}
+		took := make(chan BuildRequest, 2)
+		second := MachineOverNATS(js, "builder-b")
+		go func() {
+			_ = second.Take(ctx, func(ctx context.Context, work Build) {
+				took <- work.Request()
+				_ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "builder-b", Failed: "no"})
+				_ = work.Done()
+			})
+		}()
+		select {
+		case r := <-took:
+			if r.ID != "dies-"+name {
+				t.Errorf("%s: took %s", name, r.ID)
+			}
+		case <-time.After(15 * time.Second):
+			t.Errorf("%s: the unacknowledged check did not go to a second builder", name)
+		}
+		second.Close()
+		stop()
+	}
+}
+
+// fetchOne takes the next message of a worker and leaves it unacknowledged, as a builder that died would.
+func fetchOne(js *broker.JetStream, worker string) (*nats.Msg, error) {
+	for _, w := range broker.CheckWorkers() {
+		if w.Name != worker {
+			continue
+		}
+		sub, err := js.Context().PullSubscribe(w.Filters[0], w.Name, nats.Bind(w.Stream, w.Name), nats.ManualAck())
+		if err != nil {
+			return nil, err
+		}
+		got, err := sub.Fetch(1, nats.MaxWait(5*time.Second))
+		if err != nil {
+			return nil, err
+		}
+		return got[0], nil
+	}
+	return nil, nil
+}