Serve a tools bundle as soon as it is in place, changed or gone, with no restart; say why a bundle cannot start (issue 1945)
node-tools loaded every bundle once at start, so a bundle delivered after it started (mesh-files on shanks, 2026-10-25 12:08) went unserved until a hand restart, and a missing file was said as "not executable". The runtime now watches the entrypoints' directories with inotify, loads a module again once a burst of changes settles, and discovery answers with what is served now. A bundle that cannot start says it does not exist, is not executable with its mode, or the exec error. Tests: 1945/AC19-AC22 in internal/runtime/reload_test.go.
This commit is contained in:
@@ -198,6 +198,18 @@ func (p *provisionWatches) told(module string, changed bus.ProvisionChanged) []*
|
||||
return out
|
||||
}
|
||||
|
||||
// stopModule stops every watch of one module, whose bundle is loaded again.
|
||||
func (p *provisionWatches) stopModule(module string) {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
for w := range p.of[module] {
|
||||
w.mu.Lock()
|
||||
w.stopped = true
|
||||
w.mu.Unlock()
|
||||
}
|
||||
delete(p.of, module)
|
||||
}
|
||||
|
||||
func (p *provisionWatches) stopAll() {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
|
||||
@@ -0,0 +1,281 @@
|
||||
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 `#!/usr/bin/env python3
|
||||
import json, sys
|
||||
def say(m):
|
||||
m["jsonrpc"] = "2.0"; sys.stdout.write(json.dumps(m) + "\n"); sys.stdout.flush()
|
||||
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")
|
||||
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}})
|
||||
elif m == "tools/call":
|
||||
say({"id": rid, "result": {"content": [{"type": "text", "text": json.dumps({"tool": "` + tool + `"})}]}})
|
||||
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()
|
||||
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
tmp := path + ".new"
|
||||
if err := os.WriteFile(tmp, []byte(bundleServing(tool)), 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)
|
||||
}
|
||||
}
|
||||
@@ -135,7 +135,51 @@ type registration struct {
|
||||
tools []launch.Tool
|
||||
}
|
||||
|
||||
// Run launches, binds and serves. It answers a stop function.
|
||||
// loadedModule is one served module as its bundles were last loaded: what they registered, what of it
|
||||
// is served, and each entrypoint's signature at that load, which tells the watch that it changed.
|
||||
type loadedModule struct {
|
||||
served Served
|
||||
seen map[string]string // entrypoint → its signature when it was loaded
|
||||
children []func()
|
||||
regs []registration // every registration its bundles made
|
||||
own []registration // those under the name of a module served here
|
||||
failed string
|
||||
names []string // the tools it serves, as module.tool
|
||||
serving []func() // its tools' subscriptions
|
||||
}
|
||||
|
||||
// unload stops serving the module's tools and stops its bundles.
|
||||
func (l *loadedModule) unload() {
|
||||
for i := len(l.serving) - 1; i >= 0; i-- {
|
||||
l.serving[i]()
|
||||
}
|
||||
l.serving, l.names = nil, nil
|
||||
for i := len(l.children) - 1; i >= 0; i-- {
|
||||
l.children[i]()
|
||||
}
|
||||
l.children = nil
|
||||
}
|
||||
|
||||
// runner is the runtime at work: every served module as last loaded, loaded again when its bundle
|
||||
// appears, changes or goes (issue 1945, note 23).
|
||||
type runner struct {
|
||||
conn *bus.Conn
|
||||
logf func(string, ...any)
|
||||
envFor func(module string) []string
|
||||
events *consumers
|
||||
modules []string
|
||||
isServed map[string]bool
|
||||
|
||||
mu sync.Mutex
|
||||
of map[string]*loadedModule
|
||||
verbs map[string]func() // each module's `tools` verb, where it is answered
|
||||
|
||||
seatsAgain func()
|
||||
stops []func() // the watch, the announcement and the seats, in the order they began
|
||||
}
|
||||
|
||||
// Run launches, binds and serves, and serves each module's bundle again when it appears, changes or
|
||||
// goes, with no restart. It answers a stop function.
|
||||
func Run(conn *bus.Conn, served []Served, envs map[string]map[string]string, logf func(string, ...any)) (func(), error) {
|
||||
if account := os.Getenv(OperatorAccount); account != "" {
|
||||
home := ""
|
||||
@@ -144,153 +188,323 @@ func Run(conn *bus.Conn, served []Served, envs map[string]map[string]string, log
|
||||
}
|
||||
logf("[mesh-tools] the operator's account here is %s%s", account, home)
|
||||
}
|
||||
modules := make([]string, 0, len(served))
|
||||
isServed := map[string]bool{}
|
||||
r := &runner{conn: conn, logf: logf, isServed: map[string]bool{}, of: map[string]*loadedModule{},
|
||||
verbs: map[string]func(){}}
|
||||
for _, s := range served {
|
||||
modules = append(modules, s.Module)
|
||||
isServed[s.Module] = true
|
||||
r.modules = append(r.modules, s.Module)
|
||||
r.isServed[s.Module] = true
|
||||
conn.Follow(s.Module)
|
||||
}
|
||||
node := conn.Node()
|
||||
|
||||
base := os.Environ()
|
||||
envFor := func(module string) []string { return BundleEnv(base, envs[module], module, node) }
|
||||
r.envFor = func(module string) []string { return BundleEnv(base, envs[module], module, node) }
|
||||
|
||||
// The module's events, for every child of it that subscribes (ADR 0198): one consumer per module,
|
||||
// bound the first time any of its children subscribes, each event handed to every child that did.
|
||||
events := &consumers{conn: conn, logf: logf, of: map[string]*moduleEvents{}}
|
||||
r.events = &consumers{conn: conn, logf: logf, of: map[string]*moduleEvents{}}
|
||||
|
||||
failed := map[string]string{}
|
||||
var registrations []registration
|
||||
var stops []func()
|
||||
for _, s := range served {
|
||||
for _, entry := range s.Entrypoints {
|
||||
path, _ := filepath.Abs(entry)
|
||||
module := s.Module
|
||||
fail := func(why string) {
|
||||
failed[module] = why
|
||||
logf("[mesh-tools] %s's bundle %s failed to load: %s; its tools are not served here", module, entry, why)
|
||||
}
|
||||
if !launch.Executable(path) {
|
||||
fail(path + " is not executable; a bundle the runtime serves is started, never imported, and its build makes it executable (novox/hq ADR 0193)")
|
||||
continue
|
||||
}
|
||||
child, err := launch.Start(module, path, envFor(module), events.forModule(module), logf)
|
||||
if err != nil {
|
||||
fail(err.Error())
|
||||
continue
|
||||
}
|
||||
stops = append(stops, child.Stop)
|
||||
for _, r := range child.Registrations {
|
||||
registrations = append(registrations, registration{module: r.Module, owner: module, tools: r.Tools})
|
||||
}
|
||||
r.of[s.Module] = r.load(s)
|
||||
}
|
||||
for _, m := range r.modules {
|
||||
if err := refusal(r.of[m].own); err != nil {
|
||||
r.stop()
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
claimed := map[string]bool{}
|
||||
for _, m := range modules {
|
||||
if mem := conn.Membership(m); mem != nil {
|
||||
for _, s := range mem.Seats {
|
||||
claimed[s.Seat] = true
|
||||
}
|
||||
}
|
||||
}
|
||||
var own []registration
|
||||
for _, r := range registrations {
|
||||
switch {
|
||||
case isServed[r.module]:
|
||||
own = append(own, r)
|
||||
case claimed[r.module]:
|
||||
default:
|
||||
logf(`[mesh-tools] %s registers tools under "%s", which is neither a module served here nor a seat one of them claims; not served until the mesh issues the claim`, r.owner, r.module)
|
||||
}
|
||||
}
|
||||
stopAll := func() {
|
||||
events.stopAll()
|
||||
for i := len(stops) - 1; i >= 0; i-- {
|
||||
stops[i]()
|
||||
}
|
||||
}
|
||||
for _, r := range own {
|
||||
seen := map[string]bool{}
|
||||
for _, t := range r.tools {
|
||||
if t.Name == ToolsVerb {
|
||||
stopAll()
|
||||
return nil, fmt.Errorf(`%s names a tool "%s", which is the verb the runtime answers for every module with what it serves (novox/hq ADR 0152) — refused, rename it`, r.module, ToolsVerb)
|
||||
}
|
||||
if seen[t.Name] {
|
||||
stopAll()
|
||||
return nil, fmt.Errorf("%s exposes two tools named %s — refused", r.module, t.Name)
|
||||
}
|
||||
seen[t.Name] = true
|
||||
}
|
||||
}
|
||||
|
||||
var names []string
|
||||
byModule := map[string][]launch.Tool{}
|
||||
for _, r := range own {
|
||||
for _, t := range r.tools {
|
||||
t := t
|
||||
// Told who called, from the subject the call arrived on, never from its body (novox/hq issue 365).
|
||||
stop, err := conn.HandleCalled(r.module+"."+t.Name, func(caller string, body json.RawMessage) (any, error) {
|
||||
return t.Run(caller, argsOf(body))
|
||||
})
|
||||
if err != nil {
|
||||
logf("[mesh-tools] cannot serve %s.%s: %v", r.module, t.Name, err)
|
||||
continue
|
||||
}
|
||||
names = append(names, r.module+"."+t.Name)
|
||||
stops = append(stops, stop)
|
||||
}
|
||||
byModule[r.module] = append(byModule[r.module], r.tools...)
|
||||
}
|
||||
for _, module := range modules {
|
||||
module := module
|
||||
tools := byModule[module]
|
||||
why := failed[module]
|
||||
if len(tools) == 0 && why == "" {
|
||||
continue // a pure-events module: silent, as it always was
|
||||
}
|
||||
stop, err := conn.Handle(module+"."+ToolsVerb, func(json.RawMessage) (any, error) {
|
||||
answer := ToolsAnswer{Module: module, Tools: []ListedTool{}, Failed: why}
|
||||
for _, t := range tools {
|
||||
answer.Tools = append(answer.Tools, ListedTool{Name: t.Name, Description: t.Description,
|
||||
Input: t.Input, Subjects: subjectsOf(conn, module, t.Name)})
|
||||
}
|
||||
return answer, nil
|
||||
})
|
||||
if err == nil {
|
||||
stops = append(stops, stop)
|
||||
}
|
||||
}
|
||||
|
||||
failedNames := make([]string, 0, len(failed))
|
||||
for _, m := range modules {
|
||||
if _, f := failed[m]; f {
|
||||
r.mu.Lock()
|
||||
var names, failedNames []string
|
||||
for _, m := range r.modules {
|
||||
r.serve(r.of[m])
|
||||
names = append(names, r.of[m].names...)
|
||||
if r.of[m].failed != "" {
|
||||
failedNames = append(failedNames, m)
|
||||
}
|
||||
}
|
||||
line := fmt.Sprintf("[mesh-tools] serving %d tool(s) for %d module(s): %s", len(names), len(modules), orNone(names))
|
||||
r.bindVerbs()
|
||||
r.mu.Unlock()
|
||||
|
||||
line := fmt.Sprintf("[mesh-tools] serving %d tool(s) for %d module(s): %s", len(names), len(r.modules), orNone(names))
|
||||
if len(failedNames) > 0 {
|
||||
line += fmt.Sprintf("; not serving %s, whose bundle(s) failed to load", strings.Join(failedNames, ", "))
|
||||
}
|
||||
logf("%s", line)
|
||||
stops = append(stops, serveSeats(conn, modules, registrations, logf))
|
||||
stopSeats, again := serveSeats(conn, r.modules, r.registrations, logf)
|
||||
r.seatsAgain = again
|
||||
r.stops = append(r.stops, stopSeats)
|
||||
|
||||
// **What it serves, it announces** (novox/hq ADR 0197): the NATS services protocol's discovery,
|
||||
// answered with what is served at the moment it is asked — re-served memberships included.
|
||||
// answered with what is served at the moment it is asked — re-served memberships and bundles loaded
|
||||
// again included, so a bundle served after start is announced as one served at start.
|
||||
announced, err := announce.Serve(conn, announce.Service{
|
||||
Name: conn.Module(), ID: instanceOf(conn),
|
||||
Description: "the mesh's tool runtime on " + node + ": every assigned module's tools and the seats they hold",
|
||||
Metadata: map[string]string{"node": node},
|
||||
}, func() []announce.Endpoint { return endpointsOf(conn, own, registrations, modules) })
|
||||
}, func() []announce.Endpoint { return endpointsOf(conn, r.own(), r.registrations(), r.modules) })
|
||||
if err != nil {
|
||||
logf("[mesh-tools] cannot announce what it serves: %v", err)
|
||||
} else {
|
||||
stops = append(stops, announced)
|
||||
r.stops = append(r.stops, announced)
|
||||
}
|
||||
|
||||
// **A bundle is served as soon as it is in place** (issue 1945, note 23): the entrypoints'
|
||||
// directories are watched, and a module whose bundle appears, changes or goes is loaded again.
|
||||
var paths []string
|
||||
for _, s := range served {
|
||||
for _, entry := range s.Entrypoints {
|
||||
path, _ := filepath.Abs(entry)
|
||||
paths = append(paths, path)
|
||||
}
|
||||
}
|
||||
if len(paths) > 0 {
|
||||
stopWatch, err := watchBundles(paths, r.reconcile, logf)
|
||||
if err != nil {
|
||||
logf("[mesh-tools] cannot watch the bundles: %v; a bundle placed or changed later is served only after a restart", err)
|
||||
} else {
|
||||
r.stops = append(r.stops, stopWatch)
|
||||
}
|
||||
}
|
||||
conn.Flush()
|
||||
return stopAll, nil
|
||||
return r.stop, nil
|
||||
}
|
||||
|
||||
// stop ends the watch, the announcement and the seats, unbinds the modules' consumers, and stops every
|
||||
// module's tools and bundles.
|
||||
func (r *runner) stop() {
|
||||
for i := len(r.stops) - 1; i >= 0; i-- {
|
||||
r.stops[i]()
|
||||
}
|
||||
r.stops = nil
|
||||
r.events.stopAll()
|
||||
r.mu.Lock()
|
||||
for _, stop := range r.verbs {
|
||||
stop()
|
||||
}
|
||||
r.verbs = map[string]func(){}
|
||||
loaded := make([]*loadedModule, 0, len(r.modules))
|
||||
for _, m := range r.modules {
|
||||
if l := r.of[m]; l != nil {
|
||||
loaded = append(loaded, l)
|
||||
}
|
||||
}
|
||||
r.mu.Unlock()
|
||||
for i := len(loaded) - 1; i >= 0; i-- {
|
||||
loaded[i].unload()
|
||||
}
|
||||
}
|
||||
|
||||
// claimed is every seat a served module's membership names.
|
||||
func (r *runner) claimed() map[string]bool {
|
||||
out := map[string]bool{}
|
||||
for _, m := range r.modules {
|
||||
if mem := r.conn.Membership(m); mem != nil {
|
||||
for _, s := range mem.Seats {
|
||||
out[s.Seat] = true
|
||||
}
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// load launches one module's bundles and learns what they register; it serves nothing yet.
|
||||
func (r *runner) load(s Served) *loadedModule {
|
||||
l := &loadedModule{served: s, seen: map[string]string{}}
|
||||
claimed := r.claimed()
|
||||
for _, entry := range s.Entrypoints {
|
||||
path, _ := filepath.Abs(entry)
|
||||
module := s.Module
|
||||
l.seen[path] = signatureOf(path)
|
||||
fail := func(why string) {
|
||||
l.failed = why
|
||||
r.logf("[mesh-tools] %s's bundle %s failed to load: %s; its tools are not served here", module, entry, why)
|
||||
}
|
||||
if why := notStartable(path); why != "" {
|
||||
fail(why)
|
||||
continue
|
||||
}
|
||||
child, err := launch.Start(module, path, r.envFor(module), r.events.forModule(module), r.logf)
|
||||
if err != nil {
|
||||
fail(err.Error())
|
||||
continue
|
||||
}
|
||||
l.children = append(l.children, child.Stop)
|
||||
for _, reg := range child.Registrations {
|
||||
got := registration{module: reg.Module, owner: module, tools: reg.Tools}
|
||||
l.regs = append(l.regs, got)
|
||||
switch {
|
||||
case r.isServed[got.module]:
|
||||
l.own = append(l.own, got)
|
||||
case claimed[got.module]:
|
||||
default:
|
||||
r.logf(`[mesh-tools] %s registers tools under "%s", which is neither a module served here nor a seat one of them claims; not served until the mesh issues the claim`, got.owner, got.module)
|
||||
}
|
||||
}
|
||||
}
|
||||
return l
|
||||
}
|
||||
|
||||
// refusal is why a module's registrations cannot be served: a tool named as the verb the runtime
|
||||
// answers for every module, or two tools of one name.
|
||||
func refusal(own []registration) error {
|
||||
for _, r := range own {
|
||||
seen := map[string]bool{}
|
||||
for _, t := range r.tools {
|
||||
if t.Name == ToolsVerb {
|
||||
return fmt.Errorf(`%s names a tool "%s", which is the verb the runtime answers for every module with what it serves (novox/hq ADR 0152) — refused, rename it`, r.module, ToolsVerb)
|
||||
}
|
||||
if seen[t.Name] {
|
||||
return fmt.Errorf("%s exposes two tools named %s — refused", r.module, t.Name)
|
||||
}
|
||||
seen[t.Name] = true
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// serve binds a loaded module's tools. r.mu is held.
|
||||
func (r *runner) serve(l *loadedModule) {
|
||||
for _, reg := range l.own {
|
||||
for _, t := range reg.tools {
|
||||
t := t
|
||||
// Told who called, from the subject the call arrived on, never from its body (novox/hq issue 365).
|
||||
stop, err := r.conn.HandleCalled(reg.module+"."+t.Name, func(caller string, body json.RawMessage) (any, error) {
|
||||
return t.Run(caller, argsOf(body))
|
||||
})
|
||||
if err != nil {
|
||||
r.logf("[mesh-tools] cannot serve %s.%s: %v", reg.module, t.Name, err)
|
||||
continue
|
||||
}
|
||||
l.names = append(l.names, reg.module+"."+t.Name)
|
||||
l.serving = append(l.serving, stop)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// toolsOf is every tool served under a module's name, and why its bundle failed, if it did. r.mu is held.
|
||||
func (r *runner) toolsOf(module string) ([]launch.Tool, string) {
|
||||
var tools []launch.Tool
|
||||
for _, m := range r.modules {
|
||||
for _, reg := range r.of[m].own {
|
||||
if reg.module == module {
|
||||
tools = append(tools, reg.tools...)
|
||||
}
|
||||
}
|
||||
}
|
||||
return tools, r.of[module].failed
|
||||
}
|
||||
|
||||
// bindVerbs answers each module's `tools` verb where it serves a tool or its bundle failed, and not
|
||||
// where neither is so (a pure-events module is silent, as it always was). r.mu is held.
|
||||
func (r *runner) bindVerbs() {
|
||||
for _, module := range r.modules {
|
||||
module := module
|
||||
tools, why := r.toolsOf(module)
|
||||
want := len(tools) > 0 || why != ""
|
||||
stop, bound := r.verbs[module]
|
||||
switch {
|
||||
case want && !bound:
|
||||
stop, err := r.conn.Handle(module+"."+ToolsVerb, func(json.RawMessage) (any, error) {
|
||||
r.mu.Lock()
|
||||
tools, why := r.toolsOf(module)
|
||||
r.mu.Unlock()
|
||||
answer := ToolsAnswer{Module: module, Tools: []ListedTool{}, Failed: why}
|
||||
for _, t := range tools {
|
||||
answer.Tools = append(answer.Tools, ListedTool{Name: t.Name, Description: t.Description,
|
||||
Input: t.Input, Subjects: subjectsOf(r.conn, module, t.Name)})
|
||||
}
|
||||
return answer, nil
|
||||
})
|
||||
if err == nil {
|
||||
r.verbs[module] = stop
|
||||
}
|
||||
case !want && bound:
|
||||
stop()
|
||||
delete(r.verbs, module)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// own is every registration served under a served module's name, now.
|
||||
func (r *runner) own() []registration {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
var out []registration
|
||||
for _, m := range r.modules {
|
||||
out = append(out, r.of[m].own...)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// registrations is every registration the bundles made, now.
|
||||
func (r *runner) registrations() []registration {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
var out []registration
|
||||
for _, m := range r.modules {
|
||||
out = append(out, r.of[m].regs...)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// reconcile loads again every module whose entrypoint appeared, changed or went since it was loaded.
|
||||
// The watch calls it once a burst of changes settles; it runs on the watch's one goroutine.
|
||||
func (r *runner) reconcile() {
|
||||
for _, m := range r.modules {
|
||||
r.mu.Lock()
|
||||
l := r.of[m]
|
||||
r.mu.Unlock()
|
||||
for _, entry := range l.served.Entrypoints {
|
||||
path, _ := filepath.Abs(entry)
|
||||
was, now := l.seen[path], signatureOf(path)
|
||||
if was == now {
|
||||
continue
|
||||
}
|
||||
what := "changed"
|
||||
switch {
|
||||
case was == "":
|
||||
what = "appeared"
|
||||
case now == "":
|
||||
what = "is gone"
|
||||
}
|
||||
r.reload(m, entry+" "+what)
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// reload stops a module's tools and bundles, loads them again, serves what they register now, and
|
||||
// serves its seats' verbs again; discovery answers with it from then on.
|
||||
func (r *runner) reload(module, what string) {
|
||||
r.mu.Lock()
|
||||
old := r.of[module]
|
||||
r.of[module] = &loadedModule{served: old.served, seen: old.seen}
|
||||
r.mu.Unlock()
|
||||
old.unload()
|
||||
r.events.dropModule(module)
|
||||
|
||||
r.logf("[mesh-tools] %s's bundle %s; loading it again", module, what)
|
||||
fresh := r.load(old.served)
|
||||
if err := refusal(fresh.own); err != nil {
|
||||
fresh.unload()
|
||||
fresh.regs, fresh.own, fresh.failed = nil, nil, err.Error()
|
||||
r.events.dropModule(module)
|
||||
r.logf("[mesh-tools] %s's bundle is refused: %v; its tools are not served here", module, err)
|
||||
}
|
||||
r.mu.Lock()
|
||||
r.of[module] = fresh
|
||||
r.serve(fresh)
|
||||
r.bindVerbs()
|
||||
r.mu.Unlock()
|
||||
if fresh.failed != "" {
|
||||
r.logf("[mesh-tools] not serving %s, whose bundle failed to load", module)
|
||||
} else {
|
||||
r.logf("[mesh-tools] serving %d tool(s) for %s: %s", len(fresh.names), module, orNone(fresh.names))
|
||||
}
|
||||
if r.seatsAgain != nil {
|
||||
r.seatsAgain()
|
||||
}
|
||||
r.conn.Flush()
|
||||
}
|
||||
|
||||
// instanceOf is this runtime's instance on the bus: its machine, which is what tells two instances of
|
||||
@@ -390,26 +604,31 @@ func subjectsOf(conn *bus.Conn, module, tool string) []string {
|
||||
|
||||
// serveSeats serves every verb of every seat a served module holds, where the mesh issued it, by
|
||||
// the tool of the same name registered under the seat's name (ADR 0159, 0160) — and serves again
|
||||
// whenever a membership changes. Whether this machine holds the seat is the bus's to decide.
|
||||
func serveSeats(conn *bus.Conn, modules []string, registrations []registration, logf func(string, ...any)) func() {
|
||||
impl := map[string]map[string]launch.Tool{}
|
||||
for _, r := range registrations {
|
||||
if impl[r.module] == nil {
|
||||
impl[r.module] = map[string]launch.Tool{}
|
||||
}
|
||||
for _, t := range r.tools {
|
||||
impl[r.module][t.Name] = t
|
||||
}
|
||||
}
|
||||
// whenever a membership changes, and when again is called because a bundle was loaded again. Whether
|
||||
// this machine holds the seat is the bus's to decide. It answers its stop and again.
|
||||
func serveSeats(conn *bus.Conn, modules []string, current func() []registration, logf func(string, ...any)) (func(), func()) {
|
||||
var mu sync.Mutex
|
||||
var stops []func()
|
||||
stopped := false
|
||||
serve := func() {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
if stopped {
|
||||
return
|
||||
}
|
||||
for _, s := range stops {
|
||||
s()
|
||||
}
|
||||
stops = nil
|
||||
impl := map[string]map[string]launch.Tool{}
|
||||
for _, r := range current() {
|
||||
if impl[r.module] == nil {
|
||||
impl[r.module] = map[string]launch.Tool{}
|
||||
}
|
||||
for _, t := range r.tools {
|
||||
impl[r.module][t.Name] = t
|
||||
}
|
||||
}
|
||||
have := map[string]bool{}
|
||||
for _, module := range modules {
|
||||
m := conn.Membership(module)
|
||||
@@ -447,7 +666,8 @@ func serveSeats(conn *bus.Conn, modules []string, registrations []registration,
|
||||
s()
|
||||
}
|
||||
stops = nil
|
||||
}
|
||||
stopped = true
|
||||
}, serve
|
||||
}
|
||||
|
||||
// consumers holds, per module the runtime serves, the one durable consumer its events arrive on and
|
||||
@@ -485,6 +705,24 @@ func (c *consumers) stopAll() {
|
||||
c.provisioned.stopAll()
|
||||
}
|
||||
|
||||
// dropModule unbinds one module's consumer, its seat traffic and its provision watches, so the bundle
|
||||
// loaded again in its place binds them anew and no event is handed to a child that was stopped.
|
||||
func (c *consumers) dropModule(module string) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
if m := c.of[module]; m != nil && m.stop != nil {
|
||||
m.stop()
|
||||
}
|
||||
delete(c.of, module)
|
||||
for key, stop := range c.seatBound {
|
||||
if strings.HasPrefix(key, module+" ") {
|
||||
stop()
|
||||
delete(c.seatBound, key)
|
||||
}
|
||||
}
|
||||
c.provisioned.stopModule(module)
|
||||
}
|
||||
|
||||
// forModule is the bus one launched bundle of a module reaches the mesh through.
|
||||
func (c *consumers) forModule(module string) launch.Bus {
|
||||
return &moduleBus{all: c, module: module}
|
||||
|
||||
@@ -209,7 +209,7 @@ func TestAnEntrypointThatIsNotExecutableIsRefused(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer stop()
|
||||
if !logs.Has("plain's bundle", "many-alpha.mjs failed to load:", "is not executable; a bundle the runtime serves is started, never imported") {
|
||||
if !logs.Has("plain's bundle", "many-alpha.mjs failed to load:", "is not executable (mode -", "); a bundle the runtime serves is started, never imported") {
|
||||
t.Errorf("the non-executable entrypoint was not refused by name:\n%s", logs.All())
|
||||
}
|
||||
got, err := call(t, asker, "alpha.one@anchor", map[string]any{})
|
||||
|
||||
@@ -0,0 +1,197 @@
|
||||
package runtime
|
||||
|
||||
import (
|
||||
"encoding/binary"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io/fs"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"golang.org/x/sys/unix"
|
||||
)
|
||||
|
||||
// Settle is how long the bundles' directories stay quiet before the runtime acts on what changed there:
|
||||
// a delivery unpacks a bundle in many writes, and the runtime loads it once, when they are done
|
||||
// (issue 1945, note 23).
|
||||
var Settle = 300 * time.Millisecond
|
||||
|
||||
// watchMask is what the runtime hears of a watched directory: a file or directory made, written,
|
||||
// renamed in or out, removed or given another mode, and the directory itself removed or renamed.
|
||||
const watchMask = unix.IN_CREATE | unix.IN_CLOSE_WRITE | unix.IN_MOVED_TO | unix.IN_MOVED_FROM | unix.IN_DELETE |
|
||||
unix.IN_ATTRIB | unix.IN_DELETE_SELF | unix.IN_MOVE_SELF | unix.IN_ONLYDIR
|
||||
|
||||
// bundleWatch hears, through inotify, the directory of every entrypoint the runtime serves, or for one
|
||||
// not there yet the nearest directory above it that is, and calls changed once a burst of changes has
|
||||
// settled. Nothing is looked at on a timer: between bursts the watch sleeps in poll(2).
|
||||
type bundleWatch struct {
|
||||
fd int
|
||||
wake [2]int // a pipe: a byte written to wake[1] ends the loop
|
||||
paths []string
|
||||
byWD map[int]string
|
||||
settle time.Duration
|
||||
changed func()
|
||||
logf func(string, ...any)
|
||||
done chan struct{}
|
||||
}
|
||||
|
||||
// watchBundles starts the watch of the entrypoints at paths. It answers its stop.
|
||||
func watchBundles(paths []string, changed func(), logf func(string, ...any)) (func(), error) {
|
||||
fd, err := unix.InotifyInit1(unix.IN_CLOEXEC | unix.IN_NONBLOCK)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("inotify: %w", err)
|
||||
}
|
||||
w := &bundleWatch{fd: fd, paths: paths, byWD: map[int]string{}, settle: Settle, changed: changed, logf: logf,
|
||||
done: make(chan struct{})}
|
||||
if err := unix.Pipe2(w.wake[:], unix.O_CLOEXEC|unix.O_NONBLOCK); err != nil {
|
||||
unix.Close(fd)
|
||||
return nil, fmt.Errorf("pipe: %w", err)
|
||||
}
|
||||
w.arm()
|
||||
go w.loop()
|
||||
return func() {
|
||||
_, _ = unix.Write(w.wake[1], []byte{1})
|
||||
<-w.done
|
||||
unix.Close(w.fd)
|
||||
unix.Close(w.wake[0])
|
||||
unix.Close(w.wake[1])
|
||||
}, nil
|
||||
}
|
||||
|
||||
// arm watches, for every entrypoint, its directory, or the nearest directory above it that exists. A
|
||||
// watch already held on the same directory is kept (inotify answers its descriptor again).
|
||||
func (w *bundleWatch) arm() {
|
||||
for _, p := range w.paths {
|
||||
dir := filepath.Dir(p)
|
||||
for {
|
||||
if info, err := os.Stat(dir); err == nil && info.IsDir() {
|
||||
break
|
||||
}
|
||||
up := filepath.Dir(dir)
|
||||
if up == dir {
|
||||
break
|
||||
}
|
||||
dir = up
|
||||
}
|
||||
wd, err := unix.InotifyAddWatch(w.fd, dir, watchMask)
|
||||
if err != nil {
|
||||
w.logf("[mesh-tools] cannot watch %s for %s's bundle: %v; a bundle placed or changed there is served only after a restart", dir, p, err)
|
||||
continue
|
||||
}
|
||||
w.byWD[wd] = dir
|
||||
}
|
||||
}
|
||||
|
||||
func (w *bundleWatch) loop() {
|
||||
defer close(w.done)
|
||||
buf := make([]byte, 64*1024)
|
||||
pending := false
|
||||
var quietAt time.Time
|
||||
for {
|
||||
timeout := -1
|
||||
if pending {
|
||||
timeout = int((time.Until(quietAt) + time.Millisecond - 1) / time.Millisecond)
|
||||
if timeout < 0 {
|
||||
timeout = 0
|
||||
}
|
||||
}
|
||||
fds := []unix.PollFd{{Fd: int32(w.fd), Events: unix.POLLIN}, {Fd: int32(w.wake[0]), Events: unix.POLLIN}}
|
||||
n, err := unix.Poll(fds, timeout)
|
||||
if errors.Is(err, unix.EINTR) {
|
||||
continue
|
||||
}
|
||||
if err != nil {
|
||||
w.logf("[mesh-tools] the watch of the bundles stopped: %v; a bundle placed or changed later is served only after a restart", err)
|
||||
return
|
||||
}
|
||||
if fds[1].Revents != 0 {
|
||||
return
|
||||
}
|
||||
if n == 0 {
|
||||
// Quiet for the settle: watch the directories that came to be, then act on what changed.
|
||||
pending = false
|
||||
w.arm()
|
||||
w.changed()
|
||||
continue
|
||||
}
|
||||
if fds[0].Revents&unix.POLLIN != 0 {
|
||||
w.drain(buf)
|
||||
pending = true
|
||||
quietAt = time.Now().Add(w.settle)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// drain reads every event queued, dropping the watches of directories that went away or moved: the
|
||||
// next arm watches what is at their path now.
|
||||
func (w *bundleWatch) drain(buf []byte) {
|
||||
for {
|
||||
n, err := unix.Read(w.fd, buf)
|
||||
if err != nil || n <= 0 {
|
||||
return
|
||||
}
|
||||
for off := 0; off+unix.SizeofInotifyEvent <= n; {
|
||||
wd := int(int32(binary.NativeEndian.Uint32(buf[off:])))
|
||||
mask := binary.NativeEndian.Uint32(buf[off+4:])
|
||||
length := int(binary.NativeEndian.Uint32(buf[off+12:]))
|
||||
switch {
|
||||
case mask&unix.IN_IGNORED != 0:
|
||||
delete(w.byWD, wd)
|
||||
case mask&unix.IN_MOVE_SELF != 0:
|
||||
_, _ = unix.InotifyRmWatch(w.fd, uint32(wd))
|
||||
delete(w.byWD, wd)
|
||||
}
|
||||
off += unix.SizeofInotifyEvent + length
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// signatureOf is what the runtime compares to tell that an entrypoint changed: the file's identity,
|
||||
// mode, size and times, and those of every file beside it (a launcher starts what its build wrote
|
||||
// next to it). "" is a file that does not exist.
|
||||
func signatureOf(path string) string {
|
||||
info, err := os.Stat(path)
|
||||
if errors.Is(err, fs.ErrNotExist) {
|
||||
return ""
|
||||
}
|
||||
if err != nil {
|
||||
return "unreadable: " + err.Error()
|
||||
}
|
||||
sig := statSignature(info)
|
||||
if entries, err := os.ReadDir(filepath.Dir(path)); err == nil {
|
||||
for _, e := range entries {
|
||||
if i, err := e.Info(); err == nil {
|
||||
sig += "|" + e.Name() + " " + statSignature(i)
|
||||
}
|
||||
}
|
||||
}
|
||||
return sig
|
||||
}
|
||||
|
||||
func statSignature(info fs.FileInfo) string {
|
||||
sig := fmt.Sprintf("%s %d %d", info.Mode(), info.Size(), info.ModTime().UnixNano())
|
||||
if st, ok := info.Sys().(*syscall.Stat_t); ok {
|
||||
sig += fmt.Sprintf(" %d:%d %d.%d", st.Dev, st.Ino, st.Ctim.Sec, st.Ctim.Nsec)
|
||||
}
|
||||
return sig
|
||||
}
|
||||
|
||||
// notStartable says why the runtime cannot start an entrypoint, in its real reason, or "" when it can
|
||||
// (issue 1945, note 23: a missing file was said as "not executable").
|
||||
func notStartable(path string) string {
|
||||
info, err := os.Stat(path)
|
||||
switch {
|
||||
case errors.Is(err, fs.ErrNotExist):
|
||||
return path + " does not exist; it is served as soon as it is in place"
|
||||
case err != nil:
|
||||
return path + " cannot be read: " + err.Error()
|
||||
case info.IsDir():
|
||||
return path + " is a directory, not a bundle's entrypoint"
|
||||
case info.Mode().Perm()&0o111 == 0:
|
||||
return fmt.Sprintf("%s is not executable (mode %s); a bundle the runtime serves is started, never imported, "+
|
||||
"and its build makes it executable (novox/hq ADR 0193)", path, info.Mode().Perm())
|
||||
}
|
||||
return ""
|
||||
}
|
||||
Reference in New Issue
Block a user