diff --git a/internal/lsp/client.go b/internal/lsp/client.go index bf821223..5298e538 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 @@ -32,8 +38,30 @@ type Client struct { closeOnce sync.Once closed chan struct{} readErr error + + notifyMu sync.Mutex + 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"` @@ -81,10 +109,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 +196,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 +206,108 @@ 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, notification.seq) + } + } + } + } +} + +// enqueueNotification hands a server notification to the worker loop. It never +// blocks and never silently discards a message the handler could still act on: +// the queue grows instead, up to notifyQueueLimit. +// +// 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. 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 { + // 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 + } + 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() + + 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. + } +} + +// 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() + 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 +331,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 5ec6c39d..1a7f105d 100644 --- a/internal/lsp/client_test.go +++ b/internal/lsp/client_test.go @@ -8,6 +8,7 @@ import ( "fmt" "io" "strings" + "sync" "testing" "time" ) @@ -208,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{ @@ -227,6 +228,267 @@ 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, _ int64) { + 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, _ int64) { + 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 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) + var mu sync.Mutex + var queued []string + allQueued := make(chan struct{}) + client.SetNotificationHandler(func(method string, _ json.RawMessage, _ int64) { + if method != "blocking" { + mu.Lock() + queued = append(queued, method) + complete := len(queued) == notificationBurstSize + mu.Unlock() + if complete { + close(allQueued) + } + 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 < notificationBurstSize; 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) + } + // 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) + } + } +} + +// 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 < notificationBurstSize; i++ { + client.enqueueNotification(notification{method: fmt.Sprintf("notification-%03d", i)}) + } + + for i := 0; i < notificationBurstSize; i++ { + item, ok := client.dequeueNotification() + if !ok { + t.Fatalf("notification %d was discarded by the queue", i) + } + want := fmt.Sprintf("notification-%03d", i) + if item.method != want { + t.Fatalf("notification = %q, want %q (queue must stay FIFO)", item.method, want) + } + } + if _, ok := client.dequeueNotification(); ok { + 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)) + } +} + +// 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() @@ -243,3 +505,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, _ int64) { + 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) + } +} diff --git a/internal/lsp/documents.go b/internal/lsp/documents.go index fa7d1ec6..88c9a20c 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 66d811be..ef2bce6f 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 e646de13..ac874c8d 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.