mesh-controller/cmd/mesh-controller/queue_test.go

package main

import (
	"context"
	"reflect"
	"strings"
	"testing"
	"time"

	"github.com/nats-io/nats.go"

	"github.com/novox/mesh-controller/internal/broker"
	"github.com/novox/mesh-controller/internal/catalogue"
	"github.com/novox/mesh-controller/internal/inventory"
	"github.com/novox/mesh-controller/internal/link"
)

// The build queue, controlled by hand (novox/hq ADR 0219), and the plans that follow it.
//
//	make postgres PG_PORT=55566 PG_CONTAINER=bq-pg
//	MESH_TEST_POSTGRES='postgres://postgres:check@127.0.0.1:55566/postgres?sslmode=disable' \
//	  go test ./cmd/mesh-controller/ -run 'Queue|Cancel|Clear|Retry|Rebuild|Replay|Paused'   (each test on a bus of its own)

// asksRecorded makes every ask a plan makes return the next id in a row, and says which were asked.
func asksRecorded(t *testing.T) *[]string {
	t.Helper()
	var asked []string
	was := askABuild
	askABuild = func(_ context.Context, source buildSource, path, ref string) (string, error) {
		id := link.NewBuildID(time.Now().Add(time.Duration(len(asked)) * time.Millisecond))
		asked = append(asked, source.Repository+"#"+id)
		return id, nil
	}
	t.Cleanup(func() { askABuild = was })
	return &asked
}

// twoTiers registers two modules, b standing on a, each with a source a plan asks.
func twoTiers(t *testing.T, open *stores) {
	t.Helper()
	for _, name := range []string{"a", "b"} {
		if err := open.inventory.RegisterModule(t.Context(), catalogue.Manifest{Module: name, Version: "1"},
			inventory.Source{Repository: "novox/" + name, Seat: "git", Ref: "main", BuiltFrom: "c0ffee", Head: "c0ffee"}); err != nil {
			t.Fatal(err)
		}
	}
}

// A failed plan is retried: its failed module asked again under a new id, the plan building at that
// tier, and when that build comes in the plan goes on and asks its next tier.

// What retry refuses, and says why.
func TestRetryRefusesWhatItCannotResume(t *testing.T) {
	at := time.Date(2026, 10, 5, 12, 0, 0, 0, time.UTC)
	plan := func(id, state string, created time.Time) inventory.Plan {
		return inventory.Plan{ID: id, Repository: "novox/mesh-catalog", Branch: "main", Commit: id + "c0ffee",
			Created: created, State: state, Tiers: [][]string{{"a"}},
			Modules: map[string]*inventory.PlanModule{"a": {State: "failed"}}}
	}
	failed := plan("plan-1", inventory.PlanFailed, at)
	for _, c := range []struct {
		p      inventory.Plan
		others []inventory.Plan
		says   string
	}{
		{plan("plan-d", inventory.PlanDone, at), nil, "is done"},
		{plan("plan-s", inventory.PlanSuperseded, at), nil, "was"},
		{plan("plan-o", inventory.PlanBuilding, at), nil, "still building"},
		{failed, []inventory.Plan{plan("plan-2", inventory.PlanRolling, at.Add(time.Hour))}, "plan-2 supersedes it"},
	} {
		err := retryRefusal(c.p, c.others, ownPathOf(nil))
		if err == nil || !strings.Contains(err.Error(), c.says) {
			t.Errorf("%s: %v, wanted it to say %q", c.p.ID, err, c.says)
		}
	}
	stopped := failed
	stopped.Modules = map[string]*inventory.PlanModule{"a": {State: "built"}}
	stopped.Note = "a stopped at its first machine"
	if err := retryRefusal(stopped, nil, ownPathOf(nil)); err == nil || !strings.Contains(err.Error(), "nothing in tier 0") {
		t.Errorf("a plan with nothing failed to build was retried: %v", err)
	}
	// A failed or done plan does not supersede it.
	if err := retryRefusal(failed, []inventory.Plan{plan("plan-4", inventory.PlanFailed, at.Add(time.Hour)),
		plan("plan-5", inventory.PlanDone, at.Add(time.Hour))}, ownPathOf(nil)); err != nil {
		t.Errorf("refused for a plan that does not supersede it: %v", err)
	}
	// Another branch's open walk, or an older one, does not supersede it, and is one walk open: one at a time
	// (novox/hq ADR 0276).
	other := plan("plan-3", inventory.PlanBuilding, at.Add(time.Hour))
	other.Branch = "release"
	for _, open := range []inventory.Plan{other, plan("plan-0", inventory.PlanBuilding, at.Add(-time.Hour))} {
		if err := retryRefusal(failed, []inventory.Plan{open}, ownPathOf(nil)); err == nil || !strings.Contains(err.Error(), "one walk at a time") {
			t.Errorf("retried beside the open walk %s: %v", open.ID, err)
		}
	}
}

