fix(mcp-client): await failed generation shutdown
The MCP SDK starts a fire-and-forget close when initialization fails. Its stdio transport clears its process field before that close finishes, so our second Client.close() could return immediately and the reconnect timer could launch a replacement while the original child was still alive. Track the transport onclose signal for every client generation and gate failed-attempt backoff on both Client.close() settlement and that signal. Use the same barrier during plugin disposal. If the SDK's bounded stdio termination window expires without onclose, fail closed and report incomplete shutdown instead of risking overlapping server processes. Regression coverage models the SDK's early-returning second close, delayed and missing close signals, pending-connect disposal, close rejection, and the terminal timeout path. The reconnect Agent Note and Chinese counterpart now record the quiescence contract.
This commit is contained in:
Tianyi Cui
@@ -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 .agents/notes/implemented/feature/2026-08-06-mcp-client-auto-reconnect.md
|
||||
2026-08-06-mcp-client-auto-reconnect.md: b187f5de71a10ca3121d817f383a98669246c81f
|
||||
2026-08-06-mcp-client-auto-reconnect.zh.md: a4ec6897c34f2142ceb95abecd13d189c4a95624
|
||||
2026-08-06-mcp-client-auto-reconnect.md: e371fa13dc7bc5330f658f4d6e864969baf93cf0
|
||||
2026-08-06-mcp-client-auto-reconnect.zh.md: eb6e67dabd7c6115b9ecb507029ffa7d45338466
|
||||
|
||||
@@ -14,7 +14,7 @@ The [MCP client](2026-07-07-mcp-client-plugin.md) connected once at plugin load.
|
||||
|
||||
**Trigger.** The supervisor arms `client.onclose` per generation. The SDK fires it when the stdio child exits, so a crash is observed without polling. `StreamableHTTPClientTransport` fires `onclose` only for deliberate closes — it owns its internal SSE-stream recovery and surfaces request failures per call — so HTTP servers are effectively outside supervisor restarts; the package README records that limitation.
|
||||
|
||||
**Generations without interleaving.** Each attempt builds a fresh transport and `Client` (the SDK binds a Protocol to one transport for life). One per-supervisor queue serializes every `syncTools` call — initial syncs and `list_changed` re-syncs across all generations — and an `isCurrent` fence makes stale generations inert, so no two syncs can interleave the dispose-previous/register-next swap (which would double-dispose one generation and leak another). The queue also closes a pre-existing race where two rapid `list_changed` notifications re-synced concurrently. Failure signals are idempotent per generation: a connect rejection racing its own transport close schedules exactly one retry.
|
||||
**Generations without interleaving.** Each attempt builds a fresh transport and `Client` (the SDK binds a Protocol to one transport for life). One per-supervisor queue serializes every `syncTools` call — initial syncs and `list_changed` re-syncs across all generations — and an `isCurrent` fence makes stale generations inert, so no two syncs can interleave the dispose-previous/register-next swap (which would double-dispose one generation and leak another). The queue also closes a pre-existing race where two rapid `list_changed` notifications re-synced concurrently. Failure signals are idempotent per generation: a connect rejection racing its own transport close schedules exactly one retry. A failed attempt cannot enter backoff until both `Client.close()` settles and the transport reports `onclose`, which for stdio proves the child exited; a missing close signal stops reconnection after the SDK's bounded termination window instead of allowing two server processes to overlap. Disposal uses the same bounded close-signal barrier and reports an incomplete shutdown without ever restarting.
|
||||
|
||||
**Bounded backoff with an outage budget.** Delays double from `initialDelayMs` up to `maxDelayMs`. One outage shares `maxAttempts` consecutive failed attempts; exhaustion unregisters the server's tools, logs at error level, and stops until disposal or reload. A connection that survives past the stability window — `maxDelayMs`, derived rather than a fifth tunable, as the longest configured backoff spacing — resets the budget, so an occasionally-crashing server recovers indefinitely while a crash loop whose connects briefly succeed cannot launder its budget into a restart storm.
|
||||
|
||||
@@ -38,7 +38,7 @@ The [MCP client](2026-07-07-mcp-client-plugin.md) connected once at plugin load.
|
||||
|
||||
## Testing
|
||||
|
||||
Unit (`tests/reconnect.spec.ts`, mocked SDK): recovery swaps generations without duplication or leaks and serves post-recovery calls, the failure cap unregisters tools and stops, dispose cancels a pending backoff and quiesces an in-flight sync, a close after dispose schedules nothing, disabled mode keeps the v1 behavior, the stability window resets the budget while a crash loop exhausts it, double failure signals schedule one retry, stale generations and handlers are inert, and `resolveReconnectPolicy` rejects each invalid bound. E2E (`tests/mcp-client.e2e.ts`, keyless): the fixture server gained a `crash` tool that replies then exits; real-process tests prove a stdio crash recovers end to end and that unloading the plugin mid-outage stops reconnection promptly. Snapshot: deliberately none, per the original note's rationale — reconnection adds no new presentation shape, and a snapshot composition spawning a crashing server would make replays timing-dependent.
|
||||
Unit (`tests/reconnect.spec.ts`, mocked SDK): recovery swaps generations without duplication or leaks and serves post-recovery calls, failed initialization waits for the old generation's close signal and fails closed when that signal never arrives, disposal waits for the same signal with a bounded incomplete-shutdown path, the failure cap unregisters tools and stops, dispose cancels a pending backoff and quiesces an in-flight sync, a close after dispose schedules nothing, disabled mode keeps the v1 behavior, the stability window resets the budget while a crash loop exhausts it, double failure signals schedule one retry, stale generations and handlers are inert, and `resolveReconnectPolicy` rejects each invalid bound. E2E (`tests/mcp-client.e2e.ts`, keyless): the fixture server gained a `crash` tool that replies then exits; real-process tests prove a stdio crash recovers end to end and that unloading the plugin mid-outage stops reconnection promptly. Snapshot: deliberately none, per the original note's rationale — reconnection adds no new presentation shape, and a snapshot composition spawning a crashing server would make replays timing-dependent.
|
||||
|
||||
## Consequences
|
||||
|
||||
|
||||
@@ -14,7 +14,7 @@ Status: implemented
|
||||
|
||||
**触发条件。** 监督器在每一代上挂载 `client.onclose`。SDK 在 stdio 子进程退出时触发该回调，因此崩溃无需轮询即可感知。`StreamableHTTPClientTransport` 仅在主动关闭时触发 `onclose`——它内部拥有自己的 SSE（Server-Sent Events）流恢复机制，并将请求失败以逐调用方式暴露——因此 HTTP 服务器实际上不在监督器的重启范围内；包 README 记录了该限制。
|
||||
|
||||
**代隔离，无交错。** 每次尝试构建一个全新的 transport 和 `Client`（SDK 将一个 Protocol 绑定到一个 transport 上终身使用）。每个监督器内部有一个队列将所有 `syncTools` 调用串行化——跨所有代的初始同步和 `list_changed` 再同步——`isCurrent` 栅栏使过时的代变为惰性，从而确保不会有两次同步交错执行 dispose 上一代/注册下一代的切换（否则会对同一代执行两次 dispose 并泄漏另一代）。该队列还消除了一个先前存在的竞态：两次快速的 `list_changed` 通知同时触发重新同步。失败信号按代幂等：一次连接拒绝与其自身 transport 关闭竞态时，仅调度恰好一次重试。
|
||||
**代隔离，无交错。** 每次尝试构建一个全新的 transport 和 `Client`（SDK 将一个 Protocol 绑定到一个 transport 上终身使用）。每个监督器内部有一个队列将所有 `syncTools` 调用串行化——跨所有代的初始同步和 `list_changed` 再同步——`isCurrent` 栅栏使过时的代变为惰性，从而确保不会有两次同步交错执行 dispose 上一代/注册下一代的切换（否则会对同一代执行两次 dispose 并泄漏另一代）。该队列还消除了一个先前存在的竞态：两次快速的 `list_changed` 通知同时触发重新同步。失败信号按代幂等：一次连接拒绝与其自身 transport 关闭竞态时，仅调度恰好一次重试。失败尝试只有在 `Client.close()` 结算且 transport 报告 `onclose` 后才能进入退避；对 stdio 而言，`onclose` 证明子进程已退出；若关闭信号始终未到，则在 SDK 的有界终止窗口结束后停止重连，而不是允许两个服务器进程重叠运行。dispose 使用同一个有界关闭信号屏障；若关停未完成则予以报告，且绝不重启。
|
||||
|
||||
**有界退避与故障预算。** 延迟从 `initialDelayMs` 起逐次翻倍，上限为 `maxDelayMs`。一次故障期间共享 `maxAttempts` 次连续失败尝试的预算；耗尽后注销该服务器的工具、以 error 级别记录日志并停止，直到 dispose 或重新加载。连接在存活超过稳定窗口——即 `maxDelayMs`，作为最长退避间隔从配置推导得出而非作为第五个独立调参项——之后重置预算；因此偶尔崩溃的服务器可无限恢复，而连接短暂成功后立即再次崩溃的循环无法将其预算洗白为重启风暴。
|
||||
|
||||
@@ -38,7 +38,7 @@ Status: implemented
|
||||
|
||||
## 测试
|
||||
|
||||
单元测试（`tests/reconnect.spec.ts`，mock SDK）：恢复在不产生重复或泄漏的前提下切换代并服务恢复后的调用、失败上限注销工具并停止、dispose 取消待执行的退避并使进行中的同步完全停稳、dispose 后的关闭不调度任何操作、禁用模式保持 v1 行为、稳定窗口重置预算而崩溃循环耗尽预算、双重失败信号仅调度一次重试、过时的代和处理器为惰性、`resolveReconnectPolicy` 拒绝每个无效边界值。E2E（`tests/mcp-client.e2e.ts`，无需密钥）：fixture 服务器新增了一个 `crash` 工具（先回复再退出）；真实进程测试证明 stdio 崩溃端到端恢复，以及在故障期间卸载插件能立即停止重连。快照：刻意不做，原因与原 Agent Note 相同——重连不引入新的展示形态，而在快照组合中 spawn 崩溃服务器会使回放依赖时序。
|
||||
单元测试（`tests/reconnect.spec.ts`，mock SDK）：恢复在不产生重复或泄漏的前提下切换代并服务恢复后的调用、初始化失败会等待旧代的关闭信号，若该信号始终未到则停止重连、dispose 同样等待同一关闭信号，并在有界等待到期时报告关停未完成、失败上限注销工具并停止、dispose 取消待执行的退避并使进行中的同步完全停稳、dispose 后的关闭不调度任何操作、禁用模式保持 v1 行为、稳定窗口重置预算而崩溃循环耗尽预算、双重失败信号仅调度一次重试、过时的代和处理器为惰性、`resolveReconnectPolicy` 拒绝每个无效边界值。E2E（`tests/mcp-client.e2e.ts`，无需密钥）：fixture 服务器新增了一个 `crash` 工具（先回复再退出）；真实进程测试证明 stdio 崩溃端到端恢复，以及在故障期间卸载插件能立即停止重连。快照：刻意不做，原因与原 Agent Note 相同——重连不引入新的展示形态，而在快照组合中 spawn 崩溃服务器会使回放依赖时序。
|
||||
|
||||
## 后果
|
||||
|
||||
|
||||
@@ -44,6 +44,11 @@ export const RECONNECT_DEFAULTS: Required<ReconnectConfig> = Object.freeze({
|
||||
maxAttempts: 10,
|
||||
})
|
||||
|
||||
// The SDK's stdio transport owns two two-second termination grace periods.
|
||||
// Keep one additional second for the process-close event that proves the old
|
||||
// generation is gone; timing out fails closed instead of overlapping children.
|
||||
const GENERATION_CLOSE_TIMEOUT_MS = 5_000
|
||||
|
||||
/** Fully resolved reconnect policy captured at plugin load. */
|
||||
export type ResolvedReconnectPolicy = Readonly<Required<ReconnectConfig>>
|
||||
|
||||
@@ -133,6 +138,8 @@ export function startConnection(ctx: Context, config: Config, policy: ResolvedRe
|
||||
let disposed = false
|
||||
/** Current generation: the connecting or connected client; undefined during backoff waits and after final failure. */
|
||||
let client: Client | undefined
|
||||
/** Close signal paired with {@link client}; captured by dispose before current ownership is cleared. */
|
||||
let clientClosed: Promise<void> | undefined
|
||||
/** Live tool registrations owned by this server; only {@link enqueueSync} and dispose swap it. */
|
||||
let disposers: ToolDisposers = new Map()
|
||||
let reconnectTimer: NodeJS.Timeout | undefined
|
||||
@@ -169,9 +176,22 @@ export function startConnection(ctx: Context, config: Config, policy: ResolvedRe
|
||||
function generationDown(generation: Client): void {
|
||||
if (!isCurrent(generation)) return
|
||||
client = undefined
|
||||
clientClosed = undefined
|
||||
scheduleReconnect()
|
||||
}
|
||||
|
||||
/** Wait for the transport-owned close signal without letting a broken transport wedge teardown forever. */
|
||||
function waitForClose(closed: Promise<void>): Promise<boolean> {
|
||||
return new Promise((resolve) => {
|
||||
const timeout = setTimeout(() => { resolve(false) }, GENERATION_CLOSE_TIMEOUT_MS)
|
||||
timeout.unref()
|
||||
void closed.then(() => {
|
||||
clearTimeout(timeout)
|
||||
resolve(true)
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
function scheduleReconnect(): void {
|
||||
if (!policy.enabled) {
|
||||
const detail = connectedAt !== undefined
|
||||
@@ -216,8 +236,19 @@ export function startConnection(ctx: Context, config: Config, policy: ResolvedRe
|
||||
{ name: 'dsh-mcp-client', version: '0.0.1' },
|
||||
{ capabilities: {} },
|
||||
)
|
||||
const closed: PromiseWithResolvers<void> = Promise.withResolvers()
|
||||
let attemptSettled = false
|
||||
let closeObserved = false
|
||||
const hasClosed = (): boolean => closeObserved
|
||||
client = generation
|
||||
generation.onclose = () => { generationDown(generation) }
|
||||
clientClosed = closed.promise
|
||||
generation.onclose = () => {
|
||||
closeObserved = true
|
||||
closed.resolve()
|
||||
// A failed connect owns its close barrier in the catch path below. An
|
||||
// established generation can transition down directly from this signal.
|
||||
if (attemptSettled) generationDown(generation)
|
||||
}
|
||||
// Registered before connect so a list change during the initial sync is
|
||||
// queued behind it rather than dropped.
|
||||
generation.setNotificationHandler(
|
||||
@@ -236,12 +267,32 @@ export function startConnection(ctx: Context, config: Config, policy: ResolvedRe
|
||||
)
|
||||
try {
|
||||
await generation.connect(createTransport(config))
|
||||
if (hasClosed()) {
|
||||
attemptSettled = true
|
||||
generationDown(generation)
|
||||
return
|
||||
}
|
||||
await enqueueSync(generation)
|
||||
} catch (error) {
|
||||
if (firstAttemptError === undefined) firstAttemptError = error
|
||||
// When the transport closed first, onclose already logged and scheduled.
|
||||
// Disposal clears current ownership before it closes the generation, so
|
||||
// only a live supervisor reports an attempt failure.
|
||||
if (isCurrent(generation)) ctx.logger.warn(`${label}: connection attempt failed: ${String(error)}`)
|
||||
try { await generation.close() } catch { /* transport already gone */ }
|
||||
const quiesced = hasClosed() || await waitForClose(closed.promise)
|
||||
attemptSettled = true
|
||||
if (!isCurrent(generation)) return
|
||||
if (!quiesced) {
|
||||
client = undefined
|
||||
clientClosed = undefined
|
||||
ctx.logger.error(`${label}: failed generation did not close within ${GENERATION_CLOSE_TIMEOUT_MS}ms — reconnect stopped to avoid overlapping server processes; reload the plugin or restart the Host to retry`)
|
||||
return
|
||||
}
|
||||
generationDown(generation)
|
||||
return
|
||||
}
|
||||
attemptSettled = true
|
||||
if (hasClosed()) {
|
||||
generationDown(generation)
|
||||
return
|
||||
}
|
||||
@@ -277,9 +328,14 @@ export function startConnection(ctx: Context, config: Config, policy: ResolvedRe
|
||||
reconnectTimer = undefined
|
||||
}
|
||||
const current = client
|
||||
const currentClosed = clientClosed
|
||||
client = undefined
|
||||
clientClosed = undefined
|
||||
if (current !== undefined) {
|
||||
try { await current.close() } catch { /* transport already gone */ }
|
||||
if (currentClosed !== undefined && !await waitForClose(currentClosed)) {
|
||||
ctx.logger.error(`${label}: generation did not close within ${GENERATION_CLOSE_TIMEOUT_MS}ms during disposal — server shutdown may be incomplete`)
|
||||
}
|
||||
}
|
||||
// Quiesce, don't just request it: the in-flight attempt enqueues its
|
||||
// sync before settling, so awaiting both leaves `disposers` final.
|
||||
|
||||
@@ -159,7 +159,10 @@ describe('apply (plugin lifecycle)', () => {
|
||||
beforeEach(async () => {
|
||||
vi.clearAllMocks()
|
||||
mockConnect.mockResolvedValue(undefined)
|
||||
mockClose.mockResolvedValue(undefined)
|
||||
mockClose.mockImplementation(function (this: { onclose?: () => void }) {
|
||||
this.onclose?.()
|
||||
return Promise.resolve()
|
||||
})
|
||||
mockListTools.mockResolvedValue({
|
||||
tools: [{ name: 'remote', description: 'A remote tool', inputSchema: { type: 'object' } }],
|
||||
nextCursor: undefined,
|
||||
@@ -338,7 +341,10 @@ describe('apply (plugin lifecycle)', () => {
|
||||
})
|
||||
|
||||
it('effect disposer handles client.close failure gracefully', async () => {
|
||||
mockClose.mockRejectedValue(new Error('already closed'))
|
||||
mockClose.mockImplementation(function (this: { onclose?: () => void }) {
|
||||
this.onclose?.()
|
||||
return Promise.reject(new Error('already closed'))
|
||||
})
|
||||
|
||||
await apply(ctx, stdioConfig)
|
||||
|
||||
|
||||
@@ -59,7 +59,7 @@ vi.mock('@modelcontextprotocol/sdk/client/streamableHttp.js', () => ({
|
||||
// vi.mock is hoisted above static imports, so the modules under test see the
|
||||
// mocked SDK even through a static import.
|
||||
import { apply } from '@deepseek-ai/dsh-mcp-client/src/index.ts'
|
||||
import { RECONNECT_DEFAULTS, resolveReconnectPolicy } from '@deepseek-ai/dsh-mcp-client/src/connection.ts'
|
||||
import { RECONNECT_DEFAULTS, resolveReconnectPolicy, startConnection } from '@deepseek-ai/dsh-mcp-client/src/connection.ts'
|
||||
|
||||
// ---- Helpers ----
|
||||
|
||||
@@ -128,7 +128,10 @@ describe('reconnect supervisor', () => {
|
||||
vi.clearAllMocks()
|
||||
instances.length = 0
|
||||
mockConnect.mockResolvedValue(undefined)
|
||||
mockClose.mockResolvedValue(undefined)
|
||||
mockClose.mockImplementation(function (this: { onclose?: () => void }) {
|
||||
this.onclose?.()
|
||||
return Promise.resolve()
|
||||
})
|
||||
mockListTools.mockResolvedValue(listing('remote'))
|
||||
mockCallTool.mockResolvedValue({ content: [{ type: 'text', text: 'ok' }] })
|
||||
ctx = await mountRegistry()
|
||||
@@ -174,7 +177,10 @@ describe('reconnect supervisor', () => {
|
||||
|
||||
mockConnect.mockRejectedValue(new Error('server gone'))
|
||||
// A failing close on the failed attempt's cleanup must not break the loop.
|
||||
mockClose.mockRejectedValue(new Error('already closed'))
|
||||
mockClose.mockImplementation(function (this: { onclose?: () => void }) {
|
||||
this.onclose?.()
|
||||
return Promise.reject(new Error('already closed'))
|
||||
})
|
||||
instances[0]!.onclose?.()
|
||||
|
||||
await vi.waitFor(() => {
|
||||
@@ -189,6 +195,78 @@ describe('reconnect supervisor', () => {
|
||||
expect(mockConnect).toHaveBeenCalledTimes(3)
|
||||
})
|
||||
|
||||
it('does not start a replacement until a failed generation reports that it closed', async () => {
|
||||
mockConnect.mockRejectedValueOnce(new Error('initialize failed'))
|
||||
// Model the SDK's fire-and-forget close after initialize fails: the
|
||||
// harness's second close call returns, but the child has not exited yet.
|
||||
mockClose.mockResolvedValue(undefined)
|
||||
|
||||
const applying = apply(ctx, stdioConfig({ initialDelayMs: 2, maxDelayMs: 8, maxAttempts: 2 }))
|
||||
await vi.waitFor(() => { expect(mockClose).toHaveBeenCalled() })
|
||||
await sleep(30)
|
||||
expect(instances).toHaveLength(1)
|
||||
|
||||
instances[0]!.onclose?.()
|
||||
await applying
|
||||
await vi.waitFor(() => { expect(instances).toHaveLength(2) })
|
||||
})
|
||||
|
||||
it('stops reconnecting when a failed generation never reports that it closed', async () => {
|
||||
vi.useFakeTimers()
|
||||
try {
|
||||
const { errors } = captureLogs(ctx)
|
||||
mockConnect.mockRejectedValue(new Error('initialize failed'))
|
||||
mockClose.mockResolvedValue(undefined)
|
||||
|
||||
const applying = apply(ctx, stdioConfig({ initialDelayMs: 2, maxDelayMs: 8, maxAttempts: 2 }))
|
||||
await vi.advanceTimersByTimeAsync(5_000)
|
||||
await applying
|
||||
|
||||
expect(instances).toHaveLength(1)
|
||||
expect(errors.some(line => line.includes('reconnect stopped to avoid overlapping server processes'))).toBe(true)
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('suppresses retry reporting when disposal owns a pending connect rejection', async () => {
|
||||
const { warns } = captureLogs(ctx)
|
||||
const gate: PromiseWithResolvers<void> = Promise.withResolvers()
|
||||
mockConnect.mockImplementation(() => gate.promise)
|
||||
const handle = startConnection(ctx, stdioConfig(), resolveReconnectPolicy(undefined, 'reconnect'))
|
||||
await vi.waitFor(() => { expect(instances).toHaveLength(1) })
|
||||
|
||||
const disposing = handle.dispose()
|
||||
gate.reject(new Error('disposed connect'))
|
||||
await disposing
|
||||
await handle.ready
|
||||
|
||||
expect(warns.some(line => line.includes('connection attempt failed'))).toBe(false)
|
||||
expect(instances).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('bounds disposal while a resolving generation never reports that it closed', async () => {
|
||||
vi.useFakeTimers()
|
||||
try {
|
||||
const { errors } = captureLogs(ctx)
|
||||
const gate: PromiseWithResolvers<void> = Promise.withResolvers()
|
||||
mockConnect.mockImplementation(() => gate.promise)
|
||||
mockClose.mockResolvedValue(undefined)
|
||||
const handle = startConnection(ctx, stdioConfig(), resolveReconnectPolicy(undefined, 'reconnect'))
|
||||
await vi.advanceTimersByTimeAsync(0)
|
||||
|
||||
const disposing = handle.dispose()
|
||||
await vi.advanceTimersByTimeAsync(5_000)
|
||||
gate.resolve()
|
||||
await disposing
|
||||
|
||||
expect(mockListTools).not.toHaveBeenCalled()
|
||||
expect(errors.some(line => line.includes('server shutdown may be incomplete'))).toBe(true)
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('dispose during the backoff wait cancels the pending reconnect', async () => {
|
||||
await apply(ctx, stdioConfig({ initialDelayMs: 60_000, maxDelayMs: 60_000, maxAttempts: 5 }))
|
||||
await vi.waitFor(() => { expect(ctx.tools.get('mcp__srv__remote')).toBeDefined() })
|
||||
|
||||
Reference in New Issue
Block a user