mesh-controller/cmd/mesh-controller/superseded_test.go

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