The controller asks no build of anybody (ADR 0501 §5, ADR 0525 clause 6, ADR 0503 §3). Every module of a walk is taken from the builder's done key; a key not done yet waits, named in the plan, and never fails it. - C2b/C2c: internal/builder and cmd/mesh-controller's asks go (askABuild, onlyTheBuildersOwn, buildOneAsked, takeIn from an outcome, buildAgentsOwn, the builder's say and its fallback, the daemon hearing an outcome). The server follows the retired seats' built subjects only to let go of them, because the installer's carried user list names them. - C2d: replay, queue, cancel, clear, kill, pause, resume, delivery-check, check-here and merge-gate are refused by name. - C3: the check queues (CHECKS_READY, CHECKS_OTHER) and their workers, the checks-running and plans-building buckets and their grants, the build queue and its hold (plans_building.go, queue.go, link/holds.go, builds_nats.go, checks_nats.go). - C4: merge_gate*.go, the pull request's repo-check path, the Checked event and its layers; a pull request asks the controller nothing but to tell mesh-config what its head declares. - C5: the builds bucket is written only by key_takein. - C10 (help and refusals): the terminal's help lists only what is served. Integration tests of the plan's tiers, sends, stops and retries now run on a stand-in builder with a done key for every module (walk_keys_scaffold_integration_test.go); the tests of the asks, the queue, the merge gate and the replays are deleted.
796 lines
31 KiB
Go
796 lines
31 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
|
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
|
|
"github.com/novox/mesh-controller/internal/builderkeys"
|
|
"github.com/novox/mesh-controller/internal/conditions"
|
|
"github.com/novox/mesh-controller/internal/inventory"
|
|
)
|
|
|
|
// A module not registered yet, and the assignment that waits for it (novox/hq issue 325, ADR 0261).
|
|
//
|
|
// On 2026-10-08 the catalogue's pull request adding `sensors` merged; a minute later `assign g14 sensors`
|
|
// answered "no module of that name: sensors", and a few minutes later the same call worked. The merge had
|
|
// asked for the build (issue 300) and the build had not finished. The answer read as "you forgot to register
|
|
// it", and nothing in it said a build was on its way.
|
|
//
|
|
// So an assignment of a module the catalogue does not hold looks further before it refuses, and says which
|
|
// of three cases it is in:
|
|
//
|
|
// - **a build is in flight**: a build request for the module's directory has no outcome yet, or its build
|
|
// succeeded a moment ago and is being registered. The assignment is kept as a **pending assignment**,
|
|
// which the controller makes when the build registers the module (ADR 0261).
|
|
// - **known, not built**: the controller asked for the directory before, and the build failed, was not
|
|
// registered, said nothing within its bound, or could not be asked. Said, with `build` and assign's
|
|
// own build argument, which asks for the build and keeps the assignment pending on it.
|
|
// - **unknown**: no module registered and no build request kept by that name. Refused as before, with the
|
|
// closest names.
|
|
//
|
|
// **Pending by default while a build is in flight** (ADR 0261), without an argument to ask for it: an
|
|
// assignment is what a person meant and is kept even when its machine does not resolve (acts.go), and
|
|
// `unassign` withdraws it. Asking for a second act once the build lands is the two acts by hand issue 300
|
|
// removed. A build is never asked for by default: it runs on another machine, and a directory whose last
|
|
// build failed is a fault to look at before it is a thing to retry.
|
|
//
|
|
// **Settling is the controller's tick, never a read** (settlingPending): `status`, the board and the summary
|
|
// only read the rows.
|
|
|
|
// buildRequestBound is how long a build request may go without an outcome before it is no longer read as in
|
|
// flight, and a pending assignment waiting for it expires. Generous: a build waits behind others on the
|
|
// build seat, and a pending assignment that waits a little longer costs nothing.
|
|
const buildRequestBound = 2 * time.Hour
|
|
|
|
// registerGrace is how long a build heard as built is read as still in flight while it is not registered:
|
|
// the outcome is recorded first and registered after it, in the same take-in. Past it, the build was
|
|
// recorded and not registered, and says so.
|
|
const registerGrace = 5 * time.Minute
|
|
|
|
// pendingShownFor is how long an ended pending assignment is listed in `status`.
|
|
const pendingShownFor = 24 * time.Hour
|
|
|
|
// settleEvery is how often the controller settles the pending assignments.
|
|
const settleEvery = time.Minute
|
|
|
|
// claimStaleAfter is how long a pending assignment may be held as applying before the tick reads the claim
|
|
// as left by a controller that stopped, and settles it from what the mesh holds.
|
|
const claimStaleAfter = 5 * time.Minute
|
|
|
|
// kindPendingEnded is a pending assignment that ended without being made: expired or refused.
|
|
const kindPendingEnded = "assignment-not-made"
|
|
|
|
// assignOptions are what an assignment was asked with beside its machine and modules.
|
|
type assignOptions struct {
|
|
// Build asks for the build of a module known and not built, and keeps the assignment pending on it.
|
|
Build bool
|
|
}
|
|
|
|
// The cases of a module the catalogue does not hold (novox/hq issue 325).
|
|
const (
|
|
unknownModule = iota
|
|
buildInFlight
|
|
knownNotBuilt
|
|
// keyComing is a module the indexer holds and the catalogue does not yet: registered from the builder's done key
|
|
// of it (ADR 0501 §6), which nobody asks.
|
|
keyComing
|
|
)
|
|
|
|
// keyPendingPrefix opens the build a pending assignment waits for when it waits for a module's key: no build is asked
|
|
// (ADR 0501 §5, §6; issue 1851, A10), so it names the module's key, not a build id.
|
|
const keyPendingPrefix = "key:"
|
|
|
|
// indexedModule is the indexer's trunk value of a module the catalogue does not hold, found when it holds one. A
|
|
// variable so a test needs no bus.
|
|
var indexedModule = func(ctx context.Context, module string) (sdkdelivery.Module, bool, error) {
|
|
keys, closeKeys, err := openBuilderKeys()
|
|
if err != nil {
|
|
return sdkdelivery.Module{}, false, err
|
|
}
|
|
defer closeKeys()
|
|
v, err := keys.ModuleAt(ctx, module, "")
|
|
if errors.Is(err, builderkeys.ErrNoValue) {
|
|
return sdkdelivery.Module{}, false, nil
|
|
}
|
|
return v, err == nil, err
|
|
}
|
|
|
|
// notHeld is where a module the catalogue does not hold stands: its case, the build request it was read
|
|
// from, and the sentence that says it.
|
|
type notHeld struct {
|
|
kind int
|
|
request inventory.RequestOutcome
|
|
said string
|
|
}
|
|
|
|
// whereIs looks for a module the catalogue does not hold among the build requests the controller keeps.
|
|
// notHeldReads is what whereIs reads of the records (*inventory.Inventory, or a test's).
|
|
type notHeldReads interface {
|
|
RequestsNamed(ctx context.Context, name string) ([]inventory.RequestOutcome, error)
|
|
Catalogue(ctx context.Context) (map[string]catalogue.Manifest, error)
|
|
RequestedNames(ctx context.Context) ([]string, error)
|
|
}
|
|
|
|
func whereIs(ctx context.Context, inv notHeldReads, module string, now time.Time) (notHeld, error) {
|
|
// **Its key is coming** (ADR 0501 §6): the indexer holds it, and the controller registers it from the builder's done
|
|
// key of it. Read before the build requests of before R4e, which nothing writes any more.
|
|
if v, held, err := indexedModule(ctx, module); err != nil {
|
|
return notHeld{}, fmt.Errorf("whether the indexer holds %s cannot be read: %w", module, err)
|
|
} else if held {
|
|
return notHeld{kind: keyComing, request: inventory.RequestOutcome{BuildRequest: inventory.BuildRequest{
|
|
ID: keyPendingPrefix + module, Repository: v.Repository, Seat: gitSeat, Path: keyPath(v)}},
|
|
said: fmt.Sprintf("%s is not registered yet: the indexer holds it in %s at %.8s, and it is registered from "+
|
|
"the builder's done key of it, which no one asks", module, v.Repository, v.Commit)}, nil
|
|
}
|
|
requests, err := inv.RequestsNamed(ctx, module)
|
|
if err != nil {
|
|
return notHeld{}, err
|
|
}
|
|
if r, ok := inFlight(requests, now); ok {
|
|
being := "being built"
|
|
if r.Heard {
|
|
being = "built and being registered"
|
|
}
|
|
return notHeld{kind: buildInFlight, request: r, said: fmt.Sprintf("%s is not registered yet: it is %s "+
|
|
"from %s since %s, as build %s; it can be assigned once that build registers it", module, being,
|
|
requestSource(r), clock(r.At), r.ID)}, nil
|
|
}
|
|
if len(requests) > 0 {
|
|
return notHeld{kind: knownNotBuilt, request: requests[0], said: fmt.Sprintf("%s is known and not "+
|
|
"registered: %s", module, whyNotBuilt(requests[0], now))}, nil
|
|
}
|
|
// Only what was looked at is claimed: the catalogue, and the build requests the controller keeps.
|
|
said := fmt.Sprintf("%v: %s — the catalogue holds no module of that name, and no build request the "+
|
|
"controller kept (the last 30 days) is for a directory of that name", inventory.ErrNoSuchModule, module)
|
|
if near := closestNames(ctx, inv, module); len(near) > 0 {
|
|
said += "; the closest names it holds: " + strings.Join(near, ", ")
|
|
}
|
|
return notHeld{kind: unknownModule, said: said}, nil
|
|
}
|
|
|
|
// notInCatalogue is the modules named that the catalogue does not hold.
|
|
func notInCatalogue(ctx context.Context, open *stores, modules []string) ([]string, error) {
|
|
shelf, err := open.inventory.Catalogue(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var missing []string
|
|
for _, m := range modules {
|
|
if _, ok := shelf[m]; !ok {
|
|
missing = append(missing, m)
|
|
}
|
|
}
|
|
return missing, nil
|
|
}
|
|
|
|
// openPending is the open pending assignment of a module to a machine, if there is one.
|
|
func openPending(ctx context.Context, inv *inventory.Inventory, node, module string) (*inventory.PendingAssignment, error) {
|
|
rows, err := inv.PendingFor(ctx, node, module)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for i := range rows {
|
|
if rows[i].Open() {
|
|
return &rows[i], nil
|
|
}
|
|
}
|
|
return nil, nil
|
|
}
|
|
|
|
// notRegistered answers an assignment of a module the catalogue does not hold: pending on a build in flight,
|
|
// known and not built, or unknown (novox/hq issue 325). An answer with no error is a pending assignment kept.
|
|
func notRegistered(ctx context.Context, open *stores, node, module string, opts assignOptions) (string, error) {
|
|
inv := open.inventory
|
|
if _, err := inv.NodeByName(ctx, node); err != nil {
|
|
return "", err
|
|
}
|
|
now := time.Now()
|
|
where, err := whereIs(ctx, inv, module, now)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
waiting, err := openPending(ctx, inv, node, module)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
switch where.kind {
|
|
case keyComing:
|
|
lead := where.said
|
|
if opts.Build {
|
|
lead += "\n build \"true\" asks nothing: the builder builds every fingerprint the indexer writes (ADR 0501 §6)"
|
|
}
|
|
return keepPending(ctx, open, node, module, where.request, waiting, lead)
|
|
case buildInFlight:
|
|
lead := where.said
|
|
if opts.Build {
|
|
lead += "\n no second build was asked: this one is running"
|
|
}
|
|
return keepPending(ctx, open, node, module, where.request, waiting, lead)
|
|
case knownNotBuilt:
|
|
last := where.request
|
|
if !opts.Build {
|
|
also := ""
|
|
if waiting != nil {
|
|
also = fmt.Sprintf("\n %s already has a pending assignment of %s, waiting for build %s; build "+
|
|
"\"true\" asks for a new build and makes it wait for that one", node, module, waiting.Build)
|
|
}
|
|
return "", fmt.Errorf("%w: %s%s\n it is registered from the builder's done key once a merge adds it to %s "+
|
|
"(path %s) and the indexer holds it, or at the controller's terminal; assign it then",
|
|
inventory.ErrNoSuchModule, where.said, also, last.Repository, orRoot(last.Path))
|
|
}
|
|
// **No build is asked** (ADR 0501 §5, §6): a module the indexer does not hold has no key coming.
|
|
return "", fmt.Errorf("%w: %s\n build \"true\" asks no build: the controller asks none (ADR 0501 §5); a module "+
|
|
"is registered from the builder's done key once a merge adds it to a repository the catalogue builds from, or "+
|
|
"at the controller's terminal, and the indexer holds no %s", inventory.ErrNoSuchModule, where.said, module)
|
|
}
|
|
answer := strings.TrimPrefix(where.said, inventory.ErrNoSuchModule.Error()) +
|
|
"\n a module in a repository is registered from the builder's done key of it, then assigned"
|
|
if opts.Build {
|
|
answer += "\n build \"true\" asks for nothing: the controller asks no build (ADR 0501 §5), and the indexer " +
|
|
"holds no module of that name"
|
|
}
|
|
return "", fmt.Errorf("%w%s", inventory.ErrNoSuchModule, answer)
|
|
}
|
|
|
|
// keepPending records the assignment as pending on a build request — or, where one is already open, makes it
|
|
// wait for that build — and says so.
|
|
func keepPending(ctx context.Context, open *stores, node, module string, r inventory.RequestOutcome,
|
|
waiting *inventory.PendingAssignment, lead string) (string, error) {
|
|
inv := open.inventory
|
|
if waiting != nil {
|
|
if waiting.Build == r.ID {
|
|
return lead + fmt.Sprintf("\n %s already waits for %s on this build: nothing changed", node, module), nil
|
|
}
|
|
moved, err := inv.RepointPending(ctx, waiting.ID, r.ID, r.Repository, r.Path)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
if !moved {
|
|
return lead + fmt.Sprintf("\n the pending assignment of %s to %s ended while this was asked: `status` "+
|
|
"says how", module, node), nil
|
|
}
|
|
return lead + fmt.Sprintf("\n the pending assignment of %s to %s, which waited for build %s, now waits "+
|
|
"for build %s", module, node, waiting.Build, r.ID), nil
|
|
}
|
|
_, err := inv.RecordPending(ctx, inventory.PendingAssignment{Node: node, Module: module, Build: r.ID,
|
|
Repository: r.Repository, Path: r.Path})
|
|
if errors.Is(err, inventory.ErrAlreadyPending) {
|
|
return lead + fmt.Sprintf("\n %s already waits for %s: nothing changed", node, module), nil
|
|
}
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
// The controller makes it, not this act (ADR 0261): when the build registers the module, or at its next
|
|
// tick if the module was registered between the look and the record.
|
|
return lead + fmt.Sprintf("\n the assignment is kept as pending: the controller assigns %s to %s when the "+
|
|
"build registers it, and `status` lists it until then. If the build fails it expires, says why and "+
|
|
"raises a condition; `unassign %s %s` withdraws it", module, node, node, module), nil
|
|
}
|
|
|
|
// inFlight is the newest build request for the module still to register it: no outcome yet within the
|
|
// bound, or a successful outcome heard a moment ago and being registered.
|
|
func inFlight(requests []inventory.RequestOutcome, now time.Time) (inventory.RequestOutcome, bool) {
|
|
for _, r := range requests {
|
|
if stillComing(r, now) {
|
|
return r, true
|
|
}
|
|
}
|
|
return inventory.RequestOutcome{}, false
|
|
}
|
|
|
|
func stillComing(r inventory.RequestOutcome, now time.Time) bool {
|
|
if r.Heard {
|
|
return r.Failed == "" && now.Sub(r.HeardAt) < registerGrace
|
|
}
|
|
return r.NotAsked == "" && now.Sub(r.At) < buildRequestBound
|
|
}
|
|
|
|
// whyNotBuilt says what came of a module's last build request. What was heard comes first: an outcome is the
|
|
// last word whatever its asker said.
|
|
func whyNotBuilt(r inventory.RequestOutcome, now time.Time) string {
|
|
where := fmt.Sprintf("%s in %s", orRoot(r.Path), r.Repository)
|
|
switch {
|
|
case r.Heard && r.Failed != "":
|
|
return fmt.Sprintf("%s; its last build, %s at %s, failed: %s", where, r.ID, clock(r.HeardAt),
|
|
firstLine(r.Failed))
|
|
case r.Heard && r.Module != "" && r.Module != r.Name():
|
|
return fmt.Sprintf("%s; its last build, %s, built it as %s", where, r.ID, r.Module)
|
|
case r.Heard:
|
|
return fmt.Sprintf("%s; its last build, %s at %s, was recorded and not registered — `builds` says why",
|
|
where, r.ID, clock(r.HeardAt))
|
|
case r.NotAsked != "" && r.For == "merge":
|
|
return fmt.Sprintf("%s; the merge that added it (%s) could not ask for its build: %s", where,
|
|
short(r.Commit), firstLine(r.NotAsked))
|
|
case r.NotAsked != "":
|
|
return fmt.Sprintf("%s; build %s, asked by %s, was not handed over: %s", where, r.ID,
|
|
askerOr(r.For), firstLine(r.NotAsked))
|
|
case r.OutcomeUnknown != "":
|
|
return fmt.Sprintf("%s; build %s was asked at %s, its asker stopped waiting (%s), and no outcome has "+
|
|
"been heard in %s", where, r.ID, clock(r.At), firstLine(r.OutcomeUnknown), now.Sub(r.At).Round(time.Minute))
|
|
default:
|
|
return fmt.Sprintf("%s; build %s was asked at %s, and no outcome has been heard in %s", where, r.ID,
|
|
clock(r.At), now.Sub(r.At).Round(time.Minute))
|
|
}
|
|
}
|
|
|
|
func askerOr(who string) string {
|
|
if who == "" {
|
|
return "the controller"
|
|
}
|
|
return who
|
|
}
|
|
|
|
// requestSource is a build request's source as a person reads it: repository@commit (directory).
|
|
func requestSource(r inventory.RequestOutcome) string {
|
|
at := short(r.Commit)
|
|
if at == "" {
|
|
at = r.Ref
|
|
}
|
|
if at == "" {
|
|
at = "its default branch"
|
|
}
|
|
return fmt.Sprintf("%s@%s (%s)", r.Repository, at, orRoot(r.Path))
|
|
}
|
|
|
|
// clock is a moment as a person reads it, in the controller's local time.
|
|
func clock(t time.Time) string { return t.Local().Format("2006-01-02 15:04:05 MST") }
|
|
|
|
// closestNames is up to three names near the one asked: registered modules and directories the controller
|
|
// asked to build.
|
|
func closestNames(ctx context.Context, inv notHeldReads, name string) []string {
|
|
var candidates []string
|
|
if shelf, err := inv.Catalogue(ctx); err == nil {
|
|
for m := range shelf {
|
|
candidates = append(candidates, m)
|
|
}
|
|
}
|
|
if asked, err := inv.RequestedNames(ctx); err == nil {
|
|
candidates = append(candidates, asked...)
|
|
}
|
|
type scored struct {
|
|
name string
|
|
distance int
|
|
}
|
|
seen := map[string]bool{}
|
|
var near []scored
|
|
limit := max(2, len(name)/3)
|
|
for _, c := range candidates {
|
|
if seen[c] || c == name {
|
|
continue
|
|
}
|
|
seen[c] = true
|
|
d := editDistance(name, c)
|
|
if d <= limit || strings.Contains(c, name) || strings.Contains(name, c) {
|
|
near = append(near, scored{c, d})
|
|
}
|
|
}
|
|
sort.Slice(near, func(i, j int) bool {
|
|
if near[i].distance != near[j].distance {
|
|
return near[i].distance < near[j].distance
|
|
}
|
|
return near[i].name < near[j].name
|
|
})
|
|
var out []string
|
|
for i := 0; i < len(near) && i < 3; i++ {
|
|
out = append(out, near[i].name)
|
|
}
|
|
return out
|
|
}
|
|
|
|
// editDistance is the Levenshtein distance between two names.
|
|
func editDistance(a, b string) int {
|
|
ra, rb := []rune(a), []rune(b)
|
|
prev := make([]int, len(rb)+1)
|
|
for j := range prev {
|
|
prev[j] = j
|
|
}
|
|
for i := 1; i <= len(ra); i++ {
|
|
cur := make([]int, len(rb)+1)
|
|
cur[0] = i
|
|
for j := 1; j <= len(rb); j++ {
|
|
cost := 1
|
|
if ra[i-1] == rb[j-1] {
|
|
cost = 0
|
|
}
|
|
cur[j] = min(prev[j]+1, cur[j-1]+1, prev[j-1]+cost)
|
|
}
|
|
prev = cur
|
|
}
|
|
return prev[len(rb)]
|
|
}
|
|
|
|
// applyPending makes a pending assignment now that its module is registered (ADR 0261). Under the machine's
|
|
// hold it claims the row (waiting → applying), so a withdrawal cannot land between the look and the act,
|
|
// then assigns by the same act a person's assign is, then says what came of it: applied, or refused with the
|
|
// refusal. Returns what it said, or nothing when the row no longer waited.
|
|
func applyPending(ctx context.Context, open *stores, p inventory.PendingAssignment, by string) string {
|
|
inv := open.inventory
|
|
ctx, release, err := holdNodes(ctx, open, []string{p.Node})
|
|
if err != nil {
|
|
// Left waiting: the next tick tries again.
|
|
return fmt.Sprintf("the pending assignment of %s to %s waits: %s could not be held: %v", p.Module, p.Node,
|
|
p.Node, err)
|
|
}
|
|
defer release()
|
|
claimed, err := inv.ClaimPending(ctx, p.ID)
|
|
if err != nil {
|
|
return fmt.Sprintf("the pending assignment of %s to %s could not be taken for making: %v", p.Module, p.Node, err)
|
|
}
|
|
if !claimed {
|
|
return ""
|
|
}
|
|
answer, err := assign(ctx, open, p.Node, p.Module)
|
|
state, note := inventory.PendingApplied, ""
|
|
switch {
|
|
case err != nil:
|
|
state = inventory.PendingRefused
|
|
note = fmt.Sprintf("the pending assignment of %s to %s was refused when build %s registered it: %s",
|
|
p.Module, p.Node, by, firstLine(err.Error()))
|
|
case strings.Contains(answer, "already runs"):
|
|
note = fmt.Sprintf("%s already runs %s: the pending assignment is met", p.Node, p.Module)
|
|
default:
|
|
// The same as a person's assign: nothing is sent by this act.
|
|
note = fmt.Sprintf("%s is assigned %s: build %s registered it (pending since %s); `push %s` sends it",
|
|
p.Node, p.Module, by, clock(p.Since), p.Node)
|
|
}
|
|
settled, serr := inv.SettlePending(ctx, p.ID, inventory.PendingApplying, state, note)
|
|
if serr != nil {
|
|
return fmt.Sprintf("%s — and it could not be recorded as settled: %v", note, serr)
|
|
}
|
|
if !settled {
|
|
return ""
|
|
}
|
|
return note
|
|
}
|
|
|
|
// expirePending ends a waiting pending assignment whose build will not register its module, with why.
|
|
func expirePending(ctx context.Context, inv *inventory.Inventory, p inventory.PendingAssignment, why string) string {
|
|
note := fmt.Sprintf("the pending assignment of %s to %s expired: %s; nothing was assigned. Assign it again with "+
|
|
"build \"true\" to ask for the build again", p.Module, p.Node, why)
|
|
settled, err := inv.SettlePending(ctx, p.ID, inventory.PendingWaiting, inventory.PendingExpired, note)
|
|
if err != nil {
|
|
return fmt.Sprintf("%s — and it could not be recorded: %v", note, err)
|
|
}
|
|
if !settled {
|
|
return ""
|
|
}
|
|
return note
|
|
}
|
|
|
|
// settlingPending is the controller's tick over the pending assignments (ADR 0261), every settleEvery until
|
|
// the context ends. What it does is printed to the controller's log, kept in each row's note (which `status`
|
|
// lists for a day), and raised as a condition for an assignment that was not made.
|
|
func settlingPending(ctx context.Context, open *stores) {
|
|
for {
|
|
for _, line := range settlePending(ctx, open, time.Now()) {
|
|
fmt.Println(line)
|
|
}
|
|
for _, line := range raiseUnknownFields(ctx, open.inventory) {
|
|
fmt.Println(line)
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-time.After(settleEvery):
|
|
}
|
|
}
|
|
}
|
|
|
|
// settlePending is one tick: claims left by a controller that stopped are settled from what the mesh holds;
|
|
// a module registered meanwhile is assigned; a build heard and not registered, or silent past its bound,
|
|
// ends its assignment; ended ones are raised as conditions and cleared once answered. Returns what it did.
|
|
func settlePending(ctx context.Context, open *stores, now time.Time) []string {
|
|
inv := open.inventory
|
|
var said []string
|
|
say := func(format string, args ...any) { said = append(said, fmt.Sprintf(format, args...)) }
|
|
|
|
shelf, err := inv.Catalogue(ctx)
|
|
if err != nil {
|
|
say("the pending assignments are not settled this time: the catalogue could not be read: %v", err)
|
|
return said
|
|
}
|
|
stuck, err := inv.Stuck(ctx, now.Add(-claimStaleAfter))
|
|
if err != nil {
|
|
say("claims left by a stopped controller could not be read: %v", err)
|
|
}
|
|
for _, p := range stuck {
|
|
assigned, err := inv.Assigned(ctx, p.Node)
|
|
if err != nil {
|
|
say("the claim on %s for %s could not be settled: %v", p.Node, p.Module, err)
|
|
continue
|
|
}
|
|
if containsString(assigned, p.Module) {
|
|
if _, err := inv.SettlePending(ctx, p.ID, inventory.PendingApplying, inventory.PendingApplied,
|
|
fmt.Sprintf("%s is assigned %s (settled after the controller that made it stopped)", p.Node, p.Module)); err != nil {
|
|
say("the claim on %s for %s could not be settled: %v", p.Node, p.Module, err)
|
|
}
|
|
continue
|
|
}
|
|
if err := inv.ReleaseClaim(ctx, p.ID); err != nil {
|
|
say("the claim on %s for %s could not be released: %v", p.Node, p.Module, err)
|
|
}
|
|
}
|
|
|
|
waiting, err := inv.Pending(ctx, time.Time{})
|
|
if err != nil {
|
|
say("the pending assignments could not be read to settle them: %v", err)
|
|
}
|
|
for _, p := range waiting {
|
|
if _, ok := shelf[p.Module]; ok {
|
|
if line := applyPending(ctx, open, p, p.Build); line != "" {
|
|
said = append(said, line)
|
|
}
|
|
continue
|
|
}
|
|
if strings.HasPrefix(p.Build, keyPendingPrefix) {
|
|
why, err := keyPendingEnded(ctx, p.Module)
|
|
if err != nil {
|
|
say("the key %s waits for cannot be read: %v", p.Module, err)
|
|
continue
|
|
}
|
|
if why != "" {
|
|
if line := expirePending(ctx, inv, p, why); line != "" {
|
|
said = append(said, line)
|
|
}
|
|
}
|
|
continue
|
|
}
|
|
requests, err := inv.RequestsNamed(ctx, p.Module)
|
|
if err != nil {
|
|
say("the build requests for %s could not be read: %v", p.Module, err)
|
|
continue
|
|
}
|
|
var r *inventory.RequestOutcome
|
|
for i := range requests {
|
|
if requests[i].ID == p.Build {
|
|
r = &requests[i]
|
|
}
|
|
}
|
|
why := ""
|
|
switch {
|
|
case r == nil && now.Sub(p.Since) > buildRequestBound:
|
|
why = fmt.Sprintf("build %s is no longer on record", p.Build)
|
|
case r != nil && !stillComing(*r, now) && (r.Heard || r.NotAsked != "" || r.OutcomeUnknown != ""):
|
|
why = whyNotBuilt(*r, now)
|
|
case r != nil && !stillComing(*r, now):
|
|
why = fmt.Sprintf("no outcome of build %s was heard within %s of asking", p.Build, buildRequestBound)
|
|
}
|
|
if why == "" {
|
|
continue
|
|
}
|
|
if line := expirePending(ctx, inv, p, why); line != "" {
|
|
said = append(said, line)
|
|
}
|
|
}
|
|
|
|
said = append(said, raisePendingEnded(ctx, inv)...)
|
|
return said
|
|
}
|
|
|
|
func containsString(xs []string, x string) bool {
|
|
for _, y := range xs {
|
|
if y == x {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// pendingObservation is the condition for a pending assignment that was not made: one per pending
|
|
// assignment, its row's id the last part of its id, so one row's condition is never another's.
|
|
func pendingObservation(p inventory.PendingAssignment) conditions.Observation {
|
|
return conditions.Observation{Scope: conditions.ScopeMachine, ID: fmt.Sprintf("%s.%s.%s", p.Node, p.Module, p.ID),
|
|
Token: kindPendingEnded, Kind: kindPendingEnded, Machine: p.Node, Severity: conditions.Warning,
|
|
Resolver: conditions.ResolverOperator, Source: "pending assignments",
|
|
Summary: p.Note}
|
|
}
|
|
|
|
// raisePendingEnded raises a condition for each pending assignment that expired or was refused, once, and
|
|
// clears it when it is answered: the module was assigned to that machine since, a newer pending assignment of
|
|
// it is open, or a person took it back with `unassign`. Where this process keeps no conditions, nothing is
|
|
// stamped, and the serving controller's tick raises it.
|
|
func raisePendingEnded(ctx context.Context, inv *inventory.Inventory) []string {
|
|
keeper := conditionsFrom
|
|
if keeper == nil {
|
|
return nil
|
|
}
|
|
var said []string
|
|
toRaise, err := inv.ToRaise(ctx)
|
|
if err != nil {
|
|
return []string{fmt.Sprintf("pending assignments not made could not be read to raise them: %v", err)}
|
|
}
|
|
for _, p := range toRaise {
|
|
if _, err := keeper.Observe(ctx, pendingObservation(p)); err != nil {
|
|
said = append(said, fmt.Sprintf("the condition for %s on %s could not be raised: %v", p.Module, p.Node, err))
|
|
continue
|
|
}
|
|
if err := inv.MarkPending(ctx, p.ID, "raised"); err != nil {
|
|
said = append(said, fmt.Sprintf("the condition for %s on %s was raised and not recorded: %v", p.Module, p.Node, err))
|
|
}
|
|
}
|
|
toClear, err := inv.ToClear(ctx)
|
|
if err != nil {
|
|
return append(said, fmt.Sprintf("pending assignments raised could not be read to clear them: %v", err))
|
|
}
|
|
for _, p := range toClear {
|
|
why, answered, err := pendingAnswered(ctx, inv, p)
|
|
if err != nil {
|
|
said = append(said, fmt.Sprintf("whether %s on %s is answered could not be read: %v", p.Module, p.Node, err))
|
|
continue
|
|
}
|
|
if !answered {
|
|
continue
|
|
}
|
|
if _, err := keeper.Clear(ctx, pendingObservation(p).Key(), why); err != nil {
|
|
said = append(said, fmt.Sprintf("the condition for %s on %s could not be cleared: %v", p.Module, p.Node, err))
|
|
continue
|
|
}
|
|
if err := inv.MarkPending(ctx, p.ID, "cleared"); err != nil {
|
|
said = append(said, fmt.Sprintf("the condition for %s on %s was cleared and not recorded: %v", p.Module, p.Node, err))
|
|
}
|
|
}
|
|
return append(said, clearOrphaned(ctx, keeper, inv)...)
|
|
}
|
|
|
|
// pendingAnswered says whether a pending assignment that was not made has been answered since, and how.
|
|
func pendingAnswered(ctx context.Context, inv *inventory.Inventory, p inventory.PendingAssignment) (string, bool, error) {
|
|
if p.Acknowledged != nil {
|
|
return "taken back with unassign", true, nil
|
|
}
|
|
assigned, err := inv.Assigned(ctx, p.Node)
|
|
if err != nil {
|
|
return "", false, err
|
|
}
|
|
if containsString(assigned, p.Module) {
|
|
return p.Module + " is assigned to " + p.Node + " since", true, nil
|
|
}
|
|
rows, err := inv.PendingFor(ctx, p.Node, p.Module)
|
|
if err != nil {
|
|
return "", false, err
|
|
}
|
|
// A newer pending assignment answers it only while it is open or once it was made: one that itself
|
|
// ended unmade is its own condition, and answers nothing.
|
|
for _, r := range rows {
|
|
if r.ID != p.ID && r.Since.After(p.Since) && (r.Open() || r.State == inventory.PendingApplied) {
|
|
return "assigned again, pending on another build", true, nil
|
|
}
|
|
}
|
|
return "", false, nil
|
|
}
|
|
|
|
// clearOrphaned clears every open condition of an assignment not made whose pending assignment is no longer
|
|
// on record: its machine was removed, which takes its pending assignments with it.
|
|
func clearOrphaned(ctx context.Context, keeper *conditions.Keeper, inv *inventory.Inventory) []string {
|
|
open, err := keeper.Open(ctx)
|
|
if err != nil {
|
|
return []string{fmt.Sprintf("the open conditions could not be read to clear orphaned ones: %v", err)}
|
|
}
|
|
byID := map[string]string{}
|
|
var ids []string
|
|
for _, c := range open {
|
|
if c.Kind != kindPendingEnded {
|
|
continue
|
|
}
|
|
// `<scope>.<node>.<module>.<row>.<kind>`: the row is the part before the kind.
|
|
parts := strings.Split(c.Key, ".")
|
|
if len(parts) < 2 {
|
|
continue
|
|
}
|
|
id := parts[len(parts)-2]
|
|
if !strings.HasPrefix(id, "p-") {
|
|
continue
|
|
}
|
|
byID[id] = c.Key
|
|
ids = append(ids, id)
|
|
}
|
|
if len(ids) == 0 {
|
|
return nil
|
|
}
|
|
known, err := inv.PendingKnown(ctx, ids)
|
|
if err != nil {
|
|
return []string{fmt.Sprintf("the pending assignments could not be read to clear orphaned conditions: %v", err)}
|
|
}
|
|
var said []string
|
|
for id, key := range byID {
|
|
if known[id] {
|
|
continue
|
|
}
|
|
if _, err := keeper.Clear(ctx, key, "its pending assignment is no longer on record: its machine was removed, or the module was assigned and the wait taken back"); err != nil {
|
|
said = append(said, fmt.Sprintf("the condition %s could not be cleared: %v", key, err))
|
|
}
|
|
}
|
|
return said
|
|
}
|
|
|
|
// withdrawPending is `unassign` of a module only pending on a machine, under the machine's hold: a waiting
|
|
// pending assignment is withdrawn; one being made now is refused, never overridden; one that expired or was
|
|
// refused is taken back, which answers its condition. False when there is none.
|
|
func withdrawPending(ctx context.Context, inv *inventory.Inventory, node, module string) (string, bool, error) {
|
|
rows, err := inv.PendingFor(ctx, node, module)
|
|
if err != nil {
|
|
return "", false, err
|
|
}
|
|
var takenBack []string
|
|
for _, p := range rows {
|
|
switch p.State {
|
|
case inventory.PendingApplying:
|
|
return "", true, fmt.Errorf("%s is being assigned %s now, as build %s registered it: nothing was "+
|
|
"withdrawn; `unassign %s %s` again once it is assigned takes it off", node, module, p.Build, node, module)
|
|
case inventory.PendingWaiting:
|
|
note := fmt.Sprintf("the pending assignment of %s to %s is withdrawn; build %s goes on, and registers "+
|
|
"%s assigned nowhere", module, node, p.Build, module)
|
|
settled, err := inv.SettlePending(ctx, p.ID, inventory.PendingWaiting, inventory.PendingWithdrawn, note)
|
|
if err != nil {
|
|
return "", false, err
|
|
}
|
|
if !settled {
|
|
return "", true, fmt.Errorf("the pending assignment of %s to %s changed while it was withdrawn: "+
|
|
"`status` says how; nothing was withdrawn", module, node)
|
|
}
|
|
return note, true, nil
|
|
case inventory.PendingExpired, inventory.PendingRefused:
|
|
if p.Cleared != nil || p.Acknowledged != nil {
|
|
continue
|
|
}
|
|
if err := inv.MarkPending(ctx, p.ID, "acknowledged"); err != nil {
|
|
return "", false, err
|
|
}
|
|
takenBack = append(takenBack, fmt.Sprintf("build %s (%s)", p.Build, p.State))
|
|
}
|
|
}
|
|
if len(takenBack) > 0 {
|
|
return fmt.Sprintf("the pending assignment(s) of %s to %s that were not made are taken back: %s; their "+
|
|
"conditions clear", module, node, strings.Join(takenBack, ", ")), true, nil
|
|
}
|
|
return "", false, nil
|
|
}
|
|
|
|
// keyPendingEnded is why a pending assignment waiting for a module's key ends: the indexer no longer holds the module,
|
|
// or the builder's key of its fingerprint failed; empty while it is coming. A variable so a test needs no bus.
|
|
var keyPendingEnded = func(ctx context.Context, module string) (string, error) {
|
|
keys, closeKeys, err := openBuilderKeys()
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
defer closeKeys()
|
|
v, err := keys.ModuleAt(ctx, module, "")
|
|
if errors.Is(err, builderkeys.ErrNoValue) {
|
|
return fmt.Sprintf("the indexer no longer holds %s", module), nil
|
|
}
|
|
if err != nil || v.FP2 == "" {
|
|
return "", err
|
|
}
|
|
k, err := keys.Build(ctx, module, v.FP2)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
if k.Failed() {
|
|
return k.Says(), nil
|
|
}
|
|
return "", nil
|
|
}
|
|
|
|
// settlePendingOnKey assigns, now, every pending assignment of a module its done key just registered (ADR 0501 §6):
|
|
// what an outcome of a build did for one waiting on that build.
|
|
func settlePendingOnKey(ctx context.Context, open *stores, module string) []string {
|
|
waiting, err := open.inventory.Pending(ctx, time.Time{})
|
|
if err != nil {
|
|
return []string{fmt.Sprintf("the pending assignments could not be read after %s was registered: %v", module, err)}
|
|
}
|
|
var said []string
|
|
for _, p := range waiting {
|
|
if p.Module == module {
|
|
if line := applyPending(ctx, open, p, keyPendingPrefix+module); line != "" {
|
|
said = append(said, line)
|
|
}
|
|
}
|
|
}
|
|
return said
|
|
}
|