jschoubben
ed037784e2
mesh/delivery-group group fix/529-a-waiting-walk-is-answered-or-named checking: 0 of 2 member(s) ready
mesh/merge-gate pass: builds mesh-controller → novox; no bus step; every machine composes with the change as it did without (4 of 4 compose)
mesh/repo-check pass: its merge-check.sh passed
mesh/delivery superseded: a newer head of the same pull request
A walk whose delivery was held for a person said only that it waits for mesh-delivery's word and that plans go starts it by hand, pointing the reader at a hand start instead of the person's decision. Ask mesh-delivery what holds each waiting walk and say the held delivery, its reason and mesh-delivery.release, or the delivery not at its turn.
1255 lines
48 KiB
Go
1255 lines
48 KiB
Go
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, other []string
|
|
for _, l := range lines {
|
|
if l.State == "held" {
|
|
held = append(held, l.Says)
|
|
} else {
|
|
other = append(other, l.Says)
|
|
}
|
|
}
|
|
if 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, "; "))
|
|
}
|
|
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(other, "; "), 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 `<repository>@<commit>` (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: <repository>@<commit>", 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: <repository>`).
|
|
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 <commit> [--base main] --paths a,b [--module-dirs …] [--removed …]
|
|
// delivery order --members <json>
|
|
// delivery check --group <id> --members <json>
|
|
// delivery go <plan> [--by <who>] [--why <text>]
|
|
// delivery stop <plan> --why <text> [--by <who>]
|
|
// delivery walks [-n 50] [--plan <id>]
|
|
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, <repository>@<commit>, 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 <commit> --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 <json>: 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 <id> --members <json> (two heads or more), or --members <one head>")
|
|
}
|
|
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 <plan> [--by <who>] [--why <text>]")
|
|
}
|
|
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 <plan> --why <text> [--by <who>] [--without <repository>@<commit>,…]")
|
|
}
|
|
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
|
|
}
|