fix(lsp): dispatch notifications off read loop

internal/lsp/client.go

@@ -32,6 +32,16 @@ type Client struct {
 	closeOnce sync.Once
 	closed    chan struct{}
 	readErr   error
+
+	notifyMu     sync.Mutex
+	notifyQueue  []notification
+	notifyReady  chan struct{}
+	notifyClosed bool
+}
+
+type notification struct {
+	method string
+	params json.RawMessage
 }
 
 type rpcError struct {
@@ -81,10 +91,12 @@ type incomingMessage struct {
 // server process exits); call Close to stop using the client.
 func NewClient(r io.Reader, w io.Writer) *Client {
 	client := &Client{
-		writer:  w,
-		pending: make(map[int64]chan rpcResponse),
-		closed:  make(chan struct{}),
+		writer:      w,
+		pending:     make(map[int64]chan rpcResponse),
+		closed:      make(chan struct{}),
+		notifyReady: make(chan struct{}, 1),
 	}
+	go client.notificationLoop()
 	go client.readLoop(bufio.NewReader(r))
 	return client
 }
@@ -166,12 +178,7 @@ func (c *Client) readLoop(reader *bufio.Reader) {
 			// required or the server can block waiting on it (e.g. registerCapability).
 			_ = c.write(outgoingReply{JSONRPC: "2.0", ID: msg.ID, Result: nil})
 		case msg.Method != "":
-			c.mu.Lock()
-			handler := c.handler
-			c.mu.Unlock()
-			if handler != nil {
-				handler(msg.Method, msg.Params)
-			}
+			c.enqueueNotification(notification{method: msg.Method, params: msg.Params})
 		case hasID:
 			var id int64
 			if err := json.Unmarshal(msg.ID, &id); err == nil {
@@ -181,6 +188,83 @@ func (c *Client) readLoop(reader *bufio.Reader) {
 	}
 }
 
+func (c *Client) notificationLoop() {
+	for {
+		select {
+		case <-c.closed:
+			return
+		case <-c.notifyReady:
+			for {
+				notification, ok := c.dequeueNotification()
+				if !ok {
+					break
+				}
+				c.mu.Lock()
+				handler := c.handler
+				c.mu.Unlock()
+				if handler != nil {
+					handler(notification.method, notification.params)
+				}
+			}
+		}
+	}
+}
+
+// enqueueNotification hands a server notification to the worker loop. It never
+// blocks and never discards: the queue grows instead.
+//
+// Both alternatives are worse. Blocking the read loop when a buffer fills is the
+// deadlock this dispatch exists to avoid — a handler that calls Client.Call waits
+// for a response frame the blocked reader can no longer deliver. Dropping the
+// oldest queued item instead loses protocol state permanently: a
+// textDocument/publishDiagnostics for one URI is the server's only report for
+// that URI, so discarding it makes session.waitForDiagnostics time out and
+// Manager.Check return nothing even though the server published findings.
+//
+// Growth is bounded in practice by how much the server emits while a handler
+// runs, and the queue is released as soon as it drains. A handler that never
+// returns would grow it without limit, but that is a handler bug that queue
+// pressure makes visible rather than one this transport can paper over by
+// silently dropping messages.
+func (c *Client) enqueueNotification(item notification) {
+	c.notifyMu.Lock()
+	if c.notifyClosed {
+		// The worker loop has already stopped, so anything queued now would never
+		// be handled. Retaining it would grow the queue for as long as the
+		// transport stays readable after Close — Server.Shutdown closes the client
+		// before closing stdin, so a server emitting notifications while it handles
+		// shutdown/exit keeps the read loop feeding a queue nobody drains.
+		c.notifyMu.Unlock()
+		return
+	}
+	c.notifyQueue = append(c.notifyQueue, item)
+	c.notifyMu.Unlock()
+
+	select {
+	case c.notifyReady <- struct{}{}:
+	default:
+		// A wake-up is already pending; the worker drains the whole queue per wake,
+		// so this item is covered by it.
+	}
+}
+
+func (c *Client) dequeueNotification() (notification, bool) {
+	c.notifyMu.Lock()
+	defer c.notifyMu.Unlock()
+	if len(c.notifyQueue) == 0 {
+		return notification{}, false
+	}
+	item := c.notifyQueue[0]
+	c.notifyQueue[0] = notification{}
+	c.notifyQueue = c.notifyQueue[1:]
+	if len(c.notifyQueue) == 0 {
+		// Release the backing array once drained; re-slicing alone would keep a
+		// burst's worth of capacity (and its already-consumed items) alive.
+		c.notifyQueue = nil
+	}
+	return item, true
+}
+
 func (c *Client) deliver(id int64, resp rpcResponse) {
 	c.mu.Lock()
 	ch, ok := c.pending[id]
@@ -204,9 +288,26 @@ func (c *Client) failPending(err error) {
 			ch <- rpcResponse{Err: &rpcError{Code: -1, Message: err.Error()}}
 		}
 		close(c.closed)
+		c.closeNotifications()
 	})
 }
 
+// closeNotifications stops accepting notifications and releases anything still
+// queued, so a closed client retains nothing. It runs inside closeOnce, after
+// c.closed is signaled: an enqueue racing with shutdown therefore either sees
+// notifyClosed and drops its item, or appends just before the flag is set and has
+// its item cleared here.
+//
+// Queued-but-unhandled notifications are dropped rather than drained: the worker
+// loop is already gone by definition of close, and callers that need diagnostics
+// (Manager.Check) collect them before shutting the client down.
+func (c *Client) closeNotifications() {
+	c.notifyMu.Lock()
+	c.notifyClosed = true
+	c.notifyQueue = nil
+	c.notifyMu.Unlock()
+}
+
 func (c *Client) readError() error {
 	c.mu.Lock()
 	defer c.mu.Unlock()