From 9d8eb98d2cc4224dbdf12b9b70b4824249d67140 Mon Sep 17 00:00:00 2001 From: PierrunoYT Date: Sun, 19 Jul 2026 15:47:40 +0200 Subject: [PATCH 1/5] fix(lsp): dispatch notifications off read loop --- internal/lsp/client.go | 35 ++++++++++++--- internal/lsp/client_test.go | 85 +++++++++++++++++++++++++++++++++++++ 2 files changed, 115 insertions(+), 5 deletions(-) diff --git a/internal/lsp/client.go b/internal/lsp/client.go index bf821223c..12987a148 100644 --- a/internal/lsp/client.go +++ b/internal/lsp/client.go @@ -32,8 +32,16 @@ type Client struct { closeOnce sync.Once closed chan struct{} readErr error + notify chan notification } +type notification struct { + method string + params json.RawMessage +} + +const notificationQueueSize = 64 + type rpcError struct { Code int `json:"code"` Message string `json:"message"` @@ -84,7 +92,9 @@ func NewClient(r io.Reader, w io.Writer) *Client { writer: w, pending: make(map[int64]chan rpcResponse), closed: make(chan struct{}), + notify: make(chan notification, notificationQueueSize), } + go client.notificationLoop() go client.readLoop(bufio.NewReader(r)) return client } @@ -166,11 +176,10 @@ 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) + select { + case c.notify <- notification{method: msg.Method, params: msg.Params}: + case <-c.closed: + return } case hasID: var id int64 @@ -181,6 +190,22 @@ func (c *Client) readLoop(reader *bufio.Reader) { } } +func (c *Client) notificationLoop() { + for { + select { + case <-c.closed: + return + case notification := <-c.notify: + c.mu.Lock() + handler := c.handler + c.mu.Unlock() + if handler != nil { + handler(notification.method, notification.params) + } + } + } +} + func (c *Client) deliver(id int64, resp rpcResponse) { c.mu.Lock() ch, ok := c.pending[id] diff --git a/internal/lsp/client_test.go b/internal/lsp/client_test.go index 5ec6c39d0..3c53434a6 100644 --- a/internal/lsp/client_test.go +++ b/internal/lsp/client_test.go @@ -227,6 +227,91 @@ func TestClientNotificationHandler(t *testing.T) { } } +func TestClientNotificationHandlerCanCallClient(t *testing.T) { + clientReader, serverWriter := io.Pipe() + serverReader, clientWriter := io.Pipe() + client := NewClient(clientReader, clientWriter) + defer client.Close() + defer serverWriter.Close() + defer clientWriter.Close() + + serverDone := make(chan error, 1) + go func() { + body, err := readMessage(bufio.NewReader(serverReader)) + if err != nil { + serverDone <- err + return + } + var request incomingMessage + if err := json.Unmarshal(body, &request); err != nil { + serverDone <- err + return + } + serverDone <- writeMessage(serverWriter, map[string]any{ + "jsonrpc": "2.0", + "id": request.ID, + "result": map[string]bool{"applied": true}, + }) + }() + + handlerDone := make(chan error, 1) + client.SetNotificationHandler(func(_ string, _ json.RawMessage) { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + _, err := client.Call(ctx, "workspace/applyEdit", nil) + handlerDone <- err + }) + if err := writeMessage(serverWriter, map[string]any{ + "jsonrpc": "2.0", + "method": "workspace/requestEdit", + }); err != nil { + t.Fatal(err) + } + + select { + case err := <-handlerDone: + if err != nil { + t.Fatalf("notification handler call failed: %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("notification handler deadlocked waiting for its response") + } + if err := <-serverDone; err != nil { + t.Fatalf("server failed: %v", err) + } +} + +func TestClientNotificationHandlersPreserveOrder(t *testing.T) { + clientReader, serverWriter := io.Pipe() + client := NewClient(clientReader, io.Discard) + defer client.Close() + defer serverWriter.Close() + + received := make(chan string, 2) + client.SetNotificationHandler(func(method string, _ json.RawMessage) { + received <- method + }) + for _, method := range []string{"first", "second"} { + if err := writeMessage(serverWriter, map[string]any{ + "jsonrpc": "2.0", + "method": method, + }); err != nil { + t.Fatal(err) + } + } + + for _, want := range []string{"first", "second"} { + select { + case got := <-received: + if got != want { + t.Fatalf("notification = %q, want %q", got, want) + } + case <-time.After(2 * time.Second): + t.Fatalf("timed out waiting for notification %q", want) + } + } +} + func TestClientRejectsCallsAfterClose(t *testing.T) { clientReader, serverWriter := io.Pipe() serverReader, clientWriter := io.Pipe() From 372927e8bf32c73df926c87b189dba155a9d7be5 Mon Sep 17 00:00:00 2001 From: PierrunoYT Date: Sun, 19 Jul 2026 16:11:06 +0200 Subject: [PATCH 2/5] fix(lsp): keep notification reads non-blocking --- internal/lsp/client.go | 67 +++++++++++++++++++------- internal/lsp/client_test.go | 95 +++++++++++++++++++++++++++++++++++++ 2 files changed, 146 insertions(+), 16 deletions(-) diff --git a/internal/lsp/client.go b/internal/lsp/client.go index 12987a148..1d2b7febc 100644 --- a/internal/lsp/client.go +++ b/internal/lsp/client.go @@ -32,7 +32,10 @@ type Client struct { closeOnce sync.Once closed chan struct{} readErr error - notify chan notification + + notifyMu sync.Mutex + notifyQueue []notification + notifyReady chan struct{} } type notification struct { @@ -89,10 +92,10 @@ 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{}), - notify: make(chan notification, notificationQueueSize), + 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)) @@ -176,11 +179,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 != "": - select { - case c.notify <- notification{method: msg.Method, params: msg.Params}: - case <-c.closed: - return - } + c.enqueueNotification(notification{method: msg.Method, params: msg.Params}) case hasID: var id int64 if err := json.Unmarshal(msg.ID, &id); err == nil { @@ -195,17 +194,53 @@ func (c *Client) notificationLoop() { select { case <-c.closed: return - case notification := <-c.notify: - c.mu.Lock() - handler := c.handler - c.mu.Unlock() - if handler != nil { - handler(notification.method, notification.params) + 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) + } } } } } +func (c *Client) enqueueNotification(item notification) { + c.notifyMu.Lock() + // Never block protocol reads: at capacity, retain the newest notifications + // and preserve their FIFO order by dropping the oldest queued item. + if len(c.notifyQueue) == notificationQueueSize { + copy(c.notifyQueue, c.notifyQueue[1:]) + c.notifyQueue[len(c.notifyQueue)-1] = item + } else { + c.notifyQueue = append(c.notifyQueue, item) + } + c.notifyMu.Unlock() + + select { + case c.notifyReady <- struct{}{}: + default: + } +} + +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:] + return item, true +} + func (c *Client) deliver(id int64, resp rpcResponse) { c.mu.Lock() ch, ok := c.pending[id] diff --git a/internal/lsp/client_test.go b/internal/lsp/client_test.go index 3c53434a6..0e90e3245 100644 --- a/internal/lsp/client_test.go +++ b/internal/lsp/client_test.go @@ -312,6 +312,101 @@ func TestClientNotificationHandlersPreserveOrder(t *testing.T) { } } +func TestClientNotificationBurstDoesNotBlockResponse(t *testing.T) { + clientReader, serverWriter := io.Pipe() + serverReader, clientWriter := io.Pipe() + client := NewClient(clientReader, clientWriter) + defer client.Close() + defer serverWriter.Close() + defer clientWriter.Close() + + handlerStarted := make(chan struct{}) + handlerDone := make(chan error, 1) + client.SetNotificationHandler(func(method string, _ json.RawMessage) { + if method != "blocking" { + return + } + close(handlerStarted) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + _, err := client.Call(ctx, "workspace/applyEdit", nil) + handlerDone <- err + }) + + serverDone := make(chan error, 1) + go func() { + reader := bufio.NewReader(serverReader) + body, err := readMessage(reader) + if err != nil { + serverDone <- err + return + } + var request incomingMessage + if err := json.Unmarshal(body, &request); err != nil { + serverDone <- err + return + } + for i := 0; i < notificationQueueSize+1; i++ { + if err := writeMessage(serverWriter, map[string]any{ + "jsonrpc": "2.0", + "method": fmt.Sprintf("queued-%03d", i), + }); err != nil { + serverDone <- err + return + } + } + serverDone <- writeMessage(serverWriter, map[string]any{ + "jsonrpc": "2.0", + "id": request.ID, + "result": nil, + }) + }() + + if err := writeMessage(serverWriter, map[string]any{ + "jsonrpc": "2.0", + "method": "blocking", + }); err != nil { + t.Fatal(err) + } + select { + case <-handlerStarted: + case <-time.After(time.Second): + t.Fatal("blocking notification handler did not start") + } + select { + case err := <-handlerDone: + if err != nil { + t.Fatalf("notification handler call failed under burst: %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("notification burst blocked the response read loop") + } + if err := <-serverDone; err != nil { + t.Fatalf("server failed: %v", err) + } +} + +func TestClientNotificationOverflowDropsOldestInOrder(t *testing.T) { + client := &Client{notifyReady: make(chan struct{}, 1)} + for i := 0; i < notificationQueueSize+2; i++ { + client.enqueueNotification(notification{method: fmt.Sprintf("notification-%03d", i)}) + } + + for i := 2; i < notificationQueueSize+2; i++ { + notification, ok := client.dequeueNotification() + if !ok { + t.Fatalf("notification %d missing", i) + } + want := fmt.Sprintf("notification-%03d", i) + if notification.method != want { + t.Fatalf("notification = %q, want %q", notification.method, want) + } + } + if _, ok := client.dequeueNotification(); ok { + t.Fatal("notification queue exceeded its bound") + } +} + func TestClientRejectsCallsAfterClose(t *testing.T) { clientReader, serverWriter := io.Pipe() serverReader, clientWriter := io.Pipe() From 4b3d17b40355187beaf4c0873b0b9f8f21ac19b4 Mon Sep 17 00:00:00 2001 From: PierrunoYT Date: Sat, 25 Jul 2026 15:35:00 +0200 Subject: [PATCH 3/5] fix(lsp): make the notification queue lossless MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses the P1 on #759. The previous commit traded the read-loop deadlock for silent protocol-event loss: at 64 queued items enqueueNotification discarded the OLDEST message. A textDocument/publishDiagnostics is the server's only report for its URI, so a dropped one makes session.waitForDiagnostics time out and Manager.Check return no diagnostics even though the server published them — and the overflow test codified that loss as intended behavior. The queue is now unbounded: the read loop still never blocks (so a handler may make a re-entrant Client.Call), and nothing is discarded. Both alternatives are worse — blocking on a full buffer is the original deadlock, and dropping loses state the protocol never repeats. Growth is bounded in practice by what the server emits while a handler runs; the backing array is released as soon as the queue drains, so a burst does not retain capacity afterwards. Tests: TestClientNotificationQueueIsLossless replaces the drop-oldest test, and the end-to-end burst test now also asserts that all 512 notifications sent while a handler was blocked in a re-entrant call arrive, in order. Both were checked against the old drop-oldest policy and fail there (64 of 512 delivered, FIFO broken), so they have teeth. --- internal/lsp/client.go | 34 ++++++++++++++------- internal/lsp/client_test.go | 60 +++++++++++++++++++++++++++++++------ 2 files changed, 75 insertions(+), 19 deletions(-) diff --git a/internal/lsp/client.go b/internal/lsp/client.go index 1d2b7febc..5c96fe304 100644 --- a/internal/lsp/client.go +++ b/internal/lsp/client.go @@ -43,8 +43,6 @@ type notification struct { params json.RawMessage } -const notificationQueueSize = 64 - type rpcError struct { Code int `json:"code"` Message string `json:"message"` @@ -211,21 +209,32 @@ func (c *Client) notificationLoop() { } } +// 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() - // Never block protocol reads: at capacity, retain the newest notifications - // and preserve their FIFO order by dropping the oldest queued item. - if len(c.notifyQueue) == notificationQueueSize { - copy(c.notifyQueue, c.notifyQueue[1:]) - c.notifyQueue[len(c.notifyQueue)-1] = item - } else { - c.notifyQueue = append(c.notifyQueue, item) - } + 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. } } @@ -238,6 +247,11 @@ func (c *Client) dequeueNotification() (notification, bool) { 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 } diff --git a/internal/lsp/client_test.go b/internal/lsp/client_test.go index 0e90e3245..543aac693 100644 --- a/internal/lsp/client_test.go +++ b/internal/lsp/client_test.go @@ -8,6 +8,7 @@ import ( "fmt" "io" "strings" + "sync" "testing" "time" ) @@ -322,8 +323,18 @@ func TestClientNotificationBurstDoesNotBlockResponse(t *testing.T) { handlerStarted := make(chan struct{}) handlerDone := make(chan error, 1) + var mu sync.Mutex + var queued []string + allQueued := make(chan struct{}) client.SetNotificationHandler(func(method string, _ json.RawMessage) { if method != "blocking" { + mu.Lock() + queued = append(queued, method) + complete := len(queued) == notificationBurstSize + mu.Unlock() + if complete { + close(allQueued) + } return } close(handlerStarted) @@ -346,7 +357,7 @@ func TestClientNotificationBurstDoesNotBlockResponse(t *testing.T) { serverDone <- err return } - for i := 0; i < notificationQueueSize+1; i++ { + for i := 0; i < notificationBurstSize; i++ { if err := writeMessage(serverWriter, map[string]any{ "jsonrpc": "2.0", "method": fmt.Sprintf("queued-%03d", i), @@ -384,26 +395,57 @@ func TestClientNotificationBurstDoesNotBlockResponse(t *testing.T) { if err := <-serverDone; err != nil { t.Fatalf("server failed: %v", err) } + // Not blocking the reader is only half the requirement: every notification the + // server sent during the burst must still reach the handler, in order. A + // bounded queue that drops the oldest would silently lose the head of this + // sequence — and with it a publishDiagnostics the checker is waiting for. + select { + case <-allQueued: + case <-time.After(5 * time.Second): + mu.Lock() + received := len(queued) + mu.Unlock() + t.Fatalf("received %d of %d burst notifications; the queue lost messages", received, notificationBurstSize) + } + mu.Lock() + defer mu.Unlock() + for i, method := range queued { + if want := fmt.Sprintf("queued-%03d", i); method != want { + t.Fatalf("burst notification %d = %q, want %q (order must survive the burst)", i, method, want) + } + } } -func TestClientNotificationOverflowDropsOldestInOrder(t *testing.T) { +// notificationBurstSize is far past any buffer the dispatch used to have, so a +// bounded queue would be visible as either a blocked read loop or a lost message. +const notificationBurstSize = 512 + +// TestClientNotificationQueueIsLossless is the regression test for the overflow +// policy: a queued notification must never be discarded. A dropped +// textDocument/publishDiagnostics is the server's only report for that URI, so +// losing it makes session.waitForDiagnostics time out and Manager.Check return +// nothing even though the server published findings. +func TestClientNotificationQueueIsLossless(t *testing.T) { client := &Client{notifyReady: make(chan struct{}, 1)} - for i := 0; i < notificationQueueSize+2; i++ { + for i := 0; i < notificationBurstSize; i++ { client.enqueueNotification(notification{method: fmt.Sprintf("notification-%03d", i)}) } - for i := 2; i < notificationQueueSize+2; i++ { - notification, ok := client.dequeueNotification() + for i := 0; i < notificationBurstSize; i++ { + item, ok := client.dequeueNotification() if !ok { - t.Fatalf("notification %d missing", i) + t.Fatalf("notification %d was discarded by the queue", i) } want := fmt.Sprintf("notification-%03d", i) - if notification.method != want { - t.Fatalf("notification = %q, want %q", notification.method, want) + if item.method != want { + t.Fatalf("notification = %q, want %q (queue must stay FIFO)", item.method, want) } } if _, ok := client.dequeueNotification(); ok { - t.Fatal("notification queue exceeded its bound") + t.Fatal("queue returned more notifications than were enqueued") + } + if client.notifyQueue != nil { + t.Fatalf("drained queue retained %d slots of capacity", cap(client.notifyQueue)) } } From d8ef0f34ba791ac7dafc903e58b7fc2b8de0cde5 Mon Sep 17 00:00:00 2001 From: PierrunoYT Date: Sat, 25 Jul 2026 15:45:00 +0200 Subject: [PATCH 4/5] fix(lsp): stop retaining notifications after Close MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses the shutdown defect CodeRabbit found in the lossless queue. Client.Close signals the worker loop to exit, but readLoop keeps consuming frames until the transport ends — and Server.Shutdown closes the client BEFORE closing stdin, so a language server emitting notifications while it handles shutdown/exit fed an unbounded queue that nothing would ever drain. failPending's closeOnce now marks the queue closed and releases whatever is in it, and enqueueNotification drops instead of appending once that flag is set. Both run under notifyMu, so an enqueue racing with shutdown either sees the flag and drops its item or appends just before it and has the item cleared. Queued work is dropped rather than drained: the worker is gone by definition of close, and callers that need diagnostics collect them before shutting the client down. TestClientDropsNotificationsAfterClose keeps the transport writable after Close, as Server.Shutdown does, and requires 64 subsequent notifications to neither reach the handler nor accumulate. Verified against the unguarded enqueue, where it reports 64 retained notifications. --- internal/lsp/client.go | 33 ++++++++++++++-- internal/lsp/client_test.go | 76 +++++++++++++++++++++++++++++++++++++ 2 files changed, 106 insertions(+), 3 deletions(-) diff --git a/internal/lsp/client.go b/internal/lsp/client.go index 5c96fe304..63da3c3ee 100644 --- a/internal/lsp/client.go +++ b/internal/lsp/client.go @@ -33,9 +33,10 @@ type Client struct { closed chan struct{} readErr error - notifyMu sync.Mutex - notifyQueue []notification - notifyReady chan struct{} + notifyMu sync.Mutex + notifyQueue []notification + notifyReady chan struct{} + notifyClosed bool } type notification struct { @@ -227,6 +228,15 @@ func (c *Client) notificationLoop() { // 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() @@ -278,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() diff --git a/internal/lsp/client_test.go b/internal/lsp/client_test.go index 543aac693..a061f6570 100644 --- a/internal/lsp/client_test.go +++ b/internal/lsp/client_test.go @@ -465,3 +465,79 @@ func TestClientRejectsCallsAfterClose(t *testing.T) { t.Fatal("Notify after Close must return an error") } } + +// TestClientDropsNotificationsAfterClose covers the shutdown path: Close stops +// the worker loop, but the read loop keeps reading until the transport ends — +// Server.Shutdown closes the client BEFORE closing stdin, so a server emitting +// notifications while it handles shutdown/exit would otherwise pile them into a +// queue nobody drains. Nothing may be handled or retained after Close. +func TestClientDropsNotificationsAfterClose(t *testing.T) { + clientReader, serverWriter := io.Pipe() + serverReader, clientWriter := io.Pipe() + client := NewClient(clientReader, clientWriter) + defer serverWriter.Close() + defer clientWriter.Close() + go func() { + // Drain anything the client writes so no write can block the test. + _, _ = io.Copy(io.Discard, serverReader) + }() + + handled := make(chan string, 4) + client.SetNotificationHandler(func(method string, _ json.RawMessage) { + handled <- method + }) + + // A notification before Close still reaches the handler. + if err := writeMessage(serverWriter, map[string]any{"jsonrpc": "2.0", "method": "before-close"}); err != nil { + t.Fatal(err) + } + select { + case method := <-handled: + if method != "before-close" { + t.Fatalf("handled %q, want before-close", method) + } + case <-time.After(2 * time.Second): + t.Fatal("notification before Close was not handled") + } + + if err := client.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + // The transport stays writable, exactly as it does between Client.Close and + // stdin.Close in Server.Shutdown. + for i := 0; i < 64; i++ { + if err := writeMessage(serverWriter, map[string]any{ + "jsonrpc": "2.0", + "method": fmt.Sprintf("after-close-%03d", i), + }); err != nil { + t.Fatalf("write notification after Close: %v", err) + } + } + + select { + case method := <-handled: + t.Fatalf("handler ran for %q after Close", method) + case <-time.After(200 * time.Millisecond): + } + + // Give the read loop time to consume every frame it was handed, then require + // the queue to hold nothing: dropping on enqueue is what keeps a closed client + // from growing for as long as its reader stays open. + deadline := time.Now().Add(2 * time.Second) + for { + client.notifyMu.Lock() + queued := len(client.notifyQueue) + closed := client.notifyClosed + client.notifyMu.Unlock() + if !closed { + t.Fatal("Close did not mark the notification queue closed") + } + if queued != 0 { + t.Fatalf("closed client retained %d queued notifications", queued) + } + if time.Now().After(deadline) { + return + } + time.Sleep(10 * time.Millisecond) + } +} From 6659ee0b2aa014ab7949a510ba0f05a257bb08b8 Mon Sep 17 00:00:00 2001 From: PierrunoYT Date: Sun, 26 Jul 2026 14:56:29 +0200 Subject: [PATCH 5/5] fix(lsp): baseline diagnostics against receipt, bound the notification backlog Addresses jatmn's two #759 review findings: - Manager.Check captured its diagnostics baseline from publishCount, which only advances once session.handleNotification actually runs for a URI. Dispatch happens off a queue now, so a publish for a version Check is about to supersede could already be sitting received-but-undispatched at baseline-capture time, then run afterward and look newer than the baseline purely because handling lagged receipt. Since many servers omit the version field, the existing staleness check in handleNotification (positive-version only) couldn't catch it either. Stamp every notification with a receipt sequence at enqueue time (Client.NotificationSeq), carry it through the queue to the handler, and baseline against receipt instead of against how many publishes have been processed. - The lossless notification queue had no failure policy: a language server sustaining a higher notification rate than the single handler can drain (no stuck handler required) would retain every full json.RawMessage payload without bound. Add notifyQueueLimit; exceeding it fails and closes the client observably, letting the manager's existing dead-session eviction take over on next use, instead of growing the queue until the heap gives out. Both regression tests are verified to fail against the pre-fix code: the receipt-vs-baseline race was reproduced directly against the pre-fix API (confirmed real, not hypothetical) before porting the test to the fixed API, and the overload test was confirmed to fail with the limit check removed. Co-Authored-By: Claude Sonnet 5 --- internal/lsp/client.go | 73 +++++++++++++++++++++++++++------- internal/lsp/client_test.go | 50 +++++++++++++++++++++--- internal/lsp/documents.go | 66 +++++++++++++++++-------------- internal/lsp/manager.go | 2 +- internal/lsp/manager_test.go | 76 +++++++++++++++++++++++++++++++++++- 5 files changed, 215 insertions(+), 52 deletions(-) diff --git a/internal/lsp/client.go b/internal/lsp/client.go index 63da3c3ee..5298e5381 100644 --- a/internal/lsp/client.go +++ b/internal/lsp/client.go @@ -13,8 +13,14 @@ import ( ) // NotificationHandler receives server->client notifications (e.g. -// textDocument/publishDiagnostics). params is the raw JSON payload. -type NotificationHandler func(method string, params json.RawMessage) +// textDocument/publishDiagnostics). params is the raw JSON payload. seq is the +// notification's receipt sequence (see Client.NotificationSeq): the count of +// notifications read off the wire, including this one, at the moment it was +// enqueued — NOT when this handler happens to run, which can lag receipt when +// the queue is backed up. A caller that needs to know whether something newer +// than a given point has arrived must compare against seq, not against when +// its own handling code runs. +type NotificationHandler func(method string, params json.RawMessage, seq int64) // Client speaks JSON-RPC 2.0 with LSP framing (Content-Length headers) over a // reader/writer pair. It is transport-agnostic: server.go wires it to a process's @@ -37,13 +43,25 @@ type Client struct { notifyQueue []notification notifyReady chan struct{} notifyClosed bool + notifySeq int64 // count of notifications received (enqueued) so far } type notification struct { method string params json.RawMessage + seq int64 } +// notifyQueueLimit bounds the notification backlog. A well-behaved handler +// drains far faster than any single burst fills it; sustained overload — a +// language server emitting faster than the single handler can consume, or a +// handler stuck waiting on a re-entrant Call — is a fatal condition for this +// client, not something to paper over by growing the queue without bound. +// Hitting the limit fails the client (see enqueueNotification): IsClosed +// becomes true, and the manager evicts and restarts the session on next use, +// exactly as it does for any other dead client. +const notifyQueueLimit = 4096 + type rpcError struct { Code int `json:"code"` Message string `json:"message"` @@ -203,7 +221,7 @@ func (c *Client) notificationLoop() { handler := c.handler c.mu.Unlock() if handler != nil { - handler(notification.method, notification.params) + handler(notification.method, notification.params, notification.seq) } } } @@ -211,21 +229,26 @@ func (c *Client) notificationLoop() { } // enqueueNotification hands a server notification to the worker loop. It never -// blocks and never discards: the queue grows instead. +// blocks and never silently discards a message the handler could still act on: +// the queue grows instead, up to notifyQueueLimit. // -// 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. +// The alternatives to growing 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. +// runs, and the queue is released as soon as it drains. But "in practice" is +// not a limit: a handler that never returns, or a server that sustains a +// higher rate than the single handler can drain, would otherwise grow this +// queue's full json.RawMessage payloads without bound until the heap gives +// out. notifyQueueLimit turns that into an explicit, observable failure — +// the client is failed and closed — rather than an unbounded retention of +// protocol input this transport has no business trying to buffer forever. func (c *Client) enqueueNotification(item notification) { c.notifyMu.Lock() if c.notifyClosed { @@ -237,6 +260,13 @@ func (c *Client) enqueueNotification(item notification) { c.notifyMu.Unlock() return } + if len(c.notifyQueue) >= notifyQueueLimit { + c.notifyMu.Unlock() + c.failPending(fmt.Errorf("lsp client: notification backlog exceeded %d messages", notifyQueueLimit)) + return + } + c.notifySeq++ + item.seq = c.notifySeq c.notifyQueue = append(c.notifyQueue, item) c.notifyMu.Unlock() @@ -248,6 +278,19 @@ func (c *Client) enqueueNotification(item notification) { } } +// NotificationSeq returns the number of notifications received (read off the +// wire and enqueued) so far, including any still waiting to be dispatched to +// the handler. A caller that wants to know whether a notification newer than +// "now" has arrived should snapshot this before triggering whatever produces +// it, then require a subsequently-observed seq to be strictly greater: a +// notification already sitting in the queue at snapshot time has seq <= the +// snapshot, even if the handler doesn't run for it until afterward. +func (c *Client) NotificationSeq() int64 { + c.notifyMu.Lock() + defer c.notifyMu.Unlock() + return c.notifySeq +} + func (c *Client) dequeueNotification() (notification, bool) { c.notifyMu.Lock() defer c.notifyMu.Unlock() diff --git a/internal/lsp/client_test.go b/internal/lsp/client_test.go index a061f6570..1a7f105db 100644 --- a/internal/lsp/client_test.go +++ b/internal/lsp/client_test.go @@ -209,7 +209,7 @@ func TestClientNotificationHandler(t *testing.T) { _ = serverReader received := make(chan string, 1) - client.SetNotificationHandler(func(method string, _ json.RawMessage) { + client.SetNotificationHandler(func(method string, _ json.RawMessage, _ int64) { received <- method }) _ = writeMessage(serverWriter, map[string]any{ @@ -256,7 +256,7 @@ func TestClientNotificationHandlerCanCallClient(t *testing.T) { }() handlerDone := make(chan error, 1) - client.SetNotificationHandler(func(_ string, _ json.RawMessage) { + client.SetNotificationHandler(func(_ string, _ json.RawMessage, _ int64) { ctx, cancel := context.WithTimeout(context.Background(), time.Second) defer cancel() _, err := client.Call(ctx, "workspace/applyEdit", nil) @@ -289,7 +289,7 @@ func TestClientNotificationHandlersPreserveOrder(t *testing.T) { defer serverWriter.Close() received := make(chan string, 2) - client.SetNotificationHandler(func(method string, _ json.RawMessage) { + client.SetNotificationHandler(func(method string, _ json.RawMessage, _ int64) { received <- method }) for _, method := range []string{"first", "second"} { @@ -326,7 +326,7 @@ func TestClientNotificationBurstDoesNotBlockResponse(t *testing.T) { var mu sync.Mutex var queued []string allQueued := make(chan struct{}) - client.SetNotificationHandler(func(method string, _ json.RawMessage) { + client.SetNotificationHandler(func(method string, _ json.RawMessage, _ int64) { if method != "blocking" { mu.Lock() queued = append(queued, method) @@ -449,6 +449,46 @@ func TestClientNotificationQueueIsLossless(t *testing.T) { } } +// TestClientFailsOnNotificationBacklogOverload is the regression test for +// jatmn's #759 P2 finding: the lossless queue above had no failure policy — a +// language server sustaining a higher notification rate than the single +// handler can drain (no permanently-stuck handler required, just a sustained +// producer faster than the consumer) would retain every full json.RawMessage +// payload on Zero's heap without bound. Hitting notifyQueueLimit must fail +// (and close) the client observably instead of continuing to grow. +func TestClientFailsOnNotificationBacklogOverload(t *testing.T) { + client := &Client{ + notifyReady: make(chan struct{}, 1), + pending: make(map[int64]chan rpcResponse), + closed: make(chan struct{}), + } + for i := 0; i < notifyQueueLimit; i++ { + client.enqueueNotification(notification{method: "spam"}) + } + if client.IsClosed() { + t.Fatal("client closed before the backlog limit was reached") + } + client.notifyMu.Lock() + queued := len(client.notifyQueue) + client.notifyMu.Unlock() + if queued != notifyQueueLimit { + t.Fatalf("queued = %d, want %d before the limit is exceeded", queued, notifyQueueLimit) + } + + // One push past the limit must fail the client rather than growing the + // queue further. + client.enqueueNotification(notification{method: "spam"}) + if !client.IsClosed() { + t.Fatal("client must be closed once the notification backlog exceeds notifyQueueLimit") + } + client.notifyMu.Lock() + queuedAfter := len(client.notifyQueue) + client.notifyMu.Unlock() + if queuedAfter != 0 { + t.Fatalf("closed client retained %d queued notifications, want 0", queuedAfter) + } +} + func TestClientRejectsCallsAfterClose(t *testing.T) { clientReader, serverWriter := io.Pipe() serverReader, clientWriter := io.Pipe() @@ -483,7 +523,7 @@ func TestClientDropsNotificationsAfterClose(t *testing.T) { }() handled := make(chan string, 4) - client.SetNotificationHandler(func(method string, _ json.RawMessage) { + client.SetNotificationHandler(func(method string, _ json.RawMessage, _ int64) { handled <- method }) diff --git a/internal/lsp/documents.go b/internal/lsp/documents.go index fa7d1ec63..88c9a20c9 100644 --- a/internal/lsp/documents.go +++ b/internal/lsp/documents.go @@ -21,33 +21,33 @@ type session struct { server lspServer client *Client - mu sync.Mutex - open map[string]bool // uri -> didOpen sent - versions map[string]int // uri -> current (committed) version - diagnostics map[string][]Diagnostic // uri -> latest published diagnostics - lastPublish map[string]time.Time // uri -> time of last publish - publishCount map[string]int // uri -> monotonic publish count - waiters map[string][]chan struct{} // uri -> goroutines awaiting the next publish - fileLocks map[string]*sync.Mutex // uri -> per-document sync serializer + mu sync.Mutex + open map[string]bool // uri -> didOpen sent + versions map[string]int // uri -> current (committed) version + diagnostics map[string][]Diagnostic // uri -> latest published diagnostics + lastPublish map[string]time.Time // uri -> time of last publish + publishSeq map[string]int64 // uri -> receipt seq of latest publish (see Client.NotificationSeq) + waiters map[string][]chan struct{} // uri -> goroutines awaiting the next publish + fileLocks map[string]*sync.Mutex // uri -> per-document sync serializer } func newSession(server lspServer) *session { s := &session{ - server: server, - client: server.Client(), - open: map[string]bool{}, - versions: map[string]int{}, - diagnostics: map[string][]Diagnostic{}, - lastPublish: map[string]time.Time{}, - publishCount: map[string]int{}, - waiters: map[string][]chan struct{}{}, - fileLocks: map[string]*sync.Mutex{}, + server: server, + client: server.Client(), + open: map[string]bool{}, + versions: map[string]int{}, + diagnostics: map[string][]Diagnostic{}, + lastPublish: map[string]time.Time{}, + publishSeq: map[string]int64{}, + waiters: map[string][]chan struct{}{}, + fileLocks: map[string]*sync.Mutex{}, } s.client.SetNotificationHandler(s.handleNotification) return s } -func (s *session) handleNotification(method string, params json.RawMessage) { +func (s *session) handleNotification(method string, params json.RawMessage, seq int64) { if method != "textDocument/publishDiagnostics" { return } @@ -66,7 +66,7 @@ func (s *session) handleNotification(method string, params json.RawMessage) { } s.diagnostics[payload.URI] = payload.Diagnostics s.lastPublish[payload.URI] = time.Now() - s.publishCount[payload.URI]++ + s.publishSeq[payload.URI] = seq waiters := s.waiters[payload.URI] delete(s.waiters, payload.URI) s.mu.Unlock() @@ -133,13 +133,21 @@ func (s *session) diagnosticsFor(uri string) []Diagnostic { return append([]Diagnostic(nil), s.diagnostics[uri]...) } -// publishBaseline snapshots the current publish count for a URI, captured before -// a sync so waitForDiagnostics can wait specifically for the publish that sync -// triggers (not be satisfied by a stale earlier publish). -func (s *session) publishBaseline(uri string) int { - s.mu.Lock() - defer s.mu.Unlock() - return s.publishCount[uri] +// publishBaseline snapshots the client's current notification receipt +// sequence, captured before a sync so waitForDiagnostics can require a publish +// that was RECEIVED after this point. This must baseline against receipt, not +// against session.publishSeq (which only advances once handleNotification +// actually runs): dispatch happens off a queue, so a publish for a since- +// superseded version can already be sitting in that queue, still unprocessed, +// at the moment a later Check captures its baseline. If baseline were instead +// "how many publishes has this session recorded so far", that stale publish +// would land afterward, still satisfy "more than baseline", and hand the new +// check diagnostics for the wrong text — silently, since many servers omit the +// version field the staleness check in handleNotification relies on. +// Baselining against receipt sequence closes that: a notification already +// enqueued before this call has seq <= baseline no matter when it is handled. +func (s *session) publishBaseline() int64 { + return s.client.NotificationSeq() } // waitForDiagnostics blocks until a publish newer than baseline arrives for the @@ -149,16 +157,16 @@ func (s *session) publishBaseline(uri string) int { // for newer text. Servers don't signal "analysis complete", so the debounce // approximates it: once a fresh publish lands, wait debounce for a follow-up, // resetting on each new publish. -func (s *session) waitForDiagnostics(ctx context.Context, uri string, debounce time.Duration, baseline int) bool { +func (s *session) waitForDiagnostics(ctx context.Context, uri string, debounce time.Duration, baseline int64) bool { for { s.mu.Lock() ch := make(chan struct{}) s.waiters[uri] = append(s.waiters[uri], ch) - count := s.publishCount[uri] + seq := s.publishSeq[uri] last := s.lastPublish[uri] s.mu.Unlock() - if count <= baseline { + if seq <= baseline { select { case <-ctx.Done(): s.cancelWaiter(uri, ch) diff --git a/internal/lsp/manager.go b/internal/lsp/manager.go index 66d811bef..ef2bce6f3 100644 --- a/internal/lsp/manager.go +++ b/internal/lsp/manager.go @@ -73,7 +73,7 @@ func (m *Manager) Check(ctx context.Context, path, text string) ([]Diagnostic, e } return nil, err } - baseline := sess.publishBaseline(uri) + baseline := sess.publishBaseline() if err := sess.sync(ctx, abs, languageID, text); err != nil { return nil, err } diff --git a/internal/lsp/manager_test.go b/internal/lsp/manager_test.go index e646de138..ac874c8d4 100644 --- a/internal/lsp/manager_test.go +++ b/internal/lsp/manager_test.go @@ -228,18 +228,90 @@ func TestSessionDropsStaleVersionPublish(t *testing.T) { sess.mu.Unlock() stale, _ := json.Marshal(PublishDiagnosticsParams{URI: uri, Version: 2, Diagnostics: []Diagnostic{{Message: "stale"}}}) - sess.handleNotification("textDocument/publishDiagnostics", stale) + sess.handleNotification("textDocument/publishDiagnostics", stale, 1) if len(sess.diagnosticsFor(uri)) != 0 { t.Fatal("a publish for an older version must be ignored") } fresh, _ := json.Marshal(PublishDiagnosticsParams{URI: uri, Version: 3, Diagnostics: []Diagnostic{{Message: "fresh"}}}) - sess.handleNotification("textDocument/publishDiagnostics", fresh) + sess.handleNotification("textDocument/publishDiagnostics", fresh, 2) if d := sess.diagnosticsFor(uri); len(d) != 1 || d[0].Message != "fresh" { t.Fatalf("a current-version publish must apply, got %#v", d) } } +// TestPublishBaselineRejectsAlreadyQueuedPublish is the regression test for +// jatmn's #759 P1 finding: publishBaseline used to snapshot how many publishes +// session.handleNotification had already RUN for a URI. Dispatch happens off a +// queue, though, so a publish for a version Check is about to supersede can +// already be sitting RECEIVED but undispatched at the moment a later Check +// captures its baseline — then run only afterward, incrementing the old +// count-based baseline and wrongly looking "new". Since many servers omit the +// version field, handleNotification's own staleness check (which only fires +// for a positive version) can't catch this either, so the stale result would +// reach the caller for the new text, and the debounce could finish before the +// real response even arrives. +// +// This drives the exact interleaving through the real production path: the +// stale publish is enqueued first (fixing its receipt seq), THEN a baseline is +// captured, THEN the stale item is dispatched — reproducing "received before +// baseline, handled after" without depending on goroutine scheduling luck. A +// bare Client (no background loops, like TestClientNotificationQueueIsLossless +// uses) makes the enqueue/dequeue ordering fully explicit. +func TestPublishBaselineRejectsAlreadyQueuedPublish(t *testing.T) { + client := &Client{notifyReady: make(chan struct{}, 1)} + sess := &session{ + client: client, + versions: map[string]int{}, + diagnostics: map[string][]Diagnostic{}, + lastPublish: map[string]time.Time{}, + publishSeq: map[string]int64{}, + waiters: map[string][]chan struct{}{}, + } + uri := PathToURI("/repo/main.go") + + // The stale (version-less — the common case) publish is RECEIVED first. + stale, _ := json.Marshal(PublishDiagnosticsParams{URI: uri, Diagnostics: []Diagnostic{{Message: "stale"}}}) + client.enqueueNotification(notification{method: "textDocument/publishDiagnostics", params: stale}) + + // A later Check captures its baseline only now — after the stale publish's + // receipt, exactly as publishBaseline does between two Checks whose + // notification queue has backed up. + baseline := sess.publishBaseline() + + // Dispatch the stale item exactly as notificationLoop would: dequeue and + // call the handler with the receipt seq it was actually stamped with. + item, ok := client.dequeueNotification() + if !ok { + t.Fatal("stale notification was not queued") + } + sess.handleNotification(item.method, item.params, item.seq) + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + if sess.waitForDiagnostics(ctx, uri, 10*time.Millisecond, baseline) { + t.Fatal("a publish received before baseline must not satisfy waitForDiagnostics merely because it was handled afterward") + } + + // The real response — received (and handled) after baseline — must satisfy it. + fresh, _ := json.Marshal(PublishDiagnosticsParams{URI: uri, Diagnostics: []Diagnostic{{Message: "fresh"}}}) + client.enqueueNotification(notification{method: "textDocument/publishDiagnostics", params: fresh}) + item2, ok := client.dequeueNotification() + if !ok { + t.Fatal("fresh notification was not queued") + } + sess.handleNotification(item2.method, item2.params, item2.seq) + + ctx2, cancel2 := context.WithTimeout(context.Background(), time.Second) + defer cancel2() + if !sess.waitForDiagnostics(ctx2, uri, 10*time.Millisecond, baseline) { + t.Fatal("a publish received after baseline must satisfy waitForDiagnostics") + } + if d := sess.diagnosticsFor(uri); len(d) != 1 || d[0].Message != "fresh" { + t.Fatalf("diagnostics = %#v, want the fresh publish", d) + } +} + func TestManagerCheckDegradesWhenServerBinaryMissing(t *testing.T) { // A configured extension whose binary isn't on PATH (exec.ErrNotFound) must // degrade to no diagnostics, exactly like an unsupported extension.