Merge pull request 'The tool runner ends and restarts a bundle that writes an over-long line, and says an unanswered call and a refused answer (issue 1966 S8)' (#67) from fix/1966-s8-runner-frozen into main

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
+}

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")
+	}
+}

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)
+			rd := bufio.NewReaderSize(stdout, 64*1024)
-			scan.Buffer(make([]byte, 64*1024), 16<<20)
+			for {
-			for scan.Scan() {
+				raw, tooLong, rerr := readCapped(rd, lineCap)
-				line := strings.TrimSpace(scan.Text())
+				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)
 	}
 }
 

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)
+				}
+			}
+		})
+	}
+}