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

package main

import (
	"bufio"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"os"
	"os/exec"
	"path/filepath"
	"sort"
	"strconv"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/nats-io/nats-server/v2/server"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
)

// The acceptance (novox/hq to-be 53): "an agent opens an issue and gets the next number; two agents opening
// at the same moment never get the same one" — 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, as many
// instances as a test asks, each speaking MCP on its standard streams to a runtime played here, which does
// what the tool runner does with a bundle's state and events: each state verb on the bucket the controller
// would have made from the manifest, a lost compare-and-set answered with its code (-32010).

const bundleVar = "MESH_ISSUES_TEST_BUNDLE"

func TestMain(m *testing.M) {
	if os.Getenv(bundleVar) == "1" {
		main()
		os.Exit(0)
	}
	os.Exit(m.Run())
}

// bus is one test's JetStream server and the buckets the controller makes from the manifest.
type bus struct {
	// lost counts the compare-and-sets another writer won: proof the openers collided.
	lost atomic.Int64
	// counterLost counts only the counter's updates another opener won: proof the counter itself was contended.
	counterLost atomic.Int64
	nc          *nats.Conn
	js          jetstream.JetStream
	kvs         map[string]jetstream.KeyValue
}

func newBus(t *testing.T) *bus {
	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 := &bus{nc: nc, js: js, kvs: map[string]jetstream.KeyValue{}}
	// The controller's job: a bucket per declared state, with the options the manifest gives it.
	m, _ := readManifest(t)
	for _, d := range m.stateDeclarations(t) {
		history := uint8(1)
		if d.History > 0 {
			history = uint8(d.History)
		}
		kv, err := js.CreateKeyValue(context.Background(), jetstream.KeyValueConfig{
			Bucket: m.Module + "_" + d.Name, History: history})
		if err != nil {
			t.Fatal(err)
		}
		b.kvs[d.Name] = kv
	}
	return b
}

// instance is one running bundle, and the runtime's half of its conversation.
type instance struct {
	t      *testing.T
	b      *bus
	cmd    *exec.Cmd
	in     io.WriteCloser
	mu     sync.Mutex
	next   int
	wait   map[string]chan json.RawMessage
	events *[]event
	evMu   *sync.Mutex
}

// event is one event a bundle asked the runtime to publish.
type event struct {
	Key     string            `json:"key"`
	Body    map[string]any    `json:"body"`
	Headers map[string]string `json:"headers"`
}

func (b *bus) start(t *testing.T, settings string, events *[]event, evMu *sync.Mutex) *instance {
	t.Helper()
	cmd := exec.Command(os.Args[0], "-test.run=^$")
	cmd.Env = append(os.Environ(), bundleVar+"=1", "MESH_ISSUES_SETTINGS="+settings,
		"MESH_SERVED_MODULE=mesh-issues", "MESH_MODULE=mesh-issues", "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 = testWriter{t}
	if err := cmd.Start(); err != nil {
		t.Fatal(err)
	}
	i := &instance{t: t, b: b, cmd: cmd, in: in, wait: map[string]chan json.RawMessage{}, events: events, evMu: evMu}
	go i.read(out)
	t.Cleanup(func() { _ = in.Close(); _ = cmd.Wait() })
	return i
}

type testWriter struct{ t *testing.T }

func (w testWriter) Write(p []byte) (int, error) { w.t.Logf("bundle: %s", p); return len(p), nil }

type rpc 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 (i *instance) say(m map[string]any) {
	m["jsonrpc"] = "2.0"
	raw, _ := json.Marshal(m)
	i.mu.Lock()
	defer i.mu.Unlock()
	_, _ = i.in.Write(append(raw, '\n'))
}

func (i *instance) read(out io.Reader) {
	sc := bufio.NewScanner(out)
	sc.Buffer(make([]byte, 1<<20), 1<<24)
	for sc.Scan() {
		var m rpc
		if json.Unmarshal(sc.Bytes(), &m) != nil {
			continue
		}
		if m.Method == "" {
			var id string
			_ = json.Unmarshal(m.ID, &id)
			i.mu.Lock()
			ch := i.wait[id]
			delete(i.wait, id)
			i.mu.Unlock()
			if ch != nil {
				ch <- sc.Bytes()
			}
			continue
		}
		go i.answer(m)
	}
}

// answer is the runtime's answer to what the bundle asked: its state, and its events.
func (i *instance) answer(m rpc) {
	var p struct {
		State    string          `json:"state"`
		Key      string          `json:"key"`
		Value    json.RawMessage `json:"value"`
		Revision uint64          `json:"revision"`
	}
	_ = json.Unmarshal(m.Params, &p)
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	result, err := func() (any, error) {
		if m.Method == "mesh/publish" {
			var e event
			if err := json.Unmarshal(m.Params, &e); err != nil {
				return nil, err
			}
			i.evMu.Lock()
			*i.events = append(*i.events, e)
			i.evMu.Unlock()
			return map[string]any{}, nil
		}
		if m.Method == "mesh/ask" {
			// The records module, as a reservation asks it what numbers hq holds (issue 424).
			var a struct {
				Key  string `json:"key"`
				Body struct {
					Folder string `json:"folder"`
				} `json:"body"`
			}
			if err := json.Unmarshal(m.Params, &a); err != nil || a.Key != "records.records_list" {
				return nil, fmt.Errorf("the runtime answers no ask %s in this test", a.Key)
			}
			listing := hqToday()
			return map[string]any{"folders": listing.folders[a.Body.Folder], "documents": listing.docs[a.Body.Folder]}, nil
		}
		kv := i.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":
			if p.Revision == 0 {
				return nil, errors.New("an update names the revision it read")
			}
			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.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
			i.b.lost.Add(1)
			if m.Method == "mesh/state.update" && p.State == CountersState {
				i.b.counterLost.Add(1)
			}
		}
		i.say(map[string]any{"id": m.ID, "error": map[string]any{"code": code, "message": err.Error()}})
		return
	}
	if result == nil {
		i.say(map[string]any{"id": m.ID, "result": nil})
		return
	}
	i.say(map[string]any{"id": m.ID, "result": result})
}

