834 lines
26 KiB
Go
834 lines
26 KiB
Go
//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"
|
|
"slices"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"git.novox.be/novox/mesh-contracts/go/seats/meshconfig"
|
|
oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel"
|
|
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
"git.novox.be/novox/mesh-sdk/go/health"
|
|
"git.novox.be/novox/mesh-sdk/go/process"
|
|
"git.novox.be/novox/mesh-sdk/go/testbus"
|
|
"git.novox.be/novox/mesh-sdk/go/views"
|
|
)
|
|
|
|
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
|
|
// owned is each delivery owner's state as the owner's tool answers it: its values and the revision of its last change.
|
|
owned map[string]*ownedState
|
|
}
|
|
|
|
// ownedState is one state of an owner (the indexer's modules, the assigner's wants...), kept from its bucket.
|
|
type ownedState struct {
|
|
owner, state string
|
|
revision uint64
|
|
values map[string]json.RawMessage
|
|
}
|
|
|
|
var theCreated = time.Date(2026, 11, 4, 9, 0, 0, 0, time.UTC)
|
|
|
|
// own follows the bucket of an owner's state, and says each change after the first listing on the owner's views family
|
|
// to the module (once it subscribed), as the owner does (ADR 0615 §4).
|
|
func (p *player) own(read string) {
|
|
p.t.Helper()
|
|
owner, state, _ := strings.Cut(read, ".")
|
|
kv, err := p.kv(read)
|
|
if err != nil {
|
|
p.t.Fatal(err)
|
|
}
|
|
w, err := kv.WatchAll(p.ctx)
|
|
if err != nil {
|
|
p.t.Fatal(err)
|
|
}
|
|
st := &ownedState{owner: owner, state: state, values: map[string]json.RawMessage{}}
|
|
p.owned[read] = st
|
|
listed := make(chan struct{})
|
|
go func() {
|
|
current := true
|
|
for e := range w.Updates() {
|
|
if e == nil {
|
|
current = false
|
|
close(listed)
|
|
continue
|
|
}
|
|
p.mu.Lock()
|
|
deleted := e.Operation() != jetstream.KeyValuePut
|
|
if deleted {
|
|
delete(st.values, e.Key())
|
|
} else {
|
|
st.values[e.Key()] = append(json.RawMessage(nil), e.Value()...)
|
|
}
|
|
if !current {
|
|
st.revision++
|
|
ev := map[string]any{"v": 1, "owner": owner, "state": state, "key": e.Key(), "revision": st.revision, "created": theCreated}
|
|
if deleted {
|
|
ev["deleted"] = true
|
|
} else {
|
|
ev["value"] = json.RawMessage(e.Value())
|
|
}
|
|
p.next++
|
|
id := fmt.Sprintf("o-%d", p.next)
|
|
body, _ := json.Marshal(ev)
|
|
raw, _ := json.Marshal(map[string]any{"id": id, "jsonrpc": "2.0", "method": "mesh/event", "params": map[string]any{
|
|
"envelope": map[string]any{"key": owner + ".views." + state, "node": "novox", "body": json.RawMessage(body)}}})
|
|
_, _ = p.in.Write(append(raw, '\n'))
|
|
}
|
|
p.mu.Unlock()
|
|
}
|
|
}()
|
|
select {
|
|
case <-listed:
|
|
case <-time.After(10 * time.Second):
|
|
p.t.Fatalf("%s was not listed", read)
|
|
}
|
|
}
|
|
|
|
// ownerAnswer is an owner's read tool asked with values: the state whole at its revision.
|
|
func (p *player) ownerAnswer(key string) (any, bool) {
|
|
for read, st := range p.owned {
|
|
if key == st.owner+"."+toolOf(st) {
|
|
_ = read
|
|
return map[string]any{"v": 1, "owner": st.owner, "state": st.state, "revision": st.revision, "created": theCreated,
|
|
"values": st.values, "at": time.Now()}, true
|
|
}
|
|
}
|
|
return nil, false
|
|
}
|
|
|
|
func toolOf(st *ownedState) string {
|
|
prefix := "indexer_read_"
|
|
if st.owner == "mesh-delivery-assigner" {
|
|
prefix = "assigner_read_"
|
|
}
|
|
return prefix + strings.ReplaceAll(st.state, "-", "_")
|
|
}
|
|
|
|
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 oc.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()
|
|
if a, ok := p.ownerAnswer(key); ok {
|
|
return a
|
|
}
|
|
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)
|
|
}
|
|
}
|
|
|
|
// conforms decodes what the module published under key strictly into the seat definition's type (a field the type does
|
|
// not know is a fault here, though a reader drops it) and checks the subject the runtime publishes it on is the one the
|
|
// definition names: CAT-V4, the definition's fixtures matched by a real event.
|
|
func (p *player) conforms(key string, into any, subject func(holder string) string) {
|
|
p.t.Helper()
|
|
p.mu.Lock()
|
|
defer p.mu.Unlock()
|
|
found := false
|
|
for _, s := range p.said {
|
|
if s.Key != key {
|
|
continue
|
|
}
|
|
found = true
|
|
dec := json.NewDecoder(strings.NewReader(string(s.Body)))
|
|
dec.DisallowUnknownFields()
|
|
if err := dec.Decode(into); err != nil {
|
|
p.t.Errorf("the published %s %s does not match the seat's definition: %v", key, s.Body, err)
|
|
}
|
|
// The runtime publishes an event its manifest's own stream carries on that stream (mesh.stream), any other on
|
|
// EVENTS (mesh.mod).
|
|
got := "mesh.mod." + Module + ".event." + s.Key
|
|
if streamCarries(p.t, s.Key) {
|
|
got = process.ModuleStreamSubject(Module, s.Key)
|
|
}
|
|
if want := subject(Module); got != want {
|
|
p.t.Errorf("%s is published on %s, the definition names %s", key, got, want)
|
|
}
|
|
}
|
|
if !found {
|
|
p.t.Errorf("%s was never published", 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
|
|
}
|
|
|
|
// hearsTheViews waits until the module subscribed to every owner's views family it follows: a change said before that
|
|
// would reach it only as a gap. Waited on, never slept for.
|
|
func (p *player) hearsTheViews() {
|
|
p.t.Helper()
|
|
eventually(p.t, "the module subscribed to every views family it follows", func() bool {
|
|
p.mu.Lock()
|
|
defer p.mu.Unlock()
|
|
for _, f := range followed {
|
|
if !slices.Contains(p.subs, f.Owner+"."+views.EventName(f.State)) {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
})
|
|
}
|
|
|
|
// viewsReady waits until every first view of the module is answered (config_first_views).
|
|
func (p *player) viewsReady() {
|
|
p.t.Helper()
|
|
eventually(p.t, "every first view is answered", func() bool {
|
|
raw, err := p.call(FirstViewsTool, nil)
|
|
if err != nil {
|
|
return false
|
|
}
|
|
var a struct {
|
|
Views []struct {
|
|
Revision uint64 `json:"revision"`
|
|
Answered time.Time `json:"Answered"`
|
|
} `json:"first-views"`
|
|
}
|
|
if json.Unmarshal(raw, &a) != nil || len(a.Views) != len(followed) {
|
|
return false
|
|
}
|
|
for _, v := range a.Views {
|
|
if v.Answered.IsZero() {
|
|
return false
|
|
}
|
|
}
|
|
return 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
|
|
p := &player{t: t, ctx: ctx, js: js, router: router, in: stdin, pending: map[string]chan wire{}, haves: haves, readers: readers,
|
|
owned: map[string]*ownedState{}}
|
|
for _, f := range followed {
|
|
p.own(f.Read)
|
|
}
|
|
if err := cmd.Start(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() { _ = stdin.Close(); cancel(); _ = cmd.Wait() })
|
|
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)
|
|
p.viewsReady()
|
|
|
|
// 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)
|
|
}
|
|
p.hearsTheViews()
|
|
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 })
|
|
|
|
p.conforms(EventRenamed, &meshconfig.RenamedHolderEvent{}, meshconfig.HolderSubjectRenamed)
|
|
|
|
// 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)
|
|
p2 := start(t, bin, js, router, haves, nil)
|
|
p2.viewsReady()
|
|
if _, err := p2.call(ResyncTool, nil); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
eventually(t, "after the restart the old name answers 8", func() bool {
|
|
v, ok := resolvedValue(p2, "g14", "player", "volume")
|
|
return ok && v == "8"
|
|
})
|
|
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)
|
|
}
|
|
|
|
// The seat's verb answers the open window in the definition's shape (CAT-C10).
|
|
raw, err = p2.call(Seat+"."+RenamesVerb, map[string]any{"module": "player"})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var open meshconfig.RenamesAnswer
|
|
dec := json.NewDecoder(strings.NewReader(string(raw)))
|
|
dec.DisallowUnknownFields()
|
|
if err := dec.Decode(&open); err != nil || len(open.Renames) != 1 || open.Renames[0].To != "level" ||
|
|
strings.Join(open.Renames[0].Open, ",") != "g14,novox" {
|
|
t.Fatalf("renames answered %s (%v), want player's level open on g14 and novox", raw, err)
|
|
}
|
|
|
|
// 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(oc.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
|
|
})
|
|
p2.conforms(EventRetiring, &meshconfig.RetiringHolderEvent{}, meshconfig.HolderSubjectRetiring)
|
|
p2.conforms(EventRetired, &meshconfig.RetiredHolderEvent{}, meshconfig.HolderSubjectRetired)
|
|
eventually(t, "the retirement is dropped", func() bool { _, ok := p2.get(RetiringState, "radio.band"); return !ok })
|
|
}
|
|
|
|
// streamCarries says whether the module's manifest names event among its own streams' events.
|
|
func streamCarries(t *testing.T, event string) bool {
|
|
t.Helper()
|
|
raw, err := os.ReadFile(filepath.Join("..", "..", "module.json"))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
m, err := catalogue.Decode(raw)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for _, st := range m.Streams {
|
|
if slices.Contains(st.Events, event) {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|