diff --git a/packages/common/src/log-sanitize.ts b/packages/common/src/log-sanitize.ts index 654d2894..5b7bb09b 100644 --- a/packages/common/src/log-sanitize.ts +++ b/packages/common/src/log-sanitize.ts @@ -82,34 +82,40 @@ function skipEscSequence(line: string, start: number): number { * invisible payload into persisted or downloaded logs. */ export function sanitizeLogLine(line: string): string { - let plain = ''; + const parts: string[] = []; + let start = 0; for (let i = 0; i < line.length;) { const code = line.charCodeAt(i); + let next: number; if (code === ESC) { - i = skipEscSequence(line, i + 1); - continue; - } - - if (code === C1_CSI) { - i = skipCsi(line, i + 1); - continue; - } - - if (code === C1_OSC || C1_STRING_STARTERS.has(code)) { - i = skipControlString(line, i + 1, code === C1_OSC); - continue; - } - - if ((code < 0x20 && code !== 0x09 && code !== 0x0a) || (code >= 0x7f && code <= 0x9f)) { + next = skipEscSequence(line, i + 1); + } else if (code === C1_CSI) { + next = skipCsi(line, i + 1); + } else if (code === C1_OSC || C1_STRING_STARTERS.has(code)) { + next = skipControlString(line, i + 1, code === C1_OSC); + } else if ( + (code < 0x20 && code !== 0x09 && code !== 0x0a) || + (code >= 0x7f && code <= 0x9f) + ) { + next = i + 1; + } else { i += 1; continue; } - plain += line[i]; - i += 1; + parts.push(line.slice(start, i)); + start = next; + i = next; } - return plain; + if (start === 0) { + return line; + } + + // Join contiguous spans once. Appending each character creates a retained + // V8 cons-string node per character in the in-memory log rings. + parts.push(line.slice(start)); + return parts.join(''); } diff --git a/packages/common/src/single-flight-cache.ts b/packages/common/src/single-flight-cache.ts index fca5f230..f6cc994a 100644 --- a/packages/common/src/single-flight-cache.ts +++ b/packages/common/src/single-flight-cache.ts @@ -1,7 +1,7 @@ /** TTL caching and request deduplication to limit repeated upstream requests. */ export interface SingleFlightCacheOptions { - /** Freshness window in ms. A cached entry older than this is reloaded. */ + /** Freshness window in ms. Expired values are released even while idle. */ ttlMs: number; /** Load the value for a key. Rejections propagate and are never cached. */ load: (key: K) => Promise; @@ -22,6 +22,7 @@ export class SingleFlightCache { private readonly now: () => number; private readonly fresh = new Map(); private readonly inflight = new Map>(); + private expiryTimer: ReturnType | undefined; constructor(opts: SingleFlightCacheOptions) { this.ttlMs = opts.ttlMs; @@ -30,6 +31,11 @@ export class SingleFlightCache { this.now = opts.now ?? Date.now; } + /** Number of retained values; pending loads are not counted. */ + get size(): number { + return this.fresh.size; + } + /** * Get the value for `key`. * @@ -41,10 +47,13 @@ export class SingleFlightCache { * rather than starting a second one. */ get(key: K, opts: { force?: boolean } = {}): Promise { - if (!opts.force) { - const hit = this.fresh.get(key); + const hit = this.fresh.get(key); - if (hit && this.now() - hit.at < this.ttlMs) { + if (hit) { + if (this.now() - hit.at >= this.ttlMs) { + this.fresh.delete(key); + this.cancelExpiryIfEmpty(); + } else if (!opts.force) { return Promise.resolve(hit.value); } } @@ -58,8 +67,11 @@ export class SingleFlightCache { const load = async (): Promise => { const value = await this.load(key); - if (this.inflight.get(key) === task && this.shouldCache(value)) { + if (this.inflight.get(key) === task && this.ttlMs > 0 && this.shouldCache(value)) { + // Keep insertion order aligned with expiry, including refreshes. + this.fresh.delete(key); this.fresh.set(key, { at: this.now(), value }); + this.scheduleExpiry(); } return value; @@ -78,12 +90,52 @@ export class SingleFlightCache { invalidate(key: K): void { this.fresh.delete(key); this.inflight.delete(key); + this.cancelExpiryIfEmpty(); } /** Detach all entries and pending loads without aborting existing callers. */ clear(): void { this.fresh.clear(); this.inflight.clear(); + this.cancelExpiryIfEmpty(); + } + + private cancelExpiryIfEmpty(): void { + if (this.fresh.size === 0 && this.expiryTimer !== undefined) { + clearTimeout(this.expiryTimer); + this.expiryTimer = undefined; + } + } + + private scheduleExpiry(): void { + if (this.expiryTimer !== undefined || !Number.isFinite(this.ttlMs)) { + return; + } + + const oldest = this.fresh.values().next().value; + + if (!oldest) { + return; + } + + // One unref'ed timer per cache, not one closure/timer per cached roster. + // Clamp long TTLs to Node's timer range rather than overflowing to 1ms. + const delay = Math.max(1, Math.min(2 ** 31 - 1, oldest.at + this.ttlMs - this.now())); + this.expiryTimer = setTimeout(() => { + this.expiryTimer = undefined; + const now = this.now(); + + for (const [key, entry] of this.fresh) { + if (now - entry.at < this.ttlMs) { + break; + } + + this.fresh.delete(key); + } + + this.scheduleExpiry(); + }, delay); + this.expiryTimer.unref(); } } diff --git a/packages/common/tests/log-sanitize.test.ts b/packages/common/tests/log-sanitize.test.ts index c0d71bcb..a7a27fa9 100644 --- a/packages/common/tests/log-sanitize.test.ts +++ b/packages/common/tests/log-sanitize.test.ts @@ -23,4 +23,25 @@ describe('sanitizeLogLine', () => { it('does not leak unterminated terminal strings', () => { expect(sanitizeLogLine('visible\x1b]8;;https://example.com')).toBe('visible'); }); + + it('preserves long Unicode records and adjacent visible spans', () => { + const body = '消息 🌌 e\u0301\tpacket=0011aabb\n'.repeat(1000); + expect(sanitizeLogLine(body)).toBe(body); + expect(sanitizeLogLine(`\x1b[32m${body}\x1b[0m\x00tail\x1b`)).toBe(`${body}tail`); + }); + + it.each(['\x1bP', '\x1bX', '\x1b^', '\x1b_', '\u0090', '\u0098', '\u009e', '\u009f'])( + 'removes %j control strings between visible spans', + (opener) => { + expect(sanitizeLogLine(`before${opener}private\u009cafter`)).toBe('beforeafter'); + expect(sanitizeLogLine(`before${opener}private\x1b\\after`)).toBe('beforeafter'); + expect(sanitizeLogLine(`before${opener}private`)).toBe('before'); + }, + ); + + it('handles consecutive controls and records consisting only of controls', () => { + expect(sanitizeLogLine('\x00\x1b[31m\u009b0m\r\u007f')).toBe(''); + expect(sanitizeLogLine('a\x00\x1b[31m\u009b0m\r\u007fb')).toBe('ab'); + expect(sanitizeLogLine('')).toBe(''); + }); }); diff --git a/packages/common/tests/single-flight-cache.test.ts b/packages/common/tests/single-flight-cache.test.ts index 7a08b0cc..3c7f3c41 100644 --- a/packages/common/tests/single-flight-cache.test.ts +++ b/packages/common/tests/single-flight-cache.test.ts @@ -1,4 +1,4 @@ -import { describe, it, expect, vi } from 'vitest'; +import { afterEach, describe, it, expect, vi } from 'vitest'; import { createSingleFlightCache } from '../src/single-flight-cache'; /** A promise whose resolve/reject is exposed, for driving single-flight races. */ @@ -13,6 +13,87 @@ function deferred() { } describe('SingleFlightCache', () => { + afterEach(() => { + vi.useRealTimers(); + }); + + it('releases expired values while idle without expiring newer values early', async () => { + vi.useFakeTimers(); + const load = vi.fn(async (key: string) => ({ key })); + const cache = createSingleFlightCache({ ttlMs: 1000, load }); + await cache.get('old'); + await vi.advanceTimersByTimeAsync(500); + const newer = await cache.get('new'); + + expect(cache.size).toBe(2); + await vi.advanceTimersByTimeAsync(500); + expect(cache.size).toBe(1); + expect(await cache.get('new')).toBe(newer); + expect(load).toHaveBeenCalledTimes(2); + + await vi.advanceTimersByTimeAsync(500); + expect(cache.size).toBe(0); + expect(vi.getTimerCount()).toBe(0); + }); + + it('keeps a refreshed entry until its new deadline and expires other keys', async () => { + vi.useFakeTimers(); + const cache = createSingleFlightCache({ ttlMs: 1000, load: async (key: string) => key }); + await cache.get('refreshed'); + await cache.get('old'); + await vi.advanceTimersByTimeAsync(500); + await cache.get('refreshed', { force: true }); + + await vi.advanceTimersByTimeAsync(500); + expect(cache.size).toBe(1); + await vi.advanceTimersByTimeAsync(500); + expect(cache.size).toBe(0); + }); + + it.each(['invalidate', 'clear'] as const)( + '%s cancels idle cleanup when the last value is removed', + async (operation) => { + vi.useFakeTimers(); + const cache = createSingleFlightCache({ ttlMs: 1000, load: async () => 'value' }); + await cache.get('key'); + expect(vi.getTimerCount()).toBe(1); + cache[operation]('key'); + expect(cache.size).toBe(0); + expect(vi.getTimerCount()).toBe(0); + + await cache.get('key'); + await vi.advanceTimersByTimeAsync(1000); + expect(cache.size).toBe(0); + }, + ); + + it('does not let expiration detach a pending refresh', async () => { + vi.useFakeTimers(); + const pending = deferred(); + const load = vi.fn().mockResolvedValueOnce('old').mockReturnValue(pending.promise); + const cache = createSingleFlightCache({ ttlMs: 1000, load }); + await cache.get('key'); + const refresh = cache.get('key', { force: true }); + await vi.advanceTimersByTimeAsync(1000); + + expect(cache.size).toBe(0); + expect(cache.get('key')).toBe(refresh); + pending.resolve('new'); + await expect(refresh).resolves.toBe('new'); + expect(cache.size).toBe(1); + expect(load).toHaveBeenCalledTimes(2); + await vi.advanceTimersByTimeAsync(1000); + expect(cache.size).toBe(0); + }); + + it('does not retain values when caching is disabled by a zero TTL', async () => { + vi.useFakeTimers(); + const cache = createSingleFlightCache({ ttlMs: 0, load: async () => 'value' }); + await expect(cache.get('key')).resolves.toBe('value'); + expect(cache.size).toBe(0); + expect(vi.getTimerCount()).toBe(0); + }); + it.each(['invalidate', 'clear'] as const)( '%s detaches pending loads from the cache', async (operation) => { diff --git a/tools/bench-common-memory.mjs b/tools/bench-common-memory.mjs new file mode 100644 index 00000000..05a9eb35 --- /dev/null +++ b/tools/bench-common-memory.mjs @@ -0,0 +1,78 @@ +// Isolated retained-heap measurements; no QQ session, disk logs or live requests. +// node --expose-gc tools/bench-common-memory.mjs logs|colored-logs|cache [source-directory] +// The optional directory contains log-sanitize.ts and single-flight-cache.ts, +// allowing the same workload to run against sources from a baseline revision. +import { performance } from 'node:perf_hooks'; +import { setImmediate, setTimeout } from 'node:timers/promises'; +import { resolve } from 'node:path'; +import { pathToFileURL } from 'node:url'; + +if (!global.gc) { + throw new Error('Run with node --expose-gc'); +} + +const mode = process.argv[2] ?? 'logs'; +const source = (name) => + process.argv[3] + ? pathToFileURL(resolve(process.argv[3], name)) + : new URL(`../packages/common/src/${name}`, import.meta.url); + +async function heapUsed() { + await setImmediate(); + global.gc(); + global.gc(); + return process.memoryUsage().heapUsed; +} + +const mib = (bytes) => Number((bytes / 1024 / 1024).toFixed(2)); + +if (mode === 'logs' || mode === 'colored-logs') { + const { sanitizeLogLine } = await import(source('log-sanitize.ts')); + const before = await heapUsed(); + const records = []; + const started = performance.now(); + + for (let i = 0; i < 1000; i += 1) { + const body = `${i}: ${'0123456789abcdef'.repeat(512)}`; + const line = mode === 'colored-logs' ? `\x1b[32m${body}\x1b[0m` : body; + records.push(sanitizeLogLine(line)); + } + + const elapsedMs = Number((performance.now() - started).toFixed(1)); + const retainedHeapMiB = mib((await heapUsed()) - before); + console.log(JSON.stringify({ mode, records: records.length, retainedHeapMiB, elapsedMs })); +} else if (mode === 'cache') { + const { createSingleFlightCache } = await import(source('single-flight-cache.ts')); + const ttlMs = 2000; + const cache = createSingleFlightCache({ + ttlMs, + load: async (key) => + Array.from({ length: 1000 }, (_, i) => ({ + uin: key * 1000 + i, + nickname: `member-${key}-${i}`, + })), + }); + const before = await heapUsed(); + + for (let key = 0; key < 500; key += 1) { + await cache.get(key); + } + + const populatedHeapMiB = mib((await heapUsed()) - before); + await setTimeout(ttlMs + 100); + const idleHeapMiB = mib((await heapUsed()) - before); + // Keep the cache itself live through both samples, as it is in ContactsApi. + cache.clear(); + console.log( + JSON.stringify({ + mode, + rosters: 500, + membersPerRoster: 1000, + ttlMs, + populatedHeapMiB, + idleHeapMiB, + }), + ); +} else { + throw new Error(`Unknown workload: ${mode}`); +}