// call is one tools/call, as the tool runner makes it, and its answer: the result, or the refusal's words.
func (i *instance) call(verb string, args map[string]any) (json.RawMessage, error) {
	i.mu.Lock()
	i.next++
	id := "runtime-" + strconv.Itoa(i.next)
	ch := make(chan json.RawMessage, 1)
	i.wait[id] = ch
	i.mu.Unlock()
	i.say(map[string]any{"id": id, "method": "tools/call", "params": map[string]any{"name": Seat + "." + verb, "arguments": args}})
	select {
	case raw := <-ch:
		var m struct {
			Result struct {
				Content []struct {
					Text string `json:"text"`
				} `json:"content"`
				IsError bool `json:"isError"`
			} `json:"result"`
		}
		if err := json.Unmarshal(raw, &m); err != nil || len(m.Result.Content) == 0 {
			return nil, fmt.Errorf("an answer that is not one: %s", raw)
		}
		if m.Result.IsError {
			return nil, errors.New(m.Result.Content[0].Text)
		}
		return json.RawMessage(m.Result.Content[0].Text), nil
	case <-time.After(60 * time.Second):
		return nil, fmt.Errorf("%s did not answer in 60 s", verb)
	}
}

func writeSettings(t *testing.T, first int) string {
	t.Helper()
	path := filepath.Join(t.TempDir(), "settings.json")
	raw, _ := json.Marshal(map[string]any{"first-number": first, "owners": []string{"hq", "mesh-controller"}})
	if err := os.WriteFile(path, raw, 0o600); err != nil {
		t.Fatal(err)
	}
	return path
}

