Dev

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",

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);

src/attach.ts

@@ -39,9 +39,13 @@ export async function attach<ApiInfo extends BaseEvents = BaseEvents>({
    socket,
    client,
    logging,
+   timeouts,
 }: AttachParams): Promise<Nvim<ApiInfo>> {
    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<string, Map<number, UntypedHandler>>();
    const requestHandlers = new Map<string, UntypedHandler>();
    const pendingRequests = new Map<
@@ -56,16 +60,29 @@ export async function attach<ApiInfo extends BaseEvents = BaseEvents>({
    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<ApiInfo extends BaseEvents = BaseEvents>({
          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<ApiInfo extends BaseEvents = BaseEvents>({
 
    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<ApiInfo extends BaseEvents = BaseEvents>({
       })().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<typeof setTimeout> | 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<ApiInfo extends BaseEvents = BaseEvents>({
       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();

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;
       }
 

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, Returns> = (args: Args) => Awaitable<Returns>;
 export type NotificationHandler = EventHandler<unknown[], void>;
 export type RequestHandler = EventHandler<unknown[], unknown>;
@@ -176,10 +207,12 @@ export type Nvim<ApiInfo extends BaseEvents = BaseEvents> = {
     *
     * @param func - function name
     * @param args - function arguments, provide empty array `[]` if no args
+    * @param opts - per-call options (timeout)
     */
    call<M extends keyof NeovimApi["functions"]>(
       func: M,
       args: NeovimApi["functions"][M]["parameters"],
+      opts?: CallOptions,
    ): Promise<NeovimApi["functions"][M]["return_type"]>;
    /**
     *
@@ -242,6 +275,23 @@ export type Nvim<ApiInfo extends BaseEvents = BaseEvents> = {
       method: M,
       callback: EventHandler<ApiInfo["requests"][M], unknown>,
    ): 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.