jschoubben
f016c67a72
The service manager stops node-tools by signalling its whole group, so every bundle gets SIGTERM with it; the launcher took each exit for a fault and started the bundle again, for the 15 s the runtime took to stop. On ace's reboot of 2026-10-28 those restarted bundles dialled the system bus once polkit and the bus had stopped, the bus and polkit came back during the shutdown, nothing stopped them a second time, and the reboot waited 29 minutes on its job timeout. ShutDown, called first on SIGTERM, keeps any bundle from starting again, by a restart or by a call; a running one still answers.
253 lines
9.7 KiB
Go
253 lines
9.7 KiB
Go
// node-tools — the node's tool runtime, in Go (novox/hq ADR 0175, ADR 0193; to-be 38 WP4d).
|
|
//
|
|
// One process per machine: it connects to the bus on the node's credential, launches every assigned
|
|
// module's tools bundle and serves its tools and its seats' verbs; as the node-tools module — or
|
|
// wherever MESH_CONSOLE_LISTEN says — it is also the console, MCP over HTTP on loopback.
|
|
//
|
|
// MESH_BROKER_FILE the credential the mesh sealed to this machine for the runtime
|
|
// MESH_TOOL_MODULES <module>=<entrypoint>,… the bundles to launch
|
|
// MESH_TOOL_ENV {"<module>": {"<word>": "<value>"}} what each is given (ADR 0192)
|
|
// MESH_OPERATOR_ACCOUNT whose machine this is, and MESH_OPERATOR_HOME their home
|
|
// MESH_CONSOLE_LISTEN where the console listens, overriding 127.0.0.1:4270
|
|
package main
|
|
|
|
import (
|
|
"bytes"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"log"
|
|
"math/rand/v2"
|
|
"os"
|
|
"os/signal"
|
|
"syscall"
|
|
"time"
|
|
|
|
"github.com/novox/mesh-tools/node-tools/internal/alive"
|
|
"github.com/novox/mesh-tools/node-tools/internal/bus"
|
|
"github.com/novox/mesh-tools/node-tools/internal/console"
|
|
"github.com/novox/mesh-tools/node-tools/internal/launch"
|
|
"github.com/novox/mesh-tools/node-tools/internal/provide"
|
|
"github.com/novox/mesh-tools/node-tools/internal/runtime"
|
|
)
|
|
|
|
// runtimeModule is the module that is the node's tool runtime; on its credential, serving is also
|
|
// the console.
|
|
const runtimeModule = "node-tools"
|
|
|
|
// consoleListen is where the console listens when nothing says otherwise.
|
|
const consoleListen = "127.0.0.1:4270"
|
|
|
|
func main() {
|
|
log.SetFlags(0)
|
|
if len(os.Args) > 1 && os.Args[1] != "serve" {
|
|
fmt.Fprintf(os.Stderr, "node-tools: %q is not a mode; this runtime serves (novox/hq ADR 0193)\n", os.Args[1])
|
|
os.Exit(2)
|
|
}
|
|
until := time.Now().Add(startWait)
|
|
cred := credential(until)
|
|
conn := connectPatiently(cred, until)
|
|
|
|
served, err := runtime.ServedModulesFrom(os.Getenv(runtime.ToolModules), cred.Module)
|
|
if err != nil {
|
|
log.Fatalf("node-tools: %v", err)
|
|
}
|
|
envs, err := runtime.TakeToolEnvs()
|
|
if err != nil {
|
|
log.Fatalf("node-tools: %v", err)
|
|
}
|
|
stop, err := runtime.Run(conn, served, envs, log.Printf)
|
|
if err != nil {
|
|
log.Fatalf("node-tools: %v", err)
|
|
}
|
|
// And says it is there (novox/hq to-be 45 S11): the machine's runtime, on its machine's name. A
|
|
// runtime on another credential — a module's own, in a test — is not the machine's.
|
|
stopSaying := func() {}
|
|
if cred.Module == runtimeModule && cred.Node != "" {
|
|
stopSaying = alive.Keep(conn, cred.Node, alive.Every, log.Printf)
|
|
}
|
|
|
|
// As the machine's runtime it is also the provider of `mcp-endpoint`: it answers its consumers through the
|
|
// SDK's helper wherever it runs, whether or not the module is handed (novox/hq ADR 0349 clause 12, issue 791).
|
|
stopProviding := provide.Start(conn, cred.Module, cred.Node, time.Minute, log.Printf)
|
|
|
|
listen := os.Getenv("MESH_CONSOLE_LISTEN")
|
|
if listen == "" && cred.Module == runtimeModule {
|
|
listen = consoleListen
|
|
}
|
|
var up *console.Listening
|
|
if listen != "" {
|
|
node := cred.Node
|
|
if node == "" {
|
|
node = "?"
|
|
}
|
|
who := node + "." + cred.Module
|
|
up, err = console.Serve(console.NewSurface(conn, who), listen)
|
|
if err != nil {
|
|
log.Fatalf("node-tools: %v", err)
|
|
}
|
|
log.Printf("mesh console listening on http://%s/mcp as %s", up.Address, who)
|
|
}
|
|
|
|
signals := make(chan os.Signal, 1)
|
|
signal.Notify(signals, syscall.SIGTERM, syscall.SIGINT)
|
|
<-signals
|
|
// First of all: the service manager signalled every bundle with this process, and none is started again
|
|
// (issue 1126).
|
|
launch.ShutDown()
|
|
stopSaying()
|
|
stopProviding()
|
|
stop()
|
|
if up != nil {
|
|
_ = up.Close()
|
|
}
|
|
conn.Close()
|
|
}
|
|
|
|
// startWait is how long a runtime waits, from its start, for its broker credential to be on the machine and for the bus
|
|
// to accept the account it names (issue 1496): the node-engine starts a module's runtime before the provisioner's client
|
|
// fills the credential, and the bus carries a new account only after it reloads its users. Past it the runtime gives up,
|
|
// loudly; inside it the wait is said once.
|
|
var startWait = 5 * time.Minute
|
|
|
|
// say is a line in the runtime's journal.
|
|
func say(format string, args ...any) { fmt.Fprintf(os.Stderr, "node-tools: "+format+"\n", args...) }
|
|
|
|
// credential reads the credential the mesh delivered, waiting for it within the start's bound (issue 1496); one that can
|
|
// never be a credential (no url) is a fault of configuration, said and final.
|
|
func credential(until time.Time) bus.Credential {
|
|
file := os.Getenv("MESH_BROKER_FILE")
|
|
if file == "" {
|
|
if url := os.Getenv("MESH_BROKER_URL"); url != "" {
|
|
return bus.Credential{URL: url, Module: runtimeModule}
|
|
}
|
|
say("set MESH_BROKER_FILE (a sealed credential) or MESH_BROKER_URL — there is no broker to reach")
|
|
os.Exit(1)
|
|
}
|
|
cred, err := awaitCredential(file, until, say)
|
|
if err != nil {
|
|
say("%v", err)
|
|
os.Exit(1)
|
|
}
|
|
return cred
|
|
}
|
|
|
|
// errNotCredential is a file that is there and readable and still no broker credential: waiting will not fix it.
|
|
var errNotCredential = errors.New("it carries no url — it is not a broker credential")
|
|
|
|
// readCredential reads the credential at file. A file not there yet, empty, or not yet whole JSON is one the node-engine
|
|
// has still to write: the error says so and waiting may fix it. errNotCredential is final.
|
|
func readCredential(file string) (bus.Credential, error) {
|
|
raw, err := os.ReadFile(file)
|
|
if err != nil {
|
|
return bus.Credential{}, err
|
|
}
|
|
if len(bytes.TrimSpace(raw)) == 0 {
|
|
return bus.Credential{}, errors.New("the file is empty")
|
|
}
|
|
var cred bus.Credential
|
|
if err := json.Unmarshal(raw, &cred); err != nil {
|
|
return bus.Credential{}, err
|
|
}
|
|
if cred.URL == "" {
|
|
return bus.Credential{}, errNotCredential
|
|
}
|
|
return cred, nil
|
|
}
|
|
|
|
// awaitCredential is the credential at file, read at once and, while it is not there or not whole, again each time its
|
|
// directory changes (runtime.WatchFile, inotify), until the time until. The wait is said once; at until it ends with an
|
|
// error naming the file, how long it waited and what the last read found. A file that is not a credential ends it at once.
|
|
func awaitCredential(file string, until time.Time, say func(string, ...any)) (bus.Credential, error) {
|
|
cred, err := readCredential(file)
|
|
if err == nil {
|
|
return cred, nil
|
|
}
|
|
if errors.Is(err, errNotCredential) {
|
|
return bus.Credential{}, fmt.Errorf("%s: %w", file, err)
|
|
}
|
|
began := time.Now()
|
|
say("waits for its broker credential at %s (%v): the node-engine writes it once the provisioner's client fills it; "+
|
|
"it gives up at %s", file, err, until.Format("15:04:05"))
|
|
changed := make(chan struct{}, 1)
|
|
stop, werr := runtime.WatchFile(file, func() {
|
|
select {
|
|
case changed <- struct{}{}:
|
|
default:
|
|
}
|
|
}, func(format string, args ...any) { say(format, args...) },
|
|
fmt.Sprintf("the credential is read again every %s instead", rereadEvery))
|
|
if werr != nil {
|
|
say("cannot watch %s (%v): it is read again every %s instead", file, werr, rereadEvery)
|
|
} else {
|
|
defer stop()
|
|
}
|
|
for {
|
|
left := time.Until(until)
|
|
if left <= 0 {
|
|
return bus.Credential{}, fmt.Errorf("cannot read the broker credential at %s: waited %s for it, and the last read "+
|
|
"found: %v", file, time.Since(began).Round(time.Second), err)
|
|
}
|
|
// The watch wakes the read; the clock is only the backstop for a watch that could not be made or missed a change.
|
|
select {
|
|
case <-changed:
|
|
case <-time.After(min(left, rereadEvery)):
|
|
}
|
|
if cred, err = readCredential(file); err == nil {
|
|
say("read its broker credential at %s after %s", file, time.Since(began).Round(time.Second))
|
|
return cred, nil
|
|
}
|
|
if errors.Is(err, errNotCredential) {
|
|
return bus.Credential{}, fmt.Errorf("%s: %w", file, err)
|
|
}
|
|
}
|
|
}
|
|
|
|
// rereadEvery is the backstop of the wait for the credential, beside the watch of its directory.
|
|
var rereadEvery = 30 * time.Second
|
|
|
|
// connectPatiently retries while the bus is merely not reachable yet — the normal case at startup — and, until the start's
|
|
// bound, while the bus refuses the account: the bus carries a new account only after it reloads its users, which may come
|
|
// after the credential is on the machine (issue 1496, issue 1423). What waiting cannot fix (a pin that does not match, an
|
|
// address that is no URL) gives up at once (issue 058), and so does a refusal past the bound.
|
|
func connectPatiently(cred bus.Credential, until time.Time) *bus.Conn {
|
|
conn, err := connectWithin(cred, until, bus.Connect, time.Sleep, say)
|
|
if err != nil {
|
|
say("%v", err)
|
|
os.Exit(1)
|
|
}
|
|
return conn
|
|
}
|
|
|
|
// connectWithin is connectPatiently with its dial, its sleep and its words given, as a test gives them.
|
|
func connectWithin(cred bus.Credential, until time.Time, dial func(bus.Credential) (*bus.Conn, error), sleep func(time.Duration),
|
|
say func(string, ...any)) (*bus.Conn, error) {
|
|
refusedSaid := false
|
|
for delay := 2 * time.Second; ; delay = min(delay*2, 30*time.Second) {
|
|
conn, err := dial(cred)
|
|
if err == nil {
|
|
if refusedSaid {
|
|
say("the bus accepts the account now")
|
|
}
|
|
return conn, nil
|
|
}
|
|
wait := delay + time.Duration(rand.IntN(1000))*time.Millisecond
|
|
switch fatal := bus.Fatal(err); {
|
|
case bus.Refused(err) && time.Now().Add(wait).Before(until):
|
|
if !refusedSaid {
|
|
say("the bus refuses this account (%v): it may not carry it yet, since it loads a new account by a reload; "+
|
|
"retrying until %s", err, until.Format("15:04:05"))
|
|
refusedSaid = true
|
|
}
|
|
case bus.Refused(err) && refusedSaid:
|
|
return nil, fmt.Errorf("the bus still refuses this account (%v) at the end of the start's bound, %s; giving up", err,
|
|
until.Format("15:04:05"))
|
|
case fatal != "":
|
|
return nil, fmt.Errorf("%s — waiting will not fix this; giving up", fatal)
|
|
default:
|
|
say("the broker is not reachable yet (%v); retrying in %ds", err, int(wait.Round(time.Second)/time.Second))
|
|
}
|
|
sleep(wait)
|
|
}
|
|
}
|