From bd00b3b3c5bfb1bc3a0e46ac5296013c6b9d1eef Mon Sep 17 00:00:00 2001 From: Bogdan Date: Thu, 25 Jun 2026 03:10:01 +0200 Subject: [PATCH] fix(bg): stream session logs with bounded memory (#1762) * fix(bg): stream session logs with bounded memory * fix(bg): handle log follow cleanup edge cases * fix(bg): surface non-follow log read errors * test(bg): isolate log streaming temp dirs --- src/cli/bg.test.ts | 378 +++++++++++++++++++++++++++++++++++++++++++++ src/cli/bg.ts | 269 ++++++++++++++++++++++++++++---- 2 files changed, 618 insertions(+), 29 deletions(-) diff --git a/src/cli/bg.test.ts b/src/cli/bg.test.ts index eb5eee170..8a8b0c67c 100644 --- a/src/cli/bg.test.ts +++ b/src/cli/bg.test.ts @@ -1,12 +1,91 @@ +import { EventEmitter } from 'node:events' +import { mkdtemp, rm, unlink, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' import { describe, expect, it } from 'bun:test' import { buildBackgroundSessionLaunch, buildBackgroundChildProcessConfig, + followLogFile, + printExistingLog, terminateBackgroundProcessTree, + LOG_STREAM_CHUNK_SIZE, parseBackgroundInvocation, parseLogsInvocation, } from './bg.js' +class TestOutput extends EventEmitter { + chunks: Buffer[] = [] + destroyed = false + writableDestroyed = false + writeResults: boolean[] = [] + writeError: unknown + + write(chunk: Uint8Array): boolean { + if (this.writeError) throw this.writeError + if (this.destroyed || this.writableDestroyed) { + throw Object.assign(new Error('stdout closed'), { code: 'EPIPE' }) + } + this.chunks.push(Buffer.from(chunk)) + return this.writeResults.shift() ?? true + } + + bytes(): Buffer { + return Buffer.concat(this.chunks) + } +} + +function createManualScheduler() { + let intervalCallback: (() => void) | undefined + let cleared = false + + return { + setInterval(callback: () => void): ReturnType { + intervalCallback = callback + return 1 as unknown as ReturnType + }, + clearInterval(): void { + cleared = true + }, + tick(): void { + intervalCallback?.() + }, + get cleared(): boolean { + return cleared + }, + } +} + +function deferred() { + let resolve!: (value: T | PromiseLike) => void + let reject!: (reason?: unknown) => void + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise + reject = rejectPromise + }) + return { promise, resolve, reject } +} + +async function waitFor(condition: () => boolean): Promise { + for (let i = 0; i < 50; i++) { + if (condition()) return + await new Promise(resolve => setTimeout(resolve, 0)) + } + throw new Error('condition was not met') +} + +async function withTempFile( + name: string, + run: (path: string) => Promise, +): Promise { + const dir = await mkdtemp(join(tmpdir(), 'openclaude-bg-test-')) + try { + return await run(join(dir, name)) + } finally { + await rm(dir, { recursive: true, force: true }) + } +} + describe('background session CLI parsing', () => { it('builds a print-mode child command and preserves provider/model flags', () => { const parsed = parseBackgroundInvocation([ @@ -254,6 +333,16 @@ describe('background session CLI parsing', () => { follow: false, stream: 'stderr', }) + expect(parseLogsInvocation(['auth-refactor', '--stdout', '-f'])).toEqual({ + target: 'auth-refactor', + follow: true, + stream: 'stdout', + }) + expect(parseLogsInvocation(['auth-refactor', '-f', '--stderr'])).toEqual({ + target: 'auth-refactor', + follow: true, + stream: 'stderr', + }) }) it('preserves Node exec flags and lets the launcher manage heap relaunch state', () => { @@ -306,3 +395,292 @@ describe('background session CLI parsing', () => { expect(signals).toEqual(['SIGTERM', 'SIGKILL']) }) }) + +describe('background session log streaming', () => { + it('emits a multi-megabyte existing log exactly with bounded allocations', async () => { + await withTempFile('stdout.log', async path => { + const chunkSize = LOG_STREAM_CHUNK_SIZE + const contents = Buffer.alloc(chunkSize * 32 + 123) + for (let i = 0; i < contents.length; i++) contents[i] = i % 251 + await writeFile(path, contents) + + const allocations: number[] = [] + const output = new TestOutput() + const offset = await printExistingLog(path, { + output, + chunkSize, + createBuffer: size => { + allocations.push(size) + return Buffer.alloc(size) + }, + }) + + expect(offset).toBe(contents.length) + expect(output.bytes()).toEqual(contents) + expect(Math.max(...allocations)).toBeLessThanOrEqual(chunkSize) + expect(allocations.length).toBeGreaterThan(1) + }) + }) + + it('follow mode emits existing and appended content exactly once in order', async () => { + await withTempFile('stdout.log', async path => { + const output = new TestOutput() + await writeFile(path, Buffer.from('existing-')) + + const offset = await printExistingLog(path, { output, chunkSize: 4 }) + const scheduler = createManualScheduler() + const abort = new AbortController() + const following = followLogFile(path, offset, { + output, + chunkSize: 4, + signal: abort.signal, + setInterval: scheduler.setInterval, + clearInterval: scheduler.clearInterval, + }) + + await writeFile(path, Buffer.from('existing-appended')) + scheduler.tick() + await waitFor(() => output.bytes().toString() === 'existing-appended') + abort.abort() + await following + + expect(output.bytes().toString()).toBe('existing-appended') + }) + }) + + it('splits a large appended range into bounded chunks', async () => { + await withTempFile('stdout.log', async path => { + const chunkSize = 16 + await writeFile(path, Buffer.from('seed')) + const appended = Buffer.alloc(chunkSize * 2 + 5, 7) + await writeFile(path, Buffer.concat([Buffer.from('seed'), appended])) + + const output = new TestOutput() + const scheduler = createManualScheduler() + const abort = new AbortController() + const following = followLogFile(path, 4, { + output, + chunkSize, + signal: abort.signal, + setInterval: scheduler.setInterval, + clearInterval: scheduler.clearInterval, + }) + + scheduler.tick() + await waitFor(() => output.bytes().length === appended.length) + abort.abort() + await following + + expect(output.bytes()).toEqual(appended) + expect(output.chunks.map(chunk => chunk.length)).toEqual([16, 16, 5]) + }) + }) + + it('waits for drain before reading or writing more when stdout applies backpressure', async () => { + await withTempFile('stdout.log', async path => { + await writeFile(path, Buffer.from('abcdef')) + const output = new TestOutput() + output.writeResults.push(false) + + let settled = false + const printing = printExistingLog(path, { output, chunkSize: 3 }).then( + offset => { + settled = true + return offset + }, + ) + + await waitFor(() => output.chunks.length === 1) + expect(output.bytes().toString()).toBe('abc') + expect(settled).toBe(false) + + output.emit('drain') + await expect(printing).resolves.toBe(6) + expect(output.bytes().toString()).toBe('abcdef') + expect(output.chunks.map(chunk => chunk.length)).toEqual([3, 3]) + }) + }) + + it('resets the follow read position when the log is truncated', async () => { + await withTempFile('stdout.log', async path => { + await writeFile(path, Buffer.from('abcdef')) + const output = new TestOutput() + const scheduler = createManualScheduler() + const abort = new AbortController() + const following = followLogFile(path, 6, { + output, + chunkSize: 8, + signal: abort.signal, + setInterval: scheduler.setInterval, + clearInterval: scheduler.clearInterval, + }) + + await writeFile(path, Buffer.from('xy')) + scheduler.tick() + await waitFor(() => output.bytes().toString() === 'xy') + abort.abort() + await following + + expect(output.bytes().toString()).toBe('xy') + }) + }) + + it('tolerates temporary file disappearance while following', async () => { + await withTempFile('stdout.log', async path => { + await writeFile(path, Buffer.from('seed')) + const output = new TestOutput() + const scheduler = createManualScheduler() + const abort = new AbortController() + const following = followLogFile(path, 4, { + output, + chunkSize: 8, + signal: abort.signal, + setInterval: scheduler.setInterval, + clearInterval: scheduler.clearInterval, + }) + + await unlink(path) + scheduler.tick() + await new Promise(resolve => setTimeout(resolve, 0)) + expect(output.bytes().length).toBe(0) + + await writeFile(path, Buffer.from('new')) + scheduler.tick() + await waitFor(() => output.bytes().toString() === 'new') + abort.abort() + await following + + expect(output.bytes().toString()).toBe('new') + }) + }) + + it('prevents writes after signal cleanup during an in-flight poll', async () => { + const output = new TestOutput() + const scheduler = createManualScheduler() + const abort = new AbortController() + const readStarted = deferred() + const releaseRead = deferred() + let closed = false + + const handle = { + stat: async () => ({ size: 4 }), + read: async (buffer: Buffer) => { + readStarted.resolve() + await releaseRead.promise + buffer.write('late') + return { bytesRead: 4, buffer } + }, + close: async () => { + closed = true + }, + } + + let resolved = false + const following = followLogFile('/tmp/stdout.log', 0, { + output, + chunkSize: 4, + signal: abort.signal, + setInterval: scheduler.setInterval, + clearInterval: scheduler.clearInterval, + openFile: async () => handle, + }).then(() => { + resolved = true + }) + + scheduler.tick() + await readStarted.promise + abort.abort() + await new Promise(resolve => setTimeout(resolve, 0)) + expect(resolved).toBe(false) + releaseRead.resolve() + await following + scheduler.tick() + await new Promise(resolve => setTimeout(resolve, 0)) + + expect(scheduler.cleared).toBe(true) + expect(closed).toBe(true) + expect(output.bytes().length).toBe(0) + }) + + it('preserves streamed progress when a later close failure occurs', async () => { + const output = new TestOutput() + const scheduler = createManualScheduler() + const abort = new AbortController() + let openCount = 0 + + const following = followLogFile('/tmp/stdout.log', 0, { + output, + chunkSize: 4, + signal: abort.signal, + setInterval: scheduler.setInterval, + clearInterval: scheduler.clearInterval, + openFile: async () => { + openCount++ + return { + stat: async () => ({ size: 4 }), + read: async (buffer: Buffer) => { + buffer.write('once') + return { bytesRead: 4 } + }, + close: async () => { + if (openCount === 1) throw new Error('close failed') + }, + } + }, + }) + + scheduler.tick() + await waitFor(() => output.bytes().toString() === 'once') + scheduler.tick() + await waitFor(() => openCount === 2) + await new Promise(resolve => setTimeout(resolve, 0)) + abort.abort() + await following + + expect(output.bytes().toString()).toBe('once') + }) + + it('surfaces non-follow file read failures', async () => { + let closed = false + const readError = Object.assign(new Error('read failed'), { + code: 'EIO', + }) + + await expect( + printExistingLog('/tmp/stdout.log', { + chunkSize: 4, + openFile: async () => ({ + stat: async () => ({ size: 4 }), + read: async () => { + throw readError + }, + close: async () => { + closed = true + }, + }), + }), + ).rejects.toThrow('read failed') + expect(closed).toBe(true) + }) + + it('handles EPIPE and destroyed stdout without throwing', async () => { + await withTempFile('stdout.log', async path => { + await writeFile(path, Buffer.from('closed-pipe')) + + const epipeOutput = new TestOutput() + epipeOutput.writeError = Object.assign(new Error('broken pipe'), { + code: 'EPIPE', + }) + await expect( + printExistingLog(path, { output: epipeOutput, chunkSize: 4 }), + ).resolves.toBe(0) + + const destroyedOutput = new TestOutput() + destroyedOutput.destroyed = true + await expect( + printExistingLog(path, { output: destroyedOutput, chunkSize: 4 }), + ).resolves.toBe(0) + expect(destroyedOutput.bytes().length).toBe(0) + }) + }) +}) diff --git a/src/cli/bg.ts b/src/cli/bg.ts index d6cbcbbcd..730f31ed0 100644 --- a/src/cli/bg.ts +++ b/src/cli/bg.ts @@ -1,7 +1,7 @@ import { spawn } from 'node:child_process' import { randomUUID } from 'node:crypto' import { closeSync, openSync } from 'node:fs' -import { open, readFile, unlink } from 'node:fs/promises' +import { open, unlink } from 'node:fs/promises' import { basename } from 'node:path' import treeKill from 'tree-kill' import { argsBeforeDelimiter } from '../utils/cliArgs.js' @@ -58,6 +58,50 @@ const HEAP_RELAUNCHED_ENV = 'OPENCLAUDE_HEAP_RELAUNCHED' const DEFAULT_TERM_GRACE_MS = 2_000 const DEFAULT_KILL_GRACE_MS = 2_000 const DEFAULT_KILL_POLL_INTERVAL_MS = 100 +// Each background-log read buffer is capped at 64 KiB to avoid whole-log allocations. +export const LOG_STREAM_CHUNK_SIZE = 64 * 1024 +const LOG_FOLLOW_POLL_INTERVAL_MS = 500 + +type LogOutput = { + destroyed?: boolean + writableDestroyed?: boolean + write(chunk: Uint8Array): boolean + once(event: string, listener: (...args: unknown[]) => void): unknown + off(event: string, listener: (...args: unknown[]) => void): unknown +} + +type LogFileHandle = { + close(): Promise + stat(): Promise<{ size: number }> + read( + buffer: Buffer, + offset: number, + length: number, + position: number, + ): Promise<{ bytesRead: number }> +} + +type LogFollowTimer = ReturnType | number + +type StreamLogOptions = { + output?: LogOutput + chunkSize?: number + createBuffer?: (size: number) => Buffer + signal?: AbortSignal + openFile?: (path: string, flags: 'r') => Promise + continueOnFileError?: boolean +} + +type FollowLogOptions = StreamLogOptions & { + pollIntervalMs?: number + setInterval?: (callback: () => void, ms: number) => LogFollowTimer + clearInterval?: (timer: LogFollowTimer) => void +} + +type StreamLogResult = { + position: number + outputOpen: boolean +} // This must stay in sync with value-consuming CLI flags in main.tsx and related // handlers. If the CLI flag definitions become centralized, move this parser @@ -410,56 +454,216 @@ function printSessionTable( } } -async function printExistingLog(path: string): Promise { +function isOutputClosed(output: LogOutput): boolean { + return output.destroyed === true || output.writableDestroyed === true +} + +function normalizeChunkSize(chunkSize: number | undefined): number { + if (!Number.isFinite(chunkSize) || !chunkSize || chunkSize < 1) { + return LOG_STREAM_CHUNK_SIZE + } + return Math.floor(chunkSize) +} + +async function waitForDrain( + output: LogOutput, + signal: AbortSignal | undefined, +): Promise { + if (signal?.aborted || isOutputClosed(output)) return false + + return await new Promise(resolve => { + let settled = false + + const cleanup = () => { + output.off('drain', onDrain) + output.off('error', onError) + output.off('close', onClose) + signal?.removeEventListener('abort', onAbort) + } + const finish = (open: boolean) => { + if (settled) return + settled = true + cleanup() + resolve(open) + } + + const onDrain = () => finish(!isOutputClosed(output)) + const onError = () => finish(false) + const onClose = () => finish(false) + const onAbort = () => finish(false) + + output.once('drain', onDrain) + output.once('error', onError) + output.once('close', onClose) + signal?.addEventListener('abort', onAbort, { once: true }) + }) +} + +async function writeLogBuffer( + output: LogOutput, + buffer: Buffer, + signal: AbortSignal | undefined, +): Promise { + if (buffer.length === 0) return true + if (signal?.aborted || isOutputClosed(output)) return false + try { - const contents = await readFile(path) - if (contents.length > 0) process.stdout.write(contents) - return contents.length + if (output.write(buffer)) return !isOutputClosed(output) } catch { - return 0 + return false + } + + return await waitForDrain(output, signal) +} + +async function streamLogRange( + handle: LogFileHandle, + start: number, + endExclusive: number, + options: StreamLogOptions, +): Promise { + const output = options.output ?? process.stdout + const chunkSize = normalizeChunkSize(options.chunkSize) + const createBuffer = options.createBuffer ?? Buffer.allocUnsafe + let position = start + + while (position < endExclusive) { + if (options.signal?.aborted) return { position, outputOpen: false } + const bytesToRead = Math.min(chunkSize, endExclusive - position) + const buffer = createBuffer(bytesToRead) + if (buffer.length < bytesToRead) { + throw new Error('Log stream buffer factory returned a short buffer') + } + + let bytesRead: number + try { + const readResult = await handle.read(buffer, 0, bytesToRead, position) + bytesRead = readResult.bytesRead + } catch (error) { + if (!options.continueOnFileError) throw error + return { position, outputOpen: true } + } + if (bytesRead <= 0) break + if (options.signal?.aborted) return { position, outputOpen: false } + + const chunk = + bytesRead === buffer.length ? buffer : buffer.subarray(0, bytesRead) + if (!(await writeLogBuffer(output, chunk, options.signal))) { + return { position, outputOpen: false } + } + position += bytesRead + } + + return { position, outputOpen: true } +} + +async function streamLogSnapshot( + path: string, + offset: number, + options: StreamLogOptions, +): Promise { + let handle: LogFileHandle + try { + handle = await (options.openFile ?? open)(path, 'r') + } catch (error) { + if (!options.continueOnFileError) throw error + // Keep following; the child may create or rotate the file later. + return { position: offset, outputOpen: true } + } + + let result: StreamLogResult = { position: offset, outputOpen: true } + try { + const { size } = await handle.stat() + const start = size < offset ? 0 : offset + result = + size <= start + ? { position: start, outputOpen: true } + : await streamLogRange(handle, start, size, options) + return result + } catch (error) { + if (!options.continueOnFileError) throw error + return result + } finally { + if (options.continueOnFileError) { + await handle.close().catch(() => undefined) + } else { + await handle.close() + } } } -async function followLogFile(path: string, offset: number): Promise { +export async function printExistingLog( + path: string, + options: StreamLogOptions = {}, +): Promise { + const result = await streamLogSnapshot(path, 0, options) + return result.position +} + +export async function followLogFile( + path: string, + offset: number, + options: FollowLogOptions = {}, +): Promise { + const output = options.output ?? process.stdout + const cleanupController = new AbortController() let position = offset let reading = false + let stopped = false + let timer: LogFollowTimer | undefined + let activePoll: Promise | undefined await new Promise(resolve => { const cleanup = () => { - clearInterval(timer) + if (stopped) return + stopped = true + if (timer) (options.clearInterval ?? clearInterval)(timer) + cleanupController.abort() process.off('SIGINT', cleanup) process.off('SIGTERM', cleanup) - resolve() + options.signal?.removeEventListener('abort', cleanup) + const pendingPoll = activePoll + if (pendingPoll) { + void pendingPoll.finally(resolve) + } else { + resolve() + } } - const timer = setInterval(() => { - if (reading) return + const poll = () => { + if (stopped || reading) return reading = true - void (async () => { + const pollPromise = (async () => { try { - const handle = await open(path, 'r') - try { - const { size } = await handle.stat() - if (size < position) position = 0 - if (size > position) { - const buffer = Buffer.alloc(size - position) - await handle.read(buffer, 0, buffer.length, position) - position = size - process.stdout.write(buffer) - } - } finally { - await handle.close() - } - } catch { - // Keep following; the child may create or rotate the file later. + const result = await streamLogSnapshot(path, position, { + ...options, + output, + signal: cleanupController.signal, + continueOnFileError: true, + }) + position = result.position + if (!result.outputOpen) cleanup() } finally { reading = false } })() - }, 500) + activePoll = pollPromise + void pollPromise.finally(() => { + if (activePoll === pollPromise) activePoll = undefined + }) + } process.once('SIGINT', cleanup) process.once('SIGTERM', cleanup) + if (options.signal?.aborted) { + cleanup() + return + } + options.signal?.addEventListener('abort', cleanup, { once: true }) + timer = (options.setInterval ?? setInterval)( + poll, + options.pollIntervalMs ?? LOG_FOLLOW_POLL_INTERVAL_MS, + ) }) } @@ -571,7 +775,14 @@ export async function logsHandler( fail(`Log file does not exist: ${logPath}`) } - const offset = await printExistingLog(logPath) + let offset: number + try { + offset = await printExistingLog(logPath, { + continueOnFileError: parsed.follow, + }) + } catch (error) { + fail(`Failed to read log file: ${errorMessage(error)}`) + } if (parsed.follow) { await followLogFile(logPath, offset) }