diff --git a/.qwen/design/2026-05-21-memory-pressure-monitor-design.md b/.qwen/design/2026-05-21-memory-pressure-monitor-design.md new file mode 100644 index 00000000000..13da8c6e346 --- /dev/null +++ b/.qwen/design/2026-05-21-memory-pressure-monitor-design.md @@ -0,0 +1,136 @@ +--- +title: 'Memory Pressure Monitor' +date: '2026-05-21' +status: 'implemented' +--- + +# Memory Pressure Monitor + +## Problem + +Long-running Qwen Code sessions can accumulate memory through large tool +results, repeated file reads, chat history, and native/external allocations. +Before this change, the core package had diagnostics and session-reset cleanup, +but no runtime response when memory pressure rises during normal tool +execution. + +The highest-value cache-specific gap is `FileReadCache`: it already has a +bounded FIFO size, but it did not have a time-based eviction path. That means a +session can retain inactive file-read metadata until the hard entry limit is +hit, even when the process is under memory pressure. + +## Goals + +- Add a low-overhead memory pressure check after tool execution. +- Prefer surgical cleanup before destructive cleanup. +- Respect container memory limits when cgroup v2 or cgroup v1 memory limit + files are available. +- React to V8 heap pressure before JavaScript heap OOM on high-memory hosts. +- Keep subagent/scoped `Config` instances isolated from parent session cleanup. +- Make behavior configurable through environment variables without adding a new + user-facing settings surface. + +## Non-Goals + +- Do not add a background polling loop. +- Do not make explicit GC the default; it only runs when enabled and Node was + started with `--expose-gc`. +- Do not change prior-read enforcement semantics. Cache eviction can remove old + metadata, but it must not weaken stale-file checks for retained entries. + +## Design + +`Config.initialize()` creates one `MemoryPressureMonitor` per initialized +`Config`. `getMemoryPressureMonitor()` mirrors the existing `getFileReadCache()` +Object.create isolation pattern: when a child config is created through +prototype delegation, the getter lazily installs an own monitor bound to that +child config. + +`CoreToolScheduler.executeSingleToolCall()` calls `scheduleCheck()` in its +`finally` block after ending the tool span. `scheduleCheck()` coalesces multiple +calls in the same event-loop turn with `queueMicrotask`, so concurrent read-like +tool batches do not run one memory check per tool result. + +The monitor uses the stronger of two pressure signals: + +- RSS divided by an effective process memory limit. Prefer cgroup v2 + `/sys/fs/cgroup/memory.max` when it is a finite positive value; fall back to + cgroup v1 `/sys/fs/cgroup/memory/memory.limit_in_bytes`, then to + `os.totalmem()` otherwise. cgroup v1's huge "unlimited" sentinel values are + ignored. +- V8 `heapUsed` divided by `getHeapStatistics().heap_size_limit`. + +Using both signals matters because containers usually fail by RSS/cgroup limit, +while local high-memory machines can hit V8 heap OOM long before RSS is a large +fraction of total system memory. + +Default thresholds are intentionally conservative enough to react before the OS +or container OOM killer does: + +- `softPressureRatio = 0.50` +- `hardPressureRatio = 0.65` +- `criticalRatio = 0.80` +- `cleanupCooldownMs = 5000` +- `enableExplicitGC = false` + +Environment overrides: + +- `QWEN_MEMORY_PRESSURE_SOFT` +- `QWEN_MEMORY_PRESSURE_HARD` +- `QWEN_MEMORY_PRESSURE_CRITICAL` +- `QWEN_MEMORY_ENABLE_GC=1` + +Invalid ratios fall back to defaults. Valid ratios must be ordered as +`soft < hard < critical`, with a lower soft bound of `0.3` and an upper +critical bound of `0.98`. Ratio env vars are parsed strictly with `Number()`, +so values such as `0.8extra` are rejected instead of partially accepted. +Invalid memory-pressure env configuration writes a visible warning to stderr +and to the debug log before falling back to defaults. + +## Cleanup Policy + +Pressure levels map to increasingly strong cleanup: + +- `soft`: evict stale `FileReadCache` entries not accessed in 60 minutes. +- `hard`: evict cache entries not accessed in 30 minutes. +- `critical`: clear the file-read cache and optionally trigger `global.gc()`. + +The monitor intentionally does not force chat compaction. Compaction can call +the model backend and rewrite active chat state, so it should be triggered only +from a call site that can safely coordinate with the conversation loop. + +Cleanup is fire-and-forget from the scheduler, but the monitor guards cleanup +steps with `cleanupInProgress` and a cooldown timestamp. A higher-pressure +cleanup can bypass the cooldown and queue behind an in-progress lower-pressure +cleanup, so a `critical` check is not lost while a `soft` cleanup is finishing. +After successful cleanup it logs an RSS delta on `setImmediate()`, but RSS +movement is diagnostic only: V8 and libc may retain freed pages even when +JavaScript objects became collectible. Consecutive failures count cleanup-step +exceptions, not unchanged RSS, and the counter is reset on a new session. If +three successful cleanup attempts in a row free less than 1% RSS, the monitor +emits `memory-cleanup-ineffective` as a diagnostic signal without treating the +cleanup step itself as failed. + +## Test Coverage + +The implementation is covered by: + +- threshold validation tests; +- environment config parsing, fallback, visible warning, and explicit GC tests; +- pressure classification tests using mocked `process.memoryUsage()`; +- cgroup v2 `memory.max` and cgroup v1 `memory.limit_in_bytes` behavior; +- V8 heap limit behavior; +- `scheduleCheck()` coalescing; +- scheduler integration that invokes `scheduleCheck()` after tool execution; +- soft and critical cleanup actions; +- cleanup failure accounting for thrown cleanup steps; +- cleanup listener exception isolation and ineffective-cleanup diagnostics; +- child `Config` monitor isolation through `Object.create`; +- `FileReadCache.evictNotAccessedSince()` behavior. + +## Risks And Tradeoffs + +- RSS can stay flat after cleanup because V8 or libc may retain freed memory. + RSS deltas are logged, but unchanged RSS does not count as a cleanup failure. +- Time-based file-read cache eviction may reduce fast-path hits for old files, + but it preserves recently active entries and only runs under memory pressure. diff --git a/packages/core/src/config/config.test.ts b/packages/core/src/config/config.test.ts index a9badd5878d..baad3955e4b 100644 --- a/packages/core/src/config/config.test.ts +++ b/packages/core/src/config/config.test.ts @@ -16,6 +16,7 @@ import { } from './config.js'; import { Storage } from './storage.js'; import * as fs from 'node:fs'; +import * as os from 'node:os'; import * as path from 'node:path'; import { setGeminiMdFilename as mockSetGeminiMdFilename } from '../memory/const.js'; import { @@ -288,6 +289,13 @@ vi.mock('../ide/ide-client.js', () => ({ import { BaseLlmClient } from '../core/baseLlmClient.js'; +const MEMORY_PRESSURE_ENV_KEYS = [ + 'QWEN_MEMORY_PRESSURE_SOFT', + 'QWEN_MEMORY_PRESSURE_HARD', + 'QWEN_MEMORY_PRESSURE_CRITICAL', + 'QWEN_MEMORY_ENABLE_GC', +]; + vi.mock('../core/baseLlmClient.js'); // Mock fireNotificationHook from toolHookTriggers vi.mock('../core/toolHookTriggers.js', () => ({ @@ -328,6 +336,9 @@ describe('Server Config (config.ts)', () => { beforeEach(() => { // Reset mocks if necessary vi.clearAllMocks(); + for (const envName of MEMORY_PRESSURE_ENV_KEYS) { + delete process.env[envName]; + } (fs.existsSync as Mock).mockReturnValue(true); (fs.readdirSync as Mock).mockReturnValue([]); (fs.statSync as Mock).mockReturnValue({ @@ -422,6 +433,192 @@ describe('Server Config (config.ts)', () => { }); }); + describe('MemoryPressureMonitor isolation', () => { + it('returns a distinct monitor for child Configs created via Object.create', async () => { + const parent = new Config(baseParams); + await parent.initialize({ skipGeminiInitialization: true }); + const child = Object.create(parent) as Config; + + const parentMonitor = parent.getMemoryPressureMonitor(); + const childMonitor = child.getMemoryPressureMonitor(); + + expect(parentMonitor).toBeDefined(); + expect(childMonitor).toBeDefined(); + expect(childMonitor).not.toBe(parentMonitor); + expect(child.getMemoryPressureMonitor()).toBe(childMonitor); + }); + + it('resets monitor cleanup state when starting a new session', async () => { + const config = new Config(baseParams); + await config.initialize({ skipGeminiInitialization: true }); + const monitor = config.getMemoryPressureMonitor(); + expect(monitor).toBeDefined(); + const resetSpy = vi.spyOn(monitor!, 'resetForNewSession'); + + config.startNewSession(); + + expect(resetSpy).toHaveBeenCalledTimes(1); + }); + }); + + describe('MemoryPressure configuration environment', () => { + const restorers: Array<() => void> = []; + const originalEnv = new Map(); + + beforeEach(() => { + originalEnv.clear(); + for (const envName of MEMORY_PRESSURE_ENV_KEYS) { + originalEnv.set(envName, process.env[envName]); + delete process.env[envName]; + } + }); + + afterEach(() => { + while (restorers.length > 0) { + restorers.pop()?.(); + } + for (const [envName, value] of originalEnv) { + if (value === undefined) { + delete process.env[envName]; + } else { + process.env[envName] = value; + } + } + originalEnv.clear(); + }); + + function mockMemoryRatio(rssRatio: number, heapUsedBytes = 0): void { + const spy = vi.spyOn(process, 'memoryUsage').mockReturnValue({ + rss: Math.ceil(os.totalmem() * rssRatio), + heapTotal: 512 * 1024 * 1024, + heapUsed: heapUsedBytes, + external: 0, + arrayBuffers: 0, + }); + restorers.push(() => spy.mockRestore()); + } + + function mockStderrWrite(): Mock { + const spy = vi + .spyOn(process.stderr, 'write') + .mockImplementation(() => true); + restorers.push(() => spy.mockRestore()); + return spy as unknown as Mock; + } + + it('applies valid memory pressure env overrides', async () => { + process.env['QWEN_MEMORY_PRESSURE_SOFT'] = '0.3'; + process.env['QWEN_MEMORY_PRESSURE_HARD'] = '0.6'; + process.env['QWEN_MEMORY_PRESSURE_CRITICAL'] = '0.9'; + + const config = new Config(baseParams); + await config.initialize({ skipGeminiInitialization: true }); + mockMemoryRatio(0.35); + + expect(config.getMemoryPressureMonitor()?.getPressureLevel()).toBe( + 'soft', + ); + }); + + it('falls back to defaults and warns on strict env parse failures', async () => { + const stderrSpy = mockStderrWrite(); + process.env['QWEN_MEMORY_PRESSURE_SOFT'] = '0.3extra'; + process.env['QWEN_MEMORY_PRESSURE_HARD'] = '0.6'; + process.env['QWEN_MEMORY_PRESSURE_CRITICAL'] = '0.9'; + + const config = new Config(baseParams); + await config.initialize({ skipGeminiInitialization: true }); + mockMemoryRatio(0.35); + + expect(config.getMemoryPressureMonitor()?.getPressureLevel()).toBe( + 'normal', + ); + expect(stderrSpy).toHaveBeenCalledWith( + expect.stringContaining('Invalid memory pressure config'), + ); + }); + + it('falls back to defaults and warns on invalid threshold ordering', async () => { + const stderrSpy = mockStderrWrite(); + process.env['QWEN_MEMORY_PRESSURE_SOFT'] = '0.7'; + + const config = new Config(baseParams); + await config.initialize({ skipGeminiInitialization: true }); + + expect(config.getMemoryPressureMonitor()).toBeDefined(); + expect(stderrSpy).toHaveBeenCalledWith( + expect.stringContaining( + 'softPressureRatio must be < hardPressureRatio', + ), + ); + }); + + it.each(['NaN', 'Infinity', '0'])( + 'falls back to defaults for invalid soft threshold %s', + async (value) => { + const stderrSpy = mockStderrWrite(); + process.env['QWEN_MEMORY_PRESSURE_SOFT'] = value; + + const config = new Config(baseParams); + await config.initialize({ skipGeminiInitialization: true }); + mockMemoryRatio(0.35); + + expect(config.getMemoryPressureMonitor()?.getPressureLevel()).toBe( + 'normal', + ); + expect(stderrSpy).toHaveBeenCalledWith( + expect.stringContaining('Invalid memory pressure config'), + ); + }, + ); + + it('enables explicit GC when requested by env', async () => { + process.env['QWEN_MEMORY_ENABLE_GC'] = '1'; + const globalWithGc = global as typeof global & { gc?: () => void }; + const originalGc = globalWithGc.gc; + const gcSpy = vi.fn(); + Object.defineProperty(globalWithGc, 'gc', { + value: gcSpy, + configurable: true, + }); + restorers.push(() => { + if (originalGc) { + Object.defineProperty(globalWithGc, 'gc', { + value: originalGc, + configurable: true, + }); + } else { + delete globalWithGc.gc; + } + }); + + const config = new Config(baseParams); + await config.initialize({ skipGeminiInitialization: true }); + mockMemoryRatio(0.85); + + config.getMemoryPressureMonitor()?.performCheck(); + await Promise.resolve(); + + expect(gcSpy).toHaveBeenCalledTimes(1); + }); + + it('child Config monitors inherit the parent memory pressure config snapshot', async () => { + process.env['QWEN_MEMORY_PRESSURE_SOFT'] = '0.3'; + process.env['QWEN_MEMORY_PRESSURE_HARD'] = '0.6'; + process.env['QWEN_MEMORY_PRESSURE_CRITICAL'] = '0.9'; + const parent = new Config(baseParams); + await parent.initialize({ skipGeminiInitialization: true }); + + process.env['QWEN_MEMORY_PRESSURE_SOFT'] = '0.9'; + process.env['QWEN_MEMORY_PRESSURE_HARD'] = '0.95'; + process.env['QWEN_MEMORY_PRESSURE_CRITICAL'] = '0.97'; + const child = Object.create(parent) as Config; + mockMemoryRatio(0.35); + + expect(child.getMemoryPressureMonitor()?.getPressureLevel()).toBe('soft'); + }); + }); + describe('startNewSession', () => { it('clears the FileReadCache so a new session does not inherit prior reads', () => { // Regression guard: the file-read cache backs ReadFile's diff --git a/packages/core/src/config/config.ts b/packages/core/src/config/config.ts index 733f8bab3e8..953be9bfae8 100644 --- a/packages/core/src/config/config.ts +++ b/packages/core/src/config/config.ts @@ -48,6 +48,12 @@ import { GitService } from '../services/gitService.js'; import { GitWorktreeService } from '../services/gitWorktreeService.js'; import { cleanupStaleAgentWorktrees } from '../services/worktreeCleanup.js'; import { CronScheduler } from '../services/cronScheduler.js'; +import { + MemoryPressureMonitor, + DEFAULT_PRESSURE_CONFIG, + validateMemoryPressureConfig, + type MemoryPressureConfig, +} from '../services/memoryPressureMonitor.js'; // Tools — only lightweight imports; tool classes are lazy-loaded via dynamic import import { @@ -155,6 +161,7 @@ import { MemoryManager } from '../memory/manager.js'; import { CommitAttributionService } from '../services/commitAttribution.js'; const gitCoAuthorLogger = createDebugLogger('GIT_CO_AUTHOR'); +const memoryPressureConfigLogger = createDebugLogger('MEMORY_PRESSURE'); import { ModelsConfig, @@ -748,6 +755,53 @@ function normalizeConfigOutputFormat( } } +function loadMemoryPressureConfig(): MemoryPressureConfig { + const config: MemoryPressureConfig = { ...DEFAULT_PRESSURE_CONFIG }; + + try { + config.softPressureRatio = readMemoryPressureRatioEnv( + 'QWEN_MEMORY_PRESSURE_SOFT', + config.softPressureRatio, + ); + config.hardPressureRatio = readMemoryPressureRatioEnv( + 'QWEN_MEMORY_PRESSURE_HARD', + config.hardPressureRatio, + ); + config.criticalRatio = readMemoryPressureRatioEnv( + 'QWEN_MEMORY_PRESSURE_CRITICAL', + config.criticalRatio, + ); + + if (process.env['QWEN_MEMORY_ENABLE_GC'] === '1') { + config.enableExplicitGC = true; + } + + validateMemoryPressureConfig(config); + } catch (err) { + const fallbackMsg = + '[QWEN] WARNING: Invalid memory pressure config; using defaults. ' + + `Error: ${getErrorMessage(err)}`; + process.stderr.write(`${fallbackMsg}\n`); + memoryPressureConfigLogger.warn(fallbackMsg); + return { ...DEFAULT_PRESSURE_CONFIG }; + } + + return config; +} + +function readMemoryPressureRatioEnv(envName: string, fallback: number): number { + const raw = process.env[envName]; + if (!raw) { + return fallback; + } + + const parsed = Number(raw); + if (!Number.isFinite(parsed)) { + throw new Error(`${envName} must be a finite number`); + } + return parsed; +} + /** * Options for Config.initialize() */ @@ -789,6 +843,8 @@ export class Config { private pendingMcpBudgetCallback?: (event: McpBudgetEvent) => void; private promptRegistry!: PromptRegistry; private subagentManager!: SubagentManager; + private memoryPressureConfig?: MemoryPressureConfig; + private memoryPressureMonitor?: MemoryPressureMonitor; private readonly backgroundTaskRegistry = new BackgroundTaskRegistry(); private readonly monitorRegistry = new MonitorRegistry(); private backgroundAgentResumeService?: BackgroundAgentResumeService; @@ -1369,6 +1425,12 @@ export class Config { } this.debugLogger.debug('Skill manager initialized'); + this.memoryPressureConfig = loadMemoryPressureConfig(); + this.memoryPressureMonitor = new MemoryPressureMonitor( + this, + this.memoryPressureConfig, + ); + this.permissionManager = new PermissionManager(this); this.permissionManager.initialize(); this.debugLogger.debug('Permission manager initialized'); @@ -1864,6 +1926,7 @@ export class Config { // constructed via Object.create — those should clear their own // cache, not the parent's. this.getFileReadCache().clear(); + this.getMemoryPressureMonitor()?.resetForNewSession(); this.fileHistoryService = undefined; refreshSessionContext(this.sessionId); // The commit-attribution singleton accumulates per-file AI edits @@ -2865,6 +2928,33 @@ export class Config { return this.geminiClient; } + /** + * Session-scoped memory pressure monitor. Child Configs created with + * `Object.create(parent)` inherit the parent's monitor through the prototype + * chain until this getter installs an own monitor backed by the inherited + * pressure config snapshot. This mirrors getFileReadCache()'s isolation + * contract while keeping type-safe direct field assignment inside the class. + */ + getMemoryPressureMonitor(): MemoryPressureMonitor | undefined { + if (!Object.prototype.hasOwnProperty.call(this, 'memoryPressureMonitor')) { + const inheritedMonitor = this.memoryPressureMonitor; + if (inheritedMonitor) { + const inheritedConfig = this.memoryPressureConfig; + if (!inheritedConfig) { + throw new Error( + 'Inherited memory pressure monitor is missing config', + ); + } + this.memoryPressureConfig = { ...inheritedConfig }; + this.memoryPressureMonitor = new MemoryPressureMonitor( + this, + this.memoryPressureConfig, + ); + } + } + return this.memoryPressureMonitor; + } + getCronScheduler(): CronScheduler { if (!this.cronScheduler) { this.cronScheduler = new CronScheduler(); diff --git a/packages/core/src/core/coreToolScheduler.test.ts b/packages/core/src/core/coreToolScheduler.test.ts index bd301398f79..a52f6ff19e6 100644 --- a/packages/core/src/core/coreToolScheduler.test.ts +++ b/packages/core/src/core/coreToolScheduler.test.ts @@ -487,6 +487,7 @@ describe('CoreToolScheduler', () => { disableHooks?: boolean; onAllToolCallsComplete?: ReturnType; onToolCallsUpdate?: ReturnType; + memoryMonitor?: { scheduleCheck: () => void }; }) { const ensureTool = vi.fn( async (name: string) => @@ -536,6 +537,7 @@ describe('CoreToolScheduler', () => { getUseModelRouter: () => false, getGeminiClient: () => null, getChatRecordingService: () => undefined, + getMemoryPressureMonitor: () => options.memoryMonitor, getMessageBus: vi.fn().mockReturnValue(options.messageBus), getDisableAllHooks: vi .fn() @@ -605,6 +607,43 @@ describe('CoreToolScheduler', () => { ); }); + it('schedules a memory pressure check after tool execution', async () => { + const execute = vi.fn().mockResolvedValue({ + llmContent: 'ok', + returnDisplay: 'ok', + }); + const toolsByName = new Map([ + [ + 'mockTool', + new MockTool({ + name: 'mockTool', + execute, + }), + ], + ]); + const scheduleCheck = vi.fn(); + const { scheduler } = createSchedulerForLegacyToolTests({ + toolsByName, + memoryMonitor: { scheduleCheck }, + }); + + await scheduler.schedule( + [ + { + callId: 'memory-check', + name: 'mockTool', + args: {}, + isClientInitiated: false, + prompt_id: 'prompt-memory-check', + }, + ], + new AbortController().signal, + ); + + expect(execute).toHaveBeenCalledOnce(); + expect(scheduleCheck).toHaveBeenCalledTimes(1); + }); + it('applies canonical legacy tool names to the deny-list fallback', async () => { const execute = vi.fn().mockResolvedValue({ llmContent: 'edited', diff --git a/packages/core/src/core/coreToolScheduler.ts b/packages/core/src/core/coreToolScheduler.ts index 9b594554132..a5a02716ec4 100644 --- a/packages/core/src/core/coreToolScheduler.ts +++ b/packages/core/src/core/coreToolScheduler.ts @@ -51,6 +51,7 @@ import { fileURLToPath } from 'node:url'; import { ToolNames, ToolNamesMigration } from '../tools/tool-names.js'; import { escapeSystemReminderTags, escapeXml } from '../utils/xml.js'; import { unescapePath, PATH_ARG_KEYS } from '../utils/paths.js'; +import type { MemoryPressureMonitor } from '../services/memoryPressureMonitor.js'; import { CONCURRENCY_SAFE_KINDS } from '../tools/tools.js'; import { isShellCommandReadOnly } from '../utils/shellReadOnlyChecker.js'; import { stripShellWrapper } from '../utils/shell-utils.js'; @@ -838,6 +839,10 @@ export class CoreToolScheduler { this.chatRecordingService = options.chatRecordingService; } + private get memoryMonitor(): MemoryPressureMonitor | undefined { + return this.config.getMemoryPressureMonitor?.(); + } + private setStatusInternal( targetCallId: string, status: 'success', @@ -2560,6 +2565,7 @@ export class CoreToolScheduler { // _executeToolCallBody pre-sets status (OK / FAILURE / CANCELLED) via // setToolSpan*; finalize without metadata to preserve that. this.finalizeToolSpan(callId); + this.memoryMonitor?.scheduleCheck(); } } diff --git a/packages/core/src/services/fileReadCache.test.ts b/packages/core/src/services/fileReadCache.test.ts index 894b0cb85f0..c8f24df4355 100644 --- a/packages/core/src/services/fileReadCache.test.ts +++ b/packages/core/src/services/fileReadCache.test.ts @@ -704,4 +704,121 @@ describe('FileReadCache', () => { expect(cache.check(makeStats({ ino: 0 })).state).not.toBe('unknown'); }); }); + + describe('evictNotAccessedSince', () => { + afterEach(() => { + vi.useRealTimers(); + }); + + it('evicts entries with lastReadAt before the cutoff', () => { + const cache = new FileReadCache(); + const now = 2_000_000_000; + const old = now - 100 * 60_000; // 100 minutes ago + const recent = now - 5 * 60_000; // 5 minutes ago + + const oldStats = makeStats({ ino: 1 }); + const recentStats = makeStats({ ino: 2 }); + + vi.useFakeTimers(); + vi.setSystemTime(old); + cache.recordRead('/x/old.ts', oldStats, { full: true, cacheable: true }); + + vi.setSystemTime(recent); + cache.recordRead('/x/recent.ts', recentStats, { + full: true, + cacheable: true, + }); + + // Now set time to "now" and evict entries older than 30 minutes + vi.setSystemTime(now); + + const evicted = cache.evictNotAccessedSince(30); + expect(evicted).toBe(1); + expect(cache.check(oldStats).state).toBe('unknown'); + expect(cache.check(recentStats).state).toBe('fresh'); + }); + + it('evicts entries that were only written, never read', () => { + const cache = new FileReadCache(); + vi.useFakeTimers(); + + const pastWrite = 1000; // some fixed timestamp in the past + vi.setSystemTime(pastWrite); + cache.recordWrite('/x/old-write.ts', makeStats({ ino: 2 })); + + // Advance time by 120 minutes + vi.setSystemTime(pastWrite + 120 * 60_000); + + const evicted = cache.evictNotAccessedSince(60); + expect(evicted).toBe(1); + expect(cache.size()).toBe(0); + }); + + it('preserves recently accessed entries', () => { + const cache = new FileReadCache(); + vi.useFakeTimers(); + + const now = Date.now(); + vi.setSystemTime(now); + + cache.recordRead('/x/recent.ts', makeStats({ ino: 1 }), { + full: true, + cacheable: true, + }); + + const evicted = cache.evictNotAccessedSince(30); + expect(evicted).toBe(0); + expect(cache.size()).toBe(1); + }); + + it('returns correct eviction count', () => { + const cache = new FileReadCache(); + vi.useFakeTimers(); + + const now = Date.now(); + vi.setSystemTime(now); + + // 3 recent entries + for (let i = 0; i < 3; i++) { + cache.recordRead(`/x/recent-${i}.ts`, makeStats({ ino: i }), { + full: true, + cacheable: true, + }); + } + + // Jump 120 minutes back, add 2 old entries + vi.setSystemTime(now - 120 * 60_000); + for (let i = 10; i < 12; i++) { + cache.recordRead(`/x/old-${i}.ts`, makeStats({ ino: i }), { + full: true, + cacheable: true, + }); + } + + // Back to now + vi.setSystemTime(now); + + const evicted = cache.evictNotAccessedSince(60); + expect(evicted).toBe(2); + expect(cache.size()).toBe(3); + }); + + it('returns 0 for empty cache', () => { + const cache = new FileReadCache(); + expect(cache.evictNotAccessedSince(30)).toBe(0); + }); + + it('does not evict entries for sub-minute windows', () => { + const cache = new FileReadCache(); + cache.recordRead('/x/recent.ts', makeStats({ ino: 1 }), { + full: true, + cacheable: true, + }); + + expect(cache.evictNotAccessedSince(0)).toBe(0); + expect(cache.evictNotAccessedSince(-30)).toBe(0); + expect(cache.evictNotAccessedSince(0.0000001)).toBe(0); + expect(cache.size()).toBe(1); + }); + }); }); diff --git a/packages/core/src/services/fileReadCache.ts b/packages/core/src/services/fileReadCache.ts index 0dbd1d7b452..5f05a4661ed 100644 --- a/packages/core/src/services/fileReadCache.ts +++ b/packages/core/src/services/fileReadCache.ts @@ -317,6 +317,36 @@ export class FileReadCache { this.byInode.clear(); } + /** + * Evict entries whose most recent Read (or Write; both set + * {@link FileReadEntry.lastReadAt}) is older than `minutes`. + * + * This is a memory-pressure-driven eviction: it targets entries the + * model is least likely to need again, trading cache hit rate for lower + * memory footprint. Unlike {@link clear}, it preserves recently-read + * entries so the file_unchanged fast-path stays available for active + * files. + * + * @returns Number of entries evicted. + */ + evictNotAccessedSince(minutes: number): number { + if (!Number.isFinite(minutes) || minutes < 1) { + return 0; + } + + const cutoff = Date.now() - minutes * 60 * 1000; + let evicted = 0; + + for (const [key, entry] of this.byInode) { + if (entry.lastReadAt !== undefined && entry.lastReadAt < cutoff) { + this.byInode.delete(key); + evicted++; + } + } + + return evicted; + } + /** Number of tracked entries. Diagnostic / test use only. */ size(): number { return this.byInode.size; diff --git a/packages/core/src/services/memoryPressureMonitor.test.ts b/packages/core/src/services/memoryPressureMonitor.test.ts new file mode 100644 index 00000000000..622e560a005 --- /dev/null +++ b/packages/core/src/services/memoryPressureMonitor.test.ts @@ -0,0 +1,1233 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import { + describe, + expect, + it, + vi, + beforeEach, + beforeAll, + afterEach, +} from 'vitest'; +import { + DEFAULT_PRESSURE_CONFIG, + validateMemoryPressureConfig, +} from './memoryPressureMonitor.js'; +import type { FileReadCache } from './fileReadCache.js'; +import type { Config } from '../config/config.js'; + +// Hoisted so vi.mock can consume it. +const { mockDebugLogger } = vi.hoisted(() => ({ + mockDebugLogger: { + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }, +})); + +const { + getMockOsTotalmem, + setOsTotalmem, + getMockCgroupFile, + setCgroupMemoryMax, + setCgroupV1MemoryLimit, + getMockHeapSizeLimit, + setHeapSizeLimit, +} = vi.hoisted(() => { + let totalmem = 16 * 1024 * 1024 * 1024; // 16 GB default + let cgroupMemoryMax: string | undefined = 'max'; + let cgroupV1MemoryLimit: string | undefined; + let heapSizeLimit = 16 * 1024 * 1024 * 1024; // 16 GB default + return { + getMockOsTotalmem: () => totalmem, + setOsTotalmem: (v: number) => { + totalmem = v; + }, + getMockCgroupFile: (path: string) => { + if (path === '/sys/fs/cgroup/memory.max') { + if (cgroupMemoryMax === undefined) { + throw new Error('ENOENT'); + } + return cgroupMemoryMax; + } + if (path === '/sys/fs/cgroup/memory/memory.limit_in_bytes') { + if (cgroupV1MemoryLimit === undefined) { + throw new Error('ENOENT'); + } + return cgroupV1MemoryLimit; + } + throw new Error('ENOENT'); + }, + setCgroupMemoryMax: (v: string | undefined) => { + cgroupMemoryMax = v; + }, + setCgroupV1MemoryLimit: (v: string | undefined) => { + cgroupV1MemoryLimit = v; + }, + getMockHeapSizeLimit: () => heapSizeLimit, + setHeapSizeLimit: (v: number) => { + heapSizeLimit = v; + }, + }; +}); + +vi.mock('node:os', () => ({ + totalmem: () => getMockOsTotalmem(), +})); + +vi.mock('node:fs', () => ({ + readFileSync: (path: string) => getMockCgroupFile(path), +})); + +vi.mock('node:v8', () => ({ + getHeapStatistics: () => ({ + heap_size_limit: getMockHeapSizeLimit(), + }), +})); + +vi.mock('../utils/debugLogger.js', () => ({ + createDebugLogger: () => mockDebugLogger, +})); + +// Must be a dynamic import AFTER vi.mock so the mocked os takes effect. +// Use let + beforeAll pattern. +let MemoryPressureMonitor: typeof import('./memoryPressureMonitor.js').MemoryPressureMonitor; + +beforeAll(async () => { + const mod = await import('./memoryPressureMonitor.js'); + MemoryPressureMonitor = mod.MemoryPressureMonitor; +}); + +function createMockConfig( + overrides: { + fileReadCache?: Partial; + } = {}, +): Config { + return { + getFileReadCache: () => + ({ + clear: vi.fn(), + evictNotAccessedSince: vi.fn().mockReturnValue(0), + ...overrides.fileReadCache, + }) as unknown as FileReadCache, + } as unknown as Config; +} + +function setMemUsage(rssBytes: number, heapUsedBytes = 256 * 1024 * 1024) { + vi.spyOn(process, 'memoryUsage').mockReturnValue( + createMemUsage(rssBytes, heapUsedBytes), + ); +} + +function createMemUsage( + rssBytes: number, + heapUsedBytes = 256 * 1024 * 1024, +): ReturnType { + return { + rss: rssBytes, + heapTotal: 512 * 1024 * 1024, + heapUsed: heapUsedBytes, + external: 0, + arrayBuffers: 0, + }; +} + +async function drainCleanupMeasurement(): Promise { + await Promise.resolve(); + await Promise.resolve(); + await new Promise((resolve) => setImmediate(resolve)); + await Promise.resolve(); +} + +describe('MemoryPressureMonitor', () => { + beforeEach(() => { + mockDebugLogger.debug.mockClear(); + mockDebugLogger.info.mockClear(); + mockDebugLogger.warn.mockClear(); + mockDebugLogger.error.mockClear(); + setCgroupMemoryMax('max'); + setCgroupV1MemoryLimit(undefined); + setHeapSizeLimit(16 * 1024 * 1024 * 1024); + }); + + afterEach(() => { + vi.restoreAllMocks(); + vi.unstubAllGlobals(); + }); + + describe('validateMemoryPressureConfig', () => { + it('accepts valid config', () => { + expect(() => + validateMemoryPressureConfig({ + softPressureRatio: 0.6, + hardPressureRatio: 0.7, + criticalRatio: 0.8, + cleanupCooldownMs: 5000, + enableExplicitGC: false, + }), + ).not.toThrow(); + }); + + it('accepts zero cleanup cooldowns', () => { + expect(() => + validateMemoryPressureConfig({ + softPressureRatio: 0.6, + hardPressureRatio: 0.7, + criticalRatio: 0.8, + cleanupCooldownMs: 0, + enableExplicitGC: false, + }), + ).not.toThrow(); + }); + + it('rejects soft >= hard', () => { + expect(() => + validateMemoryPressureConfig({ + softPressureRatio: 0.8, + hardPressureRatio: 0.7, + criticalRatio: 0.9, + cleanupCooldownMs: 5000, + enableExplicitGC: false, + }), + ).toThrow('softPressureRatio must be < hardPressureRatio'); + }); + + it('rejects hard >= critical', () => { + expect(() => + validateMemoryPressureConfig({ + softPressureRatio: 0.5, + hardPressureRatio: 0.9, + criticalRatio: 0.9, + cleanupCooldownMs: 5000, + enableExplicitGC: false, + }), + ).toThrow('hardPressureRatio must be < criticalRatio'); + }); + + it('rejects non-finite ratios', () => { + expect(() => + validateMemoryPressureConfig({ + softPressureRatio: Number.NaN, + hardPressureRatio: 0.7, + criticalRatio: 0.9, + cleanupCooldownMs: 5000, + enableExplicitGC: false, + }), + ).toThrow('softPressureRatio must be a finite ratio in [0.3, 0.98]'); + + expect(() => + validateMemoryPressureConfig({ + softPressureRatio: 0.5, + hardPressureRatio: Number.POSITIVE_INFINITY, + criticalRatio: 0.9, + cleanupCooldownMs: 5000, + enableExplicitGC: false, + }), + ).toThrow('hardPressureRatio must be a finite ratio in [0.3, 0.98]'); + }); + + it('rejects ratios below 0.3', () => { + expect(() => + validateMemoryPressureConfig({ + softPressureRatio: 0.2, + hardPressureRatio: 0.7, + criticalRatio: 0.9, + cleanupCooldownMs: 5000, + enableExplicitGC: false, + }), + ).toThrow('softPressureRatio must be a finite ratio in [0.3, 0.98]'); + }); + + it('rejects ratios above 0.98', () => { + expect(() => + validateMemoryPressureConfig({ + softPressureRatio: 0.5, + hardPressureRatio: 0.7, + criticalRatio: 0.99, + cleanupCooldownMs: 5000, + enableExplicitGC: false, + }), + ).toThrow('criticalRatio must be a finite ratio in [0.3, 0.98]'); + }); + + it('rejects negative cleanup cooldowns', () => { + expect(() => + validateMemoryPressureConfig({ + softPressureRatio: 0.5, + hardPressureRatio: 0.7, + criticalRatio: 0.9, + cleanupCooldownMs: -1, + enableExplicitGC: false, + }), + ).toThrow('cleanupCooldownMs must be a non-negative number'); + }); + }); + + describe('getPressureLevel', () => { + let monitor: InstanceType; + + beforeEach(() => { + setOsTotalmem(16 * 1024 * 1024 * 1024); // 16 GB + monitor = new MemoryPressureMonitor(createMockConfig()); + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + it('returns normal when RSS is low', () => { + setMemUsage(1 * 1024 * 1024 * 1024); // 1 GB + // 1/16 = 0.0625 < 0.50: normal + expect(monitor.getPressureLevel()).toBe('normal'); + }); + + it('returns soft when RSS exceeds soft ratio', () => { + setMemUsage(9 * 1024 * 1024 * 1024); // 9 GB + // 9/16 = 0.5625 >= 0.50: soft + expect(monitor.getPressureLevel()).toBe('soft'); + }); + + it('returns hard when RSS exceeds hard ratio', () => { + setMemUsage(11 * 1024 * 1024 * 1024); // 11 GB + // 11/16 = 0.6875 >= 0.65: hard + expect(monitor.getPressureLevel()).toBe('hard'); + }); + + it('returns critical when RSS exceeds critical ratio', () => { + setMemUsage(14 * 1024 * 1024 * 1024); // 14 GB + // 14/16 = 0.875 >= 0.80: critical + expect(monitor.getPressureLevel()).toBe('critical'); + }); + + it('does not treat a zero effective memory limit as RSS pressure', () => { + setOsTotalmem(0); + setCgroupMemoryMax('max'); + setCgroupV1MemoryLimit(undefined); + monitor = new MemoryPressureMonitor(createMockConfig()); + + setMemUsage(1024 * 1024 * 1024, 0); + expect(monitor.getPressureLevel()).toBe('normal'); + expect(mockDebugLogger.warn).toHaveBeenCalledWith( + 'Effective memory limit is not positive; RSS pressure checks are disabled', + ); + }); + + it('returns normal when process memory usage cannot be read', () => { + vi.spyOn(process, 'memoryUsage').mockImplementation(() => { + throw new Error('memory API unavailable'); + }); + + expect(monitor.getPressureLevel()).toBe('normal'); + expect(mockDebugLogger.error).toHaveBeenCalledWith( + 'Failed to read memory usage for pressure check: memory API unavailable', + ); + }); + + it('uses cgroup memory.max when available', () => { + setOsTotalmem(16 * 1024 * 1024 * 1024); + setCgroupMemoryMax(String(2 * 1024 * 1024 * 1024)); // 2 GB + monitor = new MemoryPressureMonitor(createMockConfig()); + + setMemUsage(1200 * 1024 * 1024); // 1.2/2 = 0.586 >= 0.50 + expect(monitor.getPressureLevel()).toBe('soft'); + expect(mockDebugLogger.info).toHaveBeenCalledWith( + 'Using cgroup v2 memory limit: 2048 MiB', + ); + }); + + it('uses cgroup memory.max even when host total memory is unavailable', () => { + setOsTotalmem(0); + setCgroupMemoryMax(String(2 * 1024 * 1024 * 1024)); // 2 GB + monitor = new MemoryPressureMonitor(createMockConfig()); + + setMemUsage(1200 * 1024 * 1024); // 1.2/2 = 0.586 >= 0.50 + expect(monitor.getPressureLevel()).toBe('soft'); + }); + + it('uses cgroup v1 memory.limit_in_bytes when cgroup v2 is unavailable', () => { + setOsTotalmem(16 * 1024 * 1024 * 1024); + setCgroupMemoryMax(undefined); + setCgroupV1MemoryLimit(String(2 * 1024 * 1024 * 1024)); // 2 GB + monitor = new MemoryPressureMonitor(createMockConfig()); + + setMemUsage(1200 * 1024 * 1024); // 1.2/2 = 0.586 >= 0.50 + expect(monitor.getPressureLevel()).toBe('soft'); + expect(mockDebugLogger.info).toHaveBeenCalledWith( + 'Using cgroup v1 memory limit: 2048 MiB', + ); + }); + + it('ignores cgroup v1 unlimited sentinel values', () => { + setOsTotalmem(16 * 1024 * 1024 * 1024); + setCgroupMemoryMax(undefined); + setCgroupV1MemoryLimit('9223372036854771712'); + monitor = new MemoryPressureMonitor(createMockConfig()); + + setMemUsage(1200 * 1024 * 1024); // 1.2/16 = 0.073 < 0.50 + expect(monitor.getPressureLevel()).toBe('normal'); + expect(mockDebugLogger.warn).not.toHaveBeenCalledWith( + expect.stringContaining('9223372036854771712'), + ); + expect(mockDebugLogger.debug).toHaveBeenCalledWith( + 'Ignoring unlimited cgroup memory limit from ' + + '/sys/fs/cgroup/memory/memory.limit_in_bytes: 9223372036854771712', + ); + }); + + it('ignores cgroup v1 unlimited sentinel values when host total is unavailable', () => { + setOsTotalmem(0); + setCgroupMemoryMax(undefined); + setCgroupV1MemoryLimit('9223372036854771712'); + monitor = new MemoryPressureMonitor(createMockConfig()); + + setMemUsage(1200 * 1024 * 1024); + expect(monitor.getPressureLevel()).toBe('normal'); + expect(mockDebugLogger.warn).not.toHaveBeenCalledWith( + expect.stringContaining('9223372036854771712'), + ); + expect(mockDebugLogger.debug).toHaveBeenCalledWith( + 'Ignoring unlimited cgroup memory limit from ' + + '/sys/fs/cgroup/memory/memory.limit_in_bytes: 9223372036854771712', + ); + }); + + it('ignores malformed cgroup limits without partially parsing them', () => { + setOsTotalmem(16 * 1024 * 1024 * 1024); + setCgroupMemoryMax('2048garbage'); + setCgroupV1MemoryLimit(undefined); + monitor = new MemoryPressureMonitor(createMockConfig()); + + setMemUsage(1200 * 1024 * 1024); // 1.2/16 = 0.073 < 0.50 + expect(monitor.getPressureLevel()).toBe('normal'); + expect(mockDebugLogger.warn).toHaveBeenCalledWith( + 'Ignoring non-numeric cgroup memory limit from ' + + '/sys/fs/cgroup/memory.max: 2048garbage', + ); + expect(mockDebugLogger.info).toHaveBeenCalledWith( + 'Using host memory limit: 16384 MiB', + ); + }); + + it('logs cgroup read failures before falling back to host memory', () => { + setOsTotalmem(16 * 1024 * 1024 * 1024); + setCgroupMemoryMax(undefined); + setCgroupV1MemoryLimit(undefined); + monitor = new MemoryPressureMonitor(createMockConfig()); + + setMemUsage(1200 * 1024 * 1024); + expect(monitor.getPressureLevel()).toBe('normal'); + expect(mockDebugLogger.debug).toHaveBeenCalledWith( + 'Failed to read cgroup memory limit from /sys/fs/cgroup/memory.max: ' + + 'ENOENT', + ); + expect(mockDebugLogger.info).toHaveBeenCalledWith( + 'Using host memory limit: 16384 MiB', + ); + }); + + it('logs out-of-range cgroup limits distinctly', () => { + setOsTotalmem(16 * 1024 * 1024 * 1024); + setCgroupMemoryMax('0'); + setCgroupV1MemoryLimit(undefined); + monitor = new MemoryPressureMonitor(createMockConfig()); + + setMemUsage(1200 * 1024 * 1024); + expect(monitor.getPressureLevel()).toBe('normal'); + expect(mockDebugLogger.warn).toHaveBeenCalledWith( + 'Ignoring out-of-range cgroup memory limit from ' + + '/sys/fs/cgroup/memory.max: 0', + ); + }); + + it('ignores negative cgroup limits as out-of-range', () => { + setOsTotalmem(16 * 1024 * 1024 * 1024); + setCgroupMemoryMax('-1'); + setCgroupV1MemoryLimit(undefined); + monitor = new MemoryPressureMonitor(createMockConfig()); + + setMemUsage(1200 * 1024 * 1024); + expect(monitor.getPressureLevel()).toBe('normal'); + expect(mockDebugLogger.warn).toHaveBeenCalledWith( + 'Ignoring out-of-range cgroup memory limit from ' + + '/sys/fs/cgroup/memory.max: -1', + ); + }); + + it('ignores safe cgroup limits above host total memory', () => { + setOsTotalmem(16 * 1024 * 1024 * 1024); + setCgroupMemoryMax(String(32 * 1024 * 1024 * 1024)); + setCgroupV1MemoryLimit(undefined); + monitor = new MemoryPressureMonitor(createMockConfig()); + + setMemUsage(1200 * 1024 * 1024); // 1.2/16 = 0.073 < 0.50 + expect(monitor.getPressureLevel()).toBe('normal'); + expect(mockDebugLogger.debug).toHaveBeenCalledWith( + 'Ignoring cgroup memory limit above host total from ' + + '/sys/fs/cgroup/memory.max: 34359738368', + ); + }); + + it('ignores unrealistically small cgroup limits', () => { + setOsTotalmem(16 * 1024 * 1024 * 1024); + setCgroupMemoryMax('1'); + setCgroupV1MemoryLimit(undefined); + monitor = new MemoryPressureMonitor(createMockConfig()); + + setMemUsage(1200 * 1024 * 1024); // 1.2/16 = 0.073 < 0.50 + expect(monitor.getPressureLevel()).toBe('normal'); + expect(mockDebugLogger.warn).toHaveBeenCalledWith( + 'Ignoring unrealistically small cgroup memory limit from ' + + '/sys/fs/cgroup/memory.max: 1', + ); + }); + + it('does not treat heap usage as pressure when V8 heap limit is zero', () => { + setOsTotalmem(16 * 1024 * 1024 * 1024); + setHeapSizeLimit(0); + monitor = new MemoryPressureMonitor(createMockConfig()); + + setMemUsage(512 * 1024 * 1024, 12 * 1024 * 1024 * 1024); + expect(monitor.getPressureLevel()).toBe('normal'); + }); + + it('refreshes the V8 heap limit for each pressure check', () => { + setOsTotalmem(64 * 1024 * 1024 * 1024); // 64 GB + setHeapSizeLimit(1024 * 1024 * 1024); // 1 GB at construction + monitor = new MemoryPressureMonitor(createMockConfig()); + + setHeapSizeLimit(4 * 1024 * 1024 * 1024); // V8 grew the limit later + setMemUsage(512 * 1024 * 1024, 800 * 1024 * 1024); + + expect(monitor.getPressureLevel()).toBe('normal'); + }); + + it('uses V8 heap pressure even when RSS is low versus system memory', () => { + setOsTotalmem(64 * 1024 * 1024 * 1024); // 64 GB + setHeapSizeLimit(2 * 1024 * 1024 * 1024); // 2 GB + monitor = new MemoryPressureMonitor(createMockConfig()); + + setMemUsage(512 * 1024 * 1024, 1200 * 1024 * 1024); // heap 1.2/2 = 0.586 + expect(monitor.getPressureLevel()).toBe('soft'); + }); + }); + + describe('scheduleCheck', () => { + it('only schedules one check per microtask round', async () => { + setOsTotalmem(16 * 1024 * 1024 * 1024); + const evictSpy = vi.fn().mockReturnValue(0); + const monitor = new MemoryPressureMonitor( + createMockConfig({ + fileReadCache: { evictNotAccessedSince: evictSpy }, + }), + { ...DEFAULT_PRESSURE_CONFIG, cleanupCooldownMs: 0 }, + ); + + // Soft pressure should call evictNotAccessedSince(60). + vi.spyOn(process, 'memoryUsage').mockReturnValue({ + rss: 9 * 1024 * 1024 * 1024, // 9/16 = 0.5625 >= 0.50: soft + heapTotal: 0, + heapUsed: 0, + external: 0, + arrayBuffers: 0, + }); + + // Three rapid calls should be merged into one pending check. + monitor.scheduleCheck(); + monitor.scheduleCheck(); + monitor.scheduleCheck(); + + // Drain microtasks so the queued callback runs. + await new Promise((resolve) => queueMicrotask(() => resolve())); + await new Promise((resolve) => queueMicrotask(() => resolve())); + + // Verify evictNotAccessedSince was called exactly once (not 3x). + expect(evictSpy).toHaveBeenCalledTimes(1); + + vi.restoreAllMocks(); + }); + }); + + describe('performCheck with cleanup', () => { + beforeEach(() => { + setOsTotalmem(16 * 1024 * 1024 * 1024); + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + it('calls evictNotAccessedSince on soft pressure', () => { + const evictSpy = vi.fn().mockReturnValue(5); + const monitor = new MemoryPressureMonitor( + createMockConfig({ + fileReadCache: { evictNotAccessedSince: evictSpy }, + }), + { ...DEFAULT_PRESSURE_CONFIG, cleanupCooldownMs: 0 }, + ); + + setMemUsage(9 * 1024 * 1024 * 1024); // 9/16 = 0.5625 >= 0.50: soft + monitor.performCheck(); + expect(evictSpy).toHaveBeenCalledWith(60); + }); + + it('calls evictNotAccessedSince on hard pressure', () => { + const evictSpy = vi.fn().mockReturnValue(5); + const monitor = new MemoryPressureMonitor( + createMockConfig({ + fileReadCache: { evictNotAccessedSince: evictSpy }, + }), + { ...DEFAULT_PRESSURE_CONFIG, cleanupCooldownMs: 0 }, + ); + + setMemUsage(11 * 1024 * 1024 * 1024); // 11/16 = 0.6875 >= 0.65: hard + monitor.performCheck(); + expect(evictSpy).toHaveBeenCalledWith(30); + }); + + it('calls clear on critical pressure', () => { + const clearSpy = vi.fn(); + const monitor = new MemoryPressureMonitor( + createMockConfig({ + fileReadCache: { + clear: clearSpy, + evictNotAccessedSince: vi.fn(), + }, + }), + { ...DEFAULT_PRESSURE_CONFIG, cleanupCooldownMs: 0 }, + ); + + setMemUsage(14 * 1024 * 1024 * 1024); // 14/16 = 0.875 >= 0.80: critical + monitor.performCheck(); + expect(clearSpy).toHaveBeenCalled(); + }); + + it('runs escalated critical cleanup after lower cleanup finishes', async () => { + const clearSpy = vi.fn(); + const evictSpy = vi.fn().mockReturnValue(0); + const monitor = new MemoryPressureMonitor( + createMockConfig({ + fileReadCache: { + clear: clearSpy, + evictNotAccessedSince: evictSpy, + }, + }), + { ...DEFAULT_PRESSURE_CONFIG, cleanupCooldownMs: 60_000 }, + ); + + setMemUsage(9 * 1024 * 1024 * 1024); // soft pressure + monitor.performCheck(); + expect(evictSpy).toHaveBeenCalledWith(60); + + setMemUsage(14 * 1024 * 1024 * 1024); // escalates to critical + monitor.performCheck(); + expect(clearSpy).not.toHaveBeenCalled(); + + await drainCleanupMeasurement(); + expect(clearSpy).toHaveBeenCalledTimes(1); + }); + + it('keeps the strongest queued cleanup while cleanup is in progress', async () => { + const clearSpy = vi.fn(); + const evictSpy = vi.fn().mockReturnValue(0); + const monitor = new MemoryPressureMonitor( + createMockConfig({ + fileReadCache: { + clear: clearSpy, + evictNotAccessedSince: evictSpy, + }, + }), + { ...DEFAULT_PRESSURE_CONFIG, cleanupCooldownMs: 60_000 }, + ); + vi.spyOn(monitor, 'getPressureLevel') + .mockReturnValueOnce('soft') + .mockReturnValueOnce('critical') + .mockReturnValueOnce('hard'); + + monitor.performCheck(); + monitor.performCheck(); + monitor.performCheck(); + await drainCleanupMeasurement(); + + expect(clearSpy).toHaveBeenCalledTimes(1); + expect(evictSpy).toHaveBeenCalledWith(60); + expect(evictSpy).not.toHaveBeenCalledWith(30); + }); + + it('cancels queued cleanup when the session is reset', async () => { + const clearSpy = vi.fn(); + const evictSpy = vi.fn().mockReturnValue(0); + const monitor = new MemoryPressureMonitor( + createMockConfig({ + fileReadCache: { + clear: clearSpy, + evictNotAccessedSince: evictSpy, + }, + }), + { ...DEFAULT_PRESSURE_CONFIG, cleanupCooldownMs: 60_000 }, + ); + vi.spyOn(monitor, 'getPressureLevel') + .mockReturnValueOnce('soft') + .mockReturnValueOnce('critical') + .mockReturnValue('critical'); + setMemUsage(14 * 1024 * 1024 * 1024); + + monitor.performCheck(); + monitor.performCheck(); + monitor.resetForNewSession(); + await drainCleanupMeasurement(); + + expect(evictSpy).toHaveBeenCalledWith(60); + expect(clearSpy).not.toHaveBeenCalled(); + + monitor.performCheck(); + await drainCleanupMeasurement(); + expect(clearSpy).toHaveBeenCalledTimes(1); + }); + + it('blocks same-level cleanup within the cooldown window', async () => { + const evictSpy = vi.fn().mockReturnValue(0); + const monitor = new MemoryPressureMonitor( + createMockConfig({ + fileReadCache: { evictNotAccessedSince: evictSpy }, + }), + { ...DEFAULT_PRESSURE_CONFIG, cleanupCooldownMs: 60_000 }, + ); + + setMemUsage(9 * 1024 * 1024 * 1024); // soft pressure + monitor.performCheck(); + await drainCleanupMeasurement(); + monitor.performCheck(); + await drainCleanupMeasurement(); + + expect(evictSpy).toHaveBeenCalledTimes(1); + expect(evictSpy).toHaveBeenCalledWith(60); + }); + + it('does not count successful cleanup as a failure when RSS does not drop', async () => { + const monitor = new MemoryPressureMonitor( + createMockConfig({ + fileReadCache: { evictNotAccessedSince: vi.fn().mockReturnValue(0) }, + }), + { ...DEFAULT_PRESSURE_CONFIG, cleanupCooldownMs: 0 }, + ); + const cleanupFailed = vi.fn(); + monitor.on('memory-cleanup-failed', cleanupFailed); + + setMemUsage(9 * 1024 * 1024 * 1024); // soft pressure, unchanged RSS + + for (let i = 0; i < 3; i++) { + monitor.performCheck(); + await drainCleanupMeasurement(); + } + + expect(monitor.getConsecutiveFailures()).toBe(0); + expect(cleanupFailed).not.toHaveBeenCalled(); + }); + + it('emits a diagnostic event after repeated ineffective cleanups', async () => { + const monitor = new MemoryPressureMonitor( + createMockConfig({ + fileReadCache: { evictNotAccessedSince: vi.fn().mockReturnValue(0) }, + }), + { ...DEFAULT_PRESSURE_CONFIG, cleanupCooldownMs: 0 }, + ); + const cleanupIneffective = vi.fn(); + monitor.on('memory-cleanup-ineffective', cleanupIneffective); + + setMemUsage(9 * 1024 * 1024 * 1024); // soft pressure, unchanged RSS + + for (let i = 0; i < 3; i++) { + monitor.performCheck(); + await drainCleanupMeasurement(); + } + + expect(cleanupIneffective).toHaveBeenCalledTimes(1); + expect(cleanupIneffective).toHaveBeenCalledWith( + expect.objectContaining({ + consecutiveIneffectiveCleanups: 3, + freedRatio: 0, + }), + ); + }); + + it('throttles diagnostic events for continued ineffective cleanup', async () => { + const monitor = new MemoryPressureMonitor( + createMockConfig({ + fileReadCache: { evictNotAccessedSince: vi.fn().mockReturnValue(0) }, + }), + { ...DEFAULT_PRESSURE_CONFIG, cleanupCooldownMs: 0 }, + ); + const cleanupIneffective = vi.fn(); + monitor.on('memory-cleanup-ineffective', cleanupIneffective); + + setMemUsage(9 * 1024 * 1024 * 1024); // soft pressure, unchanged RSS + + for (let i = 0; i < 10; i++) { + monitor.performCheck(); + await drainCleanupMeasurement(); + } + + expect(cleanupIneffective).toHaveBeenCalledTimes(2); + expect(cleanupIneffective).toHaveBeenLastCalledWith( + expect.objectContaining({ + consecutiveIneffectiveCleanups: 10, + freedRatio: 0, + }), + ); + }); + + it('emits repeated ineffective cleanup diagnostics at the long interval', async () => { + const monitor = new MemoryPressureMonitor( + createMockConfig({ + fileReadCache: { evictNotAccessedSince: vi.fn().mockReturnValue(0) }, + }), + { ...DEFAULT_PRESSURE_CONFIG, cleanupCooldownMs: 0 }, + ); + const cleanupIneffective = vi.fn(); + monitor.on('memory-cleanup-ineffective', cleanupIneffective); + + setMemUsage(9 * 1024 * 1024 * 1024); // soft pressure, unchanged RSS + + for (let i = 0; i < 20; i++) { + monitor.performCheck(); + await drainCleanupMeasurement(); + } + + expect(cleanupIneffective).toHaveBeenCalledTimes(3); + expect(cleanupIneffective).toHaveBeenLastCalledWith( + expect.objectContaining({ + consecutiveIneffectiveCleanups: 20, + freedRatio: 0, + }), + ); + }); + + it('backs off repeated ineffective aggressive cleanup', async () => { + let now = 1_000; + vi.spyOn(Date, 'now').mockImplementation(() => now); + const clearSpy = vi.fn(); + const monitor = new MemoryPressureMonitor( + createMockConfig({ + fileReadCache: { + clear: clearSpy, + evictNotAccessedSince: vi.fn(), + }, + }), + { ...DEFAULT_PRESSURE_CONFIG, cleanupCooldownMs: 1_000 }, + ); + + setMemUsage(14 * 1024 * 1024 * 1024); // critical, unchanged RSS + + for (let i = 0; i < 3; i++) { + monitor.performCheck(); + await drainCleanupMeasurement(); + now += 1_000; + } + + expect(clearSpy).toHaveBeenCalledTimes(3); + + monitor.performCheck(); + await drainCleanupMeasurement(); + expect(clearSpy).toHaveBeenCalledTimes(3); + + now += 1_000; + monitor.performCheck(); + await drainCleanupMeasurement(); + expect(clearSpy).toHaveBeenCalledTimes(4); + }); + + it('measures a cleanup before running a queued escalation', async () => { + let rss = 9 * 1024 * 1024 * 1024; + vi.spyOn(process, 'memoryUsage').mockImplementation(() => ({ + rss, + heapTotal: 512 * 1024 * 1024, + heapUsed: 256 * 1024 * 1024, + external: 0, + arrayBuffers: 0, + })); + + const evictSpy = vi.fn(() => { + rss = 8 * 1024 * 1024 * 1024; + return 0; + }); + const clearSpy = vi.fn(() => { + rss = 4 * 1024 * 1024 * 1024; + }); + const monitor = new MemoryPressureMonitor( + createMockConfig({ + fileReadCache: { + clear: clearSpy, + evictNotAccessedSince: evictSpy, + }, + }), + { ...DEFAULT_PRESSURE_CONFIG, cleanupCooldownMs: 60_000 }, + ); + vi.spyOn(monitor, 'getPressureLevel') + .mockReturnValueOnce('soft') + .mockReturnValueOnce('critical'); + + monitor.performCheck(); + monitor.performCheck(); + await drainCleanupMeasurement(); + + expect(clearSpy).toHaveBeenCalledTimes(1); + expect(mockDebugLogger.info).toHaveBeenCalledWith( + expect.stringContaining( + 'Cleanup "light" completed; RSS delta 1073741824 bytes', + ), + ); + }); + + it('resets ineffective cleanup count after an effective cleanup', async () => { + let rss = 9 * 1024 * 1024 * 1024; + const evictSpy = vi.fn(() => { + if (evictSpy.mock.calls.length === 3) { + rss = 8 * 1024 * 1024 * 1024; + } + return 0; + }); + vi.spyOn(process, 'memoryUsage').mockImplementation(() => + createMemUsage(rss), + ); + const monitor = new MemoryPressureMonitor( + createMockConfig({ + fileReadCache: { evictNotAccessedSince: evictSpy }, + }), + { ...DEFAULT_PRESSURE_CONFIG, cleanupCooldownMs: 0 }, + ); + const cleanupIneffective = vi.fn(); + monitor.on('memory-cleanup-ineffective', cleanupIneffective); + + for (let i = 0; i < 3; i++) { + monitor.performCheck(); + await drainCleanupMeasurement(); + rss = 9 * 1024 * 1024 * 1024; + } + for (let i = 0; i < 2; i++) { + monitor.performCheck(); + await drainCleanupMeasurement(); + } + + expect(cleanupIneffective).not.toHaveBeenCalled(); + }); + + it('records queued cleanup startup failures', async () => { + const evictSpy = vi.fn().mockReturnValue(0); + const monitor = new MemoryPressureMonitor( + createMockConfig({ + fileReadCache: { + clear: vi.fn(), + evictNotAccessedSince: evictSpy, + }, + }), + { ...DEFAULT_PRESSURE_CONFIG, cleanupCooldownMs: 60_000 }, + ); + vi.spyOn(monitor, 'getPressureLevel') + .mockReturnValueOnce('soft') + .mockReturnValueOnce('critical'); + vi.spyOn(process, 'memoryUsage') + .mockReturnValueOnce(createMemUsage(9 * 1024 * 1024 * 1024)) + .mockReturnValueOnce(createMemUsage(8 * 1024 * 1024 * 1024)) + .mockImplementationOnce(() => { + throw new Error('queued RSS unavailable'); + }) + .mockReturnValueOnce(createMemUsage(8 * 1024 * 1024 * 1024)); + + monitor.performCheck(); + monitor.performCheck(); + await drainCleanupMeasurement(); + + expect(monitor.getConsecutiveFailures()).toBe(1); + expect(mockDebugLogger.error).toHaveBeenCalledWith( + 'Cleanup "aggressive" failed: queued RSS unavailable; ' + + 'consecutive failures: 1', + ); + }); + + it('warns when explicit GC is requested but unavailable', async () => { + vi.stubGlobal('gc', undefined); + const clearSpy = vi.fn(); + const monitor = new MemoryPressureMonitor( + createMockConfig({ + fileReadCache: { + clear: clearSpy, + evictNotAccessedSince: vi.fn(), + }, + }), + { + ...DEFAULT_PRESSURE_CONFIG, + cleanupCooldownMs: 0, + enableExplicitGC: true, + }, + ); + + setMemUsage(14 * 1024 * 1024 * 1024); // critical pressure + monitor.performCheck(); + await drainCleanupMeasurement(); + + expect(clearSpy).toHaveBeenCalled(); + expect(mockDebugLogger.warn).toHaveBeenCalledWith( + 'trigger_gc requested but global.gc is not available; ' + + 'start Node.js with --expose-gc', + ); + }); + + it('runs explicit GC when requested and available', async () => { + const gcSpy = vi.fn(); + vi.stubGlobal('gc', gcSpy); + const monitor = new MemoryPressureMonitor( + createMockConfig({ + fileReadCache: { + clear: vi.fn(), + evictNotAccessedSince: vi.fn(), + }, + }), + { + ...DEFAULT_PRESSURE_CONFIG, + cleanupCooldownMs: 0, + enableExplicitGC: true, + }, + ); + + setMemUsage(14 * 1024 * 1024 * 1024); // critical pressure + monitor.performCheck(); + await drainCleanupMeasurement(); + + expect(gcSpy).toHaveBeenCalledTimes(1); + expect(mockDebugLogger.debug).toHaveBeenCalledWith( + 'global.gc() freed 0 bytes', + ); + }); + + it('skips same-priority cleanup while another cleanup is in progress', () => { + const evictSpy = vi.fn().mockReturnValue(0); + const monitor = new MemoryPressureMonitor( + createMockConfig({ + fileReadCache: { evictNotAccessedSince: evictSpy }, + }), + { ...DEFAULT_PRESSURE_CONFIG, cleanupCooldownMs: 0 }, + ); + + setMemUsage(9 * 1024 * 1024 * 1024); // soft pressure + monitor.performCheck(); + monitor.performCheck(); + + expect(evictSpy).toHaveBeenCalledTimes(1); + expect(mockDebugLogger.debug).toHaveBeenCalledWith( + 'Cleanup already in progress, skipping', + ); + }); + + it('counts cleanup step exceptions as failures', async () => { + const monitor = new MemoryPressureMonitor( + createMockConfig({ + fileReadCache: { + evictNotAccessedSince: vi.fn(() => { + throw new Error('cache failure'); + }), + }, + }), + { ...DEFAULT_PRESSURE_CONFIG, cleanupCooldownMs: 0 }, + ); + const cleanupFailed = vi.fn(); + monitor.on('memory-cleanup-failed', cleanupFailed); + + setMemUsage(9 * 1024 * 1024 * 1024); // soft pressure + + for (let i = 0; i < 3; i++) { + monitor.performCheck(); + await drainCleanupMeasurement(); + } + + expect(monitor.getConsecutiveFailures()).toBe(3); + expect(cleanupFailed).toHaveBeenCalledTimes(1); + expect(cleanupFailed).toHaveBeenCalledWith( + expect.objectContaining({ + consecutiveFailures: 3, + error: 'cache failure', + }), + ); + }); + + it('resets consecutive failures after a successful cleanup', async () => { + const evictSpy = vi.fn((): number => { + throw new Error('cache failure'); + }); + const monitor = new MemoryPressureMonitor( + createMockConfig({ + fileReadCache: { + evictNotAccessedSince: evictSpy, + }, + }), + { ...DEFAULT_PRESSURE_CONFIG, cleanupCooldownMs: 0 }, + ); + + setMemUsage(9 * 1024 * 1024 * 1024); // soft pressure + monitor.performCheck(); + await drainCleanupMeasurement(); + expect(monitor.getConsecutiveFailures()).toBe(1); + + evictSpy.mockReturnValue(0); + monitor.performCheck(); + await drainCleanupMeasurement(); + + expect(monitor.getConsecutiveFailures()).toBe(0); + }); + + it('records cleanup failures when RSS cannot be read', async () => { + const monitor = new MemoryPressureMonitor( + createMockConfig({ + fileReadCache: { + evictNotAccessedSince: vi.fn(() => { + throw new Error('cache failure'); + }), + }, + }), + { ...DEFAULT_PRESSURE_CONFIG, cleanupCooldownMs: 0 }, + ); + vi.spyOn(process, 'memoryUsage') + .mockReturnValueOnce(createMemUsage(9 * 1024 * 1024 * 1024)) + .mockReturnValueOnce(createMemUsage(9 * 1024 * 1024 * 1024)) + .mockImplementationOnce(() => { + throw new Error('RSS unavailable'); + }); + + monitor.performCheck(); + await drainCleanupMeasurement(); + + expect(monitor.getConsecutiveFailures()).toBe(1); + expect(mockDebugLogger.error).toHaveBeenCalledWith( + 'Failed to read RSS after cleanup failure: RSS unavailable', + ); + expect(mockDebugLogger.error).toHaveBeenCalledWith( + 'Cleanup "light" failed: cache failure; consecutive failures: 1', + ); + }); + + it('throttles repeated cleanup failure events after the threshold', async () => { + const monitor = new MemoryPressureMonitor( + createMockConfig({ + fileReadCache: { + evictNotAccessedSince: vi.fn(() => { + throw new Error('cache failure'); + }), + }, + }), + { ...DEFAULT_PRESSURE_CONFIG, cleanupCooldownMs: 0 }, + ); + const cleanupFailed = vi.fn(); + monitor.on('memory-cleanup-failed', cleanupFailed); + + setMemUsage(9 * 1024 * 1024 * 1024); // soft pressure + + for (let i = 0; i < 10; i++) { + monitor.performCheck(); + await drainCleanupMeasurement(); + } + + expect(monitor.getConsecutiveFailures()).toBe(10); + expect(cleanupFailed).toHaveBeenCalledTimes(2); + expect(cleanupFailed).toHaveBeenLastCalledWith( + expect.objectContaining({ + consecutiveFailures: 10, + error: 'cache failure', + }), + ); + }); + + it('does not surface listener exceptions as cleanup promise rejections', async () => { + const monitor = new MemoryPressureMonitor( + createMockConfig({ + fileReadCache: { + evictNotAccessedSince: vi.fn(() => { + throw new Error('cache failure'); + }), + }, + }), + { ...DEFAULT_PRESSURE_CONFIG, cleanupCooldownMs: 0 }, + ); + monitor.on('memory-cleanup-failed', () => { + throw new Error('listener failure'); + }); + + setMemUsage(9 * 1024 * 1024 * 1024); // soft pressure + + for (let i = 0; i < 3; i++) { + monitor.performCheck(); + await drainCleanupMeasurement(); + } + + expect(monitor.getConsecutiveFailures()).toBe(3); + }); + }); + + describe('getConsecutiveFailures', () => { + afterEach(() => { + vi.restoreAllMocks(); + }); + + it('starts at zero', () => { + setOsTotalmem(16 * 1024 * 1024 * 1024); + const monitor = new MemoryPressureMonitor(createMockConfig()); + expect(monitor.getConsecutiveFailures()).toBe(0); + }); + + it('can reset consecutive failures', async () => { + setOsTotalmem(16 * 1024 * 1024 * 1024); + const monitor = new MemoryPressureMonitor( + createMockConfig({ + fileReadCache: { + evictNotAccessedSince: vi.fn(() => { + throw new Error('cache failure'); + }), + }, + }), + { ...DEFAULT_PRESSURE_CONFIG, cleanupCooldownMs: 0 }, + ); + + setMemUsage(9 * 1024 * 1024 * 1024); + monitor.performCheck(); + await drainCleanupMeasurement(); + + expect(monitor.getConsecutiveFailures()).toBe(1); + monitor.resetConsecutiveFailures(); + expect(monitor.getConsecutiveFailures()).toBe(0); + }); + + it('can reset ineffective cleanup diagnostics', async () => { + setOsTotalmem(16 * 1024 * 1024 * 1024); + const monitor = new MemoryPressureMonitor( + createMockConfig({ + fileReadCache: { evictNotAccessedSince: vi.fn().mockReturnValue(0) }, + }), + { ...DEFAULT_PRESSURE_CONFIG, cleanupCooldownMs: 0 }, + ); + const cleanupIneffective = vi.fn(); + monitor.on('memory-cleanup-ineffective', cleanupIneffective); + + setMemUsage(9 * 1024 * 1024 * 1024); + + for (let i = 0; i < 2; i++) { + monitor.performCheck(); + await drainCleanupMeasurement(); + } + monitor.resetConsecutiveFailures(); + for (let i = 0; i < 3; i++) { + monitor.performCheck(); + await drainCleanupMeasurement(); + } + + expect(cleanupIneffective).toHaveBeenCalledTimes(1); + expect(cleanupIneffective).toHaveBeenCalledWith( + expect.objectContaining({ + consecutiveIneffectiveCleanups: 3, + }), + ); + }); + }); +}); diff --git a/packages/core/src/services/memoryPressureMonitor.ts b/packages/core/src/services/memoryPressureMonitor.ts new file mode 100644 index 00000000000..956755a0308 --- /dev/null +++ b/packages/core/src/services/memoryPressureMonitor.ts @@ -0,0 +1,605 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import * as os from 'node:os'; +import { readFileSync } from 'node:fs'; +import { EventEmitter } from 'node:events'; +import { getHeapStatistics } from 'node:v8'; +import { createDebugLogger } from '../utils/debugLogger.js'; +import { getErrorMessage } from '../utils/errors.js'; +import type { Config } from '../config/config.js'; + +// Types + +export interface MemoryPressureConfig { + /** RSS / totalmem ratio at which light cleanup begins. Default 0.50. */ + softPressureRatio: number; + /** RSS / totalmem ratio at which moderate cleanup begins. Default 0.65. */ + hardPressureRatio: number; + /** RSS / totalmem ratio at which aggressive cleanup begins. Default 0.80. */ + criticalRatio: number; + /** Minimum ms between consecutive cleanups. Default 5000. */ + cleanupCooldownMs: number; + /** Allow global.gc() in aggressive cleanup. Requires --expose-gc. */ + enableExplicitGC: boolean; +} + +export interface CleanupRecommendation { + action: 'none' | 'light' | 'moderate' | 'aggressive'; + steps: CleanupStep[]; +} + +export type CleanupStep = + | 'clear_file_cache' + | 'evict_cold_cache' + | 'evict_stale_cache' + | 'trigger_gc'; + +export interface MemoryCleanupFailureEvent { + rss: number; + consecutiveFailures: number; + recommendation: CleanupRecommendation; + error: string; +} + +export interface MemoryCleanupIneffectiveEvent { + rss: number; + freedBytes: number; + freedRatio: number; + consecutiveIneffectiveCleanups: number; + recommendation: CleanupRecommendation; +} + +export const DEFAULT_PRESSURE_CONFIG: MemoryPressureConfig = { + softPressureRatio: 0.5, + hardPressureRatio: 0.65, + criticalRatio: 0.8, + cleanupCooldownMs: 5_000, + enableExplicitGC: false, +}; + +// Validation + +export function validateMemoryPressureConfig(c: MemoryPressureConfig): void { + for (const [name, ratio] of [ + ['softPressureRatio', c.softPressureRatio], + ['hardPressureRatio', c.hardPressureRatio], + ['criticalRatio', c.criticalRatio], + ] as const) { + if (!Number.isFinite(ratio) || ratio < 0.3 || ratio > 0.98) { + throw new Error(`${name} must be a finite ratio in [0.3, 0.98]`); + } + } + if (c.softPressureRatio >= c.hardPressureRatio) { + throw new Error('softPressureRatio must be < hardPressureRatio'); + } + if (c.hardPressureRatio >= c.criticalRatio) { + throw new Error('hardPressureRatio must be < criticalRatio'); + } + if (!Number.isFinite(c.cleanupCooldownMs) || c.cleanupCooldownMs < 0) { + throw new Error('cleanupCooldownMs must be a non-negative number'); + } +} + +const debugLogger = createDebugLogger('MEMORY_PRESSURE'); +const MIN_CGROUP_MEMORY_LIMIT = 64 * 1024 * 1024; + +// Monitor + +export class MemoryPressureMonitor extends EventEmitter { + private readonly config: MemoryPressureConfig; + private readonly coreConfig: Config; + + private pendingCheck = false; + private cleanupInProgress = false; + private activeCleanupAction: CleanupRecommendation['action'] = 'none'; + private lastCleanupAction: CleanupRecommendation['action'] = 'none'; + private queuedCleanupRecommendation?: CleanupRecommendation; + private lastCleanupTime = 0; + private consecutiveCleanupFailures = 0; + private consecutiveIneffectiveCleanups = 0; + private consecutiveIneffectiveAggressiveCleanups = 0; + private cleanupGeneration = 0; + private readonly effectiveMemoryLimit: number; + + constructor(coreConfig: Config, pressureConfig?: MemoryPressureConfig) { + super(); + this.coreConfig = coreConfig; + this.config = { ...(pressureConfig ?? DEFAULT_PRESSURE_CONFIG) }; + validateMemoryPressureConfig(this.config); + this.effectiveMemoryLimit = this.computeEffectiveMemoryLimit(); + const heapSizeLimit = getHeapStatistics().heap_size_limit; + debugLogger.info( + `Effective memory limit: ${formatMiB(this.effectiveMemoryLimit)} MiB; ` + + `V8 heap limit: ${formatMiB(heapSizeLimit)} MiB`, + ); + if (this.effectiveMemoryLimit <= 0) { + debugLogger.warn( + 'Effective memory limit is not positive; RSS pressure checks are disabled', + ); + } + } + + // Public API + + getConsecutiveFailures(): number { + return this.consecutiveCleanupFailures; + } + + resetConsecutiveFailures(): void { + this.consecutiveCleanupFailures = 0; + this.consecutiveIneffectiveCleanups = 0; + this.consecutiveIneffectiveAggressiveCleanups = 0; + } + + /** + * Reset session-scoped cleanup state and invalidate any async cleanup tail + * that was queued against the previous session's cache. + */ + resetForNewSession(): void { + this.cleanupGeneration++; + this.resetConsecutiveFailures(); + this.cleanupInProgress = false; + this.activeCleanupAction = 'none'; + this.queuedCleanupRecommendation = undefined; + this.lastCleanupAction = 'none'; + this.lastCleanupTime = 0; + } + + /** + * Schedule a deferred memory check after a tool finishes execution. + * Uses queueMicrotask to batch checks across concurrently-completing + * tools within the same event-loop tick. + */ + scheduleCheck(): void { + if (this.pendingCheck) return; + this.pendingCheck = true; + queueMicrotask(() => { + try { + this.performCheck(); + } finally { + this.pendingCheck = false; + } + }); + } + + /** Force an immediate check (e.g. after a concurrent batch completes). */ + performCheck(): void { + try { + this.performCheckInternal(); + } catch (err) { + debugLogger.error( + `Memory pressure check failed: ${getErrorMessage(err)}`, + ); + } + } + + private performCheckInternal(): void { + const pressure = this.getPressureLevel(); + if (pressure !== 'critical') { + this.consecutiveIneffectiveAggressiveCleanups = 0; + } + if (pressure === 'normal') return; + + const recommendation = this.recommendCleanup(pressure); + if (recommendation.action === 'none') return; + + const now = Date.now(); + const isEscalation = + cleanupActionRank(recommendation.action) > + cleanupActionRank(this.lastCleanupAction); + const cleanupCooldownMs = this.getCleanupCooldownMs(recommendation.action); + if (!isEscalation && now - this.lastCleanupTime < cleanupCooldownMs) { + return; + } + + this.executeCleanup(recommendation); + } + + /** + * Determine the current memory pressure level from the stronger of: + * - RSS as a fraction of the effective memory limit (cgroup-aware). + * - V8 heap usage as a fraction of V8's heap size limit. + */ + getPressureLevel(): 'normal' | 'soft' | 'hard' | 'critical' { + let mem: ReturnType; + try { + mem = process.memoryUsage(); + } catch (err) { + debugLogger.error( + `Failed to read memory usage for pressure check: ${getErrorMessage(err)}`, + ); + return 'normal'; + } + + const rssRatio = + this.effectiveMemoryLimit > 0 ? mem.rss / this.effectiveMemoryLimit : 0; + const heapSizeLimit = getHeapStatistics().heap_size_limit; + const heapRatio = heapSizeLimit > 0 ? mem.heapUsed / heapSizeLimit : 0; + const ratio = Math.max(rssRatio, heapRatio); + + if (ratio >= this.config.criticalRatio) return 'critical'; + if (ratio >= this.config.hardPressureRatio) return 'hard'; + if (ratio >= this.config.softPressureRatio) return 'soft'; + return 'normal'; + } + + // Cleanup + + private recommendCleanup( + pressure: 'soft' | 'hard' | 'critical', + ): CleanupRecommendation { + switch (pressure) { + case 'critical': + return { + action: 'aggressive', + steps: [ + 'clear_file_cache', + ...(this.config.enableExplicitGC ? ['trigger_gc' as const] : []), + ], + }; + case 'hard': + return { + action: 'moderate', + steps: ['evict_cold_cache'], + }; + case 'soft': + return { + action: 'light', + steps: ['evict_stale_cache'], + }; + default: + return assertNever(pressure); + } + } + + private executeCleanup(recommendation: CleanupRecommendation): void { + if (this.cleanupInProgress) { + const recommendationRank = cleanupActionRank(recommendation.action); + const activeRank = cleanupActionRank(this.activeCleanupAction); + const queuedRank = this.queuedCleanupRecommendation + ? cleanupActionRank(this.queuedCleanupRecommendation.action) + : 0; + if (recommendationRank > activeRank && recommendationRank > queuedRank) { + this.queuedCleanupRecommendation = recommendation; + debugLogger.debug( + `Queued escalated cleanup "${recommendation.action}" while ` + + `"${this.activeCleanupAction}" is in progress`, + ); + } else { + debugLogger.debug('Cleanup already in progress, skipping'); + } + return; + } + let memBefore: number; + try { + memBefore = process.memoryUsage().rss; + } catch (err) { + this.recordCleanupFailure(recommendation, err); + return; + } + + this.cleanupInProgress = true; + this.activeCleanupAction = recommendation.action; + this.lastCleanupAction = recommendation.action; + this.lastCleanupTime = Date.now(); + const cleanupGeneration = this.cleanupGeneration; + + void this.runCleanupSteps(recommendation.steps, cleanupGeneration) + .then(() => { + if (cleanupGeneration !== this.cleanupGeneration) { + return; + } + this.consecutiveCleanupFailures = 0; + setImmediate(() => { + if (cleanupGeneration !== this.cleanupGeneration) { + return; + } + try { + const memAfter = process.memoryUsage().rss; + this.logCleanupResult(memBefore, memAfter, recommendation); + } catch (err) { + debugLogger.error( + `Cleanup measurement failed: ${getErrorMessage(err)}`, + ); + } finally { + this.finishCleanupAndRunQueued(cleanupGeneration); + } + }); + }) + .catch((err) => { + if (cleanupGeneration !== this.cleanupGeneration) { + return; + } + this.recordCleanupFailure(recommendation, err); + this.finishCleanupAndRunQueued(cleanupGeneration); + }); + } + + private finishCleanupAndRunQueued(cleanupGeneration: number): void { + if (cleanupGeneration !== this.cleanupGeneration) { + return; + } + this.cleanupInProgress = false; + this.activeCleanupAction = 'none'; + const queuedRecommendation = this.queuedCleanupRecommendation; + this.queuedCleanupRecommendation = undefined; + if (queuedRecommendation) { + try { + this.executeCleanup(queuedRecommendation); + } catch (err) { + this.recordCleanupFailure(queuedRecommendation, err); + } + } + } + + private getCleanupCooldownMs( + action: CleanupRecommendation['action'], + ): number { + const baseCooldownMs = this.config.cleanupCooldownMs; + if ( + action !== 'aggressive' || + baseCooldownMs === 0 || + this.consecutiveIneffectiveAggressiveCleanups < 3 + ) { + return baseCooldownMs; + } + + const exponent = Math.min( + this.consecutiveIneffectiveAggressiveCleanups - 2, + 6, + ); + return baseCooldownMs * 2 ** exponent; + } + + private logCleanupResult( + memBefore: number, + memAfter: number, + recommendation: CleanupRecommendation, + ): void { + const freed = memBefore - memAfter; + const freedRatio = memBefore > 0 ? freed / memBefore : 0; + + debugLogger.info( + `Cleanup "${recommendation.action}" completed; RSS delta ${freed} bytes ` + + `(${(freedRatio * 100).toFixed(1)}%)`, + ); + + if (freedRatio < 0.01) { + this.consecutiveIneffectiveCleanups++; + if (recommendation.action === 'aggressive') { + this.consecutiveIneffectiveAggressiveCleanups++; + } + if (shouldEmitRepeatedDiagnostic(this.consecutiveIneffectiveCleanups)) { + const event = { + rss: memAfter, + freedBytes: freed, + freedRatio, + consecutiveIneffectiveCleanups: this.consecutiveIneffectiveCleanups, + recommendation, + } satisfies MemoryCleanupIneffectiveEvent; + debugLogger.warn( + `Cleanup "${recommendation.action}" has been ineffective ` + + `${this.consecutiveIneffectiveCleanups} times consecutively`, + ); + this.emitSafely('memory-cleanup-ineffective', event); + } + return; + } + + this.consecutiveIneffectiveCleanups = 0; + if (recommendation.action === 'aggressive') { + this.consecutiveIneffectiveAggressiveCleanups = 0; + } + } + + private recordCleanupFailure( + recommendation: CleanupRecommendation, + err: unknown, + ): void { + const error = getErrorMessage(err); + let rss = 0; + try { + rss = process.memoryUsage().rss; + } catch (rssErr) { + debugLogger.error( + `Failed to read RSS after cleanup failure: ${getErrorMessage(rssErr)}`, + ); + } + + this.consecutiveCleanupFailures++; + debugLogger.error( + `Cleanup "${recommendation.action}" failed: ${error}; ` + + `consecutive failures: ${this.consecutiveCleanupFailures}`, + ); + + if (shouldEmitRepeatedDiagnostic(this.consecutiveCleanupFailures)) { + this.emitSafely('memory-cleanup-failed', { + rss, + consecutiveFailures: this.consecutiveCleanupFailures, + recommendation, + error, + } satisfies MemoryCleanupFailureEvent); + } + } + + private emitSafely(eventName: string, event: unknown): void { + try { + this.emit(eventName, event); + } catch (err) { + debugLogger.error(`${eventName} handler threw: ${getErrorMessage(err)}`); + } + } + + private async runCleanupSteps( + steps: CleanupStep[], + cleanupGeneration: number, + ): Promise { + for (const step of steps) { + if (cleanupGeneration !== this.cleanupGeneration) { + return; + } + this.executeStep(step); + // Keep a promise boundary between steps so escalated cleanups can queue + // behind the active cleanup instead of interleaving in the same stack. + await Promise.resolve(); + } + } + + private executeStep(step: CleanupStep): void { + switch (step) { + case 'clear_file_cache': { + this.coreConfig.getFileReadCache().clear(); + debugLogger.debug('FileReadCache cleared'); + break; + } + case 'evict_cold_cache': { + const evicted = this.coreConfig + .getFileReadCache() + .evictNotAccessedSince(30); + debugLogger.debug(`FileReadCache cold eviction: ${evicted} entries`); + break; + } + case 'evict_stale_cache': { + const evicted = this.coreConfig + .getFileReadCache() + .evictNotAccessedSince(60); + debugLogger.debug(`FileReadCache stale eviction: ${evicted} entries`); + break; + } + case 'trigger_gc': { + if (typeof global.gc === 'function') { + const before = process.memoryUsage().rss; + global.gc(); + const after = process.memoryUsage().rss; + debugLogger.debug(`global.gc() freed ${before - after} bytes`); + } else { + debugLogger.warn( + 'trigger_gc requested but global.gc is not available; ' + + 'start Node.js with --expose-gc', + ); + } + break; + } + default: + return assertNever(step); + } + } + + // Memory metrics + + private computeEffectiveMemoryLimit(): number { + const hostTotal = os.totalmem(); + const cgroupV2Limit = this.readCgroupMemoryLimit( + '/sys/fs/cgroup/memory.max', + hostTotal, + ); + if (cgroupV2Limit !== undefined) { + debugLogger.info( + `Using cgroup v2 memory limit: ${formatMiB(cgroupV2Limit)} MiB`, + ); + return cgroupV2Limit; + } + + const cgroupV1Limit = this.readCgroupMemoryLimit( + '/sys/fs/cgroup/memory/memory.limit_in_bytes', + hostTotal, + ); + if (cgroupV1Limit !== undefined) { + debugLogger.info( + `Using cgroup v1 memory limit: ${formatMiB(cgroupV1Limit)} MiB`, + ); + return cgroupV1Limit; + } + + debugLogger.info(`Using host memory limit: ${formatMiB(hostTotal)} MiB`); + return hostTotal; + } + + private readCgroupMemoryLimit( + filePath: string, + hostTotal: number, + ): number | undefined { + try { + const raw = readFileSync(filePath, 'utf-8').trim(); + if (raw === 'max') return undefined; + + if (!/^-?\d+$/.test(raw)) { + debugLogger.warn( + `Ignoring non-numeric cgroup memory limit from ${filePath}: ${raw}`, + ); + return undefined; + } + + const limit = Number(raw); + if (!Number.isFinite(limit) || limit <= 0) { + debugLogger.warn( + `Ignoring out-of-range cgroup memory limit from ${filePath}: ${raw}`, + ); + return undefined; + } + if (limit < MIN_CGROUP_MEMORY_LIMIT) { + debugLogger.warn( + `Ignoring unrealistically small cgroup memory limit from ` + + `${filePath}: ${raw}`, + ); + return undefined; + } + + // cgroup v1 represents "unlimited" with huge sentinel values. + if (!Number.isSafeInteger(limit)) { + debugLogger.debug( + `Ignoring unlimited cgroup memory limit from ${filePath}: ${raw}`, + ); + return undefined; + } + if (hostTotal > 0 && limit > hostTotal) { + debugLogger.debug( + `Ignoring cgroup memory limit above host total from ${filePath}: ` + + `${raw}`, + ); + return undefined; + } + + return limit; + } catch (err) { + debugLogger.debug( + `Failed to read cgroup memory limit from ${filePath}: ` + + getErrorMessage(err), + ); + return undefined; + } + } +} + +function cleanupActionRank(action: CleanupRecommendation['action']): number { + switch (action) { + case 'aggressive': + return 3; + case 'moderate': + return 2; + case 'light': + return 1; + case 'none': + return 0; + default: + return assertNever(action); + } +} + +function shouldEmitRepeatedDiagnostic(count: number): boolean { + // Emit once at the threshold, once soon after, then every 20 repeats so + // sustained pressure remains visible without flooding event listeners. + return count === 3 || count === 10 || (count > 10 && count % 20 === 0); +} + +function formatMiB(bytes: number): string { + return (bytes / 1024 / 1024).toFixed(0); +} + +function assertNever(value: never): never { + throw new Error(`Unhandled memory pressure monitor value: ${String(value)}`); +}