jschoubben
9f072c110c
mesh/merge-gate head 9f072c110c14 merged into trunk 60cc9ec363df — pass: builds build-agent, mesh-controller → ace, g14, novox, shanks; no bus step; every …
mesh/repo-check head 9f072c110c14 merged into trunk 60cc9ec363df — pass: its merge-check.sh passed
mesh/delivery delivered
A rebuild that landed while a walk waited for mesh-delivery's word was taken as the walk's build of that module; once let go, the walk asked a tier only when every module of it was unasked, so it never asked the rest and stood still. A walk now takes only the outcomes of builds it asked, and none while it waits; a rebuild does not join a waiting walk and says it runs beside it; and a tier asked in part is asked the rest, including a module held built that the walk never asked for, which is somebody else's build and not sent as the walk's.
853 lines
30 KiB
Go
853 lines
30 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"flag"
|
|
"fmt"
|
|
"io"
|
|
"os"
|
|
"regexp"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
|
|
"github.com/novox/mesh-controller/internal/broker"
|
|
"github.com/novox/mesh-controller/internal/inventory"
|
|
"github.com/novox/mesh-controller/internal/link"
|
|
)
|
|
|
|
// The build queue, controlled by hand (novox/hq ADR 0219).
|
|
//
|
|
// Seen whole — what waits, what runs where and for how long, what was handed out as often as it
|
|
// may be and never settled — and changed through the controller: an ask cancelled, the queue
|
|
// cleared, a module rebuilt, a build replayed. What one machine is running is that machine's
|
|
// holder's to end or pause, and the controller asks it (`kill`, `pause`, `resume`).
|
|
//
|
|
// **Everything that drops an ask leaves a failed outcome for it**, taken in exactly as a build that
|
|
// failed is (takeIn, then the plan): a plan waiting on an ask a person removed fails, saying so,
|
|
// rather than waiting for ever on an answer nobody will give.
|
|
|
|
// holderAsks is how long a holder's verb is waited for; killAnswer how long its kill is — longer
|
|
// than the holder's worst case (its two passes removing containers and its wait between, 50s).
|
|
const (
|
|
holderAsks = 15 * time.Second
|
|
killAnswer = 75 * time.Second
|
|
)
|
|
|
|
// dialTheBus is the controller's connection, for a command that reads or changes the queue: the serving
|
|
// controller's own, lent, when this process is it (novox/hq issue 327).
|
|
func dialTheBus() (*broker.JetStream, error) {
|
|
return aBus()
|
|
}
|
|
|
|
// queueCommand prints every ask in the build seat's work queue.
|
|
func queueCommand(ctx context.Context, args []string) error {
|
|
return listQueue(ctx, args, os.Stdout)
|
|
}
|
|
|
|
// listQueue is `queue`: the build queue, as a person reads it or as JSON.
|
|
func listQueue(ctx context.Context, args []string, w io.Writer) error {
|
|
set := flag.NewFlagSet("queue", flag.ContinueOnError)
|
|
usageTo(set, w)
|
|
asJSON := set.Bool("json", false, "the queue as JSON")
|
|
if _, err := parseAround(set, args); err != nil {
|
|
return err
|
|
}
|
|
js, err := dialTheBus()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer js.Close()
|
|
seat := buildSeatHeld(ctx)
|
|
q, err := link.ReadQueue(ctx, js, seat)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
now := time.Now()
|
|
if *asJSON {
|
|
body, err := json.MarshalIndent(queueAnswerOf(q, now), "", " ")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
fmt.Fprintln(w, string(body))
|
|
return nil
|
|
}
|
|
fmt.Fprint(w, queueText(q, now))
|
|
return nil
|
|
}
|
|
|
|
// keepingTheQueueInOrder runs the build queue's passes while this controller acts (novox/hq issue 558):
|
|
// the one process that puts the queue in order, so no two passes pause and resume it over each other.
|
|
// The build seat is read from the catalogue at most every half minute.
|
|
func keepingTheQueueInOrder(ctx context.Context, open *stores, js *broker.JetStream) {
|
|
if js == nil {
|
|
return
|
|
}
|
|
seat, readAt := link.TheBuildMachine, time.Time{}
|
|
seatNow := func() string {
|
|
if time.Since(readAt) < watchEvery {
|
|
return seat
|
|
}
|
|
if entries, err := open.inventory.Catalogued(ctx); err == nil {
|
|
seat, readAt = buildSeatAmong(entries), time.Now()
|
|
} else {
|
|
fmt.Printf("the build seat is read as %s: the catalogue cannot be read: %v\n", seat, err)
|
|
}
|
|
return seat
|
|
}
|
|
link.KeepingInOrder(ctx, js, seatNow, link.Holding, func(format string, args ...any) {
|
|
fmt.Printf(format+"\n", args...)
|
|
})
|
|
}
|
|
|
|
// queueAnswer is the queue as data (novox/hq issue 455): what a caller reads to say how far back an ask
|
|
// waits — "waiting in the build queue behind 2" — which the table alone could not tell it. Text is the
|
|
// table, written from the same read: the verb answers it as its output and the rest as its answer, so the
|
|
// two never disagree.
|
|
type queueAnswer struct {
|
|
Seat string `json:"seat"`
|
|
Waiting int `json:"waiting"`
|
|
InFlight int `json:"in-flight"`
|
|
Dead int `json:"dead"`
|
|
MaxDeliver int `json:"max-deliver"`
|
|
Asks []queuedAnswer `json:"asks"`
|
|
Text string `json:"text"`
|
|
}
|
|
|
|
// queuedAnswer is one ask as data. Never what the request carries beyond its source — not `held`, every
|
|
// artifact the mesh has built, nor a check's checkouts beside it.
|
|
type queuedAnswer struct {
|
|
ID string `json:"id"`
|
|
State string `json:"state"`
|
|
// Kind is "delivery" for a build a walk waits for, "check" for a pull request's check and any ask
|
|
// whose kind cannot be told (link.KindOf, novox/hq issue 558).
|
|
Kind string `json:"kind"`
|
|
Repository string `json:"repository"`
|
|
Path string `json:"path,omitempty"`
|
|
Ref string `json:"ref,omitempty"`
|
|
// Commit is the ref when it names a commit; empty when the ask follows a branch or the default.
|
|
Commit string `json:"commit,omitempty"`
|
|
// Machine is the one building an ask in flight; WasOn the one that started it and handed it back.
|
|
Machine string `json:"machine,omitempty"`
|
|
WasOn string `json:"was-on,omitempty"`
|
|
AskedAt time.Time `json:"asked-at,omitzero"`
|
|
// Since is when the ask came to its state: its start on the machine for one in flight that a
|
|
// machine said it started, otherwise when it was asked.
|
|
Since time.Time `json:"since,omitzero"`
|
|
// Position is its place among the waiting, 1 being handed out next; 0 for an ask not waiting.
|
|
Position int `json:"position,omitempty"`
|
|
Seq uint64 `json:"seq"`
|
|
// Delivery is the how-manyth time an ask in flight was handed out; more than 1 is a restart
|
|
// (novox/hq issue 594).
|
|
Delivery int `json:"delivery,omitempty"`
|
|
}
|
|
|
|
// aCommitRef is a ref that names a commit by its full hash (SHA-1 or SHA-256).
|
|
var aCommitRef = regexp.MustCompile(`^(?:[0-9a-f]{40}|[0-9a-f]{64})$`)
|
|
|
|
// queueAnswerOf is the queue as data, in the table's order: in flight, the waiting from the next one to
|
|
// be handed out, then the dead. A pure function of one read (novox/hq issue 455).
|
|
func queueAnswerOf(q link.Queue, now time.Time) queueAnswer {
|
|
waiting, running, dead := q.Of(link.AskWaiting), q.Of(link.AskInFlight), q.Of(link.AskDead)
|
|
out := queueAnswer{Seat: q.Seat, Waiting: len(waiting), InFlight: len(running), Dead: len(dead),
|
|
MaxDeliver: q.MaxDeliver, Asks: []queuedAnswer{}, Text: queueText(q, now)}
|
|
one := func(a link.QueuedAsk, position int) queuedAnswer {
|
|
kind := link.KindOf(a.Request)
|
|
since := a.AskedAt
|
|
if a.State == link.AskInFlight && a.On != "" && !a.Started.IsZero() {
|
|
since = a.Started
|
|
}
|
|
commit := ""
|
|
if aCommitRef.MatchString(a.Ref) {
|
|
commit = a.Ref
|
|
}
|
|
return queuedAnswer{ID: a.ID, State: a.State, Kind: kind, Repository: link.WithoutPassword(a.Repository), Path: a.Path,
|
|
Ref: a.Ref, Commit: commit, Machine: a.On, WasOn: a.Was, AskedAt: a.AskedAt, Since: since,
|
|
Position: position, Seq: a.Seq, Delivery: a.Delivery}
|
|
}
|
|
for _, a := range running {
|
|
out.Asks = append(out.Asks, one(a, 0))
|
|
}
|
|
for i, a := range waiting {
|
|
out.Asks = append(out.Asks, one(a, i+1))
|
|
}
|
|
for _, a := range dead {
|
|
out.Asks = append(out.Asks, one(a, 0))
|
|
}
|
|
return out
|
|
}
|
|
|
|
// queueText is the queue as a person reads it: a summary line, then each kind in queue order.
|
|
// Never what an ask carries as `held` — that is every artifact the mesh has built.
|
|
func queueText(q link.Queue, now time.Time) string {
|
|
var b strings.Builder
|
|
waiting, running, dead := q.Of(link.AskWaiting), q.Of(link.AskInFlight), q.Of(link.AskDead)
|
|
fmt.Fprintf(&b, "%s: %d waiting, %d in flight, %d dead\n", q.Seat, len(waiting), len(running), len(dead))
|
|
ago := func(t time.Time) string {
|
|
if t.IsZero() {
|
|
return "asked at an unknown time"
|
|
}
|
|
return "asked " + now.Sub(t).Round(time.Second).String() + " ago"
|
|
}
|
|
what := func(a link.QueuedAsk) string {
|
|
s := link.WithoutPassword(a.Repository)
|
|
if a.Path != "" {
|
|
s += " at " + a.Path
|
|
}
|
|
if a.Ref != "" {
|
|
s += " on " + a.Ref
|
|
}
|
|
return s
|
|
}
|
|
if len(running) > 0 {
|
|
fmt.Fprintln(&b, "\nin flight:")
|
|
for _, a := range running {
|
|
where := "taken, not yet said where"
|
|
switch {
|
|
case a.On != "":
|
|
where = fmt.Sprintf("on %s for %s", a.On, now.Sub(a.Started).Round(time.Second))
|
|
// Started again after an earlier taker left it (novox/hq issue 594).
|
|
if a.Delivery > 1 {
|
|
where += fmt.Sprintf(", handed out %d times", a.Delivery)
|
|
}
|
|
case a.Was != "":
|
|
where = fmt.Sprintf("handed back by %s, to be handed out again", a.Was)
|
|
}
|
|
fmt.Fprintf(&b, " %-26s %-8s %s — %s, %s (seq %d)\n", a.ID, link.KindOf(a.Request), what(a), where,
|
|
ago(a.AskedAt), a.Seq)
|
|
}
|
|
}
|
|
if len(waiting) > 0 {
|
|
fmt.Fprintln(&b, "\nwaiting, in the order they are handed out:")
|
|
for _, a := range waiting {
|
|
fmt.Fprintf(&b, " %-26s %-8s %s — %s (seq %d)\n", a.ID, link.KindOf(a.Request), what(a), ago(a.AskedAt), a.Seq)
|
|
}
|
|
}
|
|
if len(dead) > 0 {
|
|
fmt.Fprintf(&b, "\ndead — handed out as often as the worker allows (%d) and never settled, still held:\n", q.MaxDeliver)
|
|
for _, a := range dead {
|
|
fmt.Fprintf(&b, " %-26s %-8s %s — %s (seq %d)\n", a.ID, link.KindOf(a.Request), what(a), ago(a.AskedAt), a.Seq)
|
|
}
|
|
}
|
|
switch {
|
|
case len(q.Asks) == 0:
|
|
fmt.Fprintln(&b, "nothing is asked of it")
|
|
default:
|
|
fmt.Fprintln(&b, "\n`cancel <id>` drops a waiting or dead ask, `clear` every waiting one, `kill <id>` ends one in flight")
|
|
}
|
|
return b.String()
|
|
}
|
|
|
|
// cancelCommand drops one waiting or dead ask.
|
|
func cancelCommand(ctx context.Context, args []string) error {
|
|
if len(args) != 1 {
|
|
return errors.New("cancel <build id>")
|
|
}
|
|
js, err := dialTheBus()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer js.Close()
|
|
open, err := openStores(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer open.Close()
|
|
seat := buildSeatHeld(ctx)
|
|
q, err := link.ReadQueue(ctx, js, seat)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
ask, found := q.Find(args[0])
|
|
if !found {
|
|
return fmt.Errorf("%s is not in the %s queue: it was built, cancelled, or never asked — `builds` says "+
|
|
"what came of it", args[0], seat)
|
|
}
|
|
said, err := cancelAsk(ctx, js, open, seat, ask)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
fmt.Println(said)
|
|
return nil
|
|
}
|
|
|
|
// cancelAsk drops one ask from the queue and records it failed, cancelled by hand.
|
|
//
|
|
// **In this order, and each step for a reason** (novox/hq ADR 0219):
|
|
// 1. an ask a machine says it is building is refused: deleting its message ends nothing a machine
|
|
// is running — that is `kill`, on the machine running it;
|
|
// 2. its id goes into the seat's cancelled set, so a holder that fetches it from now on ends it;
|
|
// 3. for a waiting ask, the worker is read again: one handed out since the queue was read was
|
|
// taken in the moment of cancelling, and the cancel is withdrawn and refused — the holder either
|
|
// read the mark first and ends it as cancelled, or is building it;
|
|
// 4. for an ask the worker counts handed out that no machine is building — taken and not yet said,
|
|
// handed back after a restart, or past its deliveries while nobody pulls — the holder's own look
|
|
// at the set is waited out, and a start heard since withdraws and refuses the cancel: a holder
|
|
// that took it before the mark is building it, and only `kill` ends that;
|
|
// 5. the message is deleted by its sequence;
|
|
// 6. the failed outcome is taken in as any failed build's is, and the plan that asked fails.
|
|
func cancelAsk(ctx context.Context, js *broker.JetStream, open *stores, seat string, ask link.QueuedAsk) (string, error) {
|
|
if ask.State == link.AskInFlight && ask.On != "" {
|
|
return "", fmt.Errorf("%s is in flight on %s: cancelling drops an ask nobody is building. `kill %s` "+
|
|
"ends the build where it runs", ask.ID, ask.On, ask.ID)
|
|
}
|
|
if err := link.MarkCancelled(ctx, js, seat, ask.ID); err != nil {
|
|
return "", err
|
|
}
|
|
withdraw := func() {
|
|
if err := link.UnmarkCancelled(ctx, js, seat, ask.ID); err != nil {
|
|
fmt.Fprintf(os.Stderr, "could not withdraw the cancel of %s: %v — a holder taking it ends it as cancelled\n", ask.ID, err)
|
|
}
|
|
}
|
|
worker, _ := broker.HolderConsumerFor("", "", broker.DeclaredSeat{Name: seat, Accepts: []string{"build"}})
|
|
// **Where it waits now** (novox/hq issue 558): moved into order since it was read, a waiting ask has a
|
|
// later sequence, and the worker handing out what was behind its old one is not it being taken.
|
|
if ask.State == link.AskWaiting {
|
|
if q, err := link.ReadQueue(ctx, js, seat); err == nil {
|
|
if now, ok := q.Find(ask.ID); ok && now.State == link.AskWaiting {
|
|
ask.Seq = now.Seq
|
|
}
|
|
}
|
|
}
|
|
switch ask.State {
|
|
case link.AskWaiting:
|
|
if taken, err := takenSince(js, worker, ask.Seq); err != nil {
|
|
withdraw()
|
|
return "", err
|
|
} else if taken {
|
|
withdraw()
|
|
return "", fmt.Errorf("%s was taken by a holder as it was cancelled, so the cancel is withdrawn. If the "+
|
|
"holder read it first it ends the ask as %s and its outcome says so; otherwise it is building — "+
|
|
"`queue` says which, and `kill %s` ends it", ask.ID, link.CancelledByHand, ask.ID)
|
|
}
|
|
case link.AskInFlight:
|
|
if started, err := startedSince(ctx, js, seat, ask); err != nil {
|
|
withdraw()
|
|
return "", err
|
|
} else if started != "" {
|
|
withdraw()
|
|
return "", fmt.Errorf("%s has started on %s since it was read, so the cancel is withdrawn: `kill %s` "+
|
|
"ends it there", ask.ID, started, ask.ID)
|
|
}
|
|
}
|
|
// Found by its id when it was moved into order since it was read (novox/hq issue 558).
|
|
removed, err := link.DeleteAsk(ctx, js, seat, ask)
|
|
if err != nil {
|
|
return "", fmt.Errorf("%s is marked cancelled and could not be deleted from the queue (seq %d): %w — a "+
|
|
"holder taking it ends it as cancelled", ask.ID, ask.Seq, err)
|
|
}
|
|
// A check ended because its pull request has a newer head records nothing (novox/hq issue 726): open is nil.
|
|
if open != nil {
|
|
recordCancelled(ctx, open, ask)
|
|
}
|
|
return cancelSaid(ask, seat, removed), nil
|
|
}
|
|
|
|
// cancelSaid is what a cancel says: deleted from the queue only when it was.
|
|
func cancelSaid(ask link.QueuedAsk, seat string, removed uint64) string {
|
|
where := fmt.Sprintf("deleted from the %s queue", seat)
|
|
if removed == 0 {
|
|
where = fmt.Sprintf("no longer in the %s queue to delete, and marked cancelled, so a holder taking it "+
|
|
"ends it", seat)
|
|
}
|
|
return fmt.Sprintf("cancelled %s (%s, %s): %s and recorded failed, %s — a plan "+
|
|
"that asked for it fails with that", ask.ID, ask.Repository, ask.State, where, link.CancelledByHand)
|
|
}
|
|
|
|
// holderLooks is how long a cancel waits for a holder that took the ask before the mark to say it
|
|
// started: longer than a holder's look at the cancelled set (3s) and its start that follows.
|
|
var holderLooks = 5 * time.Second
|
|
|
|
// startedSince waits out a holder's look at the cancelled set and says the machine that started the
|
|
// ask since it was read, or nothing. A start it ended as cancelled comes with its outcome and is not
|
|
// a start of a build.
|
|
func startedSince(ctx context.Context, js *broker.JetStream, seat string, ask link.QueuedAsk) (string, error) {
|
|
select {
|
|
case <-ctx.Done():
|
|
return "", ctx.Err()
|
|
case <-time.After(holderLooks):
|
|
}
|
|
since := time.Now().Add(-7 * 24 * time.Hour)
|
|
if !ask.AskedAt.IsZero() {
|
|
since = ask.AskedAt.Add(-time.Minute)
|
|
}
|
|
heard, err := link.ReadBuildEvents(ctx, js, seat, since)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
s, ok := heard.Started[ask.ID]
|
|
if !ok || heard.Outcomes[ask.ID] {
|
|
return "", nil
|
|
}
|
|
at, _ := time.Parse(time.RFC3339Nano, s.At)
|
|
if ask.Started.IsZero() || at.After(ask.Started) {
|
|
return s.On, nil
|
|
}
|
|
return "", nil
|
|
}
|
|
|
|
// takenSince is whether the worker has handed out the ask at this sequence.
|
|
func takenSince(js *broker.JetStream, worker broker.Consumer, seq uint64) (bool, error) {
|
|
info, err := js.Context().ConsumerInfo(worker.Stream, worker.Name)
|
|
if errors.Is(err, nats.ErrConsumerNotFound) {
|
|
return false, nil
|
|
}
|
|
if err != nil {
|
|
return false, fmt.Errorf("cannot read the worker of the queue again: %w", err)
|
|
}
|
|
return info.Delivered.Stream >= seq, nil
|
|
}
|
|
|
|
// recordCancelled takes the failed outcome in the way the daemon takes in any build's: recorded,
|
|
// then the plan that asked for it — by the id it asked with, or by repository and path.
|
|
func recordCancelled(ctx context.Context, open *stores, ask link.QueuedAsk) {
|
|
r := ask.Request
|
|
result := link.BuildResult{ID: ask.ID, Repository: ask.Repository, Path: ask.Path, Ref: ask.Ref,
|
|
Source: r.Source, DryRun: r.DryRun, Failed: link.CancelledByHand}
|
|
_ = builds{open.inventory, open}.Built(ctx, result)
|
|
}
|
|
|
|
// clearCommand cancels every waiting ask, and with --dead every dead one too.
|
|
func clearCommand(ctx context.Context, args []string) error {
|
|
set := flag.NewFlagSet("clear", flag.ContinueOnError)
|
|
dead := set.Bool("dead", false, "the dead asks too")
|
|
if _, err := parseAround(set, args); err != nil {
|
|
return err
|
|
}
|
|
js, err := dialTheBus()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer js.Close()
|
|
open, err := openStores(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer open.Close()
|
|
seat := buildSeatHeld(ctx)
|
|
said, err := clearQueue(ctx, js, open, seat, *dead)
|
|
fmt.Print(said)
|
|
return err
|
|
}
|
|
|
|
// clearQueue cancels what clear names, each as `cancel` would, and never anything in flight.
|
|
func clearQueue(ctx context.Context, js *broker.JetStream, open *stores, seat string, dead bool) (string, error) {
|
|
q, err := link.ReadQueue(ctx, js, seat)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
var b strings.Builder
|
|
cancelled, refused := 0, 0
|
|
for _, a := range q.Asks {
|
|
if a.State == link.AskInFlight || (a.State == link.AskDead && !dead) {
|
|
continue
|
|
}
|
|
said, err := cancelAsk(ctx, js, open, seat, a)
|
|
if err != nil {
|
|
refused++
|
|
fmt.Fprintf(&b, " %s: %v\n", a.ID, err)
|
|
continue
|
|
}
|
|
cancelled++
|
|
fmt.Fprintf(&b, " %s\n", said)
|
|
}
|
|
what := "waiting"
|
|
if dead {
|
|
what = "waiting and dead"
|
|
}
|
|
fmt.Fprintf(&b, "%d %s ask(s) cancelled", cancelled, what)
|
|
if refused > 0 {
|
|
fmt.Fprintf(&b, ", %d could not be", refused)
|
|
}
|
|
fmt.Fprintln(&b)
|
|
if n := len(q.Of(link.AskInFlight)); n > 0 {
|
|
fmt.Fprintf(&b, "%d in flight, left running: `kill <id>` ends one where it runs\n", n)
|
|
}
|
|
if n := len(q.Of(link.AskDead)); n > 0 && !dead {
|
|
fmt.Fprintf(&b, "%d dead, left: `clear --dead` cancels them too\n", n)
|
|
}
|
|
if refused > 0 {
|
|
return b.String(), fmt.Errorf("%d ask(s) could not be cancelled", refused)
|
|
}
|
|
return b.String(), nil
|
|
}
|
|
|
|
// rebuildCommand asks the module's current source again — or, given a build's id, that build's
|
|
// repository, path and ref — under a new id.
|
|
func rebuildCommand(ctx context.Context, args []string) error {
|
|
if len(args) != 1 {
|
|
return errors.New("rebuild <module | build id>")
|
|
}
|
|
open, err := openStores(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer open.Close()
|
|
inv := open.inventory
|
|
entries, err := inv.Catalogued(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var source buildSource
|
|
var path, ref, module string
|
|
if b, found, err := inv.BuildByID(ctx, args[0]); err != nil {
|
|
return err
|
|
} else if found {
|
|
source, module = sourceOfBuild(b, entries)
|
|
path, ref = b.Path, b.Ref
|
|
// **A build at a commit is rebuilt at what its module follows now** (novox/hq ADR 0219, issue
|
|
// 219): asked now, a rebuild of an old commit would be the newest ask of the module and roll
|
|
// that commit out over everything since. Building that commit again is `replay`, which says
|
|
// what it would do and refuses to register it while anything newer is asked or registered.
|
|
if e, known := entryNamed(entries, module); known && ref != "" && followedBranch(ref) == "" {
|
|
ref = followedBranch(e.Source.Ref)
|
|
follows := ref
|
|
if follows == "" {
|
|
follows = "the repository's default branch"
|
|
}
|
|
fmt.Printf("%s was built at commit %s; a rebuild asks what %s follows now, %s — `replay %s` "+
|
|
"builds that commit\n", b.ID, short(b.Ref), module, follows, b.ID)
|
|
}
|
|
} else if e, known := entryNamed(entries, args[0]); known {
|
|
// As a plan asks it: the branch it follows, never a commit a build once named (issue 215).
|
|
source = buildSource{Repository: e.Source.Repository, Seat: e.Source.Seat}
|
|
path, ref, module = e.Source.Path, followedBranch(e.Source.Ref), e.Manifest.Module
|
|
} else {
|
|
return fmt.Errorf("%s is neither a module the catalogue holds nor a build the mesh recorded", args[0])
|
|
}
|
|
|
|
// **A module a plan holds unbuilt, or failed, joins that plan** rather than running beside it: the
|
|
// plan would ask it again, or stay failed on an outcome a rebuild has replaced. A walk waiting for its word
|
|
// is not joined (novox/hq issue 754): the rebuild runs beside it, and says so.
|
|
var beside string
|
|
if module != "" {
|
|
joined, said, err := joinAPlan(ctx, open, module)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if joined {
|
|
fmt.Println(said)
|
|
return nil
|
|
}
|
|
beside = said
|
|
}
|
|
id, err := askABuild(ctx, source, path, ref)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
recordAsked(ctx, inventory.BuildRequest{ID: id, Repository: source.Repository, Seat: source.Seat, Path: path,
|
|
Ref: ref, For: "rebuild"})
|
|
if beside != "" {
|
|
fmt.Printf("rebuild asked as %s, beside the walk: %s\n", id, beside)
|
|
return nil
|
|
}
|
|
fmt.Printf("rebuild asked as %s\n", id)
|
|
return nil
|
|
}
|
|
|
|
// entryNamed is the catalogue's entry for a module.
|
|
func entryNamed(entries []inventory.Entry, name string) (inventory.Entry, bool) {
|
|
for _, e := range entries {
|
|
if e.Manifest.Module == name {
|
|
return e, true
|
|
}
|
|
}
|
|
return inventory.Entry{}, false
|
|
}
|
|
|
|
// sourceOfBuild is where a recorded build's repository is asked from: the catalogued module's own
|
|
// source when the build is of one — on its seat, so the outcome registers it as the mesh records it
|
|
// (ADR 0111) — else the repository as it was cloned.
|
|
func sourceOfBuild(b inventory.Build, entries []inventory.Entry) (buildSource, string) {
|
|
for _, e := range entries {
|
|
if (b.Module != "" && e.Manifest.Module == b.Module) ||
|
|
(b.Module == "" && repositoryMatches(e.Source.Repository, b.Repository) && e.Source.Path == b.Path) {
|
|
return buildSource{Repository: e.Source.Repository, Seat: e.Source.Seat}, e.Manifest.Module
|
|
}
|
|
}
|
|
return buildSource{Repository: b.Repository}, b.Module
|
|
}
|
|
|
|
// replayCommand asks a recorded build's repository and path again at the commit it built.
|
|
func replayCommand(ctx context.Context, args []string) error {
|
|
set := flag.NewFlagSet("replay", flag.ContinueOnError)
|
|
register := set.Bool("register", false, "register what it builds, as any build is")
|
|
older := set.Bool("older", false, "with --register: even though a newer build of the module is registered")
|
|
positionals, err := parseAround(set, args)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if len(positionals) != 1 {
|
|
return errors.New("replay <build id> [--register [--older]]")
|
|
}
|
|
open, err := openStores(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer open.Close()
|
|
inv := open.inventory
|
|
b, found, err := inv.BuildByID(ctx, positionals[0])
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !found {
|
|
return fmt.Errorf("no build %s is recorded", positionals[0])
|
|
}
|
|
entries, err := inv.Catalogued(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
source, module := sourceOfBuild(b, entries)
|
|
var history []inventory.Build
|
|
if module != "" {
|
|
if history, err = inv.Builds(ctx, module, 100); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
// What is asked of the module and not yet answered, when the replay would be registered.
|
|
var outstanding []string
|
|
if *register {
|
|
js, err := dialTheBus()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
q, err := link.ReadQueue(ctx, js, buildSeatHeld(ctx))
|
|
js.Close()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
plans, err := inv.OpenPlans(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
outstanding = outstandingFor(module, b.Repository, b.Path, q, plans)
|
|
}
|
|
if err := replayRefusal(b, module, history, *register, *older, outstanding); err != nil {
|
|
return err
|
|
}
|
|
asker := ""
|
|
if *register {
|
|
asker = "replay"
|
|
}
|
|
id, err := buildOneAsked(ctx, source, b.Path, b.Commit, 0, !*register, asker)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
fmt.Println(replaySaid(b, module, id, *register))
|
|
return nil
|
|
}
|
|
|
|
// replayRefusal is why a replay is not asked, or nothing (novox/hq ADR 0219, issue 207).
|
|
//
|
|
// A replay re-asks a recorded build at the commit it built, under a new id — and an id is when it
|
|
// was asked, which is what orders builds of one module (issue 219). So a registered replay of an
|
|
// older commit is newer than everything since, and would roll that older commit out as the module's
|
|
// current version: what issue 207 recorded happening by accident. It is therefore a dry run unless
|
|
// --register says otherwise, and --register is refused when a newer build of a different commit is
|
|
// registered, unless --older says that is the point.
|
|
//
|
|
// **And refused while anything newer is outstanding**, --older or not: an ask of the module in the
|
|
// queue, or an open plan holding it unbuilt. Asked now, the replay would be newer than those asks,
|
|
// and their builds — made from newer source — would be recorded and never registered (issue 219
|
|
// orders by ask), the plan waiting on them sending the older commit instead.
|
|
func replayRefusal(b inventory.Build, module string, history []inventory.Build, register, older bool,
|
|
outstanding []string) error {
|
|
if b.Commit == "" {
|
|
return fmt.Errorf("%s recorded no commit — it failed before it knew what it was building, so there is "+
|
|
"nothing to replay; `rebuild %s` asks its repository, path and ref again", b.ID, b.ID)
|
|
}
|
|
if older && !register {
|
|
return errors.New("--older only says what --register may do; a dry run registers nothing")
|
|
}
|
|
if register && len(outstanding) > 0 {
|
|
return fmt.Errorf("%s is asked and not yet answered — %s — and a registered replay asked now would "+
|
|
"replace what those build (novox/hq issue 219). Wait for them, or `replay %s` without --register to look",
|
|
orNone(module), strings.Join(outstanding, "; "), b.ID)
|
|
}
|
|
if !register || older {
|
|
return nil
|
|
}
|
|
for _, h := range history {
|
|
if h.ID == b.ID || !h.Worked() || h.Commit == b.Commit {
|
|
continue
|
|
}
|
|
if h.AskedOrAt().After(b.AskedOrAt()) {
|
|
return fmt.Errorf("a newer build of %s is registered — %s, from %s — and registering a replay of %s "+
|
|
"would roll that older commit out as %s's current version (novox/hq issue 207). `replay %s` "+
|
|
"without --register looks at it; --register --older registers it anyway",
|
|
module, h.ID, short(h.Commit), short(b.Commit), module, b.ID)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// replaySaid is what a replay prints once asked: what it builds, and what becomes of the outcome.
|
|
func replaySaid(b inventory.Build, module, id string, register bool) string {
|
|
what := b.Repository
|
|
if module != "" {
|
|
what = module
|
|
}
|
|
said := fmt.Sprintf("replaying %s (%s) at %s as %s", b.ID, what, short(b.Commit), id)
|
|
if !register {
|
|
return said + "\n a dry run: the outcome is looked at and not taken in — nothing is recorded or " +
|
|
"registered, and nothing is sent. `builds --log " + id + "` follows it; `--register` registers it"
|
|
}
|
|
return said + "\n registered when it is built, as the module's current version — newer than every build " +
|
|
"asked before now — and rolled out as its policy says. `builds --log " + id + "` follows it"
|
|
}
|
|
|
|
// killCommand finds the machine running a build and asks its holder to end it.
|
|
func killCommand(ctx context.Context, args []string) error {
|
|
if len(args) != 1 {
|
|
return errors.New("kill <build id>")
|
|
}
|
|
id := args[0]
|
|
js, err := dialTheBus()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer js.Close()
|
|
seat := buildSeatHeld(ctx)
|
|
since := time.Now().Add(-7 * 24 * time.Hour)
|
|
if at, ok := link.BuildAskedAt(id); ok {
|
|
since = at.Add(-time.Minute)
|
|
}
|
|
heard, err := link.ReadBuildEvents(ctx, js, seat, since)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
started, ok := heard.Started[id]
|
|
if !ok {
|
|
return fmt.Errorf("no machine has said it started %s: if it waits in the queue, `cancel %s` drops it", id, id)
|
|
}
|
|
if heard.Outcomes[id] {
|
|
return fmt.Errorf("%s has already ended on %s — `builds` says how", id, started.On)
|
|
}
|
|
answer, err := link.AskSeatTool(ctx, js.Conn(), seat, "kill", started.On, map[string]string{"id": id}, killAnswer)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return printHolderAnswer(started.On, answer)
|
|
}
|
|
|
|
// pauseCommand asks one machine's holder, or every holder's, to pause or resume.
|
|
func pauseCommand(ctx context.Context, verb string, args []string) error {
|
|
if len(args) > 1 {
|
|
return fmt.Errorf("%s [node]", verb)
|
|
}
|
|
js, err := dialTheBus()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer js.Close()
|
|
seat := buildSeatHeld(ctx)
|
|
nodes := args
|
|
if len(nodes) == 0 {
|
|
if nodes, err = buildSeatHolders(ctx, seat); err != nil {
|
|
return err
|
|
}
|
|
if len(nodes) == 0 {
|
|
return fmt.Errorf("nothing holds %s, so there is nothing to %s", seat, verb)
|
|
}
|
|
}
|
|
failed := 0
|
|
for _, node := range nodes {
|
|
answer, err := link.AskSeatTool(ctx, js.Conn(), seat, verb, node, map[string]string{}, holderAsks)
|
|
if err != nil {
|
|
fmt.Printf("%s: %v\n", node, err)
|
|
failed++
|
|
continue
|
|
}
|
|
if err := printHolderAnswer(node, answer); err != nil {
|
|
failed++
|
|
}
|
|
}
|
|
if failed > 0 {
|
|
return fmt.Errorf("%d of %d machine(s) did not %s", failed, len(nodes), verb)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// printHolderAnswer prints what a holder said, or its refusal as the command's failure.
|
|
func printHolderAnswer(node string, answer link.Answer) error {
|
|
if answer.Error != "" {
|
|
fmt.Printf("%s: %s\n", node, answer.Error)
|
|
return errors.New(answer.Error)
|
|
}
|
|
var said struct {
|
|
Said string `json:"said"`
|
|
}
|
|
if json.Unmarshal(answer.Result, &said) == nil && said.Said != "" {
|
|
fmt.Printf("%s: %s\n", node, said.Said)
|
|
return nil
|
|
}
|
|
fmt.Printf("%s: %s\n", node, string(answer.Result))
|
|
return nil
|
|
}
|
|
|
|
// buildSeatHolders is every machine an assigned module holding the build seat runs on.
|
|
func buildSeatHolders(ctx context.Context, seat string) ([]string, error) {
|
|
open, err := openStores(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer open.Close()
|
|
entries, err := open.inventory.Catalogued(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return holdersAmong(entries, seat), nil
|
|
}
|
|
|
|
// holdersAmong is the machines of every catalogued module claiming the seat, sorted, once each.
|
|
func holdersAmong(entries []inventory.Entry, seat string) []string {
|
|
seen := map[string]bool{}
|
|
var out []string
|
|
for _, e := range entries {
|
|
if !e.Manifest.ClaimsSeat(seat) {
|
|
continue
|
|
}
|
|
for _, n := range e.On {
|
|
if !seen[n] {
|
|
seen[n] = true
|
|
out = append(out, n)
|
|
}
|
|
}
|
|
}
|
|
sort.Strings(out)
|
|
return out
|
|
}
|
|
|
|
// outstandingFor is everything asked of a module and not yet answered: its asks in the build queue,
|
|
// by repository and path, and every open plan holding it not yet built.
|
|
func outstandingFor(module, repository, path string, q link.Queue, plans []inventory.Plan) []string {
|
|
var out []string
|
|
for _, a := range q.Asks {
|
|
if repositoryMatches(a.Repository, repository) && a.Path == path {
|
|
out = append(out, fmt.Sprintf("%s %s in the queue", a.ID, a.State))
|
|
}
|
|
}
|
|
if module == "" {
|
|
return out
|
|
}
|
|
for _, p := range plans {
|
|
if !p.Open() {
|
|
continue
|
|
}
|
|
for _, tier := range p.Tiers {
|
|
for _, m := range tier {
|
|
if m == module {
|
|
if st := p.Modules[m]; st == nil || st.State != "built" {
|
|
out = append(out, fmt.Sprintf("%s holds it not yet built", p.ID))
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return out
|
|
}
|