diff --git a/docs/runtime/redis.mdx b/docs/runtime/redis.mdx index 19c870ed6230..889730de774f 100644 --- a/docs/runtime/redis.mdx +++ b/docs/runtime/redis.mdx @@ -481,6 +481,7 @@ Common error codes: - `ERR_REDIS_CONNECTION_CLOSED` - Connection to the server was closed - `ERR_REDIS_AUTHENTICATION_FAILED` - Failed to authenticate with the server - `ERR_REDIS_INVALID_RESPONSE` - Received an invalid response from the server +- `ERR_REDIS_SERVER_ERROR` - The server sent an error reply, either rejecting one command or, in subscriber mode, closing the connection --- diff --git a/packages/bun-types/redis.d.ts b/packages/bun-types/redis.d.ts index 03279e9e82e9..79b394fd2fb2 100644 --- a/packages/bun-types/redis.d.ts +++ b/packages/bun-types/redis.d.ts @@ -122,7 +122,7 @@ declare module "bun" { * - An array becomes an array. * - A set becomes an array. * - A map becomes a plain object with a null prototype. - * - An error reply (`-` or `!`) rejects the promise with code ERR_REDIS_INVALID_RESPONSE. + * - An error reply (`-` or `!`) rejects the promise with code ERR_REDIS_SERVER_ERROR. */ send(command: string, args: string[]): Promise; diff --git a/src/jsc/bindings/ErrorCode.ts b/src/jsc/bindings/ErrorCode.ts index d1ce4e83dc24..10d1e80f40f9 100644 --- a/src/jsc/bindings/ErrorCode.ts +++ b/src/jsc/bindings/ErrorCode.ts @@ -378,5 +378,6 @@ const errors: ErrorCodeMapping = [ ["ERR_INSPECTOR_NOT_CONNECTED", Error], ["ERR_INSPECTOR_NOT_WORKER", Error], ["ERR_INSPECTOR_COMMAND", Error], + ["ERR_REDIS_SERVER_ERROR", Error, "RedisError"], ]; export default errors; diff --git a/src/runtime/valkey_jsc/protocol_jsc.rs b/src/runtime/valkey_jsc/protocol_jsc.rs index d8c536c8319d..2db9e9f8bf02 100644 --- a/src/runtime/valkey_jsc/protocol_jsc.rs +++ b/src/runtime/valkey_jsc/protocol_jsc.rs @@ -43,6 +43,7 @@ pub(crate) fn valkey_error_to_js( RedisError::IdleTimeout => JscError::REDIS_IDLE_TIMEOUT, RedisError::NestingDepthExceeded => JscError::REDIS_INVALID_RESPONSE, RedisError::LineTooLong => JscError::REDIS_INVALID_RESPONSE, + RedisError::ServerError => JscError::REDIS_SERVER_ERROR, RedisError::JSError => return global.take_exception(JsError::Thrown), RedisError::OutOfMemory => { let _ = global.throw_out_of_memory(); @@ -91,11 +92,7 @@ pub(crate) fn resp_value_to_js_with_options( ) -> JsResult { match this { RESPValue::SimpleString(str) => valkey_str_to_js_value(global, str, options), - RESPValue::Error(str) => Ok(valkey_error_to_js( - global, - &**str, - RedisError::InvalidResponse, - )), + RESPValue::Error(str) => Ok(valkey_error_to_js(global, &**str, RedisError::ServerError)), RESPValue::Integer(int) => Ok(JSValue::js_number(*int as f64)), RESPValue::BulkString(maybe_str) => { if let Some(str) = maybe_str { diff --git a/src/runtime/valkey_jsc/valkey.rs b/src/runtime/valkey_jsc/valkey.rs index e06e298b327a..144f407482dd 100644 --- a/src/runtime/valkey_jsc/valkey.rs +++ b/src/runtime/valkey_jsc/valkey.rs @@ -1161,7 +1161,7 @@ impl ValkeyClient { match value { RESPValue::Error(err) => { if self.parent().is_subscriber() { - self.fail(err, RedisError::InvalidResponse)?; + self.fail(err, RedisError::ServerError)?; return Ok(()); } // A raw subscription request from a client that is not (yet) a diff --git a/src/valkey/valkey_protocol.rs b/src/valkey/valkey_protocol.rs index 2c714c5d67d4..61b3d6833775 100644 --- a/src/valkey/valkey_protocol.rs +++ b/src/valkey/valkey_protocol.rs @@ -32,6 +32,8 @@ pub enum RedisError { IdleTimeout, NestingDepthExceeded, LineTooLong, + /// The server answered with a `-` or `!` error reply. + ServerError, } bun_core::impl_tag_error!(RedisError); diff --git a/test/js/valkey/valkey-incremental-scan.test.ts b/test/js/valkey/valkey-incremental-scan.test.ts index f765c56f5e1b..0df4fb497d90 100644 --- a/test/js/valkey/valkey-incremental-scan.test.ts +++ b/test/js/valkey/valkey-incremental-scan.test.ts @@ -1,4 +1,4 @@ -import { RedisClient, type TCPSocketListener } from "bun"; +import { RedisClient, type Socket, type TCPSocketListener } from "bun"; import { describe, expect, test } from "bun:test"; import net from "node:net"; @@ -11,16 +11,11 @@ const HELLO = type PerSocket = { buf: Buffer; replied: boolean }; /** - * Mock server: answers HELLO, then answers the first GET with `reply`. When - * `splitAt` is inside the reply it is split there across two event-loop turns - * so the client's empty-read-buffer stack path sees a partial frame; "bytes" - * sends one byte per turn so the reply scanner resumes at every offset. - * Subsequent commands get `+OK`. + * Mock server: parses the client's RESP command frames + * (`*N\r\n($len\r\n...\r\n){N}`) and hands each complete one to `onCommand`. */ -function createReplyServer( - reply: string, - splitAt: number | "bytes" = reply.length, - hello: string = HELLO, +function createCommandServer( + onCommand: (fields: string[], s: Socket) => void, ): TCPSocketListener { return Bun.listen({ hostname: "127.0.0.1", @@ -34,7 +29,6 @@ function createReplyServer( data(s, raw) { const st = s.data; st.buf = Buffer.concat([st.buf, raw]); - // Parse complete client RESP command frames (`*N\r\n($len\r\n...\r\n){N}`). for (;;) { const b = st.buf; if (!b.length || b[0] !== 0x2a) break; @@ -61,38 +55,55 @@ function createReplyServer( } if (!complete) break; st.buf = b.subarray(pos); - const cmd = fields[0]?.toUpperCase(); - if (cmd === "HELLO") { - s.write(hello); - } else if (cmd === "GET" && !st.replied) { - st.replied = true; - if (splitAt === "bytes") { - const bytes = Buffer.from(reply, "latin1"); - const writeByte = (i: number) => { - if (i >= bytes.length) return; - s.write(bytes.subarray(i, i + 1)); - s.flush(); - setImmediate(() => setImmediate(() => writeByte(i + 1))); - }; - writeByte(0); - } else { - s.write(reply.slice(0, splitAt)); - s.flush(); - if (splitAt < reply.length) { - // Yield twice so the first write reaches the client's `on_data` - // before the second is sent. - setImmediate(() => setImmediate(() => s.write(reply.slice(splitAt)))); - } - } - } else { - s.write(`+OK${CRLF}`); - } + onCommand(fields, s); } }, }, }); } +/** + * Mock server: answers HELLO, then answers the first GET with `reply`. When + * `splitAt` is inside the reply it is split there across two event-loop turns + * so the client's empty-read-buffer stack path sees a partial frame; "bytes" + * sends one byte per turn so the reply scanner resumes at every offset. + * Subsequent commands get `+OK`. + */ +function createReplyServer( + reply: string, + splitAt: number | "bytes" = reply.length, + hello: string = HELLO, +): TCPSocketListener { + return createCommandServer((fields, s) => { + const cmd = fields[0]?.toUpperCase(); + if (cmd === "HELLO") { + s.write(hello); + } else if (cmd === "GET" && !s.data.replied) { + s.data.replied = true; + if (splitAt === "bytes") { + const bytes = Buffer.from(reply, "latin1"); + const writeByte = (i: number) => { + if (i >= bytes.length) return; + s.write(bytes.subarray(i, i + 1)); + s.flush(); + setImmediate(() => setImmediate(() => writeByte(i + 1))); + }; + writeByte(0); + } else { + s.write(reply.slice(0, splitAt)); + s.flush(); + if (splitAt < reply.length) { + // Yield twice so the first write reaches the client's `on_data` + // before the second is sent. + setImmediate(() => setImmediate(() => s.write(reply.slice(splitAt)))); + } + } + } else { + s.write(`+OK${CRLF}`); + } + }); +} + async function withClient(server: TCPSocketListener, body: (client: RedisClient) => Promise) { const client = new RedisClient(`redis://127.0.0.1:${server.port}`, { autoReconnect: false }); client.onconnect = client.onclose = () => {}; @@ -127,12 +138,12 @@ const FRAMES: [name: string, frame: string, expected: Decoded][] = [ [ "simple error (-ERR)", `-ERR unknown command${CRLF}`, - { rejects: { code: "ERR_REDIS_INVALID_RESPONSE", message: "ERR unknown command" } }, + { rejects: { code: "ERR_REDIS_SERVER_ERROR", message: "ERR unknown command" } }, ], [ "blob error (!)", `!21${CRLF}SYNTAX invalid syntax${CRLF}`, - { rejects: { code: "ERR_REDIS_INVALID_RESPONSE", message: "SYNTAX invalid syntax" } }, + { rejects: { code: "ERR_REDIS_SERVER_ERROR", message: "SYNTAX invalid syntax" } }, ], ]; @@ -201,16 +212,51 @@ describe.concurrent("Valkey reply decoding", () => { test.each([ ["-", `-WRONGTYPE wrong kind${CRLF}`], ["!", `!20${CRLF}WRONGTYPE wrong kind${CRLF}`], - ])("error reply (%s) nested in an array resolves as an Error element", async (_kind, element) => { + ])("error reply (%s) nested in an array resolves as an ERR_REDIS_SERVER_ERROR element", async (_kind, element) => { const server = createReplyServer(`*2${CRLF}+OK${CRLF}${element}`); await withClient(server, async client => { - const result = (await client.get("k")) as unknown as [string, Error]; + const result = (await client.get("k")) as unknown as [string, Error & { code: string }]; expect(result[0]).toBe("OK"); expect(result[1]).toBeInstanceOf(Error); - expect(result[1].message).toBe("WRONGTYPE wrong kind"); + expect({ code: result[1].code, message: result[1].message }).toEqual({ + code: "ERR_REDIS_SERVER_ERROR", + message: "WRONGTYPE wrong kind", + }); expect(await client.send("PING", [])).toBe("OK"); }); }); + + test("error reply in subscriber mode fails the connection with ERR_REDIS_SERVER_ERROR", async () => { + const server = createCommandServer((fields, s) => { + switch (fields[0]?.toUpperCase()) { + case "HELLO": + s.write(HELLO); + break; + case "SUBSCRIBE": + s.write(`>3${CRLF}` + bulk("subscribe") + bulk(fields[1]) + `:1${CRLF}`); + break; + default: + s.write(`-NOPERM no permissions${CRLF}`); + } + }); + await withClient(server, async client => { + await client.subscribe("ch", () => {}); + // A subscriber fails the whole connection on an error reply. The PING + // that drew the reply is left unsettled (#32858 changes that), so the + // code is read from the second PING, which is still in flight when the + // connection fails. + client.send("PING", []).catch(() => {}); + const err = await client.send("PING", []).then( + () => null, + e => e, + ); + expect(err).toBeInstanceOf(Error); + expect({ code: err.code, message: err.message }).toEqual({ + code: "ERR_REDIS_SERVER_ERROR", + message: "NOPERM no permissions", + }); + }); + }); }); describe.concurrent("Valkey reply torn across socket reads", () => { @@ -250,7 +296,7 @@ describe.concurrent("Valkey reply torn across socket reads", () => { e => e, ); expect(err).toBeInstanceOf(Error); - expect(err.code).toBe("ERR_REDIS_INVALID_RESPONSE"); + expect(err.code).toBe("ERR_REDIS_SERVER_ERROR"); expect(err.message).toBe("SYNTAX invalid syntax"); expect(await client.send("PING", [])).toBe("OK"); });