Prove over the bus that a condition on the one subject reaches the desk (issue 1218)

2026-10-19 11:59:49 +02:00
parent de6741a335
commit 5b235a6e80
1 changed files with 209 additions and 0 deletions
@@ -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)
+		}
+	}
+}