Merge pull request 'node-tools serves a tools bundle as soon as it is placed, changed or removed, with no restart, and says why a bundle cannot start (issue 1945)' (#42) from fix/r6-reload-bundles into main

node-tools/internal/bus/bus.go

@@ -602,9 +602,13 @@ func (c *Conn) HandleCalled(key string, h CalledHandler) (func(), error) {
 	}
 	var mu sync.Mutex
 	var stops []func()
+	stopped := false
 	serve := func() {
 		mu.Lock()
 		defer mu.Unlock()
+		if stopped {
+			return
+		}
 		for _, s := range stops {
 			s()
 		}
@@ -618,14 +622,18 @@ func (c *Conn) HandleCalled(key string, h CalledHandler) (func(), error) {
 		}
 	}
 	serve()
-	c.OnMembership(func(m Membership) {
+	// Its stop drops the membership handler too: a handler left behind would bind the stopped tool again
+	// on the module's next membership, and call a handler whose bundle was replaced (issue 1945).
+	unfollow := c.onMembership(func(m Membership) {
 		if m.Module == module {
 			serve()
 		}
 	})
 	return func() {
+		unfollow()
 		mu.Lock()
 		defer mu.Unlock()
+		stopped = true
 		for _, s := range stops {
 			s()
 		}

node-tools/internal/bus/hearing_stopped_test.go

@@ -0,0 +1,48 @@
+package bus
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// heardMsg is a delivered event as a consumer's loop hands it, saying how it was settled.
+type heardMsg struct {
+	jetstream.Msg
+	mu      sync.Mutex
+	settled []string
+}
+
+func (m *heardMsg) Subject() string                  { return "mesh.mod.alpha.event.happened" }
+func (m *heardMsg) Headers() nats.Header             { return nats.Header{} }
+func (m *heardMsg) Data() []byte                     { return []byte(`{"n":1}`) }
+func (m *heardMsg) InProgress() error                { return nil }
+func (m *heardMsg) Ack() error                       { return m.say("ack") }
+func (m *heardMsg) Nak() error                       { return m.say("nak") }
+func (m *heardMsg) NakWithDelay(time.Duration) error { return m.say("nak-delayed") }
+func (m *heardMsg) Term() error                      { return m.say("term") }
+func (m *heardMsg) say(how string) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.settled = append(m.settled, how)
+	return nil
+}
+
+// An event the bus hands a hearing after it was stopped (a module's bundle loaded again, issue 1945) is
+// offered again at once and never handed to the module: the module's handler that heard it may be gone.
+func TestAHearingStoppedHandsNoEventOnAndOffersItAgain(t *testing.T) {
+	delivered := 0
+	h := newHearing(nil, nil, "alpha", func(Envelope) error { delivered++; return nil })
+	h.stop()
+	msg := &heardMsg{}
+	h.handle(msg)
+	if delivered != 0 {
+		t.Errorf("a stopped hearing handed the event to the module")
+	}
+	if len(msg.settled) != 1 || msg.settled[0] != "nak" {
+		t.Errorf("a stopped hearing settled the event as %v, want it offered again at once (nak)", msg.settled)
+	}
+}

node-tools/internal/bus/streams.go

@@ -6,6 +6,7 @@ import (
 	"sort"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/nats-io/nats.go/jetstream"
@@ -86,6 +87,9 @@ type hearing struct {
 	// serial hands the module one event at a time across its consumers (novox/hq issue 387).
 	serial sync.Mutex
 
+	// halted is stopped, read by handle without mu: the bus's loops hand events while mu is held.
+	halted atomic.Bool
+
 	mu      sync.Mutex
 	stopped bool
 	bound   map[HeardIssued]jetstream.ConsumeContext
@@ -154,11 +158,21 @@ func (h *hearing) handle(msg jetstream.Msg) {
 		}
 	}()
 	h.serial.Lock()
-	err := h.deliver(env)
+	halted := h.halted.Load()
+	var err error
+	if !halted {
+		err = h.deliver(env)
+	}
 	h.serial.Unlock()
 	close(done)
 	// No in-progress after the answer: one after a Nak would stretch its delay to the whole ack wait.
 	<-stopped
+	if halted {
+		// **A hearing stopped hands nothing on** (issue 1945): the module's bundle that subscribed may be
+		// gone, loaded again in its place; offered again at once, to the consumer bound after it.
+		_ = msg.Nak()
+		return
+	}
 	if err != nil {
 		_ = msg.NakWithDelay(NakDelay)
 		return
@@ -214,6 +228,7 @@ func (h *hearing) stop() {
 		return
 	}
 	h.stopped = true
+	h.halted.Store(true)
 	close(h.done)
 	if h.unhear != nil {
 		h.unhear()

node-tools/internal/launch/launch.go

@@ -174,6 +174,11 @@ func (c *child) write(m any) error {
 
 var stackLine = regexp.MustCompile(`^\s+at\s`)
 
+// colour is a terminal's colour or style sequence, which node writes into a stack trace when its
+// environment asks for colour (FORCE_COLOR): taken out before a line is judged or kept as a bundle's
+// reason, so the reason reads the same in every environment.
+var colour = regexp.MustCompile("\x1b\\[[0-9;]*m")
+
 // Start launches one bundle and learns its tools. It fails when the child cannot be started or does
 // not complete the handshake. A child that exits later is started again on its next call.
 func Start(module, entry string, env []string, mesh Bus, logf func(string, ...any)) (*Launched, error) {
@@ -245,9 +250,10 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an
 					continue
 				}
 				logf("[%s] %s", module, line)
-				if !stackLine.MatchString(line) && !strings.HasPrefix(line, "Node.js v") {
+				plain := colour.ReplaceAllString(line, "")
+				if !stackLine.MatchString(plain) && !strings.HasPrefix(plain, "Node.js v") {
 					saidMu.Lock()
-					lastSaid = strings.TrimSpace(line)
+					lastSaid = strings.TrimSpace(plain)
 					saidMu.Unlock()
 				}
 			}
@@ -456,14 +462,24 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an
 
 	asking := func(method string, params any, timeout time.Duration) (json.RawMessage, error) {
 		mu.Lock()
-		c := current
+		c, gone := current, stopped
 		mu.Unlock()
+		if gone {
+			// A stopped bundle is never started again by a call: a process started now would be one
+			// nothing stops (issue 1945).
+			return nil, fmt.Errorf("%s's bundle is stopped; it answers no call", module)
+		}
 		if c == nil {
 			fresh, err := start()
 			if err != nil {
 				return nil, err
 			}
 			mu.Lock()
+			if stopped {
+				mu.Unlock()
+				_ = fresh.cmd.Process.Signal(syscall.SIGTERM)
+				return nil, fmt.Errorf("%s's bundle is stopped; it answers no call", module)
+			}
 			if current == nil {
 				current = fresh
 				startedAt = time.Now()

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

node-tools/internal/runtime/reload_test.go

@@ -0,0 +1,442 @@
+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())
+	}
+}

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,324 @@ 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()
+	// Its events are refused first, so one heard while the old bundle stops is refused at once.
+	r.events.dropModule(module)
+	old.unload()
+
+	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 +605,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 +667,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 +706,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}
@@ -564,23 +803,7 @@ func (b *moduleBus) Subscribe(deliver func(json.RawMessage) error) error {
 		return nil
 	}
 	module := b.module
-	stop, err := c.conn.ConsumeAs(module, func(env bus.Envelope) error {
-		raw, err := json.Marshal(env)
-		if err != nil {
-			return err
-		}
-		c.mu.Lock()
-		targets := append([]*func(json.RawMessage) error(nil), c.of[module].delivers...)
-		c.mu.Unlock()
-		for _, d := range targets {
-			if err := (*d)(raw); err != nil {
-				c.logf("[mesh-tools] %s did not take %s%s: %s; offered again in %s", module, env.Key,
-					eventRef(env), whyNotTaken(err), bus.NakDelay)
-				return err
-			}
-		}
-		return nil
-	})
+	stop, err := c.conn.ConsumeAs(module, func(env bus.Envelope) error { return c.handOut(module, m, env) })
 	if err != nil {
 		return err
 	}
@@ -589,6 +812,35 @@ func (b *moduleBus) Subscribe(deliver func(json.RawMessage) error) error {
 	return nil
 }
 
+// errModuleReloading refuses an event heard by a module's consumer that was stopped for a reload.
+var errModuleReloading = errors.New("the module's bundle is being loaded again; offered again")
+
+// handOut hands one event of a module's consumer m to every child of the module that subscribed; an
+// event of a consumer that is no longer the module's is refused.
+func (c *consumers) handOut(module string, m *moduleEvents, env bus.Envelope) error {
+	raw, err := json.Marshal(env)
+	if err != nil {
+		return err
+	}
+	c.mu.Lock()
+	if c.of[module] != m {
+		// Heard by a consumer stopped while the module is loaded again (issue 1945): refused, so the bus
+		// offers it again, to the bundle loaded in its place once it subscribes.
+		c.mu.Unlock()
+		return errModuleReloading
+	}
+	targets := append([]*func(json.RawMessage) error(nil), m.delivers...)
+	c.mu.Unlock()
+	for _, d := range targets {
+		if err := (*d)(raw); err != nil {
+			c.logf("[mesh-tools] %s did not take %s%s: %s; offered again in %s", module, env.Key,
+				eventRef(env), whyNotTaken(err), bus.NakDelay)
+			return err
+		}
+	}
+	return nil
+}
+
 // whyNotTaken says why a bundle did not take an event, so the line names who failed: the module's
 // handler, in its own words — an error it answered — or the runtime not reaching it. A bundle's
 // handler error read bare ("Unexpected end of JSON input") looked like the runtime's (issue 276).

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{})

node-tools/internal/runtime/watch.go

@@ -0,0 +1,206 @@
+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). A directory
+// made below the watched one between its stat and its watch would be heard by nobody, so after each watch
+// the path is looked at again, and a deeper directory that came to be is watched in turn.
+func (w *bundleWatch) arm() {
+	for _, p := range w.paths {
+		target := filepath.Dir(p)
+		watched := ""
+		for {
+			dir := nearestDir(target)
+			if dir == watched {
+				break
+			}
+			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)
+				break
+			}
+			w.byWD[wd] = dir
+			watched = dir
+		}
+	}
+}
+
+// nearestDir is dir when it is a directory, else the nearest directory above it.
+func nearestDir(dir string) string {
+	for {
+		if info, err := os.Stat(dir); err == nil && info.IsDir() {
+			return dir
+		}
+		up := filepath.Dir(dir)
+		if up == dir {
+			return dir
+		}
+		dir = up
+	}
+}
+
+func (w *bundleWatch) loop() {
+	defer close(w.done)
+	buf := make([]byte, 64*1024)
+	// Looked at once as the watch begins: a bundle that came between its load and the watch is not missed.
+	pending := true
+	quietAt := time.Now()
+	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; "" is a file that does not exist. Only the entrypoint: a delivery puts every file
+// of a bundle in place anew (a directory swapped in, or each file renamed in), so the entrypoint is new
+// whenever its build is, while a file beside it may be one the bundle itself writes (a cache, a socket),
+// which would load it again after every quiet spell, and modules sharing a directory each other.
+func signatureOf(path string) string {
+	info, err := os.Stat(path)
+	if errors.Is(err, fs.ErrNotExist) {
+		return ""
+	}
+	if err != nil {
+		return "unreadable: " + err.Error()
+	}
+	return statSignature(info)
+}
+
+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 ""
+}