From 77d751fe830d4358728714572bf8869f196aaf9d Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 16 Oct 2026 17:57:52 +0200 Subject: [PATCH] Test the router whole over the real bus: a held ask is offered when a channel can carry it (issue 926, ADR 0397) --- modules/messenger/cmd/messenger/bus_test.go | 380 ++++++++++++++++++++ modules/messenger/cmd/messenger/main.go | 5 +- modules/messenger/go.mod | 21 +- modules/messenger/go.sum | 31 +- 4 files changed, 428 insertions(+), 9 deletions(-) create mode 100644 modules/messenger/cmd/messenger/bus_test.go diff --git a/modules/messenger/cmd/messenger/bus_test.go b/modules/messenger/cmd/messenger/bus_test.go new file mode 100644 index 00000000..28c8135e --- /dev/null +++ b/modules/messenger/cmd/messenger/bus_test.go @@ -0,0 +1,380 @@ +package main + +import ( + "bufio" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + "sync" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/asks" + "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" +) + +// The router started whole over a real bus (novox/hq ADR 0391, issue 926). A JetStream server is linked into the +// test and started for it alone; the module is this test binary started again as the bundle, speaking MCP on its +// standard streams to a runtime played here, which does what the tool runner does for it: its state is buckets +// of the bus, an ask submitted on the seat's subject is work handed to the bundle, what the bundle publishes on a +// seat is published on the bus, an event on the bus is handed to the bundle, and the controller's membership is +// the kinds this test says. It runs in the forge's merge check, not on the way. + +const bundleVar = "MESH_MESSENGER_TEST_BUNDLE" + +func TestMain(m *testing.M) { + if os.Getenv(bundleVar) == "1" { + kindsEvery = 200 * time.Millisecond + main() + os.Exit(0) + } + os.Exit(m.Run()) +} + +type busRuntime struct { + t *testing.T + nc *nats.Conn + js jetstream.JetStream + in io.WriteCloser + mu sync.Mutex // writes to the bundle, and the fields below + next int + wait map[string]chan json.RawMessage + + kinds []map[string]any + kindsRead int + subscribed, + taking bool + buckets map[string]jetstream.KeyValue +} + +func newBusRuntime(t *testing.T) *busRuntime { + t.Helper() + s, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true}) + if err != nil { + t.Fatal(err) + } + go s.Start() + if !s.ReadyForConnections(30 * time.Second) { + t.Fatal("no bus within 30s") + } + t.Cleanup(func() { s.Shutdown(); s.WaitForShutdown() }) + nc, err := nats.Connect(s.ClientURL()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + js, _ := jetstream.New(nc) + return &busRuntime{t: t, nc: nc, js: js, wait: map[string]chan json.RawMessage{}, buckets: map[string]jetstream.KeyValue{}} +} + +func (r *busRuntime) bucket(name string) jetstream.KeyValue { + r.mu.Lock() + defer r.mu.Unlock() + if kv, ok := r.buckets[name]; ok { + return kv + } + kv, err := r.js.CreateOrUpdateKeyValue(context.Background(), jetstream.KeyValueConfig{Bucket: name}) + if err != nil { + r.t.Fatal(err) + } + r.buckets[name] = kv + return kv +} + +func (r *busRuntime) setKinds(caps map[string][]string) { + var out []map[string]any + for kind, c := range caps { + out = append(out, map[string]any{"seat": asks.ChannelSeat, "kind": kind, "module": kind, "node": "test", "capabilities": c}) + } + r.mu.Lock() + r.kinds = out + r.mu.Unlock() +} + +func (r *busRuntime) say(m map[string]any) { + m["jsonrpc"] = "2.0" + raw, _ := json.Marshal(m) + r.mu.Lock() + defer r.mu.Unlock() + _, _ = r.in.Write(append(raw, '\n')) +} + +// request is a request to the bundle (work, an event, a tool call) and its answer. +func (r *busRuntime) request(method string, params any) (json.RawMessage, error) { + r.mu.Lock() + r.next++ + id := "runtime-" + strconv.Itoa(r.next) + ch := make(chan json.RawMessage, 1) + r.wait[id] = ch + r.mu.Unlock() + r.say(map[string]any{"id": id, "method": method, "params": params}) + select { + case raw := <-ch: + return raw, nil + case <-time.After(60 * time.Second): + return nil, fmt.Errorf("%s was not answered in 60 s", method) + } +} + +func (r *busRuntime) start(dir string) { + r.t.Helper() + settings := filepath.Join(dir, "settings.json") + _ = os.WriteFile(settings, []byte(`{}`), 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") + in, _ := cmd.StdinPipe() + out, _ := cmd.StdoutPipe() + cmd.Stderr = busLog{r.t} + if err := cmd.Start(); err != nil { + r.t.Fatal(err) + } + r.in = in + go r.read(out) + r.t.Cleanup(func() { _ = in.Close(); _ = cmd.Wait() }) +} + +type busLog struct{ t *testing.T } + +func (l busLog) Write(p []byte) (int, error) { l.t.Logf("bundle: %s", p); return len(p), nil } + +func (r *busRuntime) read(out io.Reader) { + sc := bufio.NewScanner(out) + sc.Buffer(make([]byte, 1<<20), 1<<24) + for sc.Scan() { + var m struct { + ID json.RawMessage `json:"id"` + Method string `json:"method"` + Params json.RawMessage `json:"params"` + } + if json.Unmarshal(sc.Bytes(), &m) != nil { + continue + } + if m.Method == "" { + var id string + _ = json.Unmarshal(m.ID, &id) + r.mu.Lock() + ch := r.wait[id] + delete(r.wait, id) + r.mu.Unlock() + if ch != nil { + ch <- append([]byte(nil), sc.Bytes()...) + } + continue + } + go r.runtime(m.ID, m.Method, m.Params) + } +} + +// runtime is what the tool runner does for the bundle: only what the router uses. +func (r *busRuntime) runtime(id json.RawMessage, method string, params json.RawMessage) { + var p struct { + State string `json:"state"` + Key string `json:"key"` + Value json.RawMessage `json:"value"` + Revision uint64 `json:"revision"` + Body json.RawMessage `json:"body"` + Subject string `json:"subject"` + Pattern string `json:"pattern"` + Worker string `json:"worker"` + } + _ = json.Unmarshal(params, &p) + ctx := context.Background() + const changed = -32010 + code := -32000 + result, err := func() (any, error) { + switch method { + case "mesh/state.get": + e, err := r.bucket(p.State).Get(ctx, p.Key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil, nil + } + if err != nil { + return nil, err + } + return map[string]any{"key": p.Key, "value": json.RawMessage(e.Value()), "revision": e.Revision()}, nil + case "mesh/state.put": + rev, err := r.bucket(p.State).Put(ctx, p.Key, p.Value) + return map[string]any{"revision": rev}, err + case "mesh/state.create": + rev, err := r.bucket(p.State).Create(ctx, p.Key, p.Value) + if errors.Is(err, jetstream.ErrKeyExists) { + code = changed + } + return map[string]any{"revision": rev}, err + case "mesh/state.update": + rev, err := r.bucket(p.State).Update(ctx, p.Key, p.Value, p.Revision) + if err != nil { + code = changed + } + return map[string]any{"revision": rev}, err + case "mesh/state.delete": + err := r.bucket(p.State).Delete(ctx, p.Key) + return map[string]any{}, err + case "mesh/state.keys": + keys, err := r.bucket(p.State).Keys(ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + return []string{}, nil + } + return keys, err + case "mesh/subscribe": + if strings.HasPrefix(p.Pattern, asks.IntakeSeat) { + r.mu.Lock() + r.subscribed = true + r.mu.Unlock() + } + return map[string]any{}, nil + case "mesh/seat.take": + r.mu.Lock() + r.taking = true + r.mu.Unlock() + return map[string]any{}, nil + case "mesh/seat.answer", "mesh/publish": + return map[string]any{}, nil + case "mesh/seat.kinds": + r.mu.Lock() + defer r.mu.Unlock() + r.kindsRead++ + return r.kinds, nil + case "mesh/seat.publish": + return map[string]any{}, r.nc.Publish(p.Subject, p.Body) + case "mesh/ask": + if p.Key == RootFreeVerb { + var a struct { + Machines []string `json:"machines"` + } + _ = json.Unmarshal(p.Body, &a) + var out []RootVerdict + for _, m := range a.Machines { + out = append(out, RootVerdict{Machine: m, Free: true}) + } + return map[string]any{"machines": out}, nil + } + return nil, fmt.Errorf("the test runtime answers nothing for %s", p.Key) + } + return nil, fmt.Errorf("the runtime offers no %s", method) + }() + if err != nil { + r.say(map[string]any{"id": id, "error": map[string]any{"code": code, "message": err.Error()}}) + return + } + r.say(map[string]any{"id": id, "result": result}) +} + +// forward hands the bundle what is published on the bus that it takes: an ask submitted to the seat as work, +// a channel's word as an event. +func (r *busRuntime) forward() { + if _, err := r.nc.Subscribe("mesh.seat."+asks.Seat+".accept.ask.*", func(m *nats.Msg) { + _, _ = r.request("mesh/work", map[string]any{"work": map[string]any{"worker": AsksWorker, "subject": m.Subject, "body": json.RawMessage(m.Data)}}) + }); err != nil { + r.t.Fatal(err) + } + if _, err := r.nc.Subscribe("mesh.seat."+asks.IntakeSeat+".event.>", func(m *nats.Msg) { + key := strings.Replace(strings.TrimPrefix(m.Subject, "mesh.seat."), ".event", "", 1) + _, _ = r.request("mesh/event", map[string]any{"envelope": map[string]any{"key": key, "node": "test", "body": json.RawMessage(m.Data)}}) + }); err != nil { + r.t.Fatal(err) + } +} + +func (r *busRuntime) until(what string, within time.Duration, ok func() bool) { + r.t.Helper() + for deadline := time.Now().Add(within); time.Now().Before(deadline); time.Sleep(100 * time.Millisecond) { + if ok() { + return + } + } + r.t.Fatalf("not within %s: %s", within, what) +} + +func (r *busRuntime) askRecord(key string) *AskRecord { + e, err := r.bucket("asks").Get(context.Background(), key) + if err != nil { + return nil + } + var rec AskRecord + _ = json.Unmarshal(e.Value(), &rec) + return &rec +} + +// A held ask over the real bus (ADR 0397, issue 926): taken while the controller's word on Telegram lacks a +// verified sender, it is held with the reason and its asker is told nothing of an end; once the word gives it a +// verified sender, it is offered on Telegram. +func TestAnApprovalAskIsHeldThenOfferedWhenAChannelCanCarryItOverTheBus(t *testing.T) { + r := newBusRuntime(t) + away := []string{"deliver", "reaches-away", "choice", "edit", "exact-render"} + r.setKinds(map[string][]string{"telegram": away}) + now := time.Now().UTC() + // The operator's account, linked long enough ago to approve. + id := Identity{Kind: "telegram", Identity: "42", Display: "Jo", Linked: now.Add(-3 * time.Hour), ApproveFrom: now.Add(-2 * time.Hour)} + raw, _ := json.Marshal(id) + if _, err := r.bucket("identities").Put(context.Background(), identityKey("telegram", "42"), raw); err != nil { + t.Fatal(err) + } + var told, shown []json.RawMessage // shown is kept to say nothing was shown while held + var seen sync.Mutex + collect := func(into *[]json.RawMessage) nats.MsgHandler { + return func(m *nats.Msg) { seen.Lock(); *into = append(*into, append([]byte(nil), m.Data...)); seen.Unlock() } + } + if _, err := r.nc.Subscribe(asks.DecidedSubject("mesh-delivery"), collect(&told)); err != nil { + t.Fatal(err) + } + if _, err := r.nc.Subscribe(asks.ChannelSubject(asks.Show, "telegram"), collect(&shown)); err != nil { + t.Fatal(err) + } + r.forward() + r.start(t.TempDir()) + r.until("the bundle subscribed, takes asks and read the kinds", 20*time.Second, func() bool { + r.mu.Lock() + defer r.mu.Unlock() + return r.subscribed && r.taking && r.kindsRead > 0 + }) + standing, _ := json.Marshal(Standing{Ready: true, At: time.Now().UTC()}) + if err := r.nc.Publish(asks.IntakeSubject(asks.StandingWhat, "telegram"), standing); err != nil { + t.Fatal(err) + } + time.Sleep(time.Second) + c := &clock{t: time.Now().UTC()} + submitted, _ := json.Marshal(heldAsk(c, "d-1")) + if err := r.nc.Publish(asks.AskSubject("mesh-delivery"), submitted); err != nil { + t.Fatal(err) + } + r.until("the ask is recorded held with the reason", 20*time.Second, func() bool { + rec := r.askRecord("mesh-delivery.d-1") + return rec != nil && rec.State == StateHeld && strings.Contains(rec.HeldWhy, "Telegram cannot prove who answered") + }) + rec := r.askRecord("mesh-delivery.d-1") + if len(rec.Tickets) != 0 { + t.Errorf("a held ask has tickets: %+v", rec) + } + seen.Lock() + if len(told) != 0 || len(shown) != 0 { + t.Errorf("a held ask was refused or shown: told %s, shown %s", told, shown) + } + seen.Unlock() + // The asks verb shows it held, since when, and why. + out, err := r.request("tools/call", map[string]any{"name": "operator-channel.asks", "arguments": map[string]any{}}) + if err != nil || !strings.Contains(string(out), `held_why`) || !strings.Contains(string(out), `held`) { + t.Errorf("the asks verb: %s %v", out, err) + } + // The controller's word now gives Telegram a verified sender: the ask is offered there. + r.setKinds(map[string][]string{"telegram": append(append([]string{}, away...), "verified-sender")}) + r.until("the ask is offered", 30*time.Second, func() bool { + rec := r.askRecord("mesh-delivery.d-1") + return rec != nil && rec.State == StateOpen && len(rec.Tickets) == 3 && rec.HeldWhy == "" + }) + // Its message is the router's ordinary one for a new ask, said after the burst window (BurstWindow), which a + // unit test covers with a clock; this test waits for no half minute. + if rec := r.askRecord("mesh-delivery.d-1"); rec.Tickets == nil { + t.Errorf("offered with no tickets: %+v", rec) + } +} diff --git a/modules/messenger/cmd/messenger/main.go b/modules/messenger/cmd/messenger/main.go index 484db587..649e90a5 100644 --- a/modules/messenger/cmd/messenger/main.go +++ b/modules/messenger/cmd/messenger/main.go @@ -543,6 +543,9 @@ func answerProofs(h *Holder) { } } +// kindsEvery is how often the controller's word on the kinds is read again (a test bundle reads it faster). +var kindsEvery = 30 * time.Second + // followKinds reads, every half minute, which channel kinds the controller says exist and what each // promises: never from a channel. func followKinds(h *Holder) { @@ -565,7 +568,7 @@ func followKinds(h *Holder) { h.ReleaseHeld() // a held ask is judged again when the kinds change (ADR 0397 ยง2) said = "" } - time.Sleep(30 * time.Second) + time.Sleep(kindsEvery) } } diff --git a/modules/messenger/go.mod b/modules/messenger/go.mod index 2f961edb..4b62f792 100644 --- a/modules/messenger/go.mod +++ b/modules/messenger/go.mod @@ -1,5 +1,22 @@ module messenger -go 1.22 +go 1.25.0 -require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013182924-5933eb757d44 +require ( + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013182924-5933eb757d44 + github.com/nats-io/nats-server/v2 v2.11.17 + github.com/nats-io/nats.go v1.51.0 +) + +require ( + github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect + github.com/google/go-tpm v0.9.8 // indirect + github.com/klauspost/compress v1.18.5 // indirect + github.com/minio/highwayhash v1.0.4 // indirect + github.com/nats-io/jwt/v2 v2.8.1 // indirect + github.com/nats-io/nkeys v0.4.15 // indirect + github.com/nats-io/nuid v1.0.1 // indirect + golang.org/x/crypto v0.50.0 // indirect + golang.org/x/sys v0.43.0 // indirect + golang.org/x/time v0.15.0 // indirect +) diff --git a/modules/messenger/go.sum b/modules/messenger/go.sum index 3d45e21c..f84463ba 100644 --- a/modules/messenger/go.sum +++ b/modules/messenger/go.sum @@ -1,8 +1,27 @@ -git.novox.be/novox/mesh-sdk/go v0.1.13 h1:Su4JYpZ+zhNovkGA2DgxiZdcuHpjw2tPJzc/7PljhXg= -git.novox.be/novox/mesh-sdk/go v0.1.13/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013141520-9588871d6629 h1:7TYFV1Q6Mcz+lsLP/sds87tysET4rKCn3rRIQV8gNmA= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013141520-9588871d6629/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013145622-f4745447ba6d h1:HEJyqorKktrkJPeeJdwEo0kdrJjH3fm/YVS9MGftCqY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013145622-f4745447ba6d/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013182924-5933eb757d44 h1:r5xIo5qWOXQdCSMxvqJy8kV4U+P/QWXFK6971Ha6wU8= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013182924-5933eb757d44/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= +github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= +github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= +github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= +github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY= +github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE= +github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4= +github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ= +github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU= +github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg= +github.com/nats-io/nats-server/v2 v2.11.17 h1:GKEghcFK6A+aFx11Yf1LjgLC3txAwvyhnYzhBIQZA8I= +github.com/nats-io/nats-server/v2 v2.11.17/go.mod h1:B1sFVz4StNosQ903ak4N1G01Fl/9f8e06mXpFIE2K24= +github.com/nats-io/nats.go v1.51.0 h1:ByW84XTz6W03GSSsygsZcA+xgKK8vPGaa/FCAAEHnAI= +github.com/nats-io/nats.go v1.51.0/go.mod h1:26HypzazeOkyO3/mqd1zZd53STJN0EjCYF9Uy2ZOBno= +github.com/nats-io/nkeys v0.4.15 h1:JACV5jRVO9V856KOapQ7x+EY8Jo3qw1vJt/9Jpwzkk4= +github.com/nats-io/nkeys v0.4.15/go.mod h1:CpMchTXC9fxA5zrMo4KpySxNjiDVvr8ANOSZdiNfUrs= +github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= +github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= +golang.org/x/crypto v0.50.0 h1:zO47/JPrL6vsNkINmLoo/PH1gcxpls50DNogFvB5ZGI= +golang.org/x/crypto v0.50.0/go.mod h1:3muZ7vA7PBCE6xgPX7nkzzjiUq87kRItoJQM1Yo8S+Q= +golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.43.0 h1:Rlag2XtaFTxp19wS8MXlJwTvoh8ArU6ezoyFsMyCTNI= +golang.org/x/sys v0.43.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= +golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=