Commit 71d1305a3887d4880af34789926541fa76a3fbab
1 parent
42587067
fix(session-pool): 补上机器级会话上限 —— 池子里每个 session 是一个常驻 CLI 进程
sessions.size 在此之前没有任何人过问。一次心跳泄漏就攒到过 82 个进程 / 17.5 GB(约 215MB/个),而批 4 刚把公司级并行放开(单公司上限 20)—— 机器上同时能起多少完全没有闸门,正是最容易失控的时刻。 AGENT_MAX_SESSIONS,缺省 12,**调用时读不是模块加载时快照**(快照一次就 意味着改配置要重启才生效,而且不报错 —— company-paths.js 顶上那条教训)。 非法值(0/负数/非整数)回落缺省,不当成无限。 满了怎么办:**回收最久没用的空闲会话,而不是直接拒绝**。session 是可复用的 池化资源,杀掉一个空闲的只是让它下次冷启动几秒;拒绝会让一个本来能跑的任务 因为资源原因失败 —— 那是把可恢复的代价换成不可恢复的。 只有一个空闲的都没有(全部在途)时才抛 SessionPoolFullError。刻意抛而不是 静默放行 —— 静默超出正是这个仓库栽过的那一类「超出去但不报错」。 两个实现细节: - 腾位置发生在 createConnection **之前**:起完再判等于该拦的那个已经起来了 - 空闲态用显式的 idleSince 时间戳记,不去戳 IdleTimer 的私有 _timer; getOrCreate 复用时清空它(在途 = 不可被挑走) - _evictForRoom 故意不 await reset():腾位置只需要把它从 Map 摘掉, 子进程收尾可以后台走完;await 的话一次腾位要串行等一次进程退出 守卫:server/tests/unit/session-pool-max.test.js(5 例,不起真进程, 只验判据本身:挑最久空闲的、在途的不许动、全在途就抛、没满不动、非法配置回落) 全量:129 文件 / 1245 测试全绿。 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Showing
2 changed files
with
137 additions
and
1 deletions
server/src/core/session-pool.js
| @@ -32,6 +32,33 @@ function idleMs() { | @@ -32,6 +32,33 @@ function idleMs() { | ||
| 32 | return Number(process.env.AGENT_SESSION_IDLE_MS) || 30 * 60 * 1000; | 32 | return Number(process.env.AGENT_SESSION_IDLE_MS) || 30 * 60 * 1000; |
| 33 | } | 33 | } |
| 34 | 34 | ||
| 35 | +/** | ||
| 36 | + * 机器级会话上限(27-P3C §15.5 第 2 条)。 | ||
| 37 | + * | ||
| 38 | + * 池子里每个 session 就是一个**常驻** claude CLI 子进程(stream-json 模式),实测单进程 RSS | ||
| 39 | + * 约 200-300MB。在此之前 `this.sessions.size` **没有任何人过问** —— 一次心跳泄漏就攒到过 | ||
| 40 | + * 82 个进程 / 17.5 GB。批 4 又刚把公司级并行放开(单公司上限 20),机器上同时能起多少 | ||
| 41 | + * 完全没有闸门。 | ||
| 42 | + * | ||
| 43 | + * **调用时读,不在模块加载时快照**(`company-paths.js` 顶上那条教训:快照一次就意味着 | ||
| 44 | + * 「改了配置要重启才生效」,而且不报错)。 | ||
| 45 | + */ | ||
| 46 | +const DEFAULT_MAX_SESSIONS = 12; | ||
| 47 | +function maxSessions() { | ||
| 48 | + const raw = Number(process.env.AGENT_MAX_SESSIONS); | ||
| 49 | + return Number.isInteger(raw) && raw > 0 ? raw : DEFAULT_MAX_SESSIONS; | ||
| 50 | +} | ||
| 51 | + | ||
| 52 | +/** 池子满了、且**一个空闲的都没有**(全都在途)时抛这个 —— 调用方该重试或排队,不是崩 */ | ||
| 53 | +class SessionPoolFullError extends Error { | ||
| 54 | + constructor(size, max) { | ||
| 55 | + super(`会话池已满(${size}/${max})且无空闲会话可回收:机器级并发上限,请稍后重试`); | ||
| 56 | + this.name = 'SessionPoolFullError'; | ||
| 57 | + this.size = size; | ||
| 58 | + this.max = max; | ||
| 59 | + } | ||
| 60 | +} | ||
| 61 | + | ||
| 35 | class SessionPool extends EventEmitter { | 62 | class SessionPool extends EventEmitter { |
| 36 | constructor() { | 63 | constructor() { |
| 37 | super(); | 64 | super(); |
| @@ -92,6 +119,9 @@ class SessionPool extends EventEmitter { | @@ -92,6 +119,9 @@ class SessionPool extends EventEmitter { | ||
| 92 | } else { | 119 | } else { |
| 93 | // 其他协议:组合 Launcher + Transport + JSON-RPC | 120 | // 其他协议:组合 Launcher + Transport + JSON-RPC |
| 94 | const { createConnection } = require('@mas/acp-core'); | 121 | const { createConnection } = require('@mas/acp-core'); |
| 122 | + // ⚠️ 起子进程**之前**先腾位置:起完再判等于该拦的那一个已经起来了 | ||
| 123 | + this._evictForRoom(); | ||
| 124 | + | ||
| 95 | const { transport, launcher: lnch } = await createConnection( | 125 | const { transport, launcher: lnch } = await createConnection( |
| 96 | agentConfig, launcherConfig, workDir | 126 | agentConfig, launcherConfig, workDir |
| 97 | ); | 127 | ); |
| @@ -133,6 +163,7 @@ class SessionPool extends EventEmitter { | @@ -133,6 +163,7 @@ class SessionPool extends EventEmitter { | ||
| 133 | 163 | ||
| 134 | if (existing?.protocol?.isConnected) { | 164 | if (existing?.protocol?.isConnected) { |
| 135 | existing.idleTimer?.cancel(); // 请求在途期间不计空闲 | 165 | existing.idleTimer?.cancel(); // 请求在途期间不计空闲 |
| 166 | + existing.idleSince = null; // 在途 = 不可被 _evictForRoom 挑走 | ||
| 136 | this._emit(agentName, protocolName, 'reused', existing.sessionId, key); | 167 | this._emit(agentName, protocolName, 'reused', existing.sessionId, key); |
| 137 | return existing; | 168 | return existing; |
| 138 | } | 169 | } |
| @@ -164,6 +195,8 @@ class SessionPool extends EventEmitter { | @@ -164,6 +195,8 @@ class SessionPool extends EventEmitter { | ||
| 164 | const session = this.sessions.get(sessionKey); | 195 | const session = this.sessions.get(sessionKey); |
| 165 | if (!session) return; | 196 | if (!session) return; |
| 166 | session.idleTimer?.cancel(); | 197 | session.idleTimer?.cancel(); |
| 198 | + // 空闲态用显式时间戳记,不去戳 IdleTimer 的私有 _timer —— `_evictForRoom` 靠它挑最久没用的那个 | ||
| 199 | + session.idleSince = Date.now(); | ||
| 167 | const ms = idleMs(); | 200 | const ms = idleMs(); |
| 168 | session.idleTimer = new IdleTimer( | 201 | session.idleTimer = new IdleTimer( |
| 169 | ms, | 202 | ms, |
| @@ -176,6 +209,38 @@ class SessionPool extends EventEmitter { | @@ -176,6 +209,38 @@ class SessionPool extends EventEmitter { | ||
| 176 | } | 209 | } |
| 177 | 210 | ||
| 178 | /** | 211 | /** |
| 212 | + * 起新会话之前腾位置:池子满了就回收**最久没用**的空闲会话。 | ||
| 213 | + * | ||
| 214 | + * 为什么是「回收空闲」而不是「直接拒绝」:session 是可复用的池化资源,杀掉一个空闲的 | ||
| 215 | + * 只是让它下次冷启动一次(几秒),而拒绝会让一个本来能跑的任务因为资源原因失败 —— | ||
| 216 | + * 那是把可恢复的代价换成不可恢复的。 | ||
| 217 | + * | ||
| 218 | + * 只有**一个空闲的都没有**(全部在途)时才抛 `SessionPoolFullError`:那时机器是真满了, | ||
| 219 | + * 再起就是超发。**刻意抛而不是静默放行** —— 静默放行正是这个仓库栽过的那一类 | ||
| 220 | + * 「写不进去/超出去但不报错」(27-P3C §九.4quater)。 | ||
| 221 | + * | ||
| 222 | + * 注意 `reset()` 是 async(要 await launcher.stop()),这里**故意不 await**: | ||
| 223 | + * 腾位置只需要把它从 Map 里摘掉(reset 第一件事就是),子进程的收尾可以在后台走完。 | ||
| 224 | + * await 的话,一次腾位要串行等一次进程退出,而调用方还在等着起自己的会话。 | ||
| 225 | + */ | ||
| 226 | + _evictForRoom() { | ||
| 227 | + const max = maxSessions(); | ||
| 228 | + while (this.sessions.size >= max) { | ||
| 229 | + let victim = null; | ||
| 230 | + for (const [key, sess] of this.sessions) { | ||
| 231 | + if (!sess.idleSince) continue; // 在途的不许动 | ||
| 232 | + if (!victim || sess.idleSince < victim.sess.idleSince) victim = { key, sess }; | ||
| 233 | + } | ||
| 234 | + if (!victim) throw new SessionPoolFullError(this.sessions.size, max); | ||
| 235 | + console.log(`[SessionPool] 🧹 池满(${this.sessions.size}/${max}),回收最久空闲的 ${victim.key}`); | ||
| 236 | + this.reset(victim.key).catch(() => {}); | ||
| 237 | + if (this.sessions.has(victim.key)) { // reset 没能摘掉 = 防死循环 | ||
| 238 | + this.sessions.delete(victim.key); | ||
| 239 | + } | ||
| 240 | + } | ||
| 241 | + } | ||
| 242 | + | ||
| 243 | + /** | ||
| 179 | * 重置(断开 + 清理)单个会话 | 244 | * 重置(断开 + 清理)单个会话 |
| 180 | */ | 245 | */ |
| 181 | async reset(sessionKey) { | 246 | async reset(sessionKey) { |
| @@ -262,4 +327,4 @@ class SessionPool extends EventEmitter { | @@ -262,4 +327,4 @@ class SessionPool extends EventEmitter { | ||
| 262 | } | 327 | } |
| 263 | } | 328 | } |
| 264 | 329 | ||
| 265 | -module.exports = { SessionPool }; | 330 | +module.exports = { SessionPool, SessionPoolFullError, DEFAULT_MAX_SESSIONS, maxSessions }; |
server/tests/unit/session-pool-max.test.js
0 → 100644
| 1 | +/** | ||
| 2 | + * 机器级会话上限(27-P3C §15.5 第 2 条)。 | ||
| 3 | + * | ||
| 4 | + * 为什么要这条守卫:池子里每个 session 是一个**常驻** claude CLI 子进程(约 200-300MB RSS), | ||
| 5 | + * 而 `sessions.size` 在此之前没有任何人过问 —— 一次心跳泄漏攒到过 82 个进程 / 17.5 GB。 | ||
| 6 | + * 批 4 又刚把公司级并行放开(单公司上限 20),机器上同时能起多少完全没闸门。 | ||
| 7 | + * | ||
| 8 | + * 这里**不起真进程**:直接往 `pool.sessions` 里塞假 session,只验腾位置的判据本身 | ||
| 9 | + * (挑最久空闲的、在途的不许动、全在途就抛)。真起进程的路径由 agent-manager 的用例覆盖。 | ||
| 10 | + */ | ||
| 11 | +import { describe, it, expect, beforeEach, afterEach } from 'vitest'; | ||
| 12 | +import { createRequire } from 'module'; | ||
| 13 | +const require_ = createRequire(import.meta.url); | ||
| 14 | +const { SessionPool, SessionPoolFullError, DEFAULT_MAX_SESSIONS } = require_('../../src/core/session-pool'); | ||
| 15 | + | ||
| 16 | +describe('SessionPool 机器级上限', () => { | ||
| 17 | + let pool; | ||
| 18 | + const OLD = process.env.AGENT_MAX_SESSIONS; | ||
| 19 | + | ||
| 20 | + beforeEach(() => { pool = new SessionPool(); }); | ||
| 21 | + afterEach(() => { | ||
| 22 | + if (OLD === undefined) delete process.env.AGENT_MAX_SESSIONS; | ||
| 23 | + else process.env.AGENT_MAX_SESSIONS = OLD; | ||
| 24 | + }); | ||
| 25 | + | ||
| 26 | + /** 塞一个假 session;idleSince 为 null = 在途 */ | ||
| 27 | + function seed(key, idleSince) { | ||
| 28 | + pool.sessions.set(key, { key, idleSince, protocol: null, launcher: null, idleTimer: null }); | ||
| 29 | + } | ||
| 30 | + | ||
| 31 | + it('缺省上限是 12,且从 env 调用时读(不是模块加载时快照)', () => { | ||
| 32 | + expect(DEFAULT_MAX_SESSIONS).toBe(12); | ||
| 33 | + process.env.AGENT_MAX_SESSIONS = '3'; | ||
| 34 | + seed('a', 100); seed('b', 200); seed('c', 300); | ||
| 35 | + pool._evictForRoom(); // 3/3 满 → 该腾一个 | ||
| 36 | + expect(pool.sessions.size).toBe(2); | ||
| 37 | + expect(pool.sessions.has('a')).toBe(false); // 挑的是最久空闲的那个 | ||
| 38 | + }); | ||
| 39 | + | ||
| 40 | + it('在途的会话不许被腾走', () => { | ||
| 41 | + process.env.AGENT_MAX_SESSIONS = '2'; | ||
| 42 | + seed('inflight', null); // 在途 | ||
| 43 | + seed('idle', 100); | ||
| 44 | + pool._evictForRoom(); | ||
| 45 | + expect(pool.sessions.has('inflight')).toBe(true); | ||
| 46 | + expect(pool.sessions.has('idle')).toBe(false); | ||
| 47 | + }); | ||
| 48 | + | ||
| 49 | + it('全部在途时抛 SessionPoolFullError —— 不静默放行', () => { | ||
| 50 | + process.env.AGENT_MAX_SESSIONS = '2'; | ||
| 51 | + seed('x', null); seed('y', null); | ||
| 52 | + expect(() => pool._evictForRoom()).toThrow(SessionPoolFullError); | ||
| 53 | + // 关键:一个都没被杀(在途的动不得),也没有偷偷放行 | ||
| 54 | + expect(pool.sessions.size).toBe(2); | ||
| 55 | + }); | ||
| 56 | + | ||
| 57 | + it('没满时什么都不做', () => { | ||
| 58 | + process.env.AGENT_MAX_SESSIONS = '5'; | ||
| 59 | + seed('a', 100); seed('b', 200); | ||
| 60 | + pool._evictForRoom(); | ||
| 61 | + expect(pool.sessions.size).toBe(2); | ||
| 62 | + }); | ||
| 63 | + | ||
| 64 | + it('非法配置回落缺省,不当成无限', () => { | ||
| 65 | + for (const bad of ['0', '-1', 'abc', '']) { | ||
| 66 | + process.env.AGENT_MAX_SESSIONS = bad; | ||
| 67 | + const { maxSessions } = require_('../../src/core/session-pool'); | ||
| 68 | + expect(maxSessions()).toBe(DEFAULT_MAX_SESSIONS); | ||
| 69 | + } | ||
| 70 | + }); | ||
| 71 | +}); |