diff --git a/node-tools/internal/runtime/provision.go b/node-tools/internal/runtime/provision.go index c3c22b7..66a8126 100644 --- a/node-tools/internal/runtime/provision.go +++ b/node-tools/internal/runtime/provision.go @@ -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() diff --git a/node-tools/internal/runtime/reload_test.go b/node-tools/internal/runtime/reload_test.go new file mode 100644 index 0000000..f2a3d7e --- /dev/null +++ b/node-tools/internal/runtime/reload_test.go @@ -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) + } +} diff --git a/node-tools/internal/runtime/runtime.go b/node-tools/internal/runtime/runtime.go index 678e73a..7b9f434 100644 --- a/node-tools/internal/runtime/runtime.go +++ b/node-tools/internal/runtime/runtime.go @@ -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} diff --git a/node-tools/internal/runtime/runtime_test.go b/node-tools/internal/runtime/runtime_test.go index d962bed..eb56bfa 100644 --- a/node-tools/internal/runtime/runtime_test.go +++ b/node-tools/internal/runtime/runtime_test.go @@ -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{}) diff --git a/node-tools/internal/runtime/watch.go b/node-tools/internal/runtime/watch.go new file mode 100644 index 0000000..ebb4ce7 --- /dev/null +++ b/node-tools/internal/runtime/watch.go @@ -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 "" +}