Merge remote-tracking branch 'origin/main' into feat/upgrade-safety
This commit is contained in:
@@ -800,6 +800,9 @@ type keptMerge struct {
|
||||
// FirstBuild says the merge is a first merge into a repository the mesh builds nothing from, kept to ask the first
|
||||
// build of its modules once mesh-delivery's word on it is kept (novox/hq ADR 0432, issue 1146).
|
||||
FirstBuild bool `json:"first_build,omitempty"`
|
||||
// Held says which first builds of the merge were refused for a word the running build-agent does not read: the merge
|
||||
// waits for a build-agent that reads them, and is never failed (novox/hq ADR 0482 §5, issue 1173).
|
||||
Held []heldFirstBuild `json:"held,omitempty"`
|
||||
}
|
||||
|
||||
// announcedOf is a kept merge as the forge announced it, and what it moved; empty where the record says none.
|
||||
|
||||
@@ -58,8 +58,17 @@ type stalledLine struct {
|
||||
// Severity is "urgent" for a group held or half merged and a refused member (novox/hq issues 1275, 1278); empty
|
||||
// or anything else is a warning, as before.
|
||||
Severity string `json:"severity,omitempty"`
|
||||
// Needs is "new-head" for a delivery whose cure is its author's new head (novox/hq ADR 0505 §5), Author that author
|
||||
// as the operator reads it ("the author of feat/x", with the session that pushed it when its pull request names one)
|
||||
// and Branch its branch. A mesh-delivery from before says none, and the words are as before.
|
||||
Needs string `json:"needs,omitempty"`
|
||||
Author string `json:"author,omitempty"`
|
||||
Branch string `json:"branch,omitempty"`
|
||||
}
|
||||
|
||||
// needsNewHead is what a stalled line needs when its cure is its author's new head (ADR 0505 §5).
|
||||
const needsNewHead = "new-head"
|
||||
|
||||
// severity is the condition's severity a stalled line asks: urgent when it says so, a warning otherwise.
|
||||
func (l stalledLine) severity() conditions.Severity {
|
||||
if l.Severity == string(conditions.Urgent) {
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"regexp"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
@@ -102,7 +103,7 @@ func (f following) resumeFirstBuilds(ctx context.Context, d deliveryMerge) error
|
||||
}
|
||||
for _, w := range waiting {
|
||||
k := announcedOf(w)
|
||||
if !k.FirstBuild || !strings.EqualFold(k.Owner+"/"+k.Repo, d.Owner+"/"+d.Repo) || k.Number != d.Number ||
|
||||
if !k.FirstBuild || len(k.Held) > 0 || !strings.EqualFold(k.Owner+"/"+k.Repo, d.Owner+"/"+d.Repo) || k.Number != d.Number ||
|
||||
k.HeadSHA != d.HeadSHA {
|
||||
continue
|
||||
}
|
||||
@@ -172,12 +173,122 @@ func firstBuildsOf(ctx context.Context, inv catalogueReader, m link.SourceMoved,
|
||||
return nil, "", err
|
||||
}
|
||||
|
||||
// heldFirstBuild says a first build was refused because its manifest uses a word the running build-agent does not read,
|
||||
// and its merge is kept waiting for a build-agent that reads it (novox/hq ADR 0482 §5, issue 1173).
|
||||
type heldFirstBuild struct {
|
||||
Module string `json:"module"`
|
||||
Path string `json:"path"`
|
||||
Word string `json:"word"`
|
||||
At time.Time `json:"at"`
|
||||
}
|
||||
|
||||
// judgeRefusedFirstBuild is the hold a first build's outcome puts its merge under: every module the merge holds already,
|
||||
// and this one, unless it is among them; nil unless the build failed on a word of its manifest the running build-agent
|
||||
// does not know, and was asked for a first merge, answered by the record of merges no plan answers or put back to waiting
|
||||
// by an earlier hold (a second refusal in one merge is held too, review of issue 1173).
|
||||
func judgeRefusedFirstBuild(result link.BuildResult, req inventory.BuildRequest, merge inventory.BatchedMerge) []heldFirstBuild {
|
||||
word := refusedWord(result.Failed)
|
||||
if word == "" || req.For != "merge" {
|
||||
return nil
|
||||
}
|
||||
var held []heldFirstBuild
|
||||
switch k := announcedOf(merge); {
|
||||
case merge.Plan == inventory.WaitingRecord:
|
||||
// Answered when its builds were asked: a hold from before (a build asked again) is over, and starts anew.
|
||||
case merge.Plan == "" && len(k.Held) > 0:
|
||||
held = k.Held
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
this := heldFirstBuild{Module: req.Name(), Path: strings.Trim(req.Path, "/"), Word: word, At: time.Now().UTC()}
|
||||
for i, h := range held {
|
||||
if strings.Trim(h.Path, "/") == this.Path {
|
||||
out := append([]heldFirstBuild(nil), held...)
|
||||
out[i] = this
|
||||
return out
|
||||
}
|
||||
}
|
||||
return append(append([]heldFirstBuild(nil), held...), this)
|
||||
}
|
||||
|
||||
// heldDirsOf is the directories a held merge asks again: its held modules' own, each once, "." for the repository's root.
|
||||
func heldDirsOf(k keptMerge) []string {
|
||||
var dirs []string
|
||||
seen := map[string]bool{}
|
||||
for _, h := range k.Held {
|
||||
dir := strings.Trim(h.Path, "/")
|
||||
if dir == "" {
|
||||
dir = "."
|
||||
}
|
||||
if !seen[dir] {
|
||||
seen[dir] = true
|
||||
dirs = append(dirs, dir)
|
||||
}
|
||||
}
|
||||
return dirs
|
||||
}
|
||||
|
||||
// unknownWord is how the strict reading of a manifest says a word is not one it knows: encoding/json's own words, which
|
||||
// the build-agent's answer carries whole. The build's answer has no typed field for it (Failed is the builder's words).
|
||||
var unknownWord = regexp.MustCompile(`json: unknown field "([^"]+)"`)
|
||||
|
||||
// refusedWord is the word a build's failure says the reader did not know; empty when it says none.
|
||||
func refusedWord(failed string) string {
|
||||
if m := unknownWord.FindStringSubmatch(failed); m != nil {
|
||||
return m[1]
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// holdRefusedFirstBuild keeps a first build's merge waiting when the build was refused for a word the running build-agent
|
||||
// does not read (novox/hq ADR 0482 §5, issue 1173): no failed build is recorded, nothing expires the merge, and it is
|
||||
// named by a condition until a delivery of build-agent asks the build again. Answers whether the outcome was dealt with.
|
||||
func holdRefusedFirstBuild(ctx context.Context, inv *inventory.Inventory, result link.BuildResult) (bool, error) {
|
||||
if result.Failed == "" || refusedWord(result.Failed) == "" {
|
||||
return false, nil
|
||||
}
|
||||
req, found, err := inv.BuildRequestByID(ctx, result.ID)
|
||||
if err != nil || !found {
|
||||
return false, err
|
||||
}
|
||||
merge, found, err := inv.MergeOf(ctx, req.Repository, req.Commit)
|
||||
if err != nil || !found {
|
||||
return false, err
|
||||
}
|
||||
held := judgeRefusedFirstBuild(result, req, merge)
|
||||
if len(held) == 0 {
|
||||
return false, nil
|
||||
}
|
||||
k := announcedOf(merge)
|
||||
k.Held = held
|
||||
event, err := json.Marshal(k)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
if err := inv.HoldMerge(ctx, merge.Repository, merge.Commit, event); err != nil {
|
||||
return false, err
|
||||
}
|
||||
this := held[len(held)-1]
|
||||
for _, h := range held {
|
||||
if h.Path == strings.Trim(req.Path, "/") {
|
||||
this = h
|
||||
}
|
||||
}
|
||||
fmt.Printf("%s: the first build of %s was refused for the word %q, which the running build-agent does not read; the "+
|
||||
"merge %.8s of %s is kept waiting for a build-agent that reads it (holding %s), and the build is asked again when "+
|
||||
"one is delivered\n", result.ID, this.Module, this.Word, merge.Commit, merge.Repository, heldNames(held))
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// probeFirstBuildsID is the probe that reads the first merges waiting for mesh-delivery's word.
|
||||
const probeFirstBuildsID = "D18"
|
||||
|
||||
// kindFirstBuildUnasked is what a first merge waiting too long for mesh-delivery's word raises.
|
||||
const kindFirstBuildUnasked = "first-build-unasked"
|
||||
|
||||
// kindFirstBuildHeld is what a first merge held for a build-agent that reads its manifest raises.
|
||||
const kindFirstBuildHeld = "first-build-held"
|
||||
|
||||
// firstBuildWaitBound is how long a first merge waits for mesh-delivery's announcement of the operator's Approve before
|
||||
// it is said: a bound in the sense of ADR 0426 §3, not a delay. Nothing waits on it: the word's arrival asks the builds
|
||||
// at once, and the merge is said, never dropped, once it is past this (issue 1146).
|
||||
@@ -189,6 +300,10 @@ func firstBuildObservations(waiting []inventory.BatchedMerge, now time.Time) []c
|
||||
var out []conditions.Observation
|
||||
for _, w := range waiting {
|
||||
k := announcedOf(w)
|
||||
if len(k.Held) > 0 {
|
||||
out = append(out, heldFirstBuildObservation(k))
|
||||
continue
|
||||
}
|
||||
if !k.FirstBuild || now.Sub(w.Heard) <= firstBuildWaitBound {
|
||||
continue
|
||||
}
|
||||
@@ -211,6 +326,37 @@ func firstBuildObservations(waiting []inventory.BatchedMerge, now time.Time) []c
|
||||
return out
|
||||
}
|
||||
|
||||
// heldNames is the modules a merge holds, each with the word it waits for: "watchman (key-ttl), view (per-machine)".
|
||||
func heldNames(held []heldFirstBuild) string {
|
||||
names := make([]string, 0, len(held))
|
||||
for _, h := range held {
|
||||
names = append(names, fmt.Sprintf("%s (%q)", h.Module, h.Word))
|
||||
}
|
||||
return strings.Join(names, ", ")
|
||||
}
|
||||
|
||||
// heldFirstBuildObservation names a first merge held for a build-agent that reads its manifests: at once, since nothing
|
||||
// is lost and nothing times out, but the operator can see what the merge waits for. One condition names every module held.
|
||||
func heldFirstBuildObservation(k keptMerge) conditions.Observation {
|
||||
repository := k.Owner + "/" + k.Repo
|
||||
modules := make([]string, 0, len(k.Held))
|
||||
for _, h := range k.Held {
|
||||
modules = append(modules, h.Module)
|
||||
}
|
||||
held := heldNames(k.Held)
|
||||
return conditions.Observation{Scope: conditions.ScopeMerge, ID: fmt.Sprintf("%s_%.8s", repository, k.Commit),
|
||||
Kind: kindFirstBuildHeld, Severity: conditions.Warning,
|
||||
Summary: fmt.Sprintf("the merge %.8s of %s adds modules whose manifests use words the running build-agent does not "+
|
||||
"read: %s; the merge waits for a build-agent that reads them", k.Commit, repository, held),
|
||||
Said: fmt.Sprintf("%s at %s waits for a build-agent that reads the words of %s", repository, k.Commit, held),
|
||||
Headline: fmt.Sprintf("The first build of %s waits for a newer build-agent", strings.Join(modules, ", ")),
|
||||
Explanation: fmt.Sprintf("The merge %.8s into %s adds modules whose manifests use words the build-agent now "+
|
||||
"running does not know: %s. The merge is kept, not failed: their first builds are asked again when a delivery "+
|
||||
"of build-agent ends.", k.Commit, repository, held),
|
||||
Resolved: "a delivery of build-agent ends and the first builds register the modules, or a hand `build` does",
|
||||
}
|
||||
}
|
||||
|
||||
// probeFirstBuilds is D18: no first merge waits past its bound for mesh-delivery's word unsaid.
|
||||
func probeFirstBuilds(ctx context.Context, d *doctor) ([]conditions.Observation, error) {
|
||||
waiting, err := d.open.inventory.WaitingMerges(ctx)
|
||||
@@ -219,3 +365,61 @@ func probeFirstBuilds(ctx context.Context, d *doctor) ([]conditions.Observation,
|
||||
}
|
||||
return firstBuildObservations(waiting, time.Now().UTC()), nil
|
||||
}
|
||||
|
||||
// askHeldFirstBuildsAgain asks again the first build of each first merge held for a build-agent that reads its manifest,
|
||||
// when the plan that ended delivered build-agent (novox/hq issue 1173): once for each end, since the ask answers the merge
|
||||
// and a refusal that comes again holds it anew. Answers how many merges it asked for.
|
||||
func askHeldFirstBuildsAgain(ctx context.Context, open *stores, p inventory.Plan) int {
|
||||
if !deliversBuildAgent(p) {
|
||||
return 0
|
||||
}
|
||||
actingOnMerges.Lock()
|
||||
defer actingOnMerges.Unlock()
|
||||
inv := open.inventory
|
||||
waiting, err := inv.WaitingMerges(ctx)
|
||||
if err != nil {
|
||||
fmt.Printf("%s delivered build-agent, and the first merges held for it could not be read: %v\n", p.ID, err)
|
||||
return 0
|
||||
}
|
||||
asked := 0
|
||||
for _, w := range waiting {
|
||||
k := announcedOf(w)
|
||||
if len(k.Held) == 0 {
|
||||
continue
|
||||
}
|
||||
// Its own held modules, not the rule of a first merge: a module of the merge that passed is registered by now,
|
||||
// and the repository no longer counts as one the mesh builds nothing from (review of issue 1173).
|
||||
why, err := approvedAnnounced(ctx, inv, k.Owner, k.Repo, keptMerge{SourceMoved: k.SourceMoved})
|
||||
if err != nil {
|
||||
fmt.Printf("%s delivered build-agent, and the merge %.8s of %s held for it asks no build now (%v): it stays "+
|
||||
"waiting\n", p.ID, w.Commit, w.Repository, err)
|
||||
continue
|
||||
}
|
||||
dirs := heldDirsOf(k)
|
||||
fmt.Printf("%s delivered build-agent: the first builds of %s are asked again\n", p.ID, heldNames(k.Held))
|
||||
paths := make([]string, 0, len(dirs))
|
||||
for _, dir := range dirs {
|
||||
if dir == "." {
|
||||
dir = ""
|
||||
}
|
||||
paths = append(paths, dir)
|
||||
}
|
||||
repository := k.Owner + "/" + k.Repo
|
||||
askBuilds(withFirstBuildWarrant(ctx, repository, paths, why), inv, k.SourceMoved,
|
||||
buildSource{Repository: repository, Seat: gitSeat}, dirs)
|
||||
// Answered: it no longer waits, so this end asks nothing twice; a refusal that comes again holds it anew.
|
||||
if err := inv.AnswerMerge(ctx, w.Repository, w.Commit, inventory.WaitingRecord, false); err != nil {
|
||||
fmt.Printf(" and the merge could not be marked answered: %v\n", err)
|
||||
continue
|
||||
}
|
||||
asked++
|
||||
}
|
||||
return asked
|
||||
}
|
||||
|
||||
// deliversBuildAgent says a delivery plan ended delivered with build-agent built in it: the reader a held first build
|
||||
// waits for is the one now running.
|
||||
func deliversBuildAgent(p inventory.Plan) bool {
|
||||
m := p.Modules["build-agent"]
|
||||
return p.WalkSaid == inventory.WalkEnded && p.Outcome() == inventory.WalkDelivered && m != nil && m.State == "built"
|
||||
}
|
||||
|
||||
@@ -0,0 +1,221 @@
|
||||
//go:build integration
|
||||
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
"github.com/novox/mesh-controller/internal/link"
|
||||
)
|
||||
|
||||
// 1173/AC1, 1173/AC3 whole against the store: a first merge's build refused for a word the running build-agent does not
|
||||
// read keeps the merge waiting with no failed build recorded; a delivery plan that does not deliver build-agent asks
|
||||
// nothing; the one that ends delivering it asks the first build again once, and its pass registers the module.
|
||||
func TestAHeldFirstMergeIsAskedAgainWhenABuildAgentIsDelivered(t *testing.T) {
|
||||
const head, commit = "a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1", "0177ea7a00000000"
|
||||
open := aMesh(t)
|
||||
ctx := t.Context()
|
||||
inv := open.inventory
|
||||
|
||||
a := aDeliveryMergeOf(t, head, map[string]string{"novox/mesh-delivery@a1a1a1a1a1a1": head})
|
||||
wasKept, wasAsk, realAsk := keptDeliveryMerge, routerAskOf, askABuild
|
||||
keptDeliveryMerge = func(context.Context, string, string, int, string) (deliveryMerge, bool, error) {
|
||||
return a.merge, true, nil
|
||||
}
|
||||
routerAskOf = func(context.Context, catalogueReader, string, string) (*asks.Ask, *asks.Warrant, error) {
|
||||
return &a.ask, &a.warrant, nil
|
||||
}
|
||||
var asked []string
|
||||
askABuild = func(_ context.Context, source buildSource, path, ref string) (string, error) {
|
||||
id := "build-" + string(rune('1'+len(asked)))
|
||||
asked = append(asked, id)
|
||||
return id, nil
|
||||
}
|
||||
t.Cleanup(func() { keptDeliveryMerge, routerAskOf, askABuild = wasKept, wasAsk, realAsk })
|
||||
|
||||
m := link.SourceMoved{Owner: "novox", Repo: "mesh-delivery", Base: "main", HeadSHA: head, Commit: commit, Number: 7,
|
||||
MergedAt: time.Now().Add(-time.Hour).UTC().Format(time.RFC3339Nano), Paths: []string{"module.json", "main.go"},
|
||||
ModuleDirs: []string{"."}, ModuleDirsSaid: true}
|
||||
if err := (following{open: open}).SourceMoved(ctx, m); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(asked) != 1 {
|
||||
t.Fatalf("1173/AC3: the first merge asked %v, want the one first build", asked)
|
||||
}
|
||||
|
||||
// The running build-agent refuses the manifest's word.
|
||||
refused := onTrunk("build-1", "novox/mesh-delivery", "git", "", nil)
|
||||
refused.Failed, refused.Manifest, refused.Commit = refusedKeyTTL, nil, commit
|
||||
if err := (builds{inv: inv, open: open}).Built(ctx, refused); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, recorded, _ := inv.BuildByID(ctx, "build-1"); recorded {
|
||||
t.Fatal("1173/AC1: a refused first build was recorded as a failed build")
|
||||
}
|
||||
waiting, err := inv.WaitingMerges(ctx)
|
||||
if err != nil || len(waiting) != 1 || len(announcedOf(waiting[0]).Held) == 0 {
|
||||
t.Fatalf("1173/AC1: the merges waiting are %+v (%v), want the first merge, held", waiting, err)
|
||||
}
|
||||
if obs := firstBuildObservations(waiting, time.Now()); len(obs) != 1 || obs[0].Kind != kindFirstBuildHeld {
|
||||
t.Fatalf("1173/AC2: the held merge is named by %+v", obs)
|
||||
}
|
||||
|
||||
// A plan that ended without delivering build-agent asks nothing; one that delivered it asks once.
|
||||
other := inventory.Plan{ID: "p1", State: inventory.PlanDone, WalkSaid: inventory.WalkEnded,
|
||||
Modules: map[string]*inventory.PlanModule{"mesh-delivery": {State: "built"}}}
|
||||
if n := askHeldFirstBuildsAgain(ctx, open, other); n != 0 || len(asked) != 1 {
|
||||
t.Fatalf("1173/AC3: a plan that delivered no build-agent asked %d builds again (%v)", n, asked)
|
||||
}
|
||||
delivered := inventory.Plan{ID: "p2", State: inventory.PlanDone, WalkSaid: inventory.WalkEnded,
|
||||
Modules: map[string]*inventory.PlanModule{"build-agent": {State: "built"}}}
|
||||
if n := askHeldFirstBuildsAgain(ctx, open, delivered); n != 1 || len(asked) != 2 {
|
||||
t.Fatalf("1173/AC3: the plan that delivered build-agent asked %d builds again (%v), want one", n, asked)
|
||||
}
|
||||
if n := askHeldFirstBuildsAgain(ctx, open, delivered); n != 0 || len(asked) != 2 {
|
||||
t.Fatalf("1173/AC3: the same end asked again: %d (%v)", n, asked)
|
||||
}
|
||||
|
||||
// Its pass registers the module.
|
||||
pass := onTrunk("build-2", "novox/mesh-delivery", "git", "", map[string]any{"module": "mesh-delivery-watchman", "version": "1"})
|
||||
pass.Commit = commit
|
||||
if err := (builds{inv: inv, open: open}).Built(ctx, pass); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, known := inventoryHolds(t, inv, "mesh-delivery-watchman"); !known {
|
||||
t.Fatal("1173/AC3: the module is not registered after the first build passed")
|
||||
}
|
||||
}
|
||||
|
||||
// aTwoModuleFirstMerge is a first merge into novox/mesh-delivery adding the modules watchman and view, made on the
|
||||
// operator's Approve of its head, whose first builds are asked; asked says the build id asked for each path, in order.
|
||||
func aTwoModuleFirstMerge(t *testing.T, open *stores, commit string) *[][2]string {
|
||||
t.Helper()
|
||||
const head = "b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2"
|
||||
a := aDeliveryMergeOf(t, head, map[string]string{"novox/mesh-delivery@b2b2b2b2b2b2": head})
|
||||
wasKept, wasAsk, realAsk := keptDeliveryMerge, routerAskOf, askABuild
|
||||
keptDeliveryMerge = func(context.Context, string, string, int, string) (deliveryMerge, bool, error) {
|
||||
return a.merge, true, nil
|
||||
}
|
||||
routerAskOf = func(context.Context, catalogueReader, string, string) (*asks.Ask, *asks.Warrant, error) {
|
||||
return &a.ask, &a.warrant, nil
|
||||
}
|
||||
asked := &[][2]string{}
|
||||
askABuild = func(_ context.Context, _ buildSource, path, _ string) (string, error) {
|
||||
id := fmt.Sprintf("build-%d", len(*asked)+1)
|
||||
*asked = append(*asked, [2]string{path, id})
|
||||
return id, nil
|
||||
}
|
||||
t.Cleanup(func() { keptDeliveryMerge, routerAskOf, askABuild = wasKept, wasAsk, realAsk })
|
||||
m := link.SourceMoved{Owner: "novox", Repo: "mesh-delivery", Base: "main", HeadSHA: head, Commit: commit, Number: 7,
|
||||
MergedAt: time.Now().Add(-time.Hour).UTC().Format(time.RFC3339Nano),
|
||||
Paths: []string{"view/module.json", "view/main.go", "watchman/module.json", "watchman/main.go"},
|
||||
ModuleDirs: []string{"view", "watchman"}, ModuleDirsSaid: true}
|
||||
if err := (following{open: open}).SourceMoved(t.Context(), m); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(*asked) != 2 {
|
||||
t.Fatalf("the first merge asked %v, want the two first builds", *asked)
|
||||
}
|
||||
return asked
|
||||
}
|
||||
|
||||
// idOf is the build asked last for path.
|
||||
func idOf(t *testing.T, asked [][2]string, path string) string {
|
||||
t.Helper()
|
||||
for i := len(asked) - 1; i >= 0; i-- {
|
||||
if asked[i][0] == path {
|
||||
return asked[i][1]
|
||||
}
|
||||
}
|
||||
t.Fatalf("no build asked for %s: %v", path, asked)
|
||||
return ""
|
||||
}
|
||||
|
||||
// built hands the controller a build's outcome: refused for the word, or passing as module.
|
||||
func built(t *testing.T, open *stores, id, path, commit, refusedFor, module string) {
|
||||
t.Helper()
|
||||
r := onTrunk(id, "novox/mesh-delivery", "git", path, map[string]any{"module": module, "version": "1"})
|
||||
r.Commit = commit
|
||||
if refusedFor != "" {
|
||||
r.Failed, r.Manifest = `this is not a module manifest: json: unknown field "`+refusedFor+`"`, nil
|
||||
}
|
||||
if err := (builds{inv: open.inventory, open: open}).Built(t.Context(), r); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
var buildAgentDelivered = inventory.Plan{ID: "p-ba", State: inventory.PlanDone, WalkSaid: inventory.WalkEnded,
|
||||
Modules: map[string]*inventory.PlanModule{"build-agent": {State: "built"}}}
|
||||
|
||||
// 1173/AC3 (review round 1, item 1): in a first merge of two modules where view passes and watchman is refused, the
|
||||
// held watchman is asked again when build-agent is delivered, though the repository builds view by then.
|
||||
func TestAHeldModuleIsAskedAgainAfterItsSiblingRegistered(t *testing.T) {
|
||||
const commit = "0177ea7b00000000"
|
||||
open := aMesh(t)
|
||||
ctx := t.Context()
|
||||
asked := aTwoModuleFirstMerge(t, open, commit)
|
||||
|
||||
built(t, open, idOf(t, *asked, "view"), "view", commit, "", "mesh-delivery-view")
|
||||
if _, known := inventoryHolds(t, open.inventory, "mesh-delivery-view"); !known {
|
||||
t.Fatal("view is not registered after its first build passed")
|
||||
}
|
||||
refused := idOf(t, *asked, "watchman")
|
||||
built(t, open, refused, "watchman", commit, "key-ttl", "")
|
||||
if _, recorded, _ := open.inventory.BuildByID(ctx, refused); recorded {
|
||||
t.Fatal("1173/AC1: the refused first build of watchman was recorded as a failed build")
|
||||
}
|
||||
waiting, err := open.inventory.WaitingMerges(ctx)
|
||||
if err != nil || len(waiting) != 1 || len(announcedOf(waiting[0]).Held) != 1 {
|
||||
t.Fatalf("1173/AC1: the merges waiting are %+v (%v), want the first merge, holding watchman", waiting, err)
|
||||
}
|
||||
|
||||
if n := askHeldFirstBuildsAgain(ctx, open, buildAgentDelivered); n != 1 || len(*asked) != 3 || (*asked)[2][0] != "watchman" {
|
||||
t.Fatalf("1173/AC3: the delivery of build-agent asked %d merges again (%v), want watchman's first build alone", n, *asked)
|
||||
}
|
||||
built(t, open, idOf(t, *asked, "watchman"), "watchman", commit, "", "mesh-delivery-watchman")
|
||||
if _, known := inventoryHolds(t, open.inventory, "mesh-delivery-watchman"); !known {
|
||||
t.Fatal("1173/AC3: watchman is not registered after the first build asked again passed")
|
||||
}
|
||||
}
|
||||
|
||||
// 1173/AC1, AC2, AC3 (review round 1, item 2): in a first merge of two modules both refused, neither is recorded as a
|
||||
// failed build, the condition names both, and the delivery of build-agent asks both again.
|
||||
func TestTwoRefusedModulesOfOneFirstMergeAreBothHeld(t *testing.T) {
|
||||
const commit = "0177ea7c00000000"
|
||||
open := aMesh(t)
|
||||
ctx := t.Context()
|
||||
asked := aTwoModuleFirstMerge(t, open, commit)
|
||||
|
||||
for _, path := range []string{"watchman", "view"} {
|
||||
id := idOf(t, *asked, path)
|
||||
built(t, open, id, path, commit, "key-ttl", "")
|
||||
if _, recorded, _ := open.inventory.BuildByID(ctx, id); recorded {
|
||||
t.Fatalf("1173/AC1: the refused first build of %s was recorded as a failed build", path)
|
||||
}
|
||||
}
|
||||
waiting, err := open.inventory.WaitingMerges(ctx)
|
||||
if err != nil || len(waiting) != 1 || len(announcedOf(waiting[0]).Held) != 2 {
|
||||
t.Fatalf("1173/AC1: the merges waiting are %+v (%v), want the first merge, holding both", waiting, err)
|
||||
}
|
||||
obs := firstBuildObservations(waiting, time.Now())
|
||||
if len(obs) != 1 || !strings.Contains(obs[0].Summary, "watchman") || !strings.Contains(obs[0].Summary, "view") {
|
||||
t.Fatalf("1173/AC2: the held merge is named by %+v, want one condition naming both", obs)
|
||||
}
|
||||
|
||||
if n := askHeldFirstBuildsAgain(ctx, open, buildAgentDelivered); n != 1 || len(*asked) != 4 {
|
||||
t.Fatalf("1173/AC3: the delivery of build-agent asked %d merges again (%v), want both first builds", n, *asked)
|
||||
}
|
||||
built(t, open, idOf(t, *asked, "watchman"), "watchman", commit, "", "mesh-delivery-watchman")
|
||||
built(t, open, idOf(t, *asked, "view"), "view", commit, "", "mesh-delivery-view")
|
||||
for _, module := range []string{"mesh-delivery-watchman", "mesh-delivery-view"} {
|
||||
if _, known := inventoryHolds(t, open.inventory, module); !known {
|
||||
t.Fatalf("1173/AC3: %s is not registered after the first build asked again passed", module)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,141 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/conditions"
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
"github.com/novox/mesh-controller/internal/link"
|
||||
)
|
||||
|
||||
const refusedKeyTTL = `this is not a module manifest: a state is either a name or {name, history, ttl-seconds, per-machine}: ` +
|
||||
`json: unknown field "key-ttl"`
|
||||
|
||||
// aFirstMergeAnswered is a first merge as askFirstBuilds leaves it: answered by the record of merges no plan answers.
|
||||
func aFirstMergeAnswered() (inventory.BuildRequest, inventory.BatchedMerge) {
|
||||
req := inventory.BuildRequest{ID: "build-1", Repository: "novox/mesh-delivery", Seat: "git", Path: "watchman", Ref: "main",
|
||||
Commit: "0177ea7a00", For: "merge"}
|
||||
event, _ := json.Marshal(keptMerge{SourceMoved: link.SourceMoved{Owner: "novox", Repo: "mesh-delivery", Base: "main",
|
||||
Commit: "0177ea7a00", Number: 3, HeadSHA: "a1a1a1a1a1a1"}, FirstBuild: true})
|
||||
return req, inventory.BatchedMerge{Repository: "novox/mesh-delivery", Branch: "main", Commit: "0177ea7a00",
|
||||
Plan: inventory.WaitingRecord, Event: event}
|
||||
}
|
||||
|
||||
// 1173/AC1: a first build refused for a word its manifest uses and the running build-agent does not know holds its merge.
|
||||
func TestAFirstBuildRefusedForAnUnknownWordHoldsItsMerge(t *testing.T) {
|
||||
req, merge := aFirstMergeAnswered()
|
||||
held := judgeRefusedFirstBuild(link.BuildResult{ID: "build-1", Failed: refusedKeyTTL}, req, merge)
|
||||
if len(held) != 1 || held[0].Word != "key-ttl" || held[0].Module != "watchman" || held[0].Path != "watchman" {
|
||||
t.Fatalf("1173/AC1: a first build refused for \"key-ttl\" is judged %+v, want a hold on that word for watchman", held)
|
||||
}
|
||||
}
|
||||
|
||||
// 1173/AC2: a held first merge is named at once by one condition: the merge, the module, the word and what it waits for.
|
||||
func TestAHeldFirstMergeIsNamedByACondition(t *testing.T) {
|
||||
now := time.Date(2026, 10, 22, 14, 0, 0, 0, time.UTC)
|
||||
_, merge := aFirstMergeAnswered()
|
||||
var k keptMerge
|
||||
_ = json.Unmarshal(merge.Event, &k)
|
||||
k.Held = []heldFirstBuild{{Module: "watchman", Path: "watchman", Word: "key-ttl", At: now},
|
||||
{Module: "view", Path: "view", Word: "per-machine", At: now}}
|
||||
merge.Event, _ = json.Marshal(k)
|
||||
merge.Heard = now.Add(-time.Second)
|
||||
got := firstBuildObservations([]inventory.BatchedMerge{merge}, now)
|
||||
if len(got) != 1 {
|
||||
t.Fatalf("1173/AC2: %d conditions for a held first merge, want one: %+v", len(got), got)
|
||||
}
|
||||
o := got[0]
|
||||
if o.Kind != kindFirstBuildHeld || o.Scope != conditions.ScopeMerge {
|
||||
t.Fatalf("1173/AC2: the condition is %+v", o)
|
||||
}
|
||||
for _, want := range []string{"0177ea7a", "watchman", "key-ttl", "view", "per-machine", "build-agent"} {
|
||||
if !strings.Contains(o.Summary, want) {
|
||||
t.Fatalf("1173/AC2: the summary %q does not name %q", o.Summary, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 1173/AC4: a first build that fails for any other reason is not held.
|
||||
func TestAFirstBuildFailingForAnyOtherReasonIsNotHeld(t *testing.T) {
|
||||
req, merge := aFirstMergeAnswered()
|
||||
if judgeRefusedFirstBuild(link.BuildResult{Failed: refusedKeyTTL}, req, merge) == nil {
|
||||
t.Fatal("1173/AC4: the control, a refusal for an unknown word, is not held")
|
||||
}
|
||||
for _, c := range []struct {
|
||||
what string
|
||||
failed string
|
||||
change func(*inventory.BuildRequest, *inventory.BatchedMerge)
|
||||
}{
|
||||
{what: "a compile error", failed: "main.go:3: undefined: foo"},
|
||||
{what: "no failure", failed: ""},
|
||||
{what: "a missing field", failed: `json: cannot unmarshal string into Go struct field`},
|
||||
{what: "a build a person asked", failed: refusedKeyTTL, change: func(r *inventory.BuildRequest, _ *inventory.BatchedMerge) { r.For = "build" }},
|
||||
{what: "a merge a plan answers", failed: refusedKeyTTL, change: func(_ *inventory.BuildRequest, m *inventory.BatchedMerge) { m.Plan = "plan-7" }},
|
||||
{what: "a waiting merge held for nothing", failed: refusedKeyTTL, change: func(_ *inventory.BuildRequest, m *inventory.BatchedMerge) { m.Plan = "" }},
|
||||
} {
|
||||
r, m := req, merge
|
||||
if c.change != nil {
|
||||
c.change(&r, &m)
|
||||
}
|
||||
if held := judgeRefusedFirstBuild(link.BuildResult{Failed: c.failed}, r, m); held != nil {
|
||||
t.Fatalf("1173/AC4: %s is held: %+v", c.what, held)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 1173/AC1, AC2 (review round 1, item 2): a second refusal in a merge already held is held too, its module added to the
|
||||
// hold, so no failed build is recorded and the condition names both.
|
||||
func TestASecondRefusalInAHeldMergeIsAddedToTheHold(t *testing.T) {
|
||||
req, merge := aFirstMergeAnswered()
|
||||
first := judgeRefusedFirstBuild(link.BuildResult{Failed: refusedKeyTTL}, req, merge)
|
||||
k := announcedOf(merge)
|
||||
k.Held = first
|
||||
merge.Event, _ = json.Marshal(k)
|
||||
merge.Plan = "" // HoldMerge put it back to waiting
|
||||
req.ID, req.Path = "build-2", "view"
|
||||
held := judgeRefusedFirstBuild(link.BuildResult{Failed: `json: unknown field "per-machine"`}, req, merge)
|
||||
if len(held) != 2 || held[0].Module != "watchman" || held[1].Module != "view" || held[1].Word != "per-machine" {
|
||||
t.Fatalf("1173/AC1: the second refusal in a held merge is judged %+v, want watchman and view held", held)
|
||||
}
|
||||
// The same module refused again is held once.
|
||||
k.Held = held
|
||||
merge.Event, _ = json.Marshal(k)
|
||||
if again := judgeRefusedFirstBuild(link.BuildResult{Failed: refusedKeyTTL}, req, merge); len(again) != 2 {
|
||||
t.Fatalf("1173/AC1: view refused again is held as %+v, want it once", again)
|
||||
}
|
||||
}
|
||||
|
||||
// 1173/AC3 (review round 1, item 1): a held merge is asked again for its own held modules' directories, not by the rule of
|
||||
// a first merge, which finds nothing once another module of the repository is registered.
|
||||
func TestAHeldMergeIsAskedAgainForItsHeldModulesAlone(t *testing.T) {
|
||||
k := keptMerge{Held: []heldFirstBuild{{Module: "watchman", Path: "watchman"}, {Module: "mesh-delivery", Path: ""},
|
||||
{Module: "watchman", Path: "/watchman/"}}}
|
||||
if got := heldDirsOf(k); strings.Join(got, ",") != "watchman,." {
|
||||
t.Fatalf("1173/AC3: a held merge asks again %v, want watchman and the root once each", got)
|
||||
}
|
||||
}
|
||||
|
||||
// 1173/AC3 (the part that needs no store): only a plan that ended delivered with build-agent built in it is the end a held
|
||||
// first build waits for.
|
||||
func TestOnlyADeliveredPlanOfBuildAgentIsWhatAHeldFirstBuildWaitsFor(t *testing.T) {
|
||||
plan := func(state, walk string, modules map[string]*inventory.PlanModule) inventory.Plan {
|
||||
return inventory.Plan{ID: "p", State: state, WalkSaid: walk, Modules: modules}
|
||||
}
|
||||
built := map[string]*inventory.PlanModule{"build-agent": {State: "built"}}
|
||||
if !deliversBuildAgent(plan(inventory.PlanDone, inventory.WalkEnded, built)) {
|
||||
t.Fatal("1173/AC3: a plan that delivered build-agent is not the end a held first build waits for")
|
||||
}
|
||||
for what, p := range map[string]inventory.Plan{
|
||||
"a plan still running": plan(inventory.PlanBuilding, "", built),
|
||||
"a plan that failed": plan(inventory.PlanFailed, inventory.WalkEnded, built),
|
||||
"a failed build-agent": plan(inventory.PlanDone, inventory.WalkEnded, map[string]*inventory.PlanModule{"build-agent": {State: "failed"}}),
|
||||
"another module's plan": plan(inventory.PlanDone, inventory.WalkEnded, map[string]*inventory.PlanModule{"mailu": {State: "built"}}),
|
||||
} {
|
||||
if deliversBuildAgent(p) {
|
||||
t.Fatalf("1173/AC3: %s is the end a held first build waits for", what)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -408,6 +408,13 @@ func (b builds) Built(ctx context.Context, result link.BuildResult) error {
|
||||
fmt.Printf("%s: a dry run of %s on %s, not taken in\n", result.ID, result.Repository, result.Ref)
|
||||
return nil
|
||||
}
|
||||
// **A first build refused for a word the running build-agent does not read waits for one that does** (novox/hq ADR
|
||||
// 0482 §5, issue 1173): its merge is kept, named, and never failed; no failed build is recorded.
|
||||
if held, err := holdRefusedFirstBuild(ctx, b.inv, result); err != nil {
|
||||
return err
|
||||
} else if held {
|
||||
return nil
|
||||
}
|
||||
manifest, _, err := takeIn(ctx, b.inv, result)
|
||||
// The assignments pending on this build, made or ended (novox/hq issue 325), said in the daemon's log
|
||||
// after what became of the build.
|
||||
|
||||
@@ -0,0 +1,40 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/conditions"
|
||||
)
|
||||
|
||||
// novox/hq ADR 0505 §5: a delivery whose cure is its author's new head reaches the operator in words that name its
|
||||
// author and say a new head is needed, offer no Stop and never call it the operator's; the line as mesh-delivery-gate
|
||||
// says it (ready.go readyLines: needs "new-head", author, branch).
|
||||
func TestADeliveryThatNeedsANewHeadNamesItsAuthorAndOffersNoStop(t *testing.T) {
|
||||
l := stalledLine{ID: "novox/mesh-controller@499ff5c21861", Number: 432, State: "rejected", For: "2h1m0s", Bound: "2h0m0s",
|
||||
H2: "none: a new head from its author, or a recheck that passes", Needs: needsNewHead,
|
||||
Author: "the author of feat/quiet-deliveries (quiet-deliveries on shanks)", Branch: "feat/quiet-deliveries",
|
||||
Says: "novox/mesh-controller#432 passed its checks at 499ff5c21861 and is not merged: a conflict"}
|
||||
o := stalledObservations([]stalledLine{l})[0]
|
||||
for _, words := range []string{o.Headline, o.Explanation, o.Resolved, o.Needs} {
|
||||
if why, ok := conditions.Plain(words); !ok {
|
||||
t.Errorf("%q is not plain: %s", words, why)
|
||||
}
|
||||
for _, never := range []string{"stop it", "operator's"} {
|
||||
if strings.Contains(words, never) {
|
||||
t.Errorf("%q says %q", words, never)
|
||||
}
|
||||
}
|
||||
}
|
||||
if o.Headline != "Pull request 432 of mesh-controller needs a new head from its author" ||
|
||||
!strings.Contains(o.Explanation, "until the author of feat/quiet-deliveries (quiet-deliveries on shanks) pushes a new head") ||
|
||||
!strings.Contains(o.Explanation, "for 2 hours") || len(o.Actions) != 0 ||
|
||||
o.Needs != "ask the author of feat/quiet-deliveries (quiet-deliveries on shanks) for a new head." {
|
||||
t.Fatalf("words: %q / %q / %q / %v", o.Headline, o.Explanation, o.Needs, o.Actions)
|
||||
}
|
||||
// A branch in code is not said; the author is then "its author".
|
||||
l.Author = "the author of fix/a1b2c3d4e5f6"
|
||||
if o := stalledObservations([]stalledLine{l})[0]; !strings.Contains(o.Needs, "ask its author") {
|
||||
t.Fatalf("a dotted branch was said: %q", o.Needs)
|
||||
}
|
||||
}
|
||||
@@ -966,6 +966,9 @@ func stalledWords(l stalledLine, o conditions.Observation) (headline, explanatio
|
||||
if l.State == "unanswered" {
|
||||
return unansweredWords(l, o)
|
||||
}
|
||||
if l.Needs == needsNewHead {
|
||||
return newHeadWords(l, name)
|
||||
}
|
||||
held := l.State
|
||||
if held == "" {
|
||||
held = "held"
|
||||
@@ -999,6 +1002,29 @@ func stalledWords(l stalledLine, o conditions.Observation) (headline, explanatio
|
||||
fmt.Sprintf("Delivery of %s is no longer %s", name, held), needs, actions
|
||||
}
|
||||
|
||||
// newHeadWords are the words of a delivery whose cure is its author's new head (novox/hq ADR 0505 §5): it names the
|
||||
// author and says a new head is needed; it offers no Stop and never calls it the operator's, because the operator
|
||||
// hears of it only once the author's own watch has failed, after two hours with no new head.
|
||||
func newHeadWords(l stalledLine, name string) (headline, explanation, resolved, needs string, actions []conditions.Action) {
|
||||
pull := "A pull request of " + name
|
||||
if l.Number > 0 {
|
||||
pull = fmt.Sprintf("Pull request %d of %s", l.Number, name)
|
||||
}
|
||||
author := l.Author
|
||||
if _, plain := conditions.Plain(author); !plain {
|
||||
author = "its author" // a branch named in code (a dotted name, a number) is not said
|
||||
}
|
||||
long := "for a long time"
|
||||
if d, err := time.ParseDuration(l.For); err == nil {
|
||||
long = "for " + humanDuration(d)
|
||||
}
|
||||
return pull + " needs a new head from its author",
|
||||
fmt.Sprintf("%s cannot merge until %s pushes a new head: its check failed after its Approve, and no new head "+
|
||||
"came %s.", pull, author, long),
|
||||
pull + " has a new head, or is closed",
|
||||
"ask " + author + " for a new head.", nil
|
||||
}
|
||||
|
||||
// stoppedNeeds is what ends a delivery that is already stopped, from the stalled line's healer words ("none: a
|
||||
// merge that delivers its modules ends it — the operator's"): the part that says what ends it.
|
||||
func stoppedNeeds(l stalledLine) string {
|
||||
|
||||
@@ -192,7 +192,7 @@ func startRolloutEvents(ctx context.Context, open *stores, js *broker.JetStream)
|
||||
if err := startProvAppliedLooks(ctx, js); err != nil {
|
||||
fmt.Printf("plans: the provisioner clients' words could not be watched (%v); a walk waits for them until the next look\n", err)
|
||||
}
|
||||
if err := startPlanMovedLooks(ctx, js); err != nil {
|
||||
if err := startPlanMovedLooks(ctx, js, open); err != nil {
|
||||
fmt.Printf("plans: the plans moved by other processes could not be followed (%v)\n", err)
|
||||
}
|
||||
done := startPlanLooks(ctx, open)
|
||||
@@ -209,6 +209,10 @@ func servingPlanSaved(ctx context.Context, bus link.Bus, open *stores) func(*inv
|
||||
keepBuildingKey(ctx, p)
|
||||
if p.WalkSaid == inventory.WalkEnded {
|
||||
go forgetChecksOfWalk(context.WithoutCancel(ctx), open.inventory, p)
|
||||
// A first build refused for a word only a newer build-agent reads is asked again (novox/hq issue 1173).
|
||||
if deliversBuildAgent(p) {
|
||||
go askHeldFirstBuildsAgain(context.WithoutCancel(ctx), open, p)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -563,7 +567,7 @@ func startProvAppliedLooks(ctx context.Context, js *broker.JetStream) error {
|
||||
// `plan-moved`, so a plan a person moved with a verb at the control node's terminal — a process this one did not spawn,
|
||||
// whose saves it does not hear — is looked at all the same (novox/hq ADR 0406 §2, §5, issue 974). New messages only; the
|
||||
// first look reads the rest. A look its own save asks is the same look: asks are one.
|
||||
func startPlanMovedLooks(ctx context.Context, js *broker.JetStream) error {
|
||||
func startPlanMovedLooks(ctx context.Context, js *broker.JetStream, open *stores) error {
|
||||
api, err := jetstream.New(js.Conn())
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -578,7 +582,18 @@ func startPlanMovedLooks(ctx context.Context, js *broker.JetStream) error {
|
||||
if err != nil {
|
||||
return fmt.Errorf("following the plans moved: %w", err)
|
||||
}
|
||||
consuming, err := consumer.Consume(func(jetstream.Msg) { askToLook() })
|
||||
var bus link.Bus
|
||||
if core, err := js.Conn().JetStream(); err == nil {
|
||||
bus = link.OverNATS{Conn: js.Conn(), JS: core}
|
||||
}
|
||||
consuming, err := consumer.Consume(func(m jetstream.Msg) {
|
||||
askToLook()
|
||||
if bus != nil && open != nil {
|
||||
// A walk's start or end, said by this controller or a verb: said again with the machines whose
|
||||
// declaration it changes (novox/hq ADR 0505 §2, walk_declared.go).
|
||||
go sayDeclared(context.WithoutCancel(ctx), bus, open, m.Data())
|
||||
}
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("following the plans moved: %w", err)
|
||||
}
|
||||
|
||||
@@ -174,7 +174,7 @@ func TestAPlanMovedByAnotherProcessAsksForALook(t *testing.T) {
|
||||
ctx, stop := context.WithCancel(context.Background())
|
||||
defer stop()
|
||||
go l.run(ctx)
|
||||
if err := startPlanMovedLooks(ctx, js); err != nil {
|
||||
if err := startPlanMovedLooks(ctx, js, nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
legacy, err := js.Conn().JetStream()
|
||||
|
||||
@@ -0,0 +1,163 @@
|
||||
package main
|
||||
|
||||
// The machines whose declaration a walk's merges change (novox/hq ADR 0505 §2): beside what the walk sends, a merge
|
||||
// can change what another machine is to run — an assignment, a provision or a requirement added or removed — and a
|
||||
// problem raised there as the walk runs, or as it settles, is the walk's as much as one on a machine it sends to. The
|
||||
// router holds what a walk touches (ADR 0308 §1); these are touches of their own, marked `declared`.
|
||||
//
|
||||
// **Compared as the mesh compares whether a machine is up to date.** Each machine's declaration is composed as a push
|
||||
// would compose it, with the number, epoch and generation it was last sent, and its digest held against the one it was
|
||||
// last sent (wouldSend, inventory.Waiting): byte for byte what it was sent when nothing changed, so a secret, a sequence
|
||||
// or a lease's epoch never makes a machine differ by itself. A machine whose declaration cannot be worked out now and
|
||||
// was sent one before is taken as changed: not known is changed, never "unchanged" (a merge that leaves a machine
|
||||
// unable to compose is the case this exists for, the 11:03 message of ADR 0505's context).
|
||||
//
|
||||
// **Said again, not waited for.** Composing every machine takes seconds; the walk's start and end are said at once as
|
||||
// before, and the serving controller, which hears every `plan-moved` — its own and a verb's — says a start or an end
|
||||
// again with the declared touches added, marked `declared_read`, which it never reads twice. The router merges a walk
|
||||
// said again (messenger walks.go).
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"sort"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
"github.com/novox/mesh-controller/internal/link"
|
||||
)
|
||||
|
||||
// walkDeclaredRead marks a walk's start or end said again with its declared touches: read once.
|
||||
const walkDeclaredRead = "declared_read"
|
||||
|
||||
// declaredWithin bounds the comparison of every machine for one walk's start or end.
|
||||
const declaredWithin = time.Minute
|
||||
|
||||
// changedDeclarations is every machine whose declaration differs from what it was last sent, and every machine sent
|
||||
// one before whose declaration cannot be worked out now. A seam a test replaces.
|
||||
var changedDeclarations = func(ctx context.Context, open *stores) ([]string, error) {
|
||||
nodes, err := open.inventory.Nodes(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctx = withInputCache(ctx)
|
||||
would, err := wouldSend(ctx, open, nodes)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
behind, err := open.inventory.Waiting(ctx, would)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
sent := map[string]bool{}
|
||||
for _, n := range nodes {
|
||||
if s, known, err := open.inventory.SentTo(ctx, n.Name); err == nil && known && s.Digest != "" {
|
||||
sent[n.Name] = true
|
||||
}
|
||||
}
|
||||
return changedOf(nodes, would, behind, sent), nil
|
||||
}
|
||||
|
||||
// changedOf is changedDeclarations over what was read: the machines behind, and those sent one before that could not
|
||||
// be worked out. Pure; sorted.
|
||||
func changedOf(nodes []inventory.Node, would map[string]string, behind []inventory.Machine, sent map[string]bool) []string {
|
||||
set := map[string]bool{}
|
||||
for _, m := range behind {
|
||||
set[m.Node] = true
|
||||
}
|
||||
for _, n := range nodes {
|
||||
if _, worked := would[n.Name]; !worked && sent[n.Name] {
|
||||
set[n.Name] = true
|
||||
}
|
||||
}
|
||||
out := make([]string, 0, len(set))
|
||||
for n := range set {
|
||||
out = append(out, n)
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
|
||||
// withDeclared is a walk's start or end said again with the machines whose declaration changed added as declared
|
||||
// touches: false when the body says no start or end, was said again already, or names no machine it does not touch.
|
||||
func withDeclared(body []byte, changed []string) ([]byte, bool, error) {
|
||||
said, wanted, err := toSayAgain(body)
|
||||
if err != nil || !wanted {
|
||||
return nil, false, err
|
||||
}
|
||||
var touches []walkTouch
|
||||
if raw := said["touches"]; raw != nil && string(raw) != "null" {
|
||||
if err := json.Unmarshal(raw, &touches); err != nil {
|
||||
return nil, false, fmt.Errorf("its touches are not a list of {machine, modules}: %w", err)
|
||||
}
|
||||
}
|
||||
added := declaredTouches(touches, changed)
|
||||
if len(added) == 0 {
|
||||
return nil, false, nil
|
||||
}
|
||||
touches = append(touches, added...)
|
||||
sort.Slice(touches, func(i, j int) bool { return touches[i].Machine < touches[j].Machine })
|
||||
if said["touches"], err = json.Marshal(touches); err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
said[walkDeclaredRead] = json.RawMessage(`true`)
|
||||
out, err := json.Marshal(said)
|
||||
return out, err == nil, err
|
||||
}
|
||||
|
||||
// toSayAgain reads a plan-moved: whether it says a walk's start or end not said again yet.
|
||||
func toSayAgain(body []byte) (map[string]json.RawMessage, bool, error) {
|
||||
var said map[string]json.RawMessage
|
||||
if err := json.Unmarshal(body, &said); err != nil {
|
||||
return nil, false, fmt.Errorf("a plan-moved that is not an object: %w", err)
|
||||
}
|
||||
var phase string
|
||||
_ = json.Unmarshal(said["phase"], &phase)
|
||||
return said, (phase == inventory.WalkStarted || phase == inventory.WalkEnded) && said[walkDeclaredRead] == nil, nil
|
||||
}
|
||||
|
||||
// declaredTouches are the machines changed that a walk's touches do not name, each a declared touch. Pure; sorted.
|
||||
func declaredTouches(touches []walkTouch, changed []string) []walkTouch {
|
||||
named := map[string]bool{}
|
||||
for _, t := range touches {
|
||||
named[t.Machine] = true
|
||||
}
|
||||
var out []walkTouch
|
||||
for _, n := range changed {
|
||||
if !named[n] {
|
||||
out = append(out, walkTouch{Machine: n, Modules: []string{}, Declared: true})
|
||||
named[n] = true
|
||||
}
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool { return out[i].Machine < out[j].Machine })
|
||||
return out
|
||||
}
|
||||
|
||||
// sayDeclared says a walk's start or end again with its declared touches, when it has any. What cannot be read is
|
||||
// said on the controller's log: the walk is then held as before, never less.
|
||||
func sayDeclared(ctx context.Context, bus link.Bus, open *stores, body []byte) {
|
||||
if _, wanted, _ := toSayAgain(body); !wanted {
|
||||
return // no start or end, or said again already: nothing to compare
|
||||
}
|
||||
comparing, cancel := context.WithTimeout(ctx, declaredWithin)
|
||||
defer cancel()
|
||||
changed, err := changedDeclarations(comparing, open)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "plan-moved: which machines' declarations the walk changes cannot be read: %v\n", err)
|
||||
return
|
||||
}
|
||||
again, ok, err := withDeclared(body, changed)
|
||||
if err != nil || !ok {
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "plan-moved: the walk cannot be said again with its declared machines: %v\n", err)
|
||||
}
|
||||
return
|
||||
}
|
||||
stating, stop := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
|
||||
defer stop()
|
||||
if err := bus.PublishSeatEvent(stating, link.MeshControllerSeat, link.KeyPlanMoved, again); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "plan-moved: the walk could not be said again with its declared machines: %v\n", err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,61 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"reflect"
|
||||
"testing"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
)
|
||||
|
||||
// novox/hq ADR 0505 §2: the machines whose declaration a walk's merges change are said as declared touches.
|
||||
|
||||
// The machines behind what they were last sent, and those sent one before that cannot be worked out now, are changed;
|
||||
// one never sent and not worked out is not.
|
||||
func TestAMachineBehindOrNoLongerComposableIsChanged(t *testing.T) {
|
||||
nodes := []inventory.Node{{Name: "ace"}, {Name: "g14"}, {Name: "novox"}, {Name: "shanks"}, {Name: "new"}}
|
||||
would := map[string]string{"ace": "d1", "g14": "d2", "novox": "d3"}
|
||||
behind := []inventory.Machine{{Node: "ace"}}
|
||||
sent := map[string]bool{"ace": true, "g14": true, "novox": true, "shanks": true}
|
||||
if got := changedOf(nodes, would, behind, sent); !reflect.DeepEqual(got, []string{"ace", "shanks"}) {
|
||||
t.Fatalf("changed: %v, want ace (behind) and shanks (no longer composes)", got)
|
||||
}
|
||||
}
|
||||
|
||||
// A walk's start said by plan-moved (walk_said.go's body), said again with the changed machines it does not touch as
|
||||
// declared touches, once.
|
||||
func TestAWalksStartIsSaidAgainWithTheMachinesItsMergesChange(t *testing.T) {
|
||||
body, err := json.Marshal(planMovedBody{Plan: inventory.Plan{ID: "plan-1", State: inventory.PlanBuilding},
|
||||
walkSaid: &walkSaid{Walk: "plan-1", Phase: inventory.WalkStarted, BudgetClass: "leaf", BoundSeconds: 600,
|
||||
Touches: []walkTouch{{Machine: "novox", Modules: []string{"gitea"}}}}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
again, ok, err := withDeclared(body, []string{"novox", "shanks"})
|
||||
if err != nil || !ok {
|
||||
t.Fatalf("not said again: %v %v", ok, err)
|
||||
}
|
||||
var said struct {
|
||||
ID string `json:"id"`
|
||||
Phase string `json:"phase"`
|
||||
Touches []walkTouch `json:"touches"`
|
||||
DeclaredRead bool `json:"declared_read"`
|
||||
}
|
||||
if err := json.Unmarshal(again, &said); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
want := []walkTouch{{Machine: "novox", Modules: []string{"gitea"}}, {Machine: "shanks", Modules: []string{}, Declared: true}}
|
||||
if said.ID != "plan-1" || said.Phase != inventory.WalkStarted || !said.DeclaredRead || !reflect.DeepEqual(said.Touches, want) {
|
||||
t.Fatalf("said again: %+v", said)
|
||||
}
|
||||
if _, ok, _ := withDeclared(again, []string{"ace"}); ok {
|
||||
t.Fatalf("a start said again was said a third time")
|
||||
}
|
||||
if _, ok, _ := withDeclared(body, []string{"novox"}); ok {
|
||||
t.Fatalf("said again with nothing it does not touch")
|
||||
}
|
||||
moved, _ := json.Marshal(inventory.Plan{ID: "plan-2", State: inventory.PlanBuilding})
|
||||
if _, ok, _ := withDeclared(moved, []string{"shanks"}); ok {
|
||||
t.Fatalf("a move inside a walk was said as its start")
|
||||
}
|
||||
}
|
||||
@@ -59,6 +59,9 @@ type walkTouch struct {
|
||||
// Carried names those of Modules that are there only because a moved tool runner or node-engine carries
|
||||
// them: what restarts with it, not what the walk builds.
|
||||
Carried []string `json:"carried,omitempty"`
|
||||
// Declared marks a machine the walk sends nothing to whose declaration its merges change (novox/hq ADR 0505 §2):
|
||||
// said again by the serving controller once it has compared every machine (walk_declared.go).
|
||||
Declared bool `json:"declared,omitempty"`
|
||||
}
|
||||
|
||||
// planMovedBody is `plan-moved`'s body: the walk whole, as before, and on its start and end what it touches.
|
||||
|
||||
@@ -264,3 +264,46 @@ func TestAWalkStoppedByHandAndRetriedEndsFailedWhenItFailsAgain(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// novox/hq ADR 0505 §2: a walk's start is said again, over the real bus and against the real store, with a machine it
|
||||
// sends nothing to whose declaration differs from what it was last sent (here: assigned a module and never pushed),
|
||||
// marked declared.
|
||||
func TestAWalksStartIsSaidAgainWithAMachineWhoseDeclarationChanged(t *testing.T) {
|
||||
open := aMesh(t)
|
||||
ctx := t.Context()
|
||||
next := planMovedHeard(t)
|
||||
register(t, open, catalogue.Manifest{Module: "app", Version: "1"})
|
||||
if _, err := assign(ctx, open, "anchor", "app"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
js, err := broker.Dial(testbus.URL(t))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(js.Close)
|
||||
core, err := js.Conn().JetStream()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
body, err := json.Marshal(planMovedBody{Plan: inventory.Plan{ID: "plan-505", State: inventory.PlanBuilding},
|
||||
walkSaid: &walkSaid{Walk: "plan-505", Phase: inventory.WalkStarted, BudgetClass: inventory.ClassLeaf,
|
||||
BoundSeconds: 360, Touches: []walkTouch{}}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
sayDeclared(ctx, link.OverNATS{Conn: js.Conn(), JS: core}, open, body)
|
||||
again := next(inventory.WalkStarted)
|
||||
raw, _ := json.Marshal(again["touches"])
|
||||
var touches []walkTouch
|
||||
_ = json.Unmarshal(raw, &touches)
|
||||
anchor := false
|
||||
for _, tc := range touches {
|
||||
anchor = anchor || (tc.Machine == "anchor" && tc.Declared && len(tc.Modules) == 0)
|
||||
if !tc.Declared {
|
||||
t.Fatalf("a machine the walk sends nothing to was said sent: %s", raw)
|
||||
}
|
||||
}
|
||||
if again["walk"] != "plan-505" || again[walkDeclaredRead] != true || !anchor {
|
||||
t.Fatalf("said again: %v", again)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -114,6 +114,16 @@ func (i *Inventory) AnswerMerge(ctx context.Context, repository, commit, plan st
|
||||
return i.chainSavePlans(ctx, c, nil, []MergeMove{{Repository: repository, Commit: commit, Plan: plan, Alone: alone}})
|
||||
}
|
||||
|
||||
// HoldMerge puts a merge back to waiting, no plan answering it, with the event it is kept under replaced (novox/hq issue 1173:
|
||||
// a first merge held for a build-agent that reads its manifest, the hold said in its event).
|
||||
func (i *Inventory) HoldMerge(ctx context.Context, repository, commit string, event json.RawMessage) error {
|
||||
c, err := i.theChain(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return i.chainSavePlans(ctx, c, nil, []MergeMove{{Repository: repository, Commit: commit, Event: event}})
|
||||
}
|
||||
|
||||
// Batches is every batch not yet cut: assembling or queued, oldest first. One at a time is the rule; more
|
||||
// are read so a fault is seen.
|
||||
func (i *Inventory) Batches(ctx context.Context) ([]Plan, error) {
|
||||
|
||||
@@ -634,6 +634,10 @@ func (e *recordEdit) apply(held []chain.HeldMerge) []chain.HeldMerge {
|
||||
}
|
||||
if f, has := e.flag[k]; has {
|
||||
h.Waiting, h.Alone = f.Waiting, f.Alone
|
||||
// A merge put back to waiting may carry a new announcement (HoldMerge); every other flag copies its own.
|
||||
if len(f.Event) > 0 {
|
||||
h.Event = f.Event
|
||||
}
|
||||
}
|
||||
out = append(out, h)
|
||||
}
|
||||
@@ -758,6 +762,9 @@ func (i *Inventory) chainSavePlans(ctx context.Context, c *chain.Chain, plans []
|
||||
if m.Plan == "" {
|
||||
f := w.m
|
||||
f.Waiting, f.Alone = true, m.Alone
|
||||
if m.Event != nil {
|
||||
f.Event = m.Event
|
||||
}
|
||||
editOf(w.plan).flag[key] = f
|
||||
continue
|
||||
}
|
||||
|
||||
@@ -2,6 +2,7 @@ package inventory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"slices"
|
||||
@@ -677,6 +678,8 @@ func (i *Inventory) SavePlan(ctx context.Context, p *Plan) error {
|
||||
type MergeMove struct {
|
||||
Repository, Commit, Plan string
|
||||
Alone bool
|
||||
// Event replaces the merge's kept announcement when it is put back to waiting (Plan empty); nil leaves it.
|
||||
Event json.RawMessage
|
||||
}
|
||||
|
||||
// SavePlansMoving writes plans, as SavePlan does each, and gives merges to the plans that answer them, all in one
|
||||
|
||||
Reference in New Issue
Block a user