// Two instances on one bus, fifty opens at the same moment: fifty numbers, each once, consecutive from the
// first, and an issue and an event for each. With the counter written without its revision check, numbers
// are handed out twice and this fails.
func TestFiftyOpensAtOnceFromTwoInstancesGetFiftyDistinctConsecutiveNumbers(t *testing.T) {
	b := newBus(t)
	settings := writeSettings(t, 365)
	var events []event
	var evMu sync.Mutex
	instances := []*instance{b.start(t, settings, &events, &evMu), b.start(t, settings, &events, &evMu)}
	const opens = 50
	numbers := make([]uint64, opens)
	var wg sync.WaitGroup
	start := make(chan struct{})
	for k := 0; k < opens; k++ {
		wg.Add(1)
		go func(k int) {
			defer wg.Done()
			<-start
			raw, err := instances[k%2].call("open", map[string]any{"title": fmt.Sprintf("symptom %d", k), "report": "seen at once"})
			if err != nil {
				t.Errorf("open %d: %v", k, err)
				return
			}
			var o Opened
			_ = json.Unmarshal(raw, &o)
			numbers[k] = o.Number
		}(k)
	}
	close(start)
	wg.Wait()
	if t.Failed() {
		return
	}
	sorted := append([]uint64(nil), numbers...)
	sort.Slice(sorted, func(i, j int) bool { return sorted[i] < sorted[j] })
	for k, n := range sorted {
		if n != uint64(365+k) {
			t.Fatalf("the numbers handed out are %v: not fifty distinct numbers from 365 in a row", sorted)
		}
	}
	// The openers collided: writes were refused because another opener had moved the counter first.
	t.Logf("%d writes lost to another opener and tried again, %d of them the counter's updates", b.lost.Load(), b.counterLost.Load())
	if b.counterLost.Load() < opens {
		t.Fatalf("only %d of the counter's updates were lost to another opener: the test did not make the openers "+
			"contend for the counter", b.counterLost.Load())
	}
	ctx := context.Background()
	e, err := b.kvs[CountersState].Get(ctx, IssueCounter)
	if err != nil || string(e.Value()) != "414" {
		t.Fatalf("the counter holds %v (%v), want 414", e, err)
	}
	keys, _ := b.kvs[IssuesState].Keys(ctx)
	if len(keys) != opens {
		t.Fatalf("%d issues kept, want %d", len(keys), opens)
	}
	for k, n := range numbers {
		e, err := b.kvs[IssuesState].Get(ctx, keyOf(n))
		if err != nil {
			t.Fatal(err)
		}
		var is Issue
		_ = json.Unmarshal(e.Value(), &is)
		if is.Title != fmt.Sprintf("symptom %d", k) || is.Number != n {
			t.Fatalf("issue %d holds %+v, not open %d's", n, is, k)
		}
	}
	evMu.Lock()
	defer evMu.Unlock()
	said := map[float64]bool{}
	for _, ev := range events {
		if ev.Key != EventOpened || ev.Headers["x-source"] != "mesh-issues" {
			t.Fatalf("event %+v", ev)
		}
		said[ev.Body["number"].(float64)] = true
	}
	if len(said) != opens {
		t.Fatalf("%d opened events for distinct numbers, want %d", len(said), opens)
	}
}

// Two instances on one bus, forty decision numbers reserved at the same moment: forty numbers, each once,
// consecutive from one above the highest decision hq holds, a reservation and an event for each (issue 424).
func TestFortyReservationsAtOnceFromTwoInstancesGetFortyDistinctConsecutiveNumbers(t *testing.T) {
	b := newBus(t)
	settings := writeSettings(t, 365)
	var events []event
	var evMu sync.Mutex
	instances := []*instance{b.start(t, settings, &events, &evMu), b.start(t, settings, &events, &evMu)}
	const reserves = 40
	numbers := make([]uint64, reserves)
	var wg sync.WaitGroup
	start := make(chan struct{})
	for k := 0; k < reserves; k++ {
		wg.Add(1)
		go func(k int) {
			defer wg.Done()
			<-start
			raw, err := instances[k%2].call("reserve", map[string]any{"kind": "decision", "for": fmt.Sprintf("record %d", k)})
			if err != nil {
				t.Errorf("reserve %d: %v", k, err)
				return
			}
			var r struct {
				Kind   string `json:"kind"`
				Number uint64 `json:"number"`
			}
			_ = json.Unmarshal(raw, &r)
			if r.Kind != "decision" {
				t.Errorf("reserve %d answered %s", k, raw)
			}
			numbers[k] = r.Number
		}(k)
	}
	close(start)
	wg.Wait()
	if t.Failed() {
		return
	}
	sorted := append([]uint64(nil), numbers...)
	sort.Slice(sorted, func(i, j int) bool { return sorted[i] < sorted[j] })
	for k, n := range sorted {
		if n != uint64(292+k) {
			t.Fatalf("the numbers handed out are %v: not forty distinct numbers from 292 in a row", sorted)
		}
	}
	t.Logf("%d writes lost to another caller and tried again, %d of them the counters' updates", b.lost.Load(), b.counterLost.Load())
	if b.counterLost.Load() == 0 {
		t.Fatal("no counter update was lost to another caller: the test did not make the reservations contend")
	}
	ctx := context.Background()
	if e, err := b.kvs[CountersState].Get(ctx, KindDecision); err != nil || string(e.Value()) != "331" {
		t.Fatalf("the decision counter holds %v (%v), want 331", e, err)
	}
	if _, err := b.kvs[CountersState].Get(ctx, IssueCounter); !errors.Is(err, jetstream.ErrKeyNotFound) {
		t.Fatalf("a reservation touched the issue counter: %v", err)
	}
	for k, n := range numbers {
		e, err := b.kvs[ReservationsState].Get(ctx, ReservationKey(KindDecision, n))
		if err != nil {
			t.Fatal(err)
		}
		var r Reservation
		_ = json.Unmarshal(e.Value(), &r)
		if r.Number != n || r.For != fmt.Sprintf("record %d", k) || r.By != Unproven {
			t.Fatalf("reservation %d holds %+v, not reserve %d's", n, r, k)
		}
	}
	evMu.Lock()
	defer evMu.Unlock()
	said := map[float64]bool{}
	for _, ev := range events {
		if ev.Key != EventReserved || ev.Headers["x-source"] != "mesh-issues" {
			t.Fatalf("event %+v", ev)
		}
		said[ev.Body["number"].(float64)] = true
	}
	if len(said) != reserves {
		t.Fatalf("%d reserved events for distinct numbers, want %d", len(said), reserves)
	}
}

// One instance, every verb over the bus: what an agent does with an issue from opening to resolving.
func TestAnIssueIsOpenedNotedLinkedMovedAndListedOverTheBus(t *testing.T) {
	b := newBus(t)
	var events []event
	var evMu sync.Mutex
	i := b.start(t, writeSettings(t, 365), &events, &evMu)
	must := func(raw json.RawMessage, err error) json.RawMessage {
		t.Helper()
		if err != nil {
			t.Fatal(err)
		}
		return raw
	}
	// A refused open takes no number.
	if _, err := i.call("open", map[string]any{"title": "", "report": "r"}); err == nil {
		t.Fatal("an empty title was opened")
	}
	var o Opened
	_ = json.Unmarshal(must(i.call("open", map[string]any{"title": "the gate passes and composed nothing", "report": "seen on a merge"})), &o)
	if o.Number != 365 {
		t.Fatalf("opened %d after a refusal, want 365", o.Number)
	}
	must(i.call("note", map[string]any{"number": 365, "text": "found in the gate's plan"}))
	must(i.call("link", map[string]any{"number": 365, "kind": "decision", "target": "0237"}))
	if _, err := i.call("move", map[string]any{"number": 365, "status": "resolved", "fixed-by": []string{"mesh-controller#300"}}); err == nil {
		t.Fatal("resolved without located-in")
	}
	must(i.call("move", map[string]any{"number": 365, "status": "located", "located-in": []string{"mesh-controller internal/gate"}}))
	must(i.call("move", map[string]any{"number": 365, "status": "fixing", "by": "fix-365@g14"}))
	if _, err := i.call("move", map[string]any{"number": 365, "status": "resolved", "fixed-by": []string{"mesh-controller#300"}}); err == nil {
		t.Fatal("a core issue resolved without a replay")
	}
	must(i.call("move", map[string]any{"number": 365, "status": "resolved", "fixed-by": []string{"mesh-controller#300"}, "replay": "R365"}))
	var is Issue
	_ = json.Unmarshal(must(i.call("get", map[string]any{"number": 365})), &is)
	if is.Status != StatusResolved || len(is.Notes) != 1 || len(is.Links) != 1 || is.Replay != "R365" || is.Closed == nil {
		t.Fatalf("kept %+v", is)
	}
	var l Listed
	_ = json.Unmarshal(must(i.call("list", map[string]any{"status": "resolved"})), &l)
	if len(l.Rows) != 1 || l.Rows[0].Number != 365 {
		t.Fatalf("listed %+v", l)
	}
	evMu.Lock()
	defer evMu.Unlock()
	var kinds []string
	for _, e := range events {
		kinds = append(kinds, e.Key)
		// The move to fixing carries who works on it; every other change, who acted (ADR 0283 §4).
		want := Unproven
		if e.Key == EventMoved && e.Body["to"] == StatusFixing {
			want = "fix-365@g14"
		}
		if e.Body["by"] != want || e.Body["number"] != float64(365) {
			t.Fatalf("event %+v", e)
		}
	}
	if fmt.Sprint(kinds) != "[opened noted linked moved moved moved]" {
		t.Fatalf("events %v", kinds)
	}
}