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

@@ -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)
 	}

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)

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))",

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})

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 {

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.

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
 }
 

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()
+}

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)
+		}
+	}
+}

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.