From 27ca8bbcff1013d9793c9a754e326e48de6f3b52 Mon Sep 17 00:00:00 2001 From: jochen Date: Sat, 17 Oct 2026 13:28:21 +0200 Subject: [PATCH] issue 975: the acceptance over a real bus: a ready mark and its Approve are merged within seconds of their events --- .../cmd/mesh-delivery/bus_test.go | 481 ++++++++++++++++++ modules/mesh-delivery/go.mod | 21 +- modules/mesh-delivery/go.sum | 27 +- 3 files changed, 526 insertions(+), 3 deletions(-) create mode 100644 modules/mesh-delivery/cmd/mesh-delivery/bus_test.go diff --git a/modules/mesh-delivery/cmd/mesh-delivery/bus_test.go b/modules/mesh-delivery/cmd/mesh-delivery/bus_test.go new file mode 100644 index 00000000..9d2f681f --- /dev/null +++ b/modules/mesh-delivery/cmd/mesh-delivery/bus_test.go @@ -0,0 +1,481 @@ +package main + +import ( + "bufio" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "os/exec" + "strings" + "sync" + "testing" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "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 acceptance of acting on events** (novox/hq ADR 0406, design 47 milestone 2, ADR 0391 phase 3, issue 975): a +// pull request with a passing verdict gets its `mesh/ready` label event and its `ready at ` comment event, and +// the merge is asked within five seconds of the comment event — with no tick, no poll and no reconcile to find it. +// Proved over a real bus: 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 a bundle — its state on the buckets the controller would make from the manifest, its +// events delivered from the bus's streams, the controller's verbs and the forge's tools answered by the test (the +// pattern of mesh-issues' bus_test). + +const bundleVar = "MESH_DELIVERY_TEST_BUNDLE" + +func TestMain(m *testing.M) { + if os.Getenv(bundleVar) == "1" { + main() + os.Exit(0) + } + os.Exit(m.Run()) +} + +type testBus struct { + nc *nats.Conn + js jetstream.JetStream + kvs map[string]jetstream.KeyValue +} + +func newTestBus(t *testing.T) *testBus { + t.Helper() + opts := &server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(), + NoLog: true, NoSigs: true} + s, err := server.NewServer(opts) + if err != nil { + t.Fatalf("a bus for this test could not be made: %v", err) + } + go s.Start() + if !s.ReadyForConnections(30 * time.Second) { + s.Shutdown() + t.Fatal("a bus for this test did not come up 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, err := jetstream.New(nc) + if err != nil { + t.Fatal(err) + } + b := &testBus{nc: nc, js: js, kvs: map[string]jetstream.KeyValue{}} + // The controller's job: a bucket per state the manifest declares. + m, _ := readManifest(t) + for _, name := range m.State { + kv, err := js.CreateKeyValue(context.Background(), jetstream.KeyValueConfig{Bucket: m.Module + "_" + name, History: 1}) + if err != nil { + t.Fatal(err) + } + b.kvs[name] = kv + } + // The streams the events are on (mesh.mod..event., mesh.seat..event.). + if _, err := js.CreateStream(context.Background(), jetstream.StreamConfig{Name: "EVENTS", + Subjects: []string{"mesh.mod.gitea.event.>", "mesh.seat." + ControllerSeat + ".event.>", "mesh.seat." + asks.Seat + ".event.>"}}); err != nil { + t.Fatal(err) + } + return b +} + +// runtime is the half of the conversation the tool runner plays for one running bundle. +type runtime struct { + t *testing.T + b *testBus + cmd *exec.Cmd + in io.WriteCloser + mu sync.Mutex + next int + wait map[string]chan json.RawMessage + + // what the bundle subscribed to, and asked of the mesh. + subs []string + asked []askedOfMesh + notify chan struct{} + // the asks the operator was asked, and the controller's walks and the forge's open pull requests as the test plays + // them. + operatorAsks []asks.Ask + openPulls []OpenPull +} + +type askedOfMesh struct { + Key string + Body map[string]any + At time.Time +} + +func (b *testBus) start(t *testing.T) *runtime { + t.Helper() + cmd := exec.Command(os.Args[0], "-test.run=^$") + cmd.Env = append(os.Environ(), bundleVar+"=1", "MESH_SERVED_MODULE=mesh-delivery", "MESH_MODULE=mesh-delivery", "MESH_NODE=test") + in, err := cmd.StdinPipe() + if err != nil { + t.Fatal(err) + } + out, err := cmd.StdoutPipe() + if err != nil { + t.Fatal(err) + } + cmd.Stderr = testLog{t} + if err := cmd.Start(); err != nil { + t.Fatal(err) + } + r := &runtime{t: t, b: b, cmd: cmd, in: in, wait: map[string]chan json.RawMessage{}, notify: make(chan struct{}, 64)} + go r.read(out) + t.Cleanup(func() { _ = in.Close(); _ = cmd.Wait() }) + return r +} + +type testLog struct{ t *testing.T } + +func (w testLog) Write(p []byte) (int, error) { + defer func() { _ = recover() }() // the bundle may speak after the test ended + w.t.Logf("bundle: %s", strings.TrimRight(string(p), "\n")) + return len(p), nil +} + +type rpcMessage struct { + ID json.RawMessage `json:"id,omitempty"` + Method string `json:"method,omitempty"` + Params json.RawMessage `json:"params,omitempty"` + Result json.RawMessage `json:"result,omitempty"` + Error *struct { + Code int `json:"code"` + Message string `json:"message"` + } `json:"error,omitempty"` +} + +func (r *runtime) 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')) +} + +func (r *runtime) read(out io.Reader) { + sc := bufio.NewScanner(out) + sc.Buffer(make([]byte, 1<<20), 1<<24) + for sc.Scan() { + var m rpcMessage + 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 <- sc.Bytes() + } + continue + } + go r.answer(m) + } +} + +func (r *runtime) ping() { + select { + case r.notify <- struct{}{}: + default: + } +} + +// asksOf are the asks of the mesh made so far under a key. +func (r *runtime) asksOf(key string) []askedOfMesh { + r.mu.Lock() + defer r.mu.Unlock() + var out []askedOfMesh + for _, a := range r.asked { + if a.Key == key { + out = append(out, a) + } + } + return out +} + +// until waits for what the test needs to be true, woken by every message of the bundle. +func (r *runtime) until(what string, within time.Duration, ok func() bool) { + r.t.Helper() + deadline := time.After(within) + for { + if ok() { + return + } + select { + case <-r.notify: + case <-time.After(50 * time.Millisecond): + case <-deadline: + if ok() { + return + } + r.t.Fatalf("within %s: %s did not happen", within, what) + } + } +} + +// toolAnswer is a tool's or a verb's answer as the runtime hands it over. +func toolAnswer(answer any) any { + return map[string]any{"ok": true, "output": "", "answer": answer} +} + +func (r *runtime) answer(m rpcMessage) { + var p struct { + State string `json:"state"` + Key string `json:"key"` + Value json.RawMessage `json:"value"` + Revision uint64 `json:"revision"` + Pattern string `json:"pattern"` + Subject string `json:"subject"` + Body json.RawMessage `json:"body"` + } + _ = json.Unmarshal(m.Params, &p) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + defer r.ping() + result, err := func() (any, error) { + switch m.Method { + case "mesh/subscribe": + r.mu.Lock() + r.subs = append(r.subs, p.Pattern) + r.mu.Unlock() + return map[string]any{}, nil + case "mesh/publish": + return map[string]any{}, nil // the module's own events: not needed here + case "mesh/seat.publish": + // The ask for the operator, published under the module's name. + if strings.Contains(p.Subject, ".accept.ask.") { + var a asks.Ask + if err := json.Unmarshal(p.Body, &a); err != nil { + return nil, err + } + r.mu.Lock() + r.operatorAsks = append(r.operatorAsks, a) + r.mu.Unlock() + } + return map[string]any{}, nil + case "mesh/seat.record": + return nil, nil + case "mesh/ask": + var a struct { + Key string `json:"key"` + Body map[string]any `json:"body"` + } + if err := json.Unmarshal(m.Params, &a); err != nil { + return nil, err + } + r.mu.Lock() + r.asked = append(r.asked, askedOfMesh{Key: a.Key, Body: a.Body, At: time.Now()}) + pulls := append([]OpenPull{}, r.openPulls...) + r.mu.Unlock() + switch a.Key { + case "seat:" + ControllerSeat + ".delivery-walks": + return toolAnswer(map[string]any{"held": true, "walks": []any{}}), nil + case "seat:" + ControllerSeat + ".delivery-plan": + return toolAnswer(map[string]any{"plan": DeliveryPlan{Repository: "novox/app", Moved: []string{"app"}, + Summary: "builds app, sent to the node"}}), nil + case "gitea.gitea_open_pulls": + return toolAnswer(map[string]any{"pulls": pulls}), nil + case "gitea.gitea_note_append", "gitea.gitea_delivery_view", "gitea.gitea_commit_status", "gitea.gitea_merge_pull_request", + "gitea.gitea_add_comment": + return toolAnswer(map[string]any{}), nil + } + return nil, fmt.Errorf("the runtime answers no ask %s in this test", a.Key) + } + kv := r.b.kvs[p.State] + if kv == nil { + return nil, fmt.Errorf("the state %q is not declared", p.State) + } + switch m.Method { + case "mesh/state.get": + e, err := kv.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.create": + rev, err := kv.Create(ctx, p.Key, p.Value) + return map[string]any{"revision": rev}, err + case "mesh/state.update": + rev, err := kv.Update(ctx, p.Key, p.Value, p.Revision) + return map[string]any{"revision": rev}, err + case "mesh/state.put": + rev, err := kv.Put(ctx, p.Key, p.Value) + return map[string]any{"revision": rev}, err + case "mesh/state.delete": + return map[string]any{}, kv.Delete(ctx, p.Key) + case "mesh/state.keys": + keys, err := kv.Keys(ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + return []string{}, nil + } + return keys, err + } + return nil, fmt.Errorf("the runtime offers no %s to this bundle", m.Method) + }() + if err != nil { + code := -32000 + var api *jetstream.APIError + if errors.Is(err, jetstream.ErrKeyExists) || errors.As(err, &api) && api.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequence { + code = -32010 + } + r.say(map[string]any{"id": m.ID, "error": map[string]any{"code": code, "message": err.Error()}}) + return + } + r.say(map[string]any{"id": m.ID, "result": result}) +} + +// deliverEvents is the runtime reading the events streams and handing each event to the bundle, once, as +// `mesh/event`, the way the tool runner does for a module's consumer. +func (r *runtime) deliverEvents(ctx context.Context) { + stream, err := r.b.js.Stream(ctx, "EVENTS") + if err != nil { + r.t.Fatal(err) + } + cons, err := stream.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{}) + if err != nil { + r.t.Fatal(err) + } + cc, err := cons.Consume(func(msg jetstream.Msg) { + key := keyOfSubject(msg.Subject()) + r.mu.Lock() + id := fmt.Sprintf("event-%d", r.next) + r.next++ + r.mu.Unlock() + r.say(map[string]any{"id": id, "method": "mesh/event", "params": map[string]any{"envelope": stdio.Envelope{ + Key: key, Node: "test", Body: msg.Data(), Headers: map[string]string{"x-source": strings.SplitN(key, ".", 2)[0]}}}}) + }) + if err != nil { + r.t.Fatal(err) + } + r.t.Cleanup(cc.Stop) +} + +// keyOfSubject is an event's key, as the runtime gives it to a module: mesh.mod..event. is +// ., and mesh.seat..event. is .. +func keyOfSubject(subject string) string { + parts := strings.Split(subject, ".") + if len(parts) < 5 || parts[3] != "event" { + return subject + } + return parts[2] + "." + strings.Join(parts[4:], ".") +} + +func (b *testBus) publish(t *testing.T, subject string, body any) time.Time { + t.Helper() + raw, err := json.Marshal(body) + if err != nil { + t.Fatal(err) + } + if _, err := b.js.Publish(context.Background(), subject, raw); err != nil { + t.Fatal(err) + } + return time.Now() +} + +func TestAReadyMarkAndItsApproveAreMergedWithinSecondsOfTheirEventsOverARealBus(t *testing.T) { + b := newTestBus(t) + rt := b.start(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + rt.deliverEvents(ctx) + // The module has started when it hears what it follows, and the operator's answers. + rt.until("the module to subscribe to the events it follows", 20*time.Second, func() bool { + rt.mu.Lock() + defer rt.mu.Unlock() + return len(rt.subs) >= len(followed)+1 + }) + + // A pull request into main is announced, and the controller's verdict on its head passes. + const head = "1111111111111111111111111111111111111111" + announced := pr("novox/app", 7, head, "fix/x", "src/a.go") + announced.Title = "Fix the apostrophe (issue 975)" + b.publish(t, "mesh.mod.gitea.event.pull.updated", announced) + b.publish(t, "mesh.seat."+ControllerSeat+".event.checked", judgedAt("novox/app", 7, head, "pass", "t1t1t1t1t1t1")) + rt.until("the module to keep the delivery of the head as ready", 10*time.Second, func() bool { + keys, err := b.kvs["deliveries"].Keys(context.Background()) + if err != nil { + return false + } + for _, k := range keys { + e, err := b.kvs["deliveries"].Get(context.Background(), k) + var d Delivery + if err == nil && json.Unmarshal(e.Value(), &d) == nil && d.State == Ready && d.Commit == head { + return true + } + } + return false + }) + reads := len(rt.asksOf("gitea.gitea_open_pulls")) + + // The reviewer's label and comment: the forge shows the label alone at first, and the mark whole once the comment is + // written. Nothing is asked for the label alone. + rt.mu.Lock() + rt.openPulls = []OpenPull{markedPull("novox/app", 7, head, "")} + rt.mu.Unlock() + b.publish(t, "mesh.mod.gitea.event.forge.pull-request.labeled", labeled(7, ReadyLabel)) + rt.until("the forge's open pull requests to be read for the label", 5*time.Second, func() bool { + return len(rt.asksOf("gitea.gitea_open_pulls")) > reads + }) + rt.mu.Lock() + rt.openPulls = []OpenPull{markedPull("novox/app", 7, head, head)} + nothing := len(rt.operatorAsks) + rt.mu.Unlock() + if nothing != 0 { + t.Fatalf("the operator was asked for a label with no head named") + } + commented := b.publish(t, "mesh.mod.gitea.event.forge.pull-request.commented", commentedReady(7, head)) + + // The operator is asked, within the comment's event, and answers at once (a person's Approve, proven by the router). + rt.until("the operator to be asked to Approve the merge", 5*time.Second, func() bool { + rt.mu.Lock() + defer rt.mu.Unlock() + return len(rt.operatorAsks) == 1 + }) + rt.mu.Lock() + ask := rt.operatorAsks[0] + rt.mu.Unlock() + opt, _ := ask.Option(OptionApprove) + b.publish(t, "mesh.seat."+asks.Seat+".event.decided."+Module, asks.Warrant{Ask: ask.ID, Asker: Module, + Outcome: asks.OutcomeChosen, Option: OptionApprove, Label: opt.Label, Level: opt.Level, + By: &asks.Person{Who: asks.Operator, Kind: "telegram", Verified: "user id verified"}, At: time.Now(), AskDigest: ask.Digest()}) + + rt.until("the merge to be asked of the forge", 5*time.Second, func() bool { return len(rt.asksOf("gitea.gitea_merge_pull_request")) >= 1 }) + merges := rt.asksOf("gitea.gitea_merge_pull_request") + if len(merges) != 1 || merges[0].Body["head"] != head || merges[0].Body["number"] != float64(7) { + t.Fatalf("the merge was asked %+v; wanted once, for pull request 7 at the marked head", merges) + } + if took := merges[0].At.Sub(commented); took > 5*time.Second { + t.Fatalf("the merge was asked %s after the ready comment's event; wanted within 5s", took) + } else { + t.Logf("the merge was asked %s after the ready comment's event", took) + } + // The forge's open pull requests were read for the mark by the events, not by a poll: the label's and the comment's. + if got := len(rt.asksOf("gitea.gitea_open_pulls")) - reads; got < 2 || got > 3 { + t.Fatalf("the forge's open pull requests were read %d times for one mark", got) + } + // Quiet follows: nothing asks the forge or the controller again with no event to say so. + quiet := len(rt.asked) + time.Sleep(1500 * time.Millisecond) + rt.mu.Lock() + more := len(rt.asked) - quiet + rt.mu.Unlock() + if more > 3 { + t.Fatalf("%d asks of the mesh in a quiet second and a half after the merge: something polls", more) + } +} diff --git a/modules/mesh-delivery/go.mod b/modules/mesh-delivery/go.mod index 8d6748fb..5df693f9 100644 --- a/modules/mesh-delivery/go.mod +++ b/modules/mesh-delivery/go.mod @@ -1,5 +1,22 @@ module mesh-delivery -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/mesh-delivery/go.sum b/modules/mesh-delivery/go.sum index 3c290b47..f84463ba 100644 --- a/modules/mesh-delivery/go.sum +++ b/modules/mesh-delivery/go.sum @@ -1,2 +1,27 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013182924-5933eb757d44/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=