diff --git a/Dockerfile b/Dockerfile index bdc8c9e..61b200f 100644 --- a/Dockerfile +++ b/Dockerfile @@ -52,6 +52,7 @@ RUN apt-get update \ libssl3 \ zlib1g \ lsb-release \ + procps \ wget \ xdg-utils \ && rm -rf /var/lib/apt/lists/* diff --git a/package.json b/package.json index 2110df4..552c378 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@missionsquad/mcp-api", - "version": "1.11.11", + "version": "1.11.19", "description": "MCP Servers exposed via HTTP API", "main": "dist/index.js", "repository": "missionsquad/mcp-api", @@ -30,7 +30,7 @@ "author": "Jayson Jacobs", "license": "Apache-2.0", "dependencies": { - "@missionsquad/puppeteer-scraper": "^1.1.1", + "@missionsquad/puppeteer-scraper": "^1.1.2", "@modelcontextprotocol/sdk": "^1.13.0", "@types/bcrypt": "^5.0.2", "base64-js": "^1.5.1", diff --git a/src/builtin-servers/servers/searxng.ts b/src/builtin-servers/servers/searxng.ts index 4d2211c..5c4d716 100644 --- a/src/builtin-servers/servers/searxng.ts +++ b/src/builtin-servers/servers/searxng.ts @@ -12,6 +12,8 @@ export class BuiltInSearxngServer extends BaseBuiltInServer { private scraper: PuppeteerScraper | null = null private scraperReady = false + private stopped = false + private retryTimer: NodeJS.Timeout | null = null private readonly MAX_PUPPETEER_RETRIES = 5 private readonly INITIAL_RETRY_DELAY_MS = 15000 @@ -97,14 +99,22 @@ export class BuiltInSearxngServer extends BaseBuiltInServer { } async stop(): Promise { + // Mark stopped and cancel any pending init retry so a scheduled attempt + // cannot recreate the scraper/browser after shutdown. + this.stopped = true + if (this.retryTimer) { + clearTimeout(this.retryTimer) + this.retryTimer = null + } if (this.scraper) { try { - // Puppeteer cleanup would go here - // Note: PuppeteerScraper needs a cleanup method + await this.scraper.closeBrowser() log({ level: 'info', msg: 'Stopped Puppeteer scraper' }) } catch (error) { log({ level: 'error', msg: 'Error stopping Puppeteer scraper', error }) } + this.scraper = null + this.scraperReady = false } } @@ -273,38 +283,72 @@ export class BuiltInSearxngServer extends BaseBuiltInServer { } private async initializePuppeteerWithRetries(retryCount = 0): Promise { + // Bail out if the server was stopped before this (possibly delayed) attempt ran. + if (this.stopped) { + return + } + let scraper: PuppeteerScraper | undefined try { - log({ - level: 'info', - msg: `Starting Puppeteer initialization (Attempt ${retryCount + 1}/${this.MAX_PUPPETEER_RETRIES})...` + log({ + level: 'info', + msg: `Starting Puppeteer initialization (Attempt ${retryCount + 1}/${this.MAX_PUPPETEER_RETRIES})...` }) - - this.scraper = new PuppeteerScraper({ + + scraper = new PuppeteerScraper({ headless: true, ignoreHTTPSErrors: true, blockResources: false, cacheSize: 1000, enableGPU: false }) - - await this.scraper.init() + + await scraper.init() + + // stop() may have been called while init() was in flight; if so, close the + // freshly created browser instead of publishing it. + if (this.stopped) { + await scraper.closeBrowser() + return + } + + this.scraper = scraper this.scraperReady = true log({ level: 'info', msg: 'Puppeteer initialized successfully.' }) } catch (error) { - log({ - level: 'error', - msg: `Failed to initialize Puppeteer on attempt ${retryCount + 1}:`, - error + log({ + level: 'error', + msg: `Failed to initialize Puppeteer on attempt ${retryCount + 1}:`, + error }) - + + // Close any partially-created browser before retrying so repeated init + // failures cannot leak Chromium processes. closeBrowser() is a no-op if the + // scraper already cleaned up internally. + if (scraper) { + try { + await scraper.closeBrowser() + } catch (closeError) { + log({ level: 'error', msg: 'Error closing partially-initialized Puppeteer scraper', error: closeError }) + } + } + + if (this.stopped) { + return + } + if (retryCount < this.MAX_PUPPETEER_RETRIES - 1) { const delay = this.INITIAL_RETRY_DELAY_MS * Math.pow(2, retryCount) log({ level: 'info', msg: `Retrying in ${delay / 1000} seconds...` }) - setTimeout(() => this.initializePuppeteerWithRetries(retryCount + 1), delay) + this.retryTimer = setTimeout(() => { + this.retryTimer = null + this.initializePuppeteerWithRetries(retryCount + 1) + }, delay) + // Do not keep the process alive solely for a pending retry. + this.retryTimer.unref() } else { - log({ - level: 'error', - msg: 'Max retries reached. Puppeteer initialization failed permanently for this session.' + log({ + level: 'error', + msg: 'Max retries reached. Puppeteer initialization failed permanently for this session.' }) } } diff --git a/src/env.ts b/src/env.ts index 6b56964..9b210d0 100644 --- a/src/env.ts +++ b/src/env.ts @@ -2,6 +2,18 @@ import dotenv from 'dotenv' dotenv.config() +/** + * Parses a millisecond-interval env var. Returns `fallback` when the var is + * unset, blank, or not a finite number (so a typo defaults to the intended + * interval rather than silently disabling the feature). An explicit finite + * value — including `0`, which callers treat as "disabled" — is returned as-is. + */ +function parseIntervalMs(raw: string | undefined, fallback: number): number { + if (raw === undefined || raw.trim() === '') return fallback + const parsed = Number(raw) + return Number.isFinite(parsed) ? parsed : fallback +} + export const env = { DEBUG: /true/i.test(process.env.DEBUG || 'false'), ENABLE_OAUTH_LOGGING: /true/i.test(process.env.ENABLE_OAUTH_LOGGING || 'false'), @@ -19,6 +31,7 @@ export const env = { return { repo, name } }), SEARXNG_URL: process.env.SEARXNG_URL, + RESOURCE_STATS_INTERVAL_MS: parseIntervalMs(process.env.RESOURCE_STATS_INTERVAL_MS, 60000), PYTHON_BIN: process.env.PYTHON_BIN, PYTHON_VENV_DIR: process.env.PYTHON_VENV_DIR || 'packages/python', PIP_INDEX_URL: process.env.PIP_INDEX_URL, diff --git a/src/index.ts b/src/index.ts index 02a3f75..1ed9e8c 100644 --- a/src/index.ts +++ b/src/index.ts @@ -13,6 +13,7 @@ import { McpOAuthTokens } from './services/oauthTokens' import { McpUserSessions } from './services/userSessions' import { McpUserServerInstalls } from './services/userServerInstalls' import { McpDcrClients } from './services/dcrClients' +import { ResourceStatsService } from './services/resourceStats' export type Resource = { init: () => Promise @@ -93,6 +94,17 @@ export class API { // Set up circular dependency between MCPService and PackageService mcpController.getMcpService().setPackageService(packagesController.getPackageService()) + // Periodic resource-stats logging for mcp-api and its spawned MCP servers. + // Registered at the FRONT of resources so shutDown() stops it first: sampling + // must cease before the MCP/package services it samples begin tearing down, + // otherwise a tick mid-shutdown would spawn `ps` and log stale stats. + const resourceStatsService = new ResourceStatsService({ + intervalMs: env.RESOURCE_STATS_INTERVAL_MS, + getTrackedProcesses: () => mcpController.getMcpService().getStdioServerPids() + }) + await resourceStatsService.init() + this.resources.unshift(resourceStatsService) + app.get('/healthz', (req, res) => { console.log('Health checked') res.status(200).send('OK') diff --git a/src/services/mcp.ts b/src/services/mcp.ts index 10ff30c..170d46b 100644 --- a/src/services/mcp.ts +++ b/src/services/mcp.ts @@ -43,6 +43,7 @@ import { McpServerAlreadyExistsError } from './mcpErrors' import { validateExternalMcpUrl } from '../utils/ssrf' +import { getStdioTransportPid } from '../utils/resourceStats' export interface MCPConnection { client: Client @@ -918,6 +919,29 @@ const normalizeExternalAuthError = ( return error } +/** + * Maximum number of in-memory log lines retained for each MCP server and each + * per-user connection. Older lines are displaced once this count is exceeded. + */ +export const MAX_SERVER_LOG_LINES = 500 + +/** + * Append a log line to an in-memory logs array, capping the array at + * {@link MAX_SERVER_LOG_LINES} entries by COUNT ONLY (no time-based eviction). + * When the cap is exceeded, the oldest lines are dropped so the array retains + * the most recent {@link MAX_SERVER_LOG_LINES} lines in insertion order. + * + * @param logs - The logs array to append to. A no-op when `undefined`. + * @param message - The log line to append. + */ +export function appendServerLog(logs: string[] | undefined, message: string): void { + if (!logs) return + logs.push(message) + if (logs.length > MAX_SERVER_LOG_LINES) { + logs.splice(0, logs.length - MAX_SERVER_LOG_LINES) + } +} + export class MCPService implements Resource { public servers: Record = {} public userConnections: Record = {} @@ -2254,7 +2278,7 @@ export class MCPService implements Resource { const transportErrorHandler = async (error: Error) => { log({ level: 'error', msg: `${serverKey} transport error: ${error.message}`, error }) if (this.servers[serverKey]) { - this.servers[serverKey].logs?.push(error.message) + appendServerLog(this.servers[serverKey].logs, error.message) } } @@ -2326,7 +2350,7 @@ export class MCPService implements Resource { const logMsg = data.toString().trim() log({ level: 'error', msg: `[${server.name}] stderr: ${logMsg}` }) if (this.servers[serverKey]) { - this.servers[serverKey].logs?.push(logMsg) + appendServerLog(this.servers[serverKey].logs, logMsg) } } @@ -2452,7 +2476,7 @@ export class MCPService implements Resource { log({ level: 'error', msg: `[${username}:${server.name}] transport error: ${error.message}`, error }) const conn = this.userConnections[userKey] if (conn) { - conn.logs?.push(error.message) + appendServerLog(conn.logs, error.message) } } @@ -3822,6 +3846,30 @@ export class MCPService implements Resource { return updatedServer } + /** + * Lists the OS process ids of currently running stdio MCP servers, + * labeled by server name, for resource monitoring. + * + * Only servers with status 'connected' are included: `connection` is not + * cleared on transport close, so a disconnected/error server could otherwise + * contribute a stale pid and mislabel an unrelated process if the OS reused it. + */ + public getStdioServerPids(): { name: string; pid: number }[] { + const tracked: { name: string; pid: number }[] = [] + for (const serverKey of Object.keys(this.servers)) { + const server = this.servers[serverKey] + if (server.transportType !== 'stdio') continue + if (server.status !== 'connected') continue + const transport = server.connection?.transport + if (!transport) continue + const pid = getStdioTransportPid(transport) + if (pid !== undefined) { + tracked.push({ name: server.name, pid }) + } + } + return tracked + } + public async stop() { log({ level: 'info', msg: 'Stopping MCP servers' }) diff --git a/src/services/resourceStats.ts b/src/services/resourceStats.ts new file mode 100644 index 0000000..5f217a9 --- /dev/null +++ b/src/services/resourceStats.ts @@ -0,0 +1,148 @@ +import type { Resource } from '..' +import { log } from '../utils/general' +import { + SubtreeSample, + aggregateDirectChildSubtrees, + formatMebibytes, + sampleProcessTree, + shortenProcessLabel +} from '../utils/resourceStats' + +export interface TrackedProcess { + name: string + pid: number +} + +export interface ResourceStatsServiceOptions { + /** Sampling interval in milliseconds. Values <= 0 disable sampling entirely. */ + intervalMs: number + /** Returns the currently running MCP server child processes to label by name. */ + getTrackedProcesses: () => TrackedProcess[] +} + +/** + * Periodically logs memory/CPU usage for the mcp-api process itself, every + * child process subtree it has spawned (installed stdio MCP servers, the + * Puppeteer-managed browser, package installs, ...), and a machine-facing + * total, so resource consumption is visible in the service logs over time. + */ +export class ResourceStatsService implements Resource { + private readonly intervalMs: number + private readonly getTrackedProcesses: () => TrackedProcess[] + private timer?: NodeJS.Timeout + private sampling = false + private samplingPromise?: Promise + private lastCpuUsage = process.cpuUsage() + private lastCpuSampleAt = process.hrtime.bigint() + private treeSamplingBroken = false + + constructor({ intervalMs, getTrackedProcesses }: ResourceStatsServiceOptions) { + this.intervalMs = intervalMs + this.getTrackedProcesses = getTrackedProcesses + } + + public async init(): Promise { + if (!Number.isFinite(this.intervalMs) || this.intervalMs <= 0) { + log({ + level: 'info', + msg: `[resource-stats] disabled (RESOURCE_STATS_INTERVAL_MS must be a positive number of ms; got ${this.intervalMs})` + }) + return + } + this.lastCpuUsage = process.cpuUsage() + this.lastCpuSampleAt = process.hrtime.bigint() + this.timer = setInterval(() => { + // Skip this tick if the previous sample is still running (e.g. a slow or + // blocked `ps`), so overlapping samples never pile up. + if (this.sampling) { + return + } + this.sampling = true + this.samplingPromise = this.logSample() + .catch((error) => { + log({ level: 'debug', msg: '[resource-stats] sampling failed', error }) + }) + .finally(() => { + this.sampling = false + this.samplingPromise = undefined + }) + }, this.intervalMs) + // Never keep the process alive just to report stats + this.timer.unref() + log({ level: 'info', msg: `[resource-stats] sampling every ${this.intervalMs}ms` }) + } + + public async stop(): Promise { + if (this.timer) { + clearInterval(this.timer) + this.timer = undefined + } + // Wait for any in-flight sample to finish so no sampling (spawning `ps`, + // logging) overlaps the teardown of the services this samples. + if (this.samplingPromise) { + await this.samplingPromise + } + } + + private sampleSelfCpuPercent(): number { + const now = process.hrtime.bigint() + const elapsedMicros = Number(now - this.lastCpuSampleAt) / 1000 + const usage = process.cpuUsage(this.lastCpuUsage) + this.lastCpuUsage = process.cpuUsage() + this.lastCpuSampleAt = now + if (elapsedMicros <= 0) { + return 0 + } + return ((usage.user + usage.system) / elapsedMicros) * 100 + } + + private async logSample(): Promise { + const memory = process.memoryUsage() + const cpuPercent = this.sampleSelfCpuPercent() + log({ + level: 'info', + msg: + `[resource-stats] self pid=${process.pid} rss=${formatMebibytes(memory.rss)} ` + + `heapUsed=${formatMebibytes(memory.heapUsed)} heapTotal=${formatMebibytes(memory.heapTotal)} ` + + `external=${formatMebibytes(memory.external)} cpu=${cpuPercent.toFixed(1)}%` + }) + + let subtrees: SubtreeSample[] + try { + const tree = await sampleProcessTree() + subtrees = aggregateDirectChildSubtrees(tree, process.pid) + this.treeSamplingBroken = false + } catch (error) { + // Warn once, then stay quiet: without `ps` only self stats are available + if (!this.treeSamplingBroken) { + this.treeSamplingBroken = true + log({ level: 'warn', msg: '[resource-stats] cannot sample child processes (is `ps` available?)', error }) + } + return + } + + const nameByPid = new Map() + for (const tracked of this.getTrackedProcesses()) { + nameByPid.set(tracked.pid, tracked.name) + } + + for (const subtree of subtrees) { + const name = nameByPid.get(subtree.rootPid) ?? shortenProcessLabel(subtree.command) + log({ + level: 'info', + msg: + `[resource-stats] child name=${name} pid=${subtree.rootPid} ` + + `rss=${formatMebibytes(subtree.rssBytes)} cpu=${subtree.cpuPercent.toFixed(1)}% procs=${subtree.processCount}` + }) + } + + const childRss = subtrees.reduce((sum, subtree) => sum + subtree.rssBytes, 0) + const childProcs = subtrees.reduce((sum, subtree) => sum + subtree.processCount, 0) + log({ + level: 'info', + msg: + `[resource-stats] total rss=${formatMebibytes(memory.rss + childRss)} ` + + `procs=${childProcs + 1} (self + ${childProcs} descendants)` + }) + } +} diff --git a/src/utils/resourceStats.ts b/src/utils/resourceStats.ts new file mode 100644 index 0000000..4d65380 --- /dev/null +++ b/src/utils/resourceStats.ts @@ -0,0 +1,217 @@ +import { execFile } from 'node:child_process' +import { basename } from 'node:path' +import { StdioClientTransport } from '@modelcontextprotocol/sdk/client/stdio.js' +import { Transport } from '@modelcontextprotocol/sdk/shared/transport.js' + +export interface ProcessSample { + pid: number + ppid: number + rssBytes: number + cpuPercent: number + command: string +} + +export interface SubtreeSample { + rootPid: number + command: string + rssBytes: number + cpuPercent: number + processCount: number +} + +/** + * Extracts the OS process id from a stdio client transport. + * + * @modelcontextprotocol/sdk 1.13.0 does not expose a public accessor for the + * spawned child process (verified against dist/cjs/client/stdio.d.ts: `_process` + * is private and no `pid` getter exists). This helper is the single, isolated + * place that reaches into that private field, and it validates the value at + * runtime so an SDK upgrade that changes internals degrades to `undefined` + * instead of misbehaving. + */ +export function getStdioTransportPid(transport: Transport): number | undefined { + if (!(transport instanceof StdioClientTransport)) { + return undefined + } + const internals = transport as unknown as { _process?: { pid?: unknown } } + const pid = internals._process?.pid + return typeof pid === 'number' && Number.isInteger(pid) && pid > 0 ? pid : undefined +} + +/** + * Parses `ps -eo pid=,ppid=,rss=,pcpu=,command=` output. + * rss is reported by ps in kilobytes on both Linux (procps) and macOS (BSD ps). + * `command` is the full command line (may contain spaces), so it is everything + * after the fourth column. Use {@link shortenProcessLabel} to derive a concise, + * argument-free label for logging. + */ +export function parsePsOutput(output: string): ProcessSample[] { + const samples: ProcessSample[] = [] + for (const line of output.split('\n')) { + const trimmed = line.trim() + if (!trimmed) continue + const match = trimmed.match(/^(\d+)\s+(\d+)\s+(\d+)\s+([\d.]+)\s+(.+)$/) + if (!match) continue + samples.push({ + pid: Number(match[1]), + ppid: Number(match[2]), + rssBytes: Number(match[3]) * 1024, + cpuPercent: Number(match[4]), + command: match[5].trim() + }) + } + return samples +} + +/** + * Groups every process descending from rootPid into one aggregate per direct + * child of rootPid: the child's own usage plus everything below it. This keeps + * one log line per spawned server (or browser) even when it forks helpers of + * its own (e.g. Chromium renderer processes). + */ +export function aggregateDirectChildSubtrees(samples: ProcessSample[], rootPid: number): SubtreeSample[] { + const childrenByPpid = new Map() + for (const sample of samples) { + const siblings = childrenByPpid.get(sample.ppid) + if (siblings) { + siblings.push(sample) + } else { + childrenByPpid.set(sample.ppid, [sample]) + } + } + + const collectSubtree = (pid: number, into: ProcessSample[], seen: Set): void => { + for (const child of childrenByPpid.get(pid) ?? []) { + if (seen.has(child.pid)) continue + seen.add(child.pid) + into.push(child) + collectSubtree(child.pid, into, seen) + } + } + + const subtrees: SubtreeSample[] = [] + for (const directChild of childrenByPpid.get(rootPid) ?? []) { + const members: ProcessSample[] = [directChild] + collectSubtree(directChild.pid, members, new Set([directChild.pid])) + subtrees.push({ + rootPid: directChild.pid, + command: directChild.command, + rssBytes: members.reduce((sum, m) => sum + m.rssBytes, 0), + cpuPercent: members.reduce((sum, m) => sum + m.cpuPercent, 0), + processCount: members.length + }) + } + return subtrees +} + +const RUNTIME_EXECUTABLES = new Set(['node', 'nodejs', 'python', 'python3', 'bun', 'deno', 'ts-node']) +const SCRIPT_FILE_PATTERN = /\.(m?[jt]s|cjs|py|sh)$/i +// Flags whose following token is inline code — the process has no script file and +// the value must never be surfaced (it can contain arbitrary code/secrets). +const EVAL_FLAGS = new Set(['-e', '--eval', '-p', '--print', '-c']) +// Flags whose following token is a value that is NOT the main script (a preloaded +// module, loader, etc.); the value must be skipped so the real script is found. +const VALUE_FLAGS = new Set([ + ...EVAL_FLAGS, + '-r', + '--require', + '--import', + '--loader', + '--experimental-loader' +]) + +/** + * True when a token looks like a path to a script file rather than an arbitrary + * argument value: it either contains a path separator or ends in a known script + * extension. + */ +function looksLikeScriptPath(token: string): boolean { + return token.includes('/') || SCRIPT_FILE_PATTERN.test(token) +} + +/** + * Derives a concise, human-readable label from a full `ps command=` string for + * use in log lines. Returns the executable basename; for language runtimes + * (node/python/...) it also appends the basename of the main script so that + * otherwise identical `node` processes stay distinguishable. + * + * Only executable and genuine script-file basenames are surfaced. Flags and + * their values are parsed so that inline code (`node -e `, `python -c + * `) and preloaded-module values (`node -r `) are never mistaken + * for the script — this keeps arbitrary/sensitive command-line values out of the + * logs even when they contain path separators or script-like extensions. + */ +export function shortenProcessLabel(command: string): string { + const trimmed = command.trim() + if (!trimmed) return 'unknown' + const tokens = trimmed.split(/\s+/) + const exe = basename(tokens[0]) + if (!RUNTIME_EXECUTABLES.has(exe)) { + return exe + } + + let sawEvalFlag = false + let script: string | undefined + for (let i = 1; i < tokens.length; i++) { + const token = tokens[i] + if (token.startsWith('-')) { + const name = token.includes('=') ? token.slice(0, token.indexOf('=')) : token + if (EVAL_FLAGS.has(name)) { + sawEvalFlag = true + } + // A separate value token (no inline `=`) belongs to this flag, not the script. + if (!token.includes('=') && VALUE_FLAGS.has(name)) { + i++ + } + continue + } + // First positional token — the main script/entrypoint. + script = token + break + } + + // Eval/print invocations have no script file; never surface the inline code. + if (!sawEvalFlag && script && looksLikeScriptPath(script)) { + return `${exe} ${basename(script)}` + } + return exe +} + +// Bounds a single `ps` invocation. A normal `ps -eo` completes in milliseconds; +// this ceiling exists only so a wedged `ps` cannot stall sampling (and, via +// stop() awaiting the in-flight sample, shutdown) indefinitely. +const PS_TIMEOUT_MS = 5000 + +/** + * Samples the full OS process tree with a single `ps` invocation. + * Requires the `ps` utility (procps on Linux images — installed in the + * Dockerfile; present by default on macOS/BSD). + * + * @throws Error when `ps` is unavailable, times out ({@link PS_TIMEOUT_MS}), or + * exits abnormally. + */ +export function sampleProcessTree(): Promise { + return new Promise((resolve, reject) => { + execFile( + 'ps', + ['-eo', 'pid=,ppid=,rss=,pcpu=,command='], + { maxBuffer: 4 * 1024 * 1024, timeout: PS_TIMEOUT_MS }, + (error, stdout) => { + if (error) { + reject(error) + return + } + resolve(parsePsOutput(stdout)) + } + ) + }) +} + +/** + * Formats a byte count as base-1024 mebibytes with a `MiB` suffix (matching the + * unit the math actually produces, so operators can compare against `docker + * stats` / `top` without a base-1000 vs base-1024 mismatch). + */ +export function formatMebibytes(bytes: number): string { + return `${(bytes / (1024 * 1024)).toFixed(1)}MiB` +} diff --git a/test/resource-stats-service.spec.ts b/test/resource-stats-service.spec.ts new file mode 100644 index 0000000..1d25691 --- /dev/null +++ b/test/resource-stats-service.spec.ts @@ -0,0 +1,86 @@ +// Verifies ResourceStatsService.stop() waits for an in-flight sample so no +// sampling overlaps service teardown. sampleProcessTree is mocked so the test +// controls exactly when a sample completes. +jest.mock('../src/utils/resourceStats', () => { + const actual = jest.requireActual('../src/utils/resourceStats') + return { ...actual, sampleProcessTree: jest.fn() } +}) + +import { sampleProcessTree } from '../src/utils/resourceStats' +import { ResourceStatsService } from '../src/services/resourceStats' + +const mockSampleProcessTree = sampleProcessTree as jest.Mock + +describe('ResourceStatsService.stop', () => { + beforeEach(() => { + jest.useFakeTimers() + mockSampleProcessTree.mockReset() + mockSampleProcessTree.mockResolvedValue([]) + }) + + afterEach(() => { + jest.clearAllTimers() + jest.useRealTimers() + }) + + const flushMicrotasks = async () => { + for (let i = 0; i < 5; i++) { + await Promise.resolve() + } + } + + it('waits for an in-flight sample to finish before resolving', async () => { + let resolveSample: (value: unknown[]) => void = () => undefined + mockSampleProcessTree.mockReturnValue( + new Promise((resolve) => { + resolveSample = resolve + }) + ) + + const service = new ResourceStatsService({ intervalMs: 1000, getTrackedProcesses: () => [] }) + await service.init() + + // Fire one tick: logSample() starts and suspends on the pending sampleProcessTree(). + jest.advanceTimersByTime(1000) + await flushMicrotasks() + expect(mockSampleProcessTree).toHaveBeenCalledTimes(1) + + let stopped = false + const stopPromise = service.stop().then(() => { + stopped = true + }) + + // stop() must not resolve while the sample is still in flight. + await flushMicrotasks() + expect(stopped).toBe(false) + + // Completing the sample lets stop() resolve. + resolveSample([]) + await stopPromise + expect(stopped).toBe(true) + }) + + it('resolves immediately when no sample is in flight', async () => { + const service = new ResourceStatsService({ intervalMs: 1000, getTrackedProcesses: () => [] }) + await service.init() + await expect(service.stop()).resolves.toBeUndefined() + expect(mockSampleProcessTree).not.toHaveBeenCalled() + }) + + it('is disabled (no timer, never samples) when the interval is 0', async () => { + const service = new ResourceStatsService({ intervalMs: 0, getTrackedProcesses: () => [] }) + await service.init() + jest.advanceTimersByTime(60000) + await flushMicrotasks() + expect(mockSampleProcessTree).not.toHaveBeenCalled() + await expect(service.stop()).resolves.toBeUndefined() + }) + + it('is disabled when the interval is not a finite number', async () => { + const service = new ResourceStatsService({ intervalMs: NaN, getTrackedProcesses: () => [] }) + await service.init() + jest.advanceTimersByTime(60000) + await flushMicrotasks() + expect(mockSampleProcessTree).not.toHaveBeenCalled() + }) +}) diff --git a/test/resource-stats.spec.ts b/test/resource-stats.spec.ts new file mode 100644 index 0000000..497c23e --- /dev/null +++ b/test/resource-stats.spec.ts @@ -0,0 +1,193 @@ +import { StdioClientTransport } from '@modelcontextprotocol/sdk/client/stdio.js' +import { + aggregateDirectChildSubtrees, + formatMebibytes, + getStdioTransportPid, + parsePsOutput, + sampleProcessTree, + shortenProcessLabel +} from '../src/utils/resourceStats' + +describe('parsePsOutput', () => { + it('parses pid, ppid, rss (KB -> bytes), pcpu and command', () => { + const output = [ + ' 1 0 1024 0.5 node', + ' 4321 1 20480 12.3 /usr/local/bin/node dist/index.js', + '' + ].join('\n') + + const samples = parsePsOutput(output) + expect(samples).toEqual([ + { pid: 1, ppid: 0, rssBytes: 1024 * 1024, cpuPercent: 0.5, command: 'node' }, + { pid: 4321, ppid: 1, rssBytes: 20480 * 1024, cpuPercent: 12.3, command: '/usr/local/bin/node dist/index.js' } + ]) + }) + + it('preserves spaces inside the command column', () => { + const samples = parsePsOutput('99 1 512 0.0 Google Chrome Helper (Renderer)') + expect(samples).toHaveLength(1) + expect(samples[0].command).toBe('Google Chrome Helper (Renderer)') + }) + + it('skips malformed lines', () => { + const samples = parsePsOutput('not a process line\n\n12 abc 100 0.0 zsh') + expect(samples).toEqual([]) + }) +}) + +describe('aggregateDirectChildSubtrees', () => { + const sample = (pid: number, ppid: number, rssKb: number, cpu: number, command: string) => ({ + pid, + ppid, + rssBytes: rssKb * 1024, + cpuPercent: cpu, + command + }) + + it('aggregates each direct child with all of its descendants', () => { + const root = 100 + const tree = [ + sample(root, 1, 400_000, 2.0, 'node dist/index.js'), // self (excluded from subtrees) + sample(200, root, 50_000, 0.5, 'node mcp-github'), // stdio server + sample(300, root, 300_000, 3.0, 'chrome'), // browser + sample(301, 300, 150_000, 1.5, 'chrome'), // renderer under browser + sample(302, 301, 10_000, 0.1, 'chrome'), // nested helper + sample(999, 1, 999_000, 9.9, 'unrelated') // not in our tree + ] + + const subtrees = aggregateDirectChildSubtrees(tree, root) + expect(subtrees).toHaveLength(2) + + const server = subtrees.find((s) => s.rootPid === 200) + expect(server).toEqual({ + rootPid: 200, + command: 'node mcp-github', + rssBytes: 50_000 * 1024, + cpuPercent: 0.5, + processCount: 1 + }) + + const browser = subtrees.find((s) => s.rootPid === 300) + expect(browser).toBeDefined() + expect(browser!.rssBytes).toBe((300_000 + 150_000 + 10_000) * 1024) + expect(browser!.cpuPercent).toBeCloseTo(4.6) + expect(browser!.processCount).toBe(3) + }) + + it('returns an empty list when the root has no children', () => { + expect(aggregateDirectChildSubtrees([], 100)).toEqual([]) + }) + + it('does not loop forever on cyclic ppid data', () => { + const tree = [sample(200, 100, 1, 0, 'a'), sample(201, 200, 1, 0, 'b'), sample(200, 201, 1, 0, 'a-again')] + const subtrees = aggregateDirectChildSubtrees(tree, 100) + expect(subtrees).toHaveLength(1) + }) +}) + +describe('getStdioTransportPid', () => { + it('returns undefined for a transport that has not been started', () => { + const transport = new StdioClientTransport({ command: 'true' }) + expect(getStdioTransportPid(transport)).toBeUndefined() + }) + + it('returns the pid of a started transport process', () => { + const transport = Object.create(StdioClientTransport.prototype) as StdioClientTransport + ;(transport as unknown as { _process: { pid: number } })._process = { pid: 4321 } + expect(getStdioTransportPid(transport)).toBe(4321) + }) + + it('rejects non-numeric pid values', () => { + const transport = Object.create(StdioClientTransport.prototype) as StdioClientTransport + ;(transport as unknown as { _process: { pid: unknown } })._process = { pid: 'nope' } + expect(getStdioTransportPid(transport)).toBeUndefined() + }) + + it('returns undefined for non-stdio transports', () => { + const fake = { start: async () => undefined, send: async () => undefined, close: async () => undefined } + expect(getStdioTransportPid(fake)).toBeUndefined() + }) +}) + +describe('sampleProcessTree', () => { + it('samples the live process tree and includes this process', async () => { + const samples = await sampleProcessTree() + expect(samples.length).toBeGreaterThan(0) + expect(samples.some((s) => s.pid === process.pid)).toBe(true) + }) +}) + +describe('formatMebibytes', () => { + it('formats bytes as base-1024 MiB with one decimal', () => { + expect(formatMebibytes(412.3 * 1024 * 1024)).toBe('412.3MiB') + expect(formatMebibytes(0)).toBe('0.0MiB') + }) +}) + +describe('shortenProcessLabel', () => { + it('returns the executable basename for a non-runtime command', () => { + expect(shortenProcessLabel('/opt/google/chrome/chrome --type=renderer --no-sandbox')).toBe('chrome') + expect(shortenProcessLabel('/usr/lib/chromium/chromium --headless')).toBe('chromium') + }) + + it('appends the script basename for language runtimes so siblings stay distinct', () => { + expect( + shortenProcessLabel('node ./packages/missionsquad-mcp-rss/node_modules/@missionsquad/mcp-rss/dist/index.js') + ).toBe('node index.js') + expect(shortenProcessLabel('/usr/local/bin/node /app/dist/server.js --port 8080')).toBe('node server.js') + expect(shortenProcessLabel('python3 /app/tools/worker.py')).toBe('python3 worker.py') + }) + + it('skips leading flags when locating the script token', () => { + expect(shortenProcessLabel('node --inspect --enable-source-maps /app/dist/index.js')).toBe('node index.js') + }) + + it('returns just the runtime when it has no script argument', () => { + expect(shortenProcessLabel('node')).toBe('node') + expect(shortenProcessLabel('node --version')).toBe('node') + }) + + it('does not leak trailing command-line arguments into the label', () => { + const label = shortenProcessLabel('node /app/dist/index.js --api-key=SUPER_SECRET --token abc123') + expect(label).toBe('node index.js') + expect(label).not.toContain('SUPER_SECRET') + expect(label).not.toContain('abc123') + }) + + it('does not treat inline eval/print code as a script (no arg leakage)', () => { + // The token after -e/-c is code, not a script path; it must never appear in the label. + expect(shortenProcessLabel('node -e SUPER_SECRET_CODE')).toBe('node') + expect(shortenProcessLabel('node --eval SUPER_SECRET_CODE')).toBe('node') + expect(shortenProcessLabel('python -c import os,sys;print(SECRET)')).toBe('python') + expect(shortenProcessLabel('node -p process.env.TOKEN')).toBe('node') + expect(shortenProcessLabel('node -e SUPER_SECRET_CODE')).not.toContain('SUPER_SECRET_CODE') + }) + + it('only appends a token that looks like a script path (has a separator or known extension)', () => { + // Bare non-file arg -> just the runtime. + expect(shortenProcessLabel('node server')).toBe('node') + // Recognizable script forms -> runtime + script basename. + expect(shortenProcessLabel('node worker.mjs')).toBe('node worker.mjs') + expect(shortenProcessLabel('ts-node ./src/index.ts')).toBe('ts-node index.ts') + }) + + it('skips flag values (e.g. -r/--require preloads) and selects the real script', () => { + expect(shortenProcessLabel('node -r ts-node/register /app/index.js')).toBe('node index.js') + expect(shortenProcessLabel('node --require ./setup.js /srv/main.js')).toBe('node main.js') + expect(shortenProcessLabel('node --import tsx /app/dist/server.js')).toBe('node server.js') + // The preloaded module name must not become the label. + expect(shortenProcessLabel('node -r ts-node/register /app/index.js')).not.toContain('register') + }) + + it('never surfaces inline eval code even when it contains a slash or extension', () => { + expect(shortenProcessLabel('node -e require("/etc/passwd")')).toBe('node') + expect(shortenProcessLabel('node -e require("/etc/passwd")')).not.toContain('passwd') + expect(shortenProcessLabel('python -c open("/secret/key.pem")')).toBe('python') + expect(shortenProcessLabel('python -c open("/secret/key.pem")')).not.toContain('key.pem') + }) + + it('falls back to "unknown" for an empty command', () => { + expect(shortenProcessLabel('')).toBe('unknown') + expect(shortenProcessLabel(' ')).toBe('unknown') + }) +}) diff --git a/test/searxng-lifecycle.spec.ts b/test/searxng-lifecycle.spec.ts new file mode 100644 index 0000000..1b4d3c6 --- /dev/null +++ b/test/searxng-lifecycle.spec.ts @@ -0,0 +1,101 @@ +// Verifies BuiltInSearxngServer's Puppeteer lifecycle guards without launching a +// real browser: @missionsquad/puppeteer-scraper is fully mocked. +const mockInit = jest.fn() +const mockCloseBrowser = jest.fn() +const mockScraperCtor = jest.fn() + +jest.mock('@missionsquad/puppeteer-scraper', () => ({ + PuppeteerScraper: jest.fn().mockImplementation((...args: unknown[]) => { + mockScraperCtor(...args) + return { init: mockInit, closeBrowser: mockCloseBrowser } + }) +})) + +import { BuiltInSearxngServer } from '../src/builtin-servers/servers/searxng' + +type Internals = { + initializePuppeteerWithRetries(retryCount: number): Promise + scraper: unknown + scraperReady: boolean +} +const internals = (server: BuiltInSearxngServer) => server as unknown as Internals + +describe('BuiltInSearxngServer Puppeteer lifecycle', () => { + beforeEach(() => { + jest.useFakeTimers() + mockScraperCtor.mockClear() + mockInit.mockReset() + mockCloseBrowser.mockReset() + mockCloseBrowser.mockResolvedValue(undefined) + }) + + afterEach(() => { + jest.clearAllTimers() + jest.useRealTimers() + }) + + it('schedules a retry timer after a failed init attempt', async () => { + mockInit.mockRejectedValue(new Error('init failed')) + const server = new BuiltInSearxngServer() + + await internals(server).initializePuppeteerWithRetries(0) + + expect(mockScraperCtor).toHaveBeenCalledTimes(1) + // A pending retry is scheduled (not yet fired). + expect(jest.getTimerCount()).toBe(1) + }) + + it('closes a partially-initialized browser when init fails, before retrying', async () => { + mockInit.mockRejectedValue(new Error('init failed')) + const server = new BuiltInSearxngServer() + + await internals(server).initializePuppeteerWithRetries(0) + + // The failed attempt's browser is closed in the catch so repeated failures + // cannot leak Chromium processes. + expect(mockScraperCtor).toHaveBeenCalledTimes(1) + expect(mockCloseBrowser).toHaveBeenCalledTimes(1) + }) + + it('cancels the pending retry on stop() and ignores any later retry attempt', async () => { + mockInit.mockRejectedValue(new Error('init failed')) + const server = new BuiltInSearxngServer() + + await internals(server).initializePuppeteerWithRetries(0) + expect(jest.getTimerCount()).toBe(1) + + await server.stop() + // The scheduled retry timer is cleared by stop(). + expect(jest.getTimerCount()).toBe(0) + + // Even if a retry callback still fired after stop(), the stopped guard makes it + // a no-op: no new scraper/browser is ever constructed. + mockScraperCtor.mockClear() + await internals(server).initializePuppeteerWithRetries(1) + expect(mockScraperCtor).not.toHaveBeenCalled() + }) + + it('closes a browser created during init if stop() ran mid-init (no leak, not published)', async () => { + let resolveInit: () => void = () => undefined + mockInit.mockImplementation( + () => + new Promise((resolve) => { + resolveInit = resolve + }) + ) + const server = new BuiltInSearxngServer() + + const pending = internals(server).initializePuppeteerWithRetries(0) + expect(mockScraperCtor).toHaveBeenCalledTimes(1) + + // stop() runs while init() is still in flight; the scraper is not yet published. + await server.stop() + resolveInit() + await pending + + // The freshly created browser is closed by the mid-init guard, not published. + expect(mockCloseBrowser).toHaveBeenCalledTimes(1) + expect(internals(server).scraper).toBeNull() + expect(internals(server).scraperReady).toBe(false) + }) +}) diff --git a/test/server-logs-cap.spec.ts b/test/server-logs-cap.spec.ts new file mode 100644 index 0000000..a823339 --- /dev/null +++ b/test/server-logs-cap.spec.ts @@ -0,0 +1,83 @@ +import { appendServerLog, MAX_SERVER_LOG_LINES } from '../src/services/mcp' + +describe('appendServerLog', () => { + it('exposes a cap of 500 lines', () => { + expect(MAX_SERVER_LOG_LINES).toBe(500) + }) + + it('appends normally when under the cap', () => { + const logs: string[] = [] + appendServerLog(logs, 'a') + appendServerLog(logs, 'b') + appendServerLog(logs, 'c') + expect(logs).toEqual(['a', 'b', 'c']) + expect(logs.length).toBe(3) + }) + + it('keeps exactly the last 500 entries in order once the cap is exceeded (oldest dropped)', () => { + const logs: string[] = [] + const total = MAX_SERVER_LOG_LINES + 250 // 750 entries pushed + for (let i = 0; i < total; i++) { + appendServerLog(logs, `line-${i}`) + } + expect(logs.length).toBe(MAX_SERVER_LOG_LINES) + // The oldest 250 (line-0 .. line-249) must have been dropped. + expect(logs[0]).toBe(`line-${total - MAX_SERVER_LOG_LINES}`) // line-250 + expect(logs[logs.length - 1]).toBe(`line-${total - 1}`) // line-749 + // Contents are the last 500 lines in insertion order. + const expected: string[] = [] + for (let i = total - MAX_SERVER_LOG_LINES; i < total; i++) { + expected.push(`line-${i}`) + } + expect(logs).toEqual(expected) + }) + + it('tolerates an undefined logs array without throwing', () => { + expect(() => appendServerLog(undefined, 'ignored')).not.toThrow() + }) + + it('maintains a steady-state length of 500 when called repeatedly one-at-a-time past the cap', () => { + const logs: string[] = [] + // Fill exactly to the cap. + for (let i = 0; i < MAX_SERVER_LOG_LINES; i++) { + appendServerLog(logs, `seed-${i}`) + } + expect(logs.length).toBe(MAX_SERVER_LOG_LINES) + + // Each subsequent single append must keep the length pinned at the cap, + // dropping exactly one oldest line and retaining the newest. + for (let i = 0; i < 1000; i++) { + appendServerLog(logs, `stream-${i}`) + expect(logs.length).toBe(MAX_SERVER_LOG_LINES) + expect(logs[logs.length - 1]).toBe(`stream-${i}`) + } + + // After 1000 single appends past a full buffer, every seed line is gone + // and only the last 500 streamed lines remain, in order. + const expected: string[] = [] + for (let i = 1000 - MAX_SERVER_LOG_LINES; i < 1000; i++) { + expected.push(`stream-${i}`) + } + expect(logs).toEqual(expected) + }) + + it('never exceeds the cap length under sustained high-volume appends (bounds retained lines)', () => { + // Mirrors the exact scenario the cap guards against: a chatty stdio server + // streaming stderr lines for the whole process lifetime. The array length must + // never exceed the cap regardless of how many lines are pushed — which is what + // bounds the retained-line count (and therefore this buffer's memory). + const logs: string[] = [] + const VOLUME = 100_000 + let peak = 0 + for (let i = 0; i < VOLUME; i++) { + appendServerLog(logs, `evt-${i}`) + if (logs.length > peak) peak = logs.length + } + // Peak length never exceeded the cap at any point during the run. + expect(peak).toBe(MAX_SERVER_LOG_LINES) + expect(logs.length).toBe(MAX_SERVER_LOG_LINES) + // Only the most recent 500 of the 100k lines survive, in order. + expect(logs[0]).toBe(`evt-${VOLUME - MAX_SERVER_LOG_LINES}`) + expect(logs[logs.length - 1]).toBe(`evt-${VOLUME - 1}`) + }) +}) diff --git a/yarn.lock b/yarn.lock index 9678391..5f2ea4f 100644 --- a/yarn.lock +++ b/yarn.lock @@ -1195,10 +1195,10 @@ dependencies: nanoid "3.3.10" -"@missionsquad/puppeteer-scraper@^1.1.1": - version "1.1.1" - resolved "https://registry.yarnpkg.com/@missionsquad/puppeteer-scraper/-/puppeteer-scraper-1.1.1.tgz#55165637a4530c22a28dc1112c6509257ec47f10" - integrity sha512-IfmqCt65ucxvZmHKkeUo61EhZE+3n6nt4ZTTZ0OsxemEdOmWNcio4cq5pvOinDHXYackA2Z7BLogChiFBOMPYA== +"@missionsquad/puppeteer-scraper@^1.1.2": + version "1.1.2" + resolved "https://registry.yarnpkg.com/@missionsquad/puppeteer-scraper/-/puppeteer-scraper-1.1.2.tgz#8569a5cade6715b5d4369df898975b8ebdec65d9" + integrity sha512-EUXDT3S5p+SIVEOqSGlF8zivNH2BsxagrExYamSeS9/e0NNKYCglizhSll+q+Pqn5LAj4ZK2goPDVKgMqQ7SGg== dependencies: "@missionsquad/common" "^1.1.0" browsers "^1.0.2"