Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions src/valkey/valkey_protocol.rs
Original file line number Diff line number Diff line change
Expand Up @@ -262,7 +262,6 @@ impl<'a> ValkeyReader<'a> {
if buffer.len() > Self::MAX_LINE_LEN + 1 {
return Err(RedisError::LineTooLong);
}

Err(RedisError::InvalidResponse)
}

Expand Down Expand Up @@ -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
Expand Down
70 changes: 70 additions & 0 deletions test/js/valkey/reliability/resp-nesting-depth.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
21 changes: 8 additions & 13 deletions test/js/valkey/valkey-gc.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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`.
Expand All @@ -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);
Expand All @@ -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();
}
Expand All @@ -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);
Expand Down
5 changes: 2 additions & 3 deletions test/js/valkey/valkey-incremental-scan.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
Loading