diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml new file mode 100644 index 0000000..6aa53c2 --- /dev/null +++ b/.github/workflows/test.yml @@ -0,0 +1,16 @@ +name: Protocol tests +on: [push, pull_request] +permissions: + contents: read +jobs: + test: + runs-on: ubuntu-latest + strategy: + matrix: + node: [22, 24] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-node@v4 + with: + node-version: ${{ matrix.node }} + - run: npm test diff --git a/README.md b/README.md index c2e6502..1c01d12 100644 --- a/README.md +++ b/README.md @@ -1,91 +1,71 @@ -# Codecrafters Redis in JavaScript +# A Redis learning implementation in JavaScript -[![Codecrafters Challenge Progress](https://backend.codecrafters.io/progress/redis/b26b2620-8a48-4af9-b624-3729b82c74a7)](https://app.codecrafters.io/users/HMarzban?r=2qF) +This started with the [CodeCrafters Redis challenge](https://codecrafters.io/challenges/redis) +and grew into an experiment with RESP framing, TCP streams and a shared store behind +Node.js cluster workers. It is an in-memory learning project, not a Redis replacement. -Welcome! This is my take on the [Codecrafters Redis Challenge](https://codecrafters.io/challenges/redis) Redis Challenge. I've delved deep into the heart of Redis and ended up crafting my own little version in JavaScript. Dive in and check it out! +## Run -## Introduction - -Welcome to my solution for the [Codecrafters](https://codecrafters.io/challenges/redis) Redis challenge! In this repository, I've created a toy Redis clone that not only handles the basic commands like `PING`, `GET`, and `SET` but also includes an extended `SET` of features that was not in the challenge. With benchmarking tools and extensive testing, this repository is meant to be both an educational resource and a fun exploration into the world of Redis. - -## Features - -### SET Command - -Syntax +Node.js 22+ is required. There are no npm dependencies. ```sh -SET key value [NX | XX] [GET] [EX seconds | PX milliseconds | - EXAT unix-time-seconds | PXAT unix-time-milliseconds | KEEPTTL] +npm start # one worker, 127.0.0.1:6379 +WORKER_COUNT=2 PORT=6380 npm start # shared primary store, two TCP workers +npm test # isolated TCP/cluster fixtures; no Redis install ``` -Supportet Options: - -- `EX` seconds -- Set the specified expire time, in seconds. -- `PX` milliseconds -- Set the specified expire time, in milliseconds. -- `EXAT` timestamp-seconds -- Set the specified Unix time at which the key will expire, in seconds. -- `PXAT` timestamp-milliseconds -- Set the specified Unix time at which the key will expire, in milliseconds. -- `NX` -- Only set the key if it does not already exist. -- `XX` -- Only set the key if it already exists. -- `KEEPTTL` -- Retain the time to live associated with the key. -- `GET` -- Return the old string stored at key, or nil if key did not exist. An error is returned and SET aborted if the value stored at key is not a string. - -### Other Commands - -- `DEL` -- Delete a key -- `ECHO` -- Echo the given string -- `TTL` -- Get the time to live for a key -- `PING` -- Ping the server -- `EXIT` -- Quit the server - -### Additional Features - -- **Robust Testing**: Leveraged the `test.sh` script to ensure all command combinations operate as expected. -- **Performance Benchmarking**: Introduced `benchmark.sh` for performance measurement and analysis. -- **Scalable Workers**: Easily scale worker processes using environment variables. (nodejs, singel thread give you the best perfomance) +`HOST` changes the bind address. There is no authentication; keep the learning server +on loopback. The in-memory store disappears when the primary process exits. -## Getting Started +## Implemented scope -### Prerequisites +| Area | Behavior | +| --- | --- | +| Protocol | RESP2 arrays of bulk strings; fragmented and coalesced TCP requests; binary-safe keys/values and replies. | +| Commands | `PING [message]`, `ECHO`, `GET`, `SET`, `DEL`, `TTL`, and a custom `EXIT`. | +| SET options | `NX`, `XX`, `GET`, `EX`, `PX`, `EXAT`, `PXAT`, `KEEPTTL`; conditional writes are atomic in the primary. | +| Expiry | Checked on access; updating expiry cannot be undone by an older timer. | +| Concurrency | Ordered replies per connection and correlated worker/primary IPC requests. | +| Limits | 1 MiB request/buffer limit and 1,024 command arguments; malformed frames close the connection with an error. | -Ensure [Node.js](https://nodejs.org/) and [redis-cli](https://redis.io/docs/ui/cli/) are installed on your machine. +Not implemented: persistence, replication, Redis Cluster, pub/sub, transactions, +RESP3, inline commands, authentication, eviction or production resource management. +Expired entries that are never read again remain allocated; the store has no memory +limit. Replies do not yet implement socket backpressure. These are deliberate limits +of this learning scope, not guarantees of Redis compatibility. -### Setup & Running +## Verification -1. **Clone the Repository**: +`npm test` exercises byte splits, pipelines, binary/empty values, protocol and command +errors, expiry replacement, conditional options and concurrent clients on two workers. +GitHub Actions runs these fixtures on Node 22 and 24. - ```sh - git clone https://github.com/HMarzban/codecrafters-redis-javascript.git - ``` - -2. **Start the Application:** - - ```sh - npm start - ``` - -To customize the worker count: - -- 4 Workers: `npm run start:worker4` -- 2 Workers: `npm run start:worker2` -- Single Worker: `npm run start:worker1` - -## Testing - -Run the test script to validate the command implementations: +The earlier redis-cli shell assertions remain available: ```sh -npm run test +PORT=6380 npm run test:legacy # requires a separately running server on that port ``` -## Benchmarking +## Benchmark methodology -Analyze the performance of the system with the benchmarking script: +The original harness is available as `npm run benchmark -- 1000 100`. It starts a new +redis-cli process/connection for each operation, so its mean includes client startup, +shell and network costs. It does **not** measure isolated server latency. The former +package script referenced a nonexistent filename; it now invokes `benchmark.sh`. + +For a persistent-client experiment, install the Redis tools and run against a separate +learning server: ```sh -npm run benchmark +redis-benchmark -h 127.0.0.1 -p 6380 -t set,get -n 10000 -c 20 -d 100 --csv ``` -> Note: Ensure the node application is hitting in the background before initiating tests or benchmarks. +Record the commit, Node version, OS/CPU/RAM, worker count, client count, payload size, +request count, throughput, latency percentiles and errors. Compare the same workload +and environment, repeat runs, and retain raw output. No benchmark results or speedup +claims are supplied by this maintenance pass. + +## License -I hope you find this project insightful and get a deeper understanding of how Redis works in the background. If you're curious like me, dive in and have fun learning! +The original package manifest declares MIT, but the repository has no standalone +license file. This maintenance pass does not add or change licensing terms. diff --git a/app/commands/del.js b/app/commands/del.js index a0df08f..fa54b13 100644 --- a/app/commands/del.js +++ b/app/commands/del.js @@ -1,36 +1,2 @@ import db from "../db.js"; - -const NO_KEY_PROVIDED_ERROR = "-Error: No key provided\r\n"; -const KEY_NOT_FOUND_ERROR = "-Error: Key not found\r\n"; -const DELETED_SUCCESS = "+OK\r\n"; - -/** - * Deletes one or more keys. - * For more information: https://redis.io/commands/del/ - * - * @param {object} connection - Connection object. - * @param {array} data - Data array containing the keys to be deleted. - */ -const DEL = async (connection, data) => { - if (!data || data.length === 0) { - connection.write(NO_KEY_PROVIDED_ERROR); - return; - } - - let deletedKeys = 0; - - for (const key of data) { - if (await db.has(key)) { - await db.delete(key); - deletedKeys++; - } - } - - if (data.length > 1) { - connection.write(`:${deletedKeys}\r\n`); - } else { - connection.write(deletedKeys ? DELETED_SUCCESS : KEY_NOT_FOUND_ERROR); - } -}; - -export default DEL; +export default async (connection, keys) => connection.write(`:${await db.delete(keys)}\r\n`); diff --git a/app/commands/echo.js b/app/commands/echo.js index c6e1ad9..2347236 100644 --- a/app/commands/echo.js +++ b/app/commands/echo.js @@ -1,16 +1,2 @@ -// Constants for response format -const BULK_STRING_FORMAT = `+{data}\r\n`; - -/** - * Returns the given string. - * For more information: https://redis.io/commands/echo/ - * - * @param {object} connection - Connection object. - * @param {array} data - Data array containing the string to be echoed. - */ -const ECHO = (connection, data) => { - const responseData = data.join(" "); - connection.write(BULK_STRING_FORMAT.replace("{data}", responseData)); -}; - -export default ECHO; +import { bulk } from "../resp.js"; +export default (connection, [value]) => connection.write(bulk(value)); diff --git a/app/commands/get.js b/app/commands/get.js index dca7a51..27057ad 100644 --- a/app/commands/get.js +++ b/app/commands/get.js @@ -1,20 +1,3 @@ import db from "../db.js"; - -// Constants for responses -const NULL_BULK_STRING = "$-1\r\n"; - -/** - * Retrieve the value of the specified key. - * For more information: https://redis.io/commands/get/ - * - * @param {object} connection - Connection object. - * @param {array} data - Data array containing key as the first element. - */ -const GET = async (connection, data) => { - const key = data.at(0); - const result = await db.get(key); - - connection.write(result ? `+${result.value}\r\n` : NULL_BULK_STRING); -}; - -export default GET; +import { bulk } from "../resp.js"; +export default async (connection, [key]) => connection.write(bulk(await db.get(key))); diff --git a/app/commands/index.js b/app/commands/index.js index 8623f01..4c70aaa 100644 --- a/app/commands/index.js +++ b/app/commands/index.js @@ -3,35 +3,19 @@ import GET from "./get.js"; import ECHO from "./echo.js"; import DEL from "./del.js"; import TTL from "./ttl.js"; - -// Mapping commands to their corresponding action functions +import { bulk } from "../resp.js"; const commandMap = { - ping: (connection) => connection.write("+PONG\r\n"), - echo: (connection, data) => ECHO(connection, data), - set: (connection, data) => SET(connection, data), - get: (connection, data) => GET(connection, data), - del: (connection, data) => DEL(connection, data), - ttl: (connection, data) => TTL(connection, data), - exit: (connection) => { - connection.write(`You will be disconnected, PID: ${process.pid}\r\n`); - connection.end(); - }, + ping: [0, 1, (connection, args) => connection.write(args.length ? bulk(args[0]) : "+PONG\r\n")], + echo: [1, 1, ECHO], set: [2, Infinity, SET], get: [1, 1, GET], + del: [1, Infinity, DEL], ttl: [1, 1, TTL], + exit: [0, 0, (connection) => connection.end("+OK\r\n")], }; - -/** - * Executes the given command with the provided data. - * - * @param {string} command - The command to be executed. - * @param {array} data - The data associated with the command. - * @param {object} connection - The connection object. - */ -const commands = (command, data, connection) => { +export default function commands(command, args, connection) { const action = commandMap[command]; - if (action) { - return action(connection, data); - } else { - connection.write("-Error: Unknown command\r\n"); + if (!Object.hasOwn(commandMap, command)) return connection.write("-ERR unknown command\r\n"); + const [minimum, maximum, execute] = action; + if (args.length < minimum || args.length > maximum) { + return connection.write("-ERR wrong number of arguments\r\n"); } -}; - -export default commands; + return execute(connection, args); +} diff --git a/app/commands/set.js b/app/commands/set.js index da9f38b..b077842 100644 --- a/app/commands/set.js +++ b/app/commands/set.js @@ -1,174 +1,31 @@ import db from "../db.js"; - -const computeDelay = (expiryType, expiryTime) => { - switch (expiryType) { - case "ex": - return expiryTime * 1000; - case "px": - return expiryTime; - case "exat": - return expiryTime * 1000 - Date.now(); - case "pxat": - return expiryTime - Date.now(); - default: - return null; - } -}; - -const validateExpiryType = (expiryType, timeOptions) => { - if (!expiryType && !timeOptions.includes(expiryType)) return null; - if (!timeOptions.includes(expiryType)) { - return `Error: Invalid expiryType. Allowed values are: ${timeOptions.join( - ", " - )}`; - } - return null; -}; - -const validateExpiryTime = (expiryTime) => { - if (typeof expiryTime !== "number" || expiryTime <= 0) { - return "Error: Invalid or missing expiryTime."; - } - return null; -}; - -const handleDBSet = async (key, value, setBehaviour) => { - const keyExists = await db.has(key); - - if ( - !setBehaviour || - (setBehaviour === "nx" && !keyExists) || - (setBehaviour === "xx" && keyExists) - ) { - await db.set(key, value); - return true; - } - - return false; // Handle cases that don't match any of the conditions. -}; - -//Syntax: -// SET key value [NX | XX] [GET] [EX seconds | PX milliseconds | EXAT unix-time-seconds | PXAT unix-time-milliseconds | KEEPTTL] -// URL: https://redis.io/commands/set/ -const SET = async (connection, query) => { - // Constants - const timeOptions = ["ex", "px", "exat", "pxat", "keepttl"]; - const setBehaviourOptions = ["nx", "xx"]; - const getBehaviourOptions = ["get"]; - - // Default Values - let key, value, setBehaviour, getBehaviour, expiryType, expiryTimeRaw; - - // First two are always key and value - [key, value, ...query] = query; - - query = query.map((item) => item.toLowerCase()); - - // Check for SET behaviours NX or XX - if (setBehaviourOptions.includes(query[0])) { - setBehaviour = query.shift(); - } - - // Check for GET behaviour - if (getBehaviourOptions.includes(query[0])) { - getBehaviour = query.shift(); - } - - // Check for timeOptions - if (timeOptions.includes(query[0])) { - expiryType = query.shift(); - if (["ex", "px", "exat", "pxat"].includes(expiryType)) { - expiryTimeRaw = query.shift(); - } - } - - const expiryTime = +expiryTimeRaw; - - const expiryTypeError = validateExpiryType(expiryType, timeOptions); - if (expiryTypeError) { - connection.write(expiryTypeError + "\r\n"); - return false; - } - - if (expiryType === "keepttl") { - const keyData = await db.get(key); - if (keyData) { - handleDBSet(key, { ...keyData, value }, setBehaviour); - } - if (getBehaviour && keyData) { - connection.write(`+"${value}"\r\n`); +import { bulk } from "../resp.js"; + +export default async function SET(connection, args) { + const [key, value, ...flags] = args; + const options = { condition: null, keepTTL: false, expiresAt: null }; + let returnPrevious = false; + for (let i = 0; i < flags.length; i++) { + const flag = flags[i].toString("ascii").toLowerCase(); + if ((flag === "nx" || flag === "xx") && !options.condition) options.condition = flag; + else if (flag === "get" && !returnPrevious) returnPrevious = true; + else if (flag === "keepttl" && options.expiresAt === null && !options.keepTTL) options.keepTTL = true; + else if (["ex", "px", "exat", "pxat"].includes(flag) && options.expiresAt === null && !options.keepTTL) { + const raw = flags[++i]?.toString("ascii") ?? ""; + const time = Number(raw); + const milliseconds = time * (flag === "ex" || flag === "exat" ? 1000 : 1); + const absolute = flag === "exat" || flag === "pxat"; + if (!/^[0-9]+$/.test(raw) || !Number.isSafeInteger(milliseconds) || time <= 0 || + (!absolute && !Number.isSafeInteger(Date.now() + milliseconds))) { + connection.write("-ERR invalid expire time in 'set' command\r\n"); + return; + } + options.expiresAt = absolute ? milliseconds : Date.now() + milliseconds; } else { - connection.write("+OK\r\n"); + connection.write("-ERR syntax error\r\n"); + return; } - return true; - } - - // overwrite existing key and reset expiry - if (!expiryType && !expiryTime) { - const keyData = await db.get(key); - - handleDBSet( - key, - { - ...keyData, - value, - expiryType: null, - expiryTime: null, - }, - setBehaviour - ); - if (getBehaviour && keyData) { - connection.write(`+${keyData.value}\r\n`); - } else { - connection.write("+OK\r\n"); - } - return true; - } - - const expiryTimeError = validateExpiryTime(expiryTime); - if (expiryTimeError) { - connection.write(expiryTimeError + "\r\n"); - return false; } - - const delay = computeDelay(expiryType, expiryTime); - if (delay === null || delay <= 0) { - connection.write( - "Error: Invalid delay computed or timestamp is in the past. \r\n" - ); - return false; - } - - setTimeout(async () => { - console.info("expired"); - const data = await db.get(key); - if (data?.expiryTime) { - await db.delete(key); - } - }, delay); - - if (getBehaviour) { - const keyData = await db.get(key); - handleDBSet( - key, - { value, expiryType, expiryTime, ttx: Date.now() + delay }, - setBehaviour - ); - - if (keyData) { - connection.write(`+${keyData.value}\r\n`); - } - return true; - } else { - handleDBSet( - key, - { value, expiryType, expiryTime, ttx: Date.now() + delay }, - setBehaviour - ); - } - - connection.write("+OK\r\n"); - return true; -}; - -export default SET; + const result = await db.set(key, value, options); + connection.write(returnPrevious ? bulk(result.previous) : result.applied ? "+OK\r\n" : bulk(null)); +} diff --git a/app/commands/ttl.js b/app/commands/ttl.js index 24db0d7..d5f5a3f 100644 --- a/app/commands/ttl.js +++ b/app/commands/ttl.js @@ -1,37 +1,2 @@ import db from "../db.js"; - -// Constants for error responses -const KEY_NOT_FOUND = `-2\r\n`; -const KEY_NO_EXPIRE = `-1\r\n`; - -/** - * Get the time-to-live of a key. - * For more information: https://redis.io/commands/ttl/ - * - * @param {object} connection - Connection object. - * @param {array} data - Data array containing key as the first element. - */ -const TTL = async (connection, data) => { - const key = data.at(0); - - // Check if the key exists in the database - if (!(await db.has(key))) { - connection.write(KEY_NOT_FOUND); - return; - } - - const result = await db.get(key); - - // Check if the key has an expiration time set - if (!result.ttx || result.ttx < Date.now()) { - connection.write(KEY_NO_EXPIRE); - return; - } - - // Calculate remaining time in seconds - const remainingTime = Math.round((result.ttx - Date.now()) / 1000); - - connection.write(`+${remainingTime}\r\n`); -}; - -export default TTL; +export default async (connection, [key]) => connection.write(`:${await db.ttl(key)}\r\n`); diff --git a/app/db.js b/app/db.js index 2a2af16..2c1d7e1 100644 --- a/app/db.js +++ b/app/db.js @@ -1,53 +1,29 @@ -/** - * Sends an IPC command and waits for a message in response. - * - * @param {object} payload - The data to send to the master process. - * @returns {Promise} - A promise that resolves with the message from the master process. - */ -function sendCommand(payload) { +// Correlation IDs keep concurrent clients from consuming each other's IPC replies. +let sequence = 0; +const pending = new Map(); +process.on("message", (message) => { + const waiter = pending.get(message.id); + if (!waiter) return; + pending.delete(message.id); + if (message.error) waiter.reject(new Error(message.error)); + else waiter.resolve(message.result); +}); + +function request(command, data) { return new Promise((resolve, reject) => { - process.send(payload, (error) => { - if (error) { - reject(error); - } else { - process.once("message", resolve); - } + const id = ++sequence; + pending.set(id, { resolve, reject }); + process.send({ id, command, ...data }, (error) => { + if (!error) return; + pending.delete(id); + reject(error); }); }); } - -async function set(key, value) { - const response = await sendCommand({ command: "set", key, value }); - if (response.status !== "ok") { - throw new Error("Failed to set value in master."); - } -} - -async function get(key) { - const response = await sendCommand({ command: "get", key }); - return response.value || null; -} - -async function deletekey(key) { - const response = await sendCommand({ command: "delete", key }); - switch (response.status) { - case "ok": - break; - case "not_found": - throw new Error("Key not found in master."); - default: - throw new Error("Failed to delete key from master."); - } -} - -async function has(key) { - const response = await sendCommand({ command: "has", key }); - return response.exists || false; -} - +const keyId = (key) => key.toString("base64"); export default { - set, - get, - delete: deletekey, - has, + get: (key) => request("get", { key: keyId(key) }), + set: (key, value, options) => request("set", { key: keyId(key), value, options }), + delete: (keys) => request("delete", { keys: keys.map(keyId) }), + ttl: (key) => request("ttl", { key: keyId(key) }), }; diff --git a/app/helper.js b/app/helper.js deleted file mode 100644 index 817f5b0..0000000 --- a/app/helper.js +++ /dev/null @@ -1,21 +0,0 @@ -const validCommands = ["echo", "ping", "set", "get", "del", "ttl", "exit"]; - -export const isCommand = (message) => { - const normalized = message.at(0).toLowerCase(); - return validCommands.includes(normalized); -}; - -export const getCommand = (req) => { - const dataNoLengths = req.slice(1).filter((x) => x.charAt(0) !== "$"); - - if (isCommand(dataNoLengths)) { - return { - command: dataNoLengths[0].toLowerCase(), - data: dataNoLengths.slice(1), - }; - } - return { - command: "error", - data: "error", - }; -}; diff --git a/app/ipc.js b/app/ipc.js index ce9ec96..7349e97 100644 --- a/app/ipc.js +++ b/app/ipc.js @@ -1,38 +1,39 @@ -const IPC = (cluster) => { +export default function IPC(cluster) { const dataStore = new Map(); - - const commandHandlers = { - set: (message, worker) => { - dataStore.set(message.key, message.value); - worker.send({ status: "ok" }); - }, - - has: (message, worker) => { - const hasKey = dataStore.has(message.key); - worker.send({ exists: hasKey }); - }, - - get: (message, worker) => { - const value = dataStore.get(message.key); - worker.send({ value: value }); - }, - - delete: (message, worker) => { - if (dataStore.has(message.key)) { - dataStore.delete(message.key); - worker.send({ status: "ok" }); - } else { - worker.send({ status: "not_found" }); + function get(key) { + const entry = dataStore.get(key); + if (entry?.expiresAt !== null && entry?.expiresAt <= Date.now()) { + dataStore.delete(key); + return undefined; + } + return entry; + } + // Each operation is synchronous in the primary, including conditional SET. + const handlers = { + get: ({ key }) => get(key)?.value ?? null, + set: ({ key, value, options }) => { + const previous = get(key); + if ((options.condition === "nx" && previous) || (options.condition === "xx" && !previous)) { + return { applied: false, previous: previous?.value ?? null }; } + const expiresAt = options.keepTTL ? previous?.expiresAt ?? null : options.expiresAt; + dataStore.set(key, { value, expiresAt }); + return { applied: true, previous: previous?.value ?? null }; + }, + delete: ({ keys }) => keys.reduce((count, key) => { + return get(key) && dataStore.delete(key) ? count + 1 : count; + }, 0), + ttl: ({ key }) => { + const entry = get(key); + if (!entry) return -2; + if (entry.expiresAt === null) return -1; + return Math.max(0, Math.round((entry.expiresAt - Date.now()) / 1000)); }, }; - cluster.on("message", (worker, message) => { - const handler = commandHandlers[message.command]; - if (handler) { - handler(message, worker); - } + const handler = handlers[message.command]; + if (!handler) return; + const result = handler(message); + if (worker.isConnected()) worker.send({ id: message.id, result }, () => {}); }); -}; - -export default IPC; +} diff --git a/app/main.js b/app/main.js index beaa086..d691ff4 100644 --- a/app/main.js +++ b/app/main.js @@ -1,4 +1,3 @@ -import os from "node:os"; import cluster from "node:cluster"; import startWorkerProcess from "./worker.js"; import IPC from "./ipc.js"; @@ -9,14 +8,24 @@ function startMasterProcess() { // Inter-process communication (IPC) IPC(cluster); + cluster.setupPrimary({ serialization: "advanced" }); + // Fork workers - const defaultNumWorkers = os.cpus().length; + const defaultNumWorkers = 1; const numWorkers = process.env.WORKER_COUNT || defaultNumWorkers; for (let i = 0; i < numWorkers; i++) { cluster.fork(); } + let shuttingDown = false; + for (const signal of ["SIGTERM", "SIGINT"]) { + process.on(signal, () => { + shuttingDown = true; + for (const worker of Object.values(cluster.workers)) worker.kill(signal); + }); + } + // Handle worker events cluster.on("fork", (workerInstance) => { console.info(`Worker ${workerInstance.process.pid} is forked`); @@ -27,7 +36,7 @@ function startMasterProcess() { `Worker ${workerInstance.process.pid} died with code ${code} and signal ${signal}` ); console.info("Forking a new worker..."); - cluster.fork(); + if (!shuttingDown) cluster.fork(); }); } diff --git a/app/resp.js b/app/resp.js new file mode 100644 index 0000000..ea626e4 --- /dev/null +++ b/app/resp.js @@ -0,0 +1,41 @@ +// RESP2 command arrays use byte lengths, not JavaScript string lengths. +export const MAX_REQUEST_BYTES = 1024 * 1024; +const CRLF = Buffer.from("\r\n"); + +export function parseRequest(buffer) { + let offset = 0; + function length(prefix, maximum) { + if (offset >= buffer.length) return null; + if (buffer[offset] !== prefix.charCodeAt(0)) throw new Error(`expected '${prefix}'`); + const end = buffer.indexOf(CRLF, offset); + if (end === -1) return null; + const text = buffer.toString("latin1", offset + 1, end); + if (!/^(0|[1-9][0-9]*)$/.test(text)) throw new Error("invalid length"); + const value = Number(text); + if (!Number.isSafeInteger(value) || value > maximum) throw new Error("request too large"); + offset = end + 2; + return value; + } + const count = length("*", 1024); + if (count === null) return null; + if (count === 0) throw new Error("empty command"); + const args = []; + for (let i = 0; i < count; i++) { + const size = length("$", MAX_REQUEST_BYTES); + if (size === null) return null; + if (offset + size + 2 > MAX_REQUEST_BYTES) throw new Error("request too large"); + if (buffer.length < offset + size + 2) return null; + if (!buffer.subarray(offset + size, offset + size + 2).equals(CRLF)) { + throw new Error("invalid bulk string terminator"); + } + args.push(Buffer.from(buffer.subarray(offset, offset + size))); + offset += size + 2; + } + return { args, bytes: offset }; +} + +export function bulk(value) { + if (value === null || value === undefined) return Buffer.from("$-1\r\n"); + const bytes = Buffer.isBuffer(value) ? value : Buffer.from(value); + return Buffer.concat([Buffer.from(`$${bytes.length}\r\n`), bytes, CRLF]); +} diff --git a/app/worker.js b/app/worker.js index efd1027..487ba36 100644 --- a/app/worker.js +++ b/app/worker.js @@ -1,32 +1,47 @@ import net from "node:net"; -import { getCommand } from "./helper.js"; +import { MAX_REQUEST_BYTES, parseRequest } from "./resp.js"; import commands from "./commands/index.js"; -const handleData = (connection) => (req) => { - const requestCleansed = req.toString().trim().split("\r\n"); - const { command, data } = getCommand(requestCleansed); - try { - commands(command, data, connection); - } catch (error) { - console.error("commands:", error); - } -}; - -const handleEnd = () => { - console.info("client disconnected, PID:", process.pid); -}; - -const createServer = () => { +export default function createServer() { const server = net.createServer({ keepAlive: true }, (connection) => { - console.info("client connected, PID:", process.pid); + let buffer = Buffer.alloc(0); + let processing = false; + let failed = false; - connection.on("data", handleData(connection)); - connection.on("end", handleEnd); - }); + async function drain() { + if (processing || failed) return; + processing = true; + try { + while (buffer.length && !connection.destroyed) { + const request = parseRequest(buffer); + if (!request) break; + buffer = buffer.subarray(request.bytes); + const [command, ...args] = request.args; + await commands(command.toString("utf8").toLowerCase(), args, connection); + if (connection.writableEnded) break; + } + } catch (error) { + failed = true; + connection.end("-ERR protocol or command processing error\r\n"); + } finally { + processing = false; + } + } - server.listen(6379, () => { - console.info(`server is listening, PID: ${process.pid}`); + connection.on("data", (chunk) => { + if (failed) return; + if (buffer.length + chunk.length > MAX_REQUEST_BYTES) { + failed = true; + connection.end("-ERR request buffer limit exceeded\r\n"); + return; + } + buffer = Buffer.concat([buffer, chunk]); + void drain(); + }); + connection.on("error", () => connection.destroy()); }); -}; - -export default createServer; + server.listen(Number(process.env.PORT || 6379), process.env.HOST || "127.0.0.1", () => { + console.info(`server listening on ${server.address().port}, PID: ${process.pid}`); + }); + return server; +} diff --git a/benchmark.sh b/benchmark.sh index df8bacf..cd3abfb 100755 --- a/benchmark.sh +++ b/benchmark.sh @@ -1,8 +1,15 @@ #!/bin/bash +set -euo pipefail + COUNT=${1:-1000} DATASIZE=${2:-100} +[[ "$COUNT" =~ ^[1-9][0-9]*$ && "$DATASIZE" =~ ^[1-9][0-9]*$ ]] || { echo "Counts must be positive integers" >&2; exit 1; } + +echo "Legacy shell harness: each operation starts redis-cli and opens a connection." +echo "These timings are not isolated Redis command latency." + # Prepare random data once to reuse RANDOM_DATA=$(head -c $DATASIZE /dev/null + echo "SET key$i $RANDOM_DATA" | redis-cli -p "${PORT:-6379}" > /dev/null progress_indicator $i done END_SET=$(current_time_ms) DURATION_SET=$((END_SET - START_SET)) -echo -e "\nAverage SET time: $((DURATION_SET * 1000 / COUNT)) microseconds" +echo -e "\nMean SET harness time (includes process/connection overhead): $((DURATION_SET * 1000 / COUNT)) microseconds" # Measure GET operations echo "Starting GET operations..." START_GET=$(current_time_ms) for i in $(seq 1 $COUNT); do - echo "GET key$i" | redis-cli > /dev/null + echo "GET key$i" | redis-cli -p "${PORT:-6379}" > /dev/null progress_indicator $i done END_GET=$(current_time_ms) DURATION_GET=$((END_GET - START_GET)) -echo -e "\nAverage GET time: $((DURATION_GET * 1000 / COUNT)) microseconds" +echo -e "\nMean GET harness time (includes process/connection overhead): $((DURATION_GET * 1000 / COUNT)) microseconds" # Cleanup echo "Cleaning up..." for i in $(seq 1 $COUNT); do - echo "DEL key$i" | redis-cli > /dev/null + echo "DEL key$i" | redis-cli -p "${PORT:-6379}" > /dev/null done echo "Benchmark completed!" diff --git a/package.json b/package.json index 29a1be8..fcfb8ca 100644 --- a/package.json +++ b/package.json @@ -10,7 +10,11 @@ "start:worker4": "WORKER_COUNT=4 node app/main.js", "start:worker2": "WORKER_COUNT=2 node app/main.js", "start:worker1": "WORKER_COUNT=1 node app/main.js", - "test": "sh ./test.sh", - "benchmark": "sh ./benchmark_redis.sh" + "test": "node --test", + "benchmark": "bash ./benchmark.sh", + "test:legacy": "bash ./test.sh" + }, + "engines": { + "node": ">=22" } } diff --git a/test.sh b/test.sh index 0941394..b7c3043 100644 --- a/test.sh +++ b/test.sh @@ -14,11 +14,11 @@ fail_msg() { } set_key() { - redis-cli SET "$@" + redis-cli -p "${PORT:-6379}" SET "$@" } get_key() { - redis-cli GET "$1" + redis-cli -p "${PORT:-6379}" GET "$1" } check_value() { @@ -30,7 +30,7 @@ check_value() { } cleanup() { - redis-cli DEL mykey + redis-cli -p "${PORT:-6379}" DEL mykey } echo "Starting tests..." @@ -76,7 +76,7 @@ check_value mykey "With XX" "XX" set_key mykey "With TTL" EX 6 sleep 2 set_key mykey "With KEEPTTL" KEEPTTL -remaining_ttl=$(redis-cli TTL mykey) +remaining_ttl=$(redis-cli -p "${PORT:-6379}" TTL mykey) if [[ $remaining_ttl -lt 6 && $remaining_ttl -gt 0 ]]; then pass_msg "KEEPTTL. TTL: $remaining_ttl" else @@ -85,7 +85,7 @@ fi # GET option set_key mykey "Original Value" -old_value=$(redis-cli SET mykey "With GET" GET) +old_value=$(redis-cli -p "${PORT:-6379}" SET mykey "With GET" GET) retrieved=$(get_key mykey) if [[ "$old_value" == "Original Value" && "$retrieved" == "With GET" ]]; then pass_msg "GET option. Old: $old_value, New: $retrieved" @@ -93,11 +93,11 @@ else fail_msg "GET option. Old: $old_value, New: $retrieved" fi -# 1. Multiple Simultaneous SETs +# 1. Multiple sequential SETs set_key key1 "Simultaneous" set_key key2 "Simultaneous" -check_value key1 "Simultaneous" "Simultaneous SET (key1)" -check_value key2 "Simultaneous" "Simultaneous SET (key2)" +check_value key1 "Simultaneous" "Sequential SET (key1)" +check_value key2 "Simultaneous" "Sequential SET (key2)" cleanup # 2. Setting Large Values @@ -116,7 +116,7 @@ cleanup # 4. SET then DEL Combination set_key tempkey "Temporary" -redis-cli DEL tempkey +redis-cli -p "${PORT:-6379}" DEL tempkey check_value tempkey "" "DEL After SET" cleanup diff --git a/test/protocol.test.js b/test/protocol.test.js new file mode 100644 index 0000000..3d4c0ef --- /dev/null +++ b/test/protocol.test.js @@ -0,0 +1,121 @@ +import { after, before, test } from 'node:test'; +import assert from 'node:assert/strict'; +import { spawn } from 'node:child_process'; +import net from 'node:net'; +import { once } from 'node:events'; +import { setTimeout as delay } from 'node:timers/promises'; +import { parseRequest, bulk, MAX_REQUEST_BYTES } from '../app/resp.js'; + +const frame = (...values) => Buffer.concat([ + Buffer.from(`*${values.length}\r\n`), ...values.map(value => bulk(Buffer.isBuffer(value) ? value : String(value))), +]); +let processHandle; +let port; +before(async () => { + processHandle = spawn(process.execPath, ['app/main.js'], { + env: { ...process.env, PORT: '0', WORKER_COUNT: '2' }, stdio: ['ignore', 'pipe', 'pipe'], + }); + port = await new Promise((resolve, reject) => { + const timeout = setTimeout(() => reject(new Error('server startup timed out')), 5000); + processHandle.stdout.on('data', data => { + const match = String(data).match(/server listening on (\d+)/); + if (match) { clearTimeout(timeout); resolve(Number(match[1])); } + }); + processHandle.once('error', error => { clearTimeout(timeout); reject(error); }); + processHandle.once('exit', code => { clearTimeout(timeout); reject(new Error(`server exited: ${code}`)); }); + }); +}); +after(async () => { + if (processHandle?.exitCode === null) { + const exited = once(processHandle, 'exit'); + processHandle.kill('SIGTERM'); + await exited; + } +}); + +async function client(t) { + const socket = net.connect(port, '127.0.0.1'); + await once(socket, 'connect'); + t.after(() => socket.destroy()); + let received = Buffer.alloc(0); + socket.on('data', data => { received = Buffer.concat([received, data]); }); + return { + socket, + async expect(expected) { + expected = Buffer.isBuffer(expected) ? expected : Buffer.from(expected); + const started = Date.now(); + while (received.length < expected.length && Date.now() - started < 2500) await delay(5); + assert.deepEqual(received.subarray(0, expected.length), expected); + received = received.subarray(expected.length); + }, + }; +} + +test('length parser preserves every byte and every possible split', () => { + const payload = Buffer.from([0, 255, 36, 13, 10, 226, 130, 172]); + const request = frame('ECHO', payload); + for (let split = 0; split < request.length; split++) assert.equal(parseRequest(request.subarray(0, split)), null); + const parsed = parseRequest(Buffer.concat([request, frame('PING')])); + assert.deepEqual(parsed.args, [Buffer.from('ECHO'), payload]); + assert.equal(parsed.bytes, request.length); +}); + +test('invalid RESP lengths and terminators are rejected', () => { + for (const text of ['*0\r\n', '*-1\r\n', '*1x\r\n', '*1\r\n$-1\r\n', '*1\r\n$1\r\naXX', `*1\r\n$${MAX_REQUEST_BYTES + 1}\r\n`]) { + assert.throws(() => parseRequest(Buffer.from(text))); + } + assert.throws(() => parseRequest(Buffer.from([42, 177, 13, 10]))); +}); + +test('fragmented command and pipelined commands keep reply order', async t => { + const c = await client(t); + const request = frame('SET', 'pipe', 'first'); + for (const byte of request) c.socket.write(Buffer.from([byte])); + c.socket.write(Buffer.concat([frame('GET', 'pipe'), frame('SET', 'pipe', 'second'), frame('GET', 'pipe'), frame('PING')])); + await c.expect(Buffer.concat([Buffer.from('+OK\r\n'), bulk('first'), Buffer.from('+OK\r\n'), bulk('second'), Buffer.from('+PONG\r\n')])); +}); + +test('binary keys, values, empty values and CRLF round-trip', async t => { + const c = await client(t); + const key = Buffer.from([255, 0, 13, 10]); + const value = Buffer.from([255, 36, 13, 10, 0, 195, 169]); + c.socket.write(Buffer.concat([frame('SET', key, value), frame('GET', key), frame('ECHO', value), frame('SET', 'empty', ''), frame('GET', 'empty')])); + await c.expect(Buffer.concat([Buffer.from('+OK\r\n'), bulk(value), bulk(value), Buffer.from('+OK\r\n'), bulk('')])); +}); + +test('command errors do not desynchronize a pipeline', async t => { + const c = await client(t); + c.socket.write(Buffer.concat([frame('BOGUS'), frame('GET'), frame('SET', 'x', 'y', 'PX', 'NaN'), frame('PING')])); + await c.expect('-ERR unknown command\r\n-ERR wrong number of arguments\r\n-ERR invalid expire time in \'set\' command\r\n+PONG\r\n'); +}); + +test('malformed requests return an error and close the connection', async t => { + const c = await client(t); + const ended = once(c.socket, 'end'); + c.socket.write('*1\r\n$1\r\naXX'); + await c.expect('-ERR protocol or command processing error\r\n'); + await ended; +}); + +test('expiry replacement, conditional writes, GET and KEEPTTL', async t => { + const c = await client(t); + c.socket.write(Buffer.concat([frame('SET', 'expiry', 'old', 'PX', 30), frame('SET', 'expiry', 'new', 'PX', 5000), frame('SET', 'expiry', 'ignored', 'NX'), frame('SET', 'expiry', 'kept', 'KEEPTTL', 'GET'), frame('SET', 'absent', 'no', 'XX')])); + await c.expect(Buffer.concat([Buffer.from('+OK\r\n+OK\r\n$-1\r\n'), bulk('new'), bulk(null)])); + await delay(60); + c.socket.write(Buffer.concat([frame('GET', 'expiry'), frame('TTL', 'expiry'), frame('SET', 'short', 'gone', 'PX', 20)])); + await c.expect(Buffer.concat([bulk('kept'), Buffer.from(':5\r\n+OK\r\n')])); + await delay(40); + c.socket.write(Buffer.concat([frame('GET', 'short'), frame('TTL', 'short'), frame('DEL', 'short'), frame('SET', 'plain', 'x'), frame('TTL', 'plain')])); + await c.expect('$-1\r\n:-2\r\n:0\r\n+OK\r\n:-1\r\n'); +}); + +test('concurrent clients receive their own IPC responses', async t => { + await Promise.all(Array.from({ length: 12 }, async (_, id) => { + const c = await client(t); + for (let n = 0; n < 10; n++) { + const value = `${id}:${n}`; + c.socket.write(Buffer.concat([frame('SET', `client:${id}`, value), frame('GET', `client:${id}`)])); + await c.expect(Buffer.concat([Buffer.from('+OK\r\n'), bulk(value)])); + } + })); +});