package main import ( "context" "encoding/json" "errors" "flag" "fmt" "os" "slices" "sort" "strings" "time" "github.com/nats-io/nats.go" "github.com/novox/mesh-controller/internal/catalogue" "github.com/novox/mesh-controller/internal/inventory" "github.com/novox/mesh-controller/internal/link" ) // The controller's part in a delivery (novox/hq ADR 0239, to-be 47). // // **A delivery is mesh-delivery's; the walk is the controller's.** A delivery is one commit in one // repository from its pull request's head to every machine, and a delivery group a few of them sharing a // branch name; their states, order, holds and record belong to the module holding the `mesh-delivery` // seat. The controller keeps what it already owned — the planner, the gate, registration, sending, the // rollback — and the **walk**: one trunk commit's plan, tier by tier across the machines, one machine // first and judged at the gate (ADR 0236). What changes here is when a walk starts. // // - A walk that moves the controller, the node-engine, the node tools, the bus or mesh-delivery itself is // **on the controller's own path**: started by the merge, as every plan was. mesh-delivery cannot gate // or deliver itself, and nothing the core needs to be repaired may wait for it. // - Any other walk, **while the seat has a holder on record**, is opened by the merge and waits — nothing // asked, nothing registered, nothing sent — until mesh-delivery says `deliver`, or a person says // `plans go`. Nothing of it is registered before then, so no other send can carry it to a machine // before its turn (ADR 0236 §4a carries everything registered). // - With no holder on record, every walk starts at the merge, exactly as before this existed. // // The verbs mesh-delivery asks with are here: `delivery plan` (the planner's one answer for a diffset), // `delivery order` (a group's order from the graph), `delivery check` (a group's heads composed), `delivery // go`, `delivery stop` and `delivery walks`. // onTheControllersPath are the modules whose walk never waits for the delivery's owner, and what each is. var onTheControllersPath = map[string]string{ "mesh-controller": "the controller", "mesh-host": "the node-engine", "node-tools": "the node tools", "nats": "the bus", catalogue.DeliverySeat: "the delivery's owner", } // deliverySeatHeld is whether a module claiming the delivery seat is assigned somewhere: the seat has a // holder on record. Read from the catalogue the merge handler already holds; never from whether the // holder answers, so a walk does not start by itself because mesh-delivery is down — that wait is said. func deliverySeatHeld(entries []inventory.Entry) bool { for _, e := range entries { if len(e.On) > 0 && e.Manifest.ClaimsSeat(catalogue.DeliverySeat) { return true } } return false } // awaitsFor is what a new walk waits for: nil when it starts at once — no holder on record, or a module // on the controller's own path among those it moves. func awaitsFor(entries []inventory.Entry, modules []string) *inventory.PlanDelivery { if !deliverySeatHeld(entries) { return nil } for _, m := range modules { if _, own := onTheControllersPath[m]; own { return nil } } return &inventory.PlanDelivery{Awaits: catalogue.DeliverySeat} } // waitingNote is what a walk waiting for its word says, wherever it is read. func waitingNote(p inventory.Plan) string { return fmt.Sprintf("published; its walk waits for %s's word — `plans go %s --why …` starts it by hand", p.Delivery.Awaits, p.ID) } // waitingSaid is what a walk waiting for its word says when the delivery's owner said what holds it (novox/hq // issue 529): a delivery held for a person, its reason and the verb that releases it — the person's decision, not // a hand start — or the deliveries not at their turn. Told nothing, it is waitingNote. func waitingSaid(p inventory.Plan, lines []stalledLine) string { if len(lines) == 0 { return waitingNote(p) } var held, unheld, other []string for _, l := range lines { switch l.State { case "held": held = append(held, l.Says) case "unheld": unheld = append(unheld, l.Says) default: other = append(other, l.Says) } } switch { case len(held) > 0: return fmt.Sprintf("published; its walk waits for %s's word, and a delivery it carries is held for a person: %s", p.Delivery.Awaits, strings.Join(held, "; ")) case len(other) > 0: // Not at its turn: it goes when its turn comes. A hand start here would skip its group's order (review of // novox/hq issue 529), so none is offered. return fmt.Sprintf("published; its walk waits for %s's word: %s", p.Delivery.Awaits, strings.Join(other, "; ")) } // Held by nothing: nothing will give it its word, so the hand start is the way on. return fmt.Sprintf("published; its walk waits for %s's word: %s — `plans go %s --why …` starts it by hand", p.Delivery.Awaits, strings.Join(unheld, "; "), p.ID) } // letGo gives a waiting walk its word: by the delivery's owner, or a person. The next advance asks its // first tier. Refused for a walk that does not wait. func letGo(ctx context.Context, inv *inventory.Inventory, id, by, why string) (inventory.Plan, error) { release, err := inv.HoldPlans(ctx, true) if err != nil { return inventory.Plan{}, err } defer release() p, err := inv.PlanByID(ctx, id) if err != nil { return inventory.Plan{}, err } switch { case !p.Open(): return p, fmt.Errorf("%s is %s: there is no walk to start", p.ID, p.State) case p.Delivery == nil || p.Delivery.Awaits == "": return p, fmt.Errorf("%s waits for nobody: it started at its merge", p.ID) case p.Delivery.Go != nil: return p, fmt.Errorf("%s was let go by %s at %s already", p.ID, p.Delivery.By, p.Delivery.Go.Local().Format("15:04:05")) } // **One walk at a time** (novox/hq ADR 0276): the word is taken, and the walk starts once no other walk is // started (advanceOnce), so the delivery's owner says it once and is not refused. now := time.Now().UTC() p.Delivery.Go, p.Delivery.By, p.Delivery.Why = &now, by, why p.Note = "let go by " + by + "; its first tier is asked next" if err := inv.SavePlan(ctx, &p); err != nil { return p, err } return p, nil } // stopWalk ends a walk on its delivery's word: failed, said as stopped by whom, why. What it asked still // builds and registers; nothing further is asked or sent. func stopWalk(ctx context.Context, inv *inventory.Inventory, id, by, why string) (inventory.Plan, error) { p, _, err := stopWalkWithout(ctx, inv, id, by, why, nil, time.Now().UTC()) return p, err } // stopAnswer is what a stop that leaves merges out says, as data (novox/hq ADR 0299): the record stopped, the // batch that walks its other merges again with those merges, every merge left out, the modules the stopped walk // already sent that stay on the machines named, the modules the batch's walk withholds, and every merge elsewhere // that will deliver a stopped change when it is walked — a later merge of its repository, or one moving its modules. type stopAnswer struct { Stopped string `json:"stopped"` WalkedAgainBy string `json:"walked_again_by,omitempty"` WalkedAgain []string `json:"walked_again,omitempty"` LeftOut []inventory.PlanLeftOut `json:"left_out,omitempty"` Kept []inventory.PlanKept `json:"kept,omitempty"` Withheld []string `json:"withheld,omitempty"` StillCarriedBy []string `json:"still_carried_by,omitempty"` } // stopWalkWithout ends a walk on its delivery's word and, given merges to leave out, walks every other merge it // answered again without them (novox/hq issue 459, ADR 0299): one walk carries every merge of its batch, so ending // it for one group member ended every other delivery on it. // // - Left out are the merges named, each later merge of their repository's branch, which contains one, and each // merge moving a module a left-out merge moves: a module is built at its branch, which holds the stopped merge. // - The others join the open batch of their kind — merges heard while the walk ran wait there, and an older // commit never walks after a newer one — or a new batch when none is open; its window is closed, and its walk // withholds the left-out merges' modules where it would build them only as dependents. // - Given a batch not yet cut, the merges left out leave it for a stopped record of their own, and the batch // goes on without them; with nothing left in it, the batch itself is stopped. // - The stopped record, the batch and every merge moved are written in one transaction (decision 7). // // With nothing left to walk, or no merge named, the walk is only ended. A merge named that the record does not // answer refuses the stop whole. func stopWalkWithout(ctx context.Context, inv *inventory.Inventory, id, by, why string, without []string, now time.Time) (inventory.Plan, stopAnswer, error) { said := stopAnswer{Stopped: id} release, err := inv.HoldPlans(ctx, true) if err != nil { return inventory.Plan{}, said, err } defer release() p, err := inv.PlanByID(ctx, id) if err != nil { return inventory.Plan{}, said, err } batch := p.Batch() if !p.Open() && !(batch && len(without) > 0) { return p, said, fmt.Errorf("%s is already %s", p.ID, p.State) } var merges, again []inventory.BatchedMerge if len(without) > 0 { if merges, err = inv.MergesOf(ctx, p.ID); err != nil { return p, said, err } if again, said.LeftOut, err = leaveOut(merges, without); err != nil { return p, said, fmt.Errorf("%s is not stopped: %w", p.ID, err) } } stopped := func(r *inventory.Plan) { if r.Delivery == nil { r.Delivery = &inventory.PlanDelivery{} } r.Delivery.Stopped, r.Delivery.StoppedWhy, r.Delivery.LeftOut = by, why, said.LeftOut r.State = inventory.PlanFailed r.Note = fmt.Sprintf("stopped by %s at tier %d: %s", by, r.Tier, why) } withheld := movesOf(leftOutMerges(merges, said.LeftOut)) said.Withheld = withheld if said.StillCarriedBy, err = stillCarried(ctx, inv, p.ID, merges, said.LeftOut, withheld); err != nil { return p, said, err } if batch { return p, said, stopInBatch(ctx, inv, &p, merges, again, &said, stopped, now) } entries, err := inv.Catalogued(ctx) if err != nil { return p, said, err } running := map[string][]string{} for _, e := range entries { running[e.Manifest.Module] = e.On } said.Kept = keptOf(p, movesOf(again), running) stopped(&p) p.Delivery.Kept = said.Kept if len(again) == 0 { if len(said.LeftOut) > 0 { p.Note += fmt.Sprintf("; left out %s, nothing else to walk", mergesLeftOutWords(said.LeftOut)) } return p, said, inv.SavePlan(ctx, &p) } target, err := batchToWalkAgain(ctx, inv, again, now) if err != nil { return p, said, fmt.Errorf("%s is not stopped: %w", p.ID, err) } target.Delivery.Batch.Closed = true target.Delivery.Batch.Withheld = union(target.Delivery.Batch.Withheld, withheld) sort.Strings(target.Delivery.Batch.Withheld) said.WalkedAgainBy = target.ID var moves []inventory.MergeMove for _, m := range again { said.WalkedAgain = append(said.WalkedAgain, m.Repository+"@"+m.Commit) moves = append(moves, inventory.MergeMove{Repository: m.Repository, Commit: m.Commit, Plan: target.ID, Alone: m.Alone}) } p.Delivery.WalkedAgainBy = target.ID p.Note += fmt.Sprintf("; its other merges are walked again by %s: %s", target.ID, strings.Join(said.WalkedAgain, ", ")) if len(said.LeftOut) > 0 { p.Note += "; left out " + mergesLeftOutWords(said.LeftOut) } if len(said.Kept) > 0 { p.Note += "; " + keptWords(said.Kept) } // The batch first, so no merge names a plan the store does not hold; all of it, or nothing (decision 7). if err := inv.SavePlansMoving(ctx, []*inventory.Plan{&target, &p}, moves); err != nil { return p, said, fmt.Errorf("%s is not stopped, and none of its merges moved: %w", p.ID, err) } // What the batch holds and names now: written again by the cutter's next look should this fail. if err := keepBatch(ctx, inv, &target, now, ""); err != nil { fmt.Printf(" %s holds its merges; its record is written again at the next look: %v\n", target.ID, err) } fmt.Printf("%s stopped by %s; %s walks its other merges again: %s\n", p.ID, by, target.ID, strings.Join(said.WalkedAgain, ", ")) return p, said, nil } // stopInBatch leaves merges out of a batch not yet cut (ADR 0299): they go to a stopped record of their own, which // names the batch as walking the others, and the batch withholds their modules; a batch left with none is stopped // itself. func stopInBatch(ctx context.Context, inv *inventory.Inventory, b *inventory.Plan, merges, again []inventory.BatchedMerge, said *stopAnswer, stopped func(*inventory.Plan), now time.Time) error { if len(again) == 0 { stopped(b) b.Delivery.Batch = nil b.Note = "a batch whose every merge was left out by a stop: " + b.Note return inv.SavePlan(ctx, b) } left := leftOutMerges(merges, said.LeftOut) first := left[0] r := inventory.Plan{ID: fmt.Sprintf("plan-%d", now.UnixNano()), Repository: first.Repository, Branch: first.Branch, Commit: first.Commit, Merged: first.Merged, Created: now, Tiers: [][]string{}, Modules: map[string]*inventory.PlanModule{}} if r.ID == b.ID { r.ID += "-left" } stopped(&r) r.Delivery.WalkedAgainBy = b.ID r.Note += fmt.Sprintf("; left out of %s before it was walked: %s; %s walks the others", b.ID, mergesLeftOutWords(said.LeftOut), b.ID) b.Delivery.Batch.Withheld = union(b.Delivery.Batch.Withheld, said.Withheld) sort.Strings(b.Delivery.Batch.Withheld) var moves []inventory.MergeMove for _, m := range left { moves = append(moves, inventory.MergeMove{Repository: m.Repository, Commit: m.Commit, Plan: r.ID, Alone: m.Alone}) } for _, m := range again { said.WalkedAgain = append(said.WalkedAgain, m.Repository+"@"+m.Commit) } said.Stopped, said.WalkedAgainBy = r.ID, b.ID if err := inv.SavePlansMoving(ctx, []*inventory.Plan{&r, b}, moves); err != nil { return fmt.Errorf("%s is not changed, and none of its merges moved: %w", b.ID, err) } if err := keepBatch(ctx, inv, b, now, ""); err != nil { fmt.Printf(" %s holds its merges; its record is written again at the next look: %v\n", b.ID, err) } fmt.Printf("%s: %s left out by a stop, kept by %s\n", b.ID, mergesLeftOutWords(said.LeftOut), r.ID) return nil } // batchToWalkAgain is the batch a stopped walk's merges join: the open batch of their kind, or a new one, not yet // written. The kind is read from what the merges moved. func batchToWalkAgain(ctx context.Context, inv *inventory.Inventory, again []inventory.BatchedMerge, now time.Time) (inventory.Plan, error) { own := false for _, n := range movesOf(again) { if _, on := onTheControllersPath[n]; on { own = true } } batches, err := inv.Batches(ctx) if err != nil { return inventory.Plan{}, err } for _, b := range batches { if b.OwnPath() == own { if b.Delivery == nil { b.Delivery = &inventory.PlanDelivery{} } if b.Delivery.Batch == nil { b.Delivery.Batch = &inventory.PlanBatch{Own: own} } return b, nil } } first := again[0] return inventory.Plan{ID: fmt.Sprintf("plan-%d", now.UnixNano()), Repository: first.Repository, Branch: first.Branch, Commit: first.Commit, Merged: first.Merged, Created: now, State: inventory.PlanAssembling, Tiers: [][]string{}, Modules: map[string]*inventory.PlanModule{}, Delivery: &inventory.PlanDelivery{Batch: &inventory.PlanBatch{Own: own}}}, nil } // leaveOut splits a walk's merges by the merges named to leave out, each `@` (a commit's // first characters will do): the merges walked again, and those left out — each named one; each later merge of // its repository's branch, which contains it; and each merge moving a module a left-out merge moves, which is // built at its branch and so holds the stopped merge — until none more is. A name the walk does not answer is // refused. func leaveOut(merges []inventory.BatchedMerge, without []string) ([]inventory.BatchedMerge, []inventory.PlanLeftOut, error) { var stopped []inventory.BatchedMerge for _, w := range without { repository, commit, ok := strings.Cut(strings.TrimSpace(w), "@") if !ok || repository == "" || len(commit) < 7 { return nil, nil, fmt.Errorf("%q names no merge: @", w) } i := slices.IndexFunc(merges, func(m inventory.BatchedMerge) bool { return strings.EqualFold(m.Repository, repository) && strings.HasPrefix(m.Commit, commit) }) if i < 0 { return nil, nil, fmt.Errorf("it answers no merge %s@%s", repository, commit) } stopped = append(stopped, merges[i]) } same := func(a, b inventory.BatchedMerge) bool { return strings.EqualFold(a.Repository, b.Repository) && a.Commit == b.Commit } out := map[int]inventory.PlanLeftOut{} for i, m := range merges { if slices.ContainsFunc(stopped, func(s inventory.BatchedMerge) bool { return same(s, m) }) { out[i] = inventory.PlanLeftOut{Repository: m.Repository, Commit: m.Commit} } } for grew := true; grew; { grew = false for i, m := range merges { if _, left := out[i]; left { continue } for j, o := range out { l := merges[j] contains := o.Contains if contains == "" { contains = l.Commit } if strings.EqualFold(l.Repository, m.Repository) && l.Branch == m.Branch && laterMerge(m, l) { out[i] = inventory.PlanLeftOut{Repository: m.Repository, Commit: m.Commit, Contains: contains} } else if shared := sharedMoves(m, l); len(shared) > 0 { out[i] = inventory.PlanLeftOut{Repository: m.Repository, Commit: m.Commit, Contains: contains, Shares: shared} } else { continue } grew = true break } } } var again []inventory.BatchedMerge var left []inventory.PlanLeftOut for i, m := range merges { if o, is := out[i]; is { left = append(left, o) } else { again = append(again, m) } } sort.SliceStable(left, func(i, j int) bool { return left[i].Contains == "" && left[j].Contains != "" }) return again, left, nil } // sharedMoves are the modules two kept merges both moved when they were heard. func sharedMoves(a, b inventory.BatchedMerge) []string { var out []string theirs := announcedOf(b).Moves for _, n := range announcedOf(a).Moves { if slices.Contains(theirs, n) && !slices.Contains(out, n) { out = append(out, n) } } sort.Strings(out) return out } // leftOutMerges are the kept merges a stop left out. func leftOutMerges(merges []inventory.BatchedMerge, out []inventory.PlanLeftOut) []inventory.BatchedMerge { var left []inventory.BatchedMerge for _, m := range merges { if slices.ContainsFunc(out, func(o inventory.PlanLeftOut) bool { return strings.EqualFold(o.Repository, m.Repository) && o.Commit == m.Commit }) { left = append(left, m) } } return left } // movesOf is every module kept merges moved when they were heard, sorted. func movesOf(merges []inventory.BatchedMerge) []string { var out []string for _, m := range merges { out = union(out, announcedOf(m).Moves) } sort.Strings(out) return out } // keptOf are the modules a stopped walk sent that no merge walked again moves, and the machines each reached: // those keep running the stopped walk's build. A module sent to its first machines reached those, and the rest it // was sent to after them; one sent with no first machine (its policy sends it to all together, or a later tier is // built by it) reached every machine running it. func keptOf(p inventory.Plan, walkedAgain []string, running map[string][]string) []inventory.PlanKept { var out []inventory.PlanKept for name, m := range p.Modules { if m == nil || slices.Contains(walkedAgain, name) { continue } var machines []string if m.FirstAt != nil || m.SentAt != nil { machines = union(machines, m.First) } if m.SentAt != nil && len(m.First) == 0 { machines = union(machines, running[name]) } if m.Gate != nil && m.Gate.Since != nil { machines = union(machines, m.Gate.Machines) } for machine := range m.Rest { machines = union(machines, []string{machine}) } if len(machines) == 0 { continue } sort.Strings(machines) out = append(out, inventory.PlanKept{Module: name, Machines: machines}) } sort.Slice(out, func(i, j int) bool { return out[i].Module < out[j].Module }) return out } // keptWords is what a stopped walk leaves on the machines, as a person reads it. func keptWords(kept []inventory.PlanKept) string { var words []string for _, k := range kept { words = append(words, k.Module+" on "+strings.Join(k.Machines, ", ")) } return "left running the stopped walk's build, which nothing walks again: " + strings.Join(words, "; ") } // stillCarried are the merges outside the stopped record, in a batch or an open walk, that will deliver a stopped // change when they are walked: a later merge of its repository's branch, which contains it, or one that moves a // module it moves. Holding a merged change back for good is a revert's (ADR 0299). func stillCarried(ctx context.Context, inv *inventory.Inventory, id string, merges []inventory.BatchedMerge, out []inventory.PlanLeftOut, withheld []string) ([]string, error) { var named []inventory.BatchedMerge for _, m := range leftOutMerges(merges, out) { if slices.ContainsFunc(out, func(o inventory.PlanLeftOut) bool { return o.Commit == m.Commit && o.Contains == "" }) { named = append(named, m) } } if len(named) == 0 { return nil, nil } open, err := inv.OpenPlans(ctx) if err != nil { return nil, err } batches, err := inv.Batches(ctx) if err != nil { return nil, err } var carried []string for _, p := range append(open, batches...) { if p.ID == id { continue } ms, err := inv.MergesOf(ctx, p.ID) if err != nil { return nil, err } for _, m := range ms { later := slices.ContainsFunc(named, func(s inventory.BatchedMerge) bool { return strings.EqualFold(s.Repository, m.Repository) && s.Branch == m.Branch && laterMerge(m, s) }) shares := slices.ContainsFunc(announcedOf(m).Moves, func(n string) bool { return slices.Contains(withheld, n) }) if later || shares { carried = append(carried, fmt.Sprintf("%s@%s (in %s)", m.Repository, m.Commit, p.ID)) } } } sort.Strings(carried) return carried, nil } // mergesLeftOutWords is the merges a stop left out, as a person reads them: each that contains a stopped one, or // shares its modules, says so. func mergesLeftOutWords(out []inventory.PlanLeftOut) string { var words []string for _, o := range out { s := o.Repository + "@" + short(o.Commit) switch { case len(o.Shares) > 0: s += " (which moves " + strings.Join(o.Shares, ", ") + " as " + short(o.Contains) + " does)" case o.Contains != "": s += " (which contains " + short(o.Contains) + ")" } words = append(words, s) } return strings.Join(words, ", ") } // orderMember is one member of a group, as mesh-delivery says it. type orderMember struct { ID string `json:"id"` Repository string `json:"repository"` Base string `json:"base,omitempty"` Head string `json:"head,omitempty"` Number int `json:"number,omitempty"` Paths []string `json:"paths,omitempty"` PathsTruncated bool `json:"paths_truncated,omitempty"` Removed []string `json:"removed,omitempty"` ModuleDirs []string `json:"module_dirs,omitempty"` ModuleDirsSaid bool `json:"module_dirs_said,omitempty"` CloneURL string `json:"clone_url,omitempty"` // After are the repositories its pull request says it goes after (`after: `). After []string `json:"after,omitempty"` } // pull is the member as the forge announced it. func (m orderMember) pull() link.PullUpdated { owner, repo, _ := strings.Cut(m.Repository, "/") base := m.Base if base == "" { base = "main" } return link.PullUpdated{Owner: owner, Repo: repo, Number: m.Number, Base: base, Commit: m.Head, CloneURL: m.CloneURL, Paths: m.Paths, PathsTruncated: m.PathsTruncated, Removed: m.Removed, ModuleDirs: m.ModuleDirs, ModuleDirsSaid: m.ModuleDirsSaid} } // orderPair is one "before" among a group's members, why, and the rules it won over. type orderPair struct { Before string `json:"before"` After string `json:"after"` Why string `json:"why"` // Over is the rules that ordered the pair the other way and lost to Why, by precedence (novox/hq ADR // 0249): said, so a person reading the order sees which rule won. Over []string `json:"over,omitempty"` } // orderContradiction is two members that rules of one rank order both ways: no precedence resolves it, so // the group is refused, the rules named, and Said says how a person declares the order. type orderContradiction struct { Members []string `json:"members"` Rules []orderPair `json:"rules"` Said string `json:"said"` } // orderReach is what a member moves, as the planner says. type orderReach struct { Moved []string `json:"moved,omitempty"` Dependents []string `json:"dependents,omitempty"` New []string `json:"new,omitempty"` Manifests []string `json:"manifests,omitempty"` } // groupOrder is a group's order: the members in it, every pair and why, and the cycle when there is one — // with each contradiction no precedence resolves. type groupOrder struct { Order []string `json:"order"` Pairs []orderPair `json:"pairs,omitempty"` Cycle []string `json:"cycle,omitempty"` Contradictions []orderContradiction `json:"contradictions,omitempty"` Reach map[string]orderReach `json:"reach,omitempty"` } // The reasons a pair is ordered, in the words the delivery plan shows. const ( orderDeclared = "declared" orderBuiltBy = "built by" orderVersionSkew = "version skew" orderEngineFirst = "engine before controller" ) // orderRank is a rule's precedence when two rules order one pair both ways (novox/hq ADR 0249), the lower // winning: what a person declared; then what the graph and the change say (built by, version skew); then // the rollout default, which infers a direction from the modules' names alone. func orderRank(why string) int { switch why { case orderDeclared: return 0 case orderBuiltBy, orderVersionSkew: return 1 case orderEngineFirst: return 2 } return 3 } // orderOf is a group's order (novox/hq ADR 0239 decision 4, ADR 0249), pure. A member goes before another // when: // // - its pull request is named in the other's `after:` lines (declared); // - it moves a module the other's moved modules are built by, stand on, package or declare (built by); // - it moves the controller and the other changes any module's manifest (version skew: the newer // controller parses what the newer manifest says) — the node-engine's excepted, which goes first; // - it moves the node-engine and the other moves the controller (the witness reads nothing the controller // does not yet grant, and a controller sends nothing an older engine would refuse — ADR 0236's rollout // order). // // When rules order one pair both ways, the one of higher precedence wins (orderRank) and the pair says // what it won over. Rules of one rank both ways are a contradiction: both are kept as pairs, the members // are the cycle, and the contradiction says how to declare the order. Otherwise, by repository then id, so // every reading gives one order. A cycle through three members or more is named by its members, and none // of them is ordered. func orderOf(members []orderMember, reach map[string]orderReach, edges []inventory.Edge) groupOrder { out := groupOrder{Reach: reach} byID := map[string]orderMember{} for _, m := range members { byID[m.ID] = m } var claims []orderPair add := func(before, after, why string) { if before == after { return } for _, p := range claims { if p.Before == before && p.After == after && p.Why == why { return } } claims = append(claims, orderPair{Before: before, After: after, Why: why}) } named := func(said, repository string) bool { said = strings.TrimSuffix(strings.TrimSpace(said), ".git") if strings.EqualFold(said, repository) { return true } _, repo, _ := strings.Cut(repository, "/") return strings.EqualFold(said, repo) } for _, a := range members { for _, b := range members { if a.ID == b.ID { continue } ra, rb := reach[a.ID], reach[b.ID] for _, said := range b.After { if named(said, a.Repository) { add(a.ID, b.ID, orderDeclared) } } for _, e := range edges { if slices.Contains(ra.Moved, e.To) && slices.Contains(rb.Moved, e.From) && e.From != e.To { add(a.ID, b.ID, orderBuiltBy) break } } if slices.Contains(ra.Moved, "mesh-controller") && len(rb.Manifests) > 0 && !slices.Contains(rb.Moved, "mesh-controller") && !slices.Contains(rb.Moved, "mesh-host") { add(a.ID, b.ID, orderVersionSkew) } if slices.Contains(ra.Moved, "mesh-host") && slices.Contains(rb.Moved, "mesh-controller") && !slices.Contains(ra.Moved, "mesh-controller") { add(a.ID, b.ID, orderEngineFirst) } } } out.Pairs, out.Contradictions = resolveClaims(claims, byID) sort.Slice(out.Pairs, func(i, j int) bool { if out.Pairs[i].Before != out.Pairs[j].Before { return out.Pairs[i].Before < out.Pairs[j].Before } return out.Pairs[i].After < out.Pairs[j].After }) // Kahn's walk, the next always the first by repository and id among those with nothing before them. waiting := map[string]int{} for _, m := range members { waiting[m.ID] = 0 } for _, p := range out.Pairs { waiting[p.After]++ } done := map[string]bool{} for len(done) < len(members) { var ready []orderMember for _, m := range members { if !done[m.ID] && waiting[m.ID] == 0 { ready = append(ready, m) } } if len(ready) == 0 { for _, m := range members { if !done[m.ID] { out.Cycle = append(out.Cycle, m.ID) } } sort.Strings(out.Cycle) return out } sort.Slice(ready, func(i, j int) bool { if ready[i].Repository != ready[j].Repository { return ready[i].Repository < ready[j].Repository } return ready[i].ID < ready[j].ID }) next := ready[0] done[next.ID] = true out.Order = append(out.Order, next.ID) for _, p := range out.Pairs { if p.Before == next.ID { waiting[p.After]-- } } } return out } // orderRules is the rules in the order a pair names them when more than one says the same. var orderRules = []string{orderDeclared, orderBuiltBy, orderVersionSkew, orderEngineFirst} // resolveClaims is the pairs the rules' claims come to (novox/hq ADR 0249): for each two members, the claims // of the highest precedence decide, and the claims they overrule are said on the pair. Claims of one rank // both ways are a contradiction: both pairs are kept, so the walk finds the cycle, and the contradiction // says how a person declares the order. func resolveClaims(claims []orderPair, byID map[string]orderMember) ([]orderPair, []orderContradiction) { byTwo := map[[2]string][]orderPair{} var keys [][2]string for _, c := range claims { k := [2]string{min(c.Before, c.After), max(c.Before, c.After)} if _, ok := byTwo[k]; !ok { keys = append(keys, k) } byTwo[k] = append(byTwo[k], c) } slices.SortFunc(keys, func(a, b [2]string) int { if c := strings.Compare(a[0], b[0]); c != 0 { return c } return strings.Compare(a[1], b[1]) }) ruleIndex := func(why string) int { return slices.Index(orderRules, why) } var pairs []orderPair var contradictions []orderContradiction for _, k := range keys { cs := byTwo[k] slices.SortStableFunc(cs, func(a, b orderPair) int { return ruleIndex(a.Why) - ruleIndex(b.Why) }) top := orderRank(cs[0].Why) var won []orderPair ways := map[string]bool{} for _, c := range cs { if orderRank(c.Why) == top { won = append(won, c) ways[c.Before] = true } } if len(ways) == 1 { p := orderPair{Before: won[0].Before, After: won[0].After, Why: won[0].Why} for _, c := range cs { if c.Before != p.Before && !slices.Contains(p.Over, c.Why) { p.Over = append(p.Over, c.Why) } } pairs = append(pairs, p) continue } // One rank both ways: no precedence resolves it. var first, second *orderPair var said []string for i := range won { c := won[i] if c.Before == k[0] && first == nil { first = &won[i] } if c.Before == k[1] && second == nil { second = &won[i] } said = append(said, fmt.Sprintf("%s puts %s first", c.Why, c.Before)) } pairs = append(pairs, orderPair{Before: first.Before, After: first.After, Why: first.Why}, orderPair{Before: second.Before, After: second.After, Why: second.Why}) a, b := byID[k[0]], byID[k[1]] how := fmt.Sprintf("neither outranks the other: declare the order with a line `after: %s` in %s's "+ "description, or `after: %s` in %s's", a.Repository, pullName(b), b.Repository, pullName(a)) if top == orderRank(orderDeclared) { how = "each pull request declares it goes after the other: remove one of the `after:` lines" } contradictions = append(contradictions, orderContradiction{Members: []string{k[0], k[1]}, Rules: won, Said: fmt.Sprintf("%s and %s: %s; %s", k[0], k[1], strings.Join(said, ", "), how)}) } return pairs, contradictions } // pullName is a member as a person finds it: its pull request, or its id when no number was said. func pullName(m orderMember) string { if m.Number > 0 { return fmt.Sprintf("%s#%d", m.Repository, m.Number) } return m.ID } // reachOfMembers is each member's reach, as the planner says it. func reachOfMembers(members []orderMember, entries []inventory.Entry, read map[string][]inventory.ReadRepository, edges []inventory.Edge) map[string]orderReach { out := map[string]orderReach{} for _, m := range members { s := pullScope(m.pull(), entries, read, edges) out[m.ID] = orderReach{Moved: s.Modules, Dependents: s.Dependents, New: s.New, Manifests: s.Manifests} } return out } // deliveryCommand is `delivery`, the controller's verbs for the delivery's owner: // // delivery plan --repository owner/repo --head [--base main] --paths a,b [--module-dirs …] [--removed …] // delivery order --members // delivery check --group --members // delivery go [--by ] [--why ] // delivery stop --why [--by ] // delivery walks [-n 50] [--plan ] func deliveryCommand(ctx context.Context, args []string) error { if len(args) == 0 { return errors.New("delivery plan|order|check|go|stop|walks") } sub, rest := args[0], args[1:] set := flag.NewFlagSet("delivery "+sub, flag.ContinueOnError) repository := set.String("repository", "", "owner/repository") base := set.String("base", "main", "the branch it merges into") head := set.String("head", "", "the commit at hand") paths := set.String("paths", "", "the files it changes, comma-separated") moduleDirs := set.String("module-dirs", "", "the directories holding a module.json at the head, comma-separated") removed := set.String("removed", "", "the files it deletes, comma-separated") membersJSON := set.String("members", "", "a group's members, as JSON") group := set.String("group", "", "a delivery group's id") by := set.String("by", catalogue.DeliverySeat, "who says it") why := set.String("why", "", "why") without := set.String("without", "", "merges a stop leaves out, @, comma-separated: the walk's "+ "other merges are walked again without them (novox/hq ADR 0299)") limit := set.Int("n", 50, "how many ended walks to answer beside the open ones") planID := set.String("plan", "", "one walk") positionals, err := parseAround(set, rest) if err != nil { return err } var members []orderMember if *membersJSON != "" { if err := json.Unmarshal([]byte(*membersJSON), &members); err != nil { return fmt.Errorf("the members are not readable: %w", err) } } open, err := openStores(ctx) if err != nil { return err } defer open.Close() inv := open.inventory answer := func(v any) error { enc := json.NewEncoder(os.Stdout) enc.SetIndent("", " ") return enc.Encode(v) } switch sub { case "plan": if *repository == "" || *head == "" && *paths == "" { return errors.New("delivery plan --repository owner/repo --head --paths a,b") } m := orderMember{ID: *repository + "@" + *head, Repository: *repository, Base: *base, Head: *head, Paths: splitList(*paths), Removed: splitList(*removed)} if d := moduleDirsOf(*moduleDirs); d != nil { m.ModuleDirs, m.ModuleDirsSaid = *d, true } entries, read, edges, err := theGraph(ctx, inv) if err != nil { return err } s := pullScope(m.pull(), entries, read, edges) plan := changePlanOf(*repository, *base, *head, s.Reach, entries, func(module string) (inventory.Upgrade, bool) { u, err := inv.UpgradeOf(ctx, module) return u, err == nil }) return answer(map[string]any{"plan": plan, "reach": orderReach{Moved: s.Modules, Dependents: s.Dependents, New: s.New, Manifests: s.Manifests}, "mesh": s.Mesh, "gated": s.gated()}) case "order": if len(members) == 0 { return errors.New("delivery order --members : a group has members") } entries, read, edges, err := theGraph(ctx, inv) if err != nil { return err } return answer(orderOf(members, reachOfMembers(members, entries, read, edges), edges)) case "check": 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, func() error { return (following{open}).PullUpdated(ctx, members[0].pull()) }); err != nil { return err } return answer(map[string]any{"rechecked": members[0].ID}) } if *group == "" || len(members) < 2 { return errors.New("delivery check --group --members (two heads or more), or --members ") } id, err := askGroupCheck(ctx, open, *group, members) if err != nil { return err } return answer(map[string]any{"asked": id, "group": *group}) case "go": if len(positionals) != 1 { return errors.New("delivery go [--by ] [--why ]") } var p inventory.Plan if err := sayingOnTheBus(ctx, func() (err error) { p, err = letGo(ctx, inv, positionals[0], *by, strings.TrimSpace(*why)) return err }); err != nil { return err } fmt.Printf("%s (%s at %s) is let go by %s; its first tier is asked at the next pass\n", p.ID, p.Repository, short(p.Commit), *by) return nil case "stop": if len(positionals) != 1 || strings.TrimSpace(*why) == "" { return errors.New("delivery stop --why [--by ] [--without @,…]") } var p inventory.Plan var said stopAnswer if err := sayingOnTheBus(ctx, func() (err error) { p, said, err = stopWalkWithout(ctx, inv, positionals[0], *by, strings.TrimSpace(*why), splitList(*without), time.Now().UTC()) return err }); err != nil { return err } if *without != "" { // What it left out and what it walks again, as data: the delivery's owner reads it (ADR 0299). return answer(said) } fmt.Printf("%s stopped by %s at tier %d of %d; what was asked still builds and registers, nothing further "+ "is asked or sent\n", p.ID, *by, p.Tier, len(p.Tiers)) return nil case "walks": var walks []inventory.Plan if *planID != "" { p, err := inv.PlanByID(ctx, *planID) if err != nil { return err } walks = []inventory.Plan{p} } else { if walks, err = inv.OpenPlans(ctx); err != nil { return err } // And the batch being assembled (novox/hq ADR 0276): never a walk to link or let go, shown. batches, err := inv.Batches(ctx) if err != nil { return err } walks = append(walks, batches...) recent, err := inv.RecentPlans(ctx, *limit) if err != nil { return err } for _, p := range recent { if !slices.ContainsFunc(walks, func(w inventory.Plan) bool { return w.ID == p.ID }) { walks = append(walks, p) } } } entries, err := inv.Catalogued(ctx) if err != nil { return err } // Each walk's phases, with the rest's reports (novox/hq ADR 0282 decision 6). now := time.Now() for i := range walks { walks[i].Phases = walks[i].WalkPhases(now, appliedFrom(ctx, inv)) } return answer(map[string]any{"held": deliverySeatHeld(entries), "walks": walks, "own-path": sortedKeysOf(ownPathWords())}) } return fmt.Errorf("delivery %s: plan, order, check, go, stop or walks", sub) } // appliedFrom answers a machine's first report after a send from the controller's `apply` durations; a lookup // that fails is a report not read, which leaves the walk's end unknown rather than wrong. func appliedFrom(ctx context.Context, inv *inventory.Inventory) inventory.AppliedLookup { return func(node string, sent time.Time) (inventory.AppliedReport, bool) { r, ok, err := inv.FirstAppliedAfter(ctx, node, sent) if err != nil { fmt.Fprintf(os.Stderr, "the report of %s after %s could not be read: %v\n", node, sent.Format(time.RFC3339), err) return inventory.AppliedReport{}, false } return r, ok } } // recordWalkPhases keeps, once, each phase of every walk ended lately whose end is known, as a duration of kind // walk-phase per class (novox/hq ADR 0282 decision 6): what `durations` summarises. A walk whose end is unknown // past ApplySilentAfter keeps its measured phases without its total. func recordWalkPhases(ctx context.Context, inv *inventory.Inventory, now time.Time) error { recent, err := inv.RecentPlans(ctx, 30) if err != nil { return err } for _, p := range recent { if p.State != inventory.PlanDone || p.Release != nil || now.Sub(p.Updated) > 2*time.Hour { continue } anyKept, totalKept, err := inv.WalkPhasesKept(ctx, p.ID) if err != nil { return err } if totalKept { continue } ph := p.WalkPhases(now, appliedFrom(ctx, inv)) if ph != nil && ph.End == nil && anyKept { continue // kept without its end; kept again only once its end is known } if ph == nil || (ph.End == nil && now.Sub(p.Updated) < inventory.ApplySilentAfter+time.Minute) { continue } class := ph.Class if class == "" { class = "unclassed" } for _, x := range ph.Phases { if x.State != inventory.PhaseMeasured || x.Start == nil { continue } if err := inv.RecordDuration(ctx, inventory.Duration{Kind: inventory.DurationWalkPhase, Subject: class + "/" + x.Name, Ref: fmt.Sprintf("%s/%s/%d", p.ID, x.Name, x.Tier), Started: *x.Start, Took: time.Duration(x.TookMS) * time.Millisecond, Detail: p.Named()}); err != nil { return err } } if ph.End != nil && ph.From != nil { if err := inv.RecordDuration(ctx, inventory.Duration{Kind: inventory.DurationWalkPhase, Subject: class + "/total", Ref: p.ID + "/total", Started: *ph.From, Took: time.Duration(ph.TotalMS) * time.Millisecond, Detail: ph.Said}); err != nil { return err } } } return nil } // ownPathWords is the controller's own path as words, for an answer. func ownPathWords() map[string]string { return onTheControllersPath } // theGraph is what the planner reads. func theGraph(ctx context.Context, inv *inventory.Inventory) ([]inventory.Entry, map[string][]inventory.ReadRepository, []inventory.Edge, error) { entries, err := inv.Catalogued(ctx) if err != nil { return nil, nil, nil, err } read, err := readForPlanning(ctx, inv) if err != nil { return nil, nil, nil, err } edges, err := inv.Dependencies(ctx) if err != nil { return nil, nil, nil, err } return entries, read, edges, nil } // groupPrimary is the head a group's composed check is run from: the one moving the controller, which then // judges the group by itself; else the one moving the node-engine, whose validator judges; else the first. func groupPrimary(members []orderMember, reach map[string]orderReach) int { for _, want := range []string{"mesh-controller", "mesh-host"} { for i, m := range members { if slices.Contains(reach[m.ID].Moved, want) { return i } } } return 0 } // askGroupCheck asks the build seat for a group's composed check: every head laid over the mesh in turn, // judged as one future state (novox/hq ADR 0239). The verdict comes back as `checked` with the group's id, // for mesh-delivery; the forge's holder sets no status from it. func askGroupCheck(ctx context.Context, open *stores, group string, members []orderMember) (string, error) { entries, read, edges, err := theGraph(ctx, open.inventory) if err != nil { return "", err } reach := reachOfMembers(members, entries, read, edges) primary := groupPrimary(members, reach) p := members[primary].pull() scope := pullScope(p, entries, read, edges) // The gate runs over what any member moves; a judge from the primary alone. for _, m := range members { r := reach[m.ID] scope.Modules = appendNew(scope.Modules, r.Moved...) scope.Dependents = appendNew(scope.Dependents, r.Dependents...) } request, err := checkRequestFor(ctx, open, p, scope, entries) if err != nil { return "", err } request.Check.Group = group world, err := theRestOfTheMesh(ctx, open.inventory, shelfOf(entries), "") if err != nil { return "", err } for i, m := range members { if i == primary { continue } mp := m.pull() // As the mesh clones a module built from it; a repository no module is built from, from the forge. source := inventory.Source{Seat: gitSeat, Repository: mp.Owner + "/" + mp.Repo} for _, e := range entries { if !e.Provided && sameRepository(e.Source.Repository, link.SourceMoved{Owner: mp.Owner, Repo: mp.Repo}) { source = e.Source break } } url := source.Repository if source.Seat != "" { url, err = clonedFromSeat(world, source.Seat, source.Repository) } if err != nil { return "", fmt.Errorf("%s cannot be cloned for the group's check: %w", m.ID, err) } request.Check.Members = append(request.Check.Members, link.GroupMember{Owner: mp.Owner, Repo: mp.Repo, Number: mp.Number, Repository: url, Ref: mp.Commit, Paths: mp.Paths}) } seat := buildSeatHeld(ctx) ask, err := askOverOn(seat) if err != nil { return "", err } defer ask.Close() if err := ask.Ask(ctx, request); err != nil { return "", err } fmt.Fprintf(os.Stderr, "group %s: asked %s to check %d head(s) composed together, as %s\n", group, seat, len(members), request.ID) return request.ID, nil } // appendNew appends what is not there yet. func appendNew(to []string, items ...string) []string { for _, i := range items { if !slices.Contains(to, i) { to = append(to, i) } } return to } // shelfOf is the catalogue's manifests by name. func shelfOf(entries []inventory.Entry) map[string]catalogue.Manifest { shelf := map[string]catalogue.Manifest{} for _, e := range entries { shelf[e.Manifest.Module] = e.Manifest } return shelf } // sayPlanMoved says a walk kept in a new state as the controller's `plan-moved`, the plan whole: what the // delivery it walks reads its steps from. Never in the way of the walk: said beside it, and a save that could // not be said is logged — the delivery's owner, which also asks for the walks it follows, finds it by // comparison. Records arriving out of order are told apart by the plan's revision. func sayPlanMoved(ctx context.Context, bus link.Bus, p inventory.Plan) { go publishPlanMoved(ctx, bus, p) } func publishPlanMoved(ctx context.Context, bus link.Bus, p inventory.Plan) { // Its phases so far (novox/hq ADR 0282 decision 6): the rest's reports come after the walk ends, and are // read by whoever asks for the walk (`delivery walks`). p.Phases = p.WalkPhases(time.Now(), nil) body, err := json.Marshal(p) if err != nil { return } stating, stop := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second) defer stop() if err := bus.PublishSeatEvent(stating, link.MeshControllerSeat, link.KeyPlanMoved, body); err != nil { fmt.Fprintf(os.Stderr, "%s: kept, and could not be said as plan-moved: %v\n", p.ID, err) } } // sayingOnTheBus runs a command that keeps walks or says verdicts with the bus to say them on: the serving // controller's, or a connection of the command's own — a verb runs as a command of its own, and what it kept // would otherwise be said by nobody until the delivery's owner read the walks back. Without a bus the command // still runs; what it did is found by comparison. func sayingOnTheBus(ctx context.Context, f func() error) error { if checkEvents != nil { return f() } ran := false err := onTheBus(func(conn *nats.Conn) error { js, err := conn.JetStream() if err != nil { return err } bus := link.OverNATS{Conn: conn, JS: js} checkEvents = bus inventory.PlanSaved = func(p inventory.Plan) { publishPlanMoved(ctx, bus, p) } defer func() { checkEvents, inventory.PlanSaved = nil, nil }() ran = true return f() }) if !ran { fmt.Fprintf(os.Stderr, "the bus cannot be reached (%v): what this does is not said, and is found by comparison\n", err) return f() } return err }