fix(rolldown): bound reader cache retention by log size

packages/rolldown/src/node/rolldown/__tests__/logs-manager.test.ts

@@ -0,0 +1,54 @@
+import fs from 'node:fs'
+import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'pathe'
+import { expect, it, vi } from 'vitest'
+import { RolldownEventsReader } from '../events-reader'
+import { RolldownLogsManager } from '../logs-manager'
+
+it.each(['loadSession', 'loadSessionSummary', 'loadPackageSession'] as const)(
+  '%s retains a large session across metadata reads and repeated requests',
+  async (method) => {
+    const dir = await mkdtemp(join(tmpdir(), 'rolldown-reader-cache-'))
+    const readers = new Set<RolldownEventsReader>()
+    const stat = fs.promises.stat.bind(fs.promises)
+    const get = RolldownEventsReader.get.bind(RolldownEventsReader)
+    vi.spyOn(RolldownEventsReader, 'get').mockImplementation((...args) => {
+      const reader = get(...args)
+      readers.add(reader)
+      return reader
+    })
+    // Exercise the real file readers without allocating a multi-GiB fixture.
+    vi.spyOn(fs.promises, 'stat').mockImplementation(async (...args) => {
+      const result = await stat(...args)
+      if (result && String(args[0]).endsWith('/logs.json'))
+        result.size = 300 * 1024 * 1024
+      return result
+    })
+    try {
+      for (const session of ['first', 'second']) {
+        await mkdir(join(dir, session))
+        await writeFile(join(dir, session, 'logs.json'), `${JSON.stringify({ action: 'BuildStart', timestamp: '1772529438599', session_id: session })}\n`)
+        await writeFile(join(dir, session, 'meta.json'), `${JSON.stringify({ action: 'SessionMeta', timestamp: '1772529438599', session_id: session })}\n`)
+      }
+      const manager = new RolldownLogsManager(dir)
+      const first = await manager[method]('first')
+      const filepath = join(dir, 'first', 'logs.json')
+      const key = method === 'loadPackageSession' ? `${filepath}:package-summary` : filepath
+      expect(first.meta).toMatchObject({ session_id: 'first' })
+      expect(RolldownEventsReader.peek(filepath, key)).toBe(first)
+      expect(await manager[method]('first')).toBe(first)
+
+      const second = await manager[method]('second')
+      expect(second.meta).toMatchObject({ session_id: 'second' })
+      expect(RolldownEventsReader.peek(filepath, key)).toBeUndefined()
+      expect(await manager[method]('second')).toBe(second)
+    }
+    finally {
+      for (const reader of readers)
+        reader.dispose()
+      vi.restoreAllMocks()
+      await rm(dir, { recursive: true, force: true })
+    }
+  },
+)

packages/rolldown/src/node/rolldown/__tests__/reader-cache.test.ts

