fix(broker): dispatch requests on a bounded pool and coalesce relay deliveries

services/nvpair-ui-broker/broker.go

@@ -258,15 +258,15 @@ type Broker struct {
 	// subMu guards subscribed. The discovery:nodes-changed stream is
 	// opt-in: emitNodesChanged (called on the scanner-event goroutine)
 	// reads this flag while the discovery:subscribe / discovery:unsubscribe
-	// handlers (called on the read-loop goroutine) flip it, so the two
+	// handlers (called on the dispatch-pool goroutine) flip it, so the two
 	// goroutines need a lock between them.
 	subMu      sync.Mutex
 	subscribed bool
 
 	// proxyMu guards every engineProxyRuntime.subscribed. The
 	// <namespace>:<event> streams are opt-in like discovery's: the forward
 	// hooks (on each proxy's reader goroutine) read the flag while the
-	// subscribe / unsubscribe handlers (on the read-loop goroutine) flip it.
+	// subscribe / unsubscribe handlers (on the dispatch-pool goroutine) flip it.
 	proxyMu sync.Mutex
 
 	// engineProxies holds per-engine proxy state, one entry per engine in the
@@ -278,7 +278,7 @@ type Broker struct {
 	// opt-in too: emitWorkloadEvent (called on the proxy reader goroutine
 	// for local echoes and on the workload-manager reader goroutine for
 	// peer-origin relays) reads the flag while the workloads:subscribe /
-	// workloads:unsubscribe handlers (on the read-loop goroutine) flip it.
+	// workloads:unsubscribe handlers (on the dispatch-pool goroutine) flip it.
 	workloadsMu         sync.Mutex
 	workloadsSubscribed bool
 
@@ -300,7 +300,7 @@ type Broker struct {
 	// engineMu guards engineSubscribed. The engine:<event> stream is
 	// opt-in like proxy's: forwardEngineNotification (engine-manager reader
 	// goroutine) reads the flag while the engine:subscribe /
-	// engine:unsubscribe handlers (read-loop goroutine) flip it.
+	// engine:unsubscribe handlers (dispatch-pool goroutine) flip it.
 	engineMu         sync.Mutex
 	engineSubscribed bool
 
@@ -2039,6 +2039,18 @@ func (b *Broker) runWorkloadHistoryFlusher(ctx context.Context) func() {
 	}
 }
 
+// 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")
+
+// messageDispatchConcurrency is the size of the broker's inbound dispatch
+// pool: enough worker goroutines that one slow synchronous worker relay
+// (bounded by rpcWorkerCallTimeout) cannot head-of-line block the rest of
+// the control plane, few enough that handlers stay effectively serialized
+// under normal traffic.
+const messageDispatchConcurrency = 4
+
 func (b *Broker) Serve(ctx context.Context) error {
 	ctx, cancel := context.WithCancel(ctx)
 	b.cancel = cancel
@@ -2585,6 +2597,15 @@ func setNodeIDIfEmpty(m map[string]json.RawMessage, key, nodeID string) bool {
 	return true
 }
 
+// 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,14 +2625,47 @@ 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
 		}
 	}()
 
+	// Bounded dispatch pool: handleMessage runs synchronous worker relays
+	// (proxy/cluster/settings/manual-nodes, each bounded by
+	// rpcWorkerCallTimeout) so dispatching on the read loop would let one
+	// slow worker stall every other client request for up to 5s. A small
+	// worker pool decouples them. Cross-request ordering is preserved for
+	// the channels that need it by dedicated mutexes inside the handlers
+	// (workloadEmitMu serializes workload apply→fan→emit; subscription
+	// bookkeeping is per-state mutexed), and JSON-RPC has no cross-request
+	// response-ordering guarantee — each response carries its own id. The
+	// codec's write mutex keeps concurrent responses from interleaving.
+	dispatch := make(chan *Message)
+	var dispatchWG sync.WaitGroup
+	for range messageDispatchConcurrency {
+		dispatchWG.Add(1)
+		go func() {
+			defer dispatchWG.Done()
+			for msg := range dispatch {
+				b.handleMessage(msg)
+			}
+		}()
+	}
+	defer func() {
+		close(dispatch)
+		dispatchWG.Wait()
+	}()
+
 	for {
 		select {
 		case <-ctx.Done():
@@ -2621,10 +2675,18 @@ 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
+			}
+			select {
+			case dispatch <- r.msg:
+			case <-ctx.Done():
+				return nil
 			}
-			b.handleMessage(r.msg)
 			if ctx.Err() != nil {
 				return nil
 			}
@@ -3401,7 +3463,7 @@ func (b *Broker) handleMessage(msg *Message) {
 // timeout: engine lifecycle ops (install, model pull, ...) run for minutes
 // and report progress via push events, so the broker waits for the real
 // response asynchronously rather than fabricating a timeout — meanwhile
-// other client requests keep being served on the read-loop goroutine.
+// other client requests keep being served on the dispatch-pool goroutine.
 func (b *Broker) relayToEngine(msg *Message) {
 	go b.relayToEngineNow(msg)
 }

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 (

services/nvpair-ui-broker/relay/relay.go

@@ -85,21 +85,26 @@ func registerEqual(a, b noderec.RegisterParams) bool {
 }
 
 // Subscriber is a client interested in directory changes: a service filter and a
-// callback invoked (on the caller's goroutine, under no relay lock) with the
-// subscriber's full filtered node set on every change. Consumers replace their
-// set from it rather than applying deltas, so a dropped or reordered push can't
-// leave them drifted — every push is the authoritative current list.
+// callback invoked with the subscriber's full filtered node set on every change.
+// Consumers replace their set from it rather than applying deltas, so a dropped
+// or reordered push can't leave them drifted — every push is the authoritative
+// current list.
+//
+// Deliveries are asynchronous: each subscriber owns a pump goroutine (started
+// by Directory.Subscribe) that serializes its sends and coalesces concurrent
+// triggers into one delivery that captures the snapshot at send time. A slow or
+// blocked Send therefore stalls only its own subscriber, never the directory
+// update path that feeds it (the scanner's read pump calls Apply).
 type Subscriber struct {
 	Filter noderec.SubscribeParams
 	Send   func(nodes []noderec.DirectoryNode)
 
-	// sendMu serializes deliveries to this subscriber so two concurrent
-	// deliveries — the initial post-subscribe delivery racing an Apply fan-out
-	// driven by the scanner read-pump — can't reorder and leave the subscriber
-	// holding an older set than a newer one. Combined with capturing the snapshot
-	// inside Deliver (at send time, not subscribe time), the last delivery to
-	// acquire it always carries the latest directory state.
-	sendMu sync.Mutex
+	// kick carries a pending-delivery signal (capacity 1: extra signals while
+	// one is already pending coalesce — the pump captures the latest snapshot
+	// when it wakes, so early triggers can't deliver stale state). done closes
+	// on Unsubscribe and stops the pump.
+	kick chan struct{}
+	done chan struct{}
 }
 
 // Directory is the broker's view of all LAN nodes (keyed by hostUuid) plus its
@@ -119,32 +124,69 @@ func NewDirectory() *Directory {
 	}
 }
 
-// Subscribe registers a subscriber and returns its id. The caller sends the
-// initial snapshot via Deliver after releasing its own lock — Deliver captures
-// the snapshot at send time, so a concurrent Apply can't sneak a newer snapshot
-// in and have this initial delivery overwrite it with an older one.
+// Subscribe registers a subscriber and starts its delivery pump goroutine. The
+// initial snapshot arrives via the pump after any pending Deliver call —
+// snapshot is captured at send time, so a concurrent Apply can't sneak a newer
+// snapshot in and have this initial delivery overwrite it with an older one.
 func (d *Directory) Subscribe(sub *Subscriber) (id int) {
+	sub.kick = make(chan struct{}, 1)
+	sub.done = make(chan struct{})
 	d.mu.Lock()
 	d.nextID++
 	id = d.nextID
 	d.subs[id] = sub
 	d.mu.Unlock()
+	go d.pump(sub)
 	return id
 }
 
-// Deliver pushes the subscriber its current filtered snapshot, serialized
-// per-subscriber. Capturing the snapshot here (at delivery time) rather than
-// handing Send a pre-captured slice means a delivery can never carry a set older
-// than the directory's state when it actually runs; the per-subscriber lock then
-// guarantees the initial post-subscribe delivery and a concurrent Apply fan-out
-// settle on the latest set regardless of which runs last.
+// pump serializes one subscriber's deliveries. Every wake re-captures the
+// latest filtered snapshot, so coalesced triggers always deliver current state.
+// done takes priority over a pending kick: once Unsubscribe has closed done, a
+// trigger that raced the close must not produce a Send against a consumer
+// that's gone. The pre-select alone is not enough (a token may already sit in
+// kick), so the kick case re-checks done before sending.
+func (d *Directory) pump(sub *Subscriber) {
+	for {
+		select {
+		case <-sub.done:
+			return
+		default:
+		}
+		select {
+		case <-sub.kick:
+			select {
+			case <-sub.done:
+				return
+			default:
+			}
+			sub.Send(d.filtered(sub.Filter))
+		case <-sub.done:
+			return
+		}
+	}
+}
+
+// Deliver asks for a delivery of the subscriber's current filtered snapshot.
+// Non-blocking: it schedules the send on the subscriber's pump and never blocks
+// the caller — Apply runs on the scanner read pump, and a subscriber whose Send
+// blocks (a stalled worker's stdin pipe) must not stall the directory or the
+// other subscribers. Multiple pending triggers coalesce into one send of the
+// latest state.
 func (d *Directory) Deliver(sub *Subscriber) {
-	sub.sendMu.Lock()
-	defer sub.sendMu.Unlock()
+	select {
+	case sub.kick <- struct{}{}:
+	default:
+	}
+}
+
+// filtered returns the nodes matching a subscriber's filter, sorted by
+// hostUuid for a deterministic set.
+func (d *Directory) filtered(f noderec.SubscribeParams) []noderec.DirectoryNode {
 	d.mu.Lock()
-	nodes := d.filteredLocked(sub.Filter)
+	nodes := d.filteredLocked(f)
 	d.mu.Unlock()
-	sub.Send(nodes)
+	return nodes
 }
 
 // filteredLocked returns the nodes matching a subscriber's filter, sorted by
@@ -160,11 +202,18 @@ func (d *Directory) filteredLocked(f noderec.SubscribeParams) []noderec.Director
 	return out
 }
 
-// Unsubscribe removes a subscriber.
+// Unsubscribe removes a subscriber and stops its delivery pump. It does not
+// wait for the pump to exit: closing done is enough, because the pump's kick
+// case re-checks done before every Send, so a trigger that raced the close
+// cannot deliver to a consumer that's gone.
 func (d *Directory) Unsubscribe(id int) {
 	d.mu.Lock()
+	sub := d.subs[id]
 	delete(d.subs, id)
 	d.mu.Unlock()
+	if sub != nil {
+		close(sub.done)
+	}
 }
 
 // Apply folds a daemon node-* delta into the directory, then re-sends every