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

nats removes a push consumer; gitea, mesh-conditions, mesh-issues and messenger drop what named the removed controller (issue 1126) #607

Merged
mesh-tools merged 7 commits from fix/consumer-remove-any into main 2026-11-04 10:36:41 +00:00
Conversation 1 Commits 7 Files Changed 27
+17 -178
3 changed files with 17 additions and 178 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 1572ab9b25 - Show all commits
modules/messenger/cmd/messenger/adr505_test.go
-6
View File
@@ -51,12 +51,6 @@ var shanksDeclared = Touch{Machine: "shanks", Modules: []string{"mesh-build-imag
// ADR 0505 §2b: a machine whose declaration a walk's merges change is held like a machine it sends to, and is not
// named among the machines the walk reached.
func TestAMachineWhoseDeclarationTheWalkChangesIsHeldAndNotNamedAsReached(t *testing.T) {
ev, ok, err := DecodeWalk([]byte(`{"id":"plan-d","walk":"plan-d","phase":"started","budget_class":"leaf",` +
`"bound_seconds":600,"touches":[{"machine":"novox","modules":["gitea"]},` +
`{"machine":"shanks","modules":["mesh-build-image"],"declared":true}]}`))
if err != nil || !ok || len(ev.Touches) != 2 || !ev.Touches[1].Declared || ev.Touches[0].Declared {
t.Fatalf("a declared touch was not read: %+v %v %v", ev.Touches, ok, err)
}
h, tg, dt, c, _ := newHolder(t)
walk(t, h, "plan-d", WalkStarted, "", c.now(), 600, novox("gitea"), shanksDeclared)
k := cond("machine.shanks.uncomposable", Urgent, "shanks cannot get new instructions")
modules/messenger/cmd/messenger/walks.go
+8 -111
View File
@@ -3,26 +3,14 @@ package main
// While a delivery walks, what it touches is held from the operator and told once, with the walk's result,
// when it ends (novox/hq ADR 0308, issue 532).
//
// **What this reads.** The controller says each walk's start and end on its `plan-moved` event (the plan
// whole, ADR 0282 §6), and on those two moments the body carries beside the plan (ADR 0308 §1):
// **What this reads.** The delivery view's `moving`, `settled` and `stuck` events (DecodeMotion below): a walk is a module in
// motion on a node, its id "<node>/<module>", and it carries (ADR 0308 §1):
//
// - `walk`: the plan's id (`id` is read when it is missing);
// - `phase`: `started` or `ended`; a plan-moved without it is a move inside a walk and is not read here;
// - `touches`: `[{machine, modules}]`, every module the walk moves on every machine it sends it to, and every
// module a moved tool runner or node-engine carries there;
// - `outcome`: `delivered`, `failed` or `stopped`, on `ended`;
// - `bound_seconds`: twice the largest delivery budget among its deliveries, or 900 when none has one (§5),
// on `started`; missing, 900 is taken;
// - `budget_class`: `core`, `leaf` or `none`;
// - `started_at` on the start and `ended_at` on the end (else `at`, else the plan's `updated`): when it
// happened. A start older than its bound is history and opens nothing; an end older than FreshFor with
// nothing held closes quietly.
//
// A touch may name `carried` beside `modules` (what a moved tool runner or node-engine carries); the
// controller lists those in `modules` too, which is what is read.
//
// An event of its own named `walk`, with the same body, is read the same way, should the controller say it
// there instead.
// - `touches`: `[{machine, modules}]`, the module it moves on that node;
// - `outcome`: `delivered`, `failed` or `stopped`, on the end;
// - `bound_seconds`: on the start; missing, 900 is taken (§5);
// - the start's own moment on the start and the end's on the end. A start older than its bound is history and opens
// nothing; an end older than FreshFor with nothing held closes quietly.
//
// **What is held (§2).** While a walk is open and within its bound, a message whose subject is a machine the
// walk touches, or a module it touches, is held instead of sent: kept on the walk, with when it was held and
@@ -60,11 +48,8 @@ import (
"time"
)
// The walk events, by their local names under the controller's seat.
// The walk's phases and outcomes.
const (
EventPlanMoved = "plan-moved"
EventWalk = "walk"
WalkStarted = "started"
WalkEnded = "ended"
@@ -279,94 +264,6 @@ type walkKeeper interface {
Passes() (map[string]Pass, error)
}
// DecodeWalk reads a plan-moved or walk event's body: false when it says no walk's start or end.
func DecodeWalk(body []byte) (WalkEvent, bool, error) {
var raw map[string]json.RawMessage
if err := json.Unmarshal(body, &raw); err != nil || raw == nil {
return WalkEvent{}, false, fmt.Errorf("a walk event's body is not a JSON object")
}
str := func(names ...string) string {
for _, n := range names {
var s string
if v, ok := raw[n]; ok && json.Unmarshal(v, &s) == nil && strings.TrimSpace(s) != "" {
return strings.TrimSpace(s)
}
}
return ""
}
var ev WalkEvent
ev.Phase = str("phase")
if ev.Phase == "" {
return WalkEvent{}, false, nil
}
ev.Walk = str("walk", "id")
if ev.Walk == "" {
return WalkEvent{}, false, fmt.Errorf("a walk's %s names no walk", ev.Phase)
}
switch ev.Phase {
case WalkStarted, WalkEnded:
default:
return WalkEvent{}, false, fmt.Errorf("walk %s: phase %q is neither started nor ended", ev.Walk, ev.Phase)
}
ev.Outcome = str("outcome")
if ev.Phase == WalkEnded {
switch ev.Outcome {
case WalkDelivered, WalkFailed, WalkStopped:
default:
return WalkEvent{}, false, fmt.Errorf("walk %s ended with outcome %q: neither delivered, failed nor stopped", ev.Walk, ev.Outcome)
}
}
ev.Class = str("budget_class", "budget-class")
for _, n := range []string{"bound_seconds", "bound-seconds"} {
if v, ok := raw[n]; ok && string(v) != "null" {
var f float64
if json.Unmarshal(v, &f) != nil || f < 0 {
return WalkEvent{}, false, fmt.Errorf("walk %s: %s is not a number of seconds", ev.Walk, n)
}
ev.BoundSeconds = int(f)
break
}
}
if v, ok := raw["touches"]; ok && string(v) != "null" {
var list []struct {
Machine string `json:"machine"`
Node string `json:"node"`
Modules []string `json:"modules"`
Declared bool `json:"declared"`
}
if json.Unmarshal(v, &list) != nil {
return WalkEvent{}, false, fmt.Errorf("walk %s: touches is not a list of {machine, modules}", ev.Walk)
}
for _, t := range list {
m := t.Machine
if m == "" {
m = t.Node
}
if m != "" {
ev.Touches = append(ev.Touches, Touch{Machine: m, Modules: t.Modules, Declared: t.Declared})
}
}
}
if s := str("started_at"); s != "" {
if t, err := time.Parse(time.RFC3339Nano, s); err == nil {
ev.StartedAt = t
}
}
when := []string{"started_at", "at", "updated"}
if ev.Phase == WalkEnded {
when = []string{"ended_at", "at", "updated"}
}
for _, n := range when {
if s := str(n); s != "" {
if t, err := time.Parse(time.RFC3339Nano, s); err == nil {
ev.At = t
break
}
}
}
return ev, true, nil
}
// mergeTouches is what two sayings of a walk's touches name together. A machine is declared only while no saying
// sends to it (ADR 0505 §2b).
func mergeTouches(a, b []Touch) []Touch {
modules/messenger/cmd/messenger/walks_test.go
+9 -61
View File
@@ -1,7 +1,6 @@
package main
import (
"encoding/json"
"errors"
"strings"
"testing"
@@ -11,35 +10,12 @@ import (
// The tests of novox/hq ADR 0308 "How it is checked" §2–§6, with the controller's walk events given by a fake
// source: the body the controller's plan-moved carries at a walk's start and end (ADR 0308 §1).
// walkBody is a plan-moved body as the controller says a walk's start or end (mesh-controller pull request 247).
func walkBody(t *testing.T, id, phase, outcome string, at time.Time, bound int, touches ...Touch) []byte {
t.Helper()
b := map[string]any{"id": id, "state": "walking", "walk": id, "phase": phase, "touches": touches,
"updated": at.Format(time.RFC3339Nano), "budget_class": "leaf"}
if phase == WalkStarted {
b["started_at"] = at.Format(time.RFC3339Nano)
} else {
b["ended_at"] = at.Format(time.RFC3339Nano)
}
if outcome != "" {
b["outcome"] = outcome
}
if bound > 0 {
b["bound_seconds"] = bound
}
raw, err := json.Marshal(b)
if err != nil {
t.Fatal(err)
}
return raw
}
// walk says a walk's start or end to the router, through the decoder the bus's events go through.
// walk says a walk's start or end to the router, as the delivery view's events are read into it (DecodeMotion).
func walk(t *testing.T, h *Holder, id, phase, outcome string, at time.Time, bound int, touches ...Touch) {
t.Helper()
ev, ok, err := DecodeWalk(walkBody(t, id, phase, outcome, at, bound, touches...))
if err != nil || !ok {
t.Fatalf("the walk event was not read: %v %v", ok, err)
ev := WalkEvent{Walk: id, Phase: phase, Outcome: outcome, Class: "leaf", Touches: touches, BoundSeconds: bound, At: at}
if phase == WalkStarted {
ev.StartedAt = at
}
h.Walked(ev)
}
@@ -412,27 +388,8 @@ func TestARestartSaysNothingItHasSaid(t *testing.T) {
}
}
// The controller's walk events are read as ADR 0308 §1 says them; any other plan-moved is not a walk's.
func TestTheWalkEventsAreReadAsTheControllerSaysThem(t *testing.T) {
at := time.Date(2026, 10, 12, 11, 12, 0, 0, time.UTC)
ev, ok, err := DecodeWalk(walkBody(t, "plan-1", WalkStarted, "", at, 1200, novox("mesh-tools", "node-tools")))
if err != nil || !ok || ev.Walk != "plan-1" || ev.BoundSeconds != 1200 || ev.Class != "leaf" || !ev.At.Equal(at) ||
len(ev.Touches) != 1 || ev.Touches[0].Machine != "novox" || len(ev.Touches[0].Modules) != 2 {
t.Fatalf("%+v %v %v", ev, ok, err)
}
// As mesh-controller's walk events (its pull request 247) say an end: ended_at, and carried beside modules.
end, ok, err := DecodeWalk([]byte(`{"id":"plan-1","walk":"plan-1","phase":"ended","outcome":"stopped",` +
`"touches":[{"machine":"novox","modules":["node-tools","messenger"],"carried":["messenger"]}],` +
`"started_at":"2026-10-12T11:12:00Z","ended_at":"2026-10-12T11:19:00Z","updated":"2026-10-12T11:19:01Z"}`))
if err != nil || !ok || end.Outcome != WalkStopped || !end.At.Equal(at.Add(7*time.Minute)) || len(end.Touches[0].Modules) != 2 {
t.Fatalf("%+v %v %v", end, ok, err)
}
if _, ok, err := DecodeWalk([]byte(`{"id":"plan-1","state":"walking","tier":1}`)); ok || err != nil {
t.Fatalf("a move inside a walk read as its start or end: %v %v", ok, err)
}
if _, _, err := DecodeWalk([]byte(`{"walk":"plan-1","phase":"ended"}`)); err == nil {
t.Fatalf("an end without its outcome was read")
}
// A walk without a bound is bounded at the default, and a start older than its bound is history.
func TestAWalkWithoutABoundIsBoundedAtTheDefault(t *testing.T) {
w := Walk{}
if w.bound() != DefaultWalkBound {
t.Fatalf("a walk without a bound is bounded at %s", w.bound())
@@ -540,11 +497,7 @@ func TestAStartHeardAfterItsEndOpensNothing(t *testing.T) {
h, tg, dt, c, st := newHolder(t)
started := c.now()
c.pass(20 * time.Second)
end, _, err := DecodeWalk([]byte(`{"walk":"plan-e","phase":"ended","outcome":"delivered","touches":[{"machine":"novox","modules":["mesh-tools"]}],` +
`"started_at":"` + started.Format(time.RFC3339Nano) + `","ended_at":"` + c.now().Format(time.RFC3339Nano) + `"}`))
if err != nil {
t.Fatal(err)
}
end := WalkEvent{Walk: "plan-e", Phase: WalkEnded, Outcome: WalkDelivered, Touches: []Touch{novox("mesh-tools")}, StartedAt: started, At: c.now()}
h.Walked(end)
c.pass(time.Second)
walk(t, h, "plan-e", WalkStarted, "", started, 600, novox("mesh-tools"))
@@ -747,14 +700,9 @@ func TestAStartHeardAfterItsEndOpensNothingWhenTheControllersClockIsAhead(t *tes
h, _, _, c, _ := newHolder(t)
ended := c.now().Add(10 * time.Minute) // by the controller's clock
started := ended.Add(-20 * time.Second)
body := `{"walk":"plan-k","phase":"ended","outcome":"delivered","touches":[{"machine":"novox","modules":["mesh-tools"]}],` +
`"ended_at":"` + ended.Format(time.RFC3339Nano) + `"`
end := WalkEvent{Walk: "plan-k", Phase: WalkEnded, Outcome: WalkDelivered, Touches: []Touch{novox("mesh-tools")}, At: ended}
if said == "both" {
body += `,"started_at":"` + started.Format(time.RFC3339Nano) + `"`
}
end, _, err := DecodeWalk([]byte(body + `}`))
if err != nil {
t.Fatal(err)
end.StartedAt = started
}
h.Walked(end)
c.pass(time.Second)