The node's runtime is its modules' bus: it binds each module's consumer and hands its events to the bundle (hq ADR 0198)

A launched bundle's mesh/subscribe binds the module's own durable consumer — EVENTS, <node>_<module>, by name as the module's own runtime bound it, so nothing is lost or replayed in the move — and every event goes to each child of the module that subscribed as mesh/event, acknowledged only when all answered, negatively acknowledged after a short delay when one failed or died, terminated when it is not an event. mesh/ask calls a tool as the module. Every launched bundle is started again when it exits, with backoff, since long-running code waits for no call. Requires SDK 0.1.6.
2026-10-03 22:29:20 +02:00
parent df4f492a72
commit ffe229308c
9 changed files with 601 additions and 19 deletions
@@ -0,0 +1,157 @@
+package runtime
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/novox/mesh-tools/node-tools/internal/bus"
+	"github.com/novox/mesh-tools/node-tools/internal/launch"
+	mt "github.com/novox/mesh-tools/node-tools/internal/meshtest"
+)
+
+// A logger safe to write from the runtime's goroutines.
+type lines struct {
+	mu sync.Mutex
+	l  []string
+}
+
+func (s *lines) logf(format string, args ...any) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.l = append(s.l, strings.TrimSpace(sprintf(format, args...)))
+}
+
+func (s *lines) all() string {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return strings.Join(s.l, "\n")
+}
+
+func sprintf(format string, args ...any) string { return fmtSprintf(format, args...) }
+
+func countLines(t *testing.T, path, want string) int {
+	t.Helper()
+	b, _ := os.ReadFile(path)
+	n := 0
+	for _, l := range strings.Split(string(b), "\n") {
+		if l == want {
+			n++
+		}
+	}
+	return n
+}
+
+// novox/hq ADR 0198: a module's long-running code is launched by the node's runtime, and the runtime
+// is its bus — it binds the module's own consumer, hands each event to the bundle, and acknowledges it
+// only when the bundle has taken it; one the bundle failed or died on is offered again.
+func TestTheRuntimeHandsAModulesEventsToItsBundleAndAcknowledgesThemOnlyWhenTaken(t *testing.T) {
+	mesh := mt.New(t)
+	restore := bus.NakDelay
+	bus.NakDelay = 300 * time.Millisecond
+	t.Cleanup(func() { bus.NakDelay = restore })
+	firstPause := launch.RestartFirst
+	launch.RestartFirst = 100 * time.Millisecond
+	t.Cleanup(func() { launch.RestartFirst = firstPause })
+
+	mesh.Issue(t, mt.MembershipOf("watcher", "anchor", false, nil))
+	mesh.Issue(t, mt.MembershipOf("beta", "anchor", true, nil))
+	// The controller's consumer for the module, as its own runtime bound it: anchor_watcher on EVENTS.
+	mesh.Consumer(t, "anchor", "watcher", []string{"mesh.mod.alpha.event.>"}, 2*time.Second)
+
+	nodeTools := connect(t, "node-tools", "anchor")
+	asker := connect(t, "console", "workstation")
+	dir := t.TempDir()
+	log := filepath.Join(dir, "watch.log")
+	said := &lines{}
+	stop, err := Run(nodeTools, []Served{
+		{Module: "watcher", Entrypoints: []string{mt.Fixture("watcher.serve.mjs")}},
+		{Module: "beta", Entrypoints: []string{mt.Fixture("many-beta.serve.mjs")}},
+	}, map[string]map[string]string{"watcher": {"WATCH_LOG": log}}, said.logf)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(stop)
+	if !strings.Contains(said.all(), "watcher's events arrive on its consumer anchor_watcher") {
+		t.Fatalf("the module's consumer was not bound:\n%s", said.all())
+	}
+
+	// Handled once, acknowledged once.
+	mesh.Emit(t, "mesh.mod.alpha.event.happened", map[string]any{"n": 1})
+	mt.Until(t, func() error {
+		if countLines(t, log, "handled 1") != 1 {
+			return errorf("event 1 not handled yet")
+		}
+		return nil
+	})
+	// The handler fails the first time: not acknowledged, offered again, then handled.
+	mesh.Emit(t, "mesh.mod.alpha.event.happened", map[string]any{"n": 2, "fail": true})
+	// The bundle dies on the first offer: not acknowledged, the bundle is started again and the
+	// event is offered again once its ack wait has passed.
+	mesh.Emit(t, "mesh.mod.alpha.event.happened", map[string]any{"n": 3, "die": true})
+	deadline := time.Now().Add(20 * time.Second)
+	for time.Now().Before(deadline) {
+		if countLines(t, log, "handled 2") == 1 && countLines(t, log, "handled 3") == 1 {
+			break
+		}
+		time.Sleep(200 * time.Millisecond)
+	}
+	if countLines(t, log, "handled 2") != 1 || countLines(t, log, "handled 3") != 1 {
+		b, _ := os.ReadFile(log)
+		t.Fatalf("a failed or interrupted event was not offered again and handled once:\n%s\nruntime:\n%s", b, said.all())
+	}
+	if countLines(t, log, "started") < 2 {
+		t.Errorf("the bundle that died was not started again: %s", said.all())
+	}
+	mt.Until(t, func() error {
+		ack, notYet := mesh.Pending(t, "anchor", "watcher")
+		if ack != 0 || notYet != 0 {
+			return errorf("still pending: %d unacknowledged, %d undelivered", ack, notYet)
+		}
+		return nil
+	})
+	if countLines(t, log, "handled 1") != 1 {
+		t.Errorf("event 1 was handled more than once")
+	}
+
+	// A tool asks another module's tool through the runtime, as the module.
+	got, err := call(t, asker, "watcher.relay@anchor", map[string]any{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	same(t, got, `{"beta":3}`)
+}
+
+// A long-running bundle that exits is started again, without waiting for a call (ADR 0198).
+func TestALongRunningBundleThatExitsIsStartedAgain(t *testing.T) {
+	mesh := mt.New(t)
+	firstPause := launch.RestartFirst
+	launch.RestartFirst = 100 * time.Millisecond
+	t.Cleanup(func() { launch.RestartFirst = firstPause })
+	mesh.Issue(t, mt.MembershipOf("flaky", "anchor", false, nil))
+	nodeTools := connect(t, "node-tools", "anchor")
+	log := filepath.Join(t.TempDir(), "flaky.log")
+	said := &lines{}
+	stop, err := Run(nodeTools, []Served{{Module: "flaky", Entrypoints: []string{mt.Fixture("flaky.serve.mjs")}}},
+		map[string]map[string]string{"flaky": {"FLAKY_LOG": log}}, said.logf)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(stop)
+	mt.Until(t, func() error {
+		if countLines(t, log, "started") < 3 {
+			return errorf("started %d time(s)", countLines(t, log, "started"))
+		}
+		return nil
+	})
+	if !strings.Contains(said.all(), "flaky's bundle exited (3)") || !strings.Contains(said.all(), "started again in") {
+		t.Errorf("the restart was not said:\n%s", said.all())
+	}
+}
+
+func fmtSprintf(format string, args ...any) string { return fmt.Sprintf(format, args...) }
+func errorf(format string, args ...any) error      { return fmt.Errorf(format, args...) }
@@ -157,6 +157,10 @@ func Run(conn *bus.Conn, served []Served, envs map[string]map[string]string, log
 		return out
 	}

+	// 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{}}
+
 	failed := map[string]string{}
 	var registrations []registration
 	var stops []func()
@@ -172,13 +176,7 @@ func Run(conn *bus.Conn, served []Served, envs map[string]map[string]string, log
 				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), func(params json.RawMessage) error {
-				var env bus.Envelope
-				if err := json.Unmarshal(params, &env); err != nil {
-					return fmt.Errorf("not an event envelope: %w", err)
-				}
-				return conn.PublishAs(module, env)
-			}, logf)
+			child, err := launch.Start(module, path, envFor(module), events.forModule(module), logf)
 			if err != nil {
 				fail(err.Error())
 				continue
@@ -209,6 +207,7 @@ func Run(conn *bus.Conn, served []Served, envs map[string]map[string]string, log
 		}
 	}
 	stopAll := func() {
+		events.stopAll()
 		for i := len(stops) - 1; i >= 0; i-- {
 			stops[i]()
 		}
@@ -450,3 +449,132 @@ func serveSeats(conn *bus.Conn, modules []string, registrations []registration,
 		stops = nil
 	}
 }
+
+// consumers holds, per module the runtime serves, the one durable consumer its events arrive on and
+// the children its events are handed to (novox/hq ADR 0198).
+type consumers struct {
+	conn *bus.Conn
+	logf func(string, ...any)
+	mu   sync.Mutex
+	of   map[string]*moduleEvents
+}
+
+type moduleEvents struct {
+	stop     func()
+	delivers []*func(json.RawMessage) error
+}
+
+// stopAll unbinds every module's consumer.
+func (c *consumers) stopAll() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	for _, m := range c.of {
+		if m.stop != nil {
+			m.stop()
+		}
+	}
+	c.of = map[string]*moduleEvents{}
+}
+
+// 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}
+}
+
+type moduleBus struct {
+	all     *consumers
+	module  string
+	mu      sync.Mutex
+	deliver *func(json.RawMessage) error
+}
+
+// Publish emits an event as the module (ADR 0193).
+func (b *moduleBus) Publish(params json.RawMessage) error {
+	var env bus.Envelope
+	if err := json.Unmarshal(params, &env); err != nil {
+		return fmt.Errorf("not an event envelope: %w", err)
+	}
+	return b.all.conn.PublishAs(b.module, env)
+}
+
+// Ask calls a tool as the module: `{key, body}`, answered with the tool's result (ADR 0198).
+func (b *moduleBus) Ask(params json.RawMessage) (json.RawMessage, error) {
+	var asked struct {
+		Key  string          `json:"key"`
+		Body json.RawMessage `json:"body"`
+	}
+	if err := json.Unmarshal(params, &asked); err != nil || asked.Key == "" {
+		return nil, fmt.Errorf("mesh/ask names no tool: {key, body}")
+	}
+	body := any(asked.Body)
+	if len(asked.Body) == 0 {
+		body = map[string]any{}
+	}
+	answered, err := b.all.conn.AskAs(b.module, asked.Key, body)
+	if err != nil {
+		return nil, err
+	}
+	if len(answered.Result) == 0 {
+		return json.RawMessage("null"), nil
+	}
+	return answered.Result, nil
+}
+
+// Subscribe hands this bundle the module's events. The consumer is bound once per module; each of
+// the module's children that subscribed is handed every event, and the event is acknowledged only
+// when all of them took it — one consumer split between two readers would give each half.
+func (b *moduleBus) Subscribe(deliver func(json.RawMessage) error) error {
+	b.mu.Lock()
+	if b.deliver == nil {
+		d := deliver
+		b.deliver = &d
+	} else {
+		*b.deliver = deliver
+	}
+	mine := b.deliver
+	b.mu.Unlock()
+
+	c := b.all
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	m := c.of[b.module]
+	if m == nil {
+		m = &moduleEvents{}
+		c.of[b.module] = m
+	}
+	listed := false
+	for _, d := range m.delivers {
+		if d == mine {
+			listed = true
+		}
+	}
+	if !listed {
+		m.delivers = append(m.delivers, mine)
+	}
+	if m.stop != nil {
+		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: %v; offered again", module, env.Key, err)
+				return err
+			}
+		}
+		return nil
+	})
+	if err != nil {
+		return err
+	}
+	m.stop = stop
+	c.logf("[mesh-tools] %s's events arrive on its consumer %s", module, bus.ConsumerOf(c.conn.Node(), module))
+	return nil
+}