diff --git a/.env.example b/.env.example index f8046ab..fbf768f 100644 --- a/.env.example +++ b/.env.example @@ -22,6 +22,8 @@ XFT_OPEN_API_ENABLED=true # XFT_OPEN_API_DINGTALK_APP_CODE=DDingTalk # 调优 +# 每账号 MCP 连接会话上限;1–1000,默认 32,全局总上限固定为 1000。 +# MCP_SESSION_MAX_PER_USER=32 # OFFICIAL_OA_LOOKBACK_DAYS=120 # 审批写入批量上限固定为 20,不提供环境变量覆盖。 # MCP_ACCESS_TOKEN_TTL_SECONDS=900 diff --git a/docs/deployment.md b/docs/deployment.md index c20be5e..4c6205f 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -128,7 +128,10 @@ https://approval.justconn.top/downloads/dingmcp/install.md ## 运维检查 -网关会话上限为每个已认证用户 8 个、全局 1000 个,同时计入尚未完成初始化的连接。 +网关会话上限默认为每个已认证用户 32 个、全局 1000 个,同时计入尚未完成初始化的连接。 +运维人员可在服务端环境配置中设置 `MCP_SESSION_MAX_PER_USER=64` 等正整数(范围 1–1000), +并通过上述部署流程更新网关后生效;无需修改或更新客户端。未配置时使用 32,非法值拒绝启动。 +该阈值用于限制一个账号占用的连接和内存,不是审批条数、同时执行审批的配额或上游平台限制。 现有 Bridge 客户端退出或刷新登录时,即使没有发送 MCP DELETE,网关也会在观察到 Streamable HTTP GET 长连接断开后保留 60 秒重连宽限期,再回收遗留会话;后台每 15 秒 检查一次,新连接进入时也会检查,因此通常在断开后的 60–75 秒内释放名额。 @@ -137,7 +140,7 @@ Streamable HTTP GET 长连接断开后保留 60 秒重连宽限期,再回收 实际请求处理结束前不会因为断线、取消或超时回收会话,也不会重试任何审批写入。 回收状态只保存在内存中,不记录审批参数、正文、Token 或用户身份到日志。 -如果 8 个会话确实仍在连接,网关仍返回 HTTP 429 和 `Retry-After: 60`,需要关闭不用的 +如果会话确实仍在连接并达到配置上限,网关仍返回 HTTP 429 和 `Retry-After: 60`,需要关闭不用的 Agent 连接再重试。旧客户端可能仍将 429 显示为“审批网关暂时无法连接”;此次服务端 修复无需用户更新客户端,但不会改变客户端内置的错误文案。 diff --git a/src/config.ts b/src/config.ts index b8e5c0e..34d8a4a 100644 --- a/src/config.ts +++ b/src/config.ts @@ -1,6 +1,9 @@ import { readFileSync, statSync } from "node:fs"; import { APPROVAL_BATCH_LIMIT } from "./approval-contract.js"; +export const DEFAULT_MCP_SESSION_MAX_PER_USER = 32; +export const MCP_SESSION_MAX_TOTAL = 1000; + export interface AppConfig { appMode?: "gateway"; clientId: string; @@ -11,6 +14,7 @@ export interface AppConfig { publicBaseUrl: string; bindHost?: string; port: number; + mcpSessionMaxPerUser?: number; requestTimeoutMs: number; officialLookbackDays: number; officialScanConcurrency: number; @@ -167,6 +171,10 @@ export function loadConfig(env: NodeJS.ProcessEnv = process.env): AppConfig { if (!remoteAuthEnabled || !xftOpenApiEnabled) { throw new Error("gateway 模式必须启用 MCP 远程 OAuth 和薪福通正式 OpenAPI"); } + const mcpSessionMaxPerUser = positiveInteger(env, "MCP_SESSION_MAX_PER_USER", DEFAULT_MCP_SESSION_MAX_PER_USER); + if (mcpSessionMaxPerUser > MCP_SESSION_MAX_TOTAL) { + throw new Error(`MCP_SESSION_MAX_PER_USER 不能超过全局会话上限 ${MCP_SESSION_MAX_TOTAL}`); + } return { appMode, clientId: dingTalkCredential(env, "DINGTALK_CLIENT_ID", "clientId"), @@ -177,6 +185,7 @@ export function loadConfig(env: NodeJS.ProcessEnv = process.env): AppConfig { publicBaseUrl, bindHost, port, + mcpSessionMaxPerUser, requestTimeoutMs: positiveInteger(env, "DINGTALK_REQUEST_TIMEOUT_MS", 15_000), officialLookbackDays: Math.min(120, positiveInteger(env, "OFFICIAL_OA_LOOKBACK_DAYS", 120)), officialScanConcurrency: positiveInteger(env, "OFFICIAL_OA_SCAN_CONCURRENCY", 2), diff --git a/src/mcp-http.ts b/src/mcp-http.ts index d4492dc..2836fbb 100644 --- a/src/mcp-http.ts +++ b/src/mcp-http.ts @@ -6,24 +6,24 @@ import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js"; import { isInitializeRequest } from "@modelcontextprotocol/sdk/types.js"; import type { AuthenticatedDingTalkUser } from "./xft.js"; import { McpSessionLifecycle } from "./mcp-session-lifecycle.js"; +import { DEFAULT_MCP_SESSION_MAX_PER_USER, MCP_SESSION_MAX_TOTAL } from "./config.js"; type TransportEntry = | { kind: "streamable"; transport: StreamableHTTPServerTransport; userId: string; lifecycle: McpSessionLifecycle } | { kind: "sse"; transport: SSEServerTransport; userId: string; lifecycle: McpSessionLifecycle }; -const MCP_SESSION_MAX_PER_USER = 8; -const MCP_SESSION_MAX_TOTAL = 1000; - interface McpTransportOptions { authenticate?: RequestHandler; // Production always resolves this from verified OAuth, never request parameters. requestUser: (req: Request) => AuthenticatedDingTalkUser; createServer: (user: AuthenticatedDingTalkUser, lifecycle: McpSessionLifecycle) => McpServer; now?: () => number; + maxSessionsPerUser?: number; } export function installMcpTransports(app: Express, options: McpTransportOptions) { const now = options.now ?? Date.now; + const maxSessionsPerUser = options.maxSessionsPerUser ?? DEFAULT_MCP_SESSION_MAX_PER_USER; const transports = new Map(); let pendingTransportSessions = 0; const pendingTransportSessionsByUser = new Map(); @@ -40,7 +40,7 @@ export function installMcpTransports(app: Express, options: McpTransportOptions) }; const sessionCapacityAvailable = (userId: string) => transports.size + pendingTransportSessions < MCP_SESSION_MAX_TOTAL && [...transports.values()].filter((entry) => entry.userId === userId).length - + (pendingTransportSessionsByUser.get(userId) ?? 0) < MCP_SESSION_MAX_PER_USER; + + (pendingTransportSessionsByUser.get(userId) ?? 0) < maxSessionsPerUser; const reserveSession = (userId: string) => { pendingTransportSessions += 1; pendingTransportSessionsByUser.set(userId, (pendingTransportSessionsByUser.get(userId) ?? 0) + 1); diff --git a/src/server.ts b/src/server.ts index a9a29eb..bc649e4 100644 --- a/src/server.ts +++ b/src/server.ts @@ -462,6 +462,7 @@ export async function startHttpServer(config: AppConfig): Promise { installMcpTransports(app, { authenticate: authMiddleware, + maxSessionsPerUser: config.mcpSessionMaxPerUser, requestUser: (req) => requestUser(config, oauth, req.auth), createServer: (user, lifecycle) => createApprovalMcpServer( config, service, xft, undefined, officialOaReturns, user, diff --git a/test/dingtalk.test.ts b/test/dingtalk.test.ts index ff5ffe3..45e6f40 100644 --- a/test/dingtalk.test.ts +++ b/test/dingtalk.test.ts @@ -70,6 +70,24 @@ test("loadConfig rejects the removed local DWS architecture", () => { }), /禁止配置本地 DWS/); }); +test("gateway session capacity defaults to 32 and validates operator overrides", () => { + const env = { + APP_MODE: "gateway", DINGTALK_CLIENT_ID: "mock-key", DINGTALK_CLIENT_SECRET: "mock-secret", + DINGTALK_CORP_ID: "mock-corp", MCP_PUBLIC_BASE_URL: "https://approval.example.test", + MCP_REMOTE_AUTH_ENABLED: "true", MCP_SESSION_SECRET: "mock-session-secret-with-at-least-32-bytes", + MCP_APPROVAL_REF_SECRET: "mock-reference-secret-with-at-least-32-bytes", + XFT_OPEN_API_ENABLED: "true", XFT_OPEN_API_APP_ID: "mock-app", + XFT_OPEN_API_AUTHORITY_SECRET: "1".repeat(64), XFT_OPEN_API_ENTERPRISE_ID: "mock-enterprise", + }; + assert.equal(loadConfig(env).mcpSessionMaxPerUser, 32); + for (const value of ["1", "64", "1000"]) { + assert.equal(loadConfig({ ...env, MCP_SESSION_MAX_PER_USER: value }).mcpSessionMaxPerUser, Number(value)); + } + for (const value of ["0", "-1", "1.5", "invalid", "1001"]) { + assert.throws(() => loadConfig({ ...env, MCP_SESSION_MAX_PER_USER: value }), /MCP_SESSION_MAX_PER_USER/); + } +}); + test("loadConfig reads AuthoritySecret from the second line of an XFT portal credential file", () => { const directory = mkdtempSync(join(tmpdir(), "ding-mcp-xft-openapi-config-")); const secretPath = join(directory, "xft.key"); diff --git a/test/mcp-http.test.ts b/test/mcp-http.test.ts index 038907b..b9047a2 100644 --- a/test/mcp-http.test.ts +++ b/test/mcp-http.test.ts @@ -25,7 +25,8 @@ async function until(condition: () => boolean) { assert.fail("HTTP lifecycle event did not arrive"); } -async function harness(t: TestContext, readTodos: () => Promise = async () => {}) { +async function harness(t: TestContext, readTodos: () => Promise = async () => {}, + limits: { maxSessionsPerUser?: number } = { maxSessionsPerUser: 8 }) { let time = Date.now(); let closedStreams = 0; const app = express(); @@ -36,6 +37,7 @@ async function harness(t: TestContext, readTodos: () => Promise = async () }); const routes = installMcpTransports(app, { now: () => time, + maxSessionsPerUser: limits.maxSessionsPerUser, authenticate: (req, res, next) => { // Synthetic bearer credentials; no external approval or OAuth writes. const token = req.get("authorization"); @@ -118,6 +120,22 @@ test("old bridge disconnects release eight occupied slots without a client upgra assert.equal((await h.rpc(sessions[0]!.transport.sessionId, "a", "tools/list")).response.status, 404); }); +test("default gateway capacity admits 32 live connections for one user", async (t) => { + const h = await harness(t, undefined, {}); + const sessions = []; + for (let i = 0; i < 32; i += 1) sessions.push(await h.connect()); + assert.equal((await h.rpc()).response.status, 429); + assert.ok((await sessions[0]!.client.listTools()).tools.length > 0); + assert.ok((await sessions[31]!.client.listTools()).tools.length > 0); +}); + +test("operator override permits 64 sessions while retaining its configured cap", async (t) => { + const h = await harness(t, undefined, { maxSessionsPerUser: 64 }); + const results = await Promise.all(Array.from({ length: 65 }, () => h.rpc())); + assert.equal(results.filter(({ response }) => response.status === 200).length, 64); + assert.equal(results.filter(({ response }) => response.status === 429).length, 1); +}); + test("live streams survive idle cleanup and capacity does not evict another user", async (t) => { const h = await harness(t); const sessions = [];