jschoubben
e727589e48
mesh/merge-gate head e727589e483b merged into trunk 961f2e82ea1c — pass: builds build-agent, mesh-controller, route-proxy → ace, g14, novox, shanks; no bus…
mesh/repo-check head e727589e483b merged into trunk 961f2e82ea1c — pass: its merge-check.sh passed
mesh/delivery delivered
The walk records its send to the rest only after every machine was sent, which takes 20-28 s, so the machine sent first fell outside the 2 s slack, its report was missed and the phase ended at the next walk's send to it. Keep each machine's own send time with what it was sent, match its apply duration by that send and digest, and show each machine's report in the walk's phases (ADR 0282 decision 6).
1343 lines
52 KiB
Go
1343 lines
52 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"flag"
|
|
"fmt"
|
|
"os"
|
|
"slices"
|
|
"sort"
|
|
"strings"
|
|
"sync"
|
|
"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, folded, foldMoves, err := batchToWalkAgain(ctx, inv, p.ID, 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) — the
|
|
// batches folded into it too.
|
|
if err := inv.SavePlansMoving(ctx, append([]*inventory.Plan{&target, &p}, folded...),
|
|
append(moves, foldMoves...)); 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 (novox/hq ADR 0299): a batch of their kind not yet cut
|
|
// that they are related to (ADR 0304 decision 1, ADR 0306 decision 3), or a new one, not yet written. Related to
|
|
// two or more, the newer are folded into the oldest: the plans and merge moves that fold them, to be written in
|
|
// the stop's one act. The kind is read from what the merges moved; the stopped walk itself is no walk a cut folds.
|
|
func batchToWalkAgain(ctx context.Context, inv *inventory.Inventory, stopping string, again []inventory.BatchedMerge,
|
|
now time.Time) (inventory.Plan, []*inventory.Plan, []inventory.MergeMove, error) {
|
|
own := false
|
|
for _, n := range movesOf(again) {
|
|
if _, on := onTheControllersPath[n]; on {
|
|
own = true
|
|
}
|
|
}
|
|
entries, read, edges, err := theGraph(ctx, inv)
|
|
if err != nil {
|
|
return inventory.Plan{}, nil, nil, err
|
|
}
|
|
joins, _, _, err := relatedBatches(ctx, inv, again, own, stopping, entries, read, edges)
|
|
if err != nil {
|
|
return inventory.Plan{}, nil, nil, err
|
|
}
|
|
if len(joins) > 0 {
|
|
into := joins[0]
|
|
if len(joins) == 1 {
|
|
return into, nil, nil, nil
|
|
}
|
|
plans, moves, err := foldInto(ctx, inv, &into, joins[1:])
|
|
if err != nil {
|
|
return inventory.Plan{}, nil, nil, err
|
|
}
|
|
return into, plans[1:], moves, 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, nil, 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"`
|
|
// BaseSHA is the base branch's head the forge last announced the head with (novox/hq ADR 0323): the trunk commit
|
|
// a verdict no build judges names when the head is checked again through mesh-delivery.
|
|
BaseSHA string `json:"base_sha,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, BaseSHA: m.BaseSHA,
|
|
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, inv, 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, inv, 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, inv, 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 own report of the send a walk made to it from the controller's `apply` durations
|
|
// (novox/hq issue 582); 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, send inventory.SentDeclaration, walkSent time.Time) (inventory.AppliedReport, bool) {
|
|
r, ok, err := inv.FirstAppliedAfter(ctx, node, send, walkSent)
|
|
if err != nil {
|
|
fmt.Fprintf(os.Stderr, "the report of %s after %s could not be read: %v\n", node, walkSent.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, Base: mp.Base})
|
|
}
|
|
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.
|
|
//
|
|
// On the save that starts the walk and the one that ends it, the walk also says what it touches, its bound and,
|
|
// at the end, its outcome (novox/hq ADR 0308 decision 1, walk_said.go), read from facts; nil facts say only what
|
|
// the walk recorded.
|
|
func sayPlanMoved(ctx context.Context, bus link.Bus, facts walkFacts, p inventory.Plan) {
|
|
go publishPlanMoved(ctx, bus, facts, p)
|
|
}
|
|
|
|
func publishPlanMoved(ctx context.Context, bus link.Bus, facts walkFacts, 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(planMovedBody{Plan: p, walkSaid: sayWalk(ctx, facts, 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, facts walkFacts, 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}
|
|
was := inventory.PlanSaved
|
|
checkEvents = bus
|
|
inventory.PlanSaved = func(_ *inventory.Inventory, p inventory.Plan) { publishPlanMoved(ctx, bus, facts, p) }
|
|
defer func() { checkEvents, inventory.PlanSaved = nil, was }()
|
|
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
|
|
}
|
|
|
|
// sayingWhatAVerbKept is PlanSaved for a process that is not the serving controller: each walk the verb keeps is
|
|
// said as `plan-moved` before the save returns, with the facts of the inventory that kept it, on the bus the verb
|
|
// reaches. It reaches the bus on the first save and never for a verb that keeps no walk. When the bus cannot be
|
|
// reached, a save that started or ended a walk says so on the verb's error output: its mark is kept and no later
|
|
// save says it, so the router holds what the walk touched until its bound, or not at all for a start. done lets
|
|
// the bus go and takes the hook back.
|
|
func sayingWhatAVerbKept(ctx context.Context) (said func(*inventory.Inventory, inventory.Plan), done func()) {
|
|
var (
|
|
mu sync.Mutex
|
|
reached bool
|
|
bus link.Bus
|
|
closing func()
|
|
failed error
|
|
)
|
|
said = func(inv *inventory.Inventory, p inventory.Plan) {
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
if !reached {
|
|
reached = true
|
|
if handActConn != nil {
|
|
bus, failed = overNATS(handActConn)
|
|
} else if js, err := aBus(); err != nil {
|
|
failed = err
|
|
} else if bus, failed = overNATS(js.Conn()); failed == nil {
|
|
closing = js.Close
|
|
} else {
|
|
js.Close()
|
|
}
|
|
}
|
|
if bus == nil {
|
|
switch p.WalkSaid {
|
|
case inventory.WalkStarted:
|
|
fmt.Fprintf(os.Stderr, "%s: its start is kept and could NOT be said as plan-moved (%v): the router "+
|
|
"does not hold what it touches\n", p.ID, failed)
|
|
case inventory.WalkEnded:
|
|
fmt.Fprintf(os.Stderr, "%s: its end is kept and could NOT be said as plan-moved (%v): the router holds "+
|
|
"what it touched until its bound\n", p.ID, failed)
|
|
}
|
|
return
|
|
}
|
|
publishPlanMoved(ctx, bus, inv, p)
|
|
}
|
|
done = func() {
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
inventory.PlanSaved = nil
|
|
if closing != nil {
|
|
closing()
|
|
}
|
|
}
|
|
return said, done
|
|
}
|
|
|
|
// overNATS is the bus a connection carries, with its JetStream.
|
|
func overNATS(conn *nats.Conn) (link.Bus, error) {
|
|
js, err := conn.JetStream()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return link.OverNATS{Conn: conn, JS: js}, nil
|
|
}
|