From 312e29478e423255a6c84db32fcf4e2d1f3bde9c Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 13:08:53 +0100 Subject: [PATCH] Tool runner: the cap's reason fails the call; the test proves the ending; the refused-answer limit keys on the caller's prefix and forgets old entries (issue 1966, review round 1) Claude-Session: https://claude.ai/code/session_01NcbgfFfQ4Z9UE5jK32PmE5 --- node-tools/internal/bus/bus.go | 16 ++++- node-tools/internal/bus/refused_test.go | 11 +++- node-tools/internal/launch/launch.go | 12 +++- .../internal/launch/state_failed_test.go | 61 +++++++++++++------ 4 files changed, 73 insertions(+), 27 deletions(-) diff --git a/node-tools/internal/bus/bus.go b/node-tools/internal/bus/bus.go index a46dec9..920189e 100644 --- a/node-tools/internal/bus/bus.go +++ b/node-tools/internal/bus/bus.go @@ -1229,15 +1229,25 @@ func (c *Conn) answerRefused(err error) bool { 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{} } - last, seen := c.inboxSaid[subject] - say := !seen || time.Since(last) >= time.Minute + 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] = time.Now() + c.inboxSaid[subject] = now } c.mu.Unlock() if say && c.Logf != nil { diff --git a/node-tools/internal/bus/refused_test.go b/node-tools/internal/bus/refused_test.go index dbe7a81..c9c3481 100644 --- a/node-tools/internal/bus/refused_test.go +++ b/node-tools/internal/bus/refused_test.go @@ -94,17 +94,22 @@ func TestARefusedAnswerIsSaidOncePerMinute(t *testing.T) { refuse := func(subject string) bool { return c.answerRefused(fmt.Errorf("%w: Permissions Violation for Publish to %q", nats.ErrPermissionViolation, subject)) } - if !refuse("_INBOX.user.abc") || !refuse("_INBOX.user.abc") || !refuse("_INBOX.user.def") { + // 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 subject, got %v", said) + t.Fatalf("want one line per caller, got %v", said) } c.inboxSaid["_INBOX.user.abc"] = time.Now().Add(-2 * time.Minute) - refuse("_INBOX.user.abc") + 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 560f180..dc64fdc 100644 --- a/node-tools/internal/launch/launch.go +++ b/node-tools/internal/launch/launch.go @@ -206,6 +206,8 @@ 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() @@ -351,8 +353,11 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an // 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. - logf("[mesh-tools] %s's bundle wrote a line over %d bytes (a tool's answer the bus cannot carry); ending the bundle", - module, maxLine) + reason := fmt.Sprintf("%s's bundle wrote a line over %d bytes (a tool's answer the bus cannot carry)", module, maxLine) + logf("[mesh-tools] %s; ending the bundle", reason) + c.mu.Lock() + c.endedFor = reason + c.mu.Unlock() _ = cmd.Process.Kill() break } @@ -561,6 +566,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 diff --git a/node-tools/internal/launch/state_failed_test.go b/node-tools/internal/launch/state_failed_test.go index 0463757..40595c2 100644 --- a/node-tools/internal/launch/state_failed_test.go +++ b/node-tools/internal/launch/state_failed_test.go @@ -80,7 +80,7 @@ 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"'*) printf '{"jsonrpc":"2.0","id":%s,"result":{}}\n' "$id" ;; + *'"method":"initialize"'*) echo started >> "$STARTS"; 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 @@ -91,21 +91,28 @@ done // 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 := maxLine, CallTimeout - maxLine, CallTimeout = 1024, 300*time.Millisecond - t.Cleanup(func() { maxLine, CallTimeout = oldMax, oldCall }) - for _, row := range []struct{ mode, said string }{ - {"long", "bundle wrote a line over 1024 bytes"}, - {"silent", ""}, + 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) { - entry := filepath.Join(t.TempDir(), "bundle") + CallTimeout = row.timeout + dir := t.TempDir() + entry := filepath.Join(dir, "bundle") + starts := filepath.Join(dir, "starts") 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) + l, err := Start("caller", entry, append(os.Environ(), "MODE="+row.mode, "STARTS="+starts), &refusingState{}, logf) if err != nil { t.Fatal(err) } @@ -115,25 +122,41 @@ func TestAnOverlongLineAndAnUnansweredCallAreSaid(t *testing.T) { if err == nil { t.Fatal("the call was answered") } - if took := time.Since(begun); took > 5*time.Second { - t.Fatalf("the call hung for %s", took) - } var heard []string timeout := time.After(5 * time.Second) - for { + 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 found := false; !found; { select { case line := <-said: heard = append(heard, line) - if row.mode == "long" && strings.Contains(line, row.said) && strings.Contains(line, "ending the bundle") { - return - } - if row.mode == "silent" && strings.Contains(line, "caller.t called by \"caller-1\"") && strings.Contains(line, "did not answer tools/call") { - return - } + found = strings.Contains(line, want) case <-timeout: t.Fatalf("not said; the log: %v", heard) } } + if row.mode != "long" { + return + } + // The bundle is started again: its start is written once more. + for { + if b, _ := os.ReadFile(starts); strings.Count(string(b), "started") >= 2 { + return + } + select { + case <-timeout: + t.Fatal("the bundle was not started again") + case <-time.After(10 * time.Millisecond): + } + } }) } }