diff --git a/src/runtime/valkey_jsc/valkey.rs b/src/runtime/valkey_jsc/valkey.rs index b909bff16982..a75ba6ba9cd3 100644 --- a/src/runtime/valkey_jsc/valkey.rs +++ b/src/runtime/valkey_jsc/valkey.rs @@ -627,10 +627,15 @@ impl ValkeyClient { jsvalue, ); - if !self.connection_ready() { - self.flags.is_manually_closed = true; - self.close(); - } + // A failed client rejects every later command, so the socket has to go with it. + // Leaving it open strands the client in `failed && connected`: `onclose` never + // fires and `.connected` keeps reporting true. The failure is ours, not the + // socket's (bad handshake, a frame we cannot parse, an idle timeout), so + // reconnecting would just hit it again: mark it deliberate and let `connect()` + // revive the client. A socket the peer drops never reaches here, so + // `autoReconnect` still covers it. `close()` is idempotent. + self.flags.is_manually_closed = true; + self.close(); val } diff --git a/test/js/valkey/reliability/connection-teardown.test.ts b/test/js/valkey/reliability/connection-teardown.test.ts new file mode 100644 index 000000000000..c818f4fca333 --- /dev/null +++ b/test/js/valkey/reliability/connection-teardown.test.ts @@ -0,0 +1,171 @@ +import type { RedisOptions } from "bun"; +import { describe, expect, test } from "bun:test"; +import net from "net"; + +/** + * A connection-level failure must close the socket, so `on_close` gets to run the + * client's reconnect / `onclose` policy. Without that the client is stranded in + * `failed && connected`: every command rejects, but nothing ever tells the app. + */ + +const CRLF = "\r\n"; +const bulk = (value: string) => `$${Buffer.byteLength(value)}${CRLF}${value}${CRLF}`; +const HELLO_REPLY = `%3${CRLF}${bulk("server")}${bulk("redis")}${bulk("proto")}:3${CRLF}${bulk("version")}${bulk("7.4.0")}`; +/** `\x01` is not a RESP type byte, so the client must treat the reply as a protocol failure. */ +const UNPARSEABLE_REPLY = `\x01not-a-resp-type${CRLF}`; + +/** Pull every complete `*N\r\n$len\r\n…` command frame out of `buffer`. */ +function takeCommands(buffer: Buffer): { commands: string[][]; rest: Buffer } { + const commands: string[][] = []; + let rest = buffer; + while (rest.length > 0 && rest[0] === 0x2a /* '*' */) { + const headerEnd = rest.indexOf(CRLF); + if (headerEnd < 0) break; + const argCount = Number(rest.subarray(1, headerEnd).toString()); + if (!Number.isInteger(argCount) || argCount < 0) break; + + let pos = headerEnd + 2; + const argv: string[] = []; + let complete = true; + for (let i = 0; i < argCount; i++) { + if (pos >= rest.length || rest[pos] !== 0x24 /* '$' */) { + complete = false; + break; + } + const lengthEnd = rest.indexOf(CRLF, pos); + if (lengthEnd < 0) { + complete = false; + break; + } + const length = Number(rest.subarray(pos + 1, lengthEnd).toString()); + if (!Number.isInteger(length) || length < 0 || rest.length < lengthEnd + 2 + length + 2) { + complete = false; + break; + } + argv.push(rest.subarray(lengthEnd + 2, lengthEnd + 2 + length).toString()); + pos = lengthEnd + 2 + length + 2; + } + if (!complete) break; + + commands.push(argv); + rest = rest.subarray(pos); + } + return { commands, rest }; +} + +type MockServer = Disposable & { + port: number; + /** Resolves once a client connection has been torn down. */ + disconnected: Promise; +}; + +async function startMockValkey(respond: (socket: net.Socket, argv: string[]) => void) { + const { promise: disconnected, resolve: onDisconnected } = Promise.withResolvers(); + const sockets: net.Socket[] = []; + + const server = net.createServer(socket => { + sockets.push(socket); + let buffered = Buffer.alloc(0); + socket.on("data", chunk => { + buffered = buffered.length > 0 ? Buffer.concat([buffered, chunk]) : chunk; + const { commands, rest } = takeCommands(buffered); + buffered = rest; + for (const argv of commands) respond(socket, argv); + }); + socket.on("error", () => {}); + socket.on("close", () => onDisconnected()); + }); + + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", resolve); + }); + + return { + port: (server.address() as net.AddressInfo).port, + disconnected, + [Symbol.dispose]() { + for (const socket of sockets) socket.destroy(); + server.close(); + }, + } satisfies MockServer; +} + +function createClient(port: number, options: RedisOptions) { + return new Bun.RedisClient(`redis://127.0.0.1:${port}`, { connectionTimeout: 5_000, ...options }); +} + +describe("Valkey: a failed connection is torn down", () => { + test("a protocol error closes the socket and fires onclose", async () => { + using server = await startMockValkey((socket, argv) => { + if (argv[0].toUpperCase() === "HELLO") socket.write(HELLO_REPLY); + else socket.write(UNPARSEABLE_REPLY); + }); + const client = createClient(server.port, { autoReconnect: false }); + const { promise: closed, resolve: onClosed } = Promise.withResolvers(); + client.onclose = () => onClosed(); + + try { + await client.connect(); + const rejection = await client.ping().then( + () => null, + (thrown: any) => thrown, + ); + expect(rejection?.code).toBe("ERR_REDIS_INVALID_RESPONSE_TYPE"); + + await closed; + await server.disconnected; + expect(client.connected).toBe(false); + } finally { + client.close(); + } + }); + + // `onclose` only fires on the terminal branch of `on_close` — the reconnecting branch + // never calls it — so seeing it here is proof the client did not re-dial. That matters: + // a repeatable post-handshake failure would otherwise loop forever, because a successful + // HELLO resets `retry_attempts` and `maxRetries` is never reached. + test("a protocol error is terminal even with autoReconnect enabled, and connect() revives", async () => { + let handshakes = 0; + let pings = 0; + + using server = await startMockValkey((socket, argv) => { + switch (argv[0].toUpperCase()) { + case "HELLO": + handshakes++; + socket.write(HELLO_REPLY); + break; + case "PING": + // Break the first connection mid-stream, then answer normally. + socket.write(++pings === 1 ? UNPARSEABLE_REPLY : `+PONG${CRLF}`); + break; + default: + socket.write(`+OK${CRLF}`); + break; + } + }); + const client = createClient(server.port, { autoReconnect: true }); + const { promise: closed, resolve: onClosed } = Promise.withResolvers(); + client.onclose = () => onClosed(); + + try { + await client.connect(); + const rejection = await client.ping().then( + () => null, + (thrown: any) => thrown, + ); + expect(rejection?.code).toBe("ERR_REDIS_INVALID_RESPONSE_TYPE"); + + await closed; + await server.disconnected; + expect(client.connected).toBe(false); + + // The app is handed the failure and can re-dial on its own terms. + await client.connect(); + expect(await client.ping()).toBe("PONG"); + expect({ connected: client.connected, handshakes }).toEqual({ connected: true, handshakes: 2 }); + } finally { + client.close(); + } + }); +});