valkey: close the socket on every fail() and mark the client disconnected before onclose runs

src/runtime/valkey_jsc/js_valkey.rs

@@ -1937,10 +1937,10 @@
         let _guard = this.ref_scope();
         // Ensure the socket pointer is updated.
         this.client_mut().socket = Socket::SocketTcp(uws::SocketTCP::detached());
-        let _defer = scopeguard::guard(BackRef::new(this), |p| {
-            p.client_mut().status = valkey::Status::Disconnected;
-            p.update_poll_ref();
-        });
+        // Before `on_close()`: it runs `onclose` and settles the connect()
+        // promise, and a connect() called from either must see Disconnected.
+        this.client_mut().status = valkey::Status::Disconnected;
+        let _defer = scopeguard::guard(BackRef::new(this), |p| p.update_poll_ref());
 
         let _ = this.client_mut().on_close(); // TODO: properly propagate exception upwards
     }
@@ -1962,10 +1962,8 @@
         // Ensure the socket pointer is updated.
         this.client_mut().socket = Socket::SocketTcp(uws::SocketTCP::detached());
         let _guard = this.ref_scope();
-        let _defer = scopeguard::guard(BackRef::new(this), |p| {
-            p.client_mut().status = valkey::Status::Disconnected;
-            p.update_poll_ref();
-        });
+        this.client_mut().status = valkey::Status::Disconnected;
+        let _defer = scopeguard::guard(BackRef::new(this), |p| p.update_poll_ref());
 
         narrow_terminated(this.client_mut().on_close())
     }

src/runtime/valkey_jsc/valkey.rs

@@ -33,12 +33,12 @@
     pub(crate) is_selecting_db_internal: bool,
     pub(crate) enable_offline_queue: bool,
     pub(crate) enable_auto_reconnect: bool,
-    /// Sticky until the next accepted HELLO, so it overlaps `Connecting`
-    /// (`reconnect()` reads it there) and `failed` (`update_poll_ref` reads it
-    /// there); that is why it is not a `Status` variant.
+    /// Set from the close that schedules a retry until the next accepted HELLO
+    /// or `fail()`, so it overlaps `Disconnected` and `Connecting`; that is why
+    /// it is not a `Status` variant.
     pub(crate) is_reconnecting: bool,
-    /// Sticky until `on_open`/`connect()`, and orthogonal to `Status`: `fail()`
-    /// while `Connected` leaves the socket open and `status` unchanged.
+    /// Sticky until `on_open`/`connect()`; the socket is closed when it is set,
+    /// so it overlaps `Disconnected`.
     pub(crate) failed: bool,
     pub(crate) enable_auto_pipelining: bool,
     pub(crate) finalized: bool,
@@ -607,17 +607,16 @@
             return Ok(());
         }
         self.flags.failed = true;
+        self.flags.is_reconnecting = false;
         let val = Self::reject_all_pending_commands(
             &mut self.in_flight,
             &mut self.queue,
             global_this,
             jsvalue,
         );
 
-        if !self.connection_ready() {
-            self.flags.is_manually_closed = true;
-            self.close();
-        }
+        self.flags.is_manually_closed = true;
+        self.close();
         val
     }
 

test/js/valkey/reliability/connection-failures.test.ts

@@ -1,5 +1,6 @@
 import { RedisClient } from "bun";
 import { describe, expect, mock, test } from "bun:test";
+import { bunEnv, bunExe } from "harness";
 import net from "net";
 import { DEFAULT_REDIS_OPTIONS, DEFAULT_REDIS_URL, delay, isEnabled } from "../test-utils";
 
@@ -441,3 +442,91 @@ describe("Valkey: Auto-Reconnect In-Flight Commands", () => {
     }
   });
 });
+
+describe("Valkey: Recovering After fail()", () => {
+  function helloServer(onConnection?: (socket: net.Socket, connection: number) => void) {
+    let connections = 0;
+    const server = net.createServer(socket => {
+      connections += 1;
+      onConnection?.(socket, connections);
+      socket.on("data", chunk => {
+        const text = chunk.toString("latin1");
+        if (text.includes("HELLO")) socket.write("+OK\r\n");
+        if (text.includes("PING")) socket.write("+PONG\r\n");
+      });
+    });
+    return {
+      server,
+      get connections() {
+        return connections;
+      },
+      listen: () =>
+        new Promise<number>(resolve =>
+          server.listen(0, "127.0.0.1", () => resolve((server.address() as net.AddressInfo).port)),
+        ),
+    };
+  }
+
+  test("an idle timeout while connected closes the socket, fires onclose, and connect() reconnects", async () => {
+    const closed = Promise.withResolvers<Error>();
+    const fake = helloServer();
+    const port = await fake.listen();
+    try {
+      const client = new RedisClient(`redis://127.0.0.1:${port}`, {
+        // The timer armed by connect() is only re-armed by send(), so on an idle
+        // connection the idle timeout fires once connectionTimeout elapses.
+        connectionTimeout: 100,
+        idleTimeout: 50,
+        autoReconnect: false,
+      });
+      client.onclose = err => closed.resolve(err);
+      await client.connect();
+      expect(client.connected).toBe(true);
+      const err = await closed.promise;
+      expect(err).toBeInstanceOf(Error);
+      expect(client.connected).toBe(false);
+      await client.connect();
+      expect(await client.ping()).toBe("PONG");
+      expect(fake.connections).toBe(2);
+      client.close();
+    } finally {
+      fake.server.close();
+    }
+  });
+
+  test("connect() rejects again after a failed attempt instead of hanging", async () => {
+    // Nothing listens on the port a just-closed listener used.
+    const fake = helloServer();
+    const port = await fake.listen();
+    await new Promise(resolve => fake.server.close(resolve));
+    const client = new RedisClient(`redis://127.0.0.1:${port}`, { autoReconnect: false });
+    expect(client.connect()).rejects.toThrow();
+    await client.connect().catch(() => {});
+    expect(client.connect()).rejects.toThrow();
+    await client.connect().catch(() => {});
+    client.close();
+  });
+
+  test("the process exits once auto-reconnect gives up", async () => {
+    const fake = helloServer();
+    const port = await fake.listen();
+    await new Promise(resolve => fake.server.close(resolve));
+    await using proc = Bun.spawn({
+      cmd: [
+        bunExe(),
+        "-e",
+        `
+        const client = new Bun.RedisClient("redis://127.0.0.1:${port}", { autoReconnect: true, maxRetries: 1 });
+        client.onclose = err => console.log("onclose", err.code);
+        await client.connect().catch(err => console.log("connect rejected", err.code));
+        `,
+      ],
+      env: bunEnv,
+      stdout: "pipe",
+      stderr: "pipe",
+    });
+    const [stdout, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]);
+    expect(stdout).toContain("connect rejected");
+    expect(exitCode).toBe(0);
+  });
+});