A module's runtime waits for its broker credential and its account at a start, bounded, instead of exiting (issue 1496)

The node-engine starts a module's runtime before the provisioner's client fills
its broker file, and the bus carries a new account only after it reloads its
users. The runtime exited on both and was restarted until they were there (ace,
2026-10-26 21:19: mesh-build-node's and mesh-build-image's runtimes exited twice).
It now waits up to five minutes from its start, said once: the file is read again
each time its directory changes (the bundles' inotify watch, a 30 s backstop),
and an authorization refusal is asked again with the backoff. Past the bound it
gives up loudly; a file with no url, a pin mismatch and a bad URL stay final.
This commit is contained in:
2026-10-26 22:06:39 +01:00
parent a5097dc69a
commit 3baeaece61
4 changed files with 270 additions and 30 deletions
+132 -29
View File
@@ -12,7 +12,9 @@
package main
import (
"bytes"
"encoding/json"
"errors"
"fmt"
"log"
"math/rand/v2"
@@ -41,8 +43,9 @@ func main() {
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)
}
cred := credential()
conn := connectPatiently(cred)
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 {
@@ -97,48 +100,148 @@ func main() {
conn.Close()
}
// credential reads the credential the mesh delivered; a missing or unreadable one is a fault of
// configuration, said and final.
func credential() bus.Credential {
// 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}
}
fmt.Fprintln(os.Stderr, "node-tools: set MESH_BROKER_FILE (a sealed credential) or MESH_BROKER_URL — there is no broker to reach")
say("set MESH_BROKER_FILE (a sealed credential) or MESH_BROKER_URL — there is no broker to reach")
os.Exit(1)
}
raw, err := os.ReadFile(file)
cred, err := awaitCredential(file, until, say)
if err != nil {
fmt.Fprintf(os.Stderr, "node-tools: cannot read the broker credential at %s: %v\n", file, err)
os.Exit(1)
}
var cred bus.Credential
if err := json.Unmarshal(raw, &cred); err != nil {
fmt.Fprintf(os.Stderr, "node-tools: cannot read the broker credential at %s: %v\n", file, err)
os.Exit(1)
}
if cred.URL == "" {
fmt.Fprintf(os.Stderr, "node-tools: %s carries no url — it is not a broker credential\n", file)
say("%v", err)
os.Exit(1)
}
return cred
}
// connectPatiently retries while the bus is merely not reachable yet — the normal case at startup —
// and gives up at once on what waiting cannot fix (issue 058).
func connectPatiently(cred bus.Credential) *bus.Conn {
for delay := 2 * time.Second; ; delay = min(delay*2, 30*time.Second) {
conn, err := bus.Connect(cred)
if err == nil {
return conn
// 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:
}
if fatal := bus.Fatal(err); fatal != "" {
fmt.Fprintf(os.Stderr, "node-tools: %s — waiting will not fix this; giving up\n", fatal)
os.Exit(1)
}, func(format string, args ...any) { say(format, args...) })
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
fmt.Fprintf(os.Stderr, "node-tools: the broker is not reachable yet (%v); retrying in %ds\n", err, int(wait.Round(time.Second)/time.Second))
time.Sleep(wait)
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)
}
}
+122
View File
@@ -0,0 +1,122 @@
package main
import (
"errors"
"fmt"
"os"
"path/filepath"
"strings"
"testing"
"time"
"github.com/novox/mesh-tools/node-tools/internal/bus"
)
// A runtime started before its credential is on the machine waits for it, woken by the watch of its directory, and gives
// up loudly at its bound; a file that is no credential ends the wait at once (issue 1496: on ace mesh-build-node's and
// mesh-build-image's runtimes exited twice on "cannot read the broker credential ... no such file").
func TestAwaitCredential(t *testing.T) {
const good = `{"url":"tls://bus.example:4222","module":"m"}`
rereadEvery = time.Hour // only the watch can wake the read in time
t.Cleanup(func() { rereadEvery = 30 * time.Second })
place := func(path, body string) {
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
t.Fatal(err)
}
tmp := path + ".tmp"
if err := os.WriteFile(tmp, []byte(body), 0o600); err != nil {
t.Fatal(err)
}
if err := os.Rename(tmp, path); err != nil {
t.Fatal(err)
}
}
for _, c := range []struct {
name string
before string // written before the wait, "" none
later string // written 200ms into the wait, "" none
deeper bool // the file's directory is not there at the start
bound time.Duration
want string // "" a credential; else a word of the error
said int
}{
{name: "there at the start", before: good, bound: time.Second, said: 0},
{name: "written during the wait", later: good, bound: 5 * time.Second, said: 2},
{name: "empty, then filled", before: " ", later: good, bound: 5 * time.Second, said: 2},
{name: "its directory made during the wait", later: good, deeper: true, bound: 5 * time.Second, said: 2},
{name: "never written", bound: 400 * time.Millisecond, want: "waited", said: 1},
{name: "no url", before: `{"module":"m"}`, bound: 5 * time.Second, want: "not a broker credential", said: 0},
} {
t.Run(c.name, func(t *testing.T) {
dir := t.TempDir()
file := filepath.Join(dir, "broker")
if c.deeper {
file = filepath.Join(dir, "var", "lib", "m", "broker")
}
if c.before != "" {
place(file, c.before)
}
if c.later != "" {
go func() { time.Sleep(200 * time.Millisecond); place(file, c.later) }()
}
var said []string
began := time.Now()
cred, err := awaitCredential(file, time.Now().Add(c.bound), func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) })
switch {
case c.want == "" && (err != nil || cred.URL == ""):
t.Fatalf("no credential: %v", err)
case c.want != "" && (err == nil || !strings.Contains(err.Error(), c.want)):
t.Fatalf("error %v, want one saying %q", err, c.want)
case c.want == "" && c.later != "" && time.Since(began) > 3*time.Second:
t.Fatalf("read after %s: the watch did not wake it", time.Since(began))
}
if len(said) != c.said {
t.Fatalf("said %d lines, want %d: %q", len(said), c.said, said)
}
})
}
}
// A bus that refuses the account at a start is asked again until the start's bound, said once; a refusal past it, a pin
// that does not match and an address that is no URL give up (issue 1496, issue 058).
func TestConnectWithin(t *testing.T) {
refused := errors.New("nats: Authorization Violation")
unreachable := errors.New("nats: no servers available for connection")
for _, c := range []struct {
name string
errs []error // what each dial answers, then a connection
bound time.Duration
fails string
saidAt int // lines said
}{
{name: "refused twice, then accepted", errs: []error{refused, refused}, bound: time.Hour, saidAt: 2},
{name: "unreachable, then accepted", errs: []error{unreachable}, bound: time.Hour, saidAt: 1},
{name: "refused past the bound", errs: []error{refused, refused}, bound: 0, fails: "refused this account"},
{name: "refused, then the bound ends", errs: []error{refused, refused, refused, refused, refused, refused}, bound: 5 * time.Second,
fails: "still refuses"},
{name: "pin", errs: []error{fmt.Errorf("tls: %w", bus.ErrPin)}, bound: time.Hour, fails: "pin"},
} {
t.Run(c.name, func(t *testing.T) {
dials := 0
dial := func(bus.Credential) (*bus.Conn, error) {
dials++
if dials <= len(c.errs) {
return nil, c.errs[dials-1]
}
return nil, nil
}
until := time.Now().Add(c.bound)
var said []string
_, err := connectWithin(bus.Credential{URL: "tls://bus.example:4222"}, until, dial, func(time.Duration) {},
func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) })
switch {
case c.fails == "" && err != nil:
t.Fatalf("gave up: %v", err)
case c.fails != "" && (err == nil || !strings.Contains(err.Error(), c.fails)):
t.Fatalf("error %v, want one saying %q", err, c.fails)
case c.fails == "" && len(said) != c.saidAt:
t.Fatalf("said %d lines, want %d: %q", len(said), c.saidAt, said)
}
})
}
}
+8 -1
View File
@@ -129,12 +129,19 @@ func Fatal(err error) string {
strings.Contains(msg, "nats: invalid url") {
return "the bus address is not a usable URL"
}
if regexp.MustCompile(`(?i)authorization violation|user authentication expired|permissions violation`).MatchString(msg) {
if Refused(err) || regexp.MustCompile(`(?i)user authentication expired|permissions violation`).MatchString(msg) {
return "the bus refused this account"
}
return ""
}
// Refused says whether the bus refused the account a connection named (an authorization violation): final for a running
// runtime, but at a start possibly only "not yet", since the bus carries a new account after it reloads its users (issue
// 1496). Fatal still names it; the caller decides how long a start waits.
func Refused(err error) bool {
return err != nil && regexp.MustCompile(`(?i)authorization violation`).MatchString(err.Error())
}
func normalizeFingerprint(f string) string {
f = strings.TrimSpace(f)
if len(f) > 7 && strings.EqualFold(f[:7], "sha256:") {
+8
View File
@@ -37,6 +37,14 @@ type bundleWatch struct {
done chan struct{}
}
// WatchFile hears, as the bundles are heard, the directory of the file at path, or the nearest directory above it that
// exists, and calls changed once at its start and once after each burst of changes there has settled: a runtime waiting
// for its broker credential reads it again when the node-engine writes it, never on a clock (issue 1496). It answers its
// stop.
func WatchFile(path string, changed func(), logf func(string, ...any)) (func(), error) {
return watchBundles([]string{path}, changed, logf)
}
// 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)