History

Sweep every expired MCP connection, not just the one being asked for (#1577)

* Sweep every expired MCP connection, not just the one being asked for The idle window was consulted only against idle.get(key), so an identity that was never dialled again was never examined again: its session stayed open and authenticated for the pool's lifetime, holding the credential it was dialled with. The advertised five-minute bound applied only to connections that happened to be reused. acquire now closes every entry past the window. Still lazy in the sense the pool intends -- activity drives it, no timer, no background fiber -- and the map holds at most one entry per identity, so the scan is trivial. Reuse is unchanged. * docs(mcp): state the changeset's idle bound as acquire-driven, not unconditional The headline is what lands in the published CHANGELOG, and it promised idle sessions age out with no qualifier. The sweep only runs inside acquire: a pool that sees no further activity holds a parked session until close(). Condition the headline on the pool's next acquire so the CHANGELOG does not promise a bound the code does not provide. * Bound and parallelise the MCP pool idle sweep so a hung close cannot stall an acquire ---------

Sergio Fernández committed 2 months ago

Commit 4879d471 parent 9cab2613 files · +205 −3Browse files

Unified · Split

3 files
  1. .changeset/mcp-pool-idle-sweep.md +13 −0
  2. packages/plugins/mcp/src/sdk/connection-pool-sweep.test.ts +144 −0
  3. packages/plugins/mcp/src/sdk/connection-pool.ts +48 −3
.changeset/mcp-pool-idle-sweep.md+13−0
@@ -0,0 +1,13 @@
1---
2"executor": patch
3---
4
5**Idle MCP connections age out on the pool's next acquire, even when their identity is never dialled again**
6
7The pool's five-minute idle window was only consulted against the entry being requested, so an identity that was never asked for a second time was never examined a second time. Its session stayed open and authenticated for as long as the pool lived, holding the bearer token or API key it was dialled with. The advertised bound applied only to connections that happened to be reused.
8
9`acquire` now sweeps every entry past the window, closing each one, rather than just the entry matching the key. This stays lazy in the sense the pool intends — activity drives it, there is no timer and no background fiber — and the map holds at most one entry per identity, so the scan is trivial.
10
11Because the sweep is paid for by whichever invocation acquires next, it cannot be allowed to stall that caller. The expired entries leave the pool synchronously, before any close is awaited, and the closes then run concurrently with each one bounded by a two-second timeout — so a server that accepts a close and goes quiet is abandoned rather than waited on, and cannot hold up an unrelated request or the connections queued behind it.
12
13Reuse is unchanged: an entry still inside the window is left alone, and a second call for the same identity still gets the parked session rather than a fresh dial.
packages/plugins/mcp/src/sdk/connection-pool-sweep.test.ts+144−0
@@ -0,0 +1,144 @@
1// ---------------------------------------------------------------------------
2// Idle eviction must reach every parked connection, not only the one being
3// asked for.
4//
5// The TTL used to be consulted against `idle.get(key)` alone, so an identity
6// that was never dialled again was never examined again — its session stayed
7// open and authenticated indefinitely, holding the bearer it was dialled with.
8// The advertised five-minute bound only applied to connections that happened to
9// be reused.
10//
11// Driven with a fake connector rather than a real MCP server, because the thing
12// under test is exactly WHEN `close()` is called, and a fake makes that directly
13// observable instead of inferred from session counts.
14// ---------------------------------------------------------------------------
15
16import { describe, expect, it } from "@effect/vitest";
17import { Duration, Effect, Fiber } from "effect";
18import { TestClock } from "effect/testing";
19// oxlint-disable-next-line executor/no-vitest-import -- boundary: system-time control comes from vitest itself
20import { afterEach, vi } from "vitest";
21
22import type { McpConnection, McpConnector } from "./connection";
23import { createMcpConnectionPool } from "./connection-pool";
24
25const IDLE_TTL_MS = 5 * 60 * 1_000;
26
27afterEach(() => {
28 vi.useRealTimers();
29});
30
31/** A connector whose connection records the moment it is closed. */
32const fakeConnector = (state: { closed: boolean }): McpConnector =>
33 Effect.sync(
34 () =>
35 ({
36 client: {} as McpConnection["client"],
37 close: async () => {
38 state.closed = true;
39 },
40 }) satisfies McpConnection,
41 );
42
43/** A connection whose `close()` is accepted and then never answered — the
44 * server that goes quiet mid-teardown. */
45const hangingConnector = (): McpConnector =>
46 Effect.sync(
47 () =>
48 ({
49 client: {} as McpConnection["client"],
50 close: () => new Promise<void>(() => {}),
51 }) satisfies McpConnection,
52 );
53
54describe("MCP connection pool idle sweep", () => {
55 it.effect("closes an expired connection parked under a DIFFERENT key", () =>
56 Effect.gen(function* () {
57 vi.useFakeTimers();
58 const pool = createMcpConnectionPool();
59 const stale = { closed: false };
60 const other = { closed: false };
61
62 // Park a connection under "stale" and never ask for that key again.
63 yield* pool.withConnection("stale", fakeConnector(stale), () => Effect.void);
64 expect(stale.closed).toBe(false);
65
66 vi.advanceTimersByTime(IDLE_TTL_MS + 1_000);
67
68 // Activity on an UNRELATED key is what must now reclaim it.
69 yield* pool.withConnection("other", fakeConnector(other), () => Effect.void);
70
71 expect(stale.closed).toBe(true);
72 yield* pool.close();
73 }),
74 );
75
76 it.effect("leaves a connection that is still inside the idle window alone", () =>
77 Effect.gen(function* () {
78 // The other half: sweeping must not become "close everything on any
79 // activity", which would destroy pooling while still passing the test
80 // above.
81 vi.useFakeTimers();
82 const pool = createMcpConnectionPool();
83 const fresh = { closed: false };
84 const other = { closed: false };
85
86 yield* pool.withConnection("fresh", fakeConnector(fresh), () => Effect.void);
87 vi.advanceTimersByTime(IDLE_TTL_MS - 1_000);
88 yield* pool.withConnection("other", fakeConnector(other), () => Effect.void);
89
90 expect(fresh.closed).toBe(false);
91 yield* pool.close();
92 }),
93 );
94
95 it.effect("still reuses a parked connection for the same key", () =>
96 Effect.gen(function* () {
97 // Guards the pool's whole reason for existing: a sweep that quietly broke
98 // reuse would leave both tests above green.
99 vi.useFakeTimers();
100 const pool = createMcpConnectionPool();
101 const first = { closed: false };
102 let dials = 0;
103 const counting: McpConnector = Effect.suspend(() => {
104 dials += 1;
105 return fakeConnector(first);
106 });
107
108 yield* pool.withConnection("same", counting, () => Effect.void);
109 yield* pool.withConnection("same", counting, () => Effect.void);
110
111 expect(dials).toBe(1);
112 yield* pool.close();
113 }),
114 );
115
116 it.effect("a close that never answers does not strand the acquire that swept it", () =>
117 Effect.gen(function* () {
118 // The sweep is paid for by whichever invocation happens to acquire next,
119 // so an unresponsive teardown is a live caller's latency. Only `Date` is
120 // faked here: the wait being asserted is an Effect sleep, which belongs to
121 // `it.effect`'s TestClock, and faking the platform timers underneath it
122 // would leave nothing to advance.
123 vi.useFakeTimers({ toFake: ["Date"] });
124 const pool = createMcpConnectionPool();
125 const other = { closed: false };
126
127 yield* pool.withConnection("hung", hangingConnector(), () => Effect.void);
128 vi.advanceTimersByTime(IDLE_TTL_MS + 1_000);
129
130 const fiber = yield* Effect.forkChild(
131 pool.withConnection("other", fakeConnector(other), () => Effect.void),
132 );
133
134 // Past the close timeout, but nowhere near "forever": an unbounded close
135 // would leave this fiber suspended and the join below would never return.
136 yield* TestClock.adjust(Duration.seconds(5));
137 yield* Fiber.join(fiber);
138
139 // The unrelated connection was served, not collateral damage.
140 expect(other.closed).toBe(false);
141 yield* pool.close();
142 }),
143 );
144});
packages/plugins/mcp/src/sdk/connection-pool.ts+48−3
@@ -1,4 +1,4 @@
1import { Cause, Effect, Exit, Predicate } from "effect";1import { Cause, Duration, Effect, Exit, Predicate } from "effect";
2 2
3import type { McpConnection, McpConnector } from "./connection";3import type { McpConnection, McpConnector } from "./connection";
4import type { McpInvocationError } from "./errors";4import type { McpInvocationError } from "./errors";
@@ -9,6 +9,16 @@ import type { McpInvocationError } from "./errors"; · 4 unmodified lines
9 9
10const IDLE_TTL_MS = 5 * 60 * 1_000;10const IDLE_TTL_MS = 5 * 60 * 1_000;
11 11
12/** How long a `close()` is waited on before the connection is abandoned.
13 *
14 * Eviction is driven by live traffic — the invocation that acquires a lease is
15 * the one that runs the sweep — so an unbounded close is that caller's problem:
16 * a server that accepts the close and then goes quiet would hold up a request
17 * that has nothing to do with the connection being reclaimed. Teardown is
18 * milliseconds' work when it works at all, so anything past this window is a
19 * socket that is not coming back. */
20const CLOSE_TIMEOUT = Duration.seconds(2);
21
12type IdleConnection = {22type IdleConnection = {
13 readonly connection: McpConnection;23 readonly connection: McpConnection;
14 readonly idleSince: number;24 readonly idleSince: number;
@@ -20,7 +30,7 @@ type ConnectionLease = { · 5 unmodified lines
20};30};
21 31
22const closeQuietly = (connection: McpConnection): Effect.Effect<void> =>32const closeQuietly = (connection: McpConnection): Effect.Effect<void> =>
23 Effect.tryPromise(() => connection.close()).pipe(Effect.ignore);33 Effect.tryPromise(() => connection.close()).pipe(Effect.timeout(CLOSE_TIMEOUT), Effect.ignore);
24 34
25const isMcpInvocationError = (error: unknown): error is McpInvocationError =>35const isMcpInvocationError = (error: unknown): error is McpInvocationError =>
26 Predicate.isTagged(error, "McpInvocationError");36 Predicate.isTagged(error, "McpInvocationError");
@@ -64,12 +74,47 @@ export interface McpConnectionPool { · 37 unmodified lines
64}74}
65 75
66/** Creates an MCP connection pool with lazy five-minute idle eviction and one76/** Creates an MCP connection pool with lazy five-minute idle eviction and one
67 * automatic fresh-dial retry for a reused session rejected with HTTP 404. */77 * automatic fresh-dial retry for a reused session rejected with HTTP 404.
78 *
79 * "Lazy" means activity-driven — there is no timer and no background fiber — but
80 * it applies to EVERY parked connection, not only the identity being asked for.
81 * A pooled session holds the credential it was dialled with, so an identity that
82 * is never requested again must still age out. */
68export const createMcpConnectionPool = (): McpConnectionPool => {83export const createMcpConnectionPool = (): McpConnectionPool => {
69 const idle = new Map<string, IdleConnection>();84 const idle = new Map<string, IdleConnection>();
70 85
86 /** Close and drop every entry past the idle window, not just the one being
87 * asked for.
88 *
89 * The TTL used to be consulted only against `idle.get(key)`, so an identity
90 * that was never dialled again was never examined again: its session stayed
91 * open and authenticated indefinitely, holding the bearer it was dialled
92 * with. The advertised bound only held for connections that happened to be
93 * reused.
94 *
95 * Still lazy — activity drives it, there is no timer and no background fiber.
96 * The map holds at most one entry per identity, so scanning it is trivial.
97 *
98 * The entries leave the map synchronously, before any close is awaited, so a
99 * slow teardown can never hand the same connection to a second caller. The
100 * closes themselves run concurrently and each is bounded by `CLOSE_TIMEOUT`,
101 * the same shape `close()` below uses: the sweep is work the acquiring
102 * invocation pays for, and one unresponsive server must not be able to stall
103 * it, let alone stall the connections queued behind it. */
104 const sweepExpired = Effect.suspend(() => {
105 const now = Date.now();
106 const expired: McpConnection[] = [];
107 for (const [key, entry] of idle) {
108 if (now - entry.idleSince < IDLE_TTL_MS) continue;
109 idle.delete(key);
110 expired.push(entry.connection);
111 }
112 return Effect.forEach(expired, closeQuietly, { concurrency: "unbounded", discard: true });
113 });
114
71 const acquire = (key: string, connector: McpConnector, forceFresh: boolean) =>115 const acquire = (key: string, connector: McpConnector, forceFresh: boolean) =>
72 Effect.gen(function* () {116 Effect.gen(function* () {
117 yield* sweepExpired;
73 if (forceFresh) {118 if (forceFresh) {
74 const connection = yield* connector;119 const connection = yield* connector;
75 return { connection, reused: false } satisfies ConnectionLease;120 return { connection, reused: false } satisfies ConnectionLease;