fix(runtime): close teardown gaps and simplify framing

This commit is contained in:
Tianyi Cui
2026-07-29 13:51:55 +08:00
parent 045e8462ee
commit 4fecc54998
20 changed files with 164 additions and 36 deletions

View File

@@ -2,5 +2,5 @@
# side as of the last confirmed-consistent state. Both languages carry equal authority;
# after editing either side, bring the other along and re-record with:
# pnpm run verify-translation-pairing --write packages/lsp/lsp-local/README.md
README.md: 299a5282bbca0a6d517af9342c4101cbdae33d4c
README.zh.md: edf706fdb424756aa43b303b018b11e3764c2559
README.md: c96c4febc9d047b41789f2b7a73e3eaf4d35012b
README.zh.md: 5bc2c3c8bb7a89afb673797f5a3b25bb9fc06748

View File

@@ -11,7 +11,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 target)`. A live server error is not replayed; if the selected pooled transport fails before or during a read-only query, the provider awaits its disposal and retries that query once on a fresh process.
- Uses a compatibility-first **transient-open** sequence per query: resolve and boundedly read the source through `ctx.fs`, `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.
- 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. Provider disposal aborts filesystem and protocol work, awaits workspace lookups that have not entered a queue, then drains every queue and server.
- After protocol shutdown fails, terminates the server's descendant tree through the subprocess seam (POSIX process-group signaling; Windows `taskkill /T /F`). Tree-kill delivery is contained like every group signal — it races server exit — and quiescence is confirmed by the handle's tree-liveness wait rather than by the kill's own outcome.
- Resolves the server executable, cwd, process, and protocol streams through `ctx.subprocess`; `initialize.processId` is `null` because another machine or PID namespace must not monitor the harness process.
- Uses `ctx.fs` canonical containment, file URIs, and stable bounded reads, but emits no `fs/observed`: only the LSP result is model-visible, so a query does not satisfy read-before-write policy.

View File

@@ -11,7 +11,7 @@ Namespace 插件(`name`/`inject`/`Config`/`apply`,无默认导出)
- 在注册前解析每项服务器局部设置;无效映射或注册冲突会回滚较早配置项,因此加载失败不会留下提供方路由。
- 每个 `(server id, canonical workspace target)` 惰性 single-flight 一个服务器进程。存活服务器错误不会回放;如果选中的池化传输在只读查询之前或期间失败,提供方会等待其释放,并在新进程上重试该查询一次。
- 每次查询都使用兼容性优先的**临时打开** 序列:通过 `ctx.fs` 解析源文件并进行有界读取、`textDocument/didOpen`(版本 1、完整文本)、所请求操作,然后执行 `textDocument/didClose`,该操作位于 `finally` 中。写入 `didOpen` 失败或取消时,会先终止实例再允许池复用。文档在每次调用后关闭,因此第一版不需要 `didChange`、内容 cache 或文档 LRU。
- 通过一条逐 Workspace、可中止的队列,串行执行每个源读取/打开/查询/关闭生命周期,因此排队调用只会在轮到自身时读取当前源;不同 Workspace 并行运行。
- 通过一条逐 Workspace、可中止的队列,串行执行每个源读取/打开/查询/关闭生命周期,因此排队调用只会在轮到自身时读取当前源;不同 Workspace 并行运行。提供方资源释放会中止文件系统与协议工作,等待尚未进入队列的 Workspace 查找结算,再排空所有队列并等待所有服务器结算。
- 协议 shutdown 失败后,经由进程管理器 seam 终止服务器后代树(POSIX 进程组信号;Windows `taskkill /T /F`)。树终止的投递结果与所有进程组信号一样被就地吸收,不向外抛出(投递与服务器退出存在竞态);服务器是否完全停稳,由句柄的进程树存活等待确认,而非由这次终止自身的结果确认。
- 通过 `ctx.subprocess` 解析服务器可执行文件、cwd、进程与协议流;`initialize.processId` 为 `null`,因为另一台机器或 PID 命名空间不得监控 harness 进程。
- 使用 `ctx.fs` 提供的规范 containment、文件 URI 与稳定有界读取,但不发出 `fs/observed`:只有 LSP 结果对模型可见,因此查询不满足先读后写策略。

View File

