jochen
ffe229308c
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.
581 lines
17 KiB
Go
581 lines
17 KiB
Go
// Package runtime is the node's tool runtime (novox/hq ADR 0175, ADR 0193), the Go port of
|
|
// node-tools' runtime.ts in its launch-only form: it launches every assigned module's bundle,
|
|
// serves each module's tools on that module's subjects and each held seat's verbs on the seat's,
|
|
// and answers for every module the verb that says what it serves. It imports nothing and knows no
|
|
// language.
|
|
package runtime
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"os"
|
|
"path/filepath"
|
|
"sort"
|
|
"strings"
|
|
"sync"
|
|
|
|
"github.com/novox/mesh-tools/node-tools/internal/announce"
|
|
"github.com/novox/mesh-tools/node-tools/internal/bus"
|
|
"github.com/novox/mesh-tools/node-tools/internal/launch"
|
|
)
|
|
|
|
// ToolsVerb is the verb every module's runtime answers for it (ADR 0152): its tools, from the code
|
|
// that answers them.
|
|
const ToolsVerb = "tools"
|
|
|
|
// Words the mesh sets for the runtime (ADR 0175, ADR 0192).
|
|
const (
|
|
ToolModules = "MESH_TOOL_MODULES"
|
|
ToolEnv = "MESH_TOOL_ENV"
|
|
OperatorAccount = "MESH_OPERATOR_ACCOUNT"
|
|
OperatorHome = "MESH_OPERATOR_HOME"
|
|
)
|
|
|
|
// Served is one module this runtime serves and its entrypoints.
|
|
type Served struct {
|
|
Module string
|
|
Entrypoints []string
|
|
}
|
|
|
|
// ToolsAnswer is what `tools` answers for one module.
|
|
type ToolsAnswer struct {
|
|
Module string `json:"module"`
|
|
Tools []ListedTool `json:"tools"`
|
|
Failed string `json:"failed,omitempty"`
|
|
}
|
|
|
|
// ListedTool is one tool as a module's `tools` answer lists it.
|
|
type ListedTool struct {
|
|
Name string `json:"name"`
|
|
Description string `json:"description"`
|
|
Input json.RawMessage `json:"input"`
|
|
Subjects []string `json:"subjects,omitempty"`
|
|
}
|
|
|
|
// ServedModulesFrom reads MESH_TOOL_MODULES: `<module>=<entrypoint>` entries, comma-separated, several
|
|
// per module. The one-module form — a bare path, or the runtime's own module — is the per-module
|
|
// containers' (to-be 38 WP4c) and refused here: the node's runtime imports nothing.
|
|
func ServedModulesFrom(spec, own string) ([]Served, error) {
|
|
order := []string{}
|
|
by := map[string][]string{}
|
|
for _, raw := range strings.Split(spec, ",") {
|
|
entry := strings.TrimSpace(raw)
|
|
if entry == "" {
|
|
continue
|
|
}
|
|
module, path, ok := strings.Cut(entry, "=")
|
|
module, path = strings.TrimSpace(module), strings.TrimSpace(path)
|
|
if !ok || module == "" || path == "" || module == own {
|
|
return nil, fmt.Errorf("%s: %q is not <module>=<entrypoint> of another module; the node's "+
|
|
"runtime launches the bundles it is given and imports nothing (novox/hq ADR 0193)", ToolModules, entry)
|
|
}
|
|
if _, seen := by[module]; !seen {
|
|
order = append(order, module)
|
|
}
|
|
by[module] = append(by[module], path)
|
|
}
|
|
out := make([]Served, 0, len(order))
|
|
for _, m := range order {
|
|
out = append(out, Served{Module: m, Entrypoints: by[m]})
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// TakeToolEnvs reads the composed environments (ADR 0192) and removes them from the process's, so no
|
|
// bundle finds another's there.
|
|
func TakeToolEnvs() (map[string]map[string]string, error) {
|
|
raw := os.Getenv(ToolEnv)
|
|
os.Unsetenv(ToolEnv)
|
|
out := map[string]map[string]string{}
|
|
if raw == "" {
|
|
return out, nil
|
|
}
|
|
var parsed map[string]map[string]any
|
|
if err := json.Unmarshal([]byte(raw), &parsed); err != nil {
|
|
return nil, fmt.Errorf(`%s is not JSON of the shape {"<module>": {"<word>": "<value>"}}: %w`, ToolEnv, err)
|
|
}
|
|
for module, words := range parsed {
|
|
own := map[string]string{}
|
|
for k, v := range words {
|
|
if s, ok := v.(string); ok {
|
|
own[k] = s
|
|
} else {
|
|
b, _ := json.Marshal(v)
|
|
own[k] = string(b)
|
|
}
|
|
}
|
|
out[module] = own
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
type registration struct {
|
|
module string // the module's own name, or a seat's
|
|
owner string // the module whose bundle made it
|
|
tools []launch.Tool
|
|
}
|
|
|
|
// Run launches, binds and serves. 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 := ""
|
|
if h := os.Getenv(OperatorHome); h != "" {
|
|
home = " (home " + h + ")"
|
|
}
|
|
logf("[mesh-tools] the operator's account here is %s%s", account, home)
|
|
}
|
|
modules := make([]string, 0, len(served))
|
|
isServed := map[string]bool{}
|
|
for _, s := range served {
|
|
modules = append(modules, s.Module)
|
|
isServed[s.Module] = true
|
|
conn.Follow(s.Module)
|
|
}
|
|
node := conn.Node()
|
|
|
|
base := os.Environ()
|
|
envFor := func(module string) []string {
|
|
words := map[string]string{}
|
|
for _, kv := range base {
|
|
if k, v, ok := strings.Cut(kv, "="); ok && k != ToolEnv {
|
|
words[k] = v
|
|
}
|
|
}
|
|
for k, v := range envs[module] {
|
|
words[k] = v
|
|
}
|
|
words["MESH_SERVED_MODULE"] = module
|
|
words["MESH_MODULE"] = module
|
|
if node != "" {
|
|
words["MESH_NODE"] = node
|
|
}
|
|
out := make([]string, 0, len(words))
|
|
for k, v := range words {
|
|
out = append(out, k+"="+v)
|
|
}
|
|
sort.Strings(out)
|
|
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()
|
|
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})
|
|
}
|
|
}
|
|
}
|
|
|
|
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
|
|
stop, err := conn.Handle(r.module+"."+t.Name, func(body json.RawMessage) (any, error) {
|
|
return t.Run(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 {
|
|
failedNames = append(failedNames, m)
|
|
}
|
|
}
|
|
line := fmt.Sprintf("[mesh-tools] serving %d tool(s) for %d module(s): %s", len(names), len(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))
|
|
|
|
// **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.
|
|
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) })
|
|
if err != nil {
|
|
logf("[mesh-tools] cannot announce what it serves: %v", err)
|
|
} else {
|
|
stops = append(stops, announced)
|
|
}
|
|
conn.Flush()
|
|
return stopAll, nil
|
|
}
|
|
|
|
// instanceOf is this runtime's instance on the bus: its machine, which is what tells two instances of
|
|
// one service apart; the connection's module where it has no machine.
|
|
func instanceOf(conn *bus.Conn) string {
|
|
if n := conn.Node(); n != "" {
|
|
return n
|
|
}
|
|
return conn.Module()
|
|
}
|
|
|
|
// endpointsOf is everything this runtime serves now: each served module's tools on every subject the
|
|
// mesh issued for them, and each held seat's verbs on the seat's subject (ADR 0197).
|
|
func endpointsOf(conn *bus.Conn, own []registration, registrations []registration, modules []string) []announce.Endpoint {
|
|
node := conn.Node()
|
|
var out []announce.Endpoint
|
|
for _, r := range own {
|
|
for _, t := range r.tools {
|
|
served := conn.ServedOn(r.module, t.Name)
|
|
interchangeable := false
|
|
for _, s := range served {
|
|
interchangeable = interchangeable || s.Subject == "mesh.mod."+r.module+".tool."+t.Name
|
|
}
|
|
for _, s := range served {
|
|
out = append(out, announce.Endpoint{Kind: announce.KindTool, Module: r.module, Tool: t.Name,
|
|
Node: node, Description: t.Description, Schema: t.Input, Interchangeable: interchangeable,
|
|
Subject: s.Subject, Queue: s.Queue})
|
|
}
|
|
}
|
|
}
|
|
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
|
|
}
|
|
}
|
|
have := map[string]bool{}
|
|
for _, module := range modules {
|
|
m := conn.Membership(module)
|
|
if m == nil {
|
|
continue
|
|
}
|
|
for _, v := range m.Seats {
|
|
t, ok := impl[v.Seat][v.Verb]
|
|
if !ok || have[v.Subject] {
|
|
continue
|
|
}
|
|
have[v.Subject] = true
|
|
scope := "mesh"
|
|
if node != "" && strings.HasSuffix(v.Subject, "."+node) {
|
|
scope = "node"
|
|
}
|
|
out = append(out, announce.Endpoint{Kind: announce.KindSeat, Module: module, Tool: v.Verb, Seat: v.Seat,
|
|
Scope: scope, Node: node, Description: t.Description, Schema: t.Input, Subject: v.Subject})
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
func orNone(names []string) string {
|
|
if len(names) == 0 {
|
|
return "(none)"
|
|
}
|
|
return strings.Join(names, ", ")
|
|
}
|
|
|
|
// argsOf is a call's arguments as the tool receives them: an object, `{}` for none.
|
|
func argsOf(body json.RawMessage) json.RawMessage {
|
|
trimmed := strings.TrimSpace(string(body))
|
|
if trimmed == "" || trimmed == "null" {
|
|
return json.RawMessage("{}")
|
|
}
|
|
return body
|
|
}
|
|
|
|
// subjectsOf is where a tool is answered as the mesh issued it: the plain subject first, then this
|
|
// machine's; nothing before a membership is issued.
|
|
func subjectsOf(conn *bus.Conn, module, tool string) []string {
|
|
m := conn.Membership(module)
|
|
if m == nil {
|
|
return nil
|
|
}
|
|
var plain, mine []string
|
|
for _, s := range m.Serves {
|
|
subject := strings.ReplaceAll(s.Subject, "{tool}", tool)
|
|
if s.Queue != "" {
|
|
plain = append(plain, subject)
|
|
} else {
|
|
mine = append(mine, subject)
|
|
}
|
|
}
|
|
return append(plain, mine...)
|
|
}
|
|
|
|
// 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
|
|
}
|
|
}
|
|
var mu sync.Mutex
|
|
var stops []func()
|
|
serve := func() {
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
for _, s := range stops {
|
|
s()
|
|
}
|
|
stops = nil
|
|
have := map[string]bool{}
|
|
for _, module := range modules {
|
|
m := conn.Membership(module)
|
|
if m == nil {
|
|
continue
|
|
}
|
|
for _, v := range m.Seats {
|
|
if have[v.Subject] {
|
|
continue
|
|
}
|
|
have[v.Subject] = true
|
|
t, ok := impl[v.Seat][v.Verb]
|
|
if !ok {
|
|
logf("[mesh-tools] %s claims %s and implements no %s, which that seat promises; not served", module, v.Seat, v.Verb)
|
|
continue
|
|
}
|
|
stop, err := conn.HandleSubject(v.Subject, func(body json.RawMessage) (any, error) {
|
|
return t.Run(argsOf(body))
|
|
})
|
|
if err != nil {
|
|
logf("[mesh-tools] cannot serve %s's %s on %s: %v", v.Seat, v.Verb, v.Subject, err)
|
|
continue
|
|
}
|
|
stops = append(stops, stop)
|
|
logf("[mesh-tools] serving %s's %s on %s, admitted where %s holds the seat", v.Seat, v.Verb, v.Subject, module)
|
|
}
|
|
}
|
|
}
|
|
serve()
|
|
conn.OnMembership(func(bus.Membership) { go func() { serve(); conn.Flush() }() })
|
|
return func() {
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
for _, s := range stops {
|
|
s()
|
|
}
|
|
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
|
|
}
|