From 4d21306a1177041dd7d9b4628f17c44812e1e291 Mon Sep 17 00:00:00 2001 From: Alistair Smith Date: Wed, 12 Aug 2026 19:05:33 -0700 Subject: [PATCH 01/21] valkey: close the socket on every fail() and mark the client disconnected before onclose runs --- src/runtime/valkey_jsc/js_valkey.rs | 14 ++- src/runtime/valkey_jsc/valkey.rs | 17 ++-- .../reliability/connection-failures.test.ts | 89 +++++++++++++++++++ 3 files changed, 103 insertions(+), 17 deletions(-) diff --git a/src/runtime/valkey_jsc/js_valkey.rs b/src/runtime/valkey_jsc/js_valkey.rs index cadf98d9cb57..508286af04c0 100644 --- a/src/runtime/valkey_jsc/js_valkey.rs +++ b/src/runtime/valkey_jsc/js_valkey.rs @@ -1937,10 +1937,10 @@ impl SocketHandler { 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 @@ impl SocketHandler { // 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()) } diff --git a/src/runtime/valkey_jsc/valkey.rs b/src/runtime/valkey_jsc/valkey.rs index 44542b9a6adf..b4232a61d3ba 100644 --- a/src/runtime/valkey_jsc/valkey.rs +++ b/src/runtime/valkey_jsc/valkey.rs @@ -33,12 +33,12 @@ 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()`; 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,6 +607,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, @@ -614,10 +615,8 @@ impl ValkeyClient { jsvalue, ); - if !self.connection_ready() { - self.flags.is_manually_closed = true; - self.close(); - } + self.flags.is_manually_closed = true; + self.close(); val } diff --git a/test/js/valkey/reliability/connection-failures.test.ts b/test/js/valkey/reliability/connection-failures.test.ts index bcf8964ac348..79e47b236bf8 100644 --- a/test/js/valkey/reliability/connection-failures.test.ts +++ b/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(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(); + 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); + }); +}); From 21be6f475dd8a99aacddad240bedc77bbe6e6e18 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Thu, 13 Aug 2026 07:41:31 +0000 Subject: [PATCH 02/21] valkey: fix fail_handshake/on_data clobbering a connect() issued from onclose fail_handshake closed client.socket a second time after fail_with_js_value() returned. fail_with_js_value() already closes the socket, so the only socket left for that close was the one a connect() issued from onclose had just opened. It also set is_manually_closed, which fail_with_js_value() sets too. on_data kept handling the remaining replies of a read after the reply it had just handled failed the connection, as long as status and failed looked healthy, which they do again once connect() has been called from onclose. The leftovers were then handled as the new connection's HELLO reply. Stop as soon as client.socket is no longer the socket the data came from. on_close() now frees read_buffer and resets reply_scanner like it already did for write_buffer. A reply that was only partially received when the connection went away counted as pending activity in update_poll_ref(), so close() after such a reply kept the process alive. Tests: cover the three cases above, trigger the post-handshake fail() with a protocol error instead of a 100ms connection timeout (debug builds need longer than that to connect), issue the second connect() from onclose so the test hangs without the fix regardless of microtask timing, await the rejects matchers, and assert on stderr of the spawned processes. --- src/runtime/valkey_jsc/js_valkey.rs | 3 - src/runtime/valkey_jsc/valkey.rs | 13 +- .../reliability/connection-failures.test.ts | 154 +++++++++++++++--- 3 files changed, 138 insertions(+), 32 deletions(-) diff --git a/src/runtime/valkey_jsc/js_valkey.rs b/src/runtime/valkey_jsc/js_valkey.rs index 508286af04c0..563b2914e25e 100644 --- a/src/runtime/valkey_jsc/js_valkey.rs +++ b/src/runtime/valkey_jsc/js_valkey.rs @@ -1909,9 +1909,6 @@ impl SocketHandler { err_value: JSValue, ) -> JsTerminatedResult<()> { let _exit = this.vm().enter_event_loop_scope(); - this.client_mut().flags.is_manually_closed = true; - let this_br = BackRef::new(this); - let _close = scopeguard::guard(this_br, |p| p.client_mut().close()); narrow_terminated( this.client_mut() .fail_with_js_value(&this.global_object, err_value), diff --git a/src/runtime/valkey_jsc/valkey.rs b/src/runtime/valkey_jsc/valkey.rs index b4232a61d3ba..f90bb61a3a65 100644 --- a/src/runtime/valkey_jsc/valkey.rs +++ b/src/runtime/valkey_jsc/valkey.rs @@ -651,6 +651,10 @@ impl ValkeyClient { pub fn on_close(&mut self) -> JsTerminated<()> { 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 { @@ -742,6 +746,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 @@ -809,7 +817,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(); @@ -868,8 +876,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(()); } diff --git a/test/js/valkey/reliability/connection-failures.test.ts b/test/js/valkey/reliability/connection-failures.test.ts index 79e47b236bf8..5ad145f432de 100644 --- a/test/js/valkey/reliability/connection-failures.test.ts +++ b/test/js/valkey/reliability/connection-failures.test.ts @@ -444,16 +444,19 @@ describe("Valkey: Auto-Reconnect In-Flight Commands", () => { }); describe("Valkey: Recovering After fail()", () => { - function helloServer(onConnection?: (socket: net.Socket, connection: number) => void) { + // Answers the chunk carrying HELLO with `+OK` and the one carrying PING with + // `+PONG` unless `replies` says otherwise for that connection. + function helloServer(replies: Partial string>> = {}) { let connections = 0; const server = net.createServer(socket => { connections += 1; - onConnection?.(socket, connections); + const connection = 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"); + if (text.includes("HELLO")) socket.write(replies.HELLO?.(connection) ?? "+OK\r\n"); + if (text.includes("PING")) socket.write(replies.PING?.(connection) ?? "+PONG\r\n"); }); + socket.on("error", () => {}); }); return { server, @@ -467,44 +470,140 @@ describe("Valkey: Recovering After fail()", () => { }; } - test("an idle timeout while connected closes the socket, fires onclose, and connect() reconnects", async () => { - const closed = Promise.withResolvers(); - const fake = helloServer(); + // Calls connect() from the first onclose and reports how that attempt ended. + function connectFromOnclose(client: RedisClient): Promise { + const { promise, resolve } = Promise.withResolvers(); + client.onclose = () => { + client.onclose = () => {}; + resolve( + client.connect().then( + () => "connected", + (err: Error) => `rejected: ${err.message}`, + ), + ); + }; + return promise; + } + + test("a failure while connected closes the socket, fires onclose, and connect() reconnects", async () => { + // 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") }); const port = await fake.listen(); + const closed = Promise.withResolvers(); + const client = new RedisClient(`redis://127.0.0.1:${port}`, { autoReconnect: false }); 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); + await expect(client.ping()).rejects.toMatchObject({ code: "ERR_REDIS_INVALID_RESPONSE_TYPE" }); + expect(await closed.promise).toBeInstanceOf(Error); expect(client.connected).toBe(false); await client.connect(); expect(await client.ping()).toBe("PONG"); expect(fake.connections).toBe(2); - client.close(); } finally { + client.close(); fake.server.close(); } }); - test("connect() rejects again after a failed attempt instead of hanging", async () => { + 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 }); - expect(client.connect()).rejects.toThrow(); - await client.connect().catch(() => {}); - expect(client.connect()).rejects.toThrow(); - await client.connect().catch(() => {}); - client.close(); + try { + const secondConnect = connectFromOnclose(client); + await expect(client.connect()).rejects.toThrow(); + expect(await secondConnect).toBe("rejected: Connection closed"); + } finally { + client.close(); + } + }); + + 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.toThrow(); + expect(await secondConnect).toBe("connected"); + expect(await client.ping()).toBe("PONG"); + expect(fake.connections).toBe(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(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.toThrow(); + 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 () => { @@ -525,8 +624,11 @@ describe("Valkey: Recovering After fail()", () => { stdout: "pipe", stderr: "pipe", }); - const [stdout, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]); - expect(stdout).toContain("connect rejected"); - expect(exitCode).toBe(0); + 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, + }); }); }); From 2689715e26cf34d3cfc618b2367e0dd7bfa064a4 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Thu, 13 Aug 2026 09:23:37 +0000 Subject: [PATCH 03/21] test(valkey): assert the error code of the first connect() rejection --- test/js/valkey/reliability/connection-failures.test.ts | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/test/js/valkey/reliability/connection-failures.test.ts b/test/js/valkey/reliability/connection-failures.test.ts index 5ad145f432de..cb3516af1ecd 100644 --- a/test/js/valkey/reliability/connection-failures.test.ts +++ b/test/js/valkey/reliability/connection-failures.test.ts @@ -516,7 +516,7 @@ describe("Valkey: Recovering After fail()", () => { const client = new RedisClient(`redis://127.0.0.1:${port}`, { autoReconnect: false }); try { const secondConnect = connectFromOnclose(client); - await expect(client.connect()).rejects.toThrow(); + await expect(client.connect()).rejects.toMatchObject({ code: "ERR_REDIS_CONNECTION_CLOSED" }); expect(await secondConnect).toBe("rejected: Connection closed"); } finally { client.close(); @@ -534,7 +534,7 @@ describe("Valkey: Recovering After fail()", () => { const client = new RedisClient(`redis://127.0.0.1:${port}/1`, { autoReconnect: false }); try { const secondConnect = connectFromOnclose(client); - await expect(client.connect()).rejects.toThrow(); + 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); @@ -560,7 +560,7 @@ describe("Valkey: Recovering After fail()", () => { const client = new RedisClient(`rediss://127.0.0.1:${port}`, { autoReconnect: false }); try { const secondConnect = connectFromOnclose(client); - await expect(client.connect()).rejects.toThrow(); + await expect(client.connect()).rejects.toMatchObject({ code: "ERR_REDIS_CONNECTION_CLOSED" }); expect({ secondConnect: await secondConnect, handshakes }).toEqual({ secondConnect: "rejected: Connection closed", handshakes: 2, From 682b042bc6cfa1b794bb93169d96fa36bcaeaf4d Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Thu, 13 Aug 2026 19:20:48 +0000 Subject: [PATCH 04/21] valkey: settle a reconnect whose dial fails outright, disarm the connect timer of a finished attempt reconnect() answered a connect() that failed before a socket existed (unix socket path gone, out of fds) by calling the user's onclose and nothing else: the connect() promise stayed pending, queued commands were never rejected and is_reconnecting stayed set. It now runs ValkeyClient::on_close() like a dial that fails asynchronously does, taking the ref on_close() releases in place of the socket's, so it retries with backoff or fails and settles the promise. JSValkeyClient::fail_with_js_value loses its only caller. The connection timer armed for an attempt stayed armed after that attempt failed. With connectionTimeout shorter than the retry delay it fired during the delay, and since fail() now clears is_reconnecting, the scheduled retry then did nothing and connect() never settled while the poll ref kept the process alive. on_valkey_reconnect() disarms it; reconnect() arms a new one for the next attempt. --- src/runtime/valkey_jsc/js_valkey.rs | 57 +++++++------ .../reliability/connection-failures.test.ts | 81 ++++++++++++++++++- 2 files changed, 111 insertions(+), 27 deletions(-) diff --git a/src/runtime/valkey_jsc/js_valkey.rs b/src/runtime/valkey_jsc/js_valkey.rs index 563b2914e25e..ded12e70032a 100644 --- a/src/runtime/valkey_jsc/js_valkey.rs +++ b/src/runtime/valkey_jsc/js_valkey.rs @@ -1188,6 +1188,20 @@ impl JSValkeyClient { } } + /// `ValkeyClient::on_close()` is otherwise only reached from a socket's + /// close or connect-error callback. Run it for a dial that failed before + /// there was a socket, so the connect() promise, `onclose`, the retry + /// policy and the poll ref are handled the same way. `on_close()` releases + /// the ref a socket would have held, so take one for it to release. + fn on_close_without_socket(&self) -> JsTerminatedResult<()> { + let _guard = self.ref_scope(); + self.ref_(); + self.client_mut().status = valkey::Status::Disconnected; + let result = narrow_terminated(self.client_mut().on_close()); + self.update_poll_ref(); + result + } + pub(crate) fn on_reconnect_timer(&self) { debug!("Reconnect timer fired, attempting to reconnect"); @@ -1221,15 +1235,13 @@ impl JSValkeyClient { }); if let Err(err) = self.connect() { - 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() ); - self.poll_ref.with_mut(|r| r.disable()); + // Same outcome as a dial that fails asynchronously: another retry, + // or fail() and a settled connect() promise once retries are used up. + let _ = self.on_close_without_socket(); return; } @@ -1375,11 +1387,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 `on_close_without_socket()` 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()); } @@ -1388,8 +1405,9 @@ impl JSValkeyClient { pub(crate) fn on_valkey_close(&self) -> JsTerminatedResult<()> { 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 + // `on_close_without_socket()` 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()); @@ -1436,19 +1454,6 @@ impl JSValkeyClient { narrow_terminated(self.client_mut().fail(message, err)) } - pub(crate) fn fail_with_js_value(&self, value: JSValue) { - let Some(this_value) = self.this_value.get().try_get() else { - return; - }; - let global_object = self.global_object; - if let Some(on_close) = Js::onclose_get_cached(this_value) { - let _exit = self.vm().enter_event_loop_scope(); - if let Err(e) = on_close.call(&global_object, this_value, &[value]) { - global_object.report_active_exception_as_unhandled(e); - } - } - } - fn close_socket_next_tick(&self) { if self.client.get().socket.is_closed() { return; diff --git a/test/js/valkey/reliability/connection-failures.test.ts b/test/js/valkey/reliability/connection-failures.test.ts index cb3516af1ecd..1a5287d1eb26 100644 --- a/test/js/valkey/reliability/connection-failures.test.ts +++ b/test/js/valkey/reliability/connection-failures.test.ts @@ -1,7 +1,8 @@ import { RedisClient } from "bun"; import { describe, expect, mock, test } from "bun:test"; -import { bunEnv, bunExe } from "harness"; +import { bunEnv, bunExe, isWindows, tempDir } from "harness"; import net from "net"; +import path from "path"; import { DEFAULT_REDIS_OPTIONS, DEFAULT_REDIS_URL, delay, isEnabled } from "../test-utils"; /** @@ -467,6 +468,8 @@ describe("Valkey: Recovering After fail()", () => { new Promise(resolve => server.listen(0, "127.0.0.1", () => resolve((server.address() as net.AddressInfo).port)), ), + listenUnix: (socketPath: string) => new Promise(resolve => server.listen(socketPath, resolve)), + close: () => new Promise(resolve => server.close(resolve)), }; } @@ -523,6 +526,82 @@ describe("Valkey: Recovering After fail()", () => { } }); + 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. + const client = new Bun.RedisClient("redis://127.0.0.1:${port}", { connectionTimeout: 30, 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, + }); + }); + + 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 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(); + const outcome = await client.connect().then( + () => "connected", + (err: Error & { code: string }) => `rejected: ${err.code}`, + ); + expect(outcome).toBe("rejected: ERR_REDIS_CONNECTION_CLOSED"); + await expect(client.ping()).rejects.toMatchObject({ code: "ERR_REDIS_CONNECTION_CLOSED" }); + } finally { + client.close(); + } + }, + ); + 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. From 134b3dd4c44baf3a1351fd78ea4f051feb1fb11f Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Thu, 13 Aug 2026 20:01:56 +0000 Subject: [PATCH 05/21] valkey: report a dial that fails outright from the event loop reconnect() ran on_close() inline when connect() failed before a socket existed. on_valkey_close() calls the user's onclose synchronously, so an onclose that calls connect() against an endpoint that keeps failing that way recursed through do_connect() -> reconnect() -> on_close() until the stack ran out (350 levels deep in a debug build), where the same onclose against a refused TCP port just loops through the event loop. Run on_close() from a task instead, which is also when a connect error callback would have delivered it. --- src/runtime/valkey_jsc/js_valkey.rs | 45 ++++++++++++------- .../reliability/connection-failures.test.ts | 10 ++++- 2 files changed, 38 insertions(+), 17 deletions(-) diff --git a/src/runtime/valkey_jsc/js_valkey.rs b/src/runtime/valkey_jsc/js_valkey.rs index ded12e70032a..1c49864133ca 100644 --- a/src/runtime/valkey_jsc/js_valkey.rs +++ b/src/runtime/valkey_jsc/js_valkey.rs @@ -1191,15 +1191,30 @@ impl JSValkeyClient { /// `ValkeyClient::on_close()` is otherwise only reached from a socket's /// close or connect-error callback. Run it for a dial that failed before /// there was a socket, so the connect() promise, `onclose`, the retry - /// policy and the poll ref are handled the same way. `on_close()` releases - /// the ref a socket would have held, so take one for it to release. - fn on_close_without_socket(&self) -> JsTerminatedResult<()> { - let _guard = self.ref_scope(); + /// policy and the poll ref are handled the same way. It runs from the + /// event loop like those callbacks do: `onclose` may call connect(), and + /// if that dial fails the same way it must not recurse into here. + /// `on_close()` releases the ref a socket would have held, so the task + /// takes one for it to release. + fn close_without_socket_next_tick(&self) { + fn run(this: *mut JSValkeyClient) -> bun_event_loop::JsResult<()> { + // SAFETY: adopts the ref taken at the enqueue site, which kept + // `this` alive until now and is released when this scope ends. + let _task_ref = unsafe { ScopedRef::adopt(this) }; + // SAFETY: live per the ref above; tasks run on the JS thread. + let this = unsafe { &*this }; + this.ref_(); + this.client_mut().status = valkey::Status::Disconnected; + let result = this.client_mut().on_close(); + this.update_poll_ref(); + result.map_err(Into::into) + } self.ref_(); - self.client_mut().status = valkey::Status::Disconnected; - let result = narrow_terminated(self.client_mut().on_close()); - self.update_poll_ref(); - result + let task = jsc::ManagedTask::ManagedTask::new(self.as_ctx_ptr(), run); + // 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) { @@ -1241,7 +1256,7 @@ impl JSValkeyClient { ); // Same outcome as a dial that fails asynchronously: another retry, // or fail() and a settled connect() promise once retries are used up. - let _ = self.on_close_without_socket(); + self.close_without_socket_next_tick(); return; } @@ -1387,10 +1402,10 @@ 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 (or the one `on_close_without_socket()` 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. + // socket (or the one `close_without_socket_next_tick()` 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 @@ -1406,8 +1421,8 @@ impl JSValkeyClient { let global_object = self.global_object; // SAFETY: adopts connect()'s socket keep-alive ref (or the one - // `on_close_without_socket()` took in its place); the caller holds its - // own scoped ref so count stays > 0 until this drops. + // `close_without_socket_next_tick()` 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()); diff --git a/test/js/valkey/reliability/connection-failures.test.ts b/test/js/valkey/reliability/connection-failures.test.ts index 1a5287d1eb26..e820441288e8 100644 --- a/test/js/valkey/reliability/connection-failures.test.ts +++ b/test/js/valkey/reliability/connection-failures.test.ts @@ -590,11 +590,17 @@ describe("Valkey: Recovering After fail()", () => { await client.connect(); client.close(); await fake.close(); - const outcome = await client.connect().then( + 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).toBe("rejected: ERR_REDIS_CONNECTION_CLOSED"); + 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(); From 2fcf109a3e9a31fbb28417b92747c14eae58c278 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Thu, 13 Aug 2026 23:53:22 +0000 Subject: [PATCH 06/21] valkey: fast-shutdown on close() so TLS closes synchronously too, typed task for the no-socket close close() closed with CloseCode::Normal. Past the handshake a TLS socket then sends close_notify and keeps the socket, and with it the close callback, until the peer answers, so over rediss:// a post-handshake fail() left connected true, had connect() resolve at once without clearing failed, and against a peer that had stopped answering stayed that way for good. A fast shutdown is still a FIN on TCP and closes a TLS socket inline, which is what everything downstream of close() assumes. The first recovery test now runs over redis:// and rediss:// and reads connected right after the rejection. The same test now runs with auto reconnect on, pinning that a failure after HELLO closes for good instead of retrying (an accepted HELLO resets the retry counter), with the reason next to the line in fail() that decides it. The deferred no-socket close was a ManagedTask, which drops nothing when the VM tears down with it still queued, leaking the client and whatever it holds. It is a second mode of ValkeyDeferredClose now, whose release_unrun gives the client ref and the poll ref back without running onclose. --- src/runtime/valkey_jsc/js_valkey.rs | 103 ++++++++++-------- src/runtime/valkey_jsc/valkey.rs | 14 ++- .../reliability/connection-failures.test.ts | 72 +++++++----- 3 files changed, 117 insertions(+), 72 deletions(-) diff --git a/src/runtime/valkey_jsc/js_valkey.rs b/src/runtime/valkey_jsc/js_valkey.rs index 1c49864133ca..37df5fc0e480 100644 --- a/src/runtime/valkey_jsc/js_valkey.rs +++ b/src/runtime/valkey_jsc/js_valkey.rs @@ -1188,29 +1188,22 @@ impl JSValkeyClient { } } - /// `ValkeyClient::on_close()` is otherwise only reached from a socket's - /// close or connect-error callback. Run it for a dial that failed before - /// there was a socket, so the connect() promise, `onclose`, the retry - /// policy and the poll ref are handled the same way. It runs from the - /// event loop like those callbacks do: `onclose` may call connect(), and - /// if that dial fails the same way it must not recurse into here. - /// `on_close()` releases the ref a socket would have held, so the task - /// takes one for it to release. + /// 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. fn close_without_socket_next_tick(&self) { - fn run(this: *mut JSValkeyClient) -> bun_event_loop::JsResult<()> { - // SAFETY: adopts the ref taken at the enqueue site, which kept - // `this` alive until now and is released when this scope ends. - let _task_ref = unsafe { ScopedRef::adopt(this) }; - // SAFETY: live per the ref above; tasks run on the JS thread. - let this = unsafe { &*this }; - this.ref_(); - this.client_mut().status = valkey::Status::Disconnected; - let result = this.client_mut().on_close(); - this.update_poll_ref(); - result.map_err(Into::into) - } + 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::ManagedTask::ManagedTask::new(self.as_ctx_ptr(), run); + 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); @@ -1402,10 +1395,10 @@ 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 (or the one `close_without_socket_next_tick()` 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. + // 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 @@ -1421,8 +1414,8 @@ impl JSValkeyClient { let global_object = self.global_object; // SAFETY: adopts connect()'s socket keep-alive ref (or the one - // `close_without_socket_next_tick()` took in its place); the caller - // holds its own scoped ref so count stays > 0 until this drops. + // `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()); @@ -1474,15 +1467,8 @@ impl JSValkeyClient { 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) { @@ -2091,27 +2077,58 @@ 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) { - let ctx = self.ctx; - // SAFETY: single-threaded; intrusive ref taken before enqueue guarantees liveness. - unsafe { - (*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 => this.client_mut().close(), + DeferredClose::WithoutSocket => { + // `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 _ = this.client_mut().on_close(); + this.update_poll_ref(); + } } } } 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 `reconnect()` and the enqueue 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 f90bb61a3a65..5f1ab45d524a 100644 --- a/src/runtime/valkey_jsc/valkey.rs +++ b/src/runtime/valkey_jsc/valkey.rs @@ -37,8 +37,9 @@ pub struct ConnectionFlags { /// 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()`; the socket is closed when it is set, - /// so it overlaps `Disconnected`. + /// Sticky until `on_open`/`connect()`. `fail()` closes the socket with a + /// fast shutdown, so by the time it returns the close callback has run on + /// TCP and TLS alike and this overlaps `Disconnected`. pub(crate) failed: bool, pub(crate) enable_auto_pipelining: bool, pub(crate) finalized: bool, @@ -615,6 +616,9 @@ impl ValkeyClient { jsvalue, ); + // Deliberate close rather than a retry, even with auto reconnect on: an + // accepted HELLO resets retry_attempts, so a server that keeps failing + // us after the handshake would otherwise be redialed forever. self.flags.is_manually_closed = true; self.close(); val @@ -640,7 +644,11 @@ impl ValkeyClient { // and run the close path ourselves afterwards. let is_semi_socket = matches!(socket.socket(), uws::InternalSocket::Connected(_)) && !socket.is_established(); - socket.close(uws::CloseCode::Normal); + // Still a FIN on TCP. On TLS, `Normal` would send close_notify and hold + // the socket (and its close callback) until the peer answers, which a + // peer that stopped responding never does; everything after this + // expects the close callback to have run by the time close() returns. + socket.close(uws::CloseCode::FastShutdown); if is_semi_socket { self.status = Status::Disconnected; let _ = self.on_close(); diff --git a/test/js/valkey/reliability/connection-failures.test.ts b/test/js/valkey/reliability/connection-failures.test.ts index e820441288e8..18ddde138842 100644 --- a/test/js/valkey/reliability/connection-failures.test.ts +++ b/test/js/valkey/reliability/connection-failures.test.ts @@ -1,8 +1,9 @@ import { RedisClient } from "bun"; import { describe, expect, mock, test } from "bun:test"; -import { bunEnv, bunExe, isWindows, tempDir } from "harness"; +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"; /** @@ -447,9 +448,12 @@ 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. - function helloServer(replies: Partial string>> = {}) { + function helloServer( + replies: Partial string>> = {}, + { secure = false } = {}, + ) { let connections = 0; - const server = net.createServer(socket => { + const onConnection = (socket: net.Socket) => { connections += 1; const connection = connections; socket.on("data", chunk => { @@ -458,7 +462,10 @@ describe("Valkey: Recovering After fail()", () => { if (text.includes("PING")) socket.write(replies.PING?.(connection) ?? "+PONG\r\n"); }); socket.on("error", () => {}); - }); + }; + const server: net.Server = secure + ? tls.createServer({ key: tlsCert.key, cert: tlsCert.cert }, onConnection) + : net.createServer(onConnection); return { server, get connections() { @@ -488,28 +495,41 @@ describe("Valkey: Recovering After fail()", () => { return promise; } - test("a failure while connected closes the socket, fires onclose, and connect() reconnects", async () => { - // 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") }); - const port = await fake.listen(); - const closed = Promise.withResolvers(); - const client = new RedisClient(`redis://127.0.0.1:${port}`, { autoReconnect: false }); - 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" }); - expect(await closed.promise).toBeInstanceOf(Error); - expect(client.connected).toBe(false); - await client.connect(); - expect(await client.ping()).toBe("PONG"); - expect(fake.connections).toBe(2); - } finally { - client.close(); - fake.server.close(); - } - }); + // A failure after the handshake closes the connection for good even though + // auto reconnect is left on here: an accepted HELLO resets the retry counter, + // so retrying a server that keeps failing us after it would never end. + // 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(); + 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("a connect() issued from onclose after a refused connection rejects instead of hanging", async () => { // Nothing listens on the port a just-closed listener used. From a603f2c8360454bf8eadc2a58b32e99eae9896c8 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Fri, 14 Aug 2026 00:13:02 +0000 Subject: [PATCH 07/21] valkey: close outright from fail(), fast shutdown only for disconnect() usockets defers a fast shutdown as well while the socket still owns the loop's ciphertext spill, which is the case whenever the last TLS batch flush hit a full kernel buffer, i.e. precisely the peer that stopped reading. A close with the reset code is never deferred, so fail() uses that (an RST instead of a FIN on TCP, which does not matter once everything on the connection has been rejected); disconnect() and the finalizer keep the fast shutdown. close() takes the code, and the comments describe what each one guarantees. --- src/runtime/valkey_jsc/js_valkey.rs | 4 ++-- src/runtime/valkey_jsc/valkey.rs | 26 +++++++++++++++----------- 2 files changed, 17 insertions(+), 13 deletions(-) diff --git a/src/runtime/valkey_jsc/js_valkey.rs b/src/runtime/valkey_jsc/js_valkey.rs index 37df5fc0e480..e1ac63975c2c 100644 --- a/src/runtime/valkey_jsc/js_valkey.rs +++ b/src/runtime/valkey_jsc/js_valkey.rs @@ -1902,7 +1902,7 @@ impl SocketHandler { // handshake. this.global_object.clear_exception(); this.client_mut().flags.is_manually_closed = true; - this.client_mut().close(); + this.client_mut().close(uws::CloseCode::Failure); return Ok(()); } }; @@ -2100,7 +2100,7 @@ impl ValkeyDeferredClose { // SAFETY: live per the ref above; tasks run on the JS thread. let this = unsafe { &*self.ctx }; match self.what { - DeferredClose::Socket => this.client_mut().close(), + DeferredClose::Socket => this.client_mut().close(uws::CloseCode::FastShutdown), DeferredClose::WithoutSocket => { // `on_close()` ends in `on_valkey_close`/`on_valkey_reconnect`, // which release the ref the socket would have held. diff --git a/src/runtime/valkey_jsc/valkey.rs b/src/runtime/valkey_jsc/valkey.rs index 5f1ab45d524a..d67015ce5dce 100644 --- a/src/runtime/valkey_jsc/valkey.rs +++ b/src/runtime/valkey_jsc/valkey.rs @@ -37,9 +37,9 @@ pub struct ConnectionFlags { /// 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()`. `fail()` closes the socket with a - /// fast shutdown, so by the time it returns the close callback has run on - /// TCP and TLS alike and this overlaps `Disconnected`. + /// Sticky until `on_open`/`connect()`. `fail()` closes the socket outright + /// (see `close()`), so by the time it returns the close callback has run + /// and this overlaps `Disconnected`. pub(crate) failed: bool, pub(crate) enable_auto_pipelining: bool, pub(crate) finalized: bool, @@ -620,11 +620,19 @@ impl ValkeyClient { // accepted HELLO resets retry_attempts, so a server that keeps failing // us after the handshake would otherwise be redialed forever. self.flags.is_manually_closed = true; - self.close(); + self.close(uws::CloseCode::Failure); val } - pub fn close(&mut self) { + /// `Failure` closes the socket outright, so the close callback has run by + /// the time this returns; on TCP that is an RST rather than a FIN, which + /// is fine once everything on the connection has been rejected, and it is + /// what `fail()` needs. `FastShutdown` is a FIN, and on TLS usockets still + /// holds the socket (and the callback) while ciphertext is waiting for + /// room in the kernel buffer, finishing the close once it drains or the + /// peer turns out to be gone; that is the right close for a client that + /// is merely done with the connection. + pub fn close(&mut self, code: uws::CloseCode) { let socket = core::mem::replace( &mut self.socket, AnySocket::SocketTcp(uws::SocketTCP::detached()), @@ -644,11 +652,7 @@ impl ValkeyClient { // and run the close path ourselves afterwards. let is_semi_socket = matches!(socket.socket(), uws::InternalSocket::Connected(_)) && !socket.is_established(); - // Still a FIN on TCP. On TLS, `Normal` would send close_notify and hold - // the socket (and its close callback) until the peer answers, which a - // peer that stopped responding never does; everything after this - // expects the close callback to have run by the time close() returns. - socket.close(uws::CloseCode::FastShutdown); + socket.close(code); if is_semi_socket { self.status = Status::Disconnected; let _ = self.on_close(); @@ -1513,7 +1517,7 @@ impl ValkeyClient { self.flags.is_manually_closed = true; self.unregister_auto_flusher(); if self.status == Status::Connected || self.status == Status::Connecting { - self.close(); + self.close(uws::CloseCode::FastShutdown); } } From 218faf014a00aa72af26a554cb41ca22d3af453a Mon Sep 17 00:00:00 2001 From: Alistair Smith Date: Fri, 14 Aug 2026 19:02:13 -0700 Subject: [PATCH 08/21] valkey: defer the close for a TLS context that cannot be built --- src/runtime/valkey_jsc/js_valkey.rs | 21 ++++----- .../reliability/connection-failures.test.ts | 47 +++++++++++++++++++ 2 files changed, 56 insertions(+), 12 deletions(-) diff --git a/src/runtime/valkey_jsc/js_valkey.rs b/src/runtime/valkey_jsc/js_valkey.rs index b8d9ecaeefd0..95abb53c7263 100644 --- a/src/runtime/valkey_jsc/js_valkey.rs +++ b/src/runtime/valkey_jsc/js_valkey.rs @@ -1470,12 +1470,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(); @@ -1512,11 +1506,11 @@ 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; + // Reported like a dial that fails asynchronously: Connecting keeps + // the wrapper and the event loop until the deferred close runs. + self.client_mut().status = valkey::Status::Connecting; + self.update_poll_ref(); + self.close_without_socket_next_tick(); return Ok(()); } let ssl_ctx: Option<*mut uws::SslCtx> = match &self.client.get().tls { @@ -1544,6 +1538,9 @@ impl JSValkeyClient { // `owner_ptr` opaquely (no overlapping write). let owner_ptr: *mut JSValkeyClient = std::ptr::from_ref::(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. @@ -2065,7 +2062,7 @@ impl bun_event_loop::Taskable for ValkeyDeferredClose { // Script-free bookkeeping; do it. DeferredClose::Socket => task.run(), // The VM is going away: `on_close()` would run `onclose`, so only - // give back what `reconnect()` and the enqueue took. + // give back what the dial and the enqueue took. DeferredClose::WithoutSocket => { // SAFETY: as in `run`. let _enqueue_ref = unsafe { ScopedRef::adopt(task.ctx) }; diff --git a/test/js/valkey/reliability/connection-failures.test.ts b/test/js/valkey/reliability/connection-failures.test.ts index 18ddde138842..5a128da53431 100644 --- a/test/js/valkey/reliability/connection-failures.test.ts +++ b/test/js/valkey/reliability/connection-failures.test.ts @@ -628,6 +628,53 @@ describe("Valkey: Recovering After fail()", () => { }, ); + 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. From 48e3ef50289fefcbc6acd9c69a7bf886c69fc077 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sat, 15 Aug 2026 09:34:57 +0000 Subject: [PATCH 09/21] valkey: stay Connecting until a deferred no-socket close runs, pin the remaining behaviour in tests A reconnect whose dial failed outright left the client Disconnected for the tick between the failure and the queued close. JS that ran in that tick (a timer due alongside the retry, or the caller of connect() itself) could dial again, which the queued close then stamped Disconnected over and ran on_close() against, or call close(), which returned without marking anything. close_without_socket_next_tick() now holds Connecting and the poll ref itself, as a dial in flight would, for both of its callers, and the task leaves a socket alone should one exist by the time it runs. CloseCode's docs now say which codes usockets may defer and why fail() uses the one it does; the valkey comment points there instead of repeating it. Tests added for the idle timeout on its own entry point, connected inside onclose after a server-side drop, close() over TLS against a peer that never answers close_notify, a failure while the peer has stopped reading (which distinguishes the close code: writes are issued until the kernel takes nothing twice in a row, so the socket still owns undelivered ciphertext), the two windows above, and a worker exiting with the deferred close still queued; the retry test now observes the retry through the command it rejects. --- src/runtime/valkey_jsc/js_valkey.rs | 21 +- src/runtime/valkey_jsc/valkey.rs | 13 +- src/uws_sys/us_socket_t.rs | 21 +- .../reliability/connection-failures.test.ts | 285 +++++++++++++++++- .../workers/worker-terminate-lifetime.test.ts | 96 +++--- 5 files changed, 368 insertions(+), 68 deletions(-) diff --git a/src/runtime/valkey_jsc/js_valkey.rs b/src/runtime/valkey_jsc/js_valkey.rs index 95abb53c7263..d257bcfc86ad 100644 --- a/src/runtime/valkey_jsc/js_valkey.rs +++ b/src/runtime/valkey_jsc/js_valkey.rs @@ -1147,7 +1147,16 @@ impl JSValkeyClient { /// 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() itself) then gets the cached promise from + /// connect() instead of a second dial, 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); } @@ -1506,10 +1515,6 @@ impl JSValkeyClient { b"Failed to create TLS context", protocol::RedisError::ConnectionClosed, )?; - // Reported like a dial that fails asynchronously: Connecting keeps - // the wrapper and the event loop until the deferred close runs. - self.client_mut().status = valkey::Status::Connecting; - self.update_poll_ref(); self.close_without_socket_next_tick(); return Ok(()); } @@ -2041,6 +2046,12 @@ impl ValkeyDeferredClose { 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_(); @@ -2062,7 +2073,7 @@ impl bun_event_loop::Taskable for ValkeyDeferredClose { // Script-free bookkeeping; do it. DeferredClose::Socket => task.run(), // The VM is going away: `on_close()` would run `onclose`, so only - // give back what the dial and the enqueue took. + // give back what `close_without_socket_next_tick` took. DeferredClose::WithoutSocket => { // SAFETY: as in `run`. let _enqueue_ref = unsafe { ScopedRef::adopt(task.ctx) }; diff --git a/src/runtime/valkey_jsc/valkey.rs b/src/runtime/valkey_jsc/valkey.rs index 490fde68f6be..b057b1516657 100644 --- a/src/runtime/valkey_jsc/valkey.rs +++ b/src/runtime/valkey_jsc/valkey.rs @@ -621,14 +621,11 @@ impl ValkeyClient { val.and(closed) } - /// `Failure` closes the socket outright, so the close callback has run by - /// the time this returns; on TCP that is an RST rather than a FIN, which - /// is fine once everything on the connection has been rejected, and it is - /// what `fail()` needs. `FastShutdown` is a FIN, and on TLS usockets still - /// holds the socket (and the callback) while ciphertext is waiting for - /// room in the kernel buffer, finishing the close once it drains or the - /// peer turns out to be gone; that is the right close for a client that - /// is merely done with the connection. + /// `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. /// /// For a half-open socket this runs `on_close` itself (see below) and returns what its /// `onclose` listener left pending; the caller propagates that like any other callback 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 5a128da53431..ec53850d7b87 100644 --- a/test/js/valkey/reliability/connection-failures.test.ts +++ b/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 { once } from "events"; import { bunEnv, bunExe, isWindows, tempDir, tls as tlsCert } from "harness"; import net from "net"; import path from "path"; @@ -447,29 +448,36 @@ 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. + // `+PONG` unless `replies` says otherwise for that connection (a hook that + // returns null answers nothing). Other commands are never answered. function helloServer( - replies: Partial string>> = {}, - { secure = false } = {}, + replies: Partial string | null>> = {}, + { secure = false, allowHalfOpen = false } = {}, ) { - let connections = 0; + const sockets: net.Socket[] = []; const onConnection = (socket: net.Socket) => { - connections += 1; - const connection = connections; + sockets.push(socket); + const connection = sockets.length; socket.on("data", chunk => { const text = chunk.toString("latin1"); - if (text.includes("HELLO")) socket.write(replies.HELLO?.(connection) ?? "+OK\r\n"); - if (text.includes("PING")) socket.write(replies.PING?.(connection) ?? "+PONG\r\n"); + 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 }, onConnection) - : net.createServer(onConnection); + ? tls.createServer({ key: tlsCert.key, cert: tlsCert.cert, allowHalfOpen }, onConnection) + : net.createServer({ allowHalfOpen }, onConnection); return { server, + sockets, get connections() { - return connections; + return sockets.length; }, listen: () => new Promise(resolve => @@ -480,6 +488,12 @@ describe("Valkey: Recovering After fail()", () => { }; } + // 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 { + 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 { const { promise, resolve } = Promise.withResolvers(); @@ -531,6 +545,151 @@ describe("Valkey: Recovering After fail()", () => { }, ); + 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 }>(); + // Today the timer armed by connect() carries connectionTimeout and turns + // into the idle timeout once the handshake is done; whichever way it gets + // armed, 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("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(); + 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(); + 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[] = []; + // 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. + for (let stuckFlushes = 0; stuckFlushes < 2; ) { + 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; + } + 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(); @@ -558,9 +717,13 @@ describe("Valkey: Recovering After fail()", () => { // 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); - await client.connect().catch(err => console.log("connect rejected", 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, @@ -569,12 +732,108 @@ describe("Valkey: Recovering After fail()", () => { }); 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", + 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 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 after PING, which schedules a retry + // 50ms later; the listener is gone by then, so that retry fails outright. + const first = helloServer({ + PING: (_connection, socket) => { + socket.end("+PONG\r\n"); + 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}`); + let fromTimer: Promise | undefined; + try { + await client.connect(); + expect(await client.ping()).toBe("PONG"); + await first.close(); + while (client.connected) await Bun.sleep(1); + // The retry is armed now, due before a timer armed here with the same + // delay. Blocking past both makes the loop fire them in one pass: the + // retry fails outright, then this callback runs in the window before + // the deferred close, brings the listener back and calls connect(). + const ran = Promise.withResolvers(); + setTimeout(() => { + void second.listenUnix(socketPath); + fromTimer = client.connect(); + ran.resolve(); + }, 50); + Bun.sleepSync(120); + await ran.promise; + while (second.connections === 0) await Bun.sleep(1); + // A second dial would arrive within the next retry delay (50ms); 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"); + await fromTimer; + expect({ connected: client.connected, connections: second.connections }).toEqual({ + connected: true, + connections: 1, + }); + } finally { + client.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"); diff --git a/test/js/web/workers/worker-terminate-lifetime.test.ts b/test/js/web/workers/worker-terminate-lifetime.test.ts index 88ab85eea2f4..076182683339 100644 --- a/test/js/web/workers/worker-terminate-lifetime.test.ts +++ b/test/js/web/workers/worker-terminate-lifetime.test.ts @@ -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,44 +777,24 @@ 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", - 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(); - `, - ], - env: { ...bunEnv, UV_THREADPOOL_SIZE: "4" }, - stdout: "pipe", - stderr: "pipe", - }); + () => + exitRightAfterConnecting( + "new Bun.RedisClient('redis://127.0.0.1:9', { connectionTimeout: 100, autoReconnect: false }).connect()", + ), + timeout, +); - 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); - }, +// 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). +test.skipIf(!isDebug)( + "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, ); From 404fecd20c15aa1bd81461aed11b1efade2435b0 Mon Sep 17 00:00:00 2001 From: Alistair Smith Date: Sat, 15 Aug 2026 10:12:43 -0700 Subject: [PATCH 10/21] valkey tests: cap the backpressure loop and close unix listeners in finally The stuck-flush loop in the TLS spill test now stops after 256 SETs and asserts that backpressure was actually reached, so a regression fails on an assertion instead of the per-test timeout. The unix-socket tests close their first listener in finally as well, so an early failure does not leak the fd. No-Verification-Needed: test-only change --- test/js/valkey/reliability/connection-failures.test.ts | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/test/js/valkey/reliability/connection-failures.test.ts b/test/js/valkey/reliability/connection-failures.test.ts index ec53850d7b87..9ae3e3efd809 100644 --- a/test/js/valkey/reliability/connection-failures.test.ts +++ b/test/js/valkey/reliability/connection-failures.test.ts @@ -667,13 +667,15 @@ describe("Valkey: Recovering After fail()", () => { // 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. - for (let stuckFlushes = 0; stuckFlushes < 2; ) { + 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" }); @@ -775,6 +777,7 @@ describe("Valkey: Recovering After fail()", () => { }); } finally { client.close(); + await first.close(); await second.close(); } }, @@ -829,6 +832,7 @@ describe("Valkey: Recovering After fail()", () => { }); } finally { client.close(); + await first.close(); await second.close(); } }, @@ -853,6 +857,7 @@ describe("Valkey: Recovering After fail()", () => { expect({ first: first.connections, second: second.connections }).toEqual({ first: 1, second: 1 }); } finally { client.close(); + await first.close(); await second.close(); } }); @@ -883,6 +888,7 @@ describe("Valkey: Recovering After fail()", () => { await expect(client.ping()).rejects.toMatchObject({ code: "ERR_REDIS_CONNECTION_CLOSED" }); } finally { client.close(); + await fake.close(); } }, ); From 38efbc1b430e676d423a9e82a4efefabadc66d8c Mon Sep 17 00:00:00 2001 From: robobun Date: Sat, 15 Aug 2026 17:47:11 +0000 Subject: [PATCH 11/21] valkey tests: arm the same-tick connect() from the PING rejection instead of racing the retry timer The timer that calls connect() in the window after a failed retry was armed right after the retry with the same 50ms delay, so on a release build the two were due in the same millisecond and the callback could fire first, dialling while the retry was still pending and failing the test on a second connection. The server now drops connection 1 without answering PING. The close that schedules the retry rejects that PING, and the test arms its timer and blocks from that rejection's continuation, before the loop turns, with a delay 100ms longer than the retry's, so the two always fire in one pass in due order. The connect() made by the callback reports its outcome through a handler attached immediately, so a failed assertion is no longer reported as an unhandled rejection from the close() in finally. No-Verification-Needed: test-only change --- .../reliability/connection-failures.test.ts | 61 +++++++++++-------- 1 file changed, 37 insertions(+), 24 deletions(-) diff --git a/test/js/valkey/reliability/connection-failures.test.ts b/test/js/valkey/reliability/connection-failures.test.ts index 9ae3e3efd809..e6f18bd14dc7 100644 --- a/test/js/valkey/reliability/connection-failures.test.ts +++ b/test/js/valkey/reliability/connection-failures.test.ts @@ -788,11 +788,10 @@ describe("Valkey: Recovering After fail()", () => { async () => { using dir = tempDir("valkey-unix", {}); const socketPath = path.join(String(dir), "r.sock"); - // Connection 1 is dropped by the server after PING, which schedules a retry - // 50ms later; the listener is gone by then, so that retry fails outright. + // Connection 1 is dropped by the server instead of answering PING. const first = helloServer({ PING: (_connection, socket) => { - socket.end("+PONG\r\n"); + socket.end(); return null; }, }); @@ -801,35 +800,49 @@ describe("Valkey: Recovering After fail()", () => { const second = helloServer({ HELLO: () => null }); await first.listenUnix(socketPath); const client = new RedisClient(`redis+unix://${socketPath}`); - let fromTimer: Promise | undefined; + // 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(); try { await client.connect(); - expect(await client.ping()).toBe("PONG"); - await first.close(); - while (client.connected) await Bun.sleep(1); - // The retry is armed now, due before a timer armed here with the same - // delay. Blocking past both makes the loop fire them in one pass: the - // retry fails outright, then this callback runs in the window before - // the deferred close, brings the listener back and calls connect(). - const ran = Promise.withResolvers(); + // 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 = client.connect(); - ran.resolve(); - }, 50); - Bun.sleepSync(120); - await ran.promise; + 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); - // A second dial would arrive within the next retry delay (50ms); this - // is the bound on asserting that it never comes. + // 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"); - await fromTimer; - expect({ connected: client.connected, connections: second.connections }).toEqual({ - connected: true, - connections: 1, - }); + expect({ + fromTimer: await fromTimer.promise, + connected: client.connected, + connections: second.connections, + }).toEqual({ fromTimer: "connected", connected: true, connections: 1 }); } finally { client.close(); await first.close(); From 890824df4a8aedb59f5cea4eef59d7763cb403d2 Mon Sep 17 00:00:00 2001 From: Alistair Smith Date: Sat, 15 Aug 2026 12:06:32 -0700 Subject: [PATCH 12/21] ci: rerun build canceled by queue cleanup From c793bab5f600161560e0cdffdc6dd90ca14c48cc Mon Sep 17 00:00:00 2001 From: robobun Date: Mon, 17 Aug 2026 19:41:22 +0000 Subject: [PATCH 13/21] valkey tests: make the stub's listen helpers reject when the listen fails They only wired the listening callback, so a failed bind (a unix socket path that is too long, a stale socket file) left the awaited promise pending and the test ended by timeout with nothing pointing at the listen. Waiting on the listening event through events.once() rejects with the listen error instead. No-Verification-Needed: test-only change --- .../reliability/connection-failures.test.ts | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/test/js/valkey/reliability/connection-failures.test.ts b/test/js/valkey/reliability/connection-failures.test.ts index e6f18bd14dc7..bf906f595a93 100644 --- a/test/js/valkey/reliability/connection-failures.test.ts +++ b/test/js/valkey/reliability/connection-failures.test.ts @@ -479,11 +479,16 @@ describe("Valkey: Recovering After fail()", () => { get connections() { return sockets.length; }, - listen: () => - new Promise(resolve => - server.listen(0, "127.0.0.1", () => resolve((server.address() as net.AddressInfo).port)), - ), - listenUnix: (socketPath: string) => new Promise(resolve => server.listen(socketPath, resolve)), + // 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)), }; } From 204e8a9c17872096d5f42572c18900e0038b2ad4 Mon Sep 17 00:00:00 2001 From: robobun Date: Mon, 17 Aug 2026 20:07:21 +0000 Subject: [PATCH 14/21] ci: retrigger From 000413ca7b4af7d16c065cf60d35bbbadceb9473 Mon Sep 17 00:00:00 2001 From: robobun Date: Tue, 18 Aug 2026 03:04:07 +0000 Subject: [PATCH 15/21] valkey: run a first dial that fails outright through the deferred close too do_connect() and send() handled a first dial whose connect() failed before there was a socket on their own: reject at once with SOCKET_CLOSED_BEFORE_CONNECTION, no onclose, no retry policy, and the connection promise stayed cached, so a later dial that succeeded settled it a second time. A failed re-dial already goes through close_without_socket_next_tick(); both first-dial arms now do the same, so the client stays Connecting until the task runs on_close(), which settles the promise, rejects or keeps the queued command, runs onclose and applies the retry policy exactly as for a refused dial. Tests: connect() and a command as the entry point, each with auto reconnect off (rejects with ERR_REDIS_CONNECTION_CLOSED, onclose once, and a connect() afterwards dials instead of returning the settled promise) and on (a listener that appears gets connected); a worker torn down with the task and the command still queued; a pub/sub listener that closes and reconnects while a further push is buffered behind its message, covering the buffer path of on_data(). The reason given for fail() being a deliberate close is corrected: a failure the client detects itself has always been one, only peer closes are retried. --- src/runtime/valkey_jsc/js_valkey.rs | 51 ++++--- src/runtime/valkey_jsc/valkey.rs | 6 +- .../reliability/connection-failures.test.ts | 125 +++++++++++++++++- .../workers/worker-terminate-lifetime.test.ts | 10 +- 4 files changed, 157 insertions(+), 35 deletions(-) diff --git a/src/runtime/valkey_jsc/js_valkey.rs b/src/runtime/valkey_jsc/js_valkey.rs index e40ca0ca9c62..d7dc345ebf53 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); } @@ -1150,10 +1147,11 @@ impl JSValkeyClient { /// /// 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() itself) then gets the cached promise from - /// connect() instead of a second dial, 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. + /// 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(); @@ -1579,20 +1577,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); diff --git a/src/runtime/valkey_jsc/valkey.rs b/src/runtime/valkey_jsc/valkey.rs index 07ef0acc4b38..13c6d4f4f47e 100644 --- a/src/runtime/valkey_jsc/valkey.rs +++ b/src/runtime/valkey_jsc/valkey.rs @@ -613,9 +613,9 @@ impl ValkeyClient { jsvalue, ); - // Deliberate close rather than a retry, even with auto reconnect on: an - // accepted HELLO resets retry_attempts, so a server that keeps failing - // us after the handshake would otherwise be redialed forever. + // A failure the client detected itself (idle timeout, protocol or + // handshake error) has always been a deliberate close; only closes the + // peer initiated go through the retry policy, even with auto reconnect on. self.flags.is_manually_closed = true; let closed = self.close(uws::CloseCode::Failure); // unconditionally, whatever `val` is val.and(closed) diff --git a/test/js/valkey/reliability/connection-failures.test.ts b/test/js/valkey/reliability/connection-failures.test.ts index bf906f595a93..cc3fc14dd303 100644 --- a/test/js/valkey/reliability/connection-failures.test.ts +++ b/test/js/valkey/reliability/connection-failures.test.ts @@ -514,10 +514,11 @@ describe("Valkey: Recovering After fail()", () => { return promise; } - // A failure after the handshake closes the connection for good even though - // auto reconnect is left on here: an accepted HELLO resets the retry counter, - // so retrying a server that keeps failing us after it would never end. - // onclose only fires on that terminal path, so it firing is the assertion. + // 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], @@ -911,6 +912,72 @@ describe("Valkey: Recovering After fail()", () => { }, ); + // 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][] = [ + ["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: [ @@ -979,6 +1046,56 @@ describe("Valkey: Recovering After fail()", () => { } }); + 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(); + const reconnect = Promise.withResolvers(); + 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 => { diff --git a/test/js/web/workers/worker-terminate-lifetime.test.ts b/test/js/web/workers/worker-terminate-lifetime.test.ts index 750c9783dc13..c5e0e659feb4 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 @@ -798,6 +798,14 @@ test.skipIf(!isDebug)( 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 || 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, +); + // worker.terminate() never stopped a worker parked in // Atomics.wait() (sync-over-async worker pools park exactly there). JSC wakes // the parked thread when termination is requested, but the wake-up predicate From 5573ec4478f5da34b9097964842fa5cb0db826af Mon Sep 17 00:00:00 2001 From: Alistair Smith Date: Thu, 13 Aug 2026 16:33:40 -0700 Subject: [PATCH 16/21] valkey: count idle time from connect and restart it on incoming data --- src/runtime/valkey_jsc/js_valkey.rs | 5 ++ .../reliability/connection-failures.test.ts | 53 +++++++++++++++++++ 2 files changed, 58 insertions(+) diff --git a/src/runtime/valkey_jsc/js_valkey.rs b/src/runtime/valkey_jsc/js_valkey.rs index d7dc345ebf53..d5f4daa7ac0f 100644 --- a/src/runtime/valkey_jsc/js_valkey.rs +++ b/src/runtime/valkey_jsc/js_valkey.rs @@ -1224,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| { @@ -1927,6 +1929,9 @@ impl SocketHandler { 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 } diff --git a/test/js/valkey/reliability/connection-failures.test.ts b/test/js/valkey/reliability/connection-failures.test.ts index cc3fc14dd303..8fcc0e108b7f 100644 --- a/test/js/valkey/reliability/connection-failures.test.ts +++ b/test/js/valkey/reliability/connection-failures.test.ts @@ -698,6 +698,59 @@ describe("Valkey: Recovering After fail()", () => { } }); + test("an idle connection is closed after idleTimeout, and connect() reconnects", async () => { + const closed = Promise.withResolvers(); + const fake = helloServer(); + const port = await fake.listen(); + const client = new RedisClient(`redis://127.0.0.1:${port}`, { idleTimeout: 50 }); + try { + client.onclose = err => closed.resolve(err); + await client.connect(); + expect(client.connected).toBe(true); + // Nothing is sent, so only the idle timer can end this; connectionTimeout + // is still at its 10s default, well past the test's own timeout. + expect(await closed.promise).toBeInstanceOf(Error); + expect(client.connected).toBe(false); + await client.connect(); + expect(await client.ping()).toBe("PONG"); + expect(fake.connections).toBe(2); + } finally { + client.close(); + await fake.close(); + } + }); + + test("data from the server restarts the idle timer", async () => { + let pushes = 0; + const server = net.createServer(socket => { + socket.on("data", chunk => { + if (!chunk.toString("latin1").includes("HELLO")) return; + socket.write("+OK\r\n"); + // Unsolicited pushes 25ms apart, each well inside the 100ms idle window; + // together they outlast it twice over. + const timer = setInterval(() => { + socket.write(">2\r\n$7\r\nmessage\r\n$2\r\nhi\r\n"); + if (++pushes === 8) clearInterval(timer); + }, 25); + socket.on("close", () => clearInterval(timer)); + }); + socket.on("error", () => {}); + }); + await new Promise(resolve => server.listen(0, "127.0.0.1", resolve)); + const port = (server.address() as net.AddressInfo).port; + const closed = Promise.withResolvers(); + const client = new RedisClient(`redis://127.0.0.1:${port}`, { idleTimeout: 100, autoReconnect: false }); + try { + client.onclose = () => closed.resolve(); + await client.connect(); + await closed.promise; + expect(pushes).toBe(8); + } finally { + client.close(); + 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(); From 97f835bb4ecf58b6efbba2e8ff3c84456bea2d53 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Fri, 14 Aug 2026 01:11:29 +0000 Subject: [PATCH 17/21] test(valkey): make the idle timer tests hang or fail outright without the fix instead of racing it The silent connection test dials with connectionTimeout: 0, so the timer the accepted HELLO arms is the only thing that can close the connection: without the fix nothing arms it and the test hangs, whatever the test timeout is. It awaits that close for the first connection and for the one connect() dials after it, instead of pinging over a connection whose 50ms idle timer is already running. The incoming data test answers PING with 30 pushes 20ms apart and no PONG, and waits for PING to be rejected with ERR_REDIS_IDLE_TIMEOUT. All 30 pushes have to be out by then, so a timer that data does not restart fails it at about push 20, and a stall has to exceed 380ms rather than 75ms to fail it spuriously. Both tests fail against a build without the src change. --- .../reliability/connection-failures.test.ts | 112 +++++++++--------- 1 file changed, 59 insertions(+), 53 deletions(-) diff --git a/test/js/valkey/reliability/connection-failures.test.ts b/test/js/valkey/reliability/connection-failures.test.ts index 8fcc0e108b7f..e495172c0814 100644 --- a/test/js/valkey/reliability/connection-failures.test.ts +++ b/test/js/valkey/reliability/connection-failures.test.ts @@ -600,6 +600,65 @@ describe("Valkey: Recovering After fail()", () => { } }); + 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(); + 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({ @@ -698,59 +757,6 @@ describe("Valkey: Recovering After fail()", () => { } }); - test("an idle connection is closed after idleTimeout, and connect() reconnects", async () => { - const closed = Promise.withResolvers(); - const fake = helloServer(); - const port = await fake.listen(); - const client = new RedisClient(`redis://127.0.0.1:${port}`, { idleTimeout: 50 }); - try { - client.onclose = err => closed.resolve(err); - await client.connect(); - expect(client.connected).toBe(true); - // Nothing is sent, so only the idle timer can end this; connectionTimeout - // is still at its 10s default, well past the test's own timeout. - expect(await closed.promise).toBeInstanceOf(Error); - expect(client.connected).toBe(false); - await client.connect(); - expect(await client.ping()).toBe("PONG"); - expect(fake.connections).toBe(2); - } finally { - client.close(); - await fake.close(); - } - }); - - test("data from the server restarts the idle timer", async () => { - let pushes = 0; - const server = net.createServer(socket => { - socket.on("data", chunk => { - if (!chunk.toString("latin1").includes("HELLO")) return; - socket.write("+OK\r\n"); - // Unsolicited pushes 25ms apart, each well inside the 100ms idle window; - // together they outlast it twice over. - const timer = setInterval(() => { - socket.write(">2\r\n$7\r\nmessage\r\n$2\r\nhi\r\n"); - if (++pushes === 8) clearInterval(timer); - }, 25); - socket.on("close", () => clearInterval(timer)); - }); - socket.on("error", () => {}); - }); - await new Promise(resolve => server.listen(0, "127.0.0.1", resolve)); - const port = (server.address() as net.AddressInfo).port; - const closed = Promise.withResolvers(); - const client = new RedisClient(`redis://127.0.0.1:${port}`, { idleTimeout: 100, autoReconnect: false }); - try { - client.onclose = () => closed.resolve(); - await client.connect(); - await closed.promise; - expect(pushes).toBe(8); - } finally { - client.close(); - 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(); From 9c3ff0b346bc8d0a551f030576a30271ec5b6dda Mon Sep 17 00:00:00 2001 From: Alistair Smith Date: Mon, 17 Aug 2026 22:28:06 -0700 Subject: [PATCH 18/21] test: describe how the idle timer is armed now that HELLO OK and data re-arm it --- test/js/valkey/reliability/connection-failures.test.ts | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/test/js/valkey/reliability/connection-failures.test.ts b/test/js/valkey/reliability/connection-failures.test.ts index e495172c0814..557495511ce3 100644 --- a/test/js/valkey/reliability/connection-failures.test.ts +++ b/test/js/valkey/reliability/connection-failures.test.ts @@ -558,9 +558,9 @@ describe("Valkey: Recovering After fail()", () => { const fake = helloServer({}, { secure }); const port = await fake.listen(); const closed = Promise.withResolvers<{ err: Error & { code: string }; connectedInsideOnclose: boolean }>(); - // Today the timer armed by connect() carries connectionTimeout and turns - // into the idle timeout once the handshake is done; whichever way it gets - // armed, what is pinned here is what happens when it fires on an idle + // 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, From bc6884f7d5152f7c30740733d32c8f1946dcf5cc Mon Sep 17 00:00:00 2001 From: Alistair Smith Date: Mon, 17 Aug 2026 22:28:07 -0700 Subject: [PATCH 19/21] test: pin what a rejected SELECT after an accepted HELLO does to the client --- .../reliability/connection-failures.test.ts | 40 +++++++++++++++++++ 1 file changed, 40 insertions(+) diff --git a/test/js/valkey/reliability/connection-failures.test.ts b/test/js/valkey/reliability/connection-failures.test.ts index 557495511ce3..16ffa070f9f6 100644 --- a/test/js/valkey/reliability/connection-failures.test.ts +++ b/test/js/valkey/reliability/connection-failures.test.ts @@ -1105,6 +1105,46 @@ describe("Valkey: Recovering After fail()", () => { } }); + 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 | 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 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)[]) => From a96dd901d7ff3db47b80e9d17265bff6a44636e1 Mon Sep 17 00:00:00 2001 From: Alistair Smith Date: Mon, 17 Aug 2026 22:28:07 -0700 Subject: [PATCH 20/21] valkey: keep auto-reconnect on a duplicate of a failed client --- src/runtime/valkey_jsc/valkey.rs | 16 ++++---- .../reliability/connection-failures.test.ts | 41 +++++++++++++++++++ 2 files changed, 50 insertions(+), 7 deletions(-) diff --git a/src/runtime/valkey_jsc/valkey.rs b/src/runtime/valkey_jsc/valkey.rs index 13c6d4f4f47e..ff8d11287b11 100644 --- a/src/runtime/valkey_jsc/valkey.rs +++ b/src/runtime/valkey_jsc/valkey.rs @@ -37,7 +37,8 @@ pub struct ConnectionFlags { pub(crate) is_reconnecting: bool, /// Sticky until `on_open`/`connect()`. `fail()` closes the socket outright /// (see `close()`), so by the time it returns the close callback has run - /// and this overlaps `Disconnected`. + /// (`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, @@ -614,9 +615,10 @@ impl ValkeyClient { ); // A failure the client detected itself (idle timeout, protocol or - // handshake error) has always been a deliberate close; only closes the - // peer initiated go through the retry policy, even with auto reconnect on. - self.flags.is_manually_closed = true; + // 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) } @@ -673,9 +675,9 @@ impl ValkeyClient { 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(()); diff --git a/test/js/valkey/reliability/connection-failures.test.ts b/test/js/valkey/reliability/connection-failures.test.ts index 16ffa070f9f6..6e302acdf79a 100644 --- a/test/js/valkey/reliability/connection-failures.test.ts +++ b/test/js/valkey/reliability/connection-failures.test.ts @@ -1145,6 +1145,47 @@ describe("Valkey: Recovering After fail()", () => { } }); + 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(); + 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)[]) => From 3bf03fad1e8ae84f98a9d050065c23436d117275 Mon Sep 17 00:00:00 2001 From: Alistair Smith Date: Mon, 17 Aug 2026 22:28:07 -0700 Subject: [PATCH 21/21] test: run the deferred-close teardown tests on the ASAN lane and add a main-thread variant --- .../workers/worker-terminate-lifetime.test.ts | 37 +++++++++++++++++-- 1 file changed, 34 insertions(+), 3 deletions(-) diff --git a/test/js/web/workers/worker-terminate-lifetime.test.ts b/test/js/web/workers/worker-terminate-lifetime.test.ts index c5e0e659feb4..1f3cd6f81934 100644 --- a/test/js/web/workers/worker-terminate-lifetime.test.ts +++ b/test/js/web/workers/worker-terminate-lifetime.test.ts @@ -788,8 +788,9 @@ test.skipIf(!isDebug)( // 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). -test.skipIf(!isDebug)( +// 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( @@ -800,12 +801,42 @@ test.skipIf(!isDebug)( // 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 || isWindows)( +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 () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + 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, + 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: "connect rejected ERR_REDIS_CONNECTION_CLOSED\n", + stderr: "", + exitCode: 0, + }); + }, + timeout, +); + // worker.terminate() never stopped a worker parked in // Atomics.wait() (sync-over-async worker pools park exactly there). JSC wakes // the parked thread when termination is requested, but the wake-up predicate