Files
mesh-catalog/modules/mesh-issues/cmd/mesh-issues/bus_test.go
T
jochen dcc7510cd8
mesh/delivery superseded: a newer head of the same pull request
mesh/merge-gate pass: builds mesh-issues → novox; no bus step; every machine composes with the change as it did without (4 of 4 compose)
mesh/repo-check pass: its merge-check.sh passed, each part in its toolchain: merge-check.sh (go), merge-check-typescript.sh (typescript)
mesh-issues: carry each value the issues bucket kept, read through the runtime's history, so the log begins with all the bucket still knew (hq ADR 0300 §6, ADR 0307)
2026-10-12 10:20:51 +02:00

703 lines
23 KiB
Go

package main
import (
"bufio"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"os"
"os/exec"
"path/filepath"
"sort"
"strconv"
"strings"
"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
// logs is each declared log's stream, made as the controller makes it (novox/hq ADR 0297).
logs map[string]jetstream.Stream
}
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{}, logs: map[string]jetstream.Stream{}}
// 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
}
for _, name := range m.Logs {
st, err := js.CreateStream(context.Background(), jetstream.StreamConfig{
Name: "LOG_" + m.Module + "_" + name, Subjects: []string{"mesh.log." + m.Module + "." + name + ".>"},
Storage: jetstream.FileStorage, MaxMsgsPerSubject: -1, Discard: jetstream.DiscardNew, AllowDirect: true,
DenyDelete: true, DenyPurge: true})
if err != nil {
t.Fatal(err)
}
b.logs[name] = st
}
return b
}
// logVerb is the runtime's mesh/log verbs over a stream, as the tool runner serves them (ADR 0297).
func (b *bus) logVerb(ctx context.Context, verb string, params json.RawMessage) (any, error) {
var p struct {
Log string `json:"log"`
Key string `json:"key"`
Value json.RawMessage `json:"value"`
Expect *uint64 `json:"expect"`
From uint64 `json:"from"`
Keys string `json:"keys"`
Limit int `json:"limit"`
}
if err := json.Unmarshal(params, &p); err != nil {
return nil, err
}
st := b.logs[p.Log]
if st == nil {
return nil, fmt.Errorf("the log %q is not declared", p.Log)
}
subject := func(key string) string { return "mesh.log.mesh-issues." + p.Log + "." + key }
entry := func(m *jetstream.RawStreamMsg) map[string]any {
return map[string]any{"sequence": m.Sequence, "key": m.Subject[len(subject("")):], "value": json.RawMessage(m.Data),
"at": m.Time}
}
switch verb {
case "append":
msg := nats.NewMsg(subject(p.Key))
msg.Data = p.Value
if p.Expect != nil {
msg.Header.Set(jetstream.ExpectedLastSubjSeqHeader, strconv.FormatUint(*p.Expect, 10))
}
ack, err := b.js.PublishMsg(ctx, msg)
if err != nil {
return nil, err
}
return map[string]any{"sequence": ack.Sequence}, nil
case "last":
m, err := st.GetLastMsgForSubject(ctx, subject(p.Key))
if errors.Is(err, jetstream.ErrMsgNotFound) {
return nil, nil
}
if err != nil {
return nil, err
}
return entry(m), nil
case "read":
filter := subject(">")
if p.Keys != "" {
filter = subject(p.Keys)
}
limit := p.Limit
if limit <= 0 {
limit = 100
}
info, err := st.Info(ctx)
if err != nil {
return nil, err
}
entries := []map[string]any{}
var next uint64
for seq := max(p.From, 1); seq <= info.State.LastSeq; {
m, err := st.GetMsg(ctx, seq, jetstream.WithGetMsgSubject(filter))
if errors.Is(err, jetstream.ErrMsgNotFound) {
break
}
if err != nil {
return nil, err
}
if len(entries) == limit {
next = m.Sequence
break
}
entries = append(entries, entry(m))
seq = m.Sequence + 1
}
return map[string]any{"entries": entries, "next": next, "last": info.State.LastSeq}, nil
}
return nil, fmt.Errorf("the runtime answers no mesh/log.%s", verb)
}
// 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 knowledge base, as a reservation asks it the highest numbers it holds (hq issue 488).
var a struct {
Key string `json:"key"`
}
if err := json.Unmarshal(m.Params, &a); err != nil || a.Key != HighestTool {
return nil, fmt.Errorf("the runtime answers no ask %s in this test", a.Key)
}
return map[string]any{"decision": 291, "research": 45, "records": 336}, nil
}
if verb, ok := strings.CutPrefix(m.Method, "mesh/log."); ok {
return i.b.logVerb(ctx, verb, m.Params)
}
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.history":
// As the runtime serves it (novox/hq ADR 0307): every value kept, oldest first, puts and deletes.
kept, err := kv.History(ctx, p.Key)
if errors.Is(err, jetstream.ErrKeyNotFound) {
return []any{}, nil
}
if err != nil {
return nil, err
}
out := []map[string]any{}
for _, e := range kept {
r := map[string]any{"key": e.Key(), "revision": e.Revision(), "op": "put", "at": e.Created()}
if e.Operation() == jetstream.KeyValuePut {
r["value"] = json.RawMessage(e.Value())
} else {
r["op"] = "delete"
}
out = append(out, r)
}
return out, nil
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)
}
}
// The carry over the bus (novox/hq ADR 0300 §6, ADR 0307): values the bucket kept before the switch are read
// through the runtime's history, oldest first, and carried before the write that touches the issue.
func TestAWriteOverTheBusCarriesEveryValueTheBucketKeptFirst(t *testing.T) {
b := newBus(t)
var events []event
var evMu sync.Mutex
kv := b.kvs[IssuesState]
for _, title := range []string{"first", "second", "third"} {
raw, _ := json.Marshal(Issue{Number: 300, Title: title, Status: StatusOpen, Report: "r", OpenedBy: "x", Opening: "o300"})
if _, err := kv.Put(context.Background(), "300", raw); err != nil {
t.Fatal(err)
}
}
i := b.start(t, writeSettings(t, 365), &events, &evMu)
if _, err := i.call("note", map[string]any{"number": 300, "text": "after the switch"}); err != nil {
t.Fatal(err)
}
var titles []string
var ops []string
for seq := uint64(1); ; seq++ {
m, err := b.logs[ChangesLog].GetMsg(context.Background(), seq)
if err != nil {
break
}
var op Operation
_ = json.Unmarshal(m.Data, &op)
ops = append(ops, op.Op)
if op.Issue != nil {
titles = append(titles, op.Issue.Title)
}
}
if fmt.Sprint(ops) != "[carried carried carried noted]" || fmt.Sprint(titles) != "[first second third]" {
t.Fatalf("the log holds %v of %v", ops, titles)
}
}
// 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 0284 §6).
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)
}
// The record is the log (novox/hq ADR 0300 §1): one operation per change, in the stream the controller makes,
// and the issue in the bucket folded at the last of them.
var ops []string
for seq := uint64(1); ; seq++ {
m, err := b.logs[ChangesLog].GetMsg(context.Background(), seq)
if err != nil {
break
}
var op Operation
if err := json.Unmarshal(m.Data, &op); err != nil || m.Subject != "mesh.log.mesh-issues.changes.365" {
t.Fatalf("entry %d on %s: %s", seq, m.Subject, m.Data)
}
ops = append(ops, op.Op)
}
if fmt.Sprint(ops) != fmt.Sprint(kinds) || is.Folded != uint64(len(ops)) {
t.Fatalf("the log holds %v, the issue is folded at %d", ops, is.Folded)
}
}