diff --git a/src/runtime/valkey_jsc/js_valkey.rs b/src/runtime/valkey_jsc/js_valkey.rs
index dd35845d12fe..d5f4daa7ac0f 100644
--- a/src/runtime/valkey_jsc/js_valkey.rs
+++ b/src/runtime/valkey_jsc/js_valkey.rs
@@ -1018,16 +1018,13 @@ impl JSValkeyClient {
             self.poll_ref.with_mut(|r| r.ref_(vm_event_loop_ctx()));
 
             if let Err(err) = self.connect() {
-                self.poll_ref.with_mut(|r| r.unref(vm_event_loop_ctx()));
-                self.client_mut().status = valkey::Status::NeverConnected;
-                let err_value = global_object
-                    .err(
-                        jsc::ErrorCode::SOCKET_CLOSED_BEFORE_CONNECTION,
-                        format_args!(" {} connecting to Valkey", err.name()),
-                    )
-                    .to_js();
-                let _exit = self.vm().enter_event_loop_scope();
-                promise_ptr.reject(global_object, Ok(err_value))?;
+                debug!(
+                    "first dial failed before a socket was opened: {}",
+                    err.name()
+                );
+                // Settled by the deferred close like a refused dial: the
+                // promise, onclose and the retry policy all go through on_close().
+                self.close_without_socket_next_tick();
                 return Ok(promise);
             }
 
@@ -1142,6 +1139,38 @@ impl JSValkeyClient {
         }
     }
 
+    /// Runs `ValkeyClient::on_close()` for a dial that failed before there was
+    /// a socket, so the connect() promise, `onclose`, the retry policy and the
+    /// poll ref are handled as for a dial that failed asynchronously. Deferred
+    /// because `onclose` may call connect(), and a dial that fails the same way
+    /// from in there would otherwise re-enter `on_close()` on the same stack.
+    ///
+    /// Until the task runs the client is `Connecting`, as it would be with a
+    /// dial in flight: JS that runs in between (timers due in the same tick,
+    /// or the caller of connect() or of the command that dialled) then gets
+    /// the cached promise from connect() instead of a second dial, has its
+    /// commands queued for the retry or the rejection, and a disconnect()
+    /// marks the close as manual for the task to honour. `update_poll_ref`
+    /// keeps the wrapper and the event loop alive for it like a dial would.
+    fn close_without_socket_next_tick(&self) {
+        self.client_mut().status = valkey::Status::Connecting;
+        self.update_poll_ref();
+        self.enqueue_deferred_close(DeferredClose::WithoutSocket);
+    }
+
+    fn enqueue_deferred_close(&self, what: DeferredClose) {
+        // Released by the task, whether it runs or the VM tears down first.
+        self.ref_();
+        let task = jsc::Task::from_boxed(Box::new(ValkeyDeferredClose {
+            ctx: self.as_ctx_ptr(),
+            what,
+        }));
+        // SAFETY: VM-owned event loop pointer; uniquely accessed on the JS thread.
+        unsafe {
+            (*self.vm().event_loop()).enqueue_task(task);
+        }
+    }
+
     pub(crate) fn on_reconnect_timer(&self) -> JsResult<()> {
         debug!("Reconnect timer fired, attempting to reconnect");
 
@@ -1175,15 +1204,14 @@ impl JSValkeyClient {
         });
 
         if let Err(err) = self.connect() {
-            self.poll_ref.with_mut(|r| r.disable());
-            return self.fail_with_js_value(
-                self.global_object
-                    .err(
-                        jsc::ErrorCode::SOCKET_CLOSED_BEFORE_CONNECTION,
-                        format_args!("{} reconnecting", err.name()),
-                    )
-                    .to_js(),
+            debug!(
+                "reconnect failed before a socket was opened: {}",
+                err.name()
             );
+            // Same outcome as a dial that fails asynchronously: another retry,
+            // or fail() and a settled connect() promise once retries are used up.
+            self.close_without_socket_next_tick();
+            return Ok(());
         }
 
         // Reset the socket timeout
@@ -1196,6 +1224,8 @@ impl JSValkeyClient {
         debug_assert!(self.client.get().status == valkey::Status::Connected);
         // we should always have a strong reference to the object here
         debug_assert!(self.this_value.get().is_strong());
+        // Now counting idle time, not connect time.
+        self.reset_connection_timeout();
 
         let self_ptr = self.as_ctx_ptr();
         let _defer = scopeguard::guard(self_ptr, |p| {
@@ -1221,7 +1251,6 @@ impl JSValkeyClient {
                     let error = global_object.take_exception(err);
                     let client = self.client_mut();
                     client.flags.connection_promise_returns_client = false;
-                    client.flags.is_manually_closed = true;
                     let rejected = match Js::connection_promise_get_cached(this_value) {
                         Some(promise) => {
                             Js::connection_promise_set_cached(
@@ -1234,10 +1263,16 @@ impl JSValkeyClient {
                         }
                         None => Ok(()),
                     };
-                    let failed =
-                        rejected.and_then(|()| client.fail_with_js_value(&global_object, error));
-                    let closed = self.client_mut().close();
-                    return failed.and(closed);
+                    // `fail_with_js_value` closes the socket itself; a second close here would
+                    // hit whatever a connect() from `onclose` just opened.
+                    return match rejected {
+                        Ok(()) => client.fail_with_js_value(&global_object, error),
+                        Err(_) => {
+                            client.flags.is_manually_closed = true;
+                            let closed = client.close(uws::CloseCode::Failure);
+                            rejected.and(closed)
+                        }
+                    };
                 }
             };
             Js::hello_set_cached(this_value, &global_object, hello_value);
@@ -1343,11 +1378,16 @@ impl JSValkeyClient {
     // Callback for when Valkey client needs to reconnect
     pub(crate) fn on_valkey_reconnect(&self) {
         // SAFETY: adopts connect()'s socket keep-alive ref for the just-closed
-        // socket. Reached only from `ValkeyClient::on_close()`'s reconnect
-        // branch, which never calls `on_valkey_close()`, so this scope is the
-        // sole releaser. The caller holds its own scoped ref, so count > 0.
+        // socket (or the one `ValkeyDeferredClose::run` took in its place).
+        // Reached only from `ValkeyClient::on_close()`'s reconnect branch,
+        // which never calls `on_valkey_close()`, so this scope is the sole
+        // releaser. The caller holds its own scoped ref, so count > 0.
         let _socket_ref = unsafe { ScopedRef::adopt(self.as_ctx_ptr()) };
 
+        // This timer was bounding the attempt that just ended; left armed it
+        // fires during the retry delay, and `fail()` then has no socket to
+        // close and nothing settles connect(). `reconnect()` arms a new one.
+        self.timer.disarm(self);
         self.reconnect_timer
             .arm(self, self.client.get().get_reconnect_delay());
     }
@@ -1356,8 +1396,9 @@ impl JSValkeyClient {
     pub(crate) fn on_valkey_close(&self) -> JsResult<()> {
         let global_object = self.global_object;
 
-        // SAFETY: adopts connect()'s socket keep-alive ref; the caller holds
-        // its own scoped ref so count stays > 0 until this drops.
+        // SAFETY: adopts connect()'s socket keep-alive ref (or the one
+        // `ValkeyDeferredClose::run` took in its place); the caller holds its
+        // own scoped ref so count stays > 0 until this drops.
         let _socket_ref = unsafe { ScopedRef::adopt(self.as_ctx_ptr()) };
         let _defer = scopeguard::guard(BackRef::new(self), |p| p.update_poll_ref());
 
@@ -1366,7 +1407,8 @@ impl JSValkeyClient {
         };
         this_jsvalue.ensure_still_alive();
         if global_object.has_exception() {
-            // As in `fail_with_js_value`: an exception already unwinding keeps propagating.
+            // Already unwinding an exception (a caller's, or the worker's termination): it keeps
+            // propagating; `onclose` cannot be entered on top of it.
             return Err(bun_jsc::JsError::Thrown);
         }
 
@@ -1402,37 +1444,13 @@ impl JSValkeyClient {
         self.client_mut().fail(message, err)
     }
 
-    pub(crate) fn fail_with_js_value(&self, value: JSValue) -> JsResult<()> {
-        let Some(this_value) = self.this_value.get().try_get() else {
-            return Ok(());
-        };
-        let global_object = self.global_object;
-        if global_object.has_exception() {
-            // Already unwinding an exception (a caller's, or the worker's termination): it keeps
-            // propagating; `onclose` cannot be entered on top of it.
-            return Err(bun_jsc::JsError::Thrown);
-        }
-        if let Some(on_close) = Js::onclose_get_cached(this_value) {
-            let _exit = self.vm().enter_event_loop_scope();
-            on_close.call(&global_object, this_value, &[value])?;
-        }
-        Ok(())
-    }
-
     fn close_socket_next_tick(&self) {
         if self.client.get().socket.is_closed() {
             return;
         }
 
-        self.ref_();
         // socket close can potentially call JS so we need to enqueue the deinit
-        let task = jsc::Task::from_boxed(Box::new(ValkeyDeferredClose {
-            ctx: self.as_ctx_ptr(),
-        }));
-        // SAFETY: VM-owned event loop pointer; uniquely accessed on the JS thread.
-        unsafe {
-            (*self.vm().event_loop()).enqueue_task(task);
-        }
+        self.enqueue_deferred_close(DeferredClose::Socket);
     }
 
     pub fn finalize(self: Box<Self>) {
@@ -1466,12 +1484,6 @@ impl JSValkeyClient {
 
         let _guard = self.ref_scope();
 
-        // Socket keep-alive ref, released by on_valkey_close/on_valkey_reconnect.
-        // Taken before the TLS-context check so the `tls_ctx_failed` branch's
-        // `on_valkey_close()` has a ref to consume instead of over-releasing.
-        // Forgotten on success (the socket adopts it).
-        let socket_ref = self.ref_scope();
-
         let is_tls = self.client.get().tls != valkey::TLS::None;
         let vm = self.client.get().vm.as_mut();
         let loop_ = vm.uws_loop();
@@ -1508,11 +1520,7 @@ impl JSValkeyClient {
                 b"Failed to create TLS context",
                 protocol::RedisError::ConnectionClosed,
             )?;
-            // `on_valkey_close()` consumes the socket ref; hand it over so it
-            // isn't released twice.
-            socket_ref.forget();
-            self.client_mut().on_valkey_close()?;
-            self.client_mut().status = valkey::Status::Disconnected;
+            self.close_without_socket_next_tick();
             return Ok(());
         }
         let ssl_ctx: Option<*mut uws::SslCtx> = match &self.client.get().tls {
@@ -1540,6 +1548,9 @@ impl JSValkeyClient {
         // `owner_ptr` opaquely (no overlapping write).
         let owner_ptr: *mut JSValkeyClient = std::ptr::from_ref::<JSValkeyClient>(self).cast_mut();
         let client_ptr: *mut valkey::ValkeyClient = self.client.as_ptr();
+        // Socket keep-alive ref, released by on_valkey_close/on_valkey_reconnect.
+        // Forgotten once there is a socket to own it.
+        let socket_ref = self.ref_scope();
         // SAFETY: `client_ptr` is live; `group` is the lazy-initialised per-VM
         // `SocketGroup` (stable for the VM's lifetime). `ssl_ctx` is a +1-ref
         // BoringSSL `SSL_CTX*` (or None) forwarded opaquely to usockets.
@@ -1568,20 +1579,19 @@ impl JSValkeyClient {
         if self.client.get().status == valkey::Status::NeverConnected {
             bun_core::hint::cold();
 
-            if let Err(err) = self.connect() {
-                self.client_mut().status = valkey::Status::NeverConnected;
-                let err_value = global_this
-                    .err(
-                        jsc::ErrorCode::SOCKET_CLOSED_BEFORE_CONNECTION,
-                        format_args!(" {} connecting to Valkey", err.name()),
-                    )
-                    .to_js();
-                let promise = JSPromise::create(global_this);
-                let _exit = self.vm().enter_event_loop_scope();
-                promise.reject(global_this, Ok(err_value))?;
-                return Ok(promise);
+            match self.connect() {
+                // The command is queued below as for a dial in flight; the
+                // deferred close then rejects it or a retry sends it, like a
+                // refused dial.
+                Err(err) => {
+                    debug!(
+                        "first dial failed before a socket was opened: {}",
+                        err.name()
+                    );
+                    self.close_without_socket_next_tick();
+                }
+                Ok(()) => self.reset_connection_timeout(),
             }
-            self.reset_connection_timeout();
         }
 
         let self_br = BackRef::new(self);
@@ -1853,12 +1863,8 @@ impl<const SSL: bool> SocketHandler<SSL> {
         err_value: JSValue,
     ) -> JsResult<()> {
         let _exit = this.vm().enter_event_loop_scope();
-        this.client_mut().flags.is_manually_closed = true;
-        let failed = this
-            .client_mut()
-            .fail_with_js_value(&this.global_object, err_value);
-        let closed = this.client_mut().close();
-        failed.and(closed)
+        this.client_mut()
+            .fail_with_js_value(&this.global_object, err_value)
     }
 
     pub(crate) const ON_HANDSHAKE: Option<
@@ -1875,10 +1881,10 @@ impl<const SSL: bool> SocketHandler<SSL> {
         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());
 
         this.client_mut().on_close()
     }
@@ -1900,10 +1906,8 @@ impl<const SSL: bool> SocketHandler<SSL> {
         // 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());
 
         this.client_mut().on_close()
     }
@@ -1925,6 +1929,9 @@ impl<const SSL: bool> SocketHandler<SSL> {
 
         let _guard = this.ref_scope();
         let result = this.client_mut().on_data(data);
+        if this.client.get().status == valkey::Status::Connected {
+            this.reset_connection_timeout();
+        }
         this.update_poll_ref();
         result
     }
@@ -2019,27 +2026,67 @@ impl Options {
     }
 }
 
+#[derive(Clone, Copy)]
+enum DeferredClose {
+    /// Close the socket the finalized wrapper left behind.
+    Socket,
+    /// Run the close path for a dial that never produced a socket
+    /// (`close_without_socket_next_tick`).
+    WithoutSocket,
+}
+
 pub(crate) struct ValkeyDeferredClose {
-    ctx: *const JSValkeyClient,
+    ctx: *mut JSValkeyClient,
+    what: DeferredClose,
 }
 
 impl ValkeyDeferredClose {
     #[allow(clippy::boxed_local, reason = "reclaim point for the boxed task")]
     pub(crate) fn run(self: Box<Self>) {
-        let ctx = self.ctx;
-        // SAFETY: single-threaded; intrusive ref taken before enqueue guarantees liveness.
-        unsafe {
-            crate::dispatch::fold((*ctx).client_mut().close());
-            JSValkeyClient::deref(ctx.cast_mut());
+        // SAFETY: adopts the ref `enqueue_deferred_close` took, which kept the
+        // client alive until now; released when this scope ends.
+        let _enqueue_ref = unsafe { ScopedRef::adopt(self.ctx) };
+        // SAFETY: live per the ref above; tasks run on the JS thread.
+        let this = unsafe { &*self.ctx };
+        match self.what {
+            DeferredClose::Socket => {
+                crate::dispatch::fold(this.client_mut().close(uws::CloseCode::FastShutdown))
+            }
+            DeferredClose::WithoutSocket => {
+                // Holding Connecting (see `close_without_socket_next_tick`) is
+                // what keeps a dial from starting in between; if one did, its
+                // own callbacks own the close path now, so only drop our ref.
+                if !this.client.get().socket.is_closed() {
+                    return;
+                }
+                // `on_close()` ends in `on_valkey_close`/`on_valkey_reconnect`,
+                // which release the ref the socket would have held.
+                this.ref_();
+                this.client_mut().status = valkey::Status::Disconnected;
+                let closed = this.client_mut().on_close();
+                this.update_poll_ref();
+                crate::dispatch::fold(closed);
+            }
         }
     }
 }
 
 impl bun_event_loop::Taskable for ValkeyDeferredClose {
     const TAG: bun_event_loop::TaskTag = bun_event_loop::task_tag::ValkeyDeferredClose;
-    /// The deferred close is script-free bookkeeping; do it.
     unsafe fn release_unrun(this: *mut Self) {
         // SAFETY: fn contract — boxed at the enqueue site.
-        unsafe { bun_core::heap::take(this) }.run();
+        let task = unsafe { bun_core::heap::take(this) };
+        match task.what {
+            // Script-free bookkeeping; do it.
+            DeferredClose::Socket => task.run(),
+            // The VM is going away: `on_close()` would run `onclose`, so only
+            // give back what `close_without_socket_next_tick` took.
+            DeferredClose::WithoutSocket => {
+                // SAFETY: as in `run`.
+                let _enqueue_ref = unsafe { ScopedRef::adopt(task.ctx) };
+                // SAFETY: live per the ref above.
+                unsafe { &*task.ctx }.poll_ref.with_mut(|r| r.disable());
+            }
+        }
     }
 }
diff --git a/src/runtime/valkey_jsc/valkey.rs b/src/runtime/valkey_jsc/valkey.rs
index b1f9cdf85317..ff8d11287b11 100644
--- a/src/runtime/valkey_jsc/valkey.rs
+++ b/src/runtime/valkey_jsc/valkey.rs
@@ -31,12 +31,14 @@ pub struct ConnectionFlags {
     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()`. `fail()` closes the socket outright
+    /// (see `close()`), so by the time it returns the close callback has run
+    /// (`on_close` reads this to skip the retry policy) and this overlaps
+    /// `Disconnected`.
     pub(crate) failed: bool,
     pub(crate) enable_auto_pipelining: bool,
     pub(crate) finalized: bool,
@@ -604,6 +606,7 @@ impl ValkeyClient {
             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,
@@ -611,17 +614,24 @@ impl ValkeyClient {
             jsvalue,
         );
 
-        if !self.connection_ready() {
-            self.flags.is_manually_closed = true;
-            let closed = self.close(); // unconditionally, whatever `val` is
-            return val.and(closed);
-        }
-        val
+        // A failure the client detected itself (idle timeout, protocol or
+        // handshake error) has always been a deliberate close; `on_close` reads
+        // `failed` and skips the retry policy. It is not `is_manually_closed`:
+        // that flag is copied into `duplicate()`, and a duplicate of a failed
+        // client should still reconnect.
+        let closed = self.close(uws::CloseCode::Failure); // unconditionally, whatever `val` is
+        val.and(closed)
     }
 
+    /// `fail()` passes `Failure`, the one code whose close callback has run by
+    /// the time this returns (see `CloseCode`); everything after a failure
+    /// relies on that, and an RST instead of a FIN costs nothing once the
+    /// connection's commands have been rejected. `disconnect()` and the
+    /// finalizer pass `FastShutdown`, the graceful close.
+    ///
     /// `Err` when the close event left a termination pending, or, for a half-open socket whose `on_close`
     /// runs by hand here, whatever that left.
-    pub fn close(&mut self) -> JsResult<()> {
+    pub fn close(&mut self, code: uws::CloseCode) -> JsResult<()> {
         if self.socket.is_closed() {
             return Ok(());
         }
@@ -643,7 +653,7 @@ impl ValkeyClient {
         let is_semi_socket = matches!(socket.socket(), uws::InternalSocket::Connected(_))
             && !socket.is_established();
         // TODO: make socket.close() return a JsResult.
-        socket.close(uws::CloseCode::Normal);
+        socket.close(code);
         if global.has_exception() {
             return Err(bun_jsc::JsError::Thrown);
         }
@@ -660,10 +670,14 @@ impl ValkeyClient {
     pub fn on_close(&mut self) -> JsResult<()> {
         self.unregister_auto_flusher();
         self.write_buffer.clear_and_free();
+        // A partial reply can never complete now; left in place it counts as
+        // pending activity in `update_poll_ref` and keeps the event loop alive.
+        self.read_buffer.clear_and_free();
+        self.reply_scanner.reset();
 
-        // If manually closing, don't attempt to reconnect
-        if self.flags.is_manually_closed {
-            debug!("skip reconnecting since the connection is manually closed");
+        // A manual close or a failure the client detected itself: no retry.
+        if self.flags.is_manually_closed || self.flags.failed {
+            debug!("skip reconnecting since the connection is manually closed or failed");
             self.fail(b"Connection closed", RedisError::ConnectionClosed)?;
             self.on_valkey_close()?;
             return Ok(());
@@ -751,6 +765,10 @@ impl ValkeyClient {
             data.len(),
             bstr::BStr::new(data)
         );
+        // Handling a reply can close this socket and, from `onclose` or a
+        // rejection handler, dial the next one; the remaining replies came from
+        // the closed connection and must not reach the new one.
+        let socket = *self.socket.socket();
         // Path 1: Buffer already has data, append and process from buffer
         if !self.read_buffer.remaining().is_empty() {
             self.read_buffer
@@ -818,7 +836,7 @@ impl ValkeyClient {
                 let mut value_to_handle = value; // Use temp var for defer
                 self.handle_response(&mut value_to_handle)?;
 
-                if self.status == Status::Disconnected || self.flags.failed {
+                if *self.socket.socket() != socket {
                     return Ok(());
                 }
                 self.send_next_command();
@@ -877,8 +895,7 @@ impl ValkeyClient {
             let mut value_to_handle = value; // Use temp var for defer
             self.handle_response(&mut value_to_handle)?;
 
-            // Check connection status after handling
-            if self.status == Status::Disconnected || self.flags.failed {
+            if *self.socket.socket() != socket {
                 return Ok(());
             }
 
@@ -1503,7 +1520,7 @@ impl ValkeyClient {
         self.flags.is_manually_closed = true;
         self.unregister_auto_flusher();
         if self.status == Status::Connected || self.status == Status::Connecting {
-            return self.close();
+            return self.close(uws::CloseCode::FastShutdown);
         }
         Ok(())
     }
diff --git a/src/uws_sys/us_socket_t.rs b/src/uws_sys/us_socket_t.rs
index bcdf7377fedb..43edbfb5eeb4 100644
--- a/src/uws_sys/us_socket_t.rs
+++ b/src/uws_sys/us_socket_t.rs
@@ -20,15 +20,26 @@ bun_opaque::opaque_ffi! { pub struct us_socket_t; }
 
 #[repr(i32)]
 #[derive(Copy, Clone, Eq, PartialEq, strum::IntoStaticStr)]
+/// Which of the three codes a close uses decides whether the close callback
+/// has run by the time `close()` returns. Only `failure` guarantees that: for
+/// the other two, `us_internal_ssl_close` (crypto/openssl.c) keeps a TLS
+/// socket open while it still owns the loop's ciphertext spill, i.e. when the
+/// last batch flush hit a full kernel buffer, and finishes the close from the
+/// next writable event or the peer's FIN, which a peer that stopped reading
+/// never produces.
 pub enum CloseCode {
     /// TLS: send close_notify and defer fd close until peer replies. TCP: FIN.
     normal = 0,
-    /// TLS: fast-shutdown (no wait). TCP: SO_LINGER{1,0} → RST, dropping any
-    /// unflushed send buffer. Only for `terminate()` / GC abort.
+    /// Closes now, whatever the peer does: TLS fast-shutdown with no spill
+    /// deferral; TCP SO_LINGER{1,0} → RST, dropping any unflushed send buffer.
+    /// For `terminate()` / GC abort, and for a protocol client that has given
+    /// up on the connection and rejected everything on it (the valkey client's
+    /// `fail()`), whose callers rely on the close callback having run.
     failure = 1,
-    /// TLS: fast-shutdown (no wait). TCP: FIN. For `_handle.close()` where
-    /// the JS wrapper detaches immediately so `.normal`'s deferral would
-    /// orphan the `us_socket_t`, but already-written data must still drain.
+    /// TLS: fast-shutdown, but still deferred while a spill is pending. TCP:
+    /// FIN. For `_handle.close()` where the JS wrapper detaches immediately so
+    /// `.normal`'s deferral would orphan the `us_socket_t`, but already-written
+    /// data must still drain.
     fast_shutdown = 2,
 }
 
diff --git a/test/js/valkey/reliability/connection-failures.test.ts b/test/js/valkey/reliability/connection-failures.test.ts
index bcf8964ac348..6e302acdf79a 100644
--- a/test/js/valkey/reliability/connection-failures.test.ts
+++ b/test/js/valkey/reliability/connection-failures.test.ts
@@ -1,6 +1,10 @@
 import { RedisClient } from "bun";
 import { describe, expect, mock, test } from "bun:test";
+import { once } from "events";
+import { bunEnv, bunExe, isWindows, tempDir, tls as tlsCert } from "harness";
 import net from "net";
+import path from "path";
+import tls from "tls";
 import { DEFAULT_REDIS_OPTIONS, DEFAULT_REDIS_URL, delay, isEnabled } from "../test-utils";
 
 /**
@@ -441,3 +445,882 @@ describe("Valkey: Auto-Reconnect In-Flight Commands", () => {
     }
   });
 });
+
+describe("Valkey: Recovering After fail()", () => {
+  // Answers the chunk carrying HELLO with `+OK` and the one carrying PING with
+  // `+PONG` unless `replies` says otherwise for that connection (a hook that
+  // returns null answers nothing). Other commands are never answered.
+  function helloServer(
+    replies: Partial<Record<"HELLO" | "PING", (connection: number, socket: net.Socket) => string | null>> = {},
+    { secure = false, allowHalfOpen = false } = {},
+  ) {
+    const sockets: net.Socket[] = [];
+    const onConnection = (socket: net.Socket) => {
+      sockets.push(socket);
+      const connection = sockets.length;
+      socket.on("data", chunk => {
+        const text = chunk.toString("latin1");
+        for (const command of ["HELLO", "PING"] as const) {
+          if (!text.includes(command)) continue;
+          const reply = replies[command]
+            ? replies[command](connection, socket)
+            : `+${command === "HELLO" ? "OK" : "PONG"}\r\n`;
+          if (reply !== null) socket.write(reply);
+        }
+      });
+      socket.on("error", () => {});
+    };
+    const server: net.Server = secure
+      ? tls.createServer({ key: tlsCert.key, cert: tlsCert.cert, allowHalfOpen }, onConnection)
+      : net.createServer({ allowHalfOpen }, onConnection);
+    return {
+      server,
+      sockets,
+      get connections() {
+        return sockets.length;
+      },
+      // events.once() rejects if the listen fails instead of leaving the test to time out.
+      listen: async () => {
+        server.listen(0, "127.0.0.1");
+        await once(server, "listening");
+        return (server.address() as net.AddressInfo).port;
+      },
+      listenUnix: async (socketPath: string) => {
+        server.listen(socketPath);
+        await once(server, "listening");
+      },
+      close: () => new Promise(resolve => server.close(resolve)),
+    };
+  }
+
+  // The RST a failed client sends makes the stub's socket emit an error before
+  // it closes, which events.once() would turn into a rejection.
+  function closedOnServer(socket: net.Socket): Promise<void> {
+    return socket.destroyed ? Promise.resolve() : new Promise(resolve => socket.once("close", () => resolve()));
+  }
+
+  // Calls connect() from the first onclose and reports how that attempt ended.
+  function connectFromOnclose(client: RedisClient): Promise<string> {
+    const { promise, resolve } = Promise.withResolvers<string>();
+    client.onclose = () => {
+      client.onclose = () => {};
+      resolve(
+        client.connect().then(
+          () => "connected",
+          (err: Error) => `rejected: ${err.message}`,
+        ),
+      );
+    };
+    return promise;
+  }
+
+  // A failure the client detects itself is a deliberate close, not a retry,
+  // even with auto reconnect on (left on here): only closes initiated by the
+  // peer go through the retry policy, as has always been the case and unlike
+  // ioredis. onclose only fires on that terminal path, so it firing is the
+  // assertion.
+  test.each([
+    ["redis", false],
+    ["rediss", true],
+  ])(
+    "a failure while connected over %s:// closes the socket, fires onclose, and connect() reconnects",
+    async (scheme, secure) => {
+      // 0x01 is not a RESP type byte, so the first connection fails after the
+      // handshake, on the same path as an idle timeout or any other protocol error.
+      const fake = helloServer({ PING: connection => (connection === 1 ? "\x01\r\n" : "+PONG\r\n") }, { secure });
+      const port = await fake.listen();
+      const closed = Promise.withResolvers<Error>();
+      const client = new RedisClient(`${scheme}://127.0.0.1:${port}`, secure ? { tls: { ca: tlsCert.cert } } : {});
+      try {
+        client.onclose = err => closed.resolve(err);
+        await client.connect();
+        expect(client.connected).toBe(true);
+        await expect(client.ping()).rejects.toMatchObject({ code: "ERR_REDIS_INVALID_RESPONSE_TYPE" });
+        // Already closed when the rejection is observed. Over TLS a graceful
+        // close would still be waiting for the peer's close_notify at this point.
+        expect(client.connected).toBe(false);
+        expect(await closed.promise).toBeInstanceOf(Error);
+        expect(fake.connections).toBe(1);
+        await client.connect();
+        expect(await client.ping()).toBe("PONG");
+        expect(fake.connections).toBe(2);
+      } finally {
+        client.close();
+        fake.server.close();
+      }
+    },
+  );
+
+  test.each([
+    ["redis", false],
+    ["rediss", true],
+  ])("an idle timeout over %s:// closes the connection and rejects what was in flight", async (scheme, secure) => {
+    const fake = helloServer({}, { secure });
+    const port = await fake.listen();
+    const closed = Promise.withResolvers<{ err: Error & { code: string }; connectedInsideOnclose: boolean }>();
+    // The timer armed by connect() carries connectionTimeout; the accepted
+    // HELLO re-arms it with idleTimeout and every chunk from the server re-arms
+    // it again. What is pinned here is what happens when it fires on an idle
+    // connection. 500ms leaves a debug build ample room to finish connecting.
+    const client = new RedisClient(`${scheme}://127.0.0.1:${port}`, {
+      idleTimeout: 50,
+      connectionTimeout: 500,
+      ...(secure ? { tls: { ca: tlsCert.cert } } : {}),
+    });
+    try {
+      client.onclose = err =>
+        closed.resolve({ err: err as Error & { code: string }, connectedInsideOnclose: client.connected });
+      await client.connect();
+      // GET is never answered by the stub, so it is still in flight when the timeout fires.
+      const inFlight = client.get("key").then(
+        () => "resolved",
+        (err: Error & { code: string }) => err.code,
+      );
+      const { err, connectedInsideOnclose } = await closed.promise;
+      expect({
+        onclose: err.code,
+        connectedInsideOnclose,
+        connectedAfter: client.connected,
+        inFlight: await inFlight,
+        connections: fake.connections,
+      }).toEqual({
+        onclose: "ERR_REDIS_CONNECTION_CLOSED",
+        connectedInsideOnclose: false,
+        connectedAfter: false,
+        inFlight: "ERR_REDIS_IDLE_TIMEOUT",
+        connections: 1,
+      });
+      await closedOnServer(fake.sockets[0]);
+      await client.connect();
+      expect(await client.ping()).toBe("PONG");
+      expect(fake.connections).toBe(2);
+    } finally {
+      client.close();
+      fake.server.close();
+    }
+  });
+
+  test("a connection that stays silent after the handshake is closed by its idle timeout", async () => {
+    const fake = helloServer();
+    const port = await fake.listen();
+    // The timer is armed with connectionTimeout when the socket is dialed, and
+    // accepting HELLO is what switches it to idleTimeout. With connectionTimeout
+    // off, the idle timeout is the only thing that can close this connection.
+    const client = new RedisClient(`redis://127.0.0.1:${port}`, {
+      connectionTimeout: 0,
+      idleTimeout: 50,
+      autoReconnect: false,
+    });
+    try {
+      // Once for a first connection, once for the one connect() dials after it.
+      for (const connection of [1, 2]) {
+        const closed = Promise.withResolvers<Error>();
+        client.onclose = err => closed.resolve(err);
+        await client.connect();
+        expect(client.connected).toBe(true);
+        expect(await closed.promise).toMatchObject({ code: "ERR_REDIS_CONNECTION_CLOSED" });
+        expect(client.connected).toBe(false);
+        expect(fake.connections).toBe(connection);
+      }
+    } finally {
+      client.close();
+      fake.server.close();
+    }
+  });
+
+  test("data from the server restarts the idle timeout", async () => {
+    let pushes = 0;
+    const fake = helloServer({
+      // PING is answered with 30 pushes 20ms apart, at least 600ms of traffic,
+      // and never with PONG. A client that is not subscribed discards them, so
+      // restarting its idle timer is all they can do.
+      PING: (_, socket) => {
+        const timer = setInterval(() => {
+          socket.write(">2\r\n$7\r\nmessage\r\n$2\r\nhi\r\n");
+          if (++pushes === 30) clearInterval(timer);
+        }, 20);
+        socket.on("close", () => clearInterval(timer));
+        return null;
+      },
+    });
+    const port = await fake.listen();
+    const client = new RedisClient(`redis://127.0.0.1:${port}`, { idleTimeout: 400, autoReconnect: false });
+    try {
+      await client.connect();
+      // Every push restarts the 400ms idle timer armed by the handshake, so it
+      // runs out, rejecting PING, 400ms after the last push, not in the middle
+      // of them.
+      await expect(client.ping()).rejects.toMatchObject({ code: "ERR_REDIS_IDLE_TIMEOUT" });
+      expect(pushes).toBe(30);
+      expect(client.connected).toBe(false);
+    } finally {
+      client.close();
+      fake.server.close();
+    }
+  });
+
+  test("connected reads false inside onclose when the server drops an established connection", async () => {
+    // Connection 1 is dropped by the server right after it answers PING.
+    const fake = helloServer({
+      PING: (connection, socket) => {
+        if (connection !== 1) return "+PONG\r\n";
+        socket.end("+PONG\r\n");
+        return null;
+      },
+    });
+    const port = await fake.listen();
+    const client = new RedisClient(`redis://127.0.0.1:${port}`, { autoReconnect: false });
+    try {
+      const closed = Promise.withResolvers<boolean>();
+      client.onclose = () => closed.resolve(client.connected);
+      await client.connect();
+      expect(await client.ping()).toBe("PONG");
+      expect(await closed.promise).toBe(false);
+      await client.connect();
+      expect(await client.ping()).toBe("PONG");
+      expect(fake.connections).toBe(2);
+    } finally {
+      client.close();
+      fake.server.close();
+    }
+  });
+
+  test("close() over rediss:// does not wait for the peer to answer close_notify", async () => {
+    // allowHalfOpen: the server never closes its side in response, which is
+    // what a graceful TLS close would be waiting for.
+    const fake = helloServer({}, { secure: true, allowHalfOpen: true });
+    const port = await fake.listen();
+    const client = new RedisClient(`rediss://127.0.0.1:${port}`, { tls: { ca: tlsCert.cert }, autoReconnect: false });
+    try {
+      let closes = 0;
+      client.onclose = () => closes++;
+      await client.connect();
+      const ended = once(fake.sockets[0], "end");
+      client.close();
+      expect({ connected: client.connected, closes }).toEqual({ connected: false, closes: 1 });
+      await ended;
+    } finally {
+      client.close();
+      fake.sockets[0]?.destroy();
+      fake.server.close();
+    }
+  });
+
+  test("a failure while the peer has stopped reading still closes the TLS socket at once", async () => {
+    // Pins the close code fail() uses: with a fast shutdown, usockets keeps a
+    // TLS socket whose last batch flush could not be handed to the kernel
+    // (packages/bun-usockets/src/crypto/openssl.c, us_internal_ssl_close) until
+    // the peer reads again, which this peer never does.
+    const fake = helloServer(
+      {
+        HELLO: (connection, socket) => {
+          if (connection === 1) socket.pause();
+          return "+OK\r\n";
+        },
+        PING: () => "+PONG\r\n",
+      },
+      { secure: true },
+    );
+    const port = await fake.listen();
+    const closed = Promise.withResolvers<void>();
+    const client = new RedisClient(`rediss://127.0.0.1:${port}`, { tls: { ca: tlsCert.cert }, autoReconnect: false });
+    try {
+      client.onclose = () => closed.resolve();
+      await client.connect();
+      const value = Buffer.alloc(256 * 1024, "x").toString();
+      const pending: Promise<unknown>[] = [];
+      // Each SET is flushed as soon as it is queued; once two flushes in a row
+      // hand nothing at all to the socket, the kernel buffers on both ends are
+      // full and the socket is stuck behind its undelivered ciphertext.
+      let stuckFlushes = 0;
+      while (stuckFlushes < 2 && pending.length < 256) {
+        const before = client.bufferedAmount;
+        pending.push(client.set("key", value).catch(() => {}));
+        await new Promise(resolve => setImmediate(resolve));
+        const added = client.bufferedAmount - before;
+        stuckFlushes = added >= value.length ? stuckFlushes + 1 : 0;
+      }
+      expect(stuckFlushes).toBe(2);
+      const lastSet = client.set("key", "last");
+      fake.sockets[0].write("\x01\r\n");
+      await expect(lastSet).rejects.toMatchObject({ code: "ERR_REDIS_INVALID_RESPONSE_TYPE" });
+      expect(client.connected).toBe(false);
+      await closed.promise;
+      await Promise.all(pending);
+      await client.connect();
+      expect(await client.ping()).toBe("PONG");
+      expect(fake.connections).toBe(2);
+    } finally {
+      client.close();
+      fake.sockets[0]?.destroy();
+      fake.server.close();
+    }
+  });
+
+  test("a connect() issued from onclose after a refused connection rejects 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 });
+    try {
+      const secondConnect = connectFromOnclose(client);
+      await expect(client.connect()).rejects.toMatchObject({ code: "ERR_REDIS_CONNECTION_CLOSED" });
+      expect(await secondConnect).toBe("rejected: Connection closed");
+    } finally {
+      client.close();
+    }
+  });
+
+  test("a connection timeout shorter than the retry delay does not stop the retries from settling connect()", async () => {
+    const fake = helloServer();
+    const port = await fake.listen();
+    await fake.close();
+    await using proc = Bun.spawn({
+      cmd: [
+        bunExe(),
+        "-e",
+        `
+        // The refusal comes back within a few milliseconds and the retry is
+        // scheduled 50ms after it, so a 30ms timeout armed for the first attempt
+        // would fire while no socket exists; the retry must still run and give up.
+        // The queued PING is rejected by whichever failure ends the client, so
+        // its message tells the two apart.
+        const client = new Bun.RedisClient("redis://127.0.0.1:${port}", { connectionTimeout: 30, maxRetries: 1 });
+        client.onclose = err => console.log("onclose", err.code);
+        const connected = client.connect().catch(err => console.log("connect rejected", err.code));
+        client.ping().catch(err => console.log("ping rejected:", err.message));
+        await connected;
+        `,
+      ],
+      env: bunEnv,
+      stdout: "pipe",
+      stderr: "pipe",
+    });
+    const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+    expect({ stdout, stderr, exitCode }).toEqual({
+      stdout: [
+        "onclose ERR_REDIS_CONNECTION_CLOSED",
+        "ping rejected: Max reconnection attempts reached",
+        "connect rejected ERR_REDIS_CONNECTION_CLOSED",
+        "",
+      ].join("\n"),
+      stderr: "",
+      exitCode: 0,
+    });
+  });
+
+  // The close for a dial that failed before a socket existed runs from the
+  // event loop; until then the client must look like it is dialling, so that
+  // whatever JS runs first neither dials on top of it nor is ignored.
+  test.skipIf(isWindows)(
+    "a close() issued right after a dial failed outright is honoured by the deferred close",
+    async () => {
+      using dir = tempDir("valkey-unix", {});
+      const socketPath = path.join(String(dir), "r.sock");
+      const first = helloServer();
+      const second = helloServer();
+      await first.listenUnix(socketPath);
+      const client = new RedisClient(`redis+unix://${socketPath}`);
+      try {
+        await client.connect();
+        client.close();
+        await first.close();
+        let closes = 0;
+        client.onclose = () => closes++;
+        const outcome = client.connect().then(
+          () => "connected",
+          (err: Error & { code: string }) => err.code,
+        );
+        client.close();
+        await second.listenUnix(socketPath);
+        expect({ outcome: await outcome, closes, connected: client.connected, redialled: second.connections }).toEqual({
+          outcome: "ERR_REDIS_CONNECTION_CLOSED",
+          closes: 1,
+          connected: false,
+          redialled: 0,
+        });
+      } finally {
+        client.close();
+        await first.close();
+        await second.close();
+      }
+    },
+  );
+
+  test.skipIf(isWindows)(
+    "a connect() that runs in the same tick as a retry that failed outright does not dial on top of it",
+    async () => {
+      using dir = tempDir("valkey-unix", {});
+      const socketPath = path.join(String(dir), "r.sock");
+      // Connection 1 is dropped by the server instead of answering PING.
+      const first = helloServer({
+        PING: (_connection, socket) => {
+          socket.end();
+          return null;
+        },
+      });
+      // The listener that comes back holds its HELLO reply, so a second dial on
+      // top of the first one would show up as a second connection here.
+      const second = helloServer({ HELLO: () => null });
+      await first.listenUnix(socketPath);
+      const client = new RedisClient(`redis+unix://${socketPath}`);
+      // Settled by the connect() made in the window; the handler is attached
+      // right away so that a failure elsewhere is not reported as its rejection.
+      const fromTimer = Promise.withResolvers<string>();
+      try {
+        await client.connect();
+        // Stops accepting at once while connection 1 stays up, so the retry
+        // scheduled when it drops fails outright.
+        void first.close();
+        const ping = await client.ping().then(
+          () => "answered",
+          (err: Error & { code: string }) => err.code,
+        );
+        expect(ping).toBe("ERR_REDIS_CONNECTION_CLOSED");
+        // The close that rejected the PING also armed the retry, 50ms out, and
+        // this continuation runs in its microtask checkpoint, before the loop
+        // turns again. Blocking past the retry and the timer armed here makes
+        // the loop fire both in one pass, in due order: the retry fails
+        // outright, then the callback runs in the window before the deferred
+        // close, brings the listener back and calls connect().
+        setTimeout(() => {
+          void second.listenUnix(socketPath);
+          fromTimer.resolve(
+            client.connect().then(
+              () => "connected",
+              (err: Error & { code: string }) => err.code,
+            ),
+          );
+        }, 150);
+        Bun.sleepSync(250);
+        // The deferred close schedules the next retry, which is what connects.
+        while (second.connections === 0) await Bun.sleep(1);
+        // Had the callback's connect() dialled on top, the deferred close would
+        // still have scheduled its retry, so a second connection would follow
+        // within one retry delay (at most 100ms at this point); this is the
+        // bound on asserting that it never comes.
+        await Bun.sleep(250);
+        expect(second.connections).toBe(1);
+        second.sockets[0].write("+OK\r\n");
+        expect({
+          fromTimer: await fromTimer.promise,
+          connected: client.connected,
+          connections: second.connections,
+        }).toEqual({ fromTimer: "connected", connected: true, connections: 1 });
+      } finally {
+        client.close();
+        await first.close();
+        await second.close();
+      }
+    },
+  );
+
+  test.skipIf(isWindows)("a reconnect whose dial fails outright is retried like a refused one", async () => {
+    using dir = tempDir("valkey-unix", {});
+    const socketPath = path.join(String(dir), "r.sock");
+    const first = helloServer();
+    const second = helloServer();
+    await first.listenUnix(socketPath);
+    const client = new RedisClient(`redis+unix://${socketPath}`);
+    try {
+      await client.connect();
+      client.close();
+      await first.close();
+      // connect(2) on a path nobody listens on fails before a socket exists.
+      const reconnected = client.connect();
+      await second.listenUnix(socketPath);
+      await reconnected;
+      expect(await client.ping()).toBe("PONG");
+      expect({ first: first.connections, second: second.connections }).toEqual({ first: 1, second: 1 });
+    } finally {
+      client.close();
+      await first.close();
+      await second.close();
+    }
+  });
+
+  test.skipIf(isWindows)(
+    "a reconnect whose dial fails outright rejects connect() when auto-reconnect is off",
+    async () => {
+      using dir = tempDir("valkey-unix", {});
+      const socketPath = path.join(String(dir), "r.sock");
+      const fake = helloServer();
+      await fake.listenUnix(socketPath);
+      const client = new RedisClient(`redis+unix://${socketPath}`, { autoReconnect: false });
+      try {
+        await client.connect();
+        client.close();
+        await fake.close();
+        let closes = 0;
+        client.onclose = () => closes++;
+        const attempt = client.connect();
+        // Reported from the event loop like a refused connection, not from
+        // inside connect(), so an onclose that calls connect() cannot recurse.
+        expect(closes).toBe(0);
+        const outcome = await attempt.then(
+          () => "connected",
+          (err: Error & { code: string }) => `rejected: ${err.code}`,
+        );
+        expect({ outcome, closes }).toEqual({ outcome: "rejected: ERR_REDIS_CONNECTION_CLOSED", closes: 1 });
+        await expect(client.ping()).rejects.toMatchObject({ code: "ERR_REDIS_CONNECTION_CLOSED" });
+      } finally {
+        client.close();
+        await fake.close();
+      }
+    },
+  );
+
+  // A fresh client's first dial can fail the same way, whether connect() or the
+  // first command makes it; it goes through the same close as a refused dial.
+  const firstDialFrom: [string, (client: RedisClient) => Promise<unknown>][] = [
+    ["connect()", client => client.connect()],
+    ["a command", client => client.ping()],
+  ];
+
+  test.skipIf(isWindows).each(firstDialFrom)(
+    "a first dial that fails outright, made by %s, rejects and runs onclose once when auto-reconnect is off",
+    async (_entry, dial) => {
+      using dir = tempDir("valkey-unix", {});
+      const socketPath = path.join(String(dir), "r.sock");
+      const fake = helloServer();
+      const client = new RedisClient(`redis+unix://${socketPath}`, { autoReconnect: false });
+      try {
+        let closes = 0;
+        client.onclose = () => closes++;
+        const attempt = dial(client).then(
+          () => "resolved",
+          (err: Error & { code: string }) => `rejected: ${err.code}`,
+        );
+        expect(closes).toBe(0);
+        expect({ outcome: await attempt, closes, connected: client.connected }).toEqual({
+          outcome: "rejected: ERR_REDIS_CONNECTION_CLOSED",
+          closes: 1,
+          connected: false,
+        });
+        // The failed attempt is settled and forgotten: a connect() once a
+        // listener is there dials rather than handing the same promise back.
+        await fake.listenUnix(socketPath);
+        await client.connect();
+        expect({ ping: await client.ping(), connections: fake.connections }).toEqual({ ping: "PONG", connections: 1 });
+      } finally {
+        client.close();
+        await fake.close();
+      }
+    },
+  );
+
+  test.skipIf(isWindows).each(firstDialFrom)(
+    "a first dial that fails outright, made by %s, is retried until a listener is there",
+    async (_entry, dial) => {
+      using dir = tempDir("valkey-unix", {});
+      const socketPath = path.join(String(dir), "r.sock");
+      const fake = helloServer();
+      const client = new RedisClient(`redis+unix://${socketPath}`);
+      try {
+        let closes = 0;
+        client.onclose = () => closes++;
+        const attempt = dial(client);
+        // Listening well before the first retry is due; a later retry would
+        // connect just the same, the retries are not terminal either way.
+        await fake.listenUnix(socketPath);
+        await attempt;
+        expect({ ping: await client.ping(), closes, connections: fake.connections }).toEqual({
+          ping: "PONG",
+          closes: 0,
+          connections: 1,
+        });
+      } finally {
+        client.close();
+        await fake.close();
+      }
+    },
+  );
+
+  test("a connect() issued from onclose after the TLS context cannot be built dials again from the event loop", async () => {
+    await using proc = Bun.spawn({
+      cmd: [
+        bunExe(),
+        "-e",
+        `
+        // Neither key nor cert parses, so no attempt ever gets as far as a socket.
+        const client = new Bun.RedisClient("rediss://127.0.0.1:1", {
+          tls: { key: "not a key", cert: "not a cert" },
+          autoReconnect: false,
+        });
+        const attempts = [];
+        const done = Promise.withResolvers();
+        let closes = 0, depth = 0, nested = false;
+        client.onclose = () => {
+          closes += 1;
+          nested ||= depth > 0;
+          depth += 1;
+          if (closes < 3) attempts.push(client.connect());
+          else done.resolve();
+          depth -= 1;
+        };
+        attempts.push(client.connect());
+        const closesInsideConnect = closes;
+        await done.promise;
+        const outcomes = await Promise.all(attempts.map(p => p.then(() => "connected", err => err.code)));
+        console.log(JSON.stringify({ closesInsideConnect, nested, closes, outcomes, connected: client.connected }));
+        `,
+      ],
+      env: bunEnv,
+      stdout: "pipe",
+      stderr: "pipe",
+    });
+    const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+    expect({ stdout: stdout.trim(), stderr, exitCode }).toEqual({
+      stdout: JSON.stringify({
+        closesInsideConnect: 0,
+        nested: false,
+        closes: 3,
+        outcomes: ["ERR_REDIS_CONNECTION_CLOSED", "ERR_REDIS_CONNECTION_CLOSED", "ERR_REDIS_CONNECTION_CLOSED"],
+        connected: false,
+      }),
+      stderr: "",
+      exitCode: 0,
+    });
+  });
+
+  test("a connect() issued from onclose is not fed the replies left over from the failed connection", async () => {
+    // With a database in the URL, HELLO and SELECT are written together, so a
+    // server that rejects HELLO delivers both error replies in one read.
+    const fake = helloServer({
+      HELLO: connection =>
+        connection === 1 ? "-WRONGPASS invalid password\r\n-NOAUTH Authentication required.\r\n" : "+OK\r\n+OK\r\n",
+    });
+    const port = await fake.listen();
+    const client = new RedisClient(`redis://127.0.0.1:${port}/1`, { autoReconnect: false });
+    try {
+      const secondConnect = connectFromOnclose(client);
+      await expect(client.connect()).rejects.toMatchObject({ code: "ERR_REDIS_CONNECTION_CLOSED" });
+      expect(await secondConnect).toBe("connected");
+      expect(await client.ping()).toBe("PONG");
+      expect(fake.connections).toBe(2);
+    } finally {
+      client.close();
+      fake.server.close();
+    }
+  });
+
+  test("a rejected SELECT after an accepted HELLO fails the connection once and connect() from onclose dials again", async () => {
+    // HELLO and SELECT are written together, and both replies come back in one
+    // read: connection 1 accepts HELLO and rejects SELECT, connection 2 accepts both.
+    const fake = helloServer({
+      HELLO: connection => (connection === 1 ? "+OK\r\n-ERR DB index is out of range\r\n" : "+OK\r\n+OK\r\n"),
+    });
+    const port = await fake.listen();
+    const client = new RedisClient(`redis://127.0.0.1:${port}/1`, { autoReconnect: true });
+    try {
+      const closes: { message: string; connected: boolean; connections: number }[] = [];
+      let secondConnect: Promise<string> | undefined;
+      client.onclose = err => {
+        closes.push({ message: err.message, connected: client.connected, connections: fake.connections });
+        secondConnect ??= client.connect().then(
+          () => "connected",
+          (err: Error) => `rejected: ${err.message}`,
+        );
+      };
+      // Queued behind the handshake, so it is still pending when SELECT is rejected.
+      const queued = client.get("key").then(
+        () => "resolved",
+        (err: Error & { code: string }) => `${err.code}: ${err.message}`,
+      );
+      // connect() settles on the accepted HELLO, before the SELECT reply is
+      // read, so it resolves; the failure that follows is reported by onclose.
+      await client.connect();
+      expect(await queued).toBe("ERR_REDIS_INVALID_COMMAND: ERR DB index is out of range");
+      // The rejection is a failure the client detected, so there is no retry
+      // even with autoReconnect on: onclose fires once and the only second
+      // connection is the one dialed from it.
+      expect(closes).toEqual([{ message: "Connection closed", connected: false, connections: 1 }]);
+      expect(await secondConnect).toBe("connected");
+      expect(await client.ping()).toBe("PONG");
+      expect(fake.connections).toBe(2);
+    } finally {
+      client.close();
+      fake.server.close();
+    }
+  });
+
+  test("a duplicate of a failed client still auto-reconnects", async () => {
+    // Connection 1 (the original) fails on a protocol error, connection 2 (the
+    // duplicate) is dropped by the server right after it answers PING.
+    const fake = helloServer({
+      PING: (connection, socket) => {
+        if (connection === 1) return "\x01\r\n";
+        if (connection === 2) {
+          socket.end("+PONG\r\n");
+          return null;
+        }
+        return "+PONG\r\n";
+      },
+    });
+    const port = await fake.listen();
+    const client = new RedisClient(`redis://127.0.0.1:${port}`, { autoReconnect: true });
+    let duplicate: RedisClient | undefined;
+    try {
+      await client.connect();
+      await expect(client.ping()).rejects.toMatchObject({ code: "ERR_REDIS_INVALID_RESPONSE_TYPE" });
+      expect(client.connected).toBe(false);
+      duplicate = await client.duplicate();
+      // A duplicate copies the original's manual-close state; a failure is not
+      // one, so the drop of connection 2 goes through the retry policy: no
+      // onclose, a second onconnect, and the queued PING answered by connection 3.
+      const reconnected = Promise.withResolvers<void>();
+      let connects = 0;
+      duplicate.onconnect = () => {
+        if (++connects === 2) reconnected.resolve();
+      };
+      duplicate.onclose = err => reconnected.reject(err);
+      expect(await duplicate.ping()).toBe("PONG");
+      await reconnected.promise;
+      expect(await duplicate.ping()).toBe("PONG");
+      expect(fake.connections).toBe(3);
+    } finally {
+      duplicate?.close();
+      client.close();
+      fake.server.close();
+    }
+  });
+
+  test("a message listener that closes and reconnects is not fed the pushes buffered behind its message", async () => {
+    // RESP3 push frames as the server writes them for SUBSCRIBE and for messages.
+    const push = (...items: (string | number)[]) =>
+      `>${items.length}\r\n` +
+      items.map(item => (typeof item === "number" ? `:${item}\r\n` : `$${item.length}\r\n${item}\r\n`)).join("");
+    const fake = helloServer();
+    const port = await fake.listen();
+    const client = new RedisClient(`redis://127.0.0.1:${port}`);
+    try {
+      await client.connect();
+      const connection1 = fake.sockets[0];
+      connection1.on("data", chunk => {
+        if (chunk.toString("latin1").includes("SUBSCRIBE")) connection1.write(push("subscribe", "ch", 1));
+      });
+      const delivered: string[] = [];
+      const firstDelivered = Promise.withResolvers<void>();
+      const reconnect = Promise.withResolvers<string>();
+      await client.subscribe("ch", message => {
+        delivered.push(message);
+        if (delivered.length === 1) firstDelivered.resolve();
+        if (delivered.length !== 2) return;
+        client.close();
+        reconnect.resolve(
+          client.connect().then(
+            () => "connected",
+            (err: Error) => `rejected: ${err.message}`,
+          ),
+        );
+      });
+      const [m1, m2] = [push("message", "ch", "m1"), push("message", "ch", "m2")];
+      // m0 is handled straight off this read and the start of m1 is kept in the
+      // read buffer, so everything from here on goes through the buffer path.
+      connection1.write(push("message", "ch", "m0") + m1.slice(0, 10));
+      await firstDelivered.promise;
+      // One read completes m1 and carries m2. The listener closes connection 1
+      // on m1 and dials connection 2; m2 belongs to neither (taken as the next
+      // reply, it would be read as connection 2's HELLO answer and fail it).
+      connection1.write(m1.slice(10) + m2);
+      expect({
+        reconnect: await reconnect.promise,
+        delivered,
+        connected: client.connected,
+        connections: fake.connections,
+      }).toEqual({ reconnect: "connected", delivered: ["m0", "m1"], connected: true, connections: 2 });
+    } finally {
+      client.close();
+      fake.server.close();
+    }
+  });
+
+  test("a connect() issued from onclose after a failed TLS handshake gets to dial again", async () => {
+    let handshakes = 0;
+    const server = net.createServer(socket => {
+      // The first bytes are the ClientHello; dropping the connection there
+      // fails the client's handshake.
+      socket.once("data", () => {
+        handshakes += 1;
+        socket.destroy();
+      });
+      socket.on("error", () => {});
+    });
+    await new Promise<void>(resolve => server.listen(0, "127.0.0.1", resolve));
+    const { port } = server.address() as net.AddressInfo;
+    const client = new RedisClient(`rediss://127.0.0.1:${port}`, { autoReconnect: false });
+    try {
+      const secondConnect = connectFromOnclose(client);
+      await expect(client.connect()).rejects.toMatchObject({ code: "ERR_REDIS_CONNECTION_CLOSED" });
+      expect({ secondConnect: await secondConnect, handshakes }).toEqual({
+        secondConnect: "rejected: Connection closed",
+        handshakes: 2,
+      });
+    } finally {
+      client.close();
+      server.close();
+    }
+  });
+
+  test("close() discards a half-received reply instead of letting it keep the process alive", async () => {
+    // Announces a 4 byte bulk string and stops halfway through it.
+    const fake = helloServer({ PING: () => "$4\r\nPO" });
+    const port = await fake.listen();
+    try {
+      await using proc = Bun.spawn({
+        cmd: [
+          bunExe(),
+          "-e",
+          `
+          const client = new Bun.RedisClient("redis://127.0.0.1:${port}", { autoReconnect: false });
+          await client.connect();
+          const ping = client.ping().catch(err => console.log("ping rejected", err.code));
+          while (client.bufferedAmount === 0) await Bun.sleep(1);
+          console.log("buffered before close", client.bufferedAmount);
+          client.close();
+          await ping;
+          console.log("buffered after close", client.bufferedAmount);
+          `,
+        ],
+        env: bunEnv,
+        stdout: "pipe",
+        stderr: "pipe",
+      });
+      const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+      expect({ stdout, stderr, exitCode }).toEqual({
+        stdout: "buffered before close 6\nping rejected ERR_REDIS_CONNECTION_CLOSED\nbuffered after close 0\n",
+        stderr: "",
+        exitCode: 0,
+      });
+    } finally {
+      fake.server.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, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+    expect({ stdout, stderr, exitCode }).toEqual({
+      stdout: "onclose ERR_REDIS_CONNECTION_CLOSED\nconnect rejected ERR_REDIS_CONNECTION_CLOSED\n",
+      stderr: "",
+      exitCode: 0,
+    });
+  });
+});
diff --git a/test/js/web/workers/worker-terminate-lifetime.test.ts b/test/js/web/workers/worker-terminate-lifetime.test.ts
index b66c1228019e..1f3cd6f81934 100644
--- a/test/js/web/workers/worker-terminate-lifetime.test.ts
+++ b/test/js/web/workers/worker-terminate-lifetime.test.ts
@@ -1,5 +1,5 @@
 import { describe, expect, test } from "bun:test";
-import { bunEnv, bunExe, isASAN, isDebug, tempDir, tls } from "harness";
+import { bunEnv, bunExe, isASAN, isDebug, isWindows, tempDir, tls } from "harness";
 import { join } from "path";
 
 // Worker VM startup/teardown is much slower under debug and/or ASAN; these
@@ -725,6 +725,48 @@ test(
   timeout,
 );
 
+// Spawns workers that each start `connectExpr` in one immediate and call
+// process.exit(0) in the next, so whatever that connect attempt left behind is
+// still pending when the worker's VM tears down.
+async function exitRightAfterConnecting(connectExpr: string) {
+  const workers = slow ? 8 : 24;
+  await using proc = Bun.spawn({
+    cmd: [
+      bunExe(),
+      "-e",
+      `
+      const { Worker } = require("node:worker_threads");
+      const src =
+        "const { parentPort } = require('node:worker_threads');" +
+        "Bun.file(process.execPath).slice(0, 100).json().catch(() => {});" +
+        ${JSON.stringify(`setImmediate(() => ${connectExpr}.catch(() => {}));`)} +
+        "parentPort.postMessage('up');" +
+        "setImmediate(() => process.exit(0));";
+      let started = 0, exited = 0;
+      function again() {
+        if (started >= ${workers}) {
+          if (exited === ${workers}) console.log("PASS");
+          return;
+        }
+        started++;
+        const w = new Worker(src, { eval: true });
+        w.on("error", (e) => { console.error(e); process.exit(1); });
+        w.on("exit", () => { exited++; again(); });
+      }
+      again(); again();
+    `,
+    ],
+    env: { ...bunEnv, UV_THREADPOOL_SIZE: "4" },
+    stdout: "pipe",
+    stderr: "pipe",
+  });
+
+  const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+  expect(stderr).toBe("");
+  expect(stdout).toBe("PASS\n");
+  expect(exitCode).toBe(0);
+}
+
 // For a debug build: host code that runs after the worker's own process.exit()
 // unwound script — here a redis connect started in the same immediate tick as
 // the exit, whose ECONNREFUSED then lands in that loop tick — builds JS error
@@ -735,43 +777,62 @@ test(
 // for as long as it keeps the exception.
 test.skipIf(!isDebug)(
   "process.exit() with native error completions landing in the same tick does not trip DeferTermination",
+  () =>
+    exitRightAfterConnecting(
+      "new Bun.RedisClient('redis://127.0.0.1:9', { connectionTimeout: 100, autoReconnect: false }).connect()",
+    ),
+  timeout,
+);
+
+// A TLS context that cannot be built fails the dial before there is a socket;
+// the redis client then settles that from a task it queues on the event loop,
+// holding a ref to itself and the loop. The exit in the next immediate tears
+// the VM down with that task still queued, so it has to be released without
+// running (a debug build asserts on the refcount if either ref is mishandled;
+// the ASAN build reports the leak).
+test.skipIf(!isDebug && !isASAN)(
+  "process.exit() with a redis client's deferred close still queued releases it cleanly",
+  () =>
+    exitRightAfterConnecting(
+      "new Bun.RedisClient('rediss://127.0.0.1:9', { tls: { key: 'x', cert: 'x' }, autoReconnect: false }).connect()",
+    ),
+  timeout,
+);
+
+// The same task, queued by a first command whose dial failed outright (a unix
+// socket path nobody listens on), with the command itself still queued behind it.
+test.skipIf((!isDebug && !isASAN) || isWindows)(
+  "process.exit() with a redis client's deferred close and a queued command releases both cleanly",
+  () => exitRightAfterConnecting("new Bun.RedisClient('redis+unix:///nonexistent/redis.sock').ping()"),
+  timeout,
+);
+
+// The same deferred close on the main thread, left to run: it settles the
+// connect and drops its refs, so nothing keeps the event loop alive and the
+// process exits on its own.
+test(
+  "a redis client whose TLS context cannot be built does not keep the process alive",
   async () => {
-    const workers = slow ? 8 : 24;
     await using proc = Bun.spawn({
       cmd: [
         bunExe(),
         "-e",
         `
-        const { Worker } = require("node:worker_threads");
-        const src =
-          "const { parentPort } = require('node:worker_threads');" +
-          "Bun.file(process.execPath).slice(0, 100).json().catch(() => {});" +
-          "setImmediate(() => new Bun.RedisClient('redis://127.0.0.1:9', { connectionTimeout: 100, autoReconnect: false }).connect().catch(() => {}));" +
-          "parentPort.postMessage('up');" +
-          "setImmediate(() => process.exit(0));";
-        let started = 0, exited = 0;
-        function again() {
-          if (started >= ${workers}) {
-            if (exited === ${workers}) console.log("PASS");
-            return;
-          }
-          started++;
-          const w = new Worker(src, { eval: true });
-          w.on("error", (e) => { console.error(e); process.exit(1); });
-          w.on("exit", () => { exited++; again(); });
-        }
-        again(); again();
-      `,
+        new Bun.RedisClient("rediss://127.0.0.1:1", { tls: { key: "x", cert: "x" }, autoReconnect: false })
+          .connect()
+          .catch(err => console.log("connect rejected", err.code));
+        `,
       ],
-      env: { ...bunEnv, UV_THREADPOOL_SIZE: "4" },
+      env: bunEnv,
       stdout: "pipe",
       stderr: "pipe",
     });
-
     const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
-    expect(stderr).toBe("");
-    expect(stdout).toBe("PASS\n");
-    expect(exitCode).toBe(0);
+    expect({ stdout, stderr, exitCode }).toEqual({
+      stdout: "connect rejected ERR_REDIS_CONNECTION_CLOSED\n",
+      stderr: "",
+      exitCode: 0,
+    });
   },
   timeout,
 );