From 31ecad3381ae50c1a59936fe436deedc75753355 Mon Sep 17 00:00:00 2001 From: Devin Ivy Date: Tue, 5 May 2026 20:56:47 -0400 Subject: [PATCH] fix(serve): close dup'd fd on channel-close race; wire highWaterMark - closeSync(fd) in the catch path prevents fd leak when a PushChannel is already closed during shutdown. - Pass highWaterMark through to channel() so backpressure is enforced at the cross-thread pipe layer. Co-Authored-By: Claude Opus 4.6 (1M context) --- src/serve/server-threads.ts | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/src/serve/server-threads.ts b/src/serve/server-threads.ts index e8c77f9..de07803 100644 --- a/src/serve/server-threads.ts +++ b/src/serve/server-threads.ts @@ -1,4 +1,4 @@ -import { openSync } from 'node:fs'; +import { closeSync, openSync } from 'node:fs'; import type { Server } from 'node:net'; import { channel } from '../channel.ts'; import { shared } from '../shared/shared.ts'; @@ -17,7 +17,7 @@ export function serverThreads( opts: ServerThreadsOptions = {}, ): ServerThreads { const balance = opts.balance ?? leastConns(); - const _hwm = opts.highWaterMark ?? DEFAULT_HIGH_WATER_MARK; + const hwm = opts.highWaterMark ?? DEFAULT_HIGH_WATER_MARK; const listenOpts: Required = { drainTimeout: opts.listen?.drainTimeout ?? DEFAULT_DRAIN_TIMEOUT, }; @@ -29,7 +29,7 @@ export function serverThreads( const pushChannels = Array.from({ length: n }, () => new PushChannel()); // Wrap each PushChannel in moroutine's channel() so it is recognized by // moroutine's arg-preparation layer and piped cross-thread via MessagePort. - const channels = pushChannels.map((pc) => channel(pc)); + const channels = pushChannels.map((pc) => channel(pc, { highWaterMark: hwm })); // Shared atomic counter slots, one per worker. const counters = shared(Array.from({ length: n }, () => int32atomic)); @@ -56,8 +56,9 @@ export function serverThreads( try { pushChannels[idx].send(fd); } catch { - // send threw because the channel is closed — undo the increment. + // send threw because the channel is closed — undo the increment and close the fd. counters.elements[idx].sub(1); + closeSync(fd); } }; -- 2.51.2