fix(lsp): cancel blocked document opens

packages/lsp/lsp-local/README.md

@@ -8,7 +8,7 @@ Namespace plugin (`name` / `inject` / `Config` / `apply`, no default export).
 
 - Resolves every server-local setting before registration; an invalid mapping or registration conflict rolls back earlier entries, so a failed load leaves no provider routes.
 - Lazily single-flights one server process per `(server id, canonical workspace realpath)`. A crash fails the active query without replay; a later query may replace the process.
-- Uses a compatibility-first **transient-open** sequence per query: canonicalize and read the source with Node APIs, `textDocument/didOpen` (version 1, full text), the requested request, then `textDocument/didClose` in `finally`. Documents close after each call, so the first version needs no `didChange`, content cache, or document LRU.
+- Uses a compatibility-first **transient-open** sequence per query: canonicalize and read the source with Node APIs, `textDocument/didOpen` (version 1, full text), the requested request, then `textDocument/didClose` in `finally`. A failed or canceled `didOpen` write terminates the instance before the pool can reuse it. Documents close after each call, so the first version needs no `didChange`, content cache, or document LRU.
 - Serializes each source-read/open/query/close lifecycle through one abortable per-workspace queue so queued calls read current source only when their turn starts; distinct workspaces run in parallel.
 - Reads sources through Node filesystem APIs in the subprocess's host namespace — NOT `ctx.fs`, and emits no `fs/observed`: only the LSP result is model-visible, so a query does not satisfy read-before-write policy.
 

packages/lsp/lsp-local/src/instance.ts