@@ -0,0 +1,93 @@
+import type { Stats } from 'node:fs'
+import fs from 'node:fs'
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import { RolldownEventsReader } from '../events-reader'
+
+const created: RolldownEventsReader[] = []
+const MiB = 1024 * 1024
+
+function reader(name: string, bytes = 0) {
+  const result = RolldownEventsReader.get(`/mock/${name}/logs.json`)
+  result.logBytes = bytes
+  created.push(result)
+  return result
+}
+
+afterEach(() => {
+  for (const item of created)
+    item.dispose()
+  created.length = 0
+  vi.restoreAllMocks()
+})
+
+describe('rolldown reader cache', () => {
+  it('evicts the least recently used reader when the source byte budget is exceeded', () => {
+    const first = reader('first', 100 * MiB)
+    const second = reader('second', 100 * MiB)
+    const third = reader('third', 100 * MiB)
+    expect(RolldownEventsReader.get(first.filepath)).toBe(first)
+    expect(RolldownEventsReader.peek(second.filepath)).toBeUndefined()
+    expect(RolldownEventsReader.peek(third.filepath)).toBe(third)
+  })
+
+  it('retains one oversized current reader while evicting older idle readers', () => {
+    const older = reader('older', MiB)
+    const large = reader('large', 5 * 1024 * MiB)
+    RolldownEventsReader.get(large.filepath)
+    expect(RolldownEventsReader.peek(older.filepath)).toBeUndefined()
+    expect(RolldownEventsReader.peek(large.filepath)).toBe(large)
+  })
+
+  it('keeps the count limit and preserves data still held by a caller', () => {
+    const first = reader('first')
+    first.manager.handleEvent({ action: 'BuildStart', timestamp: '1772529438599', session_id: '5173' })
+    for (let i = 0; i < 32; i++)
+      reader(`next-${i}`)
+    expect(RolldownEventsReader.peek(first.filepath)).toBeUndefined()
+    expect(first.manager.eventCount).toBe(1)
+  })
+
+  it('does not remove a replacement when an evicted reader is disposed', () => {
+    const old = reader('same', 300 * MiB)
+    reader('other')
+    const replacement = reader('same')
+    expect(replacement).not.toBe(old)
+    old.dispose()
+    expect(RolldownEventsReader.peek(old.filepath)).toBe(replacement)
+  })
+
+  it.each([
+    ['read', 'restoreCompleteSession'],
+    ['readSummary', 'restoreSummary'],
+    ['readPackageSummary', 'restorePackageSummary'],
+  ] as const)('accounts for source size when %s restores a disk cache', async (method, restore) => {
+    const older = reader('older', MiB)
+    const current = reader('current')
+    vi.spyOn(fs.promises, 'stat').mockResolvedValue({ size: 300 * MiB, mtime: new Date(1) } as Stats)
+    Object.defineProperty(current, 'logCache', {
+      value: { [restore]: vi.fn().mockResolvedValue(true), resetCompleteSessionWriteAttempt: vi.fn(), resetPackageSummaryWriteAttempt: vi.fn() },
+    })
+    await current[method]()
+    expect(current.logBytes).toBe(300 * MiB)
+    expect(RolldownEventsReader.peek(older.filepath)).toBeUndefined()
+    expect(RolldownEventsReader.peek(current.filepath)).toBe(current)
+  })
+
+  it.each(['read', 'readSummary', 'readPackageSummary'] as const)('protects a pending %s and prunes after it finishes', async (method) => {
+    const pending = reader('pending', 300 * MiB)
+    let finish!: () => void
+    const completed = new Promise<void>((resolve) => {
+      finish = resolve
+    })
+    const implementation = { read: 'readChanges', readSummary: 'readSummaryChanges', readPackageSummary: 'readPackageSummaryChanges' }[method]
+    Object.defineProperty(pending, implementation, { value: () => completed })
+    const reading = pending[method]()
+    const idle = reader('idle', MiB)
+    expect(RolldownEventsReader.peek(pending.filepath)).toBe(pending)
+    expect(RolldownEventsReader.peek(idle.filepath)).toBe(idle)
+    finish()
+    await reading
+    expect(RolldownEventsReader.peek(pending.filepath)).toBe(pending)
+    expect(RolldownEventsReader.peek(idle.filepath)).toBeUndefined()
+  })
+})

packages/rolldown/src/node/rolldown/events-reader.ts

@@ -5,13 +5,16 @@ import { Buffer } from 'node:buffer'
 import { createHash } from 'node:crypto'
 import fs from 'node:fs'
 import { parseToEvent } from '@rolldown/debug'
+import { dirname } from 'pathe'
 import { diagnostics } from '../diagnostics'
 import { getContentByteSize } from '../utils/format'
 import { getContentRef, RolldownEventsManager } from './events-manager'
 import { RolldownLogCache } from './log-cache'
 
 const readers: Map<string, RolldownEventsReader> = new Map()
 const MAX_READERS = 32
+// Source log size is a cache weight, not a measurement of retained heap.
+const MAX_READER_LOG_BYTES = 256 * 1024 * 1024
 const MAX_MODULE_METRICS_CACHE = 32
 const MAX_MODULE_METRICS_CACHE_BYTES = 64 * 1024 * 1024
 const READ_STREAM_HIGH_WATER_MARK = 1024 * 1024