@@ -208,6 +208,9 @@ class LocalLspProvider implements LspProvider {
private readonly instances = new Map<WorkspaceKey, LspInstance>()
/** One complete source-read→open→query→close serialization tail per canonical workspace. */
private readonly queues = new Map<WorkspaceKey, Promise<void>>()
/** Workspace canonicalizations that have not entered a provider-owned queue yet. */
private readonly workspaceLookups = new Set<Promise<void>>()
private readonly lifetime = new AbortController()
private disposed = false
constructor(
@@ -234,32 +237,48 @@ class LocalLspProvider implements LspProvider {
if (signal?.aborted) throw abortError(signal)
}
/** Fuse caller cancellation with provider disposal for every filesystem and protocol await. */
private querySignal(signal?: AbortSignal): AbortSignal {
return signal === undefined
? this.lifetime.signal
: AbortSignal.any([signal, this.lifetime.signal])
}
async query(request: LspProviderQuery, signal?: AbortSignal): Promise<LspQueryResult> {
// Honor an already-aborted signal before provider I/O so a canceled request never starts a server.
this.assertActive(signal)
const workspace = await canonicalizeWorkspace(this.fs, request.workspaceRoot, signal)
this.assertActive(signal)
const querySignal = this.querySignal(signal)
const workspaceResult = canonicalizeWorkspace(this.fs, request.workspaceRoot, querySignal)
const workspaceLookup = workspaceResult.then(() => undefined, () => undefined)
this.workspaceLookups.add(workspaceLookup)
let workspace: HostWorkspace
try {
workspace = await workspaceResult
} finally {
this.workspaceLookups.delete(workspaceLookup)
}
this.assertActive(querySignal)
const workspaceKey = workspace.target.targetKey
return this.enqueue(workspaceKey, signal, async () => {
this.assertActive(signal)
return this.enqueue(workspaceKey, querySignal, async () => {
this.assertActive(querySignal)
// 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(this.fs, request.filePath, workspace, this.config.maxDocumentBytes, signal)
const source = await readHostSource(this.fs, request.filePath, workspace, this.config.maxDocumentBytes, querySignal)
// 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)
this.assertActive(querySignal)
let instance = this.instanceFor(workspaceKey, workspace)
try {
return await instance.query(request, source, signal)
return await instance.query(request, source, querySignal)
} catch (error) {
// A selected child can have died while idle or fail during the next write. Queries are
// read-only, so replace that transport once and retry transparently.
if (!instance.isTransportFailure(error)) throw error
await instance.dispose()
this.evictIfCurrent(workspaceKey, instance)
this.assertActive(signal)
this.assertActive(querySignal)
instance = this.instanceFor(workspaceKey, workspace)
return await instance.query(request, source, signal)
return await instance.query(request, source, querySignal)
} finally {
// Reach quiescence before dropping a dead slot; a replacement must survive this ownership check.
if (instance.dead) {
@@ -320,13 +339,17 @@ class LocalLspProvider implements LspProvider {
/** Dispose every live instance and block further queries. */
async disposeAll(): Promise<void> {
this.disposed = true
this.lifetime.abort(new LspError('lsp-local provider is disposed', 'LSP_DISPOSED'))
const live = [...this.instances.values()]
const draining = [...this.queues.values()]
const resolving = [...this.workspaceLookups]
this.instances.clear()
await Promise.all([
...live.map(instance => instance.dispose()),
...draining,
...resolving,
])
this.queues.clear()
this.workspaceLookups.clear()
}
}

View File

@@ -301,6 +301,52 @@ describe('lsp-local end to end over a fake server', () => {
await ctx.fiber.dispose()
})
it('aborts and awaits a workspace lookup when the provider is disposed', async () => {
const ctx = await mount({ LSP_FAKE_DEF: 'null' })
const fs = ctx.fs
const resolve = fs.resolve.bind(fs)
const started = Promise.withResolvers<AbortSignal>()
const release = Promise.withResolvers<undefined>()
vi.spyOn(fs, 'resolve').mockImplementation(async (path, options) => {
if (path !== ws) return await resolve(path, options)
const signal = options?.signal
if (signal === undefined) throw new Error('workspace lookup missing provider lifetime signal')
started.resolve(signal)
return await rejectWhenAborted(signal, release.promise)
})
const pending = ctx.lsp.query(query('goToDefinition'))
const signal = await started.promise
let disposed = false
const disposing = ctx.fiber.dispose().then(() => { disposed = true })
await new Promise<void>(resolve => setImmediate(resolve))
expect(signal.aborted).toBe(true)
expect(disposed).toBe(false)
release.resolve(undefined)
await expect(pending).rejects.toThrow('provider is disposed')
await expect(disposing).resolves.toBeUndefined()
})
it('aborts a queued source read when the provider is disposed', async () => {
const ctx = await mount({ LSP_FAKE_DEF: 'null' })
const fs = ctx.fs
const started = Promise.withResolvers<AbortSignal>()
vi.spyOn(fs, 'readTextBounded').mockImplementation(async (_target, _maxBytes, signal) => {
if (signal === undefined) throw new Error('source read missing provider lifetime signal')
started.resolve(signal)
return await rejectWhenAborted(signal)
})
const pending = ctx.lsp.query(query('goToDefinition'))
const signal = await started.promise
const disposing = ctx.fiber.dispose()
await expect(pending).rejects.toThrow('provider is disposed')
await expect(disposing).resolves.toBeUndefined()
expect(signal.aborted).toBe(true)
})
it('runs distinct workspaces in parallel instances', async () => {
const ws2 = join(root, 'ws2')
await mkdir(ws2)
@@ -358,3 +404,16 @@ async function waitFor(condition: () => Promise<boolean>, timeoutMs = 3000): Pro
await new Promise<void>(resolve => setTimeout(resolve, 10))
}
}
/** Hold one fake provider operation until cancellation, optionally behind a cleanup gate. */
function rejectWhenAborted<T>(signal: AbortSignal, release: Promise<unknown> = Promise.resolve()): Promise<T> {
return new Promise((_resolve, reject) => {
const onAbort = (): void => {
void release.then(() => {
reject(signal.reason instanceof Error ? signal.reason : new Error(String(signal.reason)))
})
}
signal.addEventListener('abort', onAbort, { once: true })
if (signal.aborted) onAbort()
})
}