CAT-C9: mesh-config's first whole-module integration test, the rename window and the retirement against a real bus
The built binary, played as the node's tool runtime over a real bus: a renamed value moves and both names answer, the module restarted inside the window keeps it and says renamed once, a refused move writes its finding, g14's window closes on delivered, a value no manifest reads is asked about once when no node reads it and removed only on the Approve. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
@@ -0,0 +1,623 @@
|
||||
//go:build integration
|
||||
|
||||
package main
|
||||
|
||||
// mesh-config whole against a real bus (design 32, C20 and C24): the built binary, played over its real stdin and stdout
|
||||
// as the node's tool runtime plays it (tools/call; mesh/state.* answered from a real nats-server's key-value buckets,
|
||||
// watches handed over as the runtime hands them, mesh/event delivering the events it subscribed to, mesh/ask answered
|
||||
// as the status view, the indexer, the node-engine seat and mesh-hand-acts answer it, mesh/seat.* as the router does).
|
||||
//
|
||||
// A renamed value moves with no input, keeps its value, and both names answer until the node-engine reports the new
|
||||
// build on `delivered`; the module restarted inside the window keeps the window and says `renamed` once; a refused move
|
||||
// writes its finding. A value no manifest reads is not asked about while a node reads it, is asked once when none does,
|
||||
// stays until the operator's Approve and is then removed, `retired` said.
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
"git.novox.be/novox/mesh-sdk/go/testbus"
|
||||
)
|
||||
|
||||
type wire 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 {
|
||||
Message string `json:"message"`
|
||||
} `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
// player is the node's tool runtime, as far as mesh-config needs it, over a real bus.
|
||||
type player struct {
|
||||
t *testing.T
|
||||
ctx context.Context
|
||||
js jetstream.JetStream
|
||||
router *standIn
|
||||
mu sync.Mutex
|
||||
in io.Writer
|
||||
next int
|
||||
pending map[string]chan wire
|
||||
// haves is the status view's node_haves, by "<node>.<module>"; readers the indexer's answer.
|
||||
haves map[string]string
|
||||
readers []ReaderRow
|
||||
said []saidOnBus
|
||||
subs []string
|
||||
}
|
||||
|
||||
type saidOnBus struct {
|
||||
Key string
|
||||
Body json.RawMessage
|
||||
}
|
||||
|
||||
func (p *player) send(m map[string]any) {
|
||||
raw, _ := json.Marshal(m)
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
_, _ = p.in.Write(append(raw, '\n'))
|
||||
}
|
||||
|
||||
func (p *player) answer(id json.RawMessage, result any, err error) {
|
||||
if err != nil {
|
||||
p.send(map[string]any{"id": id, "error": map[string]any{"code": -32000, "message": err.Error()}})
|
||||
return
|
||||
}
|
||||
p.send(map[string]any{"id": id, "result": result})
|
||||
}
|
||||
|
||||
func (p *player) serve(out io.Reader) {
|
||||
scanner := bufio.NewScanner(out)
|
||||
scanner.Buffer(make([]byte, 1<<20), 1<<24)
|
||||
for scanner.Scan() {
|
||||
var m wire
|
||||
if json.Unmarshal(scanner.Bytes(), &m) != nil {
|
||||
continue
|
||||
}
|
||||
if m.Method == "" {
|
||||
var key string
|
||||
_ = json.Unmarshal(m.ID, &key)
|
||||
p.mu.Lock()
|
||||
ch := p.pending[key]
|
||||
delete(p.pending, key)
|
||||
p.mu.Unlock()
|
||||
if ch != nil {
|
||||
ch <- m
|
||||
}
|
||||
continue
|
||||
}
|
||||
go p.handle(m)
|
||||
}
|
||||
}
|
||||
|
||||
func (p *player) kv(state string) (jetstream.KeyValue, error) {
|
||||
if owner, name, read := strings.Cut(state, "."); read {
|
||||
return p.js.KeyValue(p.ctx, owner+"_"+name)
|
||||
}
|
||||
return p.js.KeyValue(p.ctx, Module+"_"+state)
|
||||
}
|
||||
|
||||
func (p *player) handle(m wire) {
|
||||
var a struct {
|
||||
State string `json:"state"`
|
||||
Key string `json:"key"`
|
||||
Value json.RawMessage `json:"value"`
|
||||
Revision uint64 `json:"revision"`
|
||||
Subject string `json:"subject"`
|
||||
Body json.RawMessage `json:"body"`
|
||||
Pattern string `json:"pattern"`
|
||||
ID string `json:"id"`
|
||||
Bucket string `json:"bucket"`
|
||||
}
|
||||
_ = json.Unmarshal(m.Params, &a)
|
||||
kv, kvErr := p.kv(a.State)
|
||||
switch m.Method {
|
||||
case "mesh/state.get":
|
||||
if kvErr != nil {
|
||||
p.answer(m.ID, nil, kvErr)
|
||||
return
|
||||
}
|
||||
e, err := kv.Get(p.ctx, a.Key)
|
||||
if errors.Is(err, jetstream.ErrKeyNotFound) {
|
||||
p.answer(m.ID, nil, nil)
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
p.answer(m.ID, nil, err)
|
||||
return
|
||||
}
|
||||
p.answer(m.ID, map[string]any{"key": e.Key(), "value": json.RawMessage(e.Value()), "revision": e.Revision()}, nil)
|
||||
case "mesh/state.create", "mesh/state.update", "mesh/state.put":
|
||||
if kvErr != nil {
|
||||
p.answer(m.ID, nil, kvErr)
|
||||
return
|
||||
}
|
||||
var rev uint64
|
||||
var err error
|
||||
switch m.Method {
|
||||
case "mesh/state.create":
|
||||
rev, err = kv.Create(p.ctx, a.Key, a.Value)
|
||||
case "mesh/state.update":
|
||||
rev, err = kv.Update(p.ctx, a.Key, a.Value, a.Revision)
|
||||
default:
|
||||
rev, err = kv.Put(p.ctx, a.Key, a.Value)
|
||||
}
|
||||
if errors.Is(err, jetstream.ErrKeyExists) || errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
|
||||
p.send(map[string]any{"id": m.ID, "error": map[string]any{"code": -32010, "message": "changed: " + err.Error()}})
|
||||
return
|
||||
}
|
||||
p.answer(m.ID, map[string]any{"revision": rev}, err)
|
||||
case "mesh/state.keys":
|
||||
if kvErr != nil {
|
||||
p.answer(m.ID, nil, kvErr)
|
||||
return
|
||||
}
|
||||
keys, err := kv.Keys(p.ctx)
|
||||
if errors.Is(err, jetstream.ErrNoKeysFound) {
|
||||
keys, err = []string{}, nil
|
||||
}
|
||||
p.answer(m.ID, keys, err)
|
||||
case "mesh/state.delete":
|
||||
err := kvErr
|
||||
if err == nil {
|
||||
err = kv.Delete(p.ctx, a.Key)
|
||||
}
|
||||
p.answer(m.ID, map[string]any{}, err)
|
||||
case "mesh/state.watch":
|
||||
if kvErr != nil {
|
||||
p.answer(m.ID, nil, kvErr)
|
||||
return
|
||||
}
|
||||
w, err := kv.WatchAll(p.ctx)
|
||||
if err != nil {
|
||||
p.answer(m.ID, nil, err)
|
||||
return
|
||||
}
|
||||
var watchID string
|
||||
_ = json.Unmarshal(m.ID, &watchID)
|
||||
current := true
|
||||
for e := range w.Updates() {
|
||||
if e == nil {
|
||||
current = false
|
||||
p.answer(m.ID, map[string]any{}, nil)
|
||||
continue
|
||||
}
|
||||
op := "put"
|
||||
if e.Operation() != jetstream.KeyValuePut {
|
||||
op = "delete"
|
||||
}
|
||||
p.mu.Lock()
|
||||
p.next++
|
||||
id := fmt.Sprintf("w-%d", p.next)
|
||||
p.mu.Unlock()
|
||||
p.send(map[string]any{"id": id, "method": "mesh/state", "params": map[string]any{"watch": watchID,
|
||||
"change": map[string]any{"state": a.State, "key": e.Key(), "op": op, "value": json.RawMessage(nullIfEmpty(e.Value())),
|
||||
"revision": e.Revision(), "current": current}}})
|
||||
}
|
||||
case "mesh/log.append":
|
||||
p.answer(m.ID, map[string]any{"sequence": 1}, nil)
|
||||
case "mesh/log.read":
|
||||
p.answer(m.ID, map[string]any{"entries": []any{}, "next": 0, "last": 0}, nil)
|
||||
case "mesh/publish":
|
||||
var e struct {
|
||||
Key string `json:"key"`
|
||||
Body json.RawMessage `json:"body"`
|
||||
}
|
||||
_ = json.Unmarshal(m.Params, &e)
|
||||
p.mu.Lock()
|
||||
p.said = append(p.said, saidOnBus{e.Key, e.Body})
|
||||
p.mu.Unlock()
|
||||
p.answer(m.ID, map[string]any{}, nil)
|
||||
case "mesh/subscribe":
|
||||
p.mu.Lock()
|
||||
p.subs = append(p.subs, a.Pattern)
|
||||
p.mu.Unlock()
|
||||
p.answer(m.ID, map[string]any{}, nil)
|
||||
case "mesh/seat.publish":
|
||||
var ask asks.Ask
|
||||
if err := json.Unmarshal(a.Body, &ask); err != nil {
|
||||
p.answer(m.ID, nil, err)
|
||||
return
|
||||
}
|
||||
p.answer(m.ID, map[string]any{"sequence": 1}, p.router.Publish(a.Subject, ask, a.ID))
|
||||
case "mesh/seat.record":
|
||||
e, err := p.router.Record(a.Key)
|
||||
if e == nil {
|
||||
p.answer(m.ID, nil, err)
|
||||
return
|
||||
}
|
||||
p.answer(m.ID, e, err)
|
||||
case "mesh/seat.kinds":
|
||||
p.answer(m.ID, p.router.kinds, nil)
|
||||
case "mesh/ask":
|
||||
var q struct {
|
||||
Key string `json:"key"`
|
||||
Body json.RawMessage `json:"body"`
|
||||
}
|
||||
_ = json.Unmarshal(m.Params, &q)
|
||||
p.answer(m.ID, p.asked(q.Key, q.Body), nil)
|
||||
default:
|
||||
p.answer(m.ID, nil, errors.New("nats: no responders available for request"))
|
||||
}
|
||||
}
|
||||
|
||||
// asked answers a tool or a seat verb mesh-config asks, as its holder does.
|
||||
func (p *player) asked(key string, body json.RawMessage) any {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
switch {
|
||||
case key == HavesAsk:
|
||||
var q struct{ Node, Module string }
|
||||
_ = json.Unmarshal(body, &q)
|
||||
hk, _ := delivery.HaveKey(q.Node, q.Module)
|
||||
keys := map[string]any{}
|
||||
if fp, ok := p.haves[q.Node+"."+q.Module]; ok {
|
||||
keys[hk] = delivery.Have{V: delivery.Version, FP: fp, Phase: "applied", Health: "healthy"}
|
||||
}
|
||||
return map[string]any{"keys": keys}
|
||||
case strings.HasPrefix(key, ManifestReadsAsk):
|
||||
return map[string]any{}
|
||||
case key == ConfigReadersAsk:
|
||||
return map[string]any{"readers": p.readers}
|
||||
case key == "seat:mesh-hand-acts.hand-act" || key == "seat:mesh-hand-acts.warranted":
|
||||
return map[string]any{"said": "recorded"}
|
||||
}
|
||||
return map[string]any{"error": "no responders available for " + key}
|
||||
}
|
||||
|
||||
func nullIfEmpty(b []byte) []byte {
|
||||
if len(b) == 0 {
|
||||
return []byte("null")
|
||||
}
|
||||
return b
|
||||
}
|
||||
|
||||
func (p *player) call(name string, args map[string]any) (json.RawMessage, error) {
|
||||
p.mu.Lock()
|
||||
p.next++
|
||||
id := fmt.Sprintf("c-%d", p.next)
|
||||
ch := make(chan wire, 1)
|
||||
p.pending[id] = ch
|
||||
p.mu.Unlock()
|
||||
p.send(map[string]any{"id": id, "method": "tools/call", "params": map[string]any{"name": name, "arguments": args,
|
||||
"_meta": map[string]any{"mesh/caller": "person.jochen"}}})
|
||||
select {
|
||||
case m := <-ch:
|
||||
if m.Error != nil {
|
||||
return nil, errors.New(m.Error.Message)
|
||||
}
|
||||
var r struct {
|
||||
Content []struct {
|
||||
Text string `json:"text"`
|
||||
} `json:"content"`
|
||||
IsError bool `json:"isError"`
|
||||
}
|
||||
if err := json.Unmarshal(m.Result, &r); err != nil || len(r.Content) != 1 {
|
||||
return nil, fmt.Errorf("%s answered %s", name, m.Result)
|
||||
}
|
||||
if r.IsError {
|
||||
return nil, errors.New(r.Content[0].Text)
|
||||
}
|
||||
return json.RawMessage(r.Content[0].Text), nil
|
||||
case <-time.After(60 * time.Second):
|
||||
return nil, fmt.Errorf("%s did not answer in 60s", name)
|
||||
}
|
||||
}
|
||||
|
||||
// event hands the module one event it subscribed to, as the runtime does, and waits for its answer.
|
||||
func (p *player) event(key string, body any) {
|
||||
p.t.Helper()
|
||||
raw, _ := json.Marshal(body)
|
||||
p.mu.Lock()
|
||||
p.next++
|
||||
id := fmt.Sprintf("e-%d", p.next)
|
||||
ch := make(chan wire, 1)
|
||||
p.pending[id] = ch
|
||||
p.mu.Unlock()
|
||||
p.send(map[string]any{"id": id, "method": "mesh/event", "params": map[string]any{"envelope": map[string]any{"key": key, "node": "novox",
|
||||
"body": json.RawMessage(raw)}}})
|
||||
select {
|
||||
case m := <-ch:
|
||||
if m.Error != nil {
|
||||
p.t.Fatalf("%s was not taken: %s", key, m.Error.Message)
|
||||
}
|
||||
case <-time.After(30 * time.Second):
|
||||
p.t.Fatalf("%s was not answered in 30s", key)
|
||||
}
|
||||
}
|
||||
|
||||
func (p *player) publishedCount(key string) int {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
n := 0
|
||||
for _, s := range p.said {
|
||||
if s.Key == key {
|
||||
n++
|
||||
}
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
func (p *player) put(state, key string, v any) {
|
||||
p.t.Helper()
|
||||
kv, err := p.kv(state)
|
||||
if err != nil {
|
||||
p.t.Fatal(err)
|
||||
}
|
||||
raw, _ := json.Marshal(v)
|
||||
if _, err := kv.Put(p.ctx, key, raw); err != nil {
|
||||
p.t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func (p *player) get(state, key string) ([]byte, bool) {
|
||||
kv, err := p.kv(state)
|
||||
if err != nil {
|
||||
return nil, false
|
||||
}
|
||||
e, err := kv.Get(p.ctx, key)
|
||||
if err != nil {
|
||||
return nil, false
|
||||
}
|
||||
return e.Value(), true
|
||||
}
|
||||
|
||||
func eventually(t *testing.T, what string, ok func() bool) {
|
||||
t.Helper()
|
||||
for deadline := time.Now().Add(30 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
|
||||
if ok() {
|
||||
return
|
||||
}
|
||||
}
|
||||
t.Fatalf("never: %s", what)
|
||||
}
|
||||
|
||||
// manifestIn writes a module's manifest into the indexer's buckets as the indexer writes it, and names it as the trunk's.
|
||||
func (p *player) manifestIn(module, raw string) string {
|
||||
p.t.Helper()
|
||||
digest := delivery.ManifestDigest([]byte(raw))
|
||||
value, err := delivery.EncodeManifest(delivery.Manifest{Module: module, JSON: raw})
|
||||
if err != nil {
|
||||
p.t.Fatal(err)
|
||||
}
|
||||
p.put("mesh-delivery-indexer.manifests", digest, json.RawMessage(value))
|
||||
p.put("mesh-delivery-indexer.modules", module, map[string]any{"v": 1, "manifest": digest, "repository": "novox/" + module, "commit": "c1"})
|
||||
return digest
|
||||
}
|
||||
|
||||
// buildIn writes a build of a module with a manifest, and the want that names it on a node.
|
||||
func (p *player) buildIn(node, module, fp, raw string) {
|
||||
p.t.Helper()
|
||||
digest := delivery.ManifestDigest([]byte(raw))
|
||||
value, _ := delivery.EncodeManifest(delivery.Manifest{Module: module, JSON: raw})
|
||||
p.put("mesh-delivery-indexer.manifests", digest, json.RawMessage(value))
|
||||
bk, _ := delivery.BuilderBuildKey(module, fp)
|
||||
p.put("mesh-delivery-builder.builds", bk, map[string]any{"v": 1, "state": "done", "manifest": digest})
|
||||
p.put("mesh-delivery-assigner.wants", "want."+node+"."+module, map[string]any{"v": 1, "fp": fp, "self": true})
|
||||
}
|
||||
|
||||
func (p *player) placed(node string, modules ...string) {
|
||||
p.t.Helper()
|
||||
pl := delivery.Placement{V: delivery.Version, Node: node}
|
||||
for _, m := range modules {
|
||||
pl.Modules = append(pl.Modules, delivery.PlacedModule{Module: m})
|
||||
}
|
||||
values, err := delivery.PlacementValues(pl)
|
||||
if err != nil {
|
||||
p.t.Fatal(err)
|
||||
}
|
||||
for k, raw := range values {
|
||||
p.put("mesh-delivery-assigner.placement", k, json.RawMessage(raw))
|
||||
}
|
||||
}
|
||||
|
||||
func resolvedValue(p *player, node, module, key string) (string, bool) {
|
||||
raw, ok := p.get(ResolvedState, ResolvedKey(node, module, key))
|
||||
if !ok {
|
||||
return "", false
|
||||
}
|
||||
var r Resolved
|
||||
_ = json.Unmarshal(raw, &r)
|
||||
return string(r.Value), true
|
||||
}
|
||||
|
||||
// start runs the built binary on the bus and serves it as the runtime.
|
||||
func start(t *testing.T, bin string, js jetstream.JetStream, router *standIn, haves map[string]string, readers []ReaderRow) *player {
|
||||
t.Helper()
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cmd := exec.CommandContext(ctx, bin)
|
||||
cmd.Env = append(os.Environ(), "MESH_NODE=novox", "MESH_MODULE="+Module)
|
||||
stdin, _ := cmd.StdinPipe()
|
||||
stdout, _ := cmd.StdoutPipe()
|
||||
cmd.Stderr = os.Stderr
|
||||
if err := cmd.Start(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = stdin.Close(); cancel(); _ = cmd.Wait() })
|
||||
p := &player{t: t, ctx: ctx, js: js, router: router, in: stdin, pending: map[string]chan wire{}, haves: haves, readers: readers}
|
||||
go p.serve(stdout)
|
||||
t.Cleanup(cancel)
|
||||
return p
|
||||
}
|
||||
|
||||
// 1966/C9 (design 32, C20 and C24): the module whole on a real bus.
|
||||
func TestConfigWholeRenamesAndRetiresAgainstTheBus(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
t.Cleanup(cancel)
|
||||
bin := filepath.Join(t.TempDir(), Module)
|
||||
build := exec.Command("go", "build", "-o", bin, ".")
|
||||
build.Stderr = os.Stderr
|
||||
if err := build.Run(); err != nil {
|
||||
t.Fatalf("mesh-config does not build: %v", err)
|
||||
}
|
||||
nc, err := nats.Connect(testbus.URL(t))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(nc.Close)
|
||||
js, err := jetstream.New(nc)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
buckets := []string{"mesh-delivery-assigner_placement", "mesh-delivery-assigner_wants", "mesh-delivery-builder_builds",
|
||||
"mesh-delivery-indexer_modules", "mesh-delivery-indexer_manifests", "mesh-delivery-indexer_predefined",
|
||||
"mesh-delivery-indexer_config-candidates"}
|
||||
for _, s := range []string{ValuesState, ResolvedState, PredefinedState, ProposalsState, PullsState, KeptState, DeclaredState,
|
||||
RenamesState, RetiringState, FindingsState} {
|
||||
buckets = append(buckets, Module+"_"+s)
|
||||
}
|
||||
for _, b := range buckets {
|
||||
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b, History: 2}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
router := newStandIn()
|
||||
haves := map[string]string{"g14.player": fp1, "novox.player": fp1, "g14.radio": fp1}
|
||||
p := start(t, bin, js, router, haves, nil)
|
||||
|
||||
// The mesh: player and radio on g14, player also on novox, all on their first build; dock declares both names.
|
||||
const oldRadio = `{"module":"radio","config":{"band":{"type":"number","default":1}}}`
|
||||
const newRadio = `{"module":"radio","config":{"gain":{"type":"number","default":1}}}`
|
||||
const dock = `{"module":"dock","config":{"a":{"type":"number","default":1},"b":{"type":"number","default":1,"renamed-from":"a"}}}`
|
||||
p.manifestIn("player", oldPlayer)
|
||||
p.manifestIn("radio", oldRadio)
|
||||
p.buildIn("g14", "player", fp1, oldPlayer)
|
||||
p.buildIn("novox", "player", fp1, oldPlayer)
|
||||
p.buildIn("g14", "radio", fp1, oldRadio)
|
||||
p.placed("g14", "player", "radio")
|
||||
p.placed("novox", "player")
|
||||
if _, err := p.call(ResyncTool, nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
time.Sleep(time.Second) // the watches start after the first call
|
||||
set := func(module, key string, value any) {
|
||||
t.Helper()
|
||||
if _, err := p.call(Seat+"."+Verb, map[string]any{"action": "set", "module": module, "node": "g14", "key": key,
|
||||
"value": value, "why": "the operator asked for it"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
set("player", "volume", 8)
|
||||
set("radio", "band", 3)
|
||||
|
||||
// C20: the new manifest renames volume to level; the value moves, both names answer.
|
||||
p.manifestIn("player", newPlayer)
|
||||
eventually(t, "the value moves to the new name", func() bool {
|
||||
raw, ok := p.get(ValuesState, "module.g14.player.level")
|
||||
return ok && strings.Contains(string(raw), `"value":8`)
|
||||
})
|
||||
if _, ok := p.get(ValuesState, "module.g14.player.volume"); ok {
|
||||
t.Error("the old name still holds a value")
|
||||
}
|
||||
eventually(t, "both names answer on g14", func() bool {
|
||||
a, okA := resolvedValue(p, "g14", "player", "level")
|
||||
b, okB := resolvedValue(p, "g14", "player", "volume")
|
||||
return okA && okB && a == "8" && b == "8"
|
||||
})
|
||||
eventually(t, "renamed is said", func() bool { return p.publishedCount(EventRenamed) == 1 })
|
||||
|
||||
// A refused move writes its finding: dock declares both names.
|
||||
p.manifestIn("dock", dock)
|
||||
p.buildIn("g14", "dock", fp1, dock)
|
||||
p.placed("g14", "player", "radio", "dock")
|
||||
eventually(t, "the refused rename's finding", func() bool {
|
||||
keys, _ := (func() ([]string, error) { kv, _ := p.kv(FindingsState); return kv.Keys(p.ctx) })()
|
||||
for _, k := range keys {
|
||||
raw, _ := p.get(FindingsState, k)
|
||||
if f, err := health.DecodeFinding(raw); err == nil && f.Kind == FindingRenameRefused && strings.Contains(f.Summary, "dock") {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
})
|
||||
|
||||
// The module restarted inside the window keeps it, and says renamed no more.
|
||||
said := p.publishedCount(EventRenamed)
|
||||
cancelOld := p.ctx
|
||||
_ = cancelOld
|
||||
p2 := start(t, bin, js, router, haves, nil)
|
||||
if _, err := p2.call(ResyncTool, nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
time.Sleep(time.Second)
|
||||
if v, ok := resolvedValue(p2, "g14", "player", "volume"); !ok || v != "8" {
|
||||
t.Errorf("after the restart the old name answers %q (%v), want 8", v, ok)
|
||||
}
|
||||
if n := p2.publishedCount(EventRenamed); n != 0 || said != 1 {
|
||||
t.Errorf("renamed was said %d times before the restart and %d after", said, n)
|
||||
}
|
||||
raw, _ := p2.get(RenamesState, "player.level")
|
||||
var rn Rename
|
||||
if json.Unmarshal(raw, &rn) != nil || strings.Join(rn.Open(), ",") != "g14,novox" {
|
||||
t.Fatalf("the window after the restart: %s", raw)
|
||||
}
|
||||
|
||||
// g14 applies the build that reads the new name; the delivery says so; its window closes, novox's stays.
|
||||
p2.buildIn("g14", "player", fp2, newPlayer)
|
||||
p2.mu.Lock()
|
||||
p2.haves["g14.player"] = fp2
|
||||
p2.mu.Unlock()
|
||||
p2.event("mesh-delivery-view.delivered", map[string]any{"revision": "r1"})
|
||||
eventually(t, "g14's window closes and the old name goes", func() bool {
|
||||
_, old := resolvedValue(p2, "g14", "player", "volume")
|
||||
v, still := resolvedValue(p2, "novox", "player", "volume")
|
||||
return !old && still && v == "5"
|
||||
})
|
||||
|
||||
// C24: radio's manifest drops band; g14 still runs the build that reads it: no ask. Then it runs the next: one ask.
|
||||
p2.manifestIn("radio", newRadio)
|
||||
p2.mu.Lock()
|
||||
p2.readers = []ReaderRow{{Module: "radio", Key: "band", Kind: KindValue, Readers: []string{}}}
|
||||
p2.mu.Unlock()
|
||||
p2.event("mesh-delivery-indexer.manifest-accepted", map[string]any{"module": "radio"})
|
||||
eventually(t, "the retirement waits on g14", func() bool {
|
||||
raw, ok := p2.get(RetiringState, "radio.band")
|
||||
return ok && strings.Contains(string(raw), RetireDraining)
|
||||
})
|
||||
if len(router.asks) != 0 || p2.publishedCount(EventRetiring) != 0 {
|
||||
t.Fatalf("asked %d times, retiring said %d times while g14 reads the value", len(router.asks), p2.publishedCount(EventRetiring))
|
||||
}
|
||||
p2.buildIn("g14", "radio", fp2, newRadio)
|
||||
p2.mu.Lock()
|
||||
p2.haves["g14.radio"] = fp2
|
||||
p2.mu.Unlock()
|
||||
p2.event("mesh-delivery-view.delivered", map[string]any{"revision": "r2"})
|
||||
eventually(t, "the operator is asked once", func() bool {
|
||||
router.mu.Lock()
|
||||
defer router.mu.Unlock()
|
||||
return len(router.asks) == 1 && p2.publishedCount(EventRetiring) == 1
|
||||
})
|
||||
if _, ok := p2.get(ValuesState, "module.g14.radio.band"); !ok {
|
||||
t.Fatal("the value was removed before the operator answered")
|
||||
}
|
||||
var id string
|
||||
router.mu.Lock()
|
||||
for k := range router.asks {
|
||||
id = k
|
||||
}
|
||||
router.mu.Unlock()
|
||||
w := router.choose(t, id, OptionApprove, nil)
|
||||
p2.event(asks.Seat+".decided."+Module, w)
|
||||
eventually(t, "the approved value is removed and retired is said", func() bool {
|
||||
_, still := p2.get(ValuesState, "module.g14.radio.band")
|
||||
return !still && p2.publishedCount(EventRetired) == 1
|
||||
})
|
||||
eventually(t, "the retirement is dropped", func() bool { _, ok := p2.get(RetiringState, "radio.band"); return !ok })
|
||||
}
|
||||
@@ -2,13 +2,21 @@ module mesh-config
|
||||
|
||||
go 1.26.0
|
||||
|
||||
require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261102191058-f02f142af128
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261102191058-f02f142af128
|
||||
github.com/nats-io/nats.go v1.54.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.20.0 // indirect
|
||||
github.com/nats-io/nats.go v1.54.0 // indirect
|
||||
github.com/minio/highwayhash v1.0.4 // indirect
|
||||
github.com/nats-io/jwt/v2 v2.8.1 // indirect
|
||||
github.com/nats-io/nats-server/v2 v2.11.17 // indirect
|
||||
github.com/nats-io/nkeys v0.4.16 // indirect
|
||||
github.com/nats-io/nuid v1.0.1 // indirect
|
||||
golang.org/x/crypto v0.57.0 // indirect
|
||||
golang.org/x/sys v0.48.0 // indirect
|
||||
golang.org/x/time v0.15.0 // indirect
|
||||
)
|
||||
|
||||
@@ -1,7 +1,17 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261102191058-f02f142af128 h1:aI1DhZAuqVBXbybSqhLNOGX5rUOrn9T0cHNCDUQF7IQ=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261102191058-f02f142af128/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
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.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA=
|
||||
github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
|
||||
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.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA=
|
||||
github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k=
|
||||
github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg=
|
||||
@@ -10,5 +20,8 @@ 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.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M=
|
||||
golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA=
|
||||
golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
|
||||
golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
|
||||
golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
|
||||
golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
|
||||
golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
|
||||
|
||||
Reference in New Issue
Block a user