diff --git a/modules/messenger/cmd/messenger/adr505_bus_test.go b/modules/messenger/cmd/messenger/adr505_bus_test.go new file mode 100644 index 00000000..440ca55e --- /dev/null +++ b/modules/messenger/cmd/messenger/adr505_bus_test.go @@ -0,0 +1,86 @@ +//go:build integration + +package main + +import ( + "encoding/json" + "strings" + "testing" + "time" +) + +// novox/hq ADR 0505 over the real bus, in the bed of condition_bus_test.go: the router whole, its walks kept in its +// state, the controller's walk and condition events handed to it as the runtime hands them. + +// controllerEvent hands the bundle one of the controller's events, as the runtime does. +func (b *conditionBed) controllerEvent(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": ControllerSeat + "." + event, + "node": "test", "body": json.RawMessage(raw)}}); err != nil { + t.Fatal(err) + } +} + +// 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) + } + return b +} + +// conditionEvent is a condition as the controller raises it (mesh-controller probes.go D1, signals.go). +func conditionEvent(key, scope, id, machine, headline string) map[string]any { + now := time.Now().UTC().Format(time.RFC3339) + return map[string]any{"key": key, "kind": key[strings.LastIndex(key, ".")+1:], "severity": Urgent, + "subject": map[string]string{"scope": scope, "id": id, "machine": machine}, "summary": headline, + "headline": headline, "explanation": "Needs you: look at it. " + headline + ".", "needs": "look at it.", + "raised": now, "at": now} +} + +func (b *conditionBed) toldWith(words string) *string { + for _, m := range b.told() { + if strings.Contains(m.Title+"\n"+m.Body, words) { + text := m.Title + "\n" + m.Body + return &text + } + } + return nil +} + +// 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) { + 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.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 { + t.Fatalf("the declared machine's condition was told during the walk") + } + b.controllerEvent(t, EventPlanMoved, walkEvent(WalkEnded, WalkDelivered, started, time.Now())) + time.Sleep(time.Second) + if b.toldWith("Shanks 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,") + }) +} diff --git a/modules/messenger/cmd/messenger/adr505_test.go b/modules/messenger/cmd/messenger/adr505_test.go index b48ee331..50ebb0e3 100644 --- a/modules/messenger/cmd/messenger/adr505_test.go +++ b/modules/messenger/cmd/messenger/adr505_test.go @@ -144,7 +144,7 @@ func TestAConditionRaisedInTheSettleIsHeldAndToldAtItsEnd(t *testing.T) { if clears { h.Condition(EventCleared, Condition{Key: k.Key, At: c.now()}) } - runFor(h, c, 2*time.Minute) // the router looks each minute: the settle's end is told at the next look + runFor(h, c, time.Minute) switch { case !clears && !heard("Still open after the delivery of gitea to novox:\n- Urgent: "+headlineOf(k.Key), tg, dt): t.Fatalf("still open at the settle's end and not told under the delivery: %+v", dt.sends) diff --git a/modules/messenger/cmd/messenger/bus_test.go b/modules/messenger/cmd/messenger/bus_test.go index 6e34d5eb..9c67bda6 100644 --- a/modules/messenger/cmd/messenger/bus_test.go +++ b/modules/messenger/cmd/messenger/bus_test.go @@ -56,6 +56,8 @@ type busRuntime struct { subscribed, taking bool buckets map[string]jetstream.KeyValue + // settings is the settings file the bundle starts with; empty is `{}`. + settings string } func newBusRuntime(t *testing.T) *busRuntime { @@ -130,7 +132,11 @@ func (r *busRuntime) request(method string, params any) (json.RawMessage, error) func (r *busRuntime) start(dir string) { r.t.Helper() settings := filepath.Join(dir, "settings.json") - _ = os.WriteFile(settings, []byte(`{}`), 0o600) + body := r.settings + if body == "" { + body = `{}` + } + _ = os.WriteFile(settings, []byte(body), 0o600) cmd := exec.Command(os.Args[0], "-test.run=^$") cmd.Env = append(os.Environ(), bundleVar+"=1", "MESH_MESSENGER_SETTINGS="+settings, "MESH_MESSENGER_STATE="+dir, "MESH_SERVED_MODULE=messenger", "MESH_MODULE=messenger", "MESH_NODE=test") diff --git a/modules/messenger/cmd/messenger/condition_bus_test.go b/modules/messenger/cmd/messenger/condition_bus_test.go index 69dd1739..9057315a 100644 --- a/modules/messenger/cmd/messenger/condition_bus_test.go +++ b/modules/messenger/cmd/messenger/condition_bus_test.go @@ -28,8 +28,15 @@ type conditionBed struct { } func newConditionBed(t *testing.T) *conditionBed { + t.Helper() + return newConditionBedWith(t, "") +} + +// newConditionBedWith is the bed with the bundle started on these settings. +func newConditionBedWith(t *testing.T, settings string) *conditionBed { t.Helper() b := &conditionBed{r: newBusRuntime(t)} + b.r.settings = settings b.r.setKinds(map[string][]string{"desktop": {"choice", "edit"}}) if _, err := b.r.nc.Subscribe("mesh.seat."+asks.ChannelSeat+".accept.*.desktop", func(m *nats.Msg) { var msg asks.Message diff --git a/modules/messenger/cmd/messenger/holder.go b/modules/messenger/cmd/messenger/holder.go index 699538e4..a2dcfaf3 100644 --- a/modules/messenger/cmd/messenger/holder.go +++ b/modules/messenger/cmd/messenger/holder.go @@ -843,6 +843,7 @@ func (h *Holder) Flush() { now := h.Now() h.mu.Unlock() h.duePasses(now) + h.settledWalks(now) // a settle's end is told at the next flush, not the next minute (ADR 0505 §3) h.flush(now, false) } diff --git a/modules/messenger/cmd/messenger/walks.go b/modules/messenger/cmd/messenger/walks.go index 201e0a80..b9f01c29 100644 --- a/modules/messenger/cmd/messenger/walks.go +++ b/modules/messenger/cmd/messenger/walks.go @@ -559,6 +559,22 @@ func (h *Holder) endWalk(w *Walk, outcome string, ended, now time.Time) { h.tellWalk(w, m, open, whats, oneLine, "ended "+outcome, now) } +// settledWalks tells the end of every delivered walk whose settle has passed (ADR 0505 §3), oldest first. +func (h *Holder) settledWalks(now time.Time) { + h.mu.Lock() + var settled []*Walk + for _, w := range h.walks { + if w.settling() && !now.Before(w.SettleUntil) { + settled = append(settled, w) + } + } + h.mu.Unlock() + sort.Slice(settled, func(i, j int) bool { return settled[i].SettleUntil.Before(settled[j].SettleUntil) }) + for _, w := range settled { + h.endSettled(w, now) + } +} + // endSettled tells the end of a delivered walk whose settle has passed (ADR 0505 §3): what it held still open, one // line each, and the cleared counted; then it is only the tomb that keeps a late start from opening it again. func (h *Holder) endSettled(w *Walk, now time.Time) { @@ -603,12 +619,8 @@ func (h *Holder) boundWalks(now time.Time) { h.mu.Lock() var due, gone []*Walk var stale []string - var settled []*Walk for _, w := range h.walks { - switch { - case w.settling() && !now.Before(w.SettleUntil): - settled = append(settled, w) - case !w.settling() && !w.BoundSaid && !now.Before(w.limit()): + if !w.settling() && !w.BoundSaid && !now.Before(w.limit()) { due = append(due, w) } if now.Sub(w.Started) >= WalkForget { @@ -641,10 +653,7 @@ func (h *Holder) boundWalks(now time.Time) { h.forgetWalk(w.ID) } }() - sort.Slice(settled, func(i, j int) bool { return settled[i].SettleUntil.Before(settled[j].SettleUntil) }) - for _, w := range settled { - h.endSettled(w, now) - } + h.settledWalks(now) sort.Slice(due, func(i, j int) bool { return due[i].Started.Before(due[j].Started) }) for _, w := range due { h.mu.Lock()