package main import ( "context" "encoding/json" "fmt" "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" "github.com/novox/mesh-controller/internal/testbus" ) // 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. func TestAFailedPlanIsRetriedAndGoesOnThroughItsLaterTiers(t *testing.T) { open := aMesh(t) ctx := t.Context() asked := asksRecorded(t) twoTiers(t, open) before := time.Now().UTC().Add(-time.Hour) failed := inventory.Plan{ID: "plan-retry", Repository: "novox/a", Branch: "main", Commit: "c0ffee", Created: before, State: inventory.PlanFailed, Tier: 0, Tiers: [][]string{{"a"}, {"b"}}, Note: "a failed to build in tier 0", Modules: map[string]*inventory.PlanModule{"a": {State: "failed", AskedAt: &before, Build: "build-1", Why: link.KilledByHand}}} if err := open.inventory.SavePlan(ctx, &failed); err != nil { t.Fatal(err) } said, err := retryPlan(ctx, open, failed.ID) if err != nil { t.Fatal(err) } p, err := open.inventory.PlanByID(ctx, failed.ID) if err != nil { t.Fatal(err) } a := p.Modules["a"] if p.State != inventory.PlanBuilding || a.State != "asked" || a.Build == "" || a.Build == "build-1" || a.Why != "" { t.Fatalf("after retry the plan is %s and a is %+v", p.State, a) } if !strings.Contains(said, a.Build) { t.Errorf("retry does not say the new id: %q", said) } if len(*asked) != 1 { t.Fatalf("asked %v", *asked) } // The killed build's own late outcome is not this ask's; the new one's is, and tier 1 follows. planBuilt(ctx, open, "a", "c0ffee", link.KilledByHand, before, "build-1") if p, _ = open.inventory.PlanByID(ctx, failed.ID); p.State != inventory.PlanBuilding { t.Fatalf("the old ask's outcome failed the retried plan: %s %q", p.State, p.Note) } asking, _ := link.BuildAskedAt(a.Build) planBuilt(ctx, open, "a", "c0ffee", "", asking, a.Build) p, err = open.inventory.PlanByID(ctx, failed.ID) if err != nil { t.Fatal(err) } if p.Tier != 1 || p.Modules["b"] == nil || p.Modules["b"].State != "asked" || p.Modules["b"].Build == "" { t.Fatalf("the retried plan did not go on to tier 1: tier %d, %s, b %+v", p.Tier, p.State, p.Modules["b"]) } if len(*asked) != 2 { t.Fatalf("asked %v", *asked) } } // 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) 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); 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))}); 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}); 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. func TestARebuildJoinsThePlanHoldingTheModule(t *testing.T) { open := aMesh(t) ctx := t.Context() asked := asksRecorded(t) twoTiers(t, open) before := time.Now().UTC().Add(-time.Hour) failed := inventory.Plan{ID: "plan-join", Repository: "novox/a", Branch: "main", Commit: "c0ffee", Created: before, State: inventory.PlanFailed, Tiers: [][]string{{"a"}, {"b"}}, Note: "a failed to build in tier 0", Modules: map[string]*inventory.PlanModule{"a": {State: "failed", AskedAt: &before, Build: "build-1", Why: link.CancelledByHand}}} if err := open.inventory.SavePlan(ctx, &failed); err != nil { t.Fatal(err) } if err := rebuildCommand(ctx, []string{"a"}); err != nil { t.Fatal(err) } p, err := open.inventory.PlanByID(ctx, failed.ID) if err != nil { t.Fatal(err) } if p.State != inventory.PlanBuilding || p.Modules["a"].State != "asked" || p.Modules["a"].Build == "build-1" { t.Fatalf("the rebuild did not join the failed plan: %s %+v", p.State, p.Modules["a"]) } if len(*asked) != 1 { t.Fatalf("asked %v", *asked) } // b is in a tier not yet reached: nothing holds it in its current tier, so it is asked alone. if err := rebuildCommand(ctx, []string{"b"}); err != nil { t.Fatal(err) } if p, _ = open.inventory.PlanByID(ctx, failed.ID); p.Modules["b"] != nil { t.Fatalf("a module the plan has not reached joined it: %+v", p.Modules["b"]) } if len(*asked) != 2 { t.Fatalf("asked %v", *asked) } } // 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. func aBuildQueue(t *testing.T) (*broker.JetStream, func(id, repository string) link.BuildRequest) { t.Helper() url := testbus.URL(t) t.Setenv(broker.NATSVar, url) js, err := broker.Dial(url) if err != nil { t.Fatal(err) } t.Cleanup(js.Close) seat := broker.DeclaredSeat{Name: link.TheBuildMachine, Accepts: []string{"build"}, Emits: []string{"started", "built", "log.*", "paused.*"}} if err := broker.AssertMeshStreams(js); err != nil { t.Fatal(err) } _ = js.Context().DeleteStream("SEAT_NODE_BUILD_AGENT") if err := broker.RaiseSeats(js, []broker.DeclaredSeat{seat}, map[string]broker.Holder{ link.TheBuildMachine: {Node: "anchor", Module: "build-agent"}}); err != nil { t.Fatal(err) } if err := broker.RaiseCancelledSets(js, []broker.DeclaredSeat{seat}); err != nil { t.Fatal(err) } t.Cleanup(func() { _ = js.Context().DeleteStream("SEAT_NODE_BUILD_AGENT") _ = js.Context().DeleteKeyValue(broker.CancelledSetName(link.TheBuildMachine)) _ = js.Context().PurgeStream(broker.EventsStream) }) ask := func(id, repository string) link.BuildRequest { r := link.BuildRequest{ID: id, Repository: repository, Held: map[string]string{"x/y": "secret-ish"}} body, _ := json.Marshal(r) if _, err := js.Context().Publish(link.BuildWork(), body); err != nil { t.Fatal(err) } return r } return js, ask } // 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. func TestCancelDeletesTheAskAndFailsThePlanThatAskedIt(t *testing.T) { js, ask := aBuildQueue(t) open := aMesh(t) ctx := t.Context() twoTiers(t, open) id := link.NewBuildID(time.Now()) ask(id, "https://forge.example/novox/a.git") asked, _ := link.BuildAskedAt(id) plan := inventory.Plan{ID: "plan-cancel", Repository: "novox/a", Commit: "c0ffee", Created: asked, State: inventory.PlanBuilding, Tiers: [][]string{{"a"}, {"b"}}, Modules: map[string]*inventory.PlanModule{"a": {State: "asked", AskedAt: &asked, Build: id}}} if err := open.inventory.SavePlan(ctx, &plan); err != nil { t.Fatal(err) } q, err := link.ReadQueue(ctx, js, link.TheBuildMachine) if err != nil { t.Fatal(err) } if len(q.Asks) != 1 || q.Asks[0].State != link.AskWaiting || q.Asks[0].ID != id { t.Fatalf("the queue reads %+v", q) } if text := queueText(q, time.Now()); !strings.Contains(text, "1 waiting, 0 in flight, 0 dead") || strings.Contains(text, "secret-ish") { t.Errorf("the queue says:\n%s", text) } if err := cancelCommand(ctx, []string{id}); err != nil { t.Fatal(err) } if q, _ = link.ReadQueue(ctx, js, link.TheBuildMachine); len(q.Asks) != 0 { t.Fatalf("the ask is still queued: %+v", q.Asks) } if cancelled, err := link.IsCancelled(js.Conn(), link.TheBuildMachine, id); err != nil || !cancelled { t.Errorf("the cancelled set does not hold it: %v %v", cancelled, err) } b, found, err := open.inventory.BuildByID(ctx, id) if err != nil || !found || b.Failed != link.CancelledByHand { t.Fatalf("the cancel is recorded as %+v (%v %v)", b, found, err) } p, err := open.inventory.PlanByID(ctx, plan.ID) if err != nil { t.Fatal(err) } if p.State != inventory.PlanFailed || p.Modules["a"].State != "failed" || p.Modules["a"].Why != link.CancelledByHand { t.Fatalf("the plan that asked is %s, a %+v", p.State, p.Modules["a"]) } if err := cancelCommand(ctx, []string{id}); err == nil { t.Error("an ask cancelled already was cancelled again") } } // clear cancels every waiting ask and leaves the dead ones unless told, and never one in flight. func TestClearCancelsTheWaitingAndTheDeadOnlyWhenTold(t *testing.T) { js, ask := aBuildQueue(t) open := aMesh(t) ctx := t.Context() start := time.Now() dead := link.NewBuildID(start) ask(dead, "https://forge.example/novox/dead.git") deadOne(t, js) var waiting []string for i := 1; i <= 2; i++ { id := link.NewBuildID(start.Add(time.Duration(i) * time.Millisecond)) waiting = append(waiting, id) ask(id, fmt.Sprintf("https://forge.example/novox/w%d.git", i)) } q, err := link.ReadQueue(ctx, js, link.TheBuildMachine) if err != nil { t.Fatal(err) } if len(q.Of(link.AskDead)) != 1 || len(q.Of(link.AskWaiting)) != 2 || q.MaxDeliver != 5 { t.Fatalf("the queue reads %+v", q) } said, err := clearQueue(ctx, js, open, link.TheBuildMachine, false) if err != nil { t.Fatal(err) } if !strings.Contains(said, "2 waiting ask(s) cancelled") || !strings.Contains(said, "1 dead, left") { t.Errorf("clear said:\n%s", said) } q, _ = link.ReadQueue(ctx, js, link.TheBuildMachine) if len(q.Asks) != 1 || q.Asks[0].ID != dead || q.Asks[0].State != link.AskDead { t.Fatalf("after clear the queue is %+v", q.Asks) } if _, err := clearQueue(ctx, js, open, link.TheBuildMachine, true); err != nil { t.Fatal(err) } if q, _ = link.ReadQueue(ctx, js, link.TheBuildMachine); len(q.Asks) != 0 { t.Fatalf("clear --dead left %+v", q.Asks) } for _, id := range append(waiting, dead) { if b, found, _ := open.inventory.BuildByID(ctx, id); !found || b.Failed != link.CancelledByHand { t.Errorf("%s is recorded as %+v", id, b) } } } // An ask in flight is not cancelled: kill ends it where it runs. func TestCancelRefusesAnAskInFlight(t *testing.T) { js, ask := aBuildQueue(t) open := aMesh(t) ctx := t.Context() id := link.NewBuildID(time.Now()) ask(id, "https://forge.example/novox/a.git") 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: id, On: "ace", At: time.Now().UTC().Format(time.RFC3339Nano)}) if _, err := js.Context().Publish(link.BuildStarted(), started); err != nil { t.Fatal(err) } q, err := link.ReadQueue(ctx, js, link.TheBuildMachine) if err != nil { t.Fatal(err) } a, _ := q.Find(id) if a.State != link.AskInFlight || a.On != "ace" { t.Fatalf("the taken ask reads %+v", a) } _, err = cancelAsk(ctx, js, open, link.TheBuildMachine, a) if err == nil || !strings.Contains(err.Error(), "kill "+id) { t.Fatalf("an ask in flight was cancelled: %v", err) } if _, found, _ := open.inventory.BuildByID(ctx, id); found { t.Error("a refused cancel recorded an outcome") } } // 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. func TestKillAndPauseAskTheHolderOnTheMachine(t *testing.T) { js, _ := aBuildQueue(t) ctx := t.Context() asked := map[string]string{} handlers := map[string]link.ToolHandler{ "kill": func(_ context.Context, raw json.RawMessage) (any, error) { var args struct{ ID string } _ = json.Unmarshal(raw, &args) asked["kill"] = args.ID if args.ID != "build-running" { return nil, fmt.Errorf("ace is not building %s", args.ID) } return map[string]any{"said": "killed " + args.ID}, nil }, "pause": func(context.Context, json.RawMessage) (any, error) { asked["pause"] = "ace" return map[string]any{"said": "ace is paused"}, nil }, } stop, err := link.OverNATS{Conn: js.Conn()}.ServeNodeSeatTools(link.TheBuildMachine, "ace", handlers, nil) if err != nil { t.Fatal(err) } defer stop() for _, id := range []string{"build-running", "build-other"} { started, _ := json.Marshal(link.BuildStart{ID: id, On: "ace", At: time.Now().UTC().Format(time.RFC3339Nano)}) if _, err := js.Context().Publish(link.BuildStarted(), started); err != nil { t.Fatal(err) } } if err := killCommand(ctx, []string{"build-running"}); err != nil { t.Fatal(err) } if asked["kill"] != "build-running" { t.Fatalf("the holder was asked %v", asked) } if err := killCommand(ctx, []string{"build-other"}); err == nil { t.Error("the holder's refusal was not the command's") } if err := killCommand(ctx, []string{"build-never"}); err == nil || !strings.Contains(err.Error(), "cancel build-never") { t.Errorf("a build nobody started: %v", err) } if err := pauseCommand(ctx, "pause", []string{"ace"}); err != nil || asked["pause"] != "ace" { t.Fatalf("pause: %v %v", err, asked) } if err := pauseCommand(ctx, "resume", []string{"g14"}); err == nil { t.Error("a machine nothing answers on was resumed") } } // 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. func TestAPlanStoppedAtItsFirstMachineIsRetried(t *testing.T) { withTheGate(t) // the first-machine gate's own path (novox/hq ADR 0360 keeps it for earlier walks) open := aMesh(t) ctx := t.Context() asked := asksRecorded(t) twoTiers(t, open) var sentTo [][]string was := sendRollout sendRollout = func(_ context.Context, _ *stores, names []string) ([]string, error) { sentTo = append(sentTo, names) return names, nil } t.Cleanup(func() { sendRollout = was }) // a records: a person's choice, since the default rolls out (novox/hq ADR 0236). if err := open.inventory.SetUpgradeOf(ctx, "a", inventory.Upgrade{Why: "test"}); err != nil { t.Fatal(err) } long := time.Now().UTC().Add(-2 * time.Hour) stopped := inventory.Plan{ID: "plan-rollout", Repository: "novox/a", Branch: "main", Commit: "c0ffee", Created: long, State: inventory.PlanFailed, Tiers: [][]string{{"a"}, {"b"}}, Note: "a stopped at its first machine in tier 0: laptop refused what it was sent", // Asked by the plan, as every module a plan holds built is (novox/hq issue 754). Modules: map[string]*inventory.PlanModule{"a": {State: "built", AskedAt: &long, Build: "build-a", BuiltAt: &long, Commit: "c0ffee", First: []string{"laptop"}, FirstAt: &long, Why: "laptop refused what it was sent"}}} if err := open.inventory.SavePlan(ctx, &stopped); err != nil { t.Fatal(err) } // A newer plan holding a refuses it: sending the older build would put it back. newer := inventory.Plan{ID: "plan-newer", Repository: "novox/other", Commit: "d00d", Created: long.Add(time.Hour), State: inventory.PlanDone, Tiers: [][]string{{"a"}}, Modules: map[string]*inventory.PlanModule{}} if err := open.inventory.SavePlan(ctx, &newer); err != nil { t.Fatal(err) } if _, err := retryPlan(ctx, open, stopped.ID); err == nil || !strings.Contains(err.Error(), "plan-newer") { t.Fatalf("retried under a newer plan: %v", err) } newer.State = inventory.PlanSuperseded if err := open.inventory.SavePlan(ctx, &newer); err != nil { t.Fatal(err) } said, err := retryPlan(ctx, open, stopped.ID) if err != nil { t.Fatal(err) } if len(sentTo) != 1 || !reflect.DeepEqual(sentTo[0], []string{"laptop"}) || !strings.Contains(said, "laptop") { t.Fatalf("sent %v; said %q", sentTo, said) } p, err := open.inventory.PlanByID(ctx, stopped.ID) if err != nil { t.Fatal(err) } a := p.Modules["a"] if p.State != inventory.PlanRolling || a.FirstAt == nil || !a.FirstAt.After(long) || a.Why != "" { t.Fatalf("after retry the plan is %s, a %+v", p.State, a) } // And it goes on: a records (its policy sends nothing more), so the next tier is asked. advancePlans(ctx, open) if p, _ = open.inventory.PlanByID(ctx, stopped.ID); p.Tier != 1 || p.Modules["b"] == nil || p.Modules["b"].State != "asked" { t.Fatalf("the retried plan did not go on: tier %d %s %+v", p.Tier, p.State, p.Modules["b"]) } if len(*asked) != 1 { t.Fatalf("asked %v", *asked) } } // 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). func TestAPlanIsAnsweredOnlyByTheBuildItAskedFor(t *testing.T) { open := aMesh(t) ctx := t.Context() asked := time.Now().UTC().Add(-time.Minute) plan := inventory.Plan{ID: "plan-own", Repository: "novox/a", Commit: "c0ffee", Created: asked, State: inventory.PlanBuilding, Tiers: [][]string{{"a"}}, Modules: map[string]*inventory.PlanModule{"a": {State: "asked", AskedAt: &asked, Build: "build-own"}}} if err := open.inventory.SavePlan(ctx, &plan); err != nil { t.Fatal(err) } planBuilt(ctx, open, "a", "0ldc0mm1t", "", time.Now().UTC(), "build-replay") p, _ := open.inventory.PlanByID(ctx, plan.ID) if p.Modules["a"].State != "asked" { t.Fatalf("a replay asked after the plan settled it: %+v", p.Modules["a"]) } // From the records too: a later build of the module recorded is not the plan's. recorded := map[string][]inventory.Build{"a": {{ID: "build-replay", Commit: "0ldc0mm1t", Asked: time.Now(), At: time.Now()}}} if settleFromRecords(&p, p.Tiers[0], recorded, nil) { t.Fatalf("the records settled it with another build: %+v", p.Modules["a"]) } planBuilt(ctx, open, "a", "c0ffee", "", asked, "build-own") if p, _ = open.inventory.PlanByID(ctx, plan.ID); p.Modules["a"].State != "built" || p.Modules["a"].Commit != "c0ffee" { t.Fatalf("its own build did not settle it: %+v", p.Modules["a"]) } } // rebuild of a build made at a commit asks what the module follows now, never the commit. func TestARebuildOfACommitAsksWhatTheModuleFollows(t *testing.T) { open := aMesh(t) ctx := t.Context() asked := asksRecorded(t) twoTiers(t, open) var refs []string was := askABuild askABuild = func(c context.Context, source buildSource, path, ref string) (string, error) { refs = append(refs, ref) return was(c, source, path, ref) } if err := open.inventory.RecordBuild(ctx, inventory.Build{ID: "build-at-commit", Repository: "novox/a", Ref: "0123456789abcdef0123456789abcdef01234567", Module: "a", Commit: "0123456789abcdef0123456789abcdef01234567"}); err != nil { t.Fatal(err) } if err := rebuildCommand(ctx, []string{"build-at-commit"}); err != nil { t.Fatal(err) } if len(refs) != 1 || refs[0] != "main" || len(*asked) != 1 { t.Fatalf("asked %v at %v", *asked, refs) } } // 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. func TestCancelOfAnAskNobodySaidIsWithdrawnWhenItStarts(t *testing.T) { js, ask := aBuildQueue(t) open := aMesh(t) ctx := t.Context() was := holderLooks holderLooks = 300 * time.Millisecond t.Cleanup(func() { holderLooks = was }) id := link.NewBuildID(time.Now()) ask(id, "https://forge.example/novox/a.git") 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) } q, err := link.ReadQueue(ctx, js, link.TheBuildMachine) if err != nil { t.Fatal(err) } a, _ := q.Find(id) if a.State != link.AskInFlight || a.On != "" { t.Fatalf("the taken ask reads %+v", a) } // The holder that took it says it started, while the cancel waits. go func() { time.Sleep(100 * time.Millisecond) started, _ := json.Marshal(link.BuildStart{ID: id, On: "ace", At: time.Now().UTC().Format(time.RFC3339Nano)}) _, _ = js.Context().Publish(link.BuildStarted(), started) }() if _, err := cancelAsk(ctx, js, open, link.TheBuildMachine, a); err == nil || !strings.Contains(err.Error(), "started on ace") { t.Fatalf("a build that started was cancelled: %v", err) } if cancelled, _ := link.IsCancelled(js.Conn(), link.TheBuildMachine, id); cancelled { t.Error("the withdrawn cancel is still marked") } if _, found, _ := open.inventory.BuildByID(ctx, id); found { t.Error("a withdrawn cancel recorded an outcome") } }