@@ -138,9 +138,16 @@ export class LspInstance {
     try {
       /* v8 ignore next -- guards an abort landing between the ready wait and didOpen; not deterministically reproducible. */
       if (signal?.aborted) throw abortError(signal)
-      await this.connection.notify('textDocument/didOpen', {
-        textDocument: { uri, languageId: request.languageId, version: 1, text: source.text },
-      })
+      try {
+        await abortable(this.connection.notify('textDocument/didOpen', {
+          textDocument: { uri, languageId: request.languageId, version: 1, text: source.text },
+        }), signal)
+      } catch (error) {
+        // A canceled backpressured write or failed stdin leaves the protocol stream unusable before
+        // `opened` can arm the didClose cleanup. Teardown here makes the pool evict the instance.
+        await this.startTeardown()
+        throw error
+      }
       opened = true
       const payload = await this.sendRequest(request.operation, uri, request.position, signal)
       return this.normalize(request.operation, payload)

packages/lsp/lsp-local/tests/fixture-server.ts

@@ -14,6 +14,9 @@
  *   simulating a server that dies while idle so the pool holds a dead instance (eviction test).
  * - LSP_FAKE_REPLY_DELAY_MS: delays each textDocument/* response by this many milliseconds.
  * - LSP_FAKE_OPEN_MARKER: appends each didOpen document text as one JSON line to this path.
+ * - LSP_FAKE_INITIALIZED_MARKER: records when the initialized notification is received.
+ * - LSP_FAKE_PAUSE_STDIN_AFTER_INITIALIZED: "1" stops consuming stdin after initialized.
+ * - LSP_FAKE_CLOSE_STDIN_AFTER_INITIALIZED: "1" closes fd 0 after the initialized notification.
  * - LSP_FAKE_CLOSE_STDIN_AFTER_REPLY: "1" closes fd 0 before sending the first query response.
  * - LSP_FAKE_EXIT_DELAY_MS / LSP_FAKE_EXIT_MARKER: delay protocol exit and record exit/termination.
  * - LSP_FAKE_NO_SHUTDOWN: "1" ignores the shutdown request (forces kill escalation).
@@ -35,6 +38,9 @@ const crashOnOpen = process.env.LSP_FAKE_CRASH_ON_OPEN === '1'
 const exitAfterReply = process.env.LSP_FAKE_EXIT_AFTER_REPLY === '1'
 const replyDelayMs = Number(process.env.LSP_FAKE_REPLY_DELAY_MS ?? 0)
 const openMarker = process.env.LSP_FAKE_OPEN_MARKER
+const initializedMarker = process.env.LSP_FAKE_INITIALIZED_MARKER
+const pauseStdinAfterInitialized = process.env.LSP_FAKE_PAUSE_STDIN_AFTER_INITIALIZED === '1'
+const closeStdinAfterInitialized = process.env.LSP_FAKE_CLOSE_STDIN_AFTER_INITIALIZED === '1'
 const closeStdinAfterReply = process.env.LSP_FAKE_CLOSE_STDIN_AFTER_REPLY === '1'
 const exitDelayMs = Number(process.env.LSP_FAKE_EXIT_DELAY_MS ?? 0)
 const exitMarker = process.env.LSP_FAKE_EXIT_MARKER
@@ -137,7 +143,13 @@ function handle(message: { id?: number; method?: string; params?: unknown; resul
     if (onOpen !== undefined) emitServerRequest(onOpen)
     return
   }
-  if (method === 'textDocument/didClose' || method === 'initialized') return
+  if (method === 'initialized') {
+    if (initializedMarker !== undefined) appendFileSync(initializedMarker, 'INITIALIZED\n')
+    if (pauseStdinAfterInitialized) process.stdin.pause()
+    if (closeStdinAfterInitialized) closeSync(0)
+    return
+  }
+  if (method === 'textDocument/didClose') return
   if (method?.startsWith('textDocument/')) {
     if (hang) return
     const reply = (): void => {
@@ -190,4 +202,6 @@ function send(message: Record<string, unknown>): void {
 
 // Keep the event loop alive.
 process.stdin.resume()
-if (closeStdinAfterReply) setInterval(() => {}, 1000)
+if (pauseStdinAfterInitialized || closeStdinAfterInitialized || closeStdinAfterReply) {
+  setInterval(() => {}, 1000)
+}

packages/lsp/lsp-local/tests/instance.spec.ts

@@ -178,6 +178,40 @@ describe('LspInstance query and abort', () => {
     await instance.dispose()
   })
 
+  it('terminates when abort interrupts a backpressured didOpen write', async () => {
+    // The fixture consumes initialized, then stops reading. A document larger than the stdio pipe
+    // keeps didOpen's write callback pending until cancellation forces bounded process teardown.
+    await writeFile(join(ws, 'a.ts'), 'x'.repeat(2_000_000))
+    const marker = join(root, 'initialized.log')
+    const instance = makeInstance({
+      LSP_FAKE_INITIALIZED_MARKER: marker,
+      LSP_FAKE_PAUSE_STDIN_AFTER_INITIALIZED: '1',
+    }, {
+      shutdownTimeoutMs: 100,
+      killGraceMs: 100,
+    })
+    const controller = new AbortController()
+    const pending = run(instance, 'goToDefinition', controller.signal)
+    await waitForFile(marker)
+    // Let the client enter the large didOpen write after the fixture has paused stdin.
+    await new Promise<void>(resolve => setTimeout(resolve, 100))
+    controller.abort(new Error('didOpen-abort'))
+    await expect(pending).rejects.toThrow(/didOpen-abort/)
+    expect(instance.dead).toBe(true)
+  })
+
+  it('terminates when stdin fails during the didOpen write', async () => {
+    // Closing stdin after initialized makes a large didOpen fail before `opened` can arm didClose;
+    // the instance must still become dead so its provider can replace it.
+    await writeFile(join(ws, 'a.ts'), 'x'.repeat(2_000_000))
+    const instance = makeInstance({ LSP_FAKE_CLOSE_STDIN_AFTER_INITIALIZED: '1' }, {
+      shutdownTimeoutMs: 100,
+      killGraceMs: 100,
+    })
+    await expect(run(instance, 'goToDefinition')).rejects.toThrow()
+    expect(instance.dead).toBe(true)
+  })
+
   it('rejects when the server lacks the operation capability', async () => {
     const instance = makeInstance({ LSP_FAKE_CAPS: JSON.stringify({ definitionProvider: false }), LSP_FAKE_DEF: 'null' })
     await expect(run(instance, 'goToDefinition')).rejects.toThrow(/does not support goToDefinition/)
@@ -287,3 +321,18 @@ function processAlive(pid: number): boolean {
     throw error
   }
 }
+
+/** Wait until a fixture marker exists, bounded so a broken handshake cannot hang the test. */
+async function waitForFile(path: string, timeoutMs = 3000): Promise<void> {
+  const started = Date.now()
+  for (;;) {
+    try {
+      await readFile(path)
+      return
+    } catch (error) {
+      if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error
+    }
+    if (Date.now() - started > timeoutMs) throw new Error('waitForFile timed out')
+    await new Promise<void>(resolve => setTimeout(resolve, 10))
+  }
+}