From e4acab32d6a20adeb9b8274125b5ec4ad7c41bdd Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 14 Oct 2026 22:04:48 +0200 Subject: [PATCH] End the check of a pull request's old head when a new head is checked A check judges one head; once the pull request has another, its verdict answers nobody and its run holds a build machine the new head waits behind. Asking a check of R#N at H now drops every waiting check of R#N at another head and kills every one in flight, saying each in the log. Announcements are unordered and a recheck asks a recorded head, so this happens only when the forge confirms H is the pull request's current head; otherwise nothing ends and that is said. A one-shot recheck ends them within the call. Tries, walk builds, group checks, other pull requests' checks and rechecks of the same head are left alone. novox/hq issue 726 --- cmd/mesh-controller/addresses_test.go | 2 +- cmd/mesh-controller/batches_test.go | 2 +- cmd/mesh-controller/checks.go | 2 + cmd/mesh-controller/delivery.go | 2 +- cmd/mesh-controller/missed_merges.go | 2 +- cmd/mesh-controller/push.go | 6 +- cmd/mesh-controller/queue.go | 5 +- cmd/mesh-controller/superseded.go | 157 ++++++++++++++++++++ cmd/mesh-controller/superseded_test.go | 192 +++++++++++++++++++++++++ cmd/mesh-controller/upgrades.go | 7 +- 10 files changed, 368 insertions(+), 9 deletions(-) create mode 100644 cmd/mesh-controller/superseded.go create mode 100644 cmd/mesh-controller/superseded_test.go diff --git a/cmd/mesh-controller/addresses_test.go b/cmd/mesh-controller/addresses_test.go index dad8dce3..bc02b37b 100644 --- a/cmd/mesh-controller/addresses_test.go +++ b/cmd/mesh-controller/addresses_test.go @@ -140,7 +140,7 @@ func TestTheRegistryTrustAndEveryImageFollowThePortTheNodeGaveTheStore(t *testin } // And a replay to the catalogue says where the store is now. - announced, err := following{open}.Announceable(ctx) + announced, err := following{open: open}.Announceable(ctx) if err != nil { t.Fatal(err) } diff --git a/cmd/mesh-controller/batches_test.go b/cmd/mesh-controller/batches_test.go index 778abe73..9e6bba74 100644 --- a/cmd/mesh-controller/batches_test.go +++ b/cmd/mesh-controller/batches_test.go @@ -758,7 +758,7 @@ func TestTheCatchUpKeepsWhatACutMadeHistory(t *testing.T) { return entries, read, err } if err := catchUpOnMerges(ctx, time.Now(), unheardMerges{announcedList(lost), open.inventory}, catalogued, - (following{open}).SourceMoved, t.Logf); err != nil { + (following{open: open}).SourceMoved, t.Logf); err != nil { t.Fatal(err) } got, _ := open.inventory.PlanByID(ctx, w.ID) diff --git a/cmd/mesh-controller/checks.go b/cmd/mesh-controller/checks.go index 1c7eefda..2b81c906 100644 --- a/cmd/mesh-controller/checks.go +++ b/cmd/mesh-controller/checks.go @@ -189,6 +189,8 @@ func (f following) PullUpdated(ctx context.Context, p link.PullUpdated) error { if err := ask.Ask(ctx, request); err != nil { return err } + // A new head ends the check of the old one (novox/hq issue 726). + endingSupersededChecks(ctx, seat, p, request.ID, f.oneShot) what := "its own merge-check.sh alone: " + noModuleTouched if scope.gated() { what = fmt.Sprintf("the gate over %s (a merge would build %d module(s) in %d tier(s))", diff --git a/cmd/mesh-controller/delivery.go b/cmd/mesh-controller/delivery.go index 40b7c489..30d78461 100644 --- a/cmd/mesh-controller/delivery.go +++ b/cmd/mesh-controller/delivery.go @@ -952,7 +952,7 @@ func deliveryCommand(ctx context.Context, args []string) error { if *group == "" && len(members) == 1 { // One head, checked again as the forge's announcement would have it (a recheck): the controller's // own path for a pull request, run because the delivery's owner asked. - if err := sayingOnTheBus(ctx, inv, func() error { return (following{open}).PullUpdated(ctx, members[0].pull()) }); err != nil { + if err := sayingOnTheBus(ctx, inv, func() error { return (following{open: open, oneShot: true}).PullUpdated(ctx, members[0].pull()) }); err != nil { return err } return answer(map[string]any{"rechecked": members[0].ID}) diff --git a/cmd/mesh-controller/missed_merges.go b/cmd/mesh-controller/missed_merges.go index 32ea7b30..1c9285e8 100644 --- a/cmd/mesh-controller/missed_merges.go +++ b/cmd/mesh-controller/missed_merges.go @@ -86,7 +86,7 @@ func owedToTheRecord(ctx context.Context, announced merges, m link.SourceMoved, // catchingUpOnMerges reads back the forge's announcements on a timer, until the context ends. func catchingUpOnMerges(ctx context.Context, open *stores, announced merges) { - f := following{open} + f := following{open: open} catalogued := func(ctx context.Context) ([]inventory.Entry, map[string][]inventory.ReadRepository, error) { entries, err := open.inventory.Catalogued(ctx) if err != nil { diff --git a/cmd/mesh-controller/push.go b/cmd/mesh-controller/push.go index 05bfc4c5..4027818e 100644 --- a/cmd/mesh-controller/push.go +++ b/cmd/mesh-controller/push.go @@ -194,7 +194,7 @@ func serve(ctx context.Context) (err error) { // And what the catalogue decided a build meant. The builder's own result is already handled // above; this is the other half — the control plane is the only one of the three that knows // which machines run the thing, so it is the one that acts (novox/hq ADR 0072). - if err := server.Follows(following{open}); err != nil { + if err := server.Follows(following{open: open}); err != nil { return err } // And the merges the bus announced and never handed over, read back on a timer and acted on late @@ -202,7 +202,7 @@ func serve(ctx context.Context) (err error) { go catchingUpOnMerges(ctx, open, server) // And a catalogue that has just started, asking for what it missed. The same type answers // both: what a build meant and what the builds were are two questions about one record. - if err := server.Answers(following{open}); err != nil { + if err := server.Answers(following{open: open}); err != nil { return err } // And what providers say about consumers they keep failing, kept for `status` (novox/hq ADR @@ -256,7 +256,7 @@ func serve(ctx context.Context) (err error) { go forgetChecksOfWalk(context.WithoutCancel(ctx), open.inventory, p) } } - if err := server.Checks(following{open}); err != nil { + if err := server.Checks(following{open: open}); err != nil { return err } // Composed now and kept current, before the verb that answers from it is served. diff --git a/cmd/mesh-controller/queue.go b/cmd/mesh-controller/queue.go index 629fe5d4..651dbcdf 100644 --- a/cmd/mesh-controller/queue.go +++ b/cmd/mesh-controller/queue.go @@ -340,7 +340,10 @@ func cancelAsk(ctx context.Context, js *broker.JetStream, open *stores, seat str return "", fmt.Errorf("%s is marked cancelled and could not be deleted from the queue (seq %d): %w — a "+ "holder taking it ends it as cancelled", ask.ID, ask.Seq, err) } - recordCancelled(ctx, open, ask) + // A check ended because its pull request has a newer head records nothing (novox/hq issue 726): open is nil. + if open != nil { + recordCancelled(ctx, open, ask) + } return cancelSaid(ask, seat, removed), nil } diff --git a/cmd/mesh-controller/superseded.go b/cmd/mesh-controller/superseded.go new file mode 100644 index 00000000..e3906713 --- /dev/null +++ b/cmd/mesh-controller/superseded.go @@ -0,0 +1,157 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "strings" + "time" + + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/link" +) + +// A new push to a pull request ends the check of its old head (novox/hq issue 726). +// +// A check judges one head; once the pull request has another, its verdict answers nobody and its run holds a build +// machine for minutes the new head waits behind. So when a check is asked for pull request N of a repository at a +// head, **every check of that pull request at another head ends**: a waiting one is dropped from the queue as +// `cancel` drops it, one in flight is ended on its machine as `kill` ends it. Each is said in the log. +// +// **Never anything else**: a try (`try`, asked for no pull request), a walk's build (no check), a delivery group's +// composed check (it answers several pull requests), another pull request's check, a dead ask (left for `clear +// --dead`), and a check of the same head — a recheck of that head is not another head. + +// **Only for the pull request's current head.** Announcements of a pull request are not ordered, one is heard again +// after a restart, and a recheck asks for the head it recorded: a check asked for a head the pull request no longer +// has must never end the check of the head it has. So the forge is asked first which head the pull request has; only +// when it names this one does anything end, and when it cannot be asked, or names another, nothing ends and that is +// said. + +// forgePullHead asks the forge, through its module's tools on the bus, which commit a pull request's head is. A +// variable so a test needs no forge. +var forgePullHead = func(ctx context.Context, js *broker.JetStream, owner, repo string, number int) (string, error) { + raw, _ := json.Marshal(map[string]any{"path": fmt.Sprintf("/repos/%s/%s/pulls/%d", owner, repo, number)}) + answer, err := link.Ask(ctx, link.OverNATS{Conn: js.Conn()}, "gitea", "gitea_api", raw, 30*time.Second) + if err != nil { + return "", err + } + if answer.Error != "" { + return "", errors.New(answer.Error) + } + var pull struct { + Result struct { + Head struct { + SHA string `json:"sha"` + } `json:"head"` + } `json:"result"` + } + if err := json.Unmarshal(answer.Result, &pull); err != nil { + return "", fmt.Errorf("the forge's answer about the pull request cannot be read: %w", err) + } + if pull.Result.Head.SHA == "" { + return "", errors.New("the forge named no head for the pull request") + } + return pull.Result.Head.SHA, nil +} + +// supersededChecks are the checks in the queue that a check of owner/repo#number at head ends: of the same pull +// request, at another head, waiting or in flight, and not the ask keep. +func supersededChecks(q link.Queue, owner, repo string, number int, head, keep string) []link.QueuedAsk { + if number <= 0 || head == "" { + return nil + } + var out []link.QueuedAsk + for _, a := range q.Asks { + c := a.Request.Check + switch { + case c == nil, link.IsTry(a.Request), c.Group != "", c.Number != number, a.ID == keep, + !strings.EqualFold(c.Owner, owner), !strings.EqualFold(c.Repo, repo), + a.State != link.AskWaiting && a.State != link.AskInFlight: + continue + } + ref := a.Request.Ref + if ref == "" { + ref = a.Ref + } + if ref == "" || ref == head || link.SameHead(ref, head) { + continue + } + out = append(out, a) + } + return out +} + +// endSupersededChecks ends the checks a check of p at its head supersedes, keep being the new check's own ask, and +// says each in the log. A waiting one is cancelled — nothing recorded: a check builds nothing, and the forge's status +// of the old head is no longer the pull request's — and one a machine builds is killed there. +func endSupersededChecks(ctx context.Context, js *broker.JetStream, seat string, p link.PullUpdated, keep string) []string { + if p.Number <= 0 || p.Commit == "" { + return nil + } + current, err := forgePullHead(ctx, js, p.Owner, p.Repo, p.Number) + if err != nil { + return []string{fmt.Sprintf("%s/%s#%d (%.8s): nothing ended — the forge could not say the pull request's "+ + "current head, so checks of its other heads keep running: %v", p.Owner, p.Repo, p.Number, p.Commit, err)} + } + if current != p.Commit && !link.SameHead(current, p.Commit) { + return []string{fmt.Sprintf("%s/%s#%d (%.8s): nothing ended — the pull request's head is %.12s now, not "+ + "this one", p.Owner, p.Repo, p.Number, p.Commit, current)} + } + q, err := link.ReadQueue(ctx, js, seat) + if err != nil { + return []string{fmt.Sprintf("%s/%s#%d (%.8s): could not read the %s queue for checks of its earlier heads: %v", + p.Owner, p.Repo, p.Number, p.Commit, seat, err)} + } + var said []string + for _, a := range supersededChecks(q, p.Owner, p.Repo, p.Number, p.Commit, keep) { + old := a.Request.Ref + if old == "" { + old = a.Ref + } + what := fmt.Sprintf("check of %s/%s#%d at %.12s", p.Owner, p.Repo, p.Number, old) + var err error + if a.State == link.AskInFlight && a.On != "" { + var answer link.Answer + answer, err = link.AskSeatTool(ctx, js.Conn(), seat, "kill", a.On, map[string]string{"id": a.ID}, killAnswer) + if err == nil && answer.Error != "" { + err = fmt.Errorf("%s refused: %s", a.On, answer.Error) + } + } else { + _, err = cancelAsk(ctx, js, nil, seat, a) + } + if err != nil { + said = append(said, fmt.Sprintf("not ended: %s (%s) — superseded by %.12s, but: %v", what, a.ID, p.Commit, err)) + continue + } + said = append(said, fmt.Sprintf("ended: %s — superseded by %.12s (%s, %s)", what, p.Commit, a.ID, a.State)) + } + return said +} + +// endingSupersededChecks ends them beside the announcement, so a kill waiting on its machine never holds the +// handler: on its own connection to the bus, within twice a kill's wait. A one-shot process (oneShot) ends them +// within the call instead, as its exit would cut them off. +func endingSupersededChecks(ctx context.Context, seat string, p link.PullUpdated, keep string, oneShot bool) { + end := func() { + ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 2*killAnswer) + defer cancel() + js, err := dialTheBus() + if err != nil { + fmt.Fprintf(os.Stderr, "%s/%s#%d (%.8s): could not reach the bus to end checks of its earlier heads: %v\n", + p.Owner, p.Repo, p.Number, p.Commit, err) + return + } + defer js.Close() + for _, line := range endSupersededChecks(ctx, js, seat, p, keep) { + fmt.Println(line) + } + } + if oneShot { + end() + return + } + go end() +} diff --git a/cmd/mesh-controller/superseded_test.go b/cmd/mesh-controller/superseded_test.go new file mode 100644 index 00000000..104fb78b --- /dev/null +++ b/cmd/mesh-controller/superseded_test.go @@ -0,0 +1,192 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "strings" + "sync" + "testing" + "time" + + "github.com/nats-io/nats.go" + + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/link" +) + +// A new push to a pull request ends the check of its old head (novox/hq issue 726): the waiting one is dropped from +// the queue, the one in flight is killed on its machine, and each is said. A try, a walk's build, another pull +// request's check, a group's check and a recheck of the new head are left alone. On the bus alone: no store. +func TestANewHeadEndsTheChecksOfTheOldOnes(t *testing.T) { + js, _ := aBuildQueue(t) + ctx := t.Context() + const ( + oldHead = "1111111111111111111111111111111111111111" + olderHead = "2222222222222222222222222222222222222222" + newHead = "3333333333333333333333333333333333333333" + ) + publish := func(r link.BuildRequest) { + t.Helper() + body, _ := json.Marshal(r) + if _, err := js.Context().Publish(link.BuildWork(), body); err != nil { + t.Fatal(err) + } + } + check := func(id string, number int, head string) link.BuildRequest { + return link.BuildRequest{ID: id, Repository: "https://forge.example/novox/a.git", Ref: head, + Check: &link.CheckRequest{Owner: "novox", Repo: "a", Number: number}} + } + // In flight first: taken by a holder, started on ace. + publish(check("check-running", 7, olderHead)) + worker, _ := broker.HolderConsumerFor("", "", broker.DeclaredSeat{Name: link.TheBuildMachine, Accepts: []string{"build"}}) + sub, err := js.Context().PullSubscribe(worker.Filters[0], worker.Name, nats.Bind(worker.Stream, worker.Name), nats.ManualAck()) + if err != nil { + t.Fatal(err) + } + defer func() { _ = sub.Unsubscribe() }() + if _, err := sub.Fetch(1, nats.MaxWait(3*time.Second)); err != nil { + t.Fatal(err) + } + started, _ := json.Marshal(link.BuildStart{ID: "check-running", On: "ace", At: time.Now().UTC().Format(time.RFC3339Nano)}) + if _, err := js.Context().Publish(link.BuildStarted(), started); err != nil { + t.Fatal(err) + } + publish(check("check-old", 7, oldHead)) + publish(check("check-other-pull", 8, oldHead)) + publish(check("check-recheck", 7, newHead)) + publish(check("check-new", 7, newHead)) + try := check("try-old", 7, oldHead) + try.Check.Number, try.Check.Try = 0, true + publish(try) + group := check("group-old", 7, oldHead) + group.Check.Group = "g1" + publish(group) + publish(link.BuildRequest{ID: "walk-build", Repository: "https://forge.example/novox/a.git", Ref: oldHead}) + + var mu sync.Mutex + var killed []string + stop, err := link.OverNATS{Conn: js.Conn()}.ServeNodeSeatTools(link.TheBuildMachine, "ace", map[string]link.ToolHandler{ + "kill": func(_ context.Context, raw json.RawMessage) (any, error) { + var args struct{ ID string } + _ = json.Unmarshal(raw, &args) + mu.Lock() + killed = append(killed, args.ID) + mu.Unlock() + return map[string]any{"said": "killed " + args.ID}, nil + }}, nil) + if err != nil { + t.Fatal(err) + } + defer stop() + + forgeSays(t, newHead, nil) + said := endSupersededChecks(ctx, js, link.TheBuildMachine, + link.PullUpdated{Owner: "novox", Repo: "a", Number: 7, Commit: newHead}, "check-new") + + mu.Lock() + if len(killed) != 1 || killed[0] != "check-running" { + t.Errorf("the holder was asked to kill %v, want only the in-flight check of the older head", killed) + } + mu.Unlock() + q, err := link.ReadQueue(ctx, js, link.TheBuildMachine) + if err != nil { + t.Fatal(err) + } + if _, found := q.Find("check-old"); found { + t.Error("the waiting check of the old head is still in the queue") + } + for _, id := range []string{"check-other-pull", "check-recheck", "check-new", "try-old", "group-old", "walk-build"} { + if _, found := q.Find(id); !found { + t.Errorf("%s was ended; only checks of the same pull request at another head end", id) + } + } + text := strings.Join(said, "\n") + for _, want := range []string{ + "ended: check of novox/a#7 at " + oldHead[:12] + " — superseded by " + newHead[:12], + "ended: check of novox/a#7 at " + olderHead[:12] + " — superseded by " + newHead[:12], + } { + if !strings.Contains(text, want) { + t.Errorf("the log does not say %q:\n%s", want, text) + } + } + if len(said) != 2 { + t.Errorf("said %d lines, want 2:\n%s", len(said), text) + } +} + +// Which asks a check supersedes, read from the queue alone. +func TestSupersededChecksAreOnlyThePullRequestsOtherHeads(t *testing.T) { + ask := func(id, state, ref string, c *link.CheckRequest) link.QueuedAsk { + return link.QueuedAsk{ID: id, State: state, Ref: ref, Request: link.BuildRequest{ID: id, Ref: ref, Check: c}} + } + pr := func(n int) *link.CheckRequest { return &link.CheckRequest{Owner: "Novox", Repo: "A", Number: n} } + head := "abcdefabcdefabcdef0123456789abcdef012345" + q := link.Queue{Asks: []link.QueuedAsk{ + ask("old", link.AskWaiting, "0123456789ab0123456789ab0123456789ab0123", pr(7)), + ask("short-same", link.AskWaiting, head[:12], pr(7)), + ask("dead", link.AskDead, "0123456789ab0123456789ab0123456789ab0123", pr(7)), + ask("other", link.AskWaiting, "0123456789ab0123456789ab0123456789ab0123", pr(8)), + ask("build", link.AskWaiting, "0123456789ab0123456789ab0123456789ab0123", nil), + ask("keep", link.AskWaiting, "0123456789ab0123456789ab0123456789ab0123", pr(7)), + }} + got := supersededChecks(q, "novox", "a", 7, head, "keep") + if len(got) != 1 || got[0].ID != "old" { + t.Fatalf("superseded %+v, want only old", got) + } + if got := supersededChecks(q, "novox", "a", 0, head, ""); got != nil { + t.Errorf("a check of no pull request superseded %+v", got) + } +} + +// forgeSays makes the forge name head as every pull request's current head, or fail with err. +func forgeSays(t *testing.T, head string, err error) { + t.Helper() + was := forgePullHead + forgePullHead = func(context.Context, *broker.JetStream, string, string, int) (string, error) { return head, err } + t.Cleanup(func() { forgePullHead = was }) +} + +// An older head heard after a newer one — an announcement out of order, one heard again after a restart, a recheck +// of a recorded head — ends nothing: the forge says the pull request's head is the newer one. Nor does a check whose +// pull request's head the forge cannot say. +func TestAnOlderHeadHeardLateEndsNothing(t *testing.T) { + js, _ := aBuildQueue(t) + ctx := t.Context() + const ( + oldHead = "1111111111111111111111111111111111111111" + newHead = "3333333333333333333333333333333333333333" + ) + for _, id := range []string{"check-new", "check-old-late"} { + head := newHead + if id == "check-old-late" { + head = oldHead + } + body, _ := json.Marshal(link.BuildRequest{ID: id, Repository: "https://forge.example/novox/a.git", Ref: head, + Check: &link.CheckRequest{Owner: "novox", Repo: "a", Number: 7}}) + if _, err := js.Context().Publish(link.BuildWork(), body); err != nil { + t.Fatal(err) + } + } + late := link.PullUpdated{Owner: "novox", Repo: "a", Number: 7, Commit: oldHead} + forgeSays(t, newHead, nil) + said := endSupersededChecks(ctx, js, link.TheBuildMachine, late, "check-old-late") + if len(said) != 1 || !strings.Contains(said[0], "nothing ended") { + t.Errorf("an older head heard late said %q", said) + } + forgeSays(t, "", errors.New("nothing serves gitea.gitea_api")) + said = endSupersededChecks(ctx, js, link.TheBuildMachine, + link.PullUpdated{Owner: "novox", Repo: "a", Number: 7, Commit: newHead}, "check-new") + if len(said) != 1 || !strings.Contains(said[0], "nothing ended") { + t.Errorf("a head the forge could not confirm said %q", said) + } + q, err := link.ReadQueue(ctx, js, link.TheBuildMachine) + if err != nil { + t.Fatal(err) + } + for _, id := range []string{"check-new", "check-old-late"} { + if _, found := q.Find(id); !found { + t.Errorf("%s was ended", id) + } + } +} diff --git a/cmd/mesh-controller/upgrades.go b/cmd/mesh-controller/upgrades.go index 7570fb79..1bf62efd 100644 --- a/cmd/mesh-controller/upgrades.go +++ b/cmd/mesh-controller/upgrades.go @@ -23,7 +23,12 @@ import ( // **It holds the stores, not a copy of the decision.** What to do about an upgrade is read when // one arrives, so changing it takes effect on the next upgrade rather than on the next restart of // the control plane. -type following struct{ open *stores } +type following struct { + open *stores + // oneShot is a process that ends after the call (a recheck through `delivery check`): what it would do beside + // the call it does within it, or the process's exit would cut it off (novox/hq issue 726). + oneShot bool +} // Upgraded sends the machines running a module the version the catalogue now considers current — // or records that they are behind, which is the default and needs no record.