diff --git a/config/daemon-blocking-call-baseline.txt b/config/daemon-blocking-call-baseline.txt index 05e615e08eb..66debece11c 100644 --- a/config/daemon-blocking-call-baseline.txt +++ b/config/daemon-blocking-call-baseline.txt @@ -3,7 +3,6 @@ # This is a RATCHET: counts may only go DOWN. # Prune after removing calls: pnpm check:daemon-blocking-call-ratchet --prune 2 src/main/daemon/daemon-cgroup-scope.ts -1 src/main/daemon/daemon-file-log.ts 1 src/main/daemon/daemon-process-identity-query.ts 1 src/main/daemon/daemon-process-start-time.ts 2 src/main/daemon/daemon-stream-backlog-probe.ts diff --git a/src/main/crash-reporting/crash-feedback-diagnostic-bundle.ts b/src/main/crash-reporting/crash-feedback-diagnostic-bundle.ts index e2a6dc7396a..b06828578fb 100644 --- a/src/main/crash-reporting/crash-feedback-diagnostic-bundle.ts +++ b/src/main/crash-reporting/crash-feedback-diagnostic-bundle.ts @@ -28,7 +28,7 @@ function skippedCrashDiagnosticBundle(): CrashDiagnosticBundleAttachment { } } -function collectCrashDiagnosticBundleAttachment(): CrashDiagnosticBundleAttachment { +async function collectCrashDiagnosticBundleAttachment(): Promise { const status = getDiagnosticsStatus() if (!status.bundleEnabled) { return { @@ -39,9 +39,9 @@ function collectCrashDiagnosticBundleAttachment(): CrashDiagnosticBundleAttachme } } - let bundle: ReturnType + let bundle: Awaited> try { - bundle = collectDiagnosticBundle({ + bundle = await collectDiagnosticBundle({ appVersion: app.getVersion(), platform: os.platform(), arch: os.arch(), @@ -70,9 +70,9 @@ function collectCrashDiagnosticBundleAttachment(): CrashDiagnosticBundleAttachme } } -export function prepareCrashDiagnosticBundle( +export async function prepareCrashDiagnosticBundle( includeDiagnosticLogs: boolean -): CrashDiagnosticBundleAttachment { +): Promise { return includeDiagnosticLogs ? collectCrashDiagnosticBundleAttachment() : skippedCrashDiagnosticBundle() diff --git a/src/main/daemon/daemon-entry.ts b/src/main/daemon/daemon-entry.ts index 7b40ac4033d..3dfc8b17dc8 100644 --- a/src/main/daemon/daemon-entry.ts +++ b/src/main/daemon/daemon-entry.ts @@ -127,6 +127,8 @@ async function main(): Promise { const readyIdentity = await readCurrentDaemonReadyIdentity(startedAtMs) const daemonLog = logFilePath ? createDaemonFileLog(logFilePath) : createNoopDaemonFileLog() daemonLog.log('startup', { protocolVersion: PROTOCOL_VERSION, socketPath }) + // Log writes are async; every exit, including process.exit below, flushes the queued tail. + process.once('exit', () => daemonLog.close()) const daemonScope = ownFreshDaemonScope(freshDaemonScope === true, launchNonce) const log = (event: string, details?: Record): void => daemonLog.log(event, details) @@ -153,6 +155,8 @@ async function main(): Promise { return } daemonLog.log('uncaught-exception-fatal', uncaughtExceptionLogDetails(err)) + // Rethrowing from this handler skips 'exit' listeners. + daemonLog.close() console.error('[daemon] Uncaught exception (fatal):', err) throw err }) @@ -184,7 +188,6 @@ async function main(): Promise { daemonLog.log('shutdown-error', { message: (err as Error)?.message }) } finally { shutdownDeadline.cancel() - daemonLog.close() process.exit(0) } } @@ -239,7 +242,6 @@ async function main(): Promise { shuttingDown = true daemonLog.log('login-session-dead-retire', details) daemon?.recordOrderlyExit() - daemonLog.close() // Skip PTY teardown so replacement daemons cold-restore scrollback from unclean session metadata. process.exit(1) } @@ -292,12 +294,10 @@ async function main(): Promise { deathWatch?.stop() shuttingDown = true daemonLog.log('shutdown', { reason: 'idle' }) - daemonLog.close() process.exit(0) }, onRpcShutdown: () => { shutdownDeadline.cancel() - daemonLog.close() process.exit(0) }, onShutdownRequested: beginShutdown diff --git a/src/main/daemon/daemon-file-log.test.ts b/src/main/daemon/daemon-file-log.test.ts index 85067507c3b..ab055af8e77 100644 --- a/src/main/daemon/daemon-file-log.test.ts +++ b/src/main/daemon/daemon-file-log.test.ts @@ -1,4 +1,4 @@ -import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs' import { tmpdir } from 'node:os' import { join } from 'node:path' @@ -21,34 +21,47 @@ function readLines(filePath: string): Record[] { } describe('createDaemonFileLog', () => { - it('appends NDJSON lines with src/ts/pid/event and terse details', () => { + it('appends NDJSON lines with src/ts/pid/event and terse details', async () => { const filePath = join(dir, 'daemon.log') const log = createDaemonFileLog(filePath) log.log('startup', { protocolVersion: 18 }) log.log('session-created', { sessionId: 'abc', pid: 42 }) + await vi.waitFor(() => expect(readLines(filePath)).toHaveLength(2), { timeout: 10_000 }) const lines = readLines(filePath) - expect(lines).toHaveLength(2) expect(lines[0]).toMatchObject({ src: 'daemon', event: 'startup', protocolVersion: 18 }) expect(typeof lines[0].ts).toBe('string') expect(lines[0].pid).toBe(process.pid) expect(lines[1]).toMatchObject({ event: 'session-created', sessionId: 'abc', pid: 42 }) }) - it('rotates at the byte cap and keeps only the configured rotated files', () => { + it('rotates at the byte cap and keeps only the configured rotated files', async () => { const filePath = join(dir, 'daemon.log') const log = createDaemonFileLog(filePath, { maxBytes: 150, maxRotatedFiles: 2 }) - for (let i = 0; i < 40; i++) { + // One line per batch, so each write sees the size left by the last. + for (let i = 0; i < 10; i++) { log.log('tick', { i }) + await vi.waitFor(() => expect(readLines(filePath).at(-1)).toMatchObject({ i }), { + timeout: 10_000 + }) } - expect(existsSync(filePath)).toBe(true) expect(existsSync(`${filePath}.1`)).toBe(true) expect(existsSync(`${filePath}.2`)).toBe(true) expect(existsSync(`${filePath}.3`)).toBe(false) + }) - const active = readLines(filePath) - expect(active.at(-1)).toMatchObject({ event: 'tick', i: 39 }) + it('counts a line too large for the queue instead of holding it', () => { + const filePath = join(dir, 'daemon.log') + const log = createDaemonFileLog(filePath, { maxQueuedChars: 2_000 }) + log.log('huge', { detail: 'x'.repeat(10_000) }) + log.log('after') + log.close() + expect(readLines(filePath).map((l) => [l.event, l.count])).toEqual([ + ['log-entries-dropped', 1], + ['after', undefined], + ['daemon-log-closed', undefined] + ]) }) it('is fail-open when the log directory cannot be created', () => { @@ -70,8 +83,9 @@ describe('createDaemonFileLog', () => { circular.self = circular expect(() => log.log('weird', circular)).not.toThrow() log.log('ok') + log.close() const lines = readLines(filePath) - expect(lines.map((l) => l.event)).toEqual(['ok']) + expect(lines.map((l) => l.event)).toEqual(['ok', 'daemon-log-closed']) }) it('close() writes a terminal marker and stops further writes', () => { diff --git a/src/main/daemon/daemon-file-log.ts b/src/main/daemon/daemon-file-log.ts index 0c4b17d96ac..934fb05f1a5 100644 --- a/src/main/daemon/daemon-file-log.ts +++ b/src/main/daemon/daemon-file-log.ts @@ -1,23 +1,26 @@ // Daemon diagnostics must never affect lifecycle or include terminal content or tokens. -import { appendFileSync, existsSync, mkdirSync, renameSync, statSync, unlinkSync } from 'node:fs' +import { mkdirSync } from 'node:fs' import { dirname } from 'node:path' +import { createBoundedLogFileWriter } from '../observability/bounded-log-file-writer' const DEFAULT_MAX_BYTES = 5 * 1024 * 1024 const DEFAULT_MAX_ROTATED_FILES = 2 -const PRIVATE_FILE_MODE = 0o600 +const BATCH_WINDOW_MS = 50 /** Includes the active file so diagnostic bundles collect the whole rotated family. */ export const DAEMON_LOG_MAX_FILES = DEFAULT_MAX_ROTATED_FILES + 1 export type DaemonFileLog = { log(event: string, details?: Record): void + /** Appends the queued tail synchronously; call right before exit. */ close(): void } export type DaemonFileLogOptions = { readonly maxBytes?: number readonly maxRotatedFiles?: number + readonly maxQueuedChars?: number } export function createNoopDaemonFileLog(): DaemonFileLog { @@ -27,88 +30,50 @@ export function createNoopDaemonFileLog(): DaemonFileLog { } } +function formatDaemonLogLine(event: string, details: Record): string { + return `${JSON.stringify({ + src: 'daemon', + ts: new Date().toISOString(), + pid: process.pid, + event, + ...details + })}\n` +} + export function createDaemonFileLog( filePath: string, opts: DaemonFileLogOptions = {} ): DaemonFileLog { - const maxBytes = opts.maxBytes ?? DEFAULT_MAX_BYTES - const maxRotatedFiles = opts.maxRotatedFiles ?? DEFAULT_MAX_ROTATED_FILES - - let disabled = false - let currentBytes = 0 - - function disable(): void { - disabled = true - } - try { mkdirSync(dirname(filePath), { recursive: true }) - currentBytes = existsSync(filePath) ? statSync(filePath).size : 0 } catch { - disable() - } - - function rotate(): void { - // Without rotation, preserve the active byte count so the overflow guard remains accurate. - if (maxRotatedFiles < 1) { - return - } - try { - for (let i = maxRotatedFiles; i >= 1; i--) { - const src = i === 1 ? filePath : `${filePath}.${i - 1}` - const dst = `${filePath}.${i}` - if (!existsSync(src)) { - continue - } - if (existsSync(dst)) { - unlinkSync(dst) - } - renameSync(src, dst) - } - currentBytes = 0 - } catch { - disable() - } + return createNoopDaemonFileLog() } + const writer = createBoundedLogFileWriter({ + filePath, + maxBytes: opts.maxBytes ?? DEFAULT_MAX_BYTES, + maxFiles: (opts.maxRotatedFiles ?? DEFAULT_MAX_ROTATED_FILES) + 1, + batchWindowMs: BATCH_WINDOW_MS, + maxQueuedChars: opts.maxQueuedChars, + formatDroppedLine: (count) => formatDaemonLogLine('log-entries-dropped', { count }) + }) function log(event: string, details: Record = {}): void { - if (disabled) { - return - } let line: string try { - line = `${JSON.stringify({ - src: 'daemon', - ts: new Date().toISOString(), - pid: process.pid, - event, - ...details - })}\n` + line = formatDaemonLogLine(event, details) } catch { // Malformed event details must not disrupt daemon lifecycle. return } - const lineBytes = Buffer.byteLength(line, 'utf8') - if (currentBytes > 0 && currentBytes + lineBytes > maxBytes) { - rotate() - if (disabled) { - return - } - } - try { - appendFileSync(filePath, line, { mode: PRIVATE_FILE_MODE }) - currentBytes += lineBytes - } catch { - disable() - } + writer.write(line) } return { log, close(): void { - // Appends are synchronous; there is nothing to flush. log('daemon-log-closed') - disabled = true + writer.close() } } } diff --git a/src/main/ipc/crash-reporting-submission.ts b/src/main/ipc/crash-reporting-submission.ts index f4cd5e0bbaf..90d2ff11620 100644 --- a/src/main/ipc/crash-reporting-submission.ts +++ b/src/main/ipc/crash-reporting-submission.ts @@ -46,7 +46,9 @@ export async function submitCrashReport( ): Promise { const report = await getRequestedCrashReport(store, args) if (!report) { - const diagnosticUpload = prepareCrashDiagnosticBundle(args.includeDiagnosticLogs !== false) + const diagnosticUpload = await prepareCrashDiagnosticBundle( + args.includeDiagnosticLogs !== false + ) const diagnosticBundle = diagnosticUpload.diagnosticBundle const reportOnlyDiagnosticBundle = diagnosticBundleForReportOnlyRetry(diagnosticUpload) const result = await submitFeedback({ @@ -98,7 +100,9 @@ export async function submitCrashReport( inFlightSubmissions.add(report.id) try { - const diagnosticUpload = prepareCrashDiagnosticBundle(args.includeDiagnosticLogs !== false) + const diagnosticUpload = await prepareCrashDiagnosticBundle( + args.includeDiagnosticLogs !== false + ) const diagnosticBundle = diagnosticUpload.diagnosticBundle const reportOnlyDiagnosticBundle = diagnosticBundleForReportOnlyRetry(diagnosticUpload) const result = await submitFeedback({ diff --git a/src/main/ipc/diagnostics.test.ts b/src/main/ipc/diagnostics.test.ts index 2868fdec19d..8a6b8d4c44a 100644 --- a/src/main/ipc/diagnostics.test.ts +++ b/src/main/ipc/diagnostics.test.ts @@ -302,7 +302,7 @@ describe('diagnostics IPC handlers', () => { collectDiagnosticBundleMock.mockReturnValue(bundle) const collect = handlers.get('diagnostics:collectBundle')! - expect(collect({}, 30)).toEqual({ + expect(await collect({}, 30)).toEqual({ bundleSubmissionId: bundle.bundleSubmissionId, bytes: bundle.bytes, spanCount: bundle.spanCount diff --git a/src/main/ipc/diagnostics.ts b/src/main/ipc/diagnostics.ts index 1b5e8377412..d81b9a71f7b 100644 --- a/src/main/ipc/diagnostics.ts +++ b/src/main/ipc/diagnostics.ts @@ -207,7 +207,7 @@ export function registerDiagnosticsHandlers(): void { ipcMain.handle( 'diagnostics:collectBundle', - (_event, lookbackMinutesIn: unknown): DiagnosticsBundlePreview => { + async (_event, lookbackMinutesIn: unknown): Promise => { // Enforce consent in main; hiding renderer controls is insufficient. const status = getDiagnosticsStatus() if (!status.bundleEnabled) { @@ -218,7 +218,7 @@ export function registerDiagnosticsHandlers(): void { typeof lookbackMinutesIn === 'number' && Number.isFinite(lookbackMinutesIn) ? Math.max(1, Math.min(30 * 24 * 60, Math.floor(lookbackMinutesIn))) : undefined - const bundle = collectDiagnosticBundle({ + const bundle = await collectDiagnosticBundle({ appVersion: app.getVersion(), platform: osPlatform(), arch: osArch(), diff --git a/src/main/observability/bounded-log-file-writer.ts b/src/main/observability/bounded-log-file-writer.ts new file mode 100644 index 00000000000..1ae78b2d22f --- /dev/null +++ b/src/main/observability/bounded-log-file-writer.ts @@ -0,0 +1,234 @@ +// Diagnostics append off the event loop so a stalled disk delays log lines, never the process. + +import { appendFileSync } from 'node:fs' +import { open, rename, stat } from 'node:fs/promises' + +const PRIVATE_FILE_MODE = 0o600 +export const DEFAULT_MAX_QUEUED_CHARS = 1024 * 1024 + +export type BoundedLogFileWriterOptions = { + readonly filePath: string + readonly maxBytes: number + /** Includes the active file; one disables rotation. */ + readonly maxFiles: number + readonly batchWindowMs: number + /** Measured in UTF-16 units so admission never scans line bytes. */ + readonly maxQueuedChars?: number + /** Written where entries were dropped. */ + readonly formatDroppedLine: (droppedEntries: number) => string +} + +export type BoundedLogFileWriter = { + /** Never blocks; when the queue is full, drops and counts its oldest lines. */ + write(line: string): void + /** Resolves once every line accepted before the call has been written or dropped. */ + flush(): Promise + /** Appends the queue synchronously, for crash boundaries; may land before an in-flight batch. */ + flushSync(): void + /** Stops accepting lines and flushes synchronously, for exit paths. */ + close(): void +} + +// A number marks where that many entries were dropped. +type QueueItem = string | number + +export function createBoundedLogFileWriter( + opts: BoundedLogFileWriterOptions +): BoundedLogFileWriter { + const { filePath, maxBytes, maxFiles, batchWindowMs, formatDroppedLine } = opts + const maxQueuedChars = opts.maxQueuedChars ?? DEFAULT_MAX_QUEUED_CHARS + + const queue: QueueItem[] = [] + // Every queued item has a sequence index; queue[0] holds headIndex. + let headIndex = 0 + // Memory stays under the cap plus one in-flight batch. + let queuedChars = 0 + // Entries lost to failed writes or oversized lines; the marker goes where the next line lands. + let droppedEntries = 0 + let inFlight: { start: number; done: Promise } | null = null + let timer: NodeJS.Timeout | null = null + let closed = false + + function nextIndex(): number { + return headIndex + queue.length + } + + /** Lines settle in order: an in-flight batch is always older than anything still queued. */ + function oldestUnsettled(): number { + return inFlight?.start ?? (queue.length > 0 ? headIndex : Number.POSITIVE_INFINITY) + } + + function queueDroppedMarker(): void { + if (droppedEntries > 0) { + queue.push(droppedEntries) + droppedEntries = 0 + } + } + + function render(items: readonly QueueItem[]): string { + return items.map((item) => (typeof item === 'number' ? formatDroppedLine(item) : item)).join('') + } + + function countDropped(items: readonly QueueItem[]): void { + for (const item of items) { + droppedEntries += typeof item === 'number' ? item : 1 + } + } + + function dequeueAll(): QueueItem[] { + headIndex += queue.length + queuedChars = 0 + return queue.splice(0) + } + + /** Drops the oldest queued entries so the newest, such as a fatal line, always fit. */ + function evictOldestFor(chars: number): void { + let evicted = 0 + while (queue.length > 0 && queuedChars + chars > maxQueuedChars) { + const item = queue[0] + evicted += typeof item === 'number' ? item : 1 + queuedChars -= typeof item === 'number' ? 0 : item.length + queue.shift() + headIndex++ + } + if (evicted > 0) { + // The marker takes the last evicted entry's place and index. + headIndex-- + queue.unshift(evicted) + } + } + + async function fileSize(): Promise { + return stat(filePath).then( + (stats) => stats.size, + () => 0 + ) + } + + async function rotate(): Promise { + // Oldest generation first, so each rename lands on a freed name. + for (let i = maxFiles - 1; i >= 1; i--) { + const from = i === 1 ? filePath : `${filePath}.${i - 1}` + await rename(from, `${filePath}.${i}`).catch(() => undefined) + } + } + + // Appends the batch in as few writes as the OS allows, normally one, so a concurrent sync + // append lands between batches rather than inside a record. + async function append(text: string): Promise { + const handle = await open(filePath, 'a', PRIVATE_FILE_MODE) + try { + const data = Buffer.from(text, 'utf8') + let offset = 0 + while (offset < data.length) { + offset += (await handle.write(data, offset)).bytesWritten + } + } finally { + await handle.close() + } + } + + // The only code that rotates; batches run one at a time, so renames never race. + async function writeBatch(items: readonly QueueItem[]): Promise { + let written = true + try { + const text = render(items) + const size = await fileSize() + // Avoid empty rotated files on new installations. + if (maxFiles > 1 && size > 0 && size + Buffer.byteLength(text, 'utf8') > maxBytes) { + await rotate() + } + if (text) { + await append(text) + } + } catch { + written = false + } + // Settle in the same synchronous block that observes the write finishing. + inFlight = null + if (!written) { + countDropped(items) + } else if (!closed) { + // Retrying a marker against a failing disk would spin; wait for the next accepted line. + queueDroppedMarker() + } + if (queue.length > 0) { + scheduleDrain() + } + } + + function startDrain(): Promise { + if (timer) { + clearTimeout(timer) + timer = null + } + if (inFlight) { + return inFlight.done + } + // An empty batch still runs the size check, for appends made by flushSync. + const start = headIndex + inFlight = { start, done: writeBatch(dequeueAll()) } + return inFlight.done + } + + function scheduleDrain(): void { + if (timer || inFlight || closed) { + return + } + timer = setTimeout(() => void startDrain(), batchWindowMs) + timer.unref?.() + } + + // Leaves the drain timer running: the next drain sizes this append and rotates if needed. + function flushSync(): void { + queueDroppedMarker() + if (queue.length === 0) { + return + } + const items = dequeueAll() + // An O_APPEND write never overwrites an in-flight batch; it may only land before it. + try { + appendFileSync(filePath, render(items), { mode: PRIVATE_FILE_MODE }) + } catch { + countDropped(items) + } + } + + return { + write(line: string): void { + if (closed) { + return + } + // A line the queue could never hold is counted where it would have gone. + if (line.length > maxQueuedChars) { + droppedEntries++ + return + } + queueDroppedMarker() + evictOldestFor(line.length) + queue.push(line) + queuedChars += line.length + scheduleDrain() + }, + async flush(): Promise { + queueDroppedMarker() + const target = nextIndex() + // Each turn awaits a real write, or the condition is already false. + while (oldestUnsettled() < target) { + await startDrain() + } + }, + flushSync, + close(): void { + if (closed) { + return + } + closed = true + if (timer) { + clearTimeout(timer) + timer = null + } + flushSync() + } + } +} diff --git a/src/main/observability/index.ts b/src/main/observability/index.ts index f1519f18d1a..c652d58309c 100644 --- a/src/main/observability/index.ts +++ b/src/main/observability/index.ts @@ -24,6 +24,7 @@ import { } from './diagnostic-bundle-upload' import { setActiveSink, startSpan } from './tracer' import { setSecurePathHardeningReporter } from '../../shared/secure-path-hardening-report' +import { withTimeout } from '../../shared/promise-timeout-fallback' const CI_ENV_VARS = [ 'CI', @@ -96,6 +97,8 @@ export function resolveObservabilityConsent(): ObservabilityConsent { export { getTraceFilePath } from './logs-directory' +const SHUTDOWN_FLUSH_TIMEOUT_MS = 500 + let sink: LocalFileSink | null = null let consent: ObservabilityConsent | null = null @@ -139,13 +142,9 @@ function installSecurePathHardeningReporter(): void { }) } -/** Flush for exits that bypass shutdown. */ -export function flushObservability(): void { - try { - sink?.flush() - } catch { - // Nothing is left to report to. - } +/** For exits that bypass shutdown. */ +export function closeObservabilityOnExit(): void { + sink?.close() } export async function shutdownObservability(): Promise { @@ -153,7 +152,9 @@ export async function shutdownObservability(): Promise { // Stop new tracer writes before closing and flushing the sink. setActiveSink(null) if (sink) { - sink.close() + // Keep an orderly quit's in-flight batch; a stalled disk costs quit this long, never a hang. + await withTimeout(sink.flushAsync(), SHUTDOWN_FLUSH_TIMEOUT_MS, undefined) + sink?.close() sink = null } consent = null @@ -181,16 +182,14 @@ export function getDiagnosticsStatus(): DiagnosticsStatus { } /** Never include install_id: bundles use independent submission IDs. */ -export function collectDiagnosticBundle( +export async function collectDiagnosticBundle( meta: Pick< CollectBundleOptions, 'appVersion' | 'platform' | 'arch' | 'osRelease' | 'orcaChannel' | 'lookbackMinutes' > -): CollectedBundle { +): Promise { // Flush before collection so the bundle includes the user’s latest action. - if (sink) { - sink.flush() - } + await sink?.flushAsync() return _collectBundle({ traceFilePath: getTraceFilePath(), maxFiles: DEFAULT_MAX_FILES, diff --git a/src/main/observability/local-file-sink-memory.test.ts b/src/main/observability/local-file-sink-memory.test.ts index 4fa56e24182..9971c4dd701 100644 --- a/src/main/observability/local-file-sink-memory.test.ts +++ b/src/main/observability/local-file-sink-memory.test.ts @@ -1,4 +1,4 @@ -import { mkdtempSync, readFileSync, rmSync, statSync } from 'node:fs' +import { existsSync, mkdtempSync, readFileSync, rmSync } from 'node:fs' import { tmpdir } from 'node:os' import { join } from 'node:path' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' @@ -31,7 +31,7 @@ function retainedHeap(): number { } describe('trace sink rejected record retention', () => { - it('keeps small-record byte scans deferred until the batch flush', () => { + it('keeps small-record byte scans deferred until the batch flush', async () => { const filePath = join(directory, 'trace.ndjson') sink = createLocalFileSink({ filePath }) const byteLength = vi.spyOn(Buffer, 'byteLength') @@ -42,16 +42,16 @@ describe('trace sink rejected record retention', () => { sink.push({ index, text: 'πŸ’‘ζΌ’ε­—' }) } beforeFlush = byteLength.mock.calls.length - sink.flush() + await sink.flushAsync() afterFlush = byteLength.mock.calls.length } finally { byteLength.mockRestore() } expect(beforeFlush).toBe(0) - expect(afterFlush).toBe(20) + expect(afterFlush).toBeLessThanOrEqual(1) }) - it('releases oversized serialized records before the pending batch flushes', () => { + it('releases oversized serialized records before the pending batch flushes', async () => { const filePath = join(directory, 'trace.ndjson') sink = createLocalFileSink({ filePath, maxBytes: 64 * 1024, batchWindowMs: 200 }) const before = retainedHeap() @@ -59,25 +59,26 @@ describe('trace sink rejected record retention', () => { sink.push({ index, payload: 'x'.repeat(1024 * 1024) }) } const retained = retainedHeap() - before - expect(statSync(filePath).size).toBe(0) + expect(existsSync(filePath)).toBe(false) expect(vi.getTimerCount()).toBe(1) expect(retained).toBeLessThan(5 * 1024 * 1024) sink.push({ valid: true }) - vi.advanceTimersByTime(200) + await sink.flushAsync() const written = readFileSync(filePath, 'utf8').split('\n').filter(Boolean).map(parseLine) expect(written.filter((entry) => entry.type === DROPPED_RECORD_TYPE)).toHaveLength(24) expect(written.at(-1)).toEqual({ valid: true }) }) - it('names the dropped record in the marker instead of leaving a silent gap', () => { + it('names the dropped record in the marker instead of leaving a silent gap', async () => { const filePath = join(directory, 'trace.ndjson') - sink = createLocalFileSink({ filePath, maxBytes: 64 * 1024, flushBufferThreshold: 1 }) + sink = createLocalFileSink({ filePath, maxBytes: 64 * 1024 }) sink.push({ type: 'effect-span', name: 'worktree.create', traceId: 'a'.repeat(32), payload: 'x'.repeat(1024 * 1024) }) + await sink.flushAsync() const [marker] = readFileSync(filePath, 'utf8').split('\n').filter(Boolean).map(parseLine) expect(marker).toMatchObject({ type: DROPPED_RECORD_TYPE, @@ -90,36 +91,25 @@ describe('trace sink rejected record retention', () => { expect(BigInt(marker.endTimeUnixNano as string)).toBeGreaterThan(0n) }) - it('omits the marker when even the marker would exceed the byte cap', () => { + it('omits the marker when even the marker would exceed the byte cap', async () => { const filePath = join(directory, 'trace.ndjson') - sink = createLocalFileSink({ filePath, maxBytes: 40, flushBufferThreshold: 1 }) + sink = createLocalFileSink({ filePath, maxBytes: 40 }) sink.push({ payload: 'x'.repeat(1_000) }) sink.push({ ok: 1 }) + await sink.flushAsync() expect(readFileSync(filePath, 'utf8')).toBe('{"ok":1}\n') }) - it('keeps the same count-triggered flush for valid records beside rejected ones', () => { - const filePath = join(directory, 'trace.ndjson') - sink = createLocalFileSink({ filePath, maxBytes: 32, flushBufferThreshold: 3 }) - sink.push({ valid: 1 }) - sink.push({ payload: 'πŸ’‘'.repeat(20) }) - expect(statSync(filePath).size).toBe(0) - sink.push({ payload: 'x'.repeat(100) }) - expect(readFileSync(filePath, 'utf8')).toBe('{"valid":1}\n') - sink.push({ valid: 2 }) - vi.advanceTimersByTime(200) - expect(readFileSync(filePath, 'utf8')).toBe('{"valid":1}\n{"valid":2}\n') - }) - - it('accepts the exact UTF-8 byte cap and preserves rotation and close flushes', () => { + it('accepts the exact UTF-8 byte cap and rotates at it', async () => { const filePath = join(directory, 'trace.ndjson') const record = { text: 'πŸ’‘' } const line = `${JSON.stringify(record)}\n` sink = createLocalFileSink({ filePath, maxBytes: Buffer.byteLength(line), maxFiles: 2 }) sink.push(record) sink.push({ text: 'πŸ’‘x' }) + await sink.flushAsync() sink.push(record) - sink.close() + await sink.flushAsync() expect(readFileSync(filePath, 'utf8')).toBe(line) expect(readFileSync(`${filePath}.1`, 'utf8')).toBe(line) expect(vi.getTimerCount()).toBe(0) diff --git a/src/main/observability/local-file-sink.test.ts b/src/main/observability/local-file-sink.test.ts index 8ea0c9bc403..541f1cce111 100644 --- a/src/main/observability/local-file-sink.test.ts +++ b/src/main/observability/local-file-sink.test.ts @@ -9,7 +9,7 @@ import { } from 'node:fs' import { tmpdir } from 'node:os' import { dirname, join } from 'node:path' -import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { createLocalFileSink, getRotatedFamilySize, listRotatedFiles } from './local-file-sink' let dir: string @@ -26,16 +26,15 @@ function makeRecord(i: number): { i: number; payload: string } { } describe('local-file-sink β€” basic write', () => { - it('writes one NDJSON line per push', () => { + it('writes one NDJSON line per push', async () => { const file = join(dir, 'test.ndjson') const sink = createLocalFileSink({ filePath: file, - batchWindowMs: 100, - flushBufferThreshold: 1 + batchWindowMs: 100 }) sink.push({ a: 1 }) sink.push({ b: 2 }) - sink.flush() + await sink.flushAsync() sink.close() const lines = readFileSync(file, 'utf8').split('\n').filter(Boolean) @@ -44,22 +43,6 @@ describe('local-file-sink β€” basic write', () => { expect(JSON.parse(lines[1])).toEqual({ b: 2 }) }) - it('coalesces writes up to the buffer threshold', () => { - const file = join(dir, 'test.ndjson') - const sink = createLocalFileSink({ - filePath: file, - batchWindowMs: 100_000, - flushBufferThreshold: 5 - }) - for (let i = 0; i < 4; i++) { - sink.push({ i }) - } - expect(statSync(file).size).toBe(0) - sink.push({ i: 4 }) - expect(statSync(file).size).toBeGreaterThan(0) - sink.close() - }) - it('creates trace directories and files with private POSIX permissions', () => { if (process.platform === 'win32') { return @@ -67,8 +50,7 @@ describe('local-file-sink β€” basic write', () => { const file = join(dir, 'logs', 'test.ndjson') const sink = createLocalFileSink({ filePath: file, - batchWindowMs: 100_000, - flushBufferThreshold: 1 + batchWindowMs: 100_000 }) sink.push({ ok: true }) sink.close() @@ -91,8 +73,7 @@ describe('local-file-sink β€” basic write', () => { const sink = createLocalFileSink({ filePath: file, maxFiles: 3, - batchWindowMs: 100_000, - flushBufferThreshold: 1 + batchWindowMs: 100_000 }) sink.close() @@ -102,19 +83,18 @@ describe('local-file-sink β€” basic write', () => { }) describe('local-file-sink β€” rotation', () => { - it('rotates when the byte cap is exceeded', () => { + it('rotates when the byte cap is exceeded', async () => { const file = join(dir, 'test.ndjson') const sink = createLocalFileSink({ filePath: file, maxBytes: 500, maxFiles: 3, - batchWindowMs: 100_000, - flushBufferThreshold: 1 + batchWindowMs: 100_000 }) for (let i = 0; i < 8; i++) { sink.push(makeRecord(i)) + await sink.flushAsync() } - sink.flush() sink.close() const files = listRotatedFiles(file, 3) @@ -122,35 +102,34 @@ describe('local-file-sink β€” rotation', () => { expect(existsSync(file)).toBe(true) }) - it('uses UTF-8 byte length for rotation accounting', () => { + it('uses UTF-8 byte length for rotation accounting', async () => { const file = join(dir, 'test.ndjson') const sink = createLocalFileSink({ filePath: file, maxBytes: 90, maxFiles: 3, - batchWindowMs: 100_000, - flushBufferThreshold: 1 + batchWindowMs: 100_000 }) sink.push({ payload: 'πŸ˜€'.repeat(15) }) + await sink.flushAsync() sink.push({ payload: 'πŸ˜€'.repeat(15) }) - sink.flush() + await sink.flushAsync() sink.close() expect(listRotatedFiles(file, 3).length).toBeGreaterThan(1) }) - it('drops an individual record that exceeds the file byte cap', () => { + it('drops an individual record that exceeds the file byte cap', async () => { const file = join(dir, 'test.ndjson') const sink = createLocalFileSink({ filePath: file, maxBytes: 100, maxFiles: 3, - batchWindowMs: 100_000, - flushBufferThreshold: 1 + batchWindowMs: 100_000 }) sink.push({ payload: 'x'.repeat(1_000) }) sink.push({ ok: true }) - sink.flush() + await sink.flushAsync() sink.close() const lines = readFileSync(file, 'utf8').split('\n').filter(Boolean) @@ -158,19 +137,18 @@ describe('local-file-sink β€” rotation', () => { expect(getRotatedFamilySize(file, 3)).toBeLessThanOrEqual(100) }) - it('splits an oversized buffered batch instead of dropping valid records', () => { + it('keeps every record of a batch larger than the byte cap', async () => { const file = join(dir, 'test.ndjson') const sink = createLocalFileSink({ filePath: file, maxBytes: 170, maxFiles: 5, - batchWindowMs: 100_000, - flushBufferThreshold: 3 + batchWindowMs: 100_000 }) sink.push({ i: 1, payload: 'x'.repeat(60) }) sink.push({ i: 2, payload: 'x'.repeat(60) }) sink.push({ i: 3, payload: 'x'.repeat(60) }) - sink.flush() + await sink.flushAsync() sink.close() const allRecords = listRotatedFiles(file, 5) @@ -181,19 +159,62 @@ describe('local-file-sink β€” rotation', () => { expect(allRecords).toEqual([1, 2, 3]) }) - it('caps total disk usage at maxFiles Γ— maxBytes (worst case)', () => { + it('flushes synchronously at crash boundaries', () => { + const file = join(dir, 'test.ndjson') + const sink = createLocalFileSink({ filePath: file, batchWindowMs: 100_000 }) + sink.push({ i: 1 }) + sink.push({ i: 2 }) + sink.flush() + + const lines = readFileSync(file, 'utf8').split('\n').filter(Boolean) + expect(lines.map((line) => JSON.parse(line))).toEqual([{ i: 1 }, { i: 2 }]) + sink.close() + }) + + it('resolves concurrent flushAsync calls only after the shared batch is written', async () => { + const file = join(dir, 'test.ndjson') + const sink = createLocalFileSink({ filePath: file, batchWindowMs: 100_000 }) + sink.push({ i: 1 }) + const first = sink.flushAsync() + await sink.flushAsync() + expect(readFileSync(file, 'utf8')).toBe('{"i":1}\n') + await first + sink.close() + }) + + it('keeps 100 crash-boundary write/flush pairs within the rotation bound', async () => { const file = join(dir, 'test.ndjson') const sink = createLocalFileSink({ filePath: file, maxBytes: 500, maxFiles: 3, - batchWindowMs: 100_000, - flushBufferThreshold: 1 + batchWindowMs: 1 + }) + const activeSize = (): number => (existsSync(file) ? statSync(file).size : 0) + for (let i = 0; i < 100; i++) { + sink.push(makeRecord(i)) + sink.flush() + // The drain the push scheduled sizes the synchronous append and rotates it. + await vi.waitFor(() => expect(activeSize()).toBeLessThanOrEqual(500), { timeout: 10_000 }) + } + expect(existsSync(`${file}.2`)).toBe(true) + // Each generation can hold one record past the cap before the drain moves it aside. + expect(getRotatedFamilySize(file, 3)).toBeLessThan(3 * 500 * 2) + sink.close() + }) + + it('caps total disk usage at maxFiles Γ— maxBytes (worst case)', async () => { + const file = join(dir, 'test.ndjson') + const sink = createLocalFileSink({ + filePath: file, + maxBytes: 500, + maxFiles: 3, + batchWindowMs: 100_000 }) for (let i = 0; i < 50; i++) { sink.push(makeRecord(i)) + await sink.flushAsync() } - sink.flush() sink.close() const total = getRotatedFamilySize(file, 3) @@ -203,19 +224,18 @@ describe('local-file-sink β€” rotation', () => { }) describe('local-file-sink β€” listing + clearing', () => { - it('lists newest β†’ oldest', () => { + it('lists newest β†’ oldest', async () => { const file = join(dir, 'test.ndjson') const sink = createLocalFileSink({ filePath: file, maxBytes: 200, maxFiles: 5, - batchWindowMs: 100_000, - flushBufferThreshold: 1 + batchWindowMs: 100_000 }) for (let i = 0; i < 20; i++) { sink.push(makeRecord(i)) + await sink.flushAsync() } - sink.flush() sink.close() const files = listRotatedFiles(file, 5) @@ -227,12 +247,11 @@ describe('local-file-sink β€” listing + clearing', () => { }) describe('local-file-sink β€” robustness', () => { - it('does not throw on circular records (drops the line)', () => { + it('does not throw on circular records (drops the line)', async () => { const file = join(dir, 'test.ndjson') const sink = createLocalFileSink({ filePath: file, - batchWindowMs: 100_000, - flushBufferThreshold: 1 + batchWindowMs: 100_000 }) const a: Record = { ok: true } a.self = a @@ -240,7 +259,7 @@ describe('local-file-sink β€” robustness', () => { sink.push(a) sink.push({ ok: 2 }) }).not.toThrow() - sink.flush() + await sink.flushAsync() sink.close() const lines = readFileSync(file, 'utf8').split('\n').filter(Boolean) expect(lines.map((l) => JSON.parse(l))).toEqual([{ ok: 2 }]) diff --git a/src/main/observability/local-file-sink.ts b/src/main/observability/local-file-sink.ts index c13c1dbf80e..f27146d02c2 100644 --- a/src/main/observability/local-file-sink.ts +++ b/src/main/observability/local-file-sink.ts @@ -1,21 +1,9 @@ -// Flush synchronously so buffered diagnostics can survive crash boundaries; batch sparse writes on a timer. +// Trace writes are queued and appended off the event loop; a stalled disk drops spans instead of freezing the process. -import { - chmodSync, - closeSync, - existsSync, - fchmodSync, - fstatSync, - mkdirSync, - openSync, - renameSync, - statSync, - unlinkSync, - writeSync -} from 'node:fs' +import { chmodSync, existsSync, mkdirSync, statSync } from 'node:fs' import { dirname } from 'node:path' +import { createBoundedLogFileWriter, DEFAULT_MAX_QUEUED_CHARS } from './bounded-log-file-writer' -const DEFAULT_FLUSH_BUFFER_THRESHOLD = 32 export const DEFAULT_MAX_BYTES = 10 * 1024 * 1024 export const DEFAULT_MAX_FILES = 10 export const DEFAULT_BATCH_WINDOW_MS = 200 @@ -29,14 +17,17 @@ export type LocalFileSinkOptions = { readonly maxBytes?: number readonly maxFiles?: number readonly batchWindowMs?: number - readonly flushBufferThreshold?: number + readonly maxQueuedChars?: number } export type LocalFileSink = { readonly filePath: string push(record: unknown): void - /** Flush buffered lines synchronously. */ + /** Writes queued records before returning; for crash boundaries only. */ flush(): void + /** Resolves once every record pushed before the call has been written or dropped. */ + flushAsync(): Promise + /** Stops accepting records and flushes like flush(); safe to call from an exit handler. */ close(): void } @@ -56,12 +47,20 @@ function tightenTraceFamilyPermissions(filePath: string, maxFiles: number): void } } +function droppedRecordMarker(fields: Record): string { + return `${JSON.stringify({ + type: DROPPED_RECORD_TYPE, + ...fields, + // Let the collector age markers out like spans. + endTimeUnixNano: `${Date.now()}000000` + })}\n` +} + export function createLocalFileSink(opts: LocalFileSinkOptions): LocalFileSink { const filePath = opts.filePath const maxBytes = opts.maxBytes ?? DEFAULT_MAX_BYTES const maxFiles = opts.maxFiles ?? DEFAULT_MAX_FILES - const batchWindowMs = opts.batchWindowMs ?? DEFAULT_BATCH_WINDOW_MS - const flushThreshold = opts.flushBufferThreshold ?? DEFAULT_FLUSH_BUFFER_THRESHOLD + const maxQueuedChars = opts.maxQueuedChars ?? DEFAULT_MAX_QUEUED_CHARS // Traces contain paths and crash context; restrict access regardless of umask. const traceDirectory = dirname(filePath) @@ -69,111 +68,15 @@ export function createLocalFileSink(opts: LocalFileSinkOptions): LocalFileSink { chmodPathIfPresent(traceDirectory, PRIVATE_DIRECTORY_MODE) tightenTraceFamilyPermissions(filePath, maxFiles) - // Size the open file descriptor so path replacement cannot race sizing. - let fd: number = openAppend(filePath) - let currentBytes: number = safeFstatSize(fd) - - let buffer: (string | null)[] = [] - let timer: NodeJS.Timeout | null = null - let closed = false - - function openAppend(path: string): number { - const handle = openSync(path, 'a', PRIVATE_FILE_MODE) - try { - fchmodSync(handle, PRIVATE_FILE_MODE) - } catch { - // Some Windows volumes reject POSIX chmod. - } - return handle - } - - function safeFstatSize(handle: number): number { - try { - return fstatSync(handle).size - } catch { - // Fall back to zero; the next write resizes. - return 0 - } - } - - function rotate(): void { - // CIFS and other filesystems may refuse to rename open files. - try { - closeSync(fd) - } catch {} - for (let i = maxFiles - 1; i >= 1; i--) { - const src = i === 1 ? filePath : `${filePath}.${i - 1}` - const dst = `${filePath}.${i}` - if (!existsSync(src)) { - continue - } - try { - if (existsSync(dst)) { - // Remove a destination left by a crashed rotation before renaming. - unlinkSync(dst) - } - renameSync(src, dst) - } catch { - // Partial rotation must not crash the process. - } - } - fd = openAppend(filePath) - currentBytes = 0 - } - - function flushBuffer(): void { - if (buffer.length === 0 || closed) { - return - } - const lines = buffer - buffer = [] - let pendingChunk: string[] = [] - let pendingChunkBytes = 0 - - function writeChunk(chunkLines: string[], chunkBytes: number): void { - if (chunkLines.length === 0) { - return - } - const chunk = chunkLines.join('') - try { - writeSync(fd, chunk) - currentBytes += chunkBytes - } catch { - // Retry once; optional diagnostics must not crash main. - try { - try { - closeSync(fd) - } catch {} - fd = openAppend(filePath) - writeSync(fd, chunk) - currentBytes = safeFstatSize(fd) - } catch {} - } - } - - function flushPendingChunk(): void { - writeChunk(pendingChunk, pendingChunkBytes) - pendingChunk = [] - pendingChunkBytes = 0 - } - - for (const line of lines) { - if (line === null) { - continue - } - const lineBytes = Buffer.byteLength(line, 'utf8') - if (pendingChunkBytes > 0 && currentBytes + pendingChunkBytes + lineBytes > maxBytes) { - flushPendingChunk() - } - // Avoid empty rotated files on new installations. - if (currentBytes > 0 && currentBytes + lineBytes > maxBytes) { - rotate() - } - pendingChunk.push(line) - pendingChunkBytes += lineBytes - } - flushPendingChunk() - } + const writer = createBoundedLogFileWriter({ + filePath, + maxBytes, + maxFiles, + batchWindowMs: opts.batchWindowMs ?? DEFAULT_BATCH_WINDOW_MS, + maxQueuedChars, + formatDroppedLine: (droppedRecords) => + droppedRecordMarker({ reason: 'queue-full', droppedRecords }) + }) /** Count dropped UTF-16 units to avoid scanning rejected payloads for byte length. */ function oversizeMarker(record: unknown, droppedChars: number): string | null { @@ -181,37 +84,18 @@ export function createLocalFileSink(opts: LocalFileSinkOptions): LocalFileSink { typeof record === 'object' && record !== null ? (record as Record) : {} const name = typeof span.name === 'string' ? span.name.slice(0, MAX_MARKER_NAME_CHARS) : null const traceId = typeof span.traceId === 'string' ? span.traceId.slice(0, 32) : null - const marker = `${JSON.stringify({ - type: DROPPED_RECORD_TYPE, + const marker = droppedRecordMarker({ reason: 'oversize', droppedChars, - // Let the collector age markers out like spans. - endTimeUnixNano: `${Date.now()}000000`, ...(name === null ? {} : { name }), ...(traceId === null ? {} : { traceId }) - })}\n` + }) return Buffer.byteLength(marker, 'utf8') > maxBytes ? null : marker } - function ensureTimer(): void { - if (timer || closed) { - return - } - timer = setTimeout(() => { - timer = null - flushBuffer() - }, batchWindowMs) - if (typeof timer.unref === 'function') { - timer.unref() - } - } - return { filePath, push(record: unknown): void { - if (closed) { - return - } let line: string try { line = `${JSON.stringify(record)}\n` @@ -220,38 +104,19 @@ export function createLocalFileSink(opts: LocalFileSinkOptions): LocalFileSink { return } // UTF-8 uses at most three bytes per UTF-16 unit; small records need no admission scan. + // A record the queue could never hold gets a named marker instead of an anonymous drop. const oversized = line.length > maxBytes || + line.length > maxQueuedChars || (line.length * 3 > maxBytes && Buffer.byteLength(line, 'utf8') > maxBytes) - // Preserve flush timing with a marker slot so rejected records leave a visible gap. - buffer.push(oversized ? oversizeMarker(record, line.length) : line) - if (buffer.length >= flushThreshold) { - flushBuffer() - } else { - ensureTimer() + const admitted = oversized ? oversizeMarker(record, line.length) : line + if (admitted !== null) { + writer.write(admitted) } }, - flush(): void { - if (timer) { - clearTimeout(timer) - timer = null - } - flushBuffer() - }, - close(): void { - if (closed) { - return - } - if (timer) { - clearTimeout(timer) - timer = null - } - flushBuffer() - try { - closeSync(fd) - } catch {} - closed = true - } + flush: () => writer.flushSync(), + flushAsync: () => writer.flush(), + close: () => writer.close() } } diff --git a/src/main/observability/stalled-log-disk.test.ts b/src/main/observability/stalled-log-disk.test.ts new file mode 100644 index 00000000000..f8a2af9730a --- /dev/null +++ b/src/main/observability/stalled-log-disk.test.ts @@ -0,0 +1,267 @@ +import type * as NodeFs from 'node:fs' +import type * as NodeFsPromises from 'node:fs/promises' +import { mkdtempSync, readFileSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { createDaemonFileLog } from '../daemon/daemon-file-log' +import { createLocalFileSink, DROPPED_RECORD_TYPE } from './local-file-sink' + +// `holdAsync` parks async appends (at open) until released; `blockSync` makes sync writes hold the thread. +const disk = vi.hoisted(() => { + const heldWrites: (() => void)[] = [] + return { holdAsync: false, blockSync: false, heldWrites, blockingWrites: 0 } +}) + +const SYNC_STALL_MS = 150 + +vi.mock('node:fs', async (importOriginal) => { + const actual = await importOriginal() + const stallSignal = new Int32Array(new SharedArrayBuffer(4)) + const blockLikeStalledDisk = (): void => { + if (disk.blockSync) { + disk.blockingWrites++ + Atomics.wait(stallSignal, 0, 0, SYNC_STALL_MS) + } + } + const mocked = { + ...actual, + appendFileSync: (...args: Parameters): void => { + blockLikeStalledDisk() + actual.appendFileSync(...args) + }, + // Overloads resolve to the last signature; the real call forwards whatever form was passed. + writeSync: (...args: Parameters): number => { + blockLikeStalledDisk() + return actual.writeSync(...args) + } + } + return { ...mocked, default: mocked } +}) + +vi.mock('node:fs/promises', async (importOriginal) => { + const actual = await importOriginal() + const mocked = { + ...actual, + open: (...args: Parameters): ReturnType => { + if (!disk.holdAsync) { + return actual.open(...args) + } + return new Promise((resolve, reject) => { + disk.heldWrites.push(() => void actual.open(...args).then(resolve, reject)) + }) + } + } + return { ...mocked, default: mocked } +}) + +function stallDisk(): void { + disk.holdAsync = true + disk.blockSync = true +} + +function releaseDisk(): void { + disk.holdAsync = false + disk.blockSync = false + disk.blockingWrites = 0 + for (const run of disk.heldWrites.splice(0)) { + run() + } +} + +async function writeIsHeld(): Promise { + await vi.waitFor(() => expect(disk.heldWrites.length).toBeGreaterThan(0), { timeout: 10_000 }) +} + +/** Resolves when a timer queued now gets to run. */ +function nextTimer(): Promise<'ran'> { + return new Promise((resolve) => setTimeout(() => resolve('ran'), 0)) +} + +let dir: string +beforeEach(() => { + dir = mkdtempSync(join(tmpdir(), 'orca-stalled-log-')) +}) +afterEach(() => { + releaseDisk() + // Windows can briefly hold a file the writer is still closing. + rmSync(dir, { recursive: true, force: true, maxRetries: 5 }) +}) + +function readJsonLines(filePath: string): Record[] { + return readFileSync(filePath, 'utf8') + .split('\n') + .filter(Boolean) + .map((line): Record => JSON.parse(line)) +} + +describe('log writes on a stalled disk', () => { + it('daemon log writes leave a concurrent timer running while the write is stalled', async () => { + const filePath = join(dir, 'daemon.log') + const log = createDaemonFileLog(filePath) + stallDisk() + for (let i = 0; i < 4; i++) { + log.log('session-killed', { i }) + } + await writeIsHeld() + await expect(nextTimer()).resolves.toBe('ran') + expect(disk.heldWrites).toHaveLength(1) + expect(disk.blockingWrites).toBe(0) + releaseDisk() + await vi.waitFor(() => expect(readJsonLines(filePath)).toHaveLength(4), { timeout: 10_000 }) + log.close() + }) + + it('trace writes leave a concurrent timer running while the write is stalled', async () => { + const filePath = join(dir, 'trace.ndjson') + const sink = createLocalFileSink({ filePath, batchWindowMs: 1 }) + stallDisk() + for (let i = 0; i < 40; i++) { + sink.push({ i }) + } + await writeIsHeld() + await expect(nextTimer()).resolves.toBe('ran') + expect(disk.heldWrites).toHaveLength(1) + expect(disk.blockingWrites).toBe(0) + releaseDisk() + await sink.flushAsync() + expect(readJsonLines(filePath)).toHaveLength(40) + sink.close() + }) + + it('daemon close writes the fatal line while an earlier write is still in flight', async () => { + const filePath = join(dir, 'daemon.log') + const log = createDaemonFileLog(filePath) + disk.holdAsync = true + log.log('before-hold') + await writeIsHeld() + log.log('uncaught-exception-fatal') + log.close() + const events = (): unknown[] => readJsonLines(filePath).map((line) => line.event) + expect(events()).toEqual(['uncaught-exception-fatal', 'daemon-log-closed']) + releaseDisk() + await vi.waitFor(() => expect(events()).toContain('before-hold'), { timeout: 10_000 }) + }) + + it('a crash-boundary trace flush lands while an earlier write is still in flight', async () => { + const filePath = join(dir, 'trace.ndjson') + const sink = createLocalFileSink({ filePath, batchWindowMs: 1 }) + disk.holdAsync = true + sink.push({ name: 'before-hold' }) + await writeIsHeld() + sink.push({ name: 'app_relaunch_requested' }) + sink.flush() + expect(readJsonLines(filePath)).toEqual([{ name: 'app_relaunch_requested' }]) + releaseDisk() + await sink.flushAsync() + expect(readJsonLines(filePath)).toContainEqual({ name: 'before-hold' }) + sink.close() + }) + + it('flushAsync waits for its own records, not ones logged while it waits', async () => { + const filePath = join(dir, 'trace.ndjson') + const sink = createLocalFileSink({ filePath, batchWindowMs: 1 }) + const heldCount = (count: number): Promise => + vi.waitFor(() => expect(disk.heldWrites).toHaveLength(count), { timeout: 10_000 }) + disk.holdAsync = true + sink.push({ batch: 'a' }) + let flushed = false + const flushing = sink.flushAsync().then(() => { + flushed = true + }) + await heldCount(1) + sink.push({ batch: 'b' }) + disk.heldWrites.shift()?.() + // Batch b is now held; flush must not wait for it. + await heldCount(1) + await vi.waitFor(() => expect(flushed).toBe(true), { timeout: 10_000 }) + expect(disk.heldWrites).toHaveLength(1) + releaseDisk() + await flushing + await sink.flushAsync() + sink.close() + }) + + it('two concurrent flushes both wait for lines queued behind an in-flight batch', async () => { + const filePath = join(dir, 'trace.ndjson') + const sink = createLocalFileSink({ filePath, batchWindowMs: 100_000 }) + disk.holdAsync = true + sink.push({ name: 'A' }) + const firstBatch = sink.flushAsync() + await writeIsHeld() + sink.push({ name: 'L' }) + const first = sink.flushAsync() + const second = sink.flushAsync() + releaseDisk() + await second + expect(readJsonLines(filePath)).toContainEqual({ name: 'L' }) + await Promise.all([first, firstBatch]) + sink.close() + }) + + it('flush resolves right after a crash-boundary flush that follows a finished batch', async () => { + const filePath = join(dir, 'trace.ndjson') + const sink = createLocalFileSink({ filePath, batchWindowMs: 100_000 }) + sink.push({ name: 'a' }) + await sink.flushAsync() + sink.push({ name: 'b' }) + sink.flush() + let flushed = false + void sink.flushAsync().then(() => { + flushed = true + }) + await vi.waitFor(() => expect(flushed).toBe(true), { timeout: 10_000 }) + expect(readJsonLines(filePath)).toEqual([{ name: 'a' }, { name: 'b' }]) + sink.close() + }) + + it('drops the oldest daemon entries when full, so the fatal line still lands', async () => { + const filePath = join(dir, 'daemon.log') + const log = createDaemonFileLog(filePath, { maxQueuedChars: 2_000 }) + disk.holdAsync = true + log.log('tick', { i: -1 }) + await writeIsHeld() + for (let i = 0; i < 200; i++) { + log.log('tick', { i }) + } + log.log('uncaught-exception-fatal') + log.close() + const events = (): unknown[] => readJsonLines(filePath).map((line) => line.event) + expect(events()).toContain('uncaught-exception-fatal') + releaseDisk() + const ticksWritten = (): unknown[] => + readJsonLines(filePath).flatMap((line) => (line.event === 'tick' ? [line.i] : [])) + // The held batch lands after the synchronous close. + await vi.waitFor(() => expect(ticksWritten()).toContain(-1), { timeout: 10_000 }) + + const lines = readJsonLines(filePath) + const ticks = lines.filter((line) => line.event === 'tick') + const markers = lines.filter((line) => line.event === 'log-entries-dropped') + expect(ticks.length).toBeLessThan(201) + expect(markers.map((line) => line.count)).toEqual([201 - ticks.length]) + expect(ticks.map((line) => line.i)).toContain(199) + }) + + it('drops trace records past the queue cap and reports the count once writes recover', async () => { + const filePath = join(dir, 'trace.ndjson') + const sink = createLocalFileSink({ filePath, batchWindowMs: 1, maxQueuedChars: 2_000 }) + disk.holdAsync = true + sink.push({ i: -1 }) + await writeIsHeld() + for (let i = 0; i < 200; i++) { + sink.push({ i, payload: 'x'.repeat(40) }) + } + releaseDisk() + await sink.flushAsync() + sink.push({ recovered: true }) + await sink.flushAsync() + + const lines = readJsonLines(filePath) + const records = lines.filter((line) => typeof line.i === 'number') + const marker = lines.find((line) => line.type === DROPPED_RECORD_TYPE) + expect(records.length).toBeLessThan(201) + expect(marker).toMatchObject({ reason: 'queue-full', droppedRecords: 201 - records.length }) + expect(lines.at(-1)).toEqual({ recovered: true }) + sink.close() + }) +}) diff --git a/src/main/orcad/orcad-entry.ts b/src/main/orcad/orcad-entry.ts index 06d8c2e0d6e..b5342028c2c 100644 --- a/src/main/orcad/orcad-entry.ts +++ b/src/main/orcad/orcad-entry.ts @@ -32,7 +32,7 @@ import { placeUnshownLiveTerminalsBestEffort } from '../runtime/terminal/adoptio export { parseArgs } -let closeOrcadObservability = (): void => {} +let closeOrcadObservability = (): Promise => Promise.resolve() /** Resolve after transport readiness publishes the desktop-compatible serve payload. */ export async function startOrcad(options: OrcadOptions = {}): Promise { @@ -45,13 +45,14 @@ export async function startOrcad(options: OrcadOptions = {}): Promise startOrcadRuntime(options, registerCleanup), - () => { + async () => { try { runOrcadQuitHandlers() } finally { // Close after quit handlers, including failed ones, so their spans reach disk. - closeOrcadObservability() - closeOrcadObservability = () => {} + const closing = closeOrcadObservability() + closeOrcadObservability = () => Promise.resolve() + await closing } } ) diff --git a/src/main/orcad/orcad-lifecycle.ts b/src/main/orcad/orcad-lifecycle.ts index f8610f601e2..2f8ca2f3922 100644 --- a/src/main/orcad/orcad-lifecycle.ts +++ b/src/main/orcad/orcad-lifecycle.ts @@ -131,7 +131,7 @@ export async function startOrcadWithLifecycle( export async function startOrcadWithHost( userDataPath: string, start: (registerCleanup: (cleanup: OrcadRuntimeCleanup) => void) => Promise, - runQuitHandlers: () => void + runQuitHandlers: () => void | Promise ): Promise; instance: OrcadManagedStopInstance }> { const instanceLock = acquireOrcadInstanceLock(userDataPath) const { pid, startedAtMs, nonce } = instanceLock.record diff --git a/src/main/orcad/orcad-observability.test.ts b/src/main/orcad/orcad-observability.test.ts index ca15c0d212b..b398e89ee61 100644 --- a/src/main/orcad/orcad-observability.test.ts +++ b/src/main/orcad/orcad-observability.test.ts @@ -62,11 +62,11 @@ describe('orcad trace file', () => { error: new Error('disk full') }) - it('writes a structured chat failure to its own file under the data root, flushed by close', () => { + it('writes a structured chat failure to its own file under the data root, flushed by close', async () => { const close = installOrcadObservability() reportLateSettlement() - close() + await close() expect(traceRecords()).toContainEqual( expect.objectContaining({ @@ -78,7 +78,7 @@ describe('orcad trace file', () => { expect(existsSync(join(dataRoot, 'logs', 'main.trace.ndjson'))).toBe(false) }) - it('flushes what is buffered when the process exits with no cleanup', () => { + it('flushes what is buffered when the process exits with no cleanup', async () => { const before = new Set(process.listeners('exit')) const close = installOrcadObservability() const onExit = process.listeners('exit').filter((listener) => !before.has(listener)) @@ -88,17 +88,17 @@ describe('orcad trace file', () => { onExit[0]?.call(process, 1) expect(traceRecords().map((record) => record.name)).toContain('agentSession.late-settlement') - close() + await close() expect(process.listeners('exit')).toHaveLength(before.size) }) - it('still starts when the logs folder cannot be opened, with tracing off', () => { + it('still starts when the logs folder cannot be opened, with tracing off', async () => { writeFileSync(join(dataRoot, 'logs'), 'not a folder') const warn = vi.mocked(console.warn) const close = installOrcadObservability() reportLateSettlement() - close() + await close() expect(warn).toHaveBeenCalledWith( expect.stringContaining('[observability] tracing is off'), diff --git a/src/main/orcad/orcad-observability.ts b/src/main/orcad/orcad-observability.ts index 262f58d31ad..84ac75ae558 100644 --- a/src/main/orcad/orcad-observability.ts +++ b/src/main/orcad/orcad-observability.ts @@ -1,14 +1,18 @@ import process from 'node:process' -import { flushObservability, initObservability, shutdownObservability } from '../observability' +import { + closeObservabilityOnExit, + initObservability, + shutdownObservability +} from '../observability' import { getOrcadTraceFilePath } from '../observability/logs-directory' /** Install before runtime startup; close after quit handlers so their spans reach disk. */ -export function installOrcadObservability(): () => void { +export function installOrcadObservability(): () => Promise { initObservability({ traceFilePath: getOrcadTraceFilePath() }) - // Exit handlers flush synchronously even when normal cleanup cannot run. - process.once('exit', flushObservability) + // Exits that skip normal cleanup still get the queued tail. + process.once('exit', closeObservabilityOnExit) return () => { - process.off('exit', flushObservability) - void shutdownObservability() + // An exit during the bounded shutdown flush still writes the queued tail. + return shutdownObservability().finally(() => process.off('exit', closeObservabilityOnExit)) } }