diff --git a/src/daemon/log-context.ts b/src/daemon/log-context.ts index 8f4731c1..4e682230 100644 --- a/src/daemon/log-context.ts +++ b/src/daemon/log-context.ts @@ -23,9 +23,18 @@ export function createLogContext(options: { if (derivedEnabled && options.logPath) { try { fsSync.mkdirSync(path.dirname(options.logPath), { recursive: true }); - context.writer = fsSync.createWriteStream(options.logPath, { + const writer = fsSync.createWriteStream(options.logPath, { flags: 'a', }); + // Attach before any write: a long-lived WriteStream with no error + // listener turns ENOSPC/EIO into an uncaughtException and kills the daemon. + writer.on('error', (error) => { + console.warn(`[daemon] Log file write error (${options.logPath}): ${error.message}`); + if (context.writer === writer) { + context.writer = undefined; + } + }); + context.writer = writer; } catch (error) { console.warn(`[daemon] Failed to open log file ${options.logPath}: ${(error as Error).message}`); } diff --git a/src/serve.ts b/src/serve.ts index e06be7b5..edaa76a2 100644 --- a/src/serve.ts +++ b/src/serve.ts @@ -1,5 +1,6 @@ import http from 'node:http'; import { Readable } from 'node:stream'; +import { pipeline } from 'node:stream/promises'; import { createMcpHandler, McpServer, @@ -232,8 +233,25 @@ async function handleNodeRequest( return; } const body = Readable.fromWeb(webResponse.body as never); - body.on('error', (error) => response.destroy(error)); - body.pipe(response); + await pipeHttpResponseBody(body, response); +} + +/** + * Pipe an MCP response body into a Node HTTP response with bidirectional + * error cleanup. Client abort must destroy the body; body failure must + * destroy the response. Bare `body.pipe(response)` only handles one direction. + */ +export async function pipeHttpResponseBody(body: Readable, response: http.ServerResponse): Promise { + try { + await pipeline(body, response); + } catch { + if (!response.destroyed) { + response.destroy(); + } + if (!body.destroyed) { + body.destroy(); + } + } } function toWebRequest(request: http.IncomingMessage): Request { diff --git a/tests/daemon-log-context.test.ts b/tests/daemon-log-context.test.ts new file mode 100644 index 00000000..24a55168 --- /dev/null +++ b/tests/daemon-log-context.test.ts @@ -0,0 +1,69 @@ +import fs from 'node:fs/promises'; +import os from 'node:os'; +import path from 'node:path'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { createLogContext, disposeLogContext, logEvent } from '../src/daemon/log-context.js'; + +describe('daemon log context stream safety', () => { + let tempDir: string | undefined; + + afterEach(async () => { + vi.restoreAllMocks(); + if (tempDir) { + await fs.rm(tempDir, { recursive: true, force: true }).catch(() => {}); + tempDir = undefined; + } + }); + + it('attaches an error listener when opening the daemon log WriteStream', async () => { + tempDir = await fs.mkdtemp(path.join(os.tmpdir(), 'mcporter-log-context-')); + const logPath = path.join(tempDir, 'daemon.log'); + const context = createLogContext({ + enabled: true, + logAllServers: true, + servers: new Set(), + logPath, + }); + + expect(context.writer).toBeDefined(); + // Without an early error listener, ENOSPC / EIO on the long-lived stream + // becomes an uncaughtException and takes down the daemon. + expect(context.writer!.listenerCount('error')).toBeGreaterThan(0); + + await disposeLogContext(context); + }); + + it('swallows log stream errors without uncaughtException and stops writing', async () => { + tempDir = await fs.mkdtemp(path.join(os.tmpdir(), 'mcporter-log-context-')); + const logPath = path.join(tempDir, 'daemon.log'); + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}); + const uncaught: Error[] = []; + const onUncaught = (error: Error) => { + uncaught.push(error); + }; + process.on('uncaughtException', onUncaught); + + const context = createLogContext({ + enabled: true, + logAllServers: true, + servers: new Set(), + logPath, + }); + expect(context.writer).toBeDefined(); + + const streamError = Object.assign(new Error('no space left on device'), { code: 'ENOSPC' }); + context.writer!.emit('error', streamError); + + await new Promise((resolve) => setImmediate(resolve)); + + expect(uncaught).toEqual([]); + expect(warn).toHaveBeenCalledWith(expect.stringContaining('no space left on device')); + expect(context.writer).toBeUndefined(); + + // logEvent must not throw after the stream is dropped + expect(() => logEvent(context, 'after stream failure')).not.toThrow(); + + process.off('uncaughtException', onUncaught); + await disposeLogContext(context); + }); +}); diff --git a/tests/serve-stream-errors.test.ts b/tests/serve-stream-errors.test.ts new file mode 100644 index 00000000..6f94d497 --- /dev/null +++ b/tests/serve-stream-errors.test.ts @@ -0,0 +1,51 @@ +import http from 'node:http'; +import { Readable, Writable } from 'node:stream'; +import { describe, expect, it, vi } from 'vitest'; +import { pipeHttpResponseBody } from '../src/serve.js'; + +function createMockResponse(): http.ServerResponse { + const dest = new Writable({ + write(_chunk, _encoding, callback) { + callback(); + }, + }); + // pipeHttpResponseBody only needs a Writable destination with destroy(). + return dest as unknown as http.ServerResponse; +} + +describe('serve HTTP body pipe error cleanup', () => { + it('destroys the response body when the client response errors', async () => { + const body = new Readable({ + read() { + this.push(Buffer.alloc(16, 1)); + }, + }); + const destroySpy = vi.spyOn(body, 'destroy'); + const response = createMockResponse(); + + const pipePromise = pipeHttpResponseBody(body, response); + // Simulate client abort / socket error on the HTTP response side. + response.emit('error', new Error('socket hang up')); + + await expect(pipePromise).resolves.toBeUndefined(); + expect(destroySpy).toHaveBeenCalled(); + expect(body.destroyed).toBe(true); + }); + + it('destroys the HTTP response when the body stream errors', async () => { + const body = new Readable({ + read() { + this.push(Buffer.alloc(8, 2)); + }, + }); + const response = createMockResponse(); + const destroySpy = vi.spyOn(response, 'destroy'); + + const pipePromise = pipeHttpResponseBody(body, response); + body.destroy(new Error('upstream truncated')); + + await expect(pipePromise).resolves.toBeUndefined(); + expect(destroySpy).toHaveBeenCalled(); + expect(response.destroyed).toBe(true); + }); +});