From 3baeaece6115cd8781a8ddd006eaca234501a81c Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 26 Oct 2026 22:03:04 +0100 Subject: [PATCH] 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. --- node-tools/cmd/node-tools/main.go | 161 ++++++++++++++++++++----- node-tools/cmd/node-tools/main_test.go | 122 +++++++++++++++++++ node-tools/internal/bus/bus.go | 9 +- node-tools/internal/runtime/watch.go | 8 ++ 4 files changed, 270 insertions(+), 30 deletions(-) create mode 100644 node-tools/cmd/node-tools/main_test.go diff --git a/node-tools/cmd/node-tools/main.go b/node-tools/cmd/node-tools/main.go index 2eef656..ab68a25 100644 --- a/node-tools/cmd/node-tools/main.go +++ b/node-tools/cmd/node-tools/main.go @@ -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) } } diff --git a/node-tools/cmd/node-tools/main_test.go b/node-tools/cmd/node-tools/main_test.go new file mode 100644 index 0000000..63aa778 --- /dev/null +++ b/node-tools/cmd/node-tools/main_test.go @@ -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) + } + }) + } +} diff --git a/node-tools/internal/bus/bus.go b/node-tools/internal/bus/bus.go index 20591a8..33dea0e 100644 --- a/node-tools/internal/bus/bus.go +++ b/node-tools/internal/bus/bus.go @@ -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:") { diff --git a/node-tools/internal/runtime/watch.go b/node-tools/internal/runtime/watch.go index 7ae673b..6486011 100644 --- a/node-tools/internal/runtime/watch.go +++ b/node-tools/internal/runtime/watch.go @@ -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)