diff --git a/node-tools/internal/bus/bus.go b/node-tools/internal/bus/bus.go index ec1648b..920189e 100644 --- a/node-tools/internal/bus/bus.go +++ b/node-tools/internal/bus/bus.go @@ -195,6 +195,8 @@ type Conn struct { calls callers // waits are the waits in flight by subject, each told when the bus refuses its subscription (wait.go). waits map[string]map[chan string]bool + // inboxSaid is when a refused answer publish was last said, by subject (answerRefused). + inboxSaid map[string]time.Time } // answered is one subject the runtime answers on: what to subscribe again with, and how often it was @@ -232,6 +234,9 @@ func (c *Conn) refused(err error) { if c.publishRefused(err) { return } + if c.answerRefused(err) { + return + } if subject := subjectRefused(err); subject != "" && c.waitRefused(subject) { return } @@ -1215,3 +1220,38 @@ func keyOf(subject string) string { } return event } + +// answerRefused says a refused publish of an answer (to a caller's inbox: a permission, or the single-use +// allowance of a reply expired) once per subject per minute, never silently (issue 1966: a caller waited out its +// time for an answer the bus had refused). It reports whether the refusal was one. +func (c *Conn) answerRefused(err error) bool { + m := refusedPublish.FindStringSubmatch(err.Error()) + if len(m) < 2 || !strings.HasPrefix(strings.Trim(m[1], `"`), "_INBOX.") { + return false + } + // Keyed without the reply's last token (a new one per request), so a caller's burst is one line. + subject := strings.Trim(m[1], `"`) + if dot := strings.LastIndex(subject, "."); dot > len("_INBOX.") { + subject = subject[:dot] + } + c.mu.Lock() + if c.inboxSaid == nil { + c.inboxSaid = map[string]time.Time{} + } + now := time.Now() + for k, at := range c.inboxSaid { + if now.Sub(at) >= time.Minute { + delete(c.inboxSaid, k) + } + } + _, seen := c.inboxSaid[subject] + say := !seen + if say { + c.inboxSaid[subject] = now + } + c.mu.Unlock() + if say && c.Logf != nil { + c.Logf("[mesh-tools] the bus refused this account an answer to %s (%v); the caller gets none", subject, err) + } + return true +} diff --git a/node-tools/internal/bus/refused_test.go b/node-tools/internal/bus/refused_test.go index 026db51..c9c3481 100644 --- a/node-tools/internal/bus/refused_test.go +++ b/node-tools/internal/bus/refused_test.go @@ -84,3 +84,33 @@ func TestARefusedSubscriptionIsAskedForAgain(t *testing.T) { t.Error("a stopped subject is still tracked") } } + +// Issue 1966: a refused publish of an answer (to a caller's inbox) is said, once per subject per minute, never dropped +// in silence while the caller waits out its time. +func TestARefusedAnswerIsSaidOncePerMinute(t *testing.T) { + c := &Conn{} + var said []string + c.Logf = func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) } + refuse := func(subject string) bool { + return c.answerRefused(fmt.Errorf("%w: Permissions Violation for Publish to %q", nats.ErrPermissionViolation, subject)) + } + // A reply subject ends in a token new to every request: a burst under one prefix is one line. + if !refuse("_INBOX.user.abc.t1") || !refuse("_INBOX.user.abc.t2") || !refuse("_INBOX.user.def.t1") { + t.Fatal("a refused answer was not taken for one") + } + if len(said) != 2 { + t.Fatalf("want one line per caller, got %v", said) + } + c.inboxSaid["_INBOX.user.abc"] = time.Now().Add(-2 * time.Minute) + c.inboxSaid["_INBOX.user.old"] = time.Now().Add(-2 * time.Minute) + refuse("_INBOX.user.abc.t3") + if len(said) != 3 || !strings.Contains(said[2], "_INBOX.user.abc") { + t.Fatalf("a minute later it is said again, got %v", said) + } + if _, kept := c.inboxSaid["_INBOX.user.old"]; kept || len(c.inboxSaid) != 2 { + t.Fatalf("entries older than a minute are dropped on write, kept %v", c.inboxSaid) + } + if refuse("mesh.mod.x.tool.y") { + t.Fatal("a refused publish that is no answer was taken for one") + } +} diff --git a/node-tools/internal/launch/launch.go b/node-tools/internal/launch/launch.go index d494da2..d293a19 100644 --- a/node-tools/internal/launch/launch.go +++ b/node-tools/internal/launch/launch.go @@ -206,11 +206,33 @@ type child struct { // restart decided — which Stop waits for (novox/hq issue 476). handled chan struct{} why error + // endedFor is why the runner itself ended the bundle (a line over the cap); its calls fail with it. + endedFor string // watches are the state watches this child asked for, stopped when it exits (ADR 0201): a child // that starts again watches again, as its code runs again. watches []func() } +// maxLine is the longest line the runner reads from a bundle; a variable so a test need not write megabytes. +var maxLine = 16 << 20 + +// errNoAnswer is a call the bundle did not answer within its bound. +var errNoAnswer = errors.New("the bundle did not answer in time") + +// readCapped reads one line of at most max bytes. tooLong says it was longer; the rest is not read. +func readCapped(r *bufio.Reader, max int) (line []byte, tooLong bool, err error) { + for { + part, isPrefix, e := r.ReadLine() + line = append(line, part...) + if len(line) > max { + return nil, true, nil + } + if e != nil || !isPrefix { + return line, false, e + } + } +} + func (c *child) write(m any) error { b, err := wire.Marshal(m) if err != nil { @@ -279,6 +301,7 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an // The restart pauses are read once, here: the exit handling runs on its own goroutine and must not // read the package's settings while a caller (a test) changes them (novox/hq issue 476). pauseFirst, pauseMost, settle := RestartFirst, RestartMost, RestartSettle + lineCap := maxLine pause := pauseFirst start := func() (*child, error) { @@ -324,10 +347,25 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an }() go func() { defer close(c.handled) - scan := bufio.NewScanner(stdout) - scan.Buffer(make([]byte, 64*1024), 16<<20) - for scan.Scan() { - line := strings.TrimSpace(scan.Text()) + rd := bufio.NewReaderSize(stdout, 64*1024) + for { + raw, tooLong, rerr := readCapped(rd, lineCap) + if tooLong { + // Never left frozen (issue 1966): a reader that stops while the bundle lives leaves the bundle's + // next write blocked for good. Said, and the bundle ended; the exit path below fails what waits + // on it and starts it again. + reason := fmt.Sprintf("%s's bundle wrote a line over %d bytes (a tool's answer the bus cannot carry)", module, lineCap) + logf("[mesh-tools] %s; ending the bundle", reason) + c.mu.Lock() + c.endedFor = reason + c.mu.Unlock() + _ = cmd.Process.Kill() + break + } + if rerr != nil && len(raw) == 0 { + break + } + line := strings.TrimSpace(string(raw)) if line == "" { continue } @@ -529,6 +567,9 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an } saidMu.Unlock() c.mu.Lock() + if c.endedFor != "" { + why = c.endedFor + "; the bundle was ended (" + code + ")" + } c.why = errors.New(why) watches := c.watches c.watches = nil @@ -697,6 +738,10 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an defer end() res, err := asking("tools/call", callParams(full, caller, call, args), CallTimeout) if err != nil { + if errors.Is(err, errNoAnswer) { + // Its caller is gone by now and nobody else would hear of it (issue 1966). + logf("[mesh-tools] %s.%s called by %q: %v", module, full, caller, err) + } return nil, err } var called struct { @@ -794,7 +839,7 @@ func (c *child) ask(module, method string, params any, timeout time.Duration) (j c.mu.Lock() delete(c.pending, id) c.mu.Unlock() - return nil, fmt.Errorf("%s's bundle did not answer %s in %ds", module, method, int(timeout/time.Second)) + return nil, fmt.Errorf("%s's bundle did not answer %s in %ds: %w", module, method, int(timeout/time.Second), errNoAnswer) } } diff --git a/node-tools/internal/launch/state_failed_test.go b/node-tools/internal/launch/state_failed_test.go index 1b9b7ff..8742e0b 100644 --- a/node-tools/internal/launch/state_failed_test.go +++ b/node-tools/internal/launch/state_failed_test.go @@ -73,3 +73,80 @@ func TestAFailedStateOperationIsSaidInTheRuntimesLog(t *testing.T) { } } } + +// A bundle that lists one tool, "t", and answers a call to it as its MODE says: "long" with a line over the runner's cap, +// "silent" with nothing at all. +const callingBundle = `#!/bin/sh +while IFS= read -r line; do + id=$(printf '%s' "$line" | sed -n 's/.*"id":\([0-9]*\)[,}].*/\1/p') + case "$line" in + *'"method":"initialize"'*) echo started >&2; printf '{"jsonrpc":"2.0","id":%s,"result":{}}\n' "$id" ;; + *'"method":"tools/list"'*) printf '{"jsonrpc":"2.0","id":%s,"result":{"tools":[{"name":"t","description":"d"}]}}\n' "$id" ;; + *'"method":"tools/call"'*) [ "$MODE" = long ] && printf '%4096s\n' x ;; + esac +done +` + +// Issue 1966: a bundle's line over the runner's cap, and a call the bundle never answers, are said in the runtime's log; +// the over-long line ends the bundle and fails the call at once, never leaving the runner reading nothing while the +// bundle's next write blocks. +func TestAnOverlongLineAndAnUnansweredCallAreSaid(t *testing.T) { + oldMax, oldCall, oldFirst := maxLine, CallTimeout, RestartFirst + maxLine, RestartFirst = 1024, 10*time.Millisecond + t.Cleanup(func() { maxLine, CallTimeout, RestartFirst = oldMax, oldCall, oldFirst }) + for _, row := range []struct { + mode string + timeout time.Duration + }{ + // The call's own bound is long: only the runner ending the bundle fails it within a second. + {"long", 10 * time.Second}, + {"silent", 300 * time.Millisecond}, + } { + t.Run(row.mode, func(t *testing.T) { + CallTimeout = row.timeout + dir := t.TempDir() + entry := filepath.Join(dir, "bundle") + if err := os.WriteFile(entry, []byte(callingBundle), 0o755); err != nil { + t.Fatal(err) + } + said := make(chan string, 32) + logf := func(f string, a ...any) { said <- fmt.Sprintf(f, a...) } + l, err := Start("caller", entry, append(os.Environ(), "MODE="+row.mode), &refusingState{}, logf) + if err != nil { + t.Fatal(err) + } + t.Cleanup(l.Stop) + begun := time.Now() + _, err = l.Registrations[0].Tools[0].Run("caller-1", nil) + if err == nil { + t.Fatal("the call was answered") + } + var heard []string + timeout := time.After(5 * time.Second) + want := "caller.t called by \"caller-1\"" + if row.mode == "long" { + want = "ending the bundle" + if took := time.Since(begun); took > time.Second { + t.Fatalf("the call hung for %s", took) + } + if !strings.Contains(err.Error(), "over 1024 bytes") || strings.Contains(err.Error(), "did not answer") { + t.Fatalf("the call failed with %q, not the cap's reason", err) + } + } + // For "long" the bundle is also started again: it says "started" on each start, so twice in all. + started, found := 0, false + for !found || (row.mode == "long" && started < 2) { + select { + case line := <-said: + heard = append(heard, line) + found = found || strings.Contains(line, want) + if line == "[caller] started" { + started++ + } + case <-timeout: + t.Fatalf("not said, or the bundle not started again; the log: %v", heard) + } + } + }) + } +}