From 8d562f9b2287a41c7a7302837fe83149a2a45566 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 22 Oct 2026 14:23:54 +0200 Subject: [PATCH 1/2] =?UTF-8?q?Say=20a=20walk's=20start=20and=20end=20agai?= =?UTF-8?q?n=20with=20the=20machines=20whose=20declaration=20its=20merges?= =?UTF-8?q?=20change=20(ADR=200505=20=C2=A72)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/mesh-controller/plan_looks.go | 17 +- .../plan_looks_integration_test.go | 2 +- cmd/mesh-controller/walk_declared.go | 163 ++++++++++++++++++ cmd/mesh-controller/walk_declared_test.go | 61 +++++++ cmd/mesh-controller/walk_said.go | 3 + .../walk_said_integration_test.go | 43 +++++ 6 files changed, 285 insertions(+), 4 deletions(-) create mode 100644 cmd/mesh-controller/walk_declared.go create mode 100644 cmd/mesh-controller/walk_declared_test.go diff --git a/cmd/mesh-controller/plan_looks.go b/cmd/mesh-controller/plan_looks.go index 29297b24..855d6e02 100644 --- a/cmd/mesh-controller/plan_looks.go +++ b/cmd/mesh-controller/plan_looks.go @@ -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) @@ -563,7 +563,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 +578,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) } diff --git a/cmd/mesh-controller/plan_looks_integration_test.go b/cmd/mesh-controller/plan_looks_integration_test.go index df7d3c9f..cf2f75fb 100644 --- a/cmd/mesh-controller/plan_looks_integration_test.go +++ b/cmd/mesh-controller/plan_looks_integration_test.go @@ -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() diff --git a/cmd/mesh-controller/walk_declared.go b/cmd/mesh-controller/walk_declared.go new file mode 100644 index 00000000..528ebc47 --- /dev/null +++ b/cmd/mesh-controller/walk_declared.go @@ -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) + } +} diff --git a/cmd/mesh-controller/walk_declared_test.go b/cmd/mesh-controller/walk_declared_test.go new file mode 100644 index 00000000..2dd47889 --- /dev/null +++ b/cmd/mesh-controller/walk_declared_test.go @@ -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") + } +} diff --git a/cmd/mesh-controller/walk_said.go b/cmd/mesh-controller/walk_said.go index 17b7984c..d1a8d3b5 100644 --- a/cmd/mesh-controller/walk_said.go +++ b/cmd/mesh-controller/walk_said.go @@ -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. diff --git a/cmd/mesh-controller/walk_said_integration_test.go b/cmd/mesh-controller/walk_said_integration_test.go index 633440e0..37190421 100644 --- a/cmd/mesh-controller/walk_said_integration_test.go +++ b/cmd/mesh-controller/walk_said_integration_test.go @@ -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) + } +} From b7d03b68974dd6ce9c5a3302558d2f4e37eea5fd Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 22 Oct 2026 14:25:26 +0200 Subject: [PATCH 2/2] =?UTF-8?q?Word=20a=20delivery=20that=20needs=20its=20?= =?UTF-8?q?author's=20new=20head=20as=20its=20author's,=20with=20no=20Stop?= =?UTF-8?q?=20(ADR=200505=20=C2=A75)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/mesh-controller/delivery_conditions.go | 9 +++++ cmd/mesh-controller/new_head_words_test.go | 40 ++++++++++++++++++++++ cmd/mesh-controller/plain_words.go | 26 ++++++++++++++ 3 files changed, 75 insertions(+) create mode 100644 cmd/mesh-controller/new_head_words_test.go diff --git a/cmd/mesh-controller/delivery_conditions.go b/cmd/mesh-controller/delivery_conditions.go index 74b14082..d0a13dc5 100644 --- a/cmd/mesh-controller/delivery_conditions.go +++ b/cmd/mesh-controller/delivery_conditions.go @@ -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) { diff --git a/cmd/mesh-controller/new_head_words_test.go b/cmd/mesh-controller/new_head_words_test.go new file mode 100644 index 00000000..86c8a959 --- /dev/null +++ b/cmd/mesh-controller/new_head_words_test.go @@ -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) + } +} diff --git a/cmd/mesh-controller/plain_words.go b/cmd/mesh-controller/plain_words.go index fce97830..1a954534 100644 --- a/cmd/mesh-controller/plain_words.go +++ b/cmd/mesh-controller/plain_words.go @@ -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 {