From 257870a4aaa090614c3c505c9a41ef8ad84f1665 Mon Sep 17 00:00:00 2001 From: wallpants <47203170+wallpants@users.noreply.github.com> Date: Sat, 18 Jul 2026 14:06:04 -0600 Subject: [PATCH 1/2] feat(attach): add onDisconnect, socket backpressure handling & timeouts - onDisconnect(callback) fires once when the connection closes for any reason (neovim exit, socket error, detach); registering after close still notifies asynchronously - socket writes respect partial writes and resume on "drain"; previously any RPC message larger than the kernel socket buffer (~8kB on macOS unix sockets) was silently truncated, permanently desyncing the msgpack stream - attach() rejects and closes the socket if the handshake goes unanswered (default 10s, configurable via timeouts.attach) - call() accepts an optional timeout, per call or via timeouts.request; disabled by default since neovim may legitimately block indefinitely --- src/attach.test.ts | 86 +++++++++++++++++++++++++++++ src/attach.ts | 106 +++++++++++++++++++++++++++++++----- src/test-utils/fake-nvim.ts | 9 +-- src/types.ts | 50 +++++++++++++++++ 4 files changed, 232 insertions(+), 19 deletions(-) diff --git a/src/attach.test.ts b/src/attach.test.ts index 2807eeb..e36fa86 100644 --- a/src/attach.test.ts +++ b/src/attach.test.ts @@ -229,6 +229,92 @@ test("pending calls reject when the connection closes", async () => { await expectReject(nvim.call("nvim_get_current_line", []), "connection closed"); }); +test("large messages survive socket backpressure", async () => { + const fake = startFake(); + const nvim = await connect(fake); + + fake.onRequest((request) => [ + MessageType.RESPONSE, + request[1], + null, + ((request[3] as string[])[0] ?? "").length, + ]); + + // far larger than the kernel socket buffer, forcing partial writes + const big = "x".repeat(8 * 1024 * 1024); + const result = await nvim.call("nvim_exec_lua", [big, []]); + expect(result).toBe(big.length); + + // stream must still be intact afterwards + fake.onRequest((request) => [MessageType.RESPONSE, request[1], null, "ok"]); + expect(await nvim.call("nvim_get_current_line", [])).toBe("ok"); +}); + +test("call rejects after the configured timeout", async () => { + const fake = startFake(); + const nvim = await connect(fake); + + fake.onRequest(() => undefined); + await expectReject( + nvim.call("nvim_get_current_line", [], { timeout: 50 }), + "timed out after 50ms", + ); + + // a late response for the timed-out request must not break the connection + const timedOut = fake.received.find( + (message) => message[0] === MessageType.REQUEST && message[2] === "nvim_get_current_line", + ) as RPCRequest | undefined; + if (!timedOut) throw new Error("unreachable"); + fake.send([MessageType.RESPONSE, timedOut[1], null, "too late"]); + await Bun.sleep(10); + + fake.onRequest((request) => [MessageType.RESPONSE, request[1], null, "still works"]); + expect(await nvim.call("nvim_get_current_line", [])).toBe("still works"); +}); + +test("attach rejects when the handshake is not answered", async () => { + const fake = startFakeNvim({ handshake: false }); + fakes.push(fake); + + await expectReject( + attach({ + socket: fake.socketPath, + client: { name: "bunvim-test" }, + timeouts: { attach: 50 }, + }), + "timed out after 50ms", + ); +}); + +test("onDisconnect fires when neovim closes the connection", async () => { + const fake = startFake(); + const nvim = await connect(fake); + + const errors: Error[] = []; + nvim.onDisconnect((error) => errors.push(error)); + + fake.close(); + + await waitFor(() => (errors.length === 1 ? true : undefined)); + expect(errors[0]?.message).toContain("connection closed"); +}); + +test("onDisconnect fires on detach and immediately when already closed", async () => { + const fake = startFake(); + const nvim = await connect(fake); + + const calls: string[] = []; + nvim.onDisconnect(() => calls.push("before")); + + nvim.detach(); + expect(calls).toEqual(["before"]); + + // registering after close still notifies (asynchronously) + nvim.onDisconnect(() => calls.push("after")); + await Bun.sleep(1); + expect(calls).toEqual(["before", "after"]); +}); + test("detach rejects pending calls and is idempotent", async () => { const fake = startFake(); const nvim = await connect(fake); diff --git a/src/attach.ts b/src/attach.ts index b15d9f3..70cf2bf 100644 --- a/src/attach.ts +++ b/src/attach.ts @@ -39,9 +39,13 @@ export async function attach({ socket, client, logging, + timeouts, }: AttachParams): Promise> { const logger = createLogger(client, logging); + const attachTimeout = timeouts?.attach ?? 10_000; + const requestTimeout = timeouts?.request; const messageOutQueue: RPCMessage[] = []; + const disconnectHandlers = new Set<(error: Error) => void>(); const notificationHandlers = new Map>(); const requestHandlers = new Map(); const pendingRequests = new Map< @@ -56,16 +60,29 @@ export async function attach({ let lastReqId = 0; let handlerId = 0; let closed = false; + let closedError: Error | undefined; + // remainder of a partially written message, pending socket "drain" + let writeBuffer: Uint8Array | undefined; function fail(error: Error) { if (closed) return; closed = true; + closedError = error; messageOutQueue.length = 0; + writeBuffer = undefined; unpackrStream.removeAllListeners(); for (const { reject } of pendingRequests.values()) { reject(error); } pendingRequests.clear(); + for (const callback of disconnectHandlers) { + try { + callback(error); + } catch (callbackError) { + logger?.error("onDisconnect handler error", callbackError); + } + } + disconnectHandlers.clear(); } const nvimSocket = await Bun.connect({ @@ -75,6 +92,10 @@ export async function attach({ data(_, data) { unpackrStream.write(data); }, + drain() { + // kernel buffer has space again, resume writing + processMessageOutQueue(); + }, error(_, error) { logger?.error("socket error", error); fail(error); @@ -92,11 +113,25 @@ export async function attach({ function processMessageOutQueue() { // All writing to neovim happens through this function. - // Outgoing RPC messages are added to the `messageOutQueue` and sent ASAP - let message: RPCMessage | undefined; - while ((message = messageOutQueue.shift()) !== undefined) { - logger?.debug(prettyRPCMessage(message, "out")); - nvimSocket.write(packr.pack(message)); + // Outgoing RPC messages are added to the `messageOutQueue` and sent ASAP. + // `socket.write` may write fewer bytes than provided when the kernel + // buffer is full; the remainder is kept in `writeBuffer` and flushed + // by the socket's "drain" callback. + if (closed) return; + for (;;) { + if (!writeBuffer) { + const message = messageOutQueue.shift(); + if (message === undefined) return; + logger?.debug(prettyRPCMessage(message, "out")); + writeBuffer = packr.pack(message); + } + const written = nvimSocket.write(writeBuffer); + if (written < writeBuffer.byteLength) { + // kernel buffer is full, resume writing on "drain" + if (written > 0) writeBuffer = writeBuffer.subarray(written); + return; + } + writeBuffer = undefined; } } @@ -179,32 +214,62 @@ export async function attach({ })().catch((error: unknown) => logger?.error("unpackrStream error", error)); }); - const call: Nvim["call"] = (func, args) => { + const call: Nvim["call"] = (func, args, opts) => { if (closed) { return Promise.reject(new Error("connection closed")); } const reqId = ++lastReqId; const request: RPCRequest = [MessageType.REQUEST, reqId, func, args]; + const timeout = opts?.timeout ?? requestTimeout; return new Promise((resolve, reject) => { + let timer: ReturnType | undefined; + if (timeout) { + timer = setTimeout(() => { + pendingRequests.delete(reqId); + reject(new Error(`request "${func}" timed out after ${timeout}ms`)); + }, timeout); + } // Register before adding request to queue to avoid // response coming in before we're ready to handle it. - pendingRequests.set(reqId, { resolve, reject }); + pendingRequests.set(reqId, { + resolve: (result) => { + clearTimeout(timer); + resolve(result); + }, + reject: (error) => { + clearTimeout(timer); + reject(error); + }, + }); messageOutQueue.push(request); processMessageOutQueue(); }); }; - await call("nvim_set_client_info", [ - client.name, - client.version ?? {}, - client.type ?? "msgpack-rpc", - client.methods ?? {}, - client.attributes ?? {}, - ]); + let channelId: number; + try { + await call( + "nvim_set_client_info", + [ + client.name, + client.version ?? {}, + client.type ?? "msgpack-rpc", + client.methods ?? {}, + client.attributes ?? {}, + ], + { timeout: attachTimeout }, + ); - const channelId = (await call("nvim_get_api_info", []))[0] as number; + channelId = (await call("nvim_get_api_info", [], { timeout: attachTimeout }))[0] as number; + } catch (error) { + // if the handshake fails or times out, don't leave a dangling socket + const failure = toError(error); + fail(failure); + nvimSocket.end(); + throw failure; + } return { call, @@ -219,6 +284,17 @@ export async function attach({ onRequest(method, callback) { requestHandlers.set(method as string, callback as UntypedHandler); }, + onDisconnect(callback) { + if (closed) { + // already disconnected, still notify, but asynchronously + const error = closedError ?? new Error("connection closed"); + queueMicrotask(() => { + callback(error); + }); + return; + } + disconnectHandlers.add(callback); + }, detach() { fail(new Error("connection closed by bunvim")); nvimSocket.end(); diff --git a/src/test-utils/fake-nvim.ts b/src/test-utils/fake-nvim.ts index 0929207..dd0fec2 100644 --- a/src/test-utils/fake-nvim.ts +++ b/src/test-utils/fake-nvim.ts @@ -24,7 +24,8 @@ export type FakeNvim = { close(): void; }; -export function startFakeNvim(): FakeNvim { +export function startFakeNvim(options?: { handshake?: boolean }): FakeNvim { + const handshake = options?.handshake ?? true; const socketPath = join(tmpdir(), `bunvim-test-${crypto.randomUUID()}.sock`); const received: RPCMessage[] = []; const unpackrStream = new UnpackrStream({ useRecords: false }); @@ -41,13 +42,13 @@ export function startFakeNvim(): FakeNvim { received.push(message); if (message[0] !== MessageType.REQUEST) return; - // auto-answer the attach handshake + // auto-answer the attach handshake (unless disabled via options) if (message[2] === "nvim_set_client_info") { - send([MessageType.RESPONSE, message[1], null, null]); + if (handshake) send([MessageType.RESPONSE, message[1], null, null]); return; } if (message[2] === "nvim_get_api_info") { - send([MessageType.RESPONSE, message[1], null, [FAKE_CHANNEL_ID, {}]]); + if (handshake) send([MessageType.RESPONSE, message[1], null, [FAKE_CHANNEL_ID, {}]]); return; } diff --git a/src/types.ts b/src/types.ts index c59da6e..c25ccdb 100644 --- a/src/types.ts +++ b/src/types.ts @@ -109,6 +109,29 @@ export type AttachParams = { /** * If left undefined, logging will be disabled. */ + /** + * Timeouts in milliseconds. + */ + timeouts?: { + /** + * Max time to wait for the initial handshake with neovim + * (`nvim_set_client_info` + `nvim_get_api_info`) before `attach` + * rejects and closes the socket. Set to `0` to disable. + * + * @default 10_000 + */ + attach?: number; + /** + * Default max time to wait for a response to any `call()` before + * rejecting. Disabled by default: neovim may legitimately take + * arbitrarily long to respond (e.g. blocked on user input). + * Can be overridden per call. + * + * @default undefined (no timeout) + */ + request?: number; + }; + logging?: { /** * @remarks @@ -155,6 +178,14 @@ export type RPCResponse = [ ]; export type RPCMessage = RPCRequest | RPCNotification | RPCResponse; +export type CallOptions = { + /** + * Reject the request if no response is received within `timeout` ms. + * Overrides `AttachParams.timeouts.request`. + */ + timeout?: number; +}; + export type EventHandler = (args: Args) => Awaitable; export type NotificationHandler = EventHandler; export type RequestHandler = EventHandler; @@ -176,10 +207,12 @@ export type Nvim = { * * @param func - function name * @param args - function arguments, provide empty array `[]` if no args + * @param opts - per-call options (timeout) */ call( func: M, args: NeovimApi["functions"][M]["parameters"], + opts?: CallOptions, ): Promise; /** * @@ -242,6 +275,23 @@ export type Nvim = { method: M, callback: EventHandler, ): void; + /** + * + * Register a handler to be called once when the connection to neovim + * closes for any reason (neovim exited, socket error, or `detach()`). + * + * If the connection is already closed when this is called, + * the handler is invoked asynchronously right away. + * + * @example + * ```typescript + * nvim.onDisconnect(() => { + * // stop servers, exit the process, etc. + * process.exit(0); + * }); + * ``` + */ + onDisconnect(callback: (error: Error) => void): void; /** * * Close socket connection to neovim. From 38a0a314705b0ffd4962d24b888a682227d38d8c Mon Sep 17 00:00:00 2001 From: wallpants <47203170+wallpants@users.noreply.github.com> Date: Sat, 18 Jul 2026 14:06:04 -0600 Subject: [PATCH 2/2] chore(publish): replace .npmignore with files allowlist --- .npmignore | 4 ---- package.json | 6 ++++++ 2 files changed, 6 insertions(+), 4 deletions(-) delete mode 100644 .npmignore diff --git a/.npmignore b/.npmignore deleted file mode 100644 index e76500a..0000000 --- a/.npmignore +++ /dev/null @@ -1,4 +0,0 @@ -docs -bun.lockb -src/**/*.test.ts -src/test-utils diff --git a/package.json b/package.json index 5fe263b..28aa433 100644 --- a/package.json +++ b/package.json @@ -11,6 +11,12 @@ "bin": { "bunvim": "src/cli/index.ts" }, + "files": [ + "src", + "!src/**/*.test.ts", + "!src/test-utils", + "!src/tests.ts" + ], "type": "module", "exports": { "types": "./src/index.ts",