jschoubben
2cd5550dc7
mesh/delivery merged: source.main moved to novox/mesh-tools@5d6cee9d844e
mesh/check pass 2cd5550dc70f on 77b05089c8df | unit 0 run 126 kept | int 0 run 128 kept | clone 9.5s | composition passed on 4 machines
Its events are refused before the old bundle stops, the bundle watch follows a directory made between its stat and its watch, and a bundle's reason is read without node's colour codes.
443 lines
15 KiB
Go
443 lines
15 KiB
Go
package runtime
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/micro"
|
|
|
|
"github.com/novox/mesh-tools/node-tools/internal/bus"
|
|
mt "github.com/novox/mesh-tools/node-tools/internal/meshtest"
|
|
)
|
|
|
|
// A runtime serves a module's bundle as soon as it is in place, again when it is replaced, and no
|
|
// more once it is gone, with no restart (issue 1945, note 23: mesh-files' bundle arrived on a machine
|
|
// a quarter of an hour after its runtime started, and its tools were missing until a hand restart).
|
|
|
|
// syncLogs is a logger that keeps the lines, safe to read while the runtime's watcher writes.
|
|
type syncLogs struct {
|
|
mu sync.Mutex
|
|
lines []string
|
|
}
|
|
|
|
func (l *syncLogs) Logf(format string, args ...any) {
|
|
l.mu.Lock()
|
|
defer l.mu.Unlock()
|
|
l.lines = append(l.lines, fmt.Sprintf(format, args...))
|
|
}
|
|
|
|
func (l *syncLogs) Has(fragments ...string) bool {
|
|
l.mu.Lock()
|
|
defer l.mu.Unlock()
|
|
for _, line := range l.lines {
|
|
all := true
|
|
for _, f := range fragments {
|
|
all = all && strings.Contains(line, f)
|
|
}
|
|
if all {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func (l *syncLogs) All() string {
|
|
l.mu.Lock()
|
|
defer l.mu.Unlock()
|
|
return strings.Join(l.lines, "\n")
|
|
}
|
|
|
|
// bundleServing is a tools bundle, MCP over stdio with no SDK, whose one tool is named tool.
|
|
func bundleServing(tool string) string { return bundleScript(tool, false, false) }
|
|
|
|
// bundleScript is a tools bundle whose one tool is named tool. With subscribe it asks the runtime for its
|
|
// module's events and writes "<tool> subscribed" and "<tool> handled <n>" to $EVENT_LOG; with writeBeside it
|
|
// writes a file beside itself as it starts, as a bundle keeping a cache or a socket there does.
|
|
func bundleScript(tool string, subscribe, writeBeside bool) string {
|
|
py := func(b bool) string {
|
|
if b {
|
|
return "True"
|
|
}
|
|
return "False"
|
|
}
|
|
return `#!/usr/bin/env python3
|
|
import json, os, sys
|
|
TOOL = "` + tool + `"
|
|
SUBSCRIBE = ` + py(subscribe) + `
|
|
if ` + py(writeBeside) + `:
|
|
with open(os.path.join(os.path.dirname(os.path.abspath(__file__)), "cache"), "w") as f:
|
|
f.write(TOOL + " " + str(os.getpid()))
|
|
def say(m):
|
|
m["jsonrpc"] = "2.0"; sys.stdout.write(json.dumps(m) + "\n"); sys.stdout.flush()
|
|
def log(line):
|
|
with open(os.environ["EVENT_LOG"], "a") as f:
|
|
f.write(line + "\n")
|
|
TOOLS = [{"name": TOOL, "description": "answers its name", "inputSchema": {"type": "object", "properties": {}}}]
|
|
for line in sys.stdin:
|
|
req = json.loads(line); rid = req.get("id"); m = req.get("method"); p = req.get("params") or {}
|
|
if m is None:
|
|
if rid == "subscribe":
|
|
log(TOOL + " subscribed")
|
|
continue
|
|
if m == "initialize":
|
|
say({"id": rid, "result": {"protocolVersion": "2025-03-26", "capabilities": {"tools": {}}, "serverInfo": {"name": "ledger", "version": "1"}}})
|
|
elif m == "tools/list":
|
|
say({"id": rid, "result": {"tools": TOOLS}})
|
|
if SUBSCRIBE:
|
|
say({"id": "subscribe", "method": "mesh/subscribe", "params": {}})
|
|
elif m == "tools/call":
|
|
say({"id": rid, "result": {"content": [{"type": "text", "text": json.dumps({"tool": TOOL})}]}})
|
|
elif m == "mesh/event":
|
|
body = (p.get("envelope") or {}).get("body") or {}
|
|
log(TOOL + " handled " + str(body.get("n")))
|
|
say({"id": rid, "result": {}})
|
|
elif rid is not None:
|
|
say({"id": rid, "result": {}})
|
|
`
|
|
}
|
|
|
|
// place puts a bundle serving tool at path the way a delivery does: written beside it, made
|
|
// executable, and renamed into place, its directories made as needed.
|
|
func place(t *testing.T, path, tool string) {
|
|
t.Helper()
|
|
placeScript(t, path, bundleServing(tool))
|
|
}
|
|
|
|
func placeScript(t *testing.T, path, script string) {
|
|
t.Helper()
|
|
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
tmp := path + ".new"
|
|
if err := os.WriteFile(tmp, []byte(script), 0o755); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := os.Rename(tmp, path); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
// fastSettle makes the watcher act on a burst of changes quickly in a test.
|
|
func fastSettle(t *testing.T) {
|
|
was := Settle
|
|
Settle = 50 * time.Millisecond
|
|
t.Cleanup(func() { Settle = was })
|
|
}
|
|
|
|
type reloadBed struct {
|
|
mesh *mt.Mesh
|
|
nodeTools *bus.Conn
|
|
asker *bus.Conn
|
|
logs *syncLogs
|
|
entry string
|
|
}
|
|
|
|
func newReloadBed(t *testing.T) *reloadBed {
|
|
t.Helper()
|
|
fastSettle(t)
|
|
mesh := mt.New(t)
|
|
mesh.Issue(t, mt.MembershipOf("ledger", "anchor", false, nil))
|
|
return &reloadBed{
|
|
mesh: mesh,
|
|
nodeTools: connect(t, "node-tools", "anchor"),
|
|
asker: connect(t, "console", "workstation"),
|
|
logs: &syncLogs{},
|
|
// Below directories that do not exist yet, as on a machine before the module's first delivery.
|
|
entry: filepath.Join(t.TempDir(), "bundles", "ledger", "tools", "serve.py"),
|
|
}
|
|
}
|
|
|
|
func (b *reloadBed) run(t *testing.T) {
|
|
t.Helper()
|
|
stop, err := Run(b.nodeTools, []Served{{"ledger", []string{b.entry}}}, nil, b.logs.Logf)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(stop)
|
|
}
|
|
|
|
func (b *reloadBed) answers(t *testing.T, tool string) {
|
|
t.Helper()
|
|
mt.Until(t, func() error {
|
|
got, err := call(t, b.asker, "ledger."+tool+"@anchor", map[string]any{})
|
|
if err != nil {
|
|
return fmt.Errorf("ledger.%s is not served: %v\n%s", tool, err, b.logs.All())
|
|
}
|
|
if !strings.Contains(got, `"tool":"`+tool+`"`) {
|
|
return fmt.Errorf("ledger.%s answered %s", tool, got)
|
|
}
|
|
return nil
|
|
})
|
|
}
|
|
|
|
func (b *reloadBed) silent(t *testing.T, tool string) {
|
|
t.Helper()
|
|
mt.Until(t, func() error {
|
|
if _, err := call(t, b.asker, "ledger."+tool+"@anchor", map[string]any{}); err == nil {
|
|
return fmt.Errorf("ledger.%s is still served\n%s", tool, b.logs.All())
|
|
}
|
|
return nil
|
|
})
|
|
}
|
|
|
|
func announced(t *testing.T) string {
|
|
t.Helper()
|
|
nc, err := nats.Connect(mt.URL(t))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer nc.Close()
|
|
msg, err := nc.Request("$SRV.INFO.node-tools.anchor", nil, 2*time.Second)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var i micro.Info
|
|
if err := json.Unmarshal(msg.Data, &i); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var out []string
|
|
for _, e := range i.Endpoints {
|
|
out = append(out, e.Subject)
|
|
}
|
|
return strings.Join(out, "\n")
|
|
}
|
|
|
|
// 1945/AC19: a bundle placed after the runtime started is served, and announced, with no restart.
|
|
func TestABundlePlacedAfterStartIsServed(t *testing.T) {
|
|
b := newReloadBed(t)
|
|
b.run(t)
|
|
if !b.logs.Has("ledger's bundle", "does not exist") {
|
|
t.Errorf("the missing bundle was not said as missing:\n%s", b.logs.All())
|
|
}
|
|
place(t, b.entry, "first")
|
|
b.answers(t, "first")
|
|
mt.Until(t, func() error {
|
|
if a := announced(t); !strings.Contains(a, "mesh.mod.ledger.tool.first.anchor") {
|
|
return fmt.Errorf("discovery does not announce the placed bundle's tool:\n%s", a)
|
|
}
|
|
return nil
|
|
})
|
|
}
|
|
|
|
// 1945/AC20: a bundle replaced while served is loaded again: its new tool is served, its old one no more.
|
|
func TestAReplacedBundleIsReloaded(t *testing.T) {
|
|
b := newReloadBed(t)
|
|
place(t, b.entry, "first")
|
|
b.run(t)
|
|
b.answers(t, "first")
|
|
place(t, b.entry, "second")
|
|
b.answers(t, "second")
|
|
b.silent(t, "first")
|
|
mt.Until(t, func() error {
|
|
a := announced(t)
|
|
if !strings.Contains(a, "mesh.mod.ledger.tool.second.anchor") || strings.Contains(a, "mesh.mod.ledger.tool.first.anchor") {
|
|
return fmt.Errorf("discovery does not follow the replaced bundle:\n%s", a)
|
|
}
|
|
return nil
|
|
})
|
|
|
|
// The bundle's whole directory swapped for a new one, as an archive unpacked beside it and renamed in.
|
|
dir := filepath.Dir(b.entry)
|
|
fresh := dir + ".unpacked"
|
|
place(t, filepath.Join(fresh, filepath.Base(b.entry)), "third")
|
|
if err := os.Rename(dir, dir+".old"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := os.Rename(fresh, dir); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
b.answers(t, "third")
|
|
b.silent(t, "second")
|
|
|
|
// And changed again inside the directory swapped in: the watch follows the new directory.
|
|
place(t, b.entry, "fourth")
|
|
b.answers(t, "fourth")
|
|
}
|
|
|
|
// A replaced bundle's old tool stays gone when the module's membership is issued again: a stopped tool
|
|
// leaves no membership handler behind to bind it once more, and a call on it starts no unmanaged second
|
|
// copy of the bundle (issue 1945, review round 2).
|
|
func TestAReplacedBundlesOldToolStaysGoneOnTheNextMembership(t *testing.T) {
|
|
b := newReloadBed(t)
|
|
place(t, b.entry, "first")
|
|
b.run(t)
|
|
b.answers(t, "first")
|
|
place(t, b.entry, "second")
|
|
b.answers(t, "second")
|
|
b.silent(t, "first")
|
|
|
|
b.mesh.Issue(t, mt.MembershipOf("ledger", "anchor", false, nil))
|
|
b.answers(t, "second")
|
|
// Long enough for the reissued membership to have been heard, before the old tool is asked.
|
|
time.Sleep(300 * time.Millisecond)
|
|
if got, err := call(t, b.asker, "ledger.first@anchor", map[string]any{}); err == nil {
|
|
t.Fatalf("ledger.first is served again after the membership was reissued, and answered %s\n%s", got, b.logs.All())
|
|
}
|
|
}
|
|
|
|
// 1945/AC21: a bundle removed while served is no longer served, and its module says why.
|
|
func TestARemovedBundleIsNoLongerServed(t *testing.T) {
|
|
b := newReloadBed(t)
|
|
place(t, b.entry, "first")
|
|
b.run(t)
|
|
b.answers(t, "first")
|
|
if err := os.RemoveAll(filepath.Dir(b.entry)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
b.silent(t, "first")
|
|
mt.Until(t, func() error {
|
|
got, err := call(t, b.asker, "ledger.tools@anchor", map[string]any{})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var answer ToolsAnswer
|
|
if err := json.Unmarshal([]byte(got), &answer); err != nil {
|
|
return err
|
|
}
|
|
if len(answer.Tools) != 0 || !strings.Contains(answer.Failed, "does not exist") {
|
|
return fmt.Errorf("ledger's tools after its bundle was removed: %s", got)
|
|
}
|
|
return nil
|
|
})
|
|
if a := announced(t); strings.Contains(a, "mesh.mod.ledger.tool.first") {
|
|
t.Errorf("discovery still announces the removed bundle's tool:\n%s", a)
|
|
}
|
|
}
|
|
|
|
// 1945/AC22: a bundle that cannot be started is said with its real reason: a missing file does not
|
|
// exist, and a file without an execute bit is not executable with its mode.
|
|
func TestABundleThatCannotStartSaysTheRealReason(t *testing.T) {
|
|
dir := t.TempDir()
|
|
missing := filepath.Join(dir, "nowhere", "serve.py")
|
|
if why := notStartable(missing); !strings.Contains(why, missing+" does not exist") || strings.Contains(why, "not executable") {
|
|
t.Errorf("a missing bundle said: %q", why)
|
|
}
|
|
plain := filepath.Join(dir, "serve.py")
|
|
if err := os.WriteFile(plain, []byte(bundleServing("first")), 0o644); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if why := notStartable(plain); !strings.Contains(why, plain+" is not executable (mode -rw-r--r--)") {
|
|
t.Errorf("a bundle without an execute bit said: %q", why)
|
|
}
|
|
if why := notStartable(dir); !strings.Contains(why, "is a directory") {
|
|
t.Errorf("a directory said: %q", why)
|
|
}
|
|
if err := os.Chmod(plain, 0o755); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if why := notStartable(plain); why != "" {
|
|
t.Errorf("a startable bundle said: %q", why)
|
|
}
|
|
}
|
|
|
|
// 1945/AC23: an event a module's consumer hears while the module is loaded again is offered again, never
|
|
// a panic of the runtime, and the bundle loaded in its place is handed the next offer.
|
|
func TestAnEventHeardDuringAReloadIsOfferedAgainNotAPanic(t *testing.T) {
|
|
c := &consumers{logf: func(string, ...any) {}, of: map[string]*moduleEvents{}}
|
|
var got []string
|
|
before := func(json.RawMessage) error { got = append(got, "before"); return nil }
|
|
old := &moduleEvents{delivers: []*func(json.RawMessage) error{&before}}
|
|
c.of["ledger"] = old
|
|
c.dropModule("ledger")
|
|
env := bus.Envelope{Key: "alpha.happened", Body: json.RawMessage(`{"n":1}`)}
|
|
hand := func(m *moduleEvents) (err error) {
|
|
defer func() {
|
|
if r := recover(); r != nil {
|
|
t.Fatalf("an event heard during a reload panicked the runtime: %v", r)
|
|
}
|
|
}()
|
|
return c.handOut("ledger", m, env)
|
|
}
|
|
if err := hand(old); err == nil {
|
|
t.Error("an event heard while the module is loaded again was taken, not offered again")
|
|
}
|
|
after := func(json.RawMessage) error { got = append(got, "after"); return nil }
|
|
fresh := &moduleEvents{delivers: []*func(json.RawMessage) error{&after}}
|
|
c.of["ledger"] = fresh
|
|
if err := hand(fresh); err != nil {
|
|
t.Errorf("the bundle loaded again was not handed the event: %v", err)
|
|
}
|
|
if err := hand(old); err == nil {
|
|
t.Error("the stopped consumer's handler handed an event on after the reload")
|
|
}
|
|
if strings.Join(got, ",") != "after" {
|
|
t.Errorf("handed to %v, want the bundle loaded again alone", got)
|
|
}
|
|
}
|
|
|
|
// 1945/AC23, whole: a module's events reach its bundle before a reload, around it, and after it.
|
|
func TestAModulesEventsReachItsBundleLoadedAgain(t *testing.T) {
|
|
fastSettle(t)
|
|
restore := bus.NakDelay
|
|
bus.NakDelay = 200 * time.Millisecond
|
|
t.Cleanup(func() { bus.NakDelay = restore })
|
|
mesh := mt.New(t)
|
|
mesh.Issue(t, mt.MembershipOf("ledger", "anchor", false, nil))
|
|
mesh.Consumer(t, "anchor", "ledger", []string{"mesh.mod.alpha.event.>"}, 2*time.Second)
|
|
nodeTools := connect(t, "node-tools", "anchor")
|
|
logs := &syncLogs{}
|
|
dir := t.TempDir()
|
|
log := filepath.Join(dir, "events.log")
|
|
entry := filepath.Join(dir, "bundles", "ledger", "tools", "serve.py")
|
|
has := func(line string) error {
|
|
b, _ := os.ReadFile(log)
|
|
for _, l := range strings.Split(string(b), "\n") {
|
|
if l == line {
|
|
return nil
|
|
}
|
|
}
|
|
return fmt.Errorf("no %q yet in:\n%s\nruntime:\n%s", line, b, logs.All())
|
|
}
|
|
placeScript(t, entry, bundleScript("first", true, false))
|
|
stop, err := Run(nodeTools, []Served{{"ledger", []string{entry}}}, map[string]map[string]string{"ledger": {"EVENT_LOG": log}}, logs.Logf)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(stop)
|
|
mt.Until(t, func() error { return has("first subscribed") })
|
|
mesh.Emit(t, "mesh.mod.alpha.event.happened", map[string]any{"n": 1})
|
|
mt.Until(t, func() error { return has("first handled 1") })
|
|
|
|
placeScript(t, entry, bundleScript("second", true, false))
|
|
mesh.Emit(t, "mesh.mod.alpha.event.happened", map[string]any{"n": 2})
|
|
mt.Until(t, func() error {
|
|
if has("first handled 2") != nil && has("second handled 2") != nil {
|
|
return has("second handled 2")
|
|
}
|
|
return nil
|
|
})
|
|
mt.Until(t, func() error { return has("second subscribed") })
|
|
mesh.Emit(t, "mesh.mod.alpha.event.happened", map[string]any{"n": 3})
|
|
mt.Until(t, func() error { return has("second handled 3") })
|
|
}
|
|
|
|
// 1945/AC24: a bundle that writes a file beside itself as it starts is loaded once at start and once
|
|
// when it is replaced, never again by its own writing.
|
|
func TestABundleWritingBesideItselfIsNotLoadedAgainByIt(t *testing.T) {
|
|
b := newReloadBed(t)
|
|
t.Setenv("EVENT_LOG", filepath.Join(t.TempDir(), "unused.log"))
|
|
placeScript(t, b.entry, bundleScript("first", false, true))
|
|
b.run(t)
|
|
b.answers(t, "first")
|
|
reloads := func() int {
|
|
return strings.Count(b.logs.All(), "loading it again")
|
|
}
|
|
time.Sleep(20 * Settle)
|
|
if n := reloads(); n != 0 {
|
|
t.Fatalf("the bundle was loaded again %d time(s) by its own writing:\n%s", n, b.logs.All())
|
|
}
|
|
placeScript(t, b.entry, bundleScript("second", false, true))
|
|
b.answers(t, "second")
|
|
time.Sleep(20 * Settle)
|
|
if n := reloads(); n != 1 {
|
|
t.Errorf("replaced once, loaded again %d time(s):\n%s", n, b.logs.All())
|
|
}
|
|
}
|