diff --git a/modules/messenger/cmd/messenger/condition_bus_test.go b/modules/messenger/cmd/messenger/condition_bus_test.go new file mode 100644 index 00000000..95c7bb48 --- /dev/null +++ b/modules/messenger/cmd/messenger/condition_bus_test.go @@ -0,0 +1,209 @@ +//go:build integration + +package main + +import ( + "context" + "encoding/json" + "strings" + "sync" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/asks" + "github.com/nats-io/nats.go" +) + +// A condition said on the operator channel's one subject over the real bus (novox/hq ADR 0437 ยง9, issue 1218), +// in the bed of bus_test.go. The fake runtime enforces no grant: that a module may say a condition only under its +// own name is the broker's, proven in the controller (issue 1216). + +// conditionBed starts the router with the desk able to send, hands it the conditions said on the one subject as +// work, and keeps what the desk is sent and whether each work item was acknowledged. +type conditionBed struct { + r *busRuntime + mu sync.Mutex + desk []asks.Message + acks []string // the answer of the router to each work item: "" when acknowledged +} + +func newConditionBed(t *testing.T) *conditionBed { + t.Helper() + b := &conditionBed{r: newBusRuntime(t)} + 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 + if json.Unmarshal(m.Data, &msg) == nil { + b.mu.Lock() + b.desk = append(b.desk, msg) + b.mu.Unlock() + } + }); err != nil { + t.Fatal(err) + } + if _, err := b.r.nc.Subscribe("mesh.seat."+asks.Seat+".accept.condition.*", func(m *nats.Msg) { + answer, err := b.r.request("mesh/work", map[string]any{"work": map[string]any{"worker": AsksWorker, "subject": m.Subject, "body": json.RawMessage(m.Data)}}) + var a struct { + Error *struct { + Message string `json:"message"` + } `json:"error"` + } + _ = json.Unmarshal(answer, &a) + res := "" + if err != nil { + res = err.Error() + } else if a.Error != nil { + res = a.Error.Message + } + b.mu.Lock() + b.acks = append(b.acks, res) + b.mu.Unlock() + }); err != nil { + t.Fatal(err) + } + b.forwardEvents(t) + b.r.start(t.TempDir()) + b.r.until("the bundle takes work and read the kinds", 20*time.Second, func() bool { + b.r.mu.Lock() + defer b.r.mu.Unlock() + return b.r.taking && b.r.kindsRead > 0 + }) + b.r.until("the bundle subscribed to the channels' word", 20*time.Second, func() bool { + b.r.mu.Lock() + defer b.r.mu.Unlock() + return b.r.subscribed + }) + standing, _ := json.Marshal(Standing{Ready: true, At: time.Now().UTC()}) + if err := b.r.nc.Publish(asks.IntakeSubject(asks.StandingWhat, "desktop"), standing); err != nil { + t.Fatal(err) + } + time.Sleep(time.Second) + return b +} + +// forwardEvents hands the bundle the channels' word, as bus_test.go's forward does for it. +func (b *conditionBed) forwardEvents(t *testing.T) { + t.Helper() + if _, err := b.r.nc.Subscribe("mesh.seat."+asks.IntakeSeat+".event.>", func(m *nats.Msg) { + key := strings.Replace(strings.TrimPrefix(m.Subject, "mesh.seat."), ".event", "", 1) + _, _ = b.r.request("mesh/event", map[string]any{"envelope": map[string]any{"key": key, "node": "test", "body": json.RawMessage(m.Data)}}) + }); err != nil { + t.Fatal(err) + } +} + +func (b *conditionBed) say(t *testing.T, publisher string, c asks.Condition) { + t.Helper() + body, err := asks.EncodeCondition(c) + if err != nil { + t.Fatal(err) + } + subject, err := asks.ConditionSubject(publisher) + if err != nil { + t.Fatal(err) + } + if err := b.r.nc.Publish(subject, body); err != nil { + t.Fatal(err) + } +} + +func (b *conditionBed) record(key string) *Record { + e, err := b.r.bucket("open").Get(context.Background(), kvKey(key)) + if err != nil { + return nil + } + var rec Record + if json.Unmarshal(e.Value(), &rec) != nil { + return nil + } + return &rec +} + +func (b *conditionBed) told() []asks.Message { + b.mu.Lock() + defer b.mu.Unlock() + return append([]asks.Message(nil), b.desk...) +} + +func (b *conditionBed) acknowledged(n int) bool { + b.mu.Lock() + defer b.mu.Unlock() + if len(b.acks) != n { + return false + } + for _, a := range b.acks { + if a != "" { + return false + } + } + return true +} + +func busCondition(key, id, headline, needs string) asks.Condition { + verdict := asks.NothingToDo + if needs != "" { + verdict = asks.NeedsYou + needs + } + return asks.Condition{Key: key, Severity: asks.SeverityWarning, Subject: asks.Scope{Scope: "merge", ID: id}, + State: asks.StateRaised, Summary: "the merge waits on a check", Headline: headline, Needs: needs, + Explanation: verdict + " It waits on a check."} +} + +// 1218/AC1 and 1218/AC2: a raised condition said on the one subject reaches the desk, is recorded and its work item +// is acknowledged; its cleared form ends it. The router holds a warning for its start grace and burst window. +func TestAConditionSaidOnTheOneSubjectReachesTheDeskAndIsClearedOverTheBus(t *testing.T) { + b := newConditionBed(t) + b.say(t, "mesh-delivery-teller", busCondition("merge.d-7.stalled", "d-7", "A merge waits on a check", "release it, or stop it.")) + // 1218/AC1: recorded, acknowledged, and told to the desk. + b.r.until("the condition is recorded open", 30*time.Second, func() bool { + rec := b.record("merge.d-7.stalled") + return rec != nil && rec.Origin == "condition" && rec.Cleared.IsZero() + }) + b.r.until("the work item is acknowledged", 30*time.Second, func() bool { return b.acknowledged(1) }) + b.r.until("the desk is sent the condition", 3*time.Minute, func() bool { + for _, m := range b.told() { + if strings.Contains(m.Title+m.Body, "A merge waits on a check") { + return true + } + } + return false + }) + time.Sleep(3 * time.Second) + if n := b.nAcks(); n != 1 { + t.Errorf("the work item was offered %d times, want once: it was not acknowledged", n) + } + // 1218/AC2: the cleared form, on the same subject, ends it. + b.say(t, "mesh-delivery-teller", asks.Condition{Key: "merge.d-7.stalled", Subject: asks.Scope{Scope: "merge", ID: "d-7"}, State: asks.StateCleared}) + b.r.until("the condition is recorded cleared", 30*time.Second, func() bool { + rec := b.record("merge.d-7.stalled") + return rec == nil || !rec.Cleared.IsZero() + }) +} + +func (b *conditionBed) nAcks() int { + b.mu.Lock() + defer b.mu.Unlock() + return len(b.acks) +} + +// 1218/AC3: a warning whose needs is empty is recorded and reaches no channel. A second warning that needs the +// operator, said after it, is told: when it is, the quiet one's turn has passed and it was not sent. +func TestAWarningThatNeedsNothingReachesNoChannelOverTheBus(t *testing.T) { + b := newConditionBed(t) + b.say(t, "mesh-delivery-teller", busCondition("merge.d-8.quiet", "d-8", "A quiet merge note", "")) + b.r.until("the quiet warning is recorded", 30*time.Second, func() bool { return b.record("merge.d-8.quiet") != nil }) + b.say(t, "mesh-delivery-teller", busCondition("merge.d-9.loud", "d-9", "A loud merge note", "release it, or stop it.")) + b.r.until("the warning that needs the operator is sent to the desk", 3*time.Minute, func() bool { + for _, m := range b.told() { + if strings.Contains(m.Title+m.Body, "A loud merge note") { + return true + } + } + return false + }) + for _, m := range b.told() { + if strings.Contains(m.Title+m.Body, "A quiet merge note") { + t.Errorf("a warning that needs nothing was sent: %+v", m) + } + } +}