perf(common): reduce retained memory in logs and TTL caches (#652)

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);
+            next = skipEscSequence(line, i + 1);
-            continue;
+        } else if (code === C1_CSI) {
-        }
+            next = skipCsi(line, i + 1);
-
+        } else if (code === C1_OSC || C1_STRING_STARTERS.has(code)) {
-        if (code === C1_CSI) {
+            next = skipControlString(line, i + 1, code === C1_OSC);
-            i = skipCsi(line, i + 1);
+        } else if (
-            continue;
+            (code < 0x20 && code !== 0x09 && code !== 0x0a) ||
-        }
+            (code >= 0x7f && code <= 0x9f)
-
+        ) {
-        if (code === C1_OSC || C1_STRING_STARTERS.has(code)) {
+            next = i + 1;
-            i = skipControlString(line, i + 1, code === C1_OSC);
+        } else {
-            continue;
-        }
-
-        if ((code < 0x20 && code !== 0x09 && code !== 0x0a) || (code >= 0x7f && code <= 0x9f)) {
             i += 1;
             continue;
         }
 
-        plain += line[i];
+        parts.push(line.slice(start, i));
-        i += 1;
+        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('');
 }

packages/common/src/single-flight-cache.ts

@@ -1,7 +1,7 @@
 /** TTL caching and request deduplication to limit repeated upstream requests. */
 
 export interface SingleFlightCacheOptions<K, V> {
-    /** 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<V>;
@@ -22,6 +22,7 @@ export class SingleFlightCache<K, V> {
     private readonly now: () => number;
     private readonly fresh = new Map<K, { at: number; value: V }>();
     private readonly inflight = new Map<K, Promise<V>>();
+    private expiryTimer: ReturnType<typeof setTimeout> | undefined;
 
     constructor(opts: SingleFlightCacheOptions<K, V>) {
         this.ttlMs = opts.ttlMs;
@@ -30,6 +31,11 @@ export class SingleFlightCache<K, V> {
         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<K, V> {
      *   rather than starting a second one.
      */
     get(key: K, opts: { force?: boolean } = {}): Promise<V> {
-        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<K, V> {
         const load = async (): Promise<V> => {
             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<K, V> {
     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();
     }
 }
 

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('');
+    });
 });

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<T>() {
 }
 
 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<string>();
+        const load = vi.fn().mockResolvedValueOnce('old').mockReturnValue(pending.promise);
+        const cache = createSingleFlightCache<string, string>({ 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) => {

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}`);
+}