From 089bd947b0030308514d267d3562670774042198 Mon Sep 17 00:00:00 2001 From: woodsonl <65194841+woodsonl@users.noreply.github.com> Date: Tue, 22 Sep 2026 10:18:35 -0500 Subject: [PATCH] fix: distinguish decode errors from terminal read errors in JSON-RPC loops MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every service's read loop treated all non-EOF read errors the same way: log and `continue`. That is right for a bad frame (the scanner advances past it) but wrong for a terminal transport error such as an over-long frame that bufio.Scanner cannot resync past — the loop then spins on the same error forever instead of ending the pump. Classify by error type instead. A DecodeError is recoverable and keeps the loop alive; any other non-EOF error is terminal. The broker's read loop returns errTerminalRead so Serve tears down cleanly, and the producer goroutine stops feeding the channel once it hits a terminal error rather than blocking forever. Signed-off-by: woodsonl <65194841+woodsonl@users.noreply.github.com> --- services/nvpair-cluster-manager/codec.go | 7 +- services/nvpair-cluster-manager/manager.go | 12 ++- services/nvpair-errors/codec.go | 7 +- services/nvpair-errors/manager.go | 12 ++- services/nvpair-job-scheduler/codec.go | 7 +- services/nvpair-job-scheduler/manager.go | 12 ++- services/nvpair-manual-nodes/codec.go | 7 +- services/nvpair-manual-nodes/manager.go | 12 ++- services/nvpair-node-scanner/codec.go | 7 +- services/nvpair-node-scanner/scanner.go | 12 ++- services/nvpair-node-settings/codec.go | 7 +- services/nvpair-node-settings/manager.go | 11 +- services/nvpair-proxy/codec.go | 7 +- services/nvpair-proxy/proxy.go | 11 +- services/nvpair-tui/rpc/client.go | 11 +- services/nvpair-tui/rpc/codec.go | 16 ++- services/nvpair-ui-broker/broker.go | 37 +++++-- services/nvpair-ui-broker/codec.go | 9 +- .../nvpair-ui-broker/terminal_read_test.go | 100 ++++++++++++++++++ .../workloadstore/persistence.go | 8 +- services/nvpair-workload-manager/codec.go | 7 +- services/nvpair-workload-manager/manager.go | 12 ++- 22 files changed, 275 insertions(+), 56 deletions(-) create mode 100644 services/nvpair-ui-broker/terminal_read_test.go diff --git a/services/nvpair-cluster-manager/codec.go b/services/nvpair-cluster-manager/codec.go index 5d6fd177..1691b8cd 100644 --- a/services/nvpair-cluster-manager/codec.go +++ b/services/nvpair-cluster-manager/codec.go @@ -10,9 +10,10 @@ package main import "nvpair-shared/jsonrpc" type ( - Message = jsonrpc.Message - RPCError = jsonrpc.RPCError - Codec = jsonrpc.Codec + Message = jsonrpc.Message + RPCError = jsonrpc.RPCError + DecodeError = jsonrpc.DecodeError + Codec = jsonrpc.Codec ) var NewCodec = jsonrpc.NewCodec diff --git a/services/nvpair-cluster-manager/manager.go b/services/nvpair-cluster-manager/manager.go index ac45bc8b..9bb84217 100644 --- a/services/nvpair-cluster-manager/manager.go +++ b/services/nvpair-cluster-manager/manager.go @@ -6,6 +6,7 @@ package main import ( "context" "encoding/json" + stderrors "errors" "fmt" "io" "log" @@ -372,8 +373,15 @@ func (m *Manager) readLoop(ctx context.Context) error { if err == io.EOF || ctx.Err() != nil { return nil } - log.Printf("JSON-RPC read error: %v", err) - continue + var de *DecodeError + if stderrors.As(err, &de) { + log.Printf("JSON-RPC decode error (skipping frame): %v", err) + continue + } + // Terminal transport/scanner error (e.g. an over-long frame — + // bufio.Scanner cannot resync) — stop instead of spinning. + log.Printf("JSON-RPC read error (terminal): %v", err) + return err } m.handleMessage(msg) if ctx.Err() != nil { diff --git a/services/nvpair-errors/codec.go b/services/nvpair-errors/codec.go index 5d6fd177..1691b8cd 100644 --- a/services/nvpair-errors/codec.go +++ b/services/nvpair-errors/codec.go @@ -10,9 +10,10 @@ package main import "nvpair-shared/jsonrpc" type ( - Message = jsonrpc.Message - RPCError = jsonrpc.RPCError - Codec = jsonrpc.Codec + Message = jsonrpc.Message + RPCError = jsonrpc.RPCError + DecodeError = jsonrpc.DecodeError + Codec = jsonrpc.Codec ) var NewCodec = jsonrpc.NewCodec diff --git a/services/nvpair-errors/manager.go b/services/nvpair-errors/manager.go index 8d4f37e0..6daa1555 100644 --- a/services/nvpair-errors/manager.go +++ b/services/nvpair-errors/manager.go @@ -6,6 +6,7 @@ package main import ( "context" "encoding/json" + stderrors "errors" "fmt" "io" "log" @@ -176,8 +177,15 @@ func (m *Manager) readLoop(ctx context.Context) error { if err == io.EOF || ctx.Err() != nil { return nil } - log.Printf("JSON-RPC read error: %v", err) - continue + var de *DecodeError + if stderrors.As(err, &de) { + log.Printf("JSON-RPC decode error (skipping frame): %v", err) + continue + } + // Terminal transport/scanner error (e.g. an over-long frame — + // bufio.Scanner cannot resync) — stop instead of spinning. + log.Printf("JSON-RPC read error (terminal): %v", err) + return err } m.handleMessage(msg) if ctx.Err() != nil { diff --git a/services/nvpair-job-scheduler/codec.go b/services/nvpair-job-scheduler/codec.go index 16601231..aaf12fef 100644 --- a/services/nvpair-job-scheduler/codec.go +++ b/services/nvpair-job-scheduler/codec.go @@ -10,9 +10,10 @@ package main import "nvpair-shared/jsonrpc" type ( - Message = jsonrpc.Message - RPCError = jsonrpc.RPCError - Codec = jsonrpc.Codec + Message = jsonrpc.Message + RPCError = jsonrpc.RPCError + DecodeError = jsonrpc.DecodeError + Codec = jsonrpc.Codec ) var NewCodec = jsonrpc.NewCodec diff --git a/services/nvpair-job-scheduler/manager.go b/services/nvpair-job-scheduler/manager.go index bdd3f366..58e19486 100644 --- a/services/nvpair-job-scheduler/manager.go +++ b/services/nvpair-job-scheduler/manager.go @@ -6,6 +6,7 @@ package main import ( "context" "encoding/json" + "errors" "fmt" "io" "log" @@ -98,8 +99,15 @@ func (m *Manager) readLoop(ctx context.Context) error { if err == io.EOF || ctx.Err() != nil { return nil } - log.Printf("JSON-RPC read error: %v", err) - continue + var de *DecodeError + if errors.As(err, &de) { + log.Printf("JSON-RPC decode error (skipping frame): %v", err) + continue + } + // Terminal transport/scanner error (e.g. an over-long frame — + // bufio.Scanner cannot resync) — stop instead of spinning. + log.Printf("JSON-RPC read error (terminal): %v", err) + return err } m.handleMessage(msg) if ctx.Err() != nil { diff --git a/services/nvpair-manual-nodes/codec.go b/services/nvpair-manual-nodes/codec.go index 5d6fd177..1691b8cd 100644 --- a/services/nvpair-manual-nodes/codec.go +++ b/services/nvpair-manual-nodes/codec.go @@ -10,9 +10,10 @@ package main import "nvpair-shared/jsonrpc" type ( - Message = jsonrpc.Message - RPCError = jsonrpc.RPCError - Codec = jsonrpc.Codec + Message = jsonrpc.Message + RPCError = jsonrpc.RPCError + DecodeError = jsonrpc.DecodeError + Codec = jsonrpc.Codec ) var NewCodec = jsonrpc.NewCodec diff --git a/services/nvpair-manual-nodes/manager.go b/services/nvpair-manual-nodes/manager.go index 55a4040a..ed3cb50d 100644 --- a/services/nvpair-manual-nodes/manager.go +++ b/services/nvpair-manual-nodes/manager.go @@ -6,6 +6,7 @@ package main import ( "context" "encoding/json" + stderrors "errors" "fmt" "io" "log" @@ -606,8 +607,15 @@ func (m *Manager) readLoop(ctx context.Context) error { if err == io.EOF || ctx.Err() != nil { return nil } - log.Printf("JSON-RPC read error: %v", err) - continue + var de *DecodeError + if stderrors.As(err, &de) { + log.Printf("JSON-RPC decode error (skipping frame): %v", err) + continue + } + // Terminal transport/scanner error (e.g. an over-long frame — + // bufio.Scanner cannot resync) — stop instead of spinning. + log.Printf("JSON-RPC read error (terminal): %v", err) + return err } m.handleMessage(msg) if ctx.Err() != nil { diff --git a/services/nvpair-node-scanner/codec.go b/services/nvpair-node-scanner/codec.go index 5d6fd177..1691b8cd 100644 --- a/services/nvpair-node-scanner/codec.go +++ b/services/nvpair-node-scanner/codec.go @@ -10,9 +10,10 @@ package main import "nvpair-shared/jsonrpc" type ( - Message = jsonrpc.Message - RPCError = jsonrpc.RPCError - Codec = jsonrpc.Codec + Message = jsonrpc.Message + RPCError = jsonrpc.RPCError + DecodeError = jsonrpc.DecodeError + Codec = jsonrpc.Codec ) var NewCodec = jsonrpc.NewCodec diff --git a/services/nvpair-node-scanner/scanner.go b/services/nvpair-node-scanner/scanner.go index 485a0275..d86da5e5 100644 --- a/services/nvpair-node-scanner/scanner.go +++ b/services/nvpair-node-scanner/scanner.go @@ -5,6 +5,7 @@ package main import ( "context" + "errors" "fmt" "io" "log" @@ -101,8 +102,15 @@ func (s *Scanner) readLoop(ctx context.Context) error { if err == io.EOF || ctx.Err() != nil { return nil } - log.Printf("JSON-RPC read error: %v", err) - continue + var de *DecodeError + if errors.As(err, &de) { + log.Printf("JSON-RPC decode error (skipping frame): %v", err) + continue + } + // Terminal transport/scanner error (e.g. an over-long frame — + // bufio.Scanner cannot resync) — stop instead of spinning. + log.Printf("JSON-RPC read error (terminal): %v", err) + return err } s.handleMessage(msg) } diff --git a/services/nvpair-node-settings/codec.go b/services/nvpair-node-settings/codec.go index 5d6fd177..1691b8cd 100644 --- a/services/nvpair-node-settings/codec.go +++ b/services/nvpair-node-settings/codec.go @@ -10,9 +10,10 @@ package main import "nvpair-shared/jsonrpc" type ( - Message = jsonrpc.Message - RPCError = jsonrpc.RPCError - Codec = jsonrpc.Codec + Message = jsonrpc.Message + RPCError = jsonrpc.RPCError + DecodeError = jsonrpc.DecodeError + Codec = jsonrpc.Codec ) var NewCodec = jsonrpc.NewCodec diff --git a/services/nvpair-node-settings/manager.go b/services/nvpair-node-settings/manager.go index a30a2eae..dfe7ed24 100644 --- a/services/nvpair-node-settings/manager.go +++ b/services/nvpair-node-settings/manager.go @@ -283,8 +283,15 @@ func (m *Manager) readLoop(ctx context.Context) error { if err == io.EOF || ctx.Err() != nil { return nil } - log.Printf("JSON-RPC read error: %v", err) - continue + var de *DecodeError + if errors.As(err, &de) { + log.Printf("JSON-RPC decode error (skipping frame): %v", err) + continue + } + // Terminal transport/scanner error (e.g. an over-long frame — + // bufio.Scanner cannot resync) — stop instead of spinning. + log.Printf("JSON-RPC read error (terminal): %v", err) + return err } m.handleMessage(msg) if ctx.Err() != nil { diff --git a/services/nvpair-proxy/codec.go b/services/nvpair-proxy/codec.go index 5d6fd177..1691b8cd 100644 --- a/services/nvpair-proxy/codec.go +++ b/services/nvpair-proxy/codec.go @@ -10,9 +10,10 @@ package main import "nvpair-shared/jsonrpc" type ( - Message = jsonrpc.Message - RPCError = jsonrpc.RPCError - Codec = jsonrpc.Codec + Message = jsonrpc.Message + RPCError = jsonrpc.RPCError + DecodeError = jsonrpc.DecodeError + Codec = jsonrpc.Codec ) var NewCodec = jsonrpc.NewCodec diff --git a/services/nvpair-proxy/proxy.go b/services/nvpair-proxy/proxy.go index 929c3d67..452bceff 100644 --- a/services/nvpair-proxy/proxy.go +++ b/services/nvpair-proxy/proxy.go @@ -2553,8 +2553,15 @@ func (p *Proxy) readLoop(ctx context.Context) error { if err == io.EOF || ctx.Err() != nil { return nil } - log.Printf("JSON-RPC read error: %v", err) - continue + var de *DecodeError + if stderrors.As(err, &de) { + log.Printf("JSON-RPC decode error (skipping frame): %v", err) + continue + } + // Terminal transport/scanner error (e.g. an over-long frame — + // bufio.Scanner cannot resync) — stop instead of spinning. + log.Printf("JSON-RPC read error (terminal): %v", err) + return err } p.handleMessage(msg) } diff --git a/services/nvpair-tui/rpc/client.go b/services/nvpair-tui/rpc/client.go index d59ce8ad..d11dec7a 100644 --- a/services/nvpair-tui/rpc/client.go +++ b/services/nvpair-tui/rpc/client.go @@ -6,6 +6,7 @@ package rpc import ( "context" "encoding/json" + "errors" "fmt" "io" "strconv" @@ -68,9 +69,15 @@ func (c *Client) Run(ctx context.Context) error { if err == io.EOF { return nil } - // A single malformed line should not kill the session; the + // A single malformed frame should not kill the session; the // broker may emit a frame we don't model. Skip and continue. - continue + // Anything else (over-long frame, transport error) is terminal: + // bufio.Scanner cannot resync, so continuing would spin. + var de *DecodeError + if errors.As(err, &de) { + continue + } + return err } switch { case msg.IsResponse(): diff --git a/services/nvpair-tui/rpc/codec.go b/services/nvpair-tui/rpc/codec.go index 8d5e8c83..2c36977e 100644 --- a/services/nvpair-tui/rpc/codec.go +++ b/services/nvpair-tui/rpc/codec.go @@ -71,6 +71,14 @@ type Codec struct { wmu sync.Mutex } +// DecodeError marks a recoverable per-frame failure (bad JSON or wrong +// version): the stream is still positioned at the next line, so callers +// may skip the frame and continue reading. Mirrors nvpair-shared/jsonrpc. +type DecodeError struct{ Err error } + +func (e *DecodeError) Error() string { return e.Err.Error() } +func (e *DecodeError) Unwrap() error { return e.Err } + // NewCodec wraps a reader/writer pair (the broker's stdout/stdin) in a // framing codec. func NewCodec(r io.Reader, w io.Writer) *Codec { @@ -79,7 +87,9 @@ func NewCodec(r io.Reader, w io.Writer) *Codec { return &Codec{scanner: scanner, writer: w} } -// Read returns the next frame, or io.EOF when the stream closes. +// Read returns the next frame, or io.EOF when the stream closes. A +// malformed frame (bad JSON or wrong version) is returned as a recoverable +// *DecodeError; a terminal scanner/transport error is a plain error. func (c *Codec) Read() (*Message, error) { if !c.scanner.Scan() { if err := c.scanner.Err(); err != nil { @@ -89,10 +99,10 @@ func (c *Codec) Read() (*Message, error) { } var msg Message if err := json.Unmarshal(c.scanner.Bytes(), &msg); err != nil { - return nil, fmt.Errorf("invalid JSON-RPC message: %w", err) + return nil, &DecodeError{fmt.Errorf("invalid JSON-RPC message: %w", err)} } if msg.JSONRPC != "2.0" { - return nil, fmt.Errorf("unsupported JSON-RPC version: %q", msg.JSONRPC) + return nil, &DecodeError{fmt.Errorf("unsupported JSON-RPC version: %q", msg.JSONRPC)} } return &msg, nil } diff --git a/services/nvpair-ui-broker/broker.go b/services/nvpair-ui-broker/broker.go index c5da7fd1..dc04cbbe 100644 --- a/services/nvpair-ui-broker/broker.go +++ b/services/nvpair-ui-broker/broker.go @@ -2585,6 +2585,20 @@ func setNodeIDIfEmpty(m map[string]json.RawMessage, key, nodeID string) bool { return true } +// errTerminalRead reports that the client stdin read loop ended on a +// non-recoverable scanner/transport error (e.g. an over-long frame that +// bufio.Scanner cannot resync past), distinct from a clean EOF. +var errTerminalRead = stderrors.New("terminal read error") + +// recoverableDecode reports whether a codec Read error is a recoverable +// per-frame decode failure (bad JSON / wrong version): the scanner advances +// past the bad frame, so both the producer and the consumer keep pumping +// instead of tearing the connection down. +func recoverableDecode(err error) bool { + var de *DecodeError + return stderrors.As(err, &de) +} + func (b *Broker) readLoop(ctx context.Context) error { // codec.Read() blocks on stdin, so we run it on its own goroutine and // select against ctx.Done(). Otherwise a SIGINT/SIGTERM (which cancels @@ -2604,11 +2618,18 @@ func (b *Broker) readLoop(ctx context.Context) error { case <-ctx.Done(): return } - // EOF is terminal (stream closed); other errors are per-line - // (e.g. a bad JSON frame) and the next Read advances past them. - if err == io.EOF { - return + // A decoded message (err nil) and a recoverable decode error + // (bad frame; the next Read advances past it) both keep the pump + // running. EOF is terminal (stream closed), and any other error + // is a terminal scanner/transport error: stop feeding the + // channel so the consumer exits instead of spinning. + if err == nil { + continue } + if recoverableDecode(err) { + continue + } + return } }() @@ -2621,8 +2642,12 @@ func (b *Broker) readLoop(ctx context.Context) error { if r.err == io.EOF || ctx.Err() != nil { return nil } - slog.Warn("JSON-RPC read error", "err", r.err) - continue + if recoverableDecode(r.err) { + slog.Warn("JSON-RPC decode error (skipping frame)", "err", r.err) + continue + } + slog.Warn("JSON-RPC read error (terminal)", "err", r.err) + return errTerminalRead } b.handleMessage(r.msg) if ctx.Err() != nil { diff --git a/services/nvpair-ui-broker/codec.go b/services/nvpair-ui-broker/codec.go index 4d7dcc80..7815d286 100644 --- a/services/nvpair-ui-broker/codec.go +++ b/services/nvpair-ui-broker/codec.go @@ -16,10 +16,11 @@ import ( ) type ( - Message = jsonrpc.Message - RPCError = jsonrpc.RPCError - Codec = jsonrpc.Codec - Peer = jsonrpc.Peer + Message = jsonrpc.Message + RPCError = jsonrpc.RPCError + DecodeError = jsonrpc.DecodeError + Codec = jsonrpc.Codec + Peer = jsonrpc.Peer ) var ( diff --git a/services/nvpair-ui-broker/terminal_read_test.go b/services/nvpair-ui-broker/terminal_read_test.go new file mode 100644 index 00000000..5435cd13 --- /dev/null +++ b/services/nvpair-ui-broker/terminal_read_test.go @@ -0,0 +1,100 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "context" + "errors" + "io" + "net" + "testing" + "time" +) + +// errFakeTerminal is a non-EOF, non-DecodeError read failure — the shape a +// terminal scanner/transport error takes at the codec boundary. +var errFakeTerminal = errors.New("fake terminal read error") + +// errorReader fails every Read with errFakeTerminal; wrapped in a real Codec it +// stands in for a dead terminal transport. +type errorReader struct{} + +func (errorReader) Read([]byte) (int, error) { return 0, errFakeTerminal } + +func newTerminalErrorCodec() *Codec { + return NewCodec(struct { + io.Reader + io.Writer + }{errorReader{}, io.Discard}) +} + +// TestReadLoopTerminalReadErrorStopsPump guards the read-loop contract: a +// non-EOF transport error must end Serve-style pumping with errTerminalRead +// (the producer goroutine has already stopped; spinning would burn CPU forever), +// while EOF and plain decode errors keep the loop alive. +func TestReadLoopTerminalReadErrorStopsPump(t *testing.T) { + t.Run("non-EOF transport error is terminal", func(t *testing.T) { + b := &Broker{codec: newTerminalErrorCodec()} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + done := make(chan error, 1) + go func() { done <- b.readLoop(ctx) }() + + select { + case err := <-done: + if !errors.Is(err, errTerminalRead) { + t.Fatalf("readLoop err = %v, want errTerminalRead", err) + } + case <-time.After(5 * time.Second): + t.Fatal("readLoop did not stop on transport error") + } + }) + + t.Run("EOF is clean exit", func(t *testing.T) { + client, server := net.Pipe() + defer client.Close() + defer server.Close() + b := &Broker{codec: NewCodec(client)} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + done := make(chan error, 1) + go func() { done <- b.readLoop(ctx) }() + server.Close() // producer sees EOF + + select { + case err := <-done: + if err != nil { + t.Fatalf("readLoop err on EOF = %v, want nil", err) + } + case <-time.After(5 * time.Second): + t.Fatal("readLoop did not stop on EOF") + } + }) + + t.Run("recoverable decode error keeps the loop alive", func(t *testing.T) { + client, server := net.Pipe() + defer client.Close() + defer server.Close() + b := &Broker{codec: NewCodec(client)} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + done := make(chan error, 1) + go func() { done <- b.readLoop(ctx) }() + if _, err := io.WriteString(server, "not-json\n"); err != nil { + t.Fatal(err) + } + // Keep the pipe open: a closed pipe is a terminal read error, while + // the bad frame must only surface as a recoverable DecodeError. + time.Sleep(100 * time.Millisecond) + select { + case err := <-done: + t.Fatalf("readLoop exited early on decode error: %v", err) + default: + } + cancel() + }) +} diff --git a/services/nvpair-ui-broker/workloadstore/persistence.go b/services/nvpair-ui-broker/workloadstore/persistence.go index 24d217ca..fb73d10a 100644 --- a/services/nvpair-ui-broker/workloadstore/persistence.go +++ b/services/nvpair-ui-broker/workloadstore/persistence.go @@ -110,7 +110,13 @@ func (s *Store) Checkpoint() error { s.mu.Unlock() rotate(path, rotations) - return writeSnapshotFile(path, infos) + if err := writeSnapshotFile(path, infos); err != nil { + s.mu.Lock() + s.dirty = true + s.mu.Unlock() + return err + } + return nil } // Run drives the coalescing flusher until ctx is cancelled: a dirty flush every diff --git a/services/nvpair-workload-manager/codec.go b/services/nvpair-workload-manager/codec.go index 5d6fd177..1691b8cd 100644 --- a/services/nvpair-workload-manager/codec.go +++ b/services/nvpair-workload-manager/codec.go @@ -10,9 +10,10 @@ package main import "nvpair-shared/jsonrpc" type ( - Message = jsonrpc.Message - RPCError = jsonrpc.RPCError - Codec = jsonrpc.Codec + Message = jsonrpc.Message + RPCError = jsonrpc.RPCError + DecodeError = jsonrpc.DecodeError + Codec = jsonrpc.Codec ) var NewCodec = jsonrpc.NewCodec diff --git a/services/nvpair-workload-manager/manager.go b/services/nvpair-workload-manager/manager.go index 84ea634d..993db486 100644 --- a/services/nvpair-workload-manager/manager.go +++ b/services/nvpair-workload-manager/manager.go @@ -6,6 +6,7 @@ package main import ( "context" "encoding/json" + "errors" "fmt" "io" "log" @@ -247,8 +248,15 @@ func (m *Manager) readLoop(ctx context.Context) error { m.cancel() return nil } - log.Printf("JSON-RPC read error: %v", err) - continue + var de *DecodeError + if errors.As(err, &de) { + log.Printf("JSON-RPC decode error (skipping frame): %v", err) + continue + } + // Terminal transport/scanner error (e.g. an over-long frame — + // bufio.Scanner cannot resync) — stop instead of spinning. + log.Printf("JSON-RPC read error (terminal): %v", err) + return err } m.handleMessage(msg) if ctx.Err() != nil {