From d8ac69904de3bab61bbe5a4a9ba0e44b5553afc1 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 15:26:02 +0100 Subject: [PATCH] Refuse an event heard during a reload instead of panicking, and judge a bundle by its entrypoint alone (issue 1945, review) An event a module's consumer heard after the module was dropped for a reload read a deleted entry and panicked node-tools, taking every tool on the machine down. The handler now refuses an event of a consumer that is no longer the module's, and a stopped hearing hands nothing on and offers the event again at once. A bundle that writes beside itself (a cache, a socket) loaded itself again after every quiet spell, because the signature took in every file in its directory. A delivery puts every file in place anew, so the entrypoint alone tells a new build. The watch also looks once as it begins, so a bundle that came between its load and the watch is not missed. Tests: 1945/AC23 and AC24 in internal/runtime/reload_test.go, and a stopped hearing in internal/bus/hearing_stopped_test.go. --- .../internal/bus/hearing_stopped_test.go | 48 ++++++ node-tools/internal/bus/streams.go | 17 +- node-tools/internal/runtime/reload_test.go | 150 +++++++++++++++++- node-tools/internal/runtime/runtime.go | 47 ++++-- node-tools/internal/runtime/watch.go | 21 +-- 5 files changed, 246 insertions(+), 37 deletions(-) create mode 100644 node-tools/internal/bus/hearing_stopped_test.go diff --git a/node-tools/internal/bus/hearing_stopped_test.go b/node-tools/internal/bus/hearing_stopped_test.go new file mode 100644 index 0000000..f004ea2 --- /dev/null +++ b/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) + } +} diff --git a/node-tools/internal/bus/streams.go b/node-tools/internal/bus/streams.go index 34a03c5..b0ce8e7 100644 --- a/node-tools/internal/bus/streams.go +++ b/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() diff --git a/node-tools/internal/runtime/reload_test.go b/node-tools/internal/runtime/reload_test.go index f2a3d7e..109a1e2 100644 --- a/node-tools/internal/runtime/reload_test.go +++ b/node-tools/internal/runtime/reload_test.go @@ -55,20 +55,49 @@ func (l *syncLogs) All() string { } // bundleServing is a tools bundle, MCP over stdio with no SDK, whose one tool is named tool. -func bundleServing(tool string) string { +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 " subscribed" and " handled " 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, sys +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() -TOOLS = [{"name": "` + tool + `", "description": "answers its name", "inputSchema": {"type": "object", "properties": {}}}] +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") + 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 + `"})}]}}) + 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": {}}) ` @@ -77,12 +106,17 @@ for line in sys.stdin: // 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(bundleServing(tool)), 0o755); err != nil { + if err := os.WriteFile(tmp, []byte(script), 0o755); err != nil { t.Fatal(err) } if err := os.Rename(tmp, path); err != nil { @@ -279,3 +313,107 @@ func TestABundleThatCannotStartSaysTheRealReason(t *testing.T) { 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()) + } +} diff --git a/node-tools/internal/runtime/runtime.go b/node-tools/internal/runtime/runtime.go index 7b9f434..d7b013c 100644 --- a/node-tools/internal/runtime/runtime.go +++ b/node-tools/internal/runtime/runtime.go @@ -802,23 +802,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 } @@ -827,6 +811,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). diff --git a/node-tools/internal/runtime/watch.go b/node-tools/internal/runtime/watch.go index ebb4ce7..d51c347 100644 --- a/node-tools/internal/runtime/watch.go +++ b/node-tools/internal/runtime/watch.go @@ -87,8 +87,9 @@ func (w *bundleWatch) arm() { func (w *bundleWatch) loop() { defer close(w.done) buf := make([]byte, 64*1024) - pending := false - var quietAt time.Time + // 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 { @@ -149,8 +150,10 @@ func (w *bundleWatch) drain(buf []byte) { } // 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. +// 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) { @@ -159,15 +162,7 @@ func signatureOf(path string) string { 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 + return statSignature(info) } func statSignature(info fs.FileInfo) string {