@@ -192,20 +195,25 @@ function summarizePluginCalls(calls: PluginBuildMetrics['calls']): Pick<PluginBu
   return summary
 }
 
-function pruneReaders() {
-  if (readers.size <= MAX_READERS)
-    return
-
-  for (const reader of Array.from(readers.values())) {
-    if (readers.size <= MAX_READERS)
-      return
-    if (!reader.hasPendingRead()) {
-      reader.dispose()
-    }
+function pruneReaders(current: RolldownEventsReader) {
+  // Log, metadata, and package-summary readers share a session directory.
+  const currentSession = dirname(current.filepath)
+  let bytes = Array.from(readers.values()).reduce((total, reader) => total + reader.logBytes, 0)
+  for (const [key, reader] of readers) {
+    if (readers.size <= MAX_READERS && bytes <= MAX_READER_LOG_BYTES)
+      break
+    if (dirname(reader.filepath) === currentSession || reader.hasPendingRead())
+      continue
+
+    // RPC handlers may still hold this reader after its read has completed.
+    // Drop cache ownership without clearing data used by those handlers.
+    readers.delete(key)
+    bytes -= reader.logBytes
   }
 }
 
 export class RolldownEventsReader {
+  logBytes: number = 0
   lastBytes: number = 0
   lastTimestamp: number = 0
   manager = new RolldownEventsManager()
@@ -242,11 +250,12 @@ export class RolldownEventsReader {
       const reader = readers.get(cacheKey)!
       readers.delete(cacheKey)
       readers.set(cacheKey, reader)
+      pruneReaders(reader)
       return reader
     }
     const reader = new RolldownEventsReader(filepath, cacheKey)
     readers.set(cacheKey, reader)
-    pruneReaders()
+    pruneReaders(reader)
     return reader
   }
 
@@ -264,6 +273,7 @@ export class RolldownEventsReader {
 
     this.pendingRead = this.readChanges().finally(() => {
       this.pendingRead = undefined
+      pruneReaders(this)
     })
     return this.pendingRead
   }
@@ -277,12 +287,15 @@ export class RolldownEventsReader {
 
     this.pendingSummaryRead = this.readSummaryChanges().finally(() => {
       this.pendingSummaryRead = undefined
+      pruneReaders(this)
     })
     return this.pendingSummaryRead
   }
 
   private async readChanges() {
     const stat = await fs.promises.stat(this.filepath)
+    this.logBytes = stat.size
+    pruneReaders(this)
     const mtime = stat.mtime.getTime()
 
     if (this.summaryOnly || this.packageSummaryOnly)
@@ -329,6 +342,8 @@ export class RolldownEventsReader {
 
   private async readSummaryChanges() {
     const stat = await fs.promises.stat(this.filepath)
+    this.logBytes = stat.size
+    pruneReaders(this)
     const mtime = stat.mtime.getTime()
 
     if (this.manager.eventCount && mtime <= this.lastTimestamp) {
@@ -548,6 +563,7 @@ export class RolldownEventsReader {
 
     this.pendingPackageSummaryRead = this.readPackageSummaryChanges().finally(() => {
       this.pendingPackageSummaryRead = undefined
+      pruneReaders(this)
     })
     return this.pendingPackageSummaryRead
   }
@@ -582,6 +598,8 @@ export class RolldownEventsReader {
 
   private async readPackageSummaryChanges() {
     const stat = await fs.promises.stat(this.filepath)
+    this.logBytes = stat.size
+    pruneReaders(this)
     const mtime = stat.mtime.getTime()
     if (mtime <= this.packageSummaryTimestamp) {
       if (this.packageSummaryOnly && this.logCache.shouldWritePackageSummary())
@@ -1332,7 +1350,8 @@ export class RolldownEventsReader {
   }
 
   dispose() {
-    readers.delete(this.cacheKey)
+    if (readers.get(this.cacheKey) === this)
+      readers.delete(this.cacheKey)
     this.disposeData()
   }