// `rebuild` of a module a failed plan holds joins that plan; one held by nothing runs alone.

// A failed plan an open plan of its repository supersedes is not joined: the open one holds the module.
func TestARebuildJoinsTheOpenPlanBeforeASupersededFailedOne(t *testing.T) {
	at := time.Date(2026, 10, 5, 12, 0, 0, 0, time.UTC)
	failed := inventory.Plan{ID: "plan-old", Repository: "novox/a", Branch: "main", Created: at, State: inventory.PlanFailed,
		Tiers: [][]string{{"a"}}, Modules: map[string]*inventory.PlanModule{"a": {State: "failed"}}}
	newer := inventory.Plan{ID: "plan-new", Repository: "novox/a", Branch: "main", Created: at.Add(time.Hour),
		State: inventory.PlanBuilding, Tiers: [][]string{{"x"}, {"a"}}, Modules: map[string]*inventory.PlanModule{}}
	if _, found := planHolding("a", []inventory.Plan{newer}, []inventory.Plan{newer, failed}); found {
		t.Fatal("joined a failed plan a newer open one supersedes")
	}
	if p, found := planHolding("a", nil, []inventory.Plan{failed}); !found || p.ID != "plan-old" {
		t.Fatalf("did not join the failed plan holding it: %v %s", found, p.ID)
	}
	built := failed
	built.Modules = map[string]*inventory.PlanModule{"a": {State: "built"}}
	if _, found := planHolding("a", nil, []inventory.Plan{built}); found {
		t.Fatal("joined a plan that has the module built")
	}
}

// replay is a dry run unless registered, and registering an older commit than one registered since
// is refused unless --older says it is meant (novox/hq issue 207).
func TestReplayRefusesToRollAnOlderCommitOutUnlessToldTo(t *testing.T) {
	asked := time.Date(2026, 10, 5, 12, 0, 0, 0, time.UTC)
	old := inventory.Build{ID: "build-old", Module: "a", Commit: "0ldc0mm1t", Asked: asked}
	newer := inventory.Build{ID: "build-new", Module: "a", Commit: "n3wc0mm1t", Asked: asked.Add(time.Hour)}
	history := []inventory.Build{newer, old}

	if err := replayRefusal(old, "a", history, false, false, nil); err != nil {
		t.Errorf("a dry run was refused: %v", err)
	}
	err := replayRefusal(old, "a", history, true, false, nil)
	if err == nil || !strings.Contains(err.Error(), "build-new") || !strings.Contains(err.Error(), "--older") {
		t.Errorf("registering an older commit than the one registered was not refused: %v", err)
	}
	if err := replayRefusal(old, "a", history, true, true, nil); err != nil {
		t.Errorf("--register --older was refused: %v", err)
	}
	if err := replayRefusal(newer, "a", history, true, false, nil); err != nil {
		t.Errorf("registering the newest build again was refused: %v", err)
	}
	// A newer build of the same commit, or one that failed, is not a newer version to roll back from.
	same := inventory.Build{ID: "build-same", Module: "a", Commit: old.Commit, Asked: asked.Add(2 * time.Hour)}
	broken := inventory.Build{ID: "build-broken", Module: "a", Failed: "no", Asked: asked.Add(3 * time.Hour)}
	if err := replayRefusal(old, "a", []inventory.Build{broken, same, old}, true, false, nil); err != nil {
		t.Errorf("refused for a newer build of the same commit or a failed one: %v", err)
	}
	if err := replayRefusal(inventory.Build{ID: "build-x", Failed: "clone"}, "", nil, false, false, nil); err == nil {
		t.Error("a build that recorded no commit was replayed")
	}
	if err := replayRefusal(old, "a", history, false, true, nil); err == nil {
		t.Error("--older without --register was taken")
	}
	// **Nothing newer outstanding**, --older or not: an ask of the module in the queue, or an open plan
	// holding it unbuilt, would be replaced by a replay asked now (issue 219).
	q := link.Queue{Asks: []link.QueuedAsk{
		{ID: "build-queued", Repository: "https://forge.example/novox/a.git", State: link.AskWaiting},
		{ID: "build-elsewhere", Repository: "https://forge.example/novox/b.git", State: link.AskWaiting},
		{ID: "build-other-path", Repository: "https://forge.example/novox/a.git", Path: "sub", State: link.AskWaiting},
	}}
	plans := []inventory.Plan{
		{ID: "plan-holds", State: inventory.PlanBuilding, Tiers: [][]string{{"x"}, {"a"}}, Modules: map[string]*inventory.PlanModule{}},
		{ID: "plan-built", State: inventory.PlanRolling, Tiers: [][]string{{"a"}}, Modules: map[string]*inventory.PlanModule{"a": {State: "built"}}},
		{ID: "plan-failed", State: inventory.PlanFailed, Tiers: [][]string{{"a"}}, Modules: map[string]*inventory.PlanModule{}},
	}
	outstanding := outstandingFor("a", "novox/a", "", q, plans)
	if len(outstanding) != 2 || !strings.Contains(outstanding[0], "build-queued") || !strings.Contains(outstanding[1], "plan-holds") {
		t.Fatalf("outstanding: %v", outstanding)
	}
	if err := replayRefusal(old, "a", history, true, true, outstanding); err == nil || !strings.Contains(err.Error(), "build-queued") {
		t.Errorf("registered over an outstanding ask: %v", err)
	}
	if err := replayRefusal(old, "a", history, false, false, outstanding); err != nil {
		t.Errorf("a dry run was refused for what is outstanding: %v", err)
	}
	if said := replaySaid(old, "a", "build-1", false); !strings.Contains(said, "dry run") || !strings.Contains(said, "--register") {
		t.Errorf("a dry replay does not say what it is: %q", said)
	}
	if said := replaySaid(old, "a", "build-1", true); !strings.Contains(said, "registered") || !strings.Contains(said, "rolled out") {
		t.Errorf("a registered replay does not say what it does: %q", said)
	}
}

// A plan waiting on builds of a seat whose every holder is paused says so and is not late; a seat
// paused on some holders only is not a reason the plan is waiting.
func TestAPlanWaitingOnAPausedSeatSaysSoAndIsNotLate(t *testing.T) {
	now := time.Date(2026, 10, 5, 12, 0, 0, 0, time.UTC)
	asked := now.Add(-12 * time.Minute)
	p := inventory.Plan{ID: "plan-p", Repository: "novox/a", Commit: "c0ffee", State: inventory.PlanBuilding,
		Updated: now.Add(-2 * time.Hour), Tiers: [][]string{{"a"}},
		Modules: map[string]*inventory.PlanModule{"a": {State: "asked", AskedAt: &asked}}}

	all := pauseOf([]string{"g14", "ace"}, map[string]link.HolderState{"ace": {Paused: true}, "g14": {Paused: true}})
	line := planLineWith(p, now, all, tierAtLeast)
	if !strings.Contains(line, "waiting: the build seat is paused on ace, g14 (asked 12m0s ago)") || strings.Contains(line, "LATE") {
		t.Errorf("a plan on a paused seat reads %q", line)
	}
	if st := planStatuses([]inventory.Plan{p}, now, all, nil)[0]; st.Late || !strings.Contains(st.Waiting, "paused on ace, g14") {
		t.Errorf("status --json says %+v", st)
	}
	if _, late := openPlans([]inventory.Plan{p}, now, all, nil); late != 0 {
		t.Errorf("a plan waiting on a paused seat counted late")
	}

	longAgo := now.Add(-25 * time.Hour)
	forgotten := p
	forgotten.Modules = map[string]*inventory.PlanModule{"a": {State: "asked", AskedAt: &longAgo}}
	if line := planLineWith(forgotten, now, all, tierAtLeast); !strings.Contains(line, "PAUSED OVER A DAY") || strings.Contains(line, "LATE") {
		t.Errorf("a plan paused over a day reads %q", line)
	}

	some := pauseOf([]string{"g14", "ace"}, map[string]link.HolderState{"ace": {Paused: true}})
	if some.All || !reflect.DeepEqual(some.Nodes, []string{"ace"}) {
		t.Fatalf("%+v", some)
	}
	if line := planLineWith(p, now, some, tierAtLeast); !strings.Contains(line, "LATE") {
		t.Errorf("a seat paused on one holder of two made the plan not late: %q", line)
	}
	if st := planStatuses([]inventory.Plan{p}, now, some, nil)[0]; !st.Late {
		t.Errorf("status --json: %+v", st)
	}
	// A plan rolling out, or with nothing asked, is not waiting on the seat.
	rolling := p
	rolling.State = inventory.PlanRolling
	if _, paused := pausedWaiting(rolling, all, now); paused {
		t.Error("a rolling plan reads as waiting on the build seat")
	}
}

// The queue's verbs are the controller seat's, each to the command it names.
func TestTheQueueVerbsRunTheirCommands(t *testing.T) {
	for _, c := range []struct {
		verb string
		args map[string]any
		want []string
	}{
		{"queue", nil, []string{"queue", "--json"}},
		{"cancel", map[string]any{"id": "build-1"}, []string{"cancel", "build-1"}},
		{"clear", nil, []string{"clear"}},
		{"clear", map[string]any{"dead": "true"}, []string{"clear", "--dead"}},
		{"rebuild", map[string]any{"what": "gitea"}, []string{"rebuild", "gitea"}},
		{"replay", map[string]any{"id": "build-1"}, []string{"replay", "build-1"}},
		{"replay", map[string]any{"id": "build-1", "register": "true", "older": "true"}, []string{"replay", "build-1", "--register", "--older"}},
		{"kill", map[string]any{"id": "build-1"}, []string{"kill", "build-1"}},
		{"pause", nil, []string{"pause"}},
		{"resume", map[string]any{"node": "ace"}, []string{"resume", "ace"}},
		{"plans", map[string]any{"retry": "plan-1"}, []string{"plans", "retry", "plan-1"}},
	} {
		got, err := argvFor(c.verb, c.args)
		if err != nil || !reflect.DeepEqual(got, c.want) {
			t.Errorf("%s %v: %v %v, want %v", c.verb, c.args, got, err, c.want)
		}
	}
	if _, err := argvFor("cancel", nil); err == nil {
		t.Error("cancel without an id was taken")
	}
	declared := map[string]bool{}
	for _, v := range catalogue.ControllerVerbs {
		declared[v.Name] = true
	}
	for _, v := range []string{"queue", "cancel", "clear", "rebuild", "replay", "kill", "pause", "resume"} {
		if !declared[v] {
			t.Errorf("%s is not a verb of the controller seat", v)
		}
	}
}

// --- against a real bus -----------------------------------------------------------------------

// aBuildQueue is the build seat's queue, worker and cancelled set on a real server, the controller
// pointed at it, and a function that asks the seat one build.

// deadOne takes the oldest ask from the worker and hands it back as often as the worker allows.
func deadOne(t *testing.T, js *broker.JetStream) {
	t.Helper()
	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() }()
	for i := 0; i < worker.MaxDeliver; i++ {
		msgs, err := sub.Fetch(1, nats.MaxWait(3*time.Second))
		if err != nil {
			t.Fatalf("delivery %d: %v", i+1, err)
		}
		_ = msgs[0].Nak()
	}
	// One more pull, as a holder always has one waiting: the server finds the ask past its deliveries
	// then, and stops counting it pending.
	if msgs, _ := sub.Fetch(1, nats.MaxWait(time.Second)); len(msgs) > 0 {
		t.Fatalf("an ask past its deliveries was delivered again")
	}
}

// cancel drops a waiting ask from the bus and records it failed, cancelled by hand — and the plan
// that asked for it, matched by the id, fails with it; an ask the plan's records name no module for
// is still found.

// clear cancels every waiting ask and leaves the dead ones unless told, and never one in flight.

// An ask in flight is not cancelled: kill ends it where it runs.

// kill finds the machine from the build's start and asks that machine's holder; pause asks the
// machine named. Each prints what the holder answered, and a refusal is the command's failure.

// A plan that stopped at its first machine is retried: the module sent to that machine again, the
// send recorded as the first anew, and the plan goes on — unless a newer plan holds the module.

// A plan module asked under an id is settled by that id's outcome alone: a replay or a rebuild beside
// the plan, asked later, never answers it (novox/hq ADR 0219).

// rebuild of a build made at a commit asks what the module follows now, never the commit.

// An ask the worker counts out that no machine said it started is cancelled only if no start comes
// while a holder looks: one that does withdraws the cancel, and kill is what ends it.