fix(lsp): align operations and harden lifecycle
This commit is contained in:
48
packages/lsp/lsp-local/src/abort.ts
Normal file
48
packages/lsp/lsp-local/src/abort.ts
Normal file
@@ -0,0 +1,48 @@
|
||||
/**
|
||||
* Shared cancellation helpers for the local LSP provider's host-I/O, queue, and protocol phases.
|
||||
* @module @deepseek-ai/dsh-lsp-local/abort
|
||||
*/
|
||||
|
||||
import { timeoutOf } from '@deepseek-ai/dsh-timeout'
|
||||
|
||||
/**
|
||||
* Build an abort Error carrying the signal's reason and preserving timeout classification.
|
||||
* @param signal - the aborted signal whose reason to surface.
|
||||
* @returns the timeout reason if present, else the Error reason, else a generic aborted Error.
|
||||
*/
|
||||
export function abortError(signal: AbortSignal): Error {
|
||||
const timeout = timeoutOf(signal)
|
||||
if (timeout !== undefined) return timeout
|
||||
const reason: unknown = signal.reason
|
||||
if (reason instanceof Error) return reason
|
||||
return new Error('LSP query aborted')
|
||||
}
|
||||
|
||||
/**
|
||||
* Throw the signal's classified abort error when it has already fired.
|
||||
* @param signal - the optional query cancellation signal.
|
||||
*/
|
||||
export function throwIfAborted(signal?: AbortSignal): void {
|
||||
if (signal?.aborted) throw abortError(signal)
|
||||
}
|
||||
|
||||
/**
|
||||
* Await work while allowing a query signal to abandon its wait; the underlying work keeps its own
|
||||
* handlers and continues to its owner-defined quiescence boundary.
|
||||
* @param work - the owned asynchronous work.
|
||||
* @param signal - optional query cancellation.
|
||||
* @returns the work result, or a rejection carrying the classified abort reason.
|
||||
*/
|
||||
export function abortable<T>(work: Promise<T>, signal?: AbortSignal): Promise<T> {
|
||||
if (signal === undefined) return work
|
||||
if (signal.aborted) return Promise.reject(abortError(signal))
|
||||
const canceled = Promise.withResolvers<never>()
|
||||
const onAbort = (): void => { canceled.reject(abortError(signal)) }
|
||||
signal.addEventListener('abort', onAbort, { once: true })
|
||||
const normalized = work.catch((error: unknown) => {
|
||||
/* v8 ignore next -- owned LSP promises reject with Error; coercion defends the generic helper. */
|
||||
throw error instanceof Error ? error : new Error(String(error))
|
||||
})
|
||||
return Promise.race([normalized, canceled.promise])
|
||||
.finally(() => { signal.removeEventListener('abort', onAbort) })
|
||||
}
|
||||
@@ -75,10 +75,10 @@ export class LspConnection {
|
||||
})
|
||||
})
|
||||
this.child.on('error', (error) => { this.fail(error) })
|
||||
// A write to the child's stdin after it exits emits an async 'error'; swallow it so an EPIPE
|
||||
// during teardown does not crash the process. Pending requests fail via the 'close' handler.
|
||||
/* v8 ignore next -- the handler only fires on an async stdin write error during teardown. */
|
||||
this.child.stdin.on('error', () => { /* swallow */ })
|
||||
// Child stdin can fail while the process itself remains alive (for example, a server closes fd
|
||||
// 0). Treat that as a fatal connection error so pending requests reject immediately instead of
|
||||
// waiting for a process-close event that may never arrive.
|
||||
this.child.stdin.on('error', (error) => { this.fail(error) })
|
||||
this.child.stdout.on('data', (chunk: Buffer) => { this.onStdout(chunk) })
|
||||
this.child.stderr.on('data', (chunk: Buffer) => { this.onStderr(chunk) })
|
||||
}
|
||||
@@ -108,15 +108,9 @@ export class LspConnection {
|
||||
return
|
||||
}
|
||||
this.pending.set(id, { resolve, reject })
|
||||
try {
|
||||
this.write({ jsonrpc: '2.0', id, method, params })
|
||||
} catch (error) {
|
||||
/* v8 ignore start -- a stdin write failure surfaces asynchronously via the swallowed
|
||||
'error' listener, so this synchronous catch is a defensive guard. */
|
||||
this.pending.delete(id)
|
||||
reject(asError(error))
|
||||
/* v8 ignore stop */
|
||||
}
|
||||
// `write()` records either synchronous or callback-delivered failures on the connection and
|
||||
// rejects every pending request. This handler only consumes the write promise itself.
|
||||
void this.write({ jsonrpc: '2.0', id, method, params }).catch(() => {})
|
||||
})
|
||||
// A caller that stops awaiting (e.g. an aborted query) can leave this promise to reject later
|
||||
// when the process closes; a benign no-op handler keeps that from surfacing as an unhandled
|
||||
@@ -129,9 +123,10 @@ export class LspConnection {
|
||||
* Send a notification (no id, no response).
|
||||
* @param method - the JSON-RPC method.
|
||||
* @param params - the notification params.
|
||||
* @returns a promise that settles when the framed notification has been written.
|
||||
*/
|
||||
notify(method: string, params: unknown): void {
|
||||
this.write({ jsonrpc: '2.0', method, params })
|
||||
notify(method: string, params: unknown): Promise<void> {
|
||||
return this.write({ jsonrpc: '2.0', method, params })
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -139,11 +134,9 @@ export class LspConnection {
|
||||
* @param requestId - the numeric id of the request to cancel.
|
||||
*/
|
||||
cancel(requestId: number): void {
|
||||
try {
|
||||
this.write({ jsonrpc: '2.0', method: '$/cancelRequest', params: { id: requestId } })
|
||||
} catch {
|
||||
// The server is already gone or unwritable; the pending request will fail on close.
|
||||
}
|
||||
// The server is already gone or unwritable when this rejects; `write()` has recorded the fatal
|
||||
// connection failure and rejected the pending request, so cancellation remains best-effort.
|
||||
void this.write({ jsonrpc: '2.0', method: '$/cancelRequest', params: { id: requestId } }).catch(() => {})
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -253,7 +246,10 @@ export class LspConnection {
|
||||
const id = frame.id
|
||||
const method = frame.method
|
||||
if (typeof method === 'string' && (typeof id === 'number' || typeof id === 'string')) {
|
||||
void this.handleServerRequest(id, method, frame.params)
|
||||
// A response-write failure has already invalidated the connection in `write()`.
|
||||
/* v8 ignore next -- protocol tests exercise response writes; only a simultaneous connection
|
||||
failure makes this consumption handler run. */
|
||||
void this.handleServerRequest(id, method, frame.params).catch(() => {})
|
||||
return
|
||||
}
|
||||
if (typeof method === 'string') {
|
||||
@@ -266,9 +262,9 @@ export class LspConnection {
|
||||
private async handleServerRequest(id: number | string, method: string, params: unknown): Promise<void> {
|
||||
try {
|
||||
const result = await this.onServerRequest(method, params)
|
||||
this.write({ jsonrpc: '2.0', id, result })
|
||||
await this.write({ jsonrpc: '2.0', id, result })
|
||||
} catch (error) {
|
||||
this.write({ jsonrpc: '2.0', id, error: { code: -32601, message: asError(error).message } })
|
||||
await this.write({ jsonrpc: '2.0', id, error: { code: -32601, message: asError(error).message } })
|
||||
}
|
||||
}
|
||||
|
||||
@@ -285,8 +281,28 @@ export class LspConnection {
|
||||
pending.resolve(frame.result)
|
||||
}
|
||||
|
||||
private write(message: unknown): void {
|
||||
this.child.stdin.write(encodeMessage(message))
|
||||
private write(message: unknown): Promise<void> {
|
||||
if (this.closeReason !== undefined) return Promise.reject(this.closeReason)
|
||||
return new Promise<void>((resolve, reject) => {
|
||||
const done = (error?: Error | null): void => {
|
||||
if (error === undefined || error === null) {
|
||||
resolve()
|
||||
return
|
||||
}
|
||||
this.fail(error)
|
||||
reject(error)
|
||||
}
|
||||
try {
|
||||
this.child.stdin.write(encodeMessage(message), done)
|
||||
/* v8 ignore start -- Node stream write failures are callback-delivered; this guards a
|
||||
nonconforming Writable implementation throwing synchronously. */
|
||||
} catch (error) {
|
||||
const failure = asError(error)
|
||||
this.fail(failure)
|
||||
reject(failure)
|
||||
}
|
||||
/* v8 ignore stop */
|
||||
})
|
||||
}
|
||||
|
||||
/** The exit-close error message, appending the retained stderr tail when the server wrote any. */
|
||||
|
||||
@@ -13,6 +13,7 @@ import { constants } from 'node:fs'
|
||||
import { open, realpath, stat } from 'node:fs/promises'
|
||||
import type { FileHandle } from 'node:fs/promises'
|
||||
import { isAbsolute, resolve as resolvePath, sep } from 'node:path'
|
||||
import { throwIfAborted } from './abort.ts'
|
||||
|
||||
/** A validated source: its canonical absolute path and current UTF-8 text. */
|
||||
export interface HostSource {
|
||||
@@ -27,17 +28,21 @@ export interface HostSource {
|
||||
* process cwd, `rootUri`, the sole `workspaceFolders` entry, and pool identity, so symlinked roots
|
||||
* collapse to one instance.
|
||||
* @param workspaceRoot - the caller's workspace root (absolute).
|
||||
* @param signal - optional cancellation observed around each filesystem operation.
|
||||
* @returns the canonical directory path.
|
||||
* @throws Error when the path is missing or not a directory.
|
||||
*/
|
||||
export async function canonicalizeWorkspace(workspaceRoot: string): Promise<string> {
|
||||
export async function canonicalizeWorkspace(workspaceRoot: string, signal?: AbortSignal): Promise<string> {
|
||||
throwIfAborted(signal)
|
||||
let canonical: string
|
||||
try {
|
||||
canonical = await realpath(workspaceRoot)
|
||||
} catch (error) {
|
||||
throw new Error(`workspace root "${workspaceRoot}" cannot be resolved: ${messageOf(error)}`)
|
||||
}
|
||||
throwIfAborted(signal)
|
||||
const info = await stat(canonical)
|
||||
throwIfAborted(signal)
|
||||
if (!info.isDirectory()) {
|
||||
throw new Error(`workspace root "${workspaceRoot}" is not a directory`)
|
||||
}
|
||||
@@ -52,6 +57,7 @@ export async function canonicalizeWorkspace(workspaceRoot: string): Promise<stri
|
||||
* @param filePath - the model-supplied source path (relative or absolute).
|
||||
* @param canonicalWorkspace - the already-canonicalized workspace root.
|
||||
* @param maxDocumentBytes - the largest source this host will open.
|
||||
* @param signal - optional cancellation observed throughout resolution, validation, and reading.
|
||||
* @returns the canonical path and current UTF-8 text.
|
||||
* @throws Error when the source is missing, non-regular, oversized, non-UTF-8, or out of workspace.
|
||||
*/
|
||||
@@ -59,7 +65,9 @@ export async function readHostSource(
|
||||
filePath: string,
|
||||
canonicalWorkspace: string,
|
||||
maxDocumentBytes: number,
|
||||
signal?: AbortSignal,
|
||||
): Promise<HostSource> {
|
||||
throwIfAborted(signal)
|
||||
const requested = isAbsolute(filePath) ? filePath : resolvePath(canonicalWorkspace, filePath)
|
||||
let canonicalPath: string
|
||||
try {
|
||||
@@ -67,6 +75,7 @@ export async function readHostSource(
|
||||
} catch (error) {
|
||||
throw new Error(`source "${filePath}" cannot be resolved: ${messageOf(error)}`)
|
||||
}
|
||||
throwIfAborted(signal)
|
||||
if (!isInside(canonicalWorkspace, canonicalPath)) {
|
||||
throw new Error(`source "${filePath}" resolves outside the workspace`)
|
||||
}
|
||||
@@ -74,9 +83,12 @@ export async function readHostSource(
|
||||
// realpath and read cannot swap the target, so the regular-file and size checks bind the bytes we
|
||||
// actually read (no path-based TOCTOU). O_NOFOLLOW rejects the final component being swapped for a
|
||||
// symlink between realpath and open (which would otherwise escape the workspace).
|
||||
const handle = await open(canonicalPath, constants.O_RDONLY | constants.O_NOFOLLOW)
|
||||
// O_NONBLOCK prevents a FIFO with no writer from hanging before fstat can reject it as nonregular.
|
||||
const handle = await open(canonicalPath, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK)
|
||||
try {
|
||||
throwIfAborted(signal)
|
||||
const info = await handle.stat()
|
||||
throwIfAborted(signal)
|
||||
if (!info.isFile()) {
|
||||
throw new Error(`source "${filePath}" is not a regular file`)
|
||||
}
|
||||
@@ -85,8 +97,9 @@ export async function readHostSource(
|
||||
}
|
||||
// Bound the read to the cap even if the file grew after stat: read one extra byte and reject on
|
||||
// overflow, so a concurrent grow cannot defeat the memory bound.
|
||||
const buffer = await readCapped(handle, maxDocumentBytes, filePath)
|
||||
const buffer = await readCapped(handle, maxDocumentBytes, filePath, signal)
|
||||
const text = decodeUtf8Strict(buffer, filePath)
|
||||
throwIfAborted(signal)
|
||||
return { canonicalPath, text }
|
||||
} finally {
|
||||
await handle.close()
|
||||
@@ -94,12 +107,19 @@ export async function readHostSource(
|
||||
}
|
||||
|
||||
/** Read at most `maxBytes` from the handle, rejecting when the source overflows the cap. */
|
||||
async function readCapped(handle: FileHandle, maxBytes: number, filePath: string): Promise<Buffer> {
|
||||
async function readCapped(
|
||||
handle: FileHandle,
|
||||
maxBytes: number,
|
||||
filePath: string,
|
||||
signal?: AbortSignal,
|
||||
): Promise<Buffer> {
|
||||
const limit = maxBytes + 1
|
||||
const chunk = Buffer.allocUnsafe(limit)
|
||||
let total = 0
|
||||
for (;;) {
|
||||
throwIfAborted(signal)
|
||||
const { bytesRead } = await handle.read(chunk, total, limit - total, total)
|
||||
throwIfAborted(signal)
|
||||
if (bytesRead === 0) break
|
||||
total += bytesRead
|
||||
/* v8 ignore next 3 -- overflow requires the file to grow past the cap between stat and read (a concurrent mutation); defensive. */
|
||||
|
||||
@@ -15,14 +15,16 @@ import { accessSync, constants, statSync } from 'node:fs'
|
||||
import { delimiter, isAbsolute, join } from 'node:path'
|
||||
import type { Context } from 'cordis'
|
||||
import z from 'schemastery'
|
||||
import { LspProviderId } from '@deepseek-ai/dsh-lsp'
|
||||
import { LspError, LspProviderId } from '@deepseek-ai/dsh-lsp'
|
||||
import type {
|
||||
LspProvider,
|
||||
LspProviderQuery,
|
||||
LspQueryResult,
|
||||
} from '@deepseek-ai/dsh-lsp'
|
||||
import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'
|
||||
import { abortable, abortError } from './abort.ts'
|
||||
import { canonicalizeWorkspace, readHostSource } from './host.ts'
|
||||
import { abortError, LspInstance } from './instance.ts'
|
||||
import { LspInstance } from './instance.ts'
|
||||
import type { InstanceSpec } from './instance.ts'
|
||||
|
||||
export { canonicalizeWorkspace, readHostSource } from './host.ts'
|
||||
@@ -75,7 +77,7 @@ export interface LspLocalServerConfig {
|
||||
maxDocumentBytes?: number
|
||||
/** Graceful `shutdown`/`exit` budget before escalation (ms). Default 5000. */
|
||||
shutdownTimeoutMs?: number
|
||||
/** SIGTERM→SIGKILL grace after graceful shutdown fails (ms). Default 2000. */
|
||||
/** Request-cancel and SIGTERM→SIGKILL grace (ms). Default 2000. */
|
||||
killGraceMs?: number
|
||||
}
|
||||
|
||||
@@ -98,8 +100,8 @@ const LspLocalServerConfig: z<LspLocalServerConfig> = z.object({
|
||||
maxMessageBytes: z.number().default(DEFAULT_MAX_MESSAGE_BYTES),
|
||||
maxStderrBytes: z.number().default(DEFAULT_MAX_STDERR_BYTES),
|
||||
maxDocumentBytes: z.number().default(DEFAULT_MAX_DOCUMENT_BYTES),
|
||||
shutdownTimeoutMs: z.number().default(DEFAULT_SHUTDOWN_TIMEOUT_MS),
|
||||
killGraceMs: z.number().default(DEFAULT_KILL_GRACE_MS),
|
||||
shutdownTimeoutMs: z.number().max(MAX_TIMER_DELAY_MS).default(DEFAULT_SHUTDOWN_TIMEOUT_MS),
|
||||
killGraceMs: z.number().max(MAX_TIMER_DELAY_MS).default(DEFAULT_KILL_GRACE_MS),
|
||||
})
|
||||
|
||||
export const Config: z<Config> = z.object({
|
||||
@@ -148,8 +150,8 @@ export function apply(ctx: Context, config: Config): void {
|
||||
function validateServerConfig(providerId: string, resolved: ResolvedServerConfig): void {
|
||||
// Teardown budgets feed `deadline()`, whose `<= 0` is the internal no-timeout sentinel; a
|
||||
// nonpositive value would let a server that ignores shutdown hang disposal forever. Fail at load.
|
||||
assertPositiveInteger(providerId, 'shutdownTimeoutMs', resolved.shutdownTimeoutMs)
|
||||
assertPositiveInteger(providerId, 'killGraceMs', resolved.killGraceMs)
|
||||
assertTimer(providerId, 'shutdownTimeoutMs', resolved.shutdownTimeoutMs)
|
||||
assertTimer(providerId, 'killGraceMs', resolved.killGraceMs)
|
||||
// Byte caps must be positive: a nonpositive stderr cap defeats the retained-tail bound
|
||||
// (`slice(-0)` keeps everything), `maxMessageBytes: 0` makes every response fatal, and a bad
|
||||
// document cap fails later in the read path instead of at load.
|
||||
@@ -158,6 +160,13 @@ function validateServerConfig(providerId: string, resolved: ResolvedServerConfig
|
||||
assertPositiveInteger(providerId, 'maxDocumentBytes', resolved.maxDocumentBytes)
|
||||
}
|
||||
|
||||
/** Reject a timer value Node would clamp instead of scheduling as configured. */
|
||||
function assertTimer(providerId: string, name: string, value: number): void {
|
||||
if (!Number.isInteger(value) || value < 1 || value > MAX_TIMER_DELAY_MS) {
|
||||
throw new Error(`lsp-local: servers.${providerId}.${name} must be a positive integer no greater than ${MAX_TIMER_DELAY_MS}`)
|
||||
}
|
||||
}
|
||||
|
||||
/** Reject a nonpositive or non-integer config value at load, so misconfiguration fails loud. */
|
||||
function assertPositiveInteger(providerId: string, name: string, value: number): void {
|
||||
if (!Number.isInteger(value) || value < 1) {
|
||||
@@ -171,6 +180,8 @@ class LocalLspProvider implements LspProvider {
|
||||
readonly extensionToLanguage: Readonly<Record<string, string>>
|
||||
/** One live instance per canonical workspace realpath. */
|
||||
private readonly instances = new Map<string, LspInstance>()
|
||||
/** One complete source-read→open→query→close serialization tail per canonical workspace. */
|
||||
private readonly queues = new Map<string, Promise<void>>()
|
||||
private disposed = false
|
||||
|
||||
constructor(
|
||||
@@ -192,35 +203,49 @@ class LocalLspProvider implements LspProvider {
|
||||
private assertActive(signal?: AbortSignal): void {
|
||||
/* v8 ignore next -- the seam unregisters this provider before disposal; direct in-flight calls
|
||||
exercise the post-await check instead. */
|
||||
if (this.isDisposed()) throw new Error('lsp-local provider is disposed')
|
||||
if (this.isDisposed()) throw new LspError('lsp-local provider is disposed', 'LSP_DISPOSED')
|
||||
if (signal?.aborted) throw abortError(signal)
|
||||
}
|
||||
|
||||
async query(request: LspProviderQuery, signal?: AbortSignal): Promise<LspQueryResult> {
|
||||
// Honor an already-aborted signal before host I/O so a canceled request never starts a server.
|
||||
this.assertActive(signal)
|
||||
const workspace = await canonicalizeWorkspace(request.workspaceRoot)
|
||||
// Validate and read the source BEFORE spawning a server: a missing/external/non-regular/oversized
|
||||
// source must fail without leaving an idle process pooled (the pre-start rejection contract), and
|
||||
// the single-handle read preserves the containment/size checks against a mid-read swap.
|
||||
const source = await readHostSource(request.filePath, workspace, this.config.maxDocumentBytes)
|
||||
// Re-check disposal after the awaits: disposeAll() may have snapshotted the instance map while we
|
||||
// were canonicalizing/reading, so creating a server now would leave it unowned by teardown.
|
||||
const workspace = await canonicalizeWorkspace(request.workspaceRoot, signal)
|
||||
this.assertActive(signal)
|
||||
let instance = this.instanceFor(workspace)
|
||||
// Eviction and replacement are synchronous so disposal cannot snapshot the pool between them and
|
||||
// miss a newly spawned process.
|
||||
if (instance.dead) {
|
||||
this.evictIfCurrent(workspace, instance)
|
||||
instance = this.instanceFor(workspace)
|
||||
}
|
||||
try {
|
||||
return await instance.query(request, source, signal)
|
||||
} finally {
|
||||
// A crashed/closed process must not be reused: drop its slot so the next query starts fresh,
|
||||
// but only if the slot still holds THIS instance (a concurrent replacement must survive).
|
||||
if (instance.dead) this.evictIfCurrent(workspace, instance)
|
||||
}
|
||||
return this.enqueue(workspace, signal, async () => {
|
||||
this.assertActive(signal)
|
||||
// Read inside the workspace queue but before spawning: a queued query sees current bytes when
|
||||
// its turn starts, while an invalid source still cannot leave an idle process pooled.
|
||||
const source = await readHostSource(request.filePath, workspace, this.config.maxDocumentBytes, signal)
|
||||
// Disposal may have snapshotted the instance map while host I/O was pending. Re-check before a
|
||||
// synchronous get-or-create so every spawned process remains owned by teardown.
|
||||
this.assertActive(signal)
|
||||
let instance = this.instanceFor(workspace)
|
||||
if (instance.dead) {
|
||||
this.evictIfCurrent(workspace, instance)
|
||||
instance = this.instanceFor(workspace)
|
||||
}
|
||||
try {
|
||||
return await instance.query(request, source, signal)
|
||||
} finally {
|
||||
// Drop a crashed slot only when it still owns this instance; a replacement must survive.
|
||||
if (instance.dead) this.evictIfCurrent(workspace, instance)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/** Serialize one complete query lifecycle for a canonical workspace. */
|
||||
private enqueue<T>(workspace: string, signal: AbortSignal | undefined, run: () => Promise<T>): Promise<T> {
|
||||
const previous = this.queues.get(workspace) ?? Promise.resolve()
|
||||
const result = abortable(previous, signal).then(run)
|
||||
// The tail follows the actual prior work even when this caller aborts its wait. It never rejects,
|
||||
// so later callers serialize without inheriting an earlier query's outcome.
|
||||
const tail = previous.then(() => result).then(() => undefined, () => undefined)
|
||||
this.queues.set(workspace, tail)
|
||||
void tail.then(() => {
|
||||
if (this.queues.get(workspace) === tail) this.queues.delete(workspace)
|
||||
})
|
||||
return result
|
||||
}
|
||||
|
||||
/** Return or synchronously publish the one instance for a canonical workspace. */
|
||||
@@ -259,8 +284,13 @@ class LocalLspProvider implements LspProvider {
|
||||
async disposeAll(): Promise<void> {
|
||||
this.disposed = true
|
||||
const live = [...this.instances.values()]
|
||||
const draining = [...this.queues.values()]
|
||||
this.instances.clear()
|
||||
await Promise.all(live.map(instance => instance.dispose()))
|
||||
await Promise.all([
|
||||
...live.map(instance => instance.dispose()),
|
||||
...draining,
|
||||
])
|
||||
this.queues.clear()
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -14,7 +14,8 @@ import type {
|
||||
LspProviderQuery,
|
||||
LspQueryResult,
|
||||
} from '@deepseek-ai/dsh-lsp'
|
||||
import { deadline, timeoutOf } from '@deepseek-ai/dsh-timeout'
|
||||
import { deadline } from '@deepseek-ai/dsh-timeout'
|
||||
import { abortable, abortError } from './abort.ts'
|
||||
import { LspConnection } from './connection.ts'
|
||||
import type { ConnectionSpec } from './connection.ts'
|
||||
import type { HostSource } from './host.ts'
|
||||
@@ -83,7 +84,7 @@ export class LspInstance {
|
||||
// Serialize behind prior work, but observe abort DURING the queue wait too: if an earlier query
|
||||
// hangs (e.g. a signal-less seam caller), a later tool's timeout must still be able to give up
|
||||
// rather than block on the shared tail forever.
|
||||
const run = this.abortable(this.queue, signal).then(() => this.runQuery(request, source, signal))
|
||||
const run = abortable(this.queue, signal).then(() => this.runQuery(request, source, signal))
|
||||
// Keep the tail alive regardless of this query's outcome so the next caller still serializes. The
|
||||
// tail follows the ACTUAL prior work (this.queue), not the abortable view, so a caller giving up
|
||||
// on the wait does not deserialize the queue.
|
||||
@@ -103,11 +104,11 @@ export class LspInstance {
|
||||
// An omitted encoding defaults to utf-16; any other value is a protocol error we reject here.
|
||||
negotiatePositionEncoding(capabilities.positionEncoding)
|
||||
this.capabilities = capabilities
|
||||
this.connection.notify('initialized', {})
|
||||
await this.connection.notify('initialized', {})
|
||||
}
|
||||
|
||||
private async runQuery(request: LspProviderQuery, source: HostSource, signal?: AbortSignal): Promise<LspQueryResult> {
|
||||
if (this.disposed) throw new Error('LSP instance was disposed')
|
||||
if (this.disposed) throw new LspError('LSP instance was disposed', 'LSP_DISPOSED')
|
||||
/* v8 ignore next -- the abortable queue wait rejects a pre-aborted signal before runQuery; this is a belt-and-suspenders guard. */
|
||||
if (signal?.aborted) throw abortError(signal)
|
||||
// Observe abort during the handshake wait, and never pool a poisoned instance: if the wait ends
|
||||
@@ -115,11 +116,10 @@ export class LspInstance {
|
||||
// negotiation, malformed result) without the process exiting — tear the instance down so a
|
||||
// permanently-rejecting/pending `ready` can't make every later query for this workspace fail.
|
||||
try {
|
||||
await this.abortable(this.ready, signal)
|
||||
await abortable(this.ready, signal)
|
||||
} catch (error) {
|
||||
if (!this.dead) {
|
||||
/* v8 ignore next -- ready rejects with an Error (abort reason or initialize failure); the String() fallback is defensive. */
|
||||
await this.startTeardown(error instanceof Error ? error : new Error(String(error)))
|
||||
await this.startTeardown()
|
||||
}
|
||||
throw error
|
||||
}
|
||||
@@ -138,7 +138,7 @@ 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)
|
||||
this.connection.notify('textDocument/didOpen', {
|
||||
await this.connection.notify('textDocument/didOpen', {
|
||||
textDocument: { uri, languageId: request.languageId, version: 1, text: source.text },
|
||||
})
|
||||
opened = true
|
||||
@@ -150,34 +150,21 @@ export class LspInstance {
|
||||
// the next queued query's document lifecycle overlap the still-active request.
|
||||
if (opened && !this.dead) {
|
||||
try {
|
||||
this.connection.notify('textDocument/didClose', { textDocument: { uri } })
|
||||
} catch (error) {
|
||||
/* v8 ignore start -- stdin write errors surface asynchronously via the swallowed 'error'
|
||||
listener, so a synchronous didClose write failure is a defensive path. */
|
||||
await this.connection.notify('textDocument/didClose', { textDocument: { uri } })
|
||||
} catch {
|
||||
// A close-write failure does not replace the settled result/error, but the instance can no
|
||||
// longer be trusted: invalidate it and await bounded process termination.
|
||||
void this.startTeardown(error instanceof Error ? error : new Error(String(error)))
|
||||
/* v8 ignore stop */
|
||||
try {
|
||||
await this.startTeardown()
|
||||
} catch {
|
||||
/* v8 ignore next -- teardown owns all expected process races; this only preserves the
|
||||
already-settled query outcome if an unexpected cleanup primitive itself rejects. */
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Await `work`, but reject as soon as `signal` aborts. The underlying `work` promise keeps its own
|
||||
* handlers, so an orphaned rejection after abort is not unhandled.
|
||||
*/
|
||||
private abortable<T>(work: Promise<T>, signal: AbortSignal | undefined): Promise<T> {
|
||||
if (signal === undefined) return work
|
||||
/* v8 ignore next -- callers either pre-check the signal or pass a freshly armed teardown deadline. */
|
||||
if (signal.aborted) return Promise.reject(abortError(signal))
|
||||
return new Promise<T>((resolve, reject) => {
|
||||
const onAbort = (): void => { reject(abortError(signal)) }
|
||||
signal.addEventListener('abort', onAbort, { once: true })
|
||||
work.then(resolve, reject).finally(() => { signal.removeEventListener('abort', onAbort) })
|
||||
})
|
||||
}
|
||||
|
||||
private async sendRequest(
|
||||
operation: LspOperation,
|
||||
uri: string,
|
||||
@@ -187,9 +174,9 @@ export class LspInstance {
|
||||
const params = {
|
||||
textDocument: { uri },
|
||||
position: { line: position.line, character: position.character },
|
||||
// references always includes declarations: the caller gets no flag and impact analysis never
|
||||
// omits the defining site.
|
||||
...(operation === 'references' ? { context: { includeDeclaration: true } } : {}),
|
||||
// findReferences always includes declarations: the caller gets no flag and impact analysis
|
||||
// never omits the defining site.
|
||||
...(operation === 'findReferences' ? { context: { includeDeclaration: true } } : {}),
|
||||
}
|
||||
const requestId = this.connection.peekNextId()
|
||||
const send = this.connection.request(requestMethod(operation), params)
|
||||
@@ -204,7 +191,7 @@ export class LspInstance {
|
||||
*/
|
||||
private async raceAbort(send: Promise<unknown>, requestId: number, signal: AbortSignal): Promise<unknown> {
|
||||
try {
|
||||
return await this.abortable(send, signal)
|
||||
return await abortable(send, signal)
|
||||
} catch (error) {
|
||||
if (!signal.aborted) throw error
|
||||
this.connection.cancel(requestId)
|
||||
@@ -221,7 +208,7 @@ export class LspInstance {
|
||||
grace.signal.addEventListener('abort', () => { resolve(false) }, { once: true })
|
||||
}),
|
||||
])
|
||||
if (!settled) await this.startTeardown(abortError(signal))
|
||||
if (!settled) await this.startTeardown()
|
||||
} finally {
|
||||
grace[Symbol.dispose]()
|
||||
}
|
||||
@@ -263,17 +250,17 @@ export class LspInstance {
|
||||
* process close so nothing outlives disposal.
|
||||
*/
|
||||
async dispose(): Promise<void> {
|
||||
await this.startTeardown(new Error('LSP instance disposed'))
|
||||
await this.startTeardown()
|
||||
}
|
||||
|
||||
/** Publish disposal once and make every caller await the same quiescence boundary. */
|
||||
private startTeardown(reason: Error): Promise<void> {
|
||||
private startTeardown(): Promise<void> {
|
||||
this.disposed = true
|
||||
this.teardownPromise ??= this.tearDown(reason)
|
||||
this.teardownPromise ??= this.tearDown()
|
||||
return this.teardownPromise
|
||||
}
|
||||
|
||||
private async tearDown(_reason: Error): Promise<void> {
|
||||
private async tearDown(): Promise<void> {
|
||||
const shutdownDeadline = deadline(undefined, this.spec.shutdownTimeoutMs, 'LSP_SHUTDOWN')
|
||||
try {
|
||||
await this.gracefulShutdown(shutdownDeadline.signal)
|
||||
@@ -287,9 +274,9 @@ export class LspInstance {
|
||||
|
||||
/** Best-effort LSP `shutdown`/`exit`, including process close, bounded by `signal`. */
|
||||
private async gracefulShutdown(signal: AbortSignal): Promise<void> {
|
||||
await this.abortable(this.connection.request('shutdown', null), signal)
|
||||
this.connection.notify('exit', null)
|
||||
await this.abortable(this.connection.closed, signal)
|
||||
await abortable(this.connection.request('shutdown', null), signal)
|
||||
await this.connection.notify('exit', null)
|
||||
await abortable(this.connection.closed, signal)
|
||||
}
|
||||
|
||||
/** SIGTERM the group, escalate after `killGraceMs`, then await leader and helper exit. */
|
||||
@@ -322,19 +309,6 @@ function markSettled(): boolean {
|
||||
return true
|
||||
}
|
||||
|
||||
/**
|
||||
* Build an abort Error carrying the signal's reason (preserving a timeout classification).
|
||||
* @param signal - the aborted signal whose reason to surface.
|
||||
* @returns the timeout reason if the signal carries one, else the signal's Error reason, else a generic aborted Error.
|
||||
*/
|
||||
export function abortError(signal: AbortSignal): Error {
|
||||
const timeout = timeoutOf(signal)
|
||||
if (timeout !== undefined) return timeout
|
||||
const reason: unknown = signal.reason
|
||||
if (reason instanceof Error) return reason
|
||||
return new Error('LSP query aborted')
|
||||
}
|
||||
|
||||
/**
|
||||
* The client capabilities advertised at `initialize`: UTF-16 positions, workspace folders and
|
||||
* configuration, markdown/plaintext hover, and link support for definition/implementation. No
|
||||
|
||||
@@ -11,6 +11,7 @@ import type {
|
||||
LspOperation,
|
||||
LspRange,
|
||||
} from '@deepseek-ai/dsh-lsp'
|
||||
import { LspError } from '@deepseek-ai/dsh-lsp'
|
||||
import { assertNever } from '@deepseek-ai/dsh-llm'
|
||||
import type {
|
||||
WireHover,
|
||||
@@ -30,9 +31,9 @@ import type {
|
||||
*/
|
||||
export function requestMethod(operation: LspOperation): string {
|
||||
switch (operation) {
|
||||
case 'definition': return 'textDocument/definition'
|
||||
case 'references': return 'textDocument/references'
|
||||
case 'implementation': return 'textDocument/implementation'
|
||||
case 'goToDefinition': return 'textDocument/definition'
|
||||
case 'findReferences': return 'textDocument/references'
|
||||
case 'goToImplementation': return 'textDocument/implementation'
|
||||
case 'hover': return 'textDocument/hover'
|
||||
/* v8 ignore next -- exhaustive over the closed LspOperation union; unreachable. */
|
||||
default: return assertNever(operation, 'requestMethod')
|
||||
@@ -42,9 +43,9 @@ export function requestMethod(operation: LspOperation): string {
|
||||
/** The `ServerCapabilities` provider field backing each operation. */
|
||||
function capabilityValue(capabilities: WireServerCapabilities, operation: LspOperation): WireProviderCapability {
|
||||
switch (operation) {
|
||||
case 'definition': return capabilities.definitionProvider
|
||||
case 'references': return capabilities.referencesProvider
|
||||
case 'implementation': return capabilities.implementationProvider
|
||||
case 'goToDefinition': return capabilities.definitionProvider
|
||||
case 'findReferences': return capabilities.referencesProvider
|
||||
case 'goToImplementation': return capabilities.implementationProvider
|
||||
case 'hover': return capabilities.hoverProvider
|
||||
/* v8 ignore next -- exhaustive over the closed LspOperation union; unreachable. */
|
||||
default: return assertNever(operation, 'capabilityValue')
|
||||
@@ -127,7 +128,12 @@ function isRange(value: unknown): value is WireRange {
|
||||
function isPosition(value: unknown): boolean {
|
||||
if (value === null || typeof value !== 'object') return false
|
||||
const position = value as Record<string, unknown>
|
||||
return typeof position.line === 'number' && typeof position.character === 'number'
|
||||
return isProtocolCoordinate(position.line) && isProtocolCoordinate(position.character)
|
||||
}
|
||||
|
||||
/** Whether a wire coordinate is a valid nonnegative integer. */
|
||||
function isProtocolCoordinate(value: unknown): value is number {
|
||||
return typeof value === 'number' && Number.isInteger(value) && value >= 0
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -138,12 +144,13 @@ function isPosition(value: unknown): boolean {
|
||||
* @throws Error when an element is neither a `Location` nor a `LocationLink`.
|
||||
*/
|
||||
export function normalizeLocations(payload: unknown): LspLocation[] {
|
||||
if (payload === null || payload === undefined) return []
|
||||
if (payload === null) return []
|
||||
if (payload === undefined) throw malformedResponse('LSP navigation result was missing')
|
||||
const elements = Array.isArray(payload) ? payload : [payload]
|
||||
const locations: LspLocation[] = []
|
||||
for (const element of elements) {
|
||||
if (element === null || typeof element !== 'object') {
|
||||
throw new Error('LSP navigation result contained a non-object entry')
|
||||
throw malformedResponse('LSP navigation result contained a non-object entry')
|
||||
}
|
||||
const record = element as Record<string, unknown>
|
||||
if (isLocationLink(record)) {
|
||||
@@ -153,7 +160,7 @@ export function normalizeLocations(payload: unknown): LspLocation[] {
|
||||
const location = record as unknown as WireLocation
|
||||
locations.push({ uri: location.uri, range: toRange(location.range) })
|
||||
} else {
|
||||
throw new Error('LSP navigation result contained neither a Location nor a LocationLink')
|
||||
throw malformedResponse('LSP navigation result contained neither a Location nor a LocationLink')
|
||||
}
|
||||
}
|
||||
return locations
|
||||
@@ -168,39 +175,61 @@ function renderMarkedString(value: WireMarkedString): string {
|
||||
/**
|
||||
* Normalize a `Hover` (or `null`) to the seam's hover. `MarkupContent` uses its `value`; a string
|
||||
* `MarkedString` is verbatim; a language-tagged `MarkedString` becomes a fenced code block; an array
|
||||
* joins its rendered parts with one blank line. `maxHoverChars` is NOT applied here — the tool caps.
|
||||
* joins its rendered parts with one blank line. The model-facing tool owns the complete result cap.
|
||||
* @param payload - the raw `textDocument/hover` result.
|
||||
* @returns the normalized hover, or `null` when there is no content.
|
||||
* @throws Error when the payload is a non-null, non-object, or structurally invalid hover.
|
||||
*/
|
||||
export function normalizeHover(payload: unknown): LspHover | null {
|
||||
if (payload === null || payload === undefined) return null
|
||||
if (typeof payload !== 'object') throw new Error('LSP hover result was not an object')
|
||||
if (payload === null) return null
|
||||
if (payload === undefined) throw malformedResponse('LSP hover result was missing')
|
||||
if (typeof payload !== 'object') throw malformedResponse('LSP hover result was not an object')
|
||||
const hover = payload as unknown as WireHover
|
||||
const contents = renderHoverContents(hover.contents)
|
||||
if (contents === '') return null
|
||||
const range = hover.range
|
||||
return range !== undefined && isRange(range) ? { contents, range: toRange(range) } : { contents }
|
||||
if (range === undefined) return { contents }
|
||||
if (!isRange(range)) throw malformedResponse('LSP hover result contained a malformed range')
|
||||
return { contents, range: toRange(range) }
|
||||
}
|
||||
|
||||
/** Render the three `Hover.contents` encodings into one string (input is untrusted wire data). */
|
||||
function renderHoverContents(contents: unknown): string {
|
||||
if (contents === null || contents === undefined) {
|
||||
throw new Error('LSP hover result had no contents')
|
||||
throw malformedResponse('LSP hover result had no contents')
|
||||
}
|
||||
if (typeof contents === 'string') return contents
|
||||
if (Array.isArray(contents)) {
|
||||
return contents.map(renderMarkedString).join('\n\n')
|
||||
return contents.map((value) => {
|
||||
if (isMarkedString(value)) return renderMarkedString(value)
|
||||
throw malformedResponse('LSP hover contents contained a malformed MarkedString')
|
||||
}).join('\n\n')
|
||||
}
|
||||
if (typeof contents !== 'object') {
|
||||
throw new Error('LSP hover contents were not MarkupContent, MarkedString, or an array')
|
||||
throw malformedResponse('LSP hover contents were not MarkupContent, MarkedString, or an array')
|
||||
}
|
||||
const record = contents as Record<string, unknown>
|
||||
if (record.kind === 'markdown' || record.kind === 'plaintext') {
|
||||
return typeof record.value === 'string' ? record.value : ''
|
||||
if (typeof record.value !== 'string') {
|
||||
throw malformedResponse('LSP hover MarkupContent value was not a string')
|
||||
}
|
||||
return record.value
|
||||
}
|
||||
if (typeof record.language === 'string' && typeof record.value === 'string') {
|
||||
return renderMarkedString({ language: record.language, value: record.value })
|
||||
}
|
||||
throw new Error('LSP hover contents were not MarkupContent, MarkedString, or an array')
|
||||
throw malformedResponse('LSP hover contents were not MarkupContent, MarkedString, or an array')
|
||||
}
|
||||
|
||||
/** Whether an untrusted value is either form of `MarkedString`. */
|
||||
function isMarkedString(value: unknown): value is WireMarkedString {
|
||||
if (typeof value === 'string') return true
|
||||
if (value === null || typeof value !== 'object') return false
|
||||
const record = value as Record<string, unknown>
|
||||
return typeof record.language === 'string' && typeof record.value === 'string'
|
||||
}
|
||||
|
||||
/** Create the stable structured error used for malformed server result payloads. */
|
||||
function malformedResponse(message: string): LspError {
|
||||
return new LspError(message, 'LSP_MALFORMED_RESPONSE')
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user