issue 975: the acceptance over a real bus: a ready mark and its Approve are merged within seconds of their events

This commit is contained in:
2026-10-17 13:28:21 +02:00
parent 0809a1f982
commit 27ca8bbcff
3 changed files with 526 additions and 3 deletions
@@ -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 <head>` 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.<module>.event.<name>, mesh.seat.<seat>.event.<name>).
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.<module>.event.<name> is
// <module>.<name>, and mesh.seat.<seat>.event.<name> is <seat>.<name>.
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)
}
}
+19 -2
View File
@@ -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
)
+26 -1
View File
@@ -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=