novox/mesh-catalog
18
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

The messenger holds and tells what the delivery view says is in motion, not the controller's plans (issue 1946)

Browse Source
This commit is contained in:
jschoubben
2026-10-24 18:25:38 +02:00
parent 9fa6710296
commit a17d0c8e93
6 changed files with 175 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/messenger/cmd/messenger/adr505_bus_test.go
+19 -18
View File
@@ -25,17 +25,17 @@ func (b *conditionBed) controllerEvent(t *testing.T, event string, body any) {
}
}
// walkEvent is a plan-moved body as the controller says a walk's start or end (mesh-controller walk_said.go), with a
// machine it sends nothing to and whose declaration its merges change (`declared`, ADR 0505 §2b).
func walkEvent(phase, outcome string, started, at time.Time) map[string]any {
b := map[string]any{"id": "plan-bus", "state": "walking", "walk": "plan-bus", "phase": phase, "budget_class": "leaf",
"bound_seconds": 600, "started_at": started.UTC().Format(time.RFC3339Nano),
"touches": []map[string]any{{"machine": "novox", "modules": []string{"gitea"}},
{"machine": "shanks", "modules": []string{"mesh-build-image"}, "declared": true}}}
if phase == WalkEnded {
b["outcome"], b["ended_at"] = outcome, at.UTC().Format(time.RFC3339Nano)
// viewEvent hands the bundle one of the delivery view's events (moving, settled, stuck), as the runtime does.
func (b *conditionBed) viewEvent(t *testing.T, event string, body any) {
t.Helper()
raw, err := json.Marshal(body)
if err != nil {
t.Fatal(err)
}
if _, err := b.r.request("mesh/event", map[string]any{"envelope": map[string]any{"key": ViewModule + "." + event,
"node": "test", "body": json.RawMessage(raw)}}); err != nil {
t.Fatal(err)
}
return b
}
// conditionEvent is a condition as the controller raises it (mesh-controller probes.go D1, signals.go).
@@ -60,27 +60,28 @@ func (b *conditionBed) toldWith(words string) *string {
// ADR 0505 §1, §2b, §3: during a walk, a condition on a machine whose declaration it changes is held, and one on an
// untouched machine reaches the desk at once ending "No delivery running."; after the walk delivered, the held one
// reaches the desk only once the settle (two seconds here) has passed, inside the walk's end message.
func TestAWalksDeclaredMachineIsHeldUntilItSettledOverTheBus(t *testing.T) {
func TestAMovingModulesMachineIsHeldUntilItSettledOverTheBus(t *testing.T) {
b := newConditionBedWith(t, `{"settle-seconds": 2}`)
started := time.Now()
b.controllerEvent(t, EventPlanMoved, walkEvent(WalkStarted, "", started, started))
b.controllerEvent(t, EventRaised, conditionEvent("machine.shanks.uncomposable", "machine", "shanks", "shanks",
"Shanks cannot get new instructions"))
b.viewEvent(t, EventMoving, map[string]any{"node": "novox", "module": "gitea", "from_fp": "a", "to_fp": "b",
"since": started.UTC().Format(time.RFC3339Nano), "bound_seconds": 600})
b.controllerEvent(t, EventRaised, conditionEvent("machine.novox.uncomposable", "machine", "novox", "novox",
"Novox cannot get new instructions"))
b.controllerEvent(t, EventRaised, conditionEvent("machine.ace.silent", "machine", "ace", "ace", "Ace is silent"))
b.r.until("the untouched machine's condition reaches the desk with its delivery line", 15*time.Second, func() bool {
m := b.toldWith("Ace is silent")
return m != nil && strings.HasSuffix(*m, "\nNo delivery running.")
})
if b.toldWith("Shanks cannot get new instructions") != nil {
if b.toldWith("Novox cannot get new instructions") != nil {
t.Fatalf("the declared machine's condition was told during the walk")
}
b.controllerEvent(t, EventPlanMoved, walkEvent(WalkEnded, WalkDelivered, started, time.Now()))
b.viewEvent(t, EventSettled, map[string]any{"node": "novox", "module": "gitea", "at": time.Now().UTC().Format(time.RFC3339Nano)})
time.Sleep(time.Second)
if b.toldWith("Shanks cannot get new instructions") != nil || b.toldWith("Delivered") != nil {
if b.toldWith("Novox cannot get new instructions") != nil || b.toldWith("Delivered") != nil {
t.Fatalf("told before the settle passed: %+v", b.told())
}
b.r.until("the walk's end tells what it held once it settled", 30*time.Second, func() bool {
m := b.toldWith("Still open after the delivery of gitea to novox:")
return m != nil && strings.Contains(*m, "Shanks cannot get new instructions") && !strings.Contains(*m, "shanks,")
return m != nil && strings.Contains(*m, "Novox cannot get new instructions")
})
}
modules/messenger/cmd/messenger/main.go
+15 -3
View File
@@ -434,9 +434,9 @@ func run(h *Holder, l *listening) {
}
}()
handle := func(e stdio.Envelope) error {
if e.Key == ControllerSeat+"."+EventPlanMoved || e.Key == ControllerSeat+"."+EventWalk {
// A walk's start or end (novox/hq ADR 0308 §1); any other move of a plan is not read here.
ev, isWalk, err := DecodeWalk(e.Body)
if event, ok := strings.CutPrefix(e.Key, ViewModule+"."); ok {
// A module in motion on a node, settled or stuck: a walk's start or end now that plans are gone (ADR 0308 §1, ADR 0539 §17).
ev, isWalk, err := DecodeMotion(event, e.Body)
if err != nil {
h.Unreadable(e.Key, err)
return nil
@@ -462,6 +462,18 @@ func run(h *Holder, l *listening) {
go takeAsks(h)
go answerProofs(h)
go followKinds(h)
// The delivery view's events say what is in motion (ADR 0539 §17), heard beside the controller's condition events.
go func() {
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
if err := stdio.Subscribe(ViewModule+".*", handle); err == nil {
logf("[messenger] listening for the delivery view's moving, settled and stuck")
return
} else {
logf("[messenger] not hearing the delivery view yet (%v); asking again in %s", err, wait)
}
time.Sleep(wait)
}
}()
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
err := stdio.Subscribe(ControllerSeat+".*", handle)
if err == nil {
modules/messenger/cmd/messenger/manifest_test.go
+1 -1
View File
@@ -81,7 +81,7 @@ func TestItDeclaresTheSeatsOfAnAskAndHoldsTheOperatorChannel(t *testing.T) { //
}
if !reflect.DeepEqual(m.Consumes, []string{
ControllerSeat + "." + EventRaised, ControllerSeat + "." + EventChanged, ControllerSeat + "." + EventCleared,
ControllerSeat + "." + EventPlanMoved, "intake.choice", "intake.link", "intake.failed", "intake.standing"}) {
ViewModule + "." + EventMoving, ViewModule + "." + EventSettled, ViewModule + "." + EventStuck, "intake.choice", "intake.link", "intake.failed", "intake.standing"}) {
t.Fatalf("consumes %v", m.Consumes)
}
// It performs nothing: the verbs it calls read the open conditions, whether root is free, and which issues
modules/messenger/cmd/messenger/motion_test.go
+76
View File
@@ -0,0 +1,76 @@
package main
import (
"testing"
"time"
)
// R6 (ADR 0539 §17): plans and their plan-moved are gone, so a walk is a module in motion on a node, as the delivery view says
// it: `moving`, then `settled` or `stuck`. The holding, the bound, the settle and the end message are the walk's, unchanged. The
// subjects and bodies are the view's as the lead relayed them (node, module, from_fp, to_fp, composed, since, bound_seconds).
// motion says a view event to the router, through the decoder the bus's events go through.
func motion(t *testing.T, h *Holder, event, body string) {
t.Helper()
ev, ok, err := DecodeMotion(event, []byte(body))
if err != nil || !ok {
t.Fatalf("the %s event was not read: %v %v", event, ok, err)
}
h.Walked(ev)
}
func TestTheViewsEventsAreReadAsAWalkOfOneModuleOnOneNode(t *testing.T) {
since := time.Date(2026, 10, 24, 12, 0, 0, 0, time.UTC)
start, ok, err := DecodeMotion(EventMoving, []byte(`{"node":"ace","module":"node-tools","from_fp":"a","to_fp":"b","composed":"c",`+
`"since":"2026-10-24T12:00:00Z","bound_seconds":600}`))
if err != nil || !ok || start.Walk != "ace/node-tools" || start.Phase != WalkStarted || start.BoundSeconds != 600 ||
!start.StartedAt.Equal(since) || len(start.Touches) != 1 || start.Touches[0].Machine != "ace" || start.Touches[0].Modules[0] != "node-tools" {
t.Fatalf("moving read as %+v %v %v", start, ok, err)
}
end, ok, err := DecodeMotion(EventSettled, []byte(`{"node":"ace","module":"node-tools","at":"2026-10-24T12:05:00Z"}`))
if err != nil || !ok || end.Walk != "ace/node-tools" || end.Phase != WalkEnded || end.Outcome != WalkDelivered {
t.Fatalf("settled read as %+v %v %v", end, ok, err)
}
stuck, _, _ := DecodeMotion(EventStuck, []byte(`{"node":"ace","module":"node-tools","outcome":"stopped","words":"held for a backup"}`))
failed, _, _ := DecodeMotion(EventStuck, []byte(`{"node":"ace","module":"node-tools","words":"unhealthy"}`))
if stuck.Outcome != WalkStopped || stuck.Words != "held for a backup" || failed.Outcome != WalkFailed {
t.Fatalf("stuck read as %+v and %+v", stuck, failed)
}
if _, ok, _ := DecodeMotion("other", []byte(`{}`)); ok {
t.Fatal("an event that is no motion was read as one")
}
if _, _, err := DecodeMotion(EventMoving, []byte(`{"module":"x"}`)); err == nil {
t.Fatal("a motion that names no node was read")
}
}
func TestAModuleInMotionHoldsWhatItTouchesAndIsToldAtSettle(t *testing.T) {
h, tg, dt, c, _ := newHolder(t)
motion(t, h, EventMoving, `{"node":"novox","module":"node-tools","from_fp":"a","to_fp":"b","since":"`+c.now().Format(time.RFC3339Nano)+`","bound_seconds":600}`)
k := cond("module.node-tools.novox.unhealthy", Urgent, "the tool runner on novox does not answer")
k.Machine, k.Raised = "novox", c.now()
h.Condition(EventRaised, k)
runFor(h, c, time.Minute)
if saidOn(tg, dt) != 0 || !has(outcomes(h, k.Key), "held") {
t.Fatalf("a module in motion did not hold its condition: %v", outcomes(h, k.Key))
}
motion(t, h, EventSettled, `{"node":"novox","module":"node-tools","at":"`+c.now().Format(time.RFC3339Nano)+`"}`)
runFor(h, c, DefaultSettle+time.Minute)
if !heard("Delivered: node-tools on novox", tg, dt) || !heard("Still open after the delivery of node-tools to novox", tg, dt) {
t.Fatalf("the settle did not tell the end and what is still open: %+v %+v", tg.sends, dt.sends)
}
}
func TestAStuckModuleIsToldAtOnceWithTheViewsWords(t *testing.T) {
h, tg, dt, c, _ := newHolder(t)
motion(t, h, EventMoving, `{"node":"novox","module":"node-tools","since":"`+c.now().Format(time.RFC3339Nano)+`","bound_seconds":600}`)
k := cond("machine.novox.tools-silent", Warning, "the tools on novox are silent")
k.Raised = c.now()
h.Condition(EventRaised, k)
runFor(h, c, time.Minute)
motion(t, h, EventStuck, `{"node":"novox","module":"node-tools","outcome":"failed","words":"the new build did not come healthy on novox","at":"`+c.now().Format(time.RFC3339Nano)+`"}`)
runFor(h, c, time.Minute)
if !heard("Delivery failed: node-tools on novox", tg, dt) || !heard("the new build did not come healthy on novox", tg, dt) {
t.Fatalf("a stuck module was not told at once with the view's words: %+v %+v", tg.sends, dt.sends)
}
}
modules/messenger/cmd/messenger/walks.go
+61
View File
@@ -86,6 +86,59 @@ const (
MostNamed = 6
)
// The delivery view's events that stand for a walk now that plans and their plan-moved are gone (release 6, ADR 0539 §17): a
// walk is the set of (node, module) in motion. `moving` opens one for a module on a node, `settled` ends it delivered, `stuck`
// ends it failed or stopped with the view's words. Each is stated as the WalkEvent a plan-moved once was, so the holding, the
// bound, the settle and the end message below stay as they were.
const (
ViewModule = "mesh-delivery-view"
EventMoving = "moving"
EventSettled = "settled"
EventStuck = "stuck"
)
// DecodeMotion reads a view event's body as a walk's start or end: moving {node, module, from_fp, to_fp, composed, since,
// bound_seconds}, settled {node, module, at}, stuck {node, module, outcome, words, at}. The walk's id is "<node>/<module>".
func DecodeMotion(event string, body []byte) (WalkEvent, bool, error) {
var b struct {
Node string `json:"node"`
Module string `json:"module"`
Since time.Time `json:"since"`
At time.Time `json:"at"`
BoundSeconds float64 `json:"bound_seconds"`
Outcome string `json:"outcome"`
Words string `json:"words"`
}
switch event {
case EventMoving, EventSettled, EventStuck:
default:
return WalkEvent{}, false, nil
}
if err := json.Unmarshal(body, &b); err != nil {
return WalkEvent{}, false, fmt.Errorf("a %s event's body is not a JSON object", event)
}
if b.Node == "" || b.Module == "" {
return WalkEvent{}, false, fmt.Errorf("a %s event names no node or no module", event)
}
if b.BoundSeconds < 0 {
return WalkEvent{}, false, fmt.Errorf("%s/%s: bound_seconds is negative", b.Node, b.Module)
}
ev := WalkEvent{Walk: b.Node + "/" + b.Module, Touches: []Touch{{Machine: b.Node, Modules: []string{b.Module}}}, Words: b.Words}
switch event {
case EventMoving:
ev.Phase, ev.BoundSeconds, ev.StartedAt, ev.At = WalkStarted, int(b.BoundSeconds), b.Since, b.Since
case EventSettled:
ev.Phase, ev.Outcome, ev.At = WalkEnded, WalkDelivered, b.At
case EventStuck:
ev.Phase, ev.Outcome, ev.At = WalkEnded, WalkFailed, b.At
if b.Outcome == WalkStopped {
ev.Outcome = WalkStopped
}
}
return ev, true, nil
}
// settle is how long a delivered walk still holds after its end (ADR 0505 §3).
func (h *Holder) settle() time.Duration {
if h.Settle == nil {
@@ -119,6 +172,8 @@ type WalkEvent struct {
// StartedAt is the walk's own start, on both events (`started_at`): the bound runs from it, so a router
// that hears a start late, or starts again, ends the hold on time.
StartedAt time.Time
// Words are the view's own words of why a walk ended failed or stopped (`stuck`); empty for a plan's.
Words string
}
// HeldMessage is one message a walk holds: when it was first held and what was to be said.
@@ -156,6 +211,8 @@ type Walk struct {
// is how it ended. Its end is told when SettleUntil passes.
SettleUntil time.Time `json:"settle_until,omitempty"`
Outcome string `json:"outcome,omitempty"`
// Words are the view's words of why it ended failed or stopped, told in the end message.
Words string `json:"words,omitempty"`
}
func (w *Walk) bound() time.Duration {
@@ -404,6 +461,7 @@ func (h *Holder) Walked(ev WalkEvent) {
}
if w != nil {
w.Touches = mergeTouches(w.Touches, ev.Touches)
w.Words = ev.Words
if !ev.StartedAt.IsZero() {
w.Started = ev.StartedAt
}
@@ -570,6 +628,9 @@ func (h *Holder) endWalk(w *Walk, outcome string, ended, now time.Time) {
m = Message{Title: head + ": " + reached + ", in " + roughly(took), Quiet: true}
} else {
lines := []string{"It took " + roughly(took) + "."}
if w.Words != "" {
lines = append(lines, w.Words)
}
lines = append(lines, h.heldLines(open, whats, cleared, "Still open after the delivery of "+deliveryWords(w.Touches)+":")...)
m = Message{Title: head + ": " + reached, Body: strings.Join(lines, "\n")}
}
modules/messenger/module.json
+3 -1
View File
@@ -77,7 +77,9 @@
"mesh-controller.condition-raised",
"mesh-controller.condition-changed",
"mesh-controller.condition-cleared",
"mesh-controller.plan-moved",
"mesh-delivery-view.moving",
"mesh-delivery-view.settled",
"mesh-delivery-view.stuck",
"intake.choice",
"intake.link",
"intake.failed",