mesh-tools/node-tools/internal/runtime/reload_test.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 {
	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{
		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")
}

// 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())
	}
}