diff --git a/cmd/mesh-controller/checks.go b/cmd/mesh-controller/checks.go index b3ba6d6a..e0104e34 100644 --- a/cmd/mesh-controller/checks.go +++ b/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)) } diff --git a/cmd/mesh-controller/checks_test.go b/cmd/mesh-controller/checks_test.go index 8ea4242a..dd6a3b8e 100644 --- a/cmd/mesh-controller/checks_test.go +++ b/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): + } +} diff --git a/internal/link/checks_nats.go b/internal/link/checks_nats.go index cf09cad0..fe900eb2 100644 --- a/internal/link/checks_nats.go +++ b/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. diff --git a/internal/link/checks_nats_test.go b/internal/link/checks_nats_test.go index 34c36447..b5a2f51c 100644 --- a/internal/link/checks_nats_test.go +++ b/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 +}