redis: raise the RESP line-length cap from 512KB to 512MB

src/valkey/valkey_protocol.rs

@@ -262,7 +262,6 @@ impl<'a> ValkeyReader<'a> {
         if buffer.len() > Self::MAX_LINE_LEN + 1 {
             return Err(RedisError::LineTooLong);
         }
-
         Err(RedisError::InvalidResponse)
     }
 
@@ -344,7 +343,10 @@ impl<'a> ValkeyReader<'a> {
     /// attacker-chosen size.
     const MAX_BULK_LEN: i64 = 512 * 1024 * 1024;
 
-    const MAX_LINE_LEN: usize = 512 * 1024;
+    /// Maximum accepted length for a CRLF-terminated RESP line (`+ - : _ , # (`).
+    /// Mirrors `MAX_BULK_LEN` so line-terminated replies get the same
+    /// buffer-growth bound as length-prefixed blobs; the spec places no limit.
+    const MAX_LINE_LEN: usize = Self::MAX_BULK_LEN as usize;
 
     /// Caps an aggregate's `Vec::with_capacity` so the total bytes reserved
     /// across the whole parse — every nesting level combined — never exceed

test/js/valkey/reliability/resp-nesting-depth.test.ts

@@ -239,6 +239,76 @@ describe("Valkey: RESP Nesting Depth Handling", () => {
   });
 });
 
+describe("Valkey: RESP line-terminated replies (>512KB)", () => {
+  // The Zig parser scanned for CRLF over the whole buffer. A later hardening
+  // pass capped the scan window at 512KB, which rejects valid RESP simple
+  // strings and error replies that a real server emits (Lua
+  // redis.status_reply(...) / redis.error_reply(...) with a long payload).
+  // https://redis.io/docs/latest/develop/reference/protocol-spec/ places no
+  // length limit on `+` / `-` lines.
+
+  async function roundtrip(payload: Buffer): Promise<any> {
+    const { server, port } = await createMockRedisServer([payload, Buffer.from("+PONG\r\n")]);
+    try {
+      const client = new Bun.RedisClient(`redis://127.0.0.1:${port}`, {
+        autoReconnect: false,
+        connectionTimeout: 5000,
+      });
+      try {
+        const reply = await client.send("PING", []);
+        // The client must still be usable for the next command: a parse
+        // failure here used to latch `failed` and reject forever.
+        const pong = await client.send("PING", []);
+        return { reply, pong };
+      } finally {
+        client.close();
+      }
+    } finally {
+      server.close();
+    }
+  }
+
+  test("accepts a simple string (`+`) reply longer than 512KB", async () => {
+    const body = Buffer.alloc(600_000, "x").toString();
+    const { reply, pong } = await roundtrip(Buffer.from(`+${body}\r\n`));
+    expect(typeof reply).toBe("string");
+    expect(reply.length).toBe(600_000);
+    expect(pong).toBe("PONG");
+  });
+
+  test("accepts a simple string at exactly 512KB + 1 bytes", async () => {
+    // 524288 worked, 524289 was rejected by an off-by-one in the scan window.
+    const body = Buffer.alloc(512 * 1024 + 1, "y").toString();
+    const { reply, pong } = await roundtrip(Buffer.from(`+${body}\r\n`));
+    expect(reply.length).toBe(512 * 1024 + 1);
+    expect(pong).toBe("PONG");
+  });
+
+  test("surfaces an error (`-`) reply longer than 512KB with its original text", async () => {
+    const body = "ERR user_script:1: " + Buffer.alloc(600_000, "e").toString();
+    const { server, port } = await createMockRedisServer([Buffer.from(`-${body}\r\n`), Buffer.from("+PONG\r\n")]);
+    try {
+      const client = new Bun.RedisClient(`redis://127.0.0.1:${port}`, {
+        autoReconnect: false,
+        connectionTimeout: 5000,
+      });
+      try {
+        const rejection = await client.send("EVAL", ["script", "0"]).then(
+          () => null,
+          (e: any) => e,
+        );
+        expect(rejection?.message).toBe(body);
+        // Client must survive and serve the next command.
+        expect(await client.send("PING", [])).toBe("PONG");
+      } finally {
+        client.close();
+      }
+    } finally {
+      server.close();
+    }
+  });
+});
+
 describe("Valkey: RESP push frame routing", () => {
   test("resolves a pending command with its own reply when an out-of-band push frame precedes it", async () => {
     const payload = Buffer.from(

test/js/valkey/valkey-gc.test.ts

@@ -132,10 +132,9 @@ test.concurrent("RedisClient survives GC across many short-lived instances", asy
 });
 
 // A RESP scalar line (simple string, error, integer, ...) must end with CRLF.
-// The reader bounds how many bytes it will accumulate while waiting for that
-// terminator (MAX_LINE_LEN = 512 KiB), so a server that streams an endless
-// unterminated line gets a protocol error promptly instead of the client
-// buffering and rescanning the whole line on every socket read.
+// The reader caps line-terminated replies at MAX_BULK_LEN (512 MB), so a 600 KB
+// unterminated line is treated as a partial reply; when the server closes
+// mid-line the pending command is rejected as connection-closed.
 test.concurrent("rejects a RESP simple-string reply whose line terminator never arrives", async () => {
   // Minimal mock Redis server: replies +OK to the HELLO handshake, then
   // answers the next command with `payload`.
@@ -161,12 +160,9 @@ test.concurrent("rejects a RESP simple-string reply whose line terminator never
     });
   }
 
-  // 1) A simple-string reply whose CRLF terminator never arrives. Once more
-  //    than 512 KiB of the line has accumulated, the client must fail the
-  //    reply with a protocol error rather than keep waiting for a terminator
-  //    that never comes. (The server closes the socket after the payload so
-  //    that a client which keeps waiting still settles the promise -- with a
-  //    connection-closed error instead of the expected protocol error.)
+  // 1) A simple-string reply whose CRLF terminator never arrives. The reader
+  //    treats the unterminated bytes as a partial reply and keeps waiting; when
+  //    the server closes, the pending command is rejected as connection-closed.
   {
     const unterminated = Buffer.from("+" + Buffer.alloc(600_000, "A").toString());
     const { server, port } = await listen(unterminated, true);
@@ -179,7 +175,7 @@ test.concurrent("rejects a RESP simple-string reply whose line terminator never
         await client.send("PING", []);
         expect.unreachable();
       } catch (error: any) {
-        expect(error.code).toBe("ERR_REDIS_INVALID_RESPONSE");
+        expect(error.code).toBe("ERR_REDIS_CONNECTION_CLOSED");
       } finally {
         client.close();
       }
@@ -188,8 +184,7 @@ test.concurrent("rejects a RESP simple-string reply whose line terminator never
     }
   }
 
-  // 2) A large but properly terminated simple string under the bound still
-  //    parses.
+  // 2) A large, properly terminated simple string still parses.
   {
     const value = Buffer.alloc(100_000, "B").toString();
     const { server, port } = await listen(Buffer.from("+" + value + "\r\n"), false);

test/js/valkey/valkey-incremental-scan.test.ts

@@ -127,9 +127,8 @@ describe.concurrent("Valkey reply torn across socket reads", () => {
 });
 
 describe("Valkey incremental reply scanning", () => {
-  // Sizes chosen so the reply line stays under the protocol's 512 KiB line
-  // limit while still being large enough that re-scanning the accumulated
-  // partial line on every socket read would dominate the runtime.
+  // Sizes chosen large enough that re-scanning the accumulated partial line on
+  // every socket read would dominate the runtime.
   const HEAD_BYTES = 410_000;
   const CHUNK_BYTES = 2;
   const CHUNK_COUNT = 25_000;