diff --git a/docs/runtime/redis.mdx b/docs/runtime/redis.mdx index 653cb9f3c9db..19c870ed6230 100644 --- a/docs/runtime/redis.mdx +++ b/docs/runtime/redis.mdx @@ -340,8 +340,9 @@ The client automatically converts Redis responses to JavaScript values: - Integer responses are returned as JavaScript numbers - Bulk strings are returned as JavaScript strings - Simple strings are returned as JavaScript strings -- Null bulk strings are returned as `null` +- Null bulk strings and null arrays are returned as `null` - Array responses are returned as JavaScript arrays +- Big number responses (RESP3) are returned as `BigInt`. A payload that is not an integer literal is returned as a string. `getBuffer` returns the payload as a `Buffer`. - Error responses throw JavaScript errors with appropriate error codes - Boolean responses (RESP3) are returned as JavaScript booleans - Map responses (RESP3) are returned as JavaScript objects diff --git a/packages/bun-types/redis.d.ts b/packages/bun-types/redis.d.ts index f254de647b23..03279e9e82e9 100644 --- a/packages/bun-types/redis.d.ts +++ b/packages/bun-types/redis.d.ts @@ -110,6 +110,19 @@ declare module "bun" { * @param command The command to send * @param args The arguments to the command * @returns A promise that resolves with the command result + * + * The reply is converted as follows. See the type conversion section in + * docs/runtime/redis.mdx. + * - A simple, bulk or verbatim string becomes a string. Methods that return a Buffer, such as getBuffer, keep the bytes. + * - An integer becomes a number. + * - A double becomes a number. + * - A big number becomes a bigint. When its payload is not an integer literal it becomes a string. + * - A boolean becomes a boolean. + * - A null, a null bulk string and a null array become null. + * - 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. */ send(command: string, args: string[]): Promise; @@ -3536,6 +3549,8 @@ declare module "bun" { * await redis.eval("return ARGV[1]", 0, "hello"); // "hello" * await redis.eval("return redis.call('GET', KEYS[1])", 1, "mykey"); * ``` + * + * The reply is converted the same way as for `send`. */ eval(script: string, numkeys: number, ...keysAndArgs: (string | number)[]): Promise; @@ -3551,6 +3566,8 @@ declare module "bun" { * const sha = await redis.script("LOAD", "return ARGV[1]"); * await redis.evalsha(sha, 0, "hello"); // "hello" * ``` + * + * The reply is converted the same way as for `send`. */ evalsha(sha1: string, numkeys: number, ...keysAndArgs: (string | number)[]): Promise; @@ -3574,6 +3591,8 @@ declare module "bun" { * @param numkeys The number of keys * @param keysAndArgs The keys followed by additional arguments * @returns Promise that resolves with the function's return value + * + * The reply is converted the same way as for `send`. */ fcall(name: string, numkeys: number, ...keysAndArgs: (string | number)[]): Promise; diff --git a/src/jsc/JSValue.rs b/src/jsc/JSValue.rs index e41e0015940c..f4e08f90c79c 100644 --- a/src/jsc/JSValue.rs +++ b/src/jsc/JSValue.rs @@ -666,6 +666,24 @@ impl JSValue { pub fn from_uint64_no_truncate(global: &JSGlobalObject, i: u64) -> JsResult { host_fn::from_js_host_call(global, || JSC__JSValue__fromUInt64NoTruncate(global, i)) } + /// A BigInt from a decimal integer literal (optional `-`, then digits). + /// Returns `Ok(None)` when `digits` is not such a literal. + #[track_caller] + pub fn big_int_from_decimal( + global: &JSGlobalObject, + digits: &[u8], + ) -> JsResult> { + let unsigned = digits.strip_prefix(b"-").unwrap_or(digits); + if unsigned.is_empty() || !unsigned.iter().all(u8::is_ascii_digit) { + return Ok(None); + } + // Only text StringToBigInt accepts gets here, so empty means it threw (too large). + let value = host_fn::from_js_host_call(global, || { + // SAFETY: `digits` is a live slice for the duration of the call. + unsafe { JSC__JSValue__bigIntFromLatin1(global, digits.as_ptr(), digits.len()) } + })?; + Ok(Some(value)) + } /// `JSValue.fromTimevalNoTruncate` — encode a `struct timeval` /// as a BigInt (`sec * 1_000_000 + nsec`) without precision loss. May allocate /// a heap BigInt, so wrapped in `from_js_host_call` for exception checking. @@ -1984,6 +2002,11 @@ unsafe extern "C" { safe fn JSC__JSValue__dateInstanceFromNumber(global: &JSGlobalObject, n: f64) -> JSValue; safe fn JSC__JSValue__fromInt64NoTruncate(global: &JSGlobalObject, i: i64) -> JSValue; safe fn JSC__JSValue__fromUInt64NoTruncate(global: &JSGlobalObject, i: u64) -> JSValue; + fn JSC__JSValue__bigIntFromLatin1( + global: &JSGlobalObject, + ptr: *const u8, + len: usize, + ) -> JSValue; safe fn JSC__JSValue__fromTimevalNoTruncate( global: &JSGlobalObject, nsec: i64, diff --git a/src/jsc/bindings/bindings.cpp b/src/jsc/bindings/bindings.cpp index 248c256d432a..147f9a5ec434 100644 --- a/src/jsc/bindings/bindings.cpp +++ b/src/jsc/bindings/bindings.cpp @@ -4487,6 +4487,13 @@ JSC::EncodedJSValue JSC__JSValue__fromUInt64NoTruncate(JSC::JSGlobalObject* glob return JSC::JSValue::encode(JSC::JSBigInt::createFrom(globalObject, val)); } +// Decimal integer literal (Latin-1) -> BigInt. Returns the empty value when +// the text is not a valid StringToBigInt input. +JSC::EncodedJSValue JSC__JSValue__bigIntFromLatin1(JSC::JSGlobalObject* globalObject, const uint8_t* ptr, size_t len) +{ + return JSC::JSValue::encode(JSC::JSBigInt::stringToBigInt(globalObject, WTF::StringView(std::span { reinterpret_cast(ptr), len }))); +} + uint64_t JSC__JSValue__toUInt64NoTruncate(JSC::EncodedJSValue val) { JSC::JSValue value = JSC::JSValue::decode(val); diff --git a/src/jsc/bindings/headers.h b/src/jsc/bindings/headers.h index 13b241a473b1..8a06f69c54c7 100644 --- a/src/jsc/bindings/headers.h +++ b/src/jsc/bindings/headers.h @@ -225,6 +225,7 @@ CPP_DECL void JSC__JSValue__forEachPropertyOrdered(JSC::EncodedJSValue JSValue0, CPP_DECL JSC::EncodedJSValue JSC__JSValue__fromEntries(JSC::JSGlobalObject* arg0, ZigString* arg1, ZigString* arg2, size_t arg3, bool arg4); CPP_DECL JSC::EncodedJSValue JSC__JSValue__fromInt64NoTruncate(JSC::JSGlobalObject* arg0, int64_t arg1); CPP_DECL JSC::EncodedJSValue JSC__JSValue__fromUInt64NoTruncate(JSC::JSGlobalObject* arg0, uint64_t arg1); +CPP_DECL JSC::EncodedJSValue JSC__JSValue__bigIntFromLatin1(JSC::JSGlobalObject* arg0, const uint8_t* arg1, size_t arg2); CPP_DECL JSC::EncodedJSValue JSC__JSValue__fromTimevalNoTruncate(JSC::JSGlobalObject* arg0, int64_t nsec, int64_t sec); CPP_DECL JSC::EncodedJSValue JSC__JSValue__bigIntSum(JSC::JSGlobalObject* arg0, JSC::EncodedJSValue JSValue0, JSC::EncodedJSValue JSValue1); CPP_DECL void JSC__JSValue__getClassName(JSC::EncodedJSValue JSValue0, JSC::JSGlobalObject* arg1, ZigString* arg2); diff --git a/src/runtime/valkey_jsc/protocol_jsc.rs b/src/runtime/valkey_jsc/protocol_jsc.rs index a936ad0cf7ed..d8c536c8319d 100644 --- a/src/runtime/valkey_jsc/protocol_jsc.rs +++ b/src/runtime/valkey_jsc/protocol_jsc.rs @@ -112,11 +112,6 @@ pub(crate) fn resp_value_to_js_with_options( RESPValue::Null => Ok(JSValue::NULL), RESPValue::Double(d) => Ok(JSValue::js_number(*d)), RESPValue::Boolean(b) => Ok(JSValue::from(*b)), - RESPValue::BlobError(str) => Ok(valkey_error_to_js( - global, - &**str, - RedisError::InvalidBlobError, - )), RESPValue::VerbatimString(verbatim) => { valkey_str_to_js_value(global, &mut verbatim.content, options) } @@ -158,14 +153,15 @@ pub(crate) fn resp_value_to_js_with_options( Ok(js_obj) } + // BigInt when the payload is an integer literal; modules and Lua can + // put anything after `(`, so other text stays a string. RESPValue::BigNumber(str) => { - // Try to parse as number if possible - if let Ok(int) = bun_core::fmt::parse_int::(str, 10) { - Ok(JSValue::js_number(int as f64)) - } else { - // If it doesn't fit in an i64, return as string - bun_string_jsc::create_utf8_for_js(global, str) + if !options.return_as_buffer + && let Some(big) = JSValue::big_int_from_decimal(global, str)? + { + return Ok(big); } + valkey_str_to_js_value(global, str, options) } } } diff --git a/src/valkey/valkey_protocol.rs b/src/valkey/valkey_protocol.rs index 1fe52ef4d5ee..2c714c5d67d4 100644 --- a/src/valkey/valkey_protocol.rs +++ b/src/valkey/valkey_protocol.rs @@ -88,6 +88,7 @@ impl RESPType { pub enum RESPValue { // RESP2 types SimpleString(Box<[u8]>), + /// A `-` simple error or a `!` blob error reply, holding the server's message. Error(Box<[u8]>), Integer(i64), BulkString(Option>), @@ -97,7 +98,6 @@ pub enum RESPValue { Null, Double(f64), Boolean(bool), - BlobError(Box<[u8]>), VerbatimString(VerbatimString), Map(Vec), Set(Vec), @@ -134,7 +134,6 @@ impl fmt::Display for RESPValue { RESPValue::Null => writer.write_str("(nil)"), RESPValue::Double(d) => write!(writer, "{}", d), RESPValue::Boolean(b) => write!(writer, "{}", b), - RESPValue::BlobError(str) => write!(writer, "Error: {}", BStr::new(str)), RESPValue::VerbatimString(verbatim) => { write!( writer, @@ -398,7 +397,8 @@ impl<'a> ValkeyReader<'a> { } let len = self.read_integer()?; if len < 0 { - return Ok(RESPValue::Array(Vec::new())); + // RESP2 null array. + return Ok(RESPValue::Null); } let len = usize::try_from(len).expect("int cast"); let mut array = @@ -414,7 +414,9 @@ impl<'a> ValkeyReader<'a> { // RESP3 types RESPType::Null => { - let _ = self.read_until_crlf()?; // Read and discard CRLF + if !self.read_until_crlf()?.is_empty() { + return Err(RedisError::InvalidNull); + } Ok(RESPValue::Null) } RESPType::Double => { @@ -441,7 +443,7 @@ impl<'a> ValkeyReader<'a> { return Err(RedisError::InvalidBlobError); } let owned = Box::<[u8]>::from(str); - Ok(RESPValue::BlobError(owned)) + Ok(RESPValue::Error(owned)) } RESPType::VerbatimString => Ok(RESPValue::VerbatimString(self.read_verbatim_string()?)), RESPType::Map => { @@ -668,13 +670,18 @@ impl ReplyScanner { RESPType::SimpleString | RESPType::Error | RESPType::Integer - | RESPType::Null | RESPType::Double | RESPType::Boolean | RESPType::BigNumber => { let _ = reader.read_until_crlf()?; Ok(None) } + RESPType::Null => { + if !reader.read_until_crlf()?.is_empty() { + return Err(RedisError::InvalidNull); + } + Ok(None) + } RESPType::BulkString | RESPType::BlobError | RESPType::VerbatimString => { let invalid = match ty { RESPType::BlobError => RedisError::InvalidBlobError, diff --git a/test/js/valkey/valkey-incremental-scan.test.ts b/test/js/valkey/valkey-incremental-scan.test.ts index 4cd4f306be9e..f765c56f5e1b 100644 --- a/test/js/valkey/valkey-incremental-scan.test.ts +++ b/test/js/valkey/valkey-incremental-scan.test.ts @@ -2,91 +2,218 @@ import { RedisClient, type TCPSocketListener } from "bun"; import { describe, expect, test } from "bun:test"; import net from "node:net"; -describe.concurrent("Valkey reply torn across socket reads", () => { - const CRLF = "\r\n"; - const bulk = (s: string) => `$${Buffer.byteLength(s)}${CRLF}${s}${CRLF}`; - // Minimal RESP3 HELLO map so the client enters the Connected state. - const HELLO = - `%3${CRLF}` + bulk("server") + bulk("redis") + bulk("proto") + `:3${CRLF}` + bulk("version") + bulk("7.4.0"); +const CRLF = "\r\n"; +const bulk = (s: string) => `$${Buffer.byteLength(s)}${CRLF}${s}${CRLF}`; +// Minimal RESP3 HELLO map so the client enters the Connected state. +const HELLO = + `%3${CRLF}` + bulk("server") + bulk("redis") + bulk("proto") + `:3${CRLF}` + bulk("version") + bulk("7.4.0"); - type PerSocket = { buf: Buffer; replied: boolean }; +type PerSocket = { buf: Buffer; replied: boolean }; - /** - * Mock server: answers HELLO, then answers the first GET with `reply` split at - * `splitAt` across two event-loop turns so the client's empty-read-buffer - * stack path sees a partial blob body. Subsequent commands get `+OK`. - */ - function createTornReplyServer(reply: string, splitAt: number): TCPSocketListener { - return Bun.listen({ - hostname: "127.0.0.1", - port: 0, - socket: { - open(s) { - s.data = { buf: Buffer.alloc(0), replied: false }; - }, - error() {}, - close() {}, - 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; - const headerEnd = b.indexOf(CRLF); - if (headerEnd < 0) break; - const argc = parseInt(b.subarray(1, headerEnd).toString("latin1"), 10); - let pos = headerEnd + 2; - const fields: string[] = []; - let complete = true; - for (let i = 0; i < argc; i++) { - const lenEnd = b.indexOf(CRLF, pos); - if (lenEnd < 0 || b[pos] !== 0x24) { - complete = false; - break; - } - const len = parseInt(b.subarray(pos + 1, lenEnd).toString("latin1"), 10); - const next = lenEnd + 2 + len + 2; - if (next > b.length) { - complete = false; - break; - } - fields.push(b.subarray(lenEnd + 2, lenEnd + 2 + len).toString("latin1")); - pos = next; +/** + * 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 Bun.listen({ + hostname: "127.0.0.1", + port: 0, + socket: { + open(s) { + s.data = { buf: Buffer.alloc(0), replied: false }; + }, + error() {}, + close() {}, + 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; + const headerEnd = b.indexOf(CRLF); + if (headerEnd < 0) break; + const argc = parseInt(b.subarray(1, headerEnd).toString("latin1"), 10); + let pos = headerEnd + 2; + const fields: string[] = []; + let complete = true; + for (let i = 0; i < argc; i++) { + const lenEnd = b.indexOf(CRLF, pos); + if (lenEnd < 0 || b[pos] !== 0x24) { + complete = false; + break; + } + const len = parseInt(b.subarray(pos + 1, lenEnd).toString("latin1"), 10); + const next = lenEnd + 2 + len + 2; + if (next > b.length) { + complete = false; + break; } - 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; + fields.push(b.subarray(lenEnd + 2, lenEnd + 2 + len).toString("latin1")); + pos = next; + } + 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(); - // 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}`); + 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 = () => {}; + try { + await client.connect(); + return await body(client); + } finally { + client.close(); + server.stop(true); } +} + +type Decoded = { value: unknown } | { rejects: { code: string; message?: string }; connectionFails?: boolean }; + +// One entry per RESP frame shape the decoder changed. Each is sent whole and +// one byte per socket read, so both the tree parser and the reply scanner see +// every torn prefix. +const FRAMES: [name: string, frame: string, expected: Decoded][] = [ + ["RESP2 null array (*-1)", `*-1${CRLF}`, { value: null }], + ["RESP2 null array nested in an array", `*2${CRLF}*-1${CRLF}$3${CRLF}abc${CRLF}`, { value: [null, "abc"] }], + ["RESP2 null bulk string ($-1)", `$-1${CRLF}`, { value: null }], + ["RESP3 null (_)", `_${CRLF}`, { value: null }], + [ + "RESP3 null with trailing bytes (_junk)", + `_junk${CRLF}`, + { rejects: { code: "ERR_REDIS_INVALID_RESPONSE" }, connectionFails: true }, + ], + ["big number above 2^53", `(9007199254740993${CRLF}`, { value: 9007199254740993n }], + ["negative big number", `(-42${CRLF}`, { value: -42n }], + ["big number above 2^64", `(340282366920938463463374607431768211456${CRLF}`, { value: 2n ** 128n }], + ["big number with a non-integer payload", `(12abc${CRLF}`, { value: "12abc" }], + [ + "simple error (-ERR)", + `-ERR unknown command${CRLF}`, + { rejects: { code: "ERR_REDIS_INVALID_RESPONSE", message: "ERR unknown command" } }, + ], + [ + "blob error (!)", + `!21${CRLF}SYNTAX invalid syntax${CRLF}`, + { rejects: { code: "ERR_REDIS_INVALID_RESPONSE", message: "SYNTAX invalid syntax" } }, + ], +]; + +describe.concurrent.each([ + ["whole", (reply: string) => createReplyServer(reply)], + ["one byte per read", (reply: string) => createReplyServer(reply, "bytes")], +])("Valkey reply decoding, frame sent %s", (_mode, serve) => { + test.each(FRAMES)("%s", async (_name, frame, expected) => { + await withClient(serve(frame), async client => { + const outcome = await client.get("k").then( + value => ({ value }), + error => ({ error }), + ); + if ("value" in expected) { + expect(outcome).toEqual({ value: expected.value }); + expect(typeof (outcome as { value: unknown }).value).toBe(typeof expected.value); + } else { + expect(outcome).toHaveProperty("error"); + const { error } = outcome as { error: Error & { code: string } }; + expect(error).toBeInstanceOf(Error); + expect(error.code).toBe(expected.rejects.code); + if (expected.rejects.message !== undefined) expect(error.message).toBe(expected.rejects.message); + } + if (!("connectionFails" in expected && expected.connectionFails)) { + expect(await client.send("PING", [])).toBe("OK"); + } + }); + }); +}); + +describe.concurrent("Valkey reply decoding", () => { + test("big number resolves a Buffer of the digits for getBuffer", async () => { + const server = createReplyServer(`(9007199254740993${CRLF}`); + await withClient(server, async client => { + const value = await client.getBuffer("k"); + expect(value).toBeInstanceOf(Buffer); + expect(value!.toString()).toBe("9007199254740993"); + }); + }); - async function withClient(server: TCPSocketListener, body: (client: RedisClient) => Promise) { + test.each([ + ["-", `-NOAUTH nope!${CRLF}`], + ["!", `!12${CRLF}NOAUTH nope!${CRLF}`], + ])("error reply (%s) to HELLO rejects queued commands with the server text", async (_kind, hello) => { + const server = createReplyServer(`+OK${CRLF}`, undefined, hello); const client = new RedisClient(`redis://127.0.0.1:${server.port}`, { autoReconnect: false }); client.onconnect = client.onclose = () => {}; try { - await client.connect(); - return await body(client); + // Queued before the handshake finishes, so the rejection carries the + // HELLO error. connect() itself rejects with a generic "Connection closed". + const queued = client.get("k").then( + () => null, + e => e, + ); + await client.connect().catch(() => {}); + const err = await queued; + expect(err).toBeInstanceOf(Error); + expect(err.code).toBe("ERR_REDIS_AUTHENTICATION_FAILED"); + expect(err.message).toBe("NOAUTH nope!"); } finally { client.close(); server.stop(true); } - } + }); + 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) => { + const server = createReplyServer(`*2${CRLF}+OK${CRLF}${element}`); + await withClient(server, async client => { + const result = (await client.get("k")) as unknown as [string, Error]; + expect(result[0]).toBe("OK"); + expect(result[1]).toBeInstanceOf(Error); + expect(result[1].message).toBe("WRONGTYPE wrong kind"); + expect(await client.send("PING", [])).toBe("OK"); + }); + }); +}); + +describe.concurrent("Valkey reply torn across socket reads", () => { // `$15`/`=15` frames: 5-byte header, 15-byte body at [5,20), trailing CRLF at [20,22). // `!21` frame: 5-byte header, 21-byte body at [5,26), trailing CRLF at [26,28). // Offsets cover: body start, mid-body, last body byte, and mid-CRLF. @@ -94,7 +221,7 @@ describe.concurrent("Valkey reply torn across socket reads", () => { const LONG_SPLITS = [5, 10, 25, 27] as const; test.each(SHORT_SPLITS)("BulkString ($) torn at byte %i decodes (baseline)", async splitAt => { - const server = createTornReplyServer(`$15${CRLF}xxx:Some string${CRLF}`, splitAt); + const server = createReplyServer(`$15${CRLF}xxx:Some string${CRLF}`, splitAt); await withClient(server, async client => { expect(await client.get("k")).toBe("xxx:Some string"); expect(await client.send("PING", [])).toBe("OK"); @@ -104,7 +231,7 @@ describe.concurrent("Valkey reply torn across socket reads", () => { test.each(SHORT_SPLITS)( "VerbatimString (=) torn at byte %i decodes instead of failing the connection", async splitAt => { - const server = createTornReplyServer(`=15${CRLF}txt:Some string${CRLF}`, splitAt); + const server = createReplyServer(`=15${CRLF}txt:Some string${CRLF}`, splitAt); await withClient(server, async client => { expect(await client.get("k")).toBe("Some string"); expect(await client.send("PING", [])).toBe("OK"); @@ -113,14 +240,18 @@ describe.concurrent("Valkey reply torn across socket reads", () => { ); test.each(LONG_SPLITS)("BlobError (!) torn at byte %i decodes instead of failing the connection", async splitAt => { - const server = createTornReplyServer(`!21${CRLF}SYNTAX invalid syntax${CRLF}`, splitAt); + const server = createReplyServer(`!21${CRLF}SYNTAX invalid syntax${CRLF}`, splitAt); await withClient(server, async client => { - // A parsed BlobError resolves (not rejects) with an Error carrying the - // server's message. Before the fix this rejected with - // "Failed to read data (stack path)" and killed the connection. - const result = await client.get("k"); - expect(result).toBeInstanceOf(Error); - expect((result as unknown as Error).message).toBe("SYNTAX invalid syntax"); + // A parsed BlobError rejects only this command with the server's + // message. Before the fix this rejected with "Failed to read data + // (stack path)" and killed the connection. + const err = await client.get("k").then( + () => null, + e => e, + ); + expect(err).toBeInstanceOf(Error); + expect(err.code).toBe("ERR_REDIS_INVALID_RESPONSE"); + expect(err.message).toBe("SYNTAX invalid syntax"); expect(await client.send("PING", [])).toBe("OK"); }); });