diff --git a/modules/messenger/cmd/messenger/adr505_bus_test.go b/modules/messenger/cmd/messenger/adr505_bus_test.go index 440ca55e..254babc8 100644 --- a/modules/messenger/cmd/messenger/adr505_bus_test.go +++ b/modules/messenger/cmd/messenger/adr505_bus_test.go @@ -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") }) } diff --git a/modules/messenger/cmd/messenger/main.go b/modules/messenger/cmd/messenger/main.go index 57bccaaa..458dd236 100644 --- a/modules/messenger/cmd/messenger/main.go +++ b/modules/messenger/cmd/messenger/main.go @@ -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 { diff --git a/modules/messenger/cmd/messenger/manifest_test.go b/modules/messenger/cmd/messenger/manifest_test.go index 8fc2f9e5..2281501b 100644 --- a/modules/messenger/cmd/messenger/manifest_test.go +++ b/modules/messenger/cmd/messenger/manifest_test.go @@ -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 diff --git a/modules/messenger/cmd/messenger/motion_test.go b/modules/messenger/cmd/messenger/motion_test.go new file mode 100644 index 00000000..a31e0603 --- /dev/null +++ b/modules/messenger/cmd/messenger/motion_test.go @@ -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) + } +} diff --git a/modules/messenger/cmd/messenger/walks.go b/modules/messenger/cmd/messenger/walks.go index ada2b575..98788aec 100644 --- a/modules/messenger/cmd/messenger/walks.go +++ b/modules/messenger/cmd/messenger/walks.go @@ -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 "/". +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")} } diff --git a/modules/messenger/module.json b/modules/messenger/module.json index d7da1d68..4aceaebe 100644 --- a/modules/messenger/module.json +++ b/modules/messenger/module.json @@ -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",