mesh-catalog/modules/mesh-delivery/cmd/mesh-delivery/bus_test.go

//go:build integration

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; ADR
// 0421, issue 1022): a pull request whose verdict passes is asked of the operator within seconds of the verdict's event,
// with no label, comment or mark written by anyone, and the merge is asked within five seconds of the Approve's 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
	// operatorCancels are the ids of the asks taken back from the router, in order (issue 1079).
	operatorCancels []string
	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()
			}
			if strings.Contains(p.Subject, ".accept.cancel.") {
				var c struct {
					ID string `json:"id"`
				}
				if err := json.Unmarshal(p.Body, &c); err != nil {
					return nil, err
				}
				r.mu.Lock()
				r.operatorCancels = append(r.operatorCancels, c.ID)
				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 "seat:" + ControllerSeat + ".delivery-check":
				return toolAnswer(map[string]any{"asked": "check-1"}), 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 TestAPassingVerdictAsksTheApproveAndItsApproveMergesWithinSecondsOverARealBus(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
	})
	for _, k := range followed {
		if strings.Contains(k, "labeled") || strings.Contains(k, "commented") {
			t.Fatalf("the module follows %s: a label or a comment starts nothing here (ADR 0421 §4)", k)
		}
	}

	// 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.BaseSHA = "t1t1t1t1t1t1"
	announced.Title = "Fix the apostrophe (issue 975)"
	announced.Body = "The description, with no review line yet."
	b.publish(t, "mesh.mod.gitea.event.pull.updated", announced)
	passed := b.publish(t, "mesh.seat."+ControllerSeat+".event.checked", judgedAt("novox/app", 7, head, "pass", "t1t1t1t1t1t1"))

	// The operator is asked, within seconds of the verdict's event, with no label, comment or mark written by anyone.
	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]
	asked := time.Now()
	rt.mu.Unlock()
	if !strings.Contains(ask.Explanation, "not reviewed at this head") {
		t.Fatalf("the first ask does not say the record is missing: %s", ask.Explanation)
	}
	t.Logf("the operator was asked %s after the verdict's event", asked.Sub(passed))
	// The description is edited to add the review record (issue 1079): the forge module's edit event carries it. The unanswered
	// ask is withdrawn from the router and asked again, once, with the record read; nothing else is asked of the mesh.
	first := ask
	b.publish(t, "mesh.mod.gitea.event.forge.pull-request.edited", map[string]any{"owner": "novox", "repo": "app", "number": 7,
		"title": announced.Title, "head_sha": head, "base": "main", "body": "Reviewed-by: lead@shanks at " + head})
	rt.until("the first ask to be withdrawn and the operator asked again", 5*time.Second, func() bool {
		rt.mu.Lock()
		defer rt.mu.Unlock()
		return len(rt.operatorCancels) >= 1 && len(rt.operatorAsks) >= 2
	})
	rt.mu.Lock()
	if len(rt.operatorCancels) != 1 || rt.operatorCancels[0] != first.ID || len(rt.operatorAsks) != 2 {
		t.Fatalf("withdrawn %v, asked %d; wanted the first ask %s withdrawn once and exactly one new ask", rt.operatorCancels, len(rt.operatorAsks), first.ID)
	}
	ask = rt.operatorAsks[1]
	rt.mu.Unlock()
	if ask.ID == first.ID || !strings.Contains(ask.Explanation, "reviewed by lead on shanks") || strings.Contains(ask.Explanation, "not reviewed") {
		t.Fatalf("the new ask is not a fresh one naming the reviewer: %s %s", ask.ID, ask.Explanation)
	}
	if len(rt.asksOf("gitea.gitea_merge_pull_request")) != 0 {
		t.Fatalf("merged before the Approve")
	}
	// The pull request's key holds its head, the trunk commit it was asked against and the verdict.
	rt.until("the pull request's key to hold head, trunk and verdict", 5*time.Second, func() bool {
		e, err := b.kvs["pulls"].Get(context.Background(), "novox.app.7")
		var k PullKey
		return err == nil && json.Unmarshal(e.Value(), &k) == nil && k == PullKey{Head: head, Trunk: "t1t1t1t1t1t1", Verdict: "pass"}
	})

	// The operator approves (a person's Approve, proven by the router): the merge is asked within seconds.
	opt, _ := ask.Option(OptionApprove)
	approved := 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 approved head", merges)
	}
	if got := fmt.Sprint(merges[0].Body["approved_heads"]); got != "[novox/app@"+head+"]" {
		t.Fatalf("the merge tool was handed the approved heads %s", got)
	}
	if took := merges[0].At.Sub(approved); took > 5*time.Second {
		t.Fatalf("the merge was asked %s after the Approve's event; wanted within 5s", took)
	} else {
		t.Logf("the merge was asked %s after the Approve's event", took)
	}
	// 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)
	}
}

// **The acceptance of one group, one Approve** (novox/hq ADR 0418 §4, ADR 0421, design 47 milestone 3, issues 978 and
// 1022): a group of two pull requests on one branch name, in two repositories, with green checks, gets one ask of the
// operator — no label, comment or mark written by anyone — and on the Approve both merge in one pass, each handed the
// approved heads of both.
func TestAGroupOfTwoWithGreenChecksGetsOneAskAndOnItsApproveBothMergeInOnePassOverARealBus(t *testing.T) {
	b := newTestBus(t)
	rt := b.start(t)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	rt.deliverEvents(ctx)
	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
	})
	const ctlHead, catHead = "c1c1c1c1c1c1c1c1c1c1c1c1c1c1c1c1c1c1c1c1", "d2d2d2d2d2d2d2d2d2d2d2d2d2d2d2d2d2d2d2d2"
	for _, p := range []PullEvent{pr("novox/mesh-controller", 1, ctlHead, "feat/x", "internal/a.go"),
		pr("novox/mesh-catalog", 2, catHead, "feat/x", "modules/app/module.json")} {
		p.BaseSHA = "t1t1t1t1t1t1"
		p.Body = "Reviewed-by: lead@shanks at " + p.HeadSHA
		b.publish(t, "mesh.mod.gitea.event.pull.updated", p)
	}
	b.publish(t, "mesh.seat."+ControllerSeat+".event.checked", judgedAt("novox/mesh-controller", 1, ctlHead, "pass", "t1t1t1t1t1t1"))
	b.publish(t, "mesh.seat."+ControllerSeat+".event.checked", judgedAt("novox/mesh-catalog", 2, catHead, "pass", "t1t1t1t1t1t1"))

	// Both members green: the composed check is asked, once, for both.
	// Each member's own check is asked at once (issue 1115), before the group's: wait for the one carrying the group.
	rt.until("the composed check to be asked", 10*time.Second, func() bool {
		for _, c := range rt.asksOf("seat:" + ControllerSeat + ".delivery-check") {
			if c.Body["group"] == "feat/x" {
				return true
			}
		}
		return false
	})
	if n := len(rt.asksOf("seat:" + ControllerSeat + ".delivery-order")); n != 0 {
		t.Fatalf("the controller's delivery-order was asked %d times: no order is asked for", n)
	}
	rt.mu.Lock()
	nothing := len(rt.operatorAsks)
	rt.mu.Unlock()
	if nothing != 0 {
		t.Fatalf("the operator was asked before the group's composed check passed")
	}
	// The composed check passes at both heads: the operator is asked once, for the group.
	b.publish(t, "mesh.seat."+ControllerSeat+".event.checked", map[string]any{"group": "feat/x", "id": "check-1", "verdict": "pass",
		"summary": "both compose", "members": []map[string]any{
			{"owner": "novox", "repo": "mesh-controller", "commit": ctlHead, "trunk": "t1t1t1t1t1t1"},
			{"owner": "novox", "repo": "mesh-catalog", "commit": catHead, "trunk": "t1t1t1t1t1t1"}}})
	rt.until("the operator to be asked once for the group", 10*time.Second, func() bool {
		rt.mu.Lock()
		defer rt.mu.Unlock()
		return len(rt.operatorAsks) == 1
	})
	time.Sleep(500 * time.Millisecond)
	rt.mu.Lock()
	asked, ask := len(rt.operatorAsks), rt.operatorAsks[0]
	rt.mu.Unlock()
	if asked != 1 {
		t.Fatalf("the operator was asked %d times for one group", asked)
	}
	if len(rt.asksOf("gitea.gitea_merge_pull_request")) != 0 {
		t.Fatalf("merged before the Approve")
	}
	opt, _ := ask.Option(OptionApprove)
	approved := 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()})

	// Both merge on that one Approve, in one pass: asked within seconds of the Approve's event, each with both heads.
	rt.until("both merges to be asked of the forge", 5*time.Second, func() bool { return len(rt.asksOf("gitea.gitea_merge_pull_request")) >= 2 })
	merges := rt.asksOf("gitea.gitea_merge_pull_request")
	if len(merges) != 2 {
		t.Fatalf("the merge was asked %d times; wanted once for each member: %+v", len(merges), merges)
	}
	heads := map[string]bool{}
	for _, m := range merges {
		heads[fmt.Sprint(m.Body["repo"], "@", m.Body["head"])] = true
		if got := fmt.Sprint(m.Body["approved_heads"]); got != "[novox/mesh-catalog@"+catHead+" novox/mesh-controller@"+ctlHead+"]" {
			t.Fatalf("a merge was handed the approved heads %s", got)
		}
		if took := m.At.Sub(approved); took > 5*time.Second {
			t.Fatalf("a merge was asked %s after the Approve's event; wanted within 5s", took)
		}
	}
	if !heads["mesh-controller@"+ctlHead] || !heads["mesh-catalog@"+catHead] {
		t.Fatalf("the merges were asked for %v", heads)
	}
	rt.mu.Lock()
	again := len(rt.operatorAsks)
	rt.mu.Unlock()
	if again != 1 {
		t.Fatalf("a second ask was made: %d", again)
	}
}

// **The acceptance of a failed backup before the send, on the mesh-delivery side** (novox/hq ADR 0404 clause 4, design 47
// milestone 4, issue 1068): a merged delivery whose walk runs is held when the walk's record, as the controller's
// `plan-moved` carries it, names a backup that failed and nothing was sent; the hold names the machine, the module and the
// backup's words, and nothing asks the controller to let it go. One test function, one bus. The walk's backups field is the controller's
// contract (issue 1065).
func TestAPlanMovedCarryingAFailedBackupHoldsTheDeliveryWithTheBackupsWordsOverARealBus(t *testing.T) {
	b := newTestBus(t)
	rt := b.start(t)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	rt.deliverEvents(ctx)
	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
	})
	const head, merge = "3333333333333333333333333333333333333333", "4444444444444444444444444444444444444444"
	announced := pr("novox/app", 9, head, "fix/y", "src/a.go")
	announced.BaseSHA = "t1t1t1t1t1t1"
	deliveryOf := func() Delivery {
		e, err := b.kvs["deliveries"].Get(context.Background(), kvKey(IDOf("novox/app", head)))
		var d Delivery
		if err == nil {
			_ = json.Unmarshal(e.Value(), &d)
		}
		return d
	}
	// Each event is published after the module has taken the one before it: they are on two subjects, and the runner
	// may hand over a later one first, which the table would rightly judge on its own (merged before checked).
	b.publish(t, "mesh.mod.gitea.event.pull.updated", announced)
	rt.until("the pull request to be taken", 10*time.Second, func() bool { return deliveryOf().State != "" })
	b.publish(t, "mesh.seat."+ControllerSeat+".event.checked", judgedAt("novox/app", 9, head, "pass", "t1t1t1t1t1t1"))
	rt.until("the verdict to be taken", 10*time.Second, func() bool { d := deliveryOf(); return d.State != "" && d.State != Proposed })
	b.publish(t, "mesh.mod.gitea.event.pull.merged", merged("novox/app", 9, head, merge))
	rt.until("the merge to be taken", 10*time.Second, func() bool { return deliveryOf().MergedAs != "" })

	// Its walk runs, let go: the record names no failed backup yet.
	now := time.Now()
	walk := aWalk("plan-1", "novox/app", merge, true)
	walk.Created, walk.State, walk.Revision = now, walkRolling, 1
	walk.Delivery.Go, walk.Delivery.By = &now, "mesh-delivery"
	b.publish(t, "mesh.seat."+ControllerSeat+".event.plan-moved", walk)
	rt.until("the delivery to be delivering", 10*time.Second, func() bool { return deliveryOf().State == Delivering })

	// A backup failed before the send: the delivery is held with the backup's words.
	walk.Revision, walk.Backups = 2, &WalkBackups{State: "failed", Asked: []WalkBackupAsked{{Machine: "ace", Module: "pg", State: "failed", Words: "no space left on device"}}}
	b.publish(t, "mesh.seat."+ControllerSeat+".event.plan-moved", walk)
	rt.until("the delivery to be held for the failed backup", 10*time.Second, func() bool { return deliveryOf().State == Held })
	d := deliveryOf()
	for _, want := range []string{"ace", "pg", "no space left on device"} {
		if !strings.Contains(d.HeldWhy, want) {
			t.Fatalf("the held reason %q does not name %q", d.HeldWhy, want)
		}
	}
	// It stays held, and the controller is asked nothing to let it go.
	time.Sleep(500 * time.Millisecond)
	if s := deliveryOf().State; s != Held {
		t.Fatalf("the hold lifted by itself: %s", s)
	}
	if n := len(rt.asksOf("seat:"+ControllerSeat+".deliver")) + len(rt.asksOf("seat:"+ControllerSeat+".delivery-backups-again")); n != 0 {
		t.Fatalf("the controller was asked %d times to let the held walk go or take its backups again", n)
	}
}