initial commit

Author: sarfarazahmedkl

README.md

@@ -1,14 +1,28 @@
-# Redis Cluster Pipeline Library
+# Redis Cluster Pipeline Library with <img src="./images/pw.jpg" alt="drawing" width="30"/>
 
-## Overview
+This is a custom wrapper around the [ioredis](https://www.npmjs.com/package/ioredis) that adds a new method `clusterPipeline` to handle Redis Cluster pipelines more efficiently.
 
-This is a custom wrapper library around the popular ioredis client that adds a new method called `clusterPipeline` to handle Redis Cluster pipelines more efficiently. This library was developed by the PhysicsWallah Private Limited tech team to solve issues with executing multiple Redis commands across Redis Cluster nodes.
+## Why This Library?
+
+### Problem at PhysicsWallah
+
+At PhysicsWallah, we needed an efficient way to handle Redis Cluster commands when multiple Redis nodes were involved. The standard Redis pipeline approach didn't take slot distribution into account, leading to inefficiencies and complexity. We developed this library to:
+
+- Simplify the execution of multiple Redis commands across a Redis Cluster.
+- Maintain the order of results while executing commands in parallel.
+- Make our Redis Cluster usage more efficient and less error-prone.
 
-## Problem Statement
+## Limitations of ioredis in Clustered Redis Topology
 
-While using Redis Cluster with the ioredis client, handling multiple commands in a way that correctly distributes them across nodes while maintaining the order of execution was challenging. To address this, we developed this library that adds the `clusterPipeline` method, making it easy to manage Redis commands in a Redis Cluster environment.
+As stated in the official [ioredis documentation](https://github.com/redis/ioredis?tab=readme-ov-file#transaction-and-pipeline-in-cluster-mode):
 
-This library is simply a wrapper around ioredis, extending it with an additional method that enables Redis Cluster commands to be executed in parallel, with the results returned in the correct order.
+> "All keys in a pipeline should belong to slots served by the same node, since ioredis sends all commands in a pipeline to the same node."
+
+This means that when executing multiple commands in a pipeline, all keys must belong to the same Redis slot, as `ioredis` sends the pipeline commands to a single node. If the keys interact with different shards, pipeline commands would fail with error `All the keys in a pipeline command should belong to the same slot`
+
+Due to this limitation, pipelining cannot be reliably used in Redis clusters with ioredis, leading to increased network latency as each command is executed individually instead of being batched together.
+
+Our custom library resolves this issue by enabling efficient and reliable pipelining support in clustered Redis environments, improving performance and reducing latency.
 
 ## Features
 
@@ -22,40 +36,40 @@ This library is simply a wrapper around ioredis, extending it with an additional
 To install the library, use npm or yarn to add it to your project.
 
 ```bash
-npm install @pw-tech/ioredis
+npm install @pw-tech/pw-redis
 ```
 
 Or with yarn:
 
 ```bash
-yarn add @pw-tech/ioredis
+yarn add @pw-tech/pw-redis
 ```
 
 ## Usage
 
 After installing, you can use the Cluster class just as you would with ioredis, but with the added functionality of the `clusterPipeline` method for Redis Cluster commands.
+```javascript
+clusterPipeline(commands: [string, ...any][]): Promise<[error: Error | null, result: unknown][] | null>
+```
 
-### Example
+##### Parameters:
 
-1. Basic Redis Usage (Without Clusters)
+- `commands`:  An array of Redis commands, where each command is represented as an array with the command name as the first element, followed by its respective arguments.
+  - Example:  [['set', 'key1', 'value1'], ['get', 'key1']].
 
-```javascript
-import { Redis } from '@pw-tech/ioredis';
+##### Returns:
 
-const redis = new Redis({
-  host: 'localhost',
-  port: 6379,
-});
+- A Promise that resolves to an array of results for each command, in the same order as the input.
+  - Example: [ ['OK'], ['value1'] ].
 
-redis.set('key1', 'value1').then(() => redis.get('key1')).then(console.log);
-```
+#### Examples
 
-2. Redis Cluster Usage (With `clusterPipeline`)
+1. Cluster Redis Usage
 
-In case you are using a Redis Cluster, you can now use the `clusterPipeline` method to handle pipelines across multiple nodes.
+When working with Redis Cluster, you can use the clusterPipeline method to efficiently handle multiple commands across different nodes.
 
 ```javascript
-import { Cluster } from '@pw-tech/ioredis';
+import { Cluster } from '@pw-tech/pw-redis';
 
 const clusterRedis = new Cluster({
   redisOptions: [{ host: 'localhost', port: 6379 }],
@@ -66,19 +80,40 @@ const commands = [
   ['get', 'key1'],
 ];
 
-clusterRedis.clusterPipeline(commands).then(results => {
-  console.log(results); // Output: [ ['OK'], ['value1'] ]
+clusterRedis.clusterPipeline(commands)
+```
+
+And the output will be:
+
+```javascript
+[[null,"OK"],[null,"value1"]]
+```
+
+
+2. Normal Usage
+
+```javascript
+import { Redis } from '@pw-tech/pw-redis';
+
+const redis = new Redis({
+  host: 'localhost',
+  port: 6379,
 });
+
+redis.set('key1', 'value1').then(() => redis.get('key1')).then(console.log);
 ```
 
 ## Explanation of `clusterPipeline`
 
-`clusterPipeline` takes an array of Redis commands (e.g., [['set', 'key', 'value'], ['get', 'key']]).
-It automatically divides the commands across the Redis Cluster nodes based on the slot of the key.
-It executes the commands in parallel and ensures the results are returned in the same order as the original commands.
+The `clusterPipeline` method takes an array of Redis commands (e.g., [['set', 'key', 'value'], ['get', 'key']]) and automatically distributes them across the appropriate Redis Cluster nodes based on key slots. It executes the commands in parallel, ensuring that the results are returned in the same order as the input commands.
 
 ## How it Works
 
+- **Slot Calculation**: Redis Cluster distributes keys across multiple nodes using hash slots. The library automatically determines the correct node for each key and groups commands accordingly, using the Redis `SHARDS` method to map and store the slot-to-node mapping in memory. In case of any slot changes in Redis, detected through error handling, the slot ranges are refreshed dynamically.
+- **Pipeline Execution**: Once the commands are grouped by node, they are executed in parallel, improving performance when processing large sets of commands.
+- **Result Handling**: The results from each node are collected and returned in the same order as the original set of commands, ensuring consistency and ease of use.
+
+
 - **Slot Calculation**: Redis Cluster splits keys across multiple nodes using slots. This library automatically calculates which node each key belongs to and groups the commands accordingly.
 - **Pipeline Execution**: Once the commands are grouped by node, they are executed in parallel on each node, ensuring better performance when dealing with large pipelines.
 - **Result Handling**: Results from each node are merged and returned in the same order as the original commands.
@@ -115,6 +150,43 @@ At PhysicsWallah Private Limited, we needed an efficient way to handle Redis Clu
 
 By adding the `clusterPipeline` method, we ensure commands are distributed to the appropriate Redis node, executed in parallel, and results are returned in the correct order. This approach eliminates the need to manually manage slot calculations and node selection.
 
+## Performance Improvement in Production
+
+After implementing partial pipeline optimization, we observed significant API response time improvements in production. Some queries had to remain sequential due to business logic dependencies, but wherever possible, parallel and pipeline execution were combined.
+
+### API 1
+
+| Date       | Avg Latency | P95 Latency | P99 Latency |
+|------------|-------------|-------------|-------------|
+| Without Pipeline  | 52ms        | 119ms       | 264ms       |
+| With Pipeline     | 42ms (↓19%) | 101ms (↓15%)| 224ms (↓15%)|
+
+#### Latency Comparison
+![API 1 Avg Latency Comparison](./images/api1_avg_latency.png)
+
+![API 1 P95 Latency Comparison](./images/api1_p95_latency.png)
+
+![API 1 P99 Latency Comparison](./images/api1_p99_latency.png)
+
+### API 2
+
+| Date       | Avg Latency | P95 Latency | P99 Latency |
+|------------|-------------|-------------|-------------|
+| Without Pipeline | 36ms        | 78ms        | 234ms       |
+| With Pipeline    | 31ms (↓14%) | 57ms (↓27%) | 202ms (↓13%)|
+
+#### Latency Comparison
+![API 2 Avg Latency Comparison](./images/api2_avg_latency.png)
+
+![API 2 P95 Latency Comparison](./images/api2_p95_latency.png)
+
+![API 2 P99 Latency Comparison](./images/api2_p99_latency.png)
+
+## Key Learnings & Next Steps
+- **Pipeline execution improved response times** significantly in cluster mode.
+- **Sequential logic in APIs limits optimization**, we will continue identifying more queries for batching.
+- **Next Steps**: Expanding this approach across more APIs & optimizing Redis cluster configurations.
+
 ## License
 
 This library is open-source and licensed under the MIT License.

dist/clusterPipeline.d.ts

@@ -1,6 +1,9 @@
 import Redis, { Cluster } from 'ioredis';
 declare class ExtendedClusterRedis extends Cluster {
-    clusterPipeline(commands: [string, ...any][]): Promise<any[]>;
+    private nodeSlotRanges;
+    private updateRedisClusterSlots;
+    clusterPipeline(commands: [string, ...any][]): Promise<[error: Error | null, result: unknown][] | null>;
+    private executePipelineForCluster;
     private calculateSlot;
     private hashCRC16;
     private findNodeForSlot;

dist/clusterPipeline.js

@@ -4,10 +4,13 @@ exports.Cluster = exports.Redis = void 0;
 const ioredis_1 = require("ioredis");
 exports.Redis = ioredis_1.default;
 class ExtendedClusterRedis extends ioredis_1.Cluster {
-    async clusterPipeline(commands) {
-        const pipelinesByNode = {};
+    constructor() {
+        super(...arguments);
+        this.nodeSlotRanges = [];
+    }
+    async updateRedisClusterSlots() {
         const clusterSlots = await this.cluster('SHARDS');
-        const nodeSlotRanges = clusterSlots.flatMap(([, slotRanges, , [node]]) => {
+        this.nodeSlotRanges = clusterSlots.flatMap(([, slotRanges, , [node]]) => {
             // slotRanges can contain multiple start-end pairs
             const ranges = [];
             for (let i = 0; i < slotRanges.length; i += 2) {
@@ -19,12 +22,28 @@ class ExtendedClusterRedis extends ioredis_1.Cluster {
             }
             return ranges;
         });
+    }
+    async clusterPipeline(commands) {
+        try {
+            let res = await this.executePipelineForCluster(commands);
+            return res;
+        }
+        catch (error) {
+            if (error instanceof Error && error.message.includes("slots")) {
+                await this.updateRedisClusterSlots();
+                return this.executePipelineForCluster(commands);
+            }
+            throw error;
+        }
+    }
+    async executePipelineForCluster(commands) {
+        const pipelinesByNode = {};
         // Group commands by node
         for (let i = 0; i < commands.length; i++) {
             const command = commands[i];
             const key = command[1];
             const slot = this.calculateSlot(key);
-            const node = this.findNodeForSlot(nodeSlotRanges, slot);
+            const node = this.findNodeForSlot(this.nodeSlotRanges, slot);
             if (!pipelinesByNode[node]) {
                 pipelinesByNode[node] = { commands: [], originalIndices: [] };
             }
@@ -33,7 +52,7 @@ class ExtendedClusterRedis extends ioredis_1.Cluster {
         }
         // Execute pipelines per node
         const results = [];
-        for (const node in pipelinesByNode) {
+        const promises = Object.keys(pipelinesByNode).map(async (node) => {
             const { commands, originalIndices } = pipelinesByNode[node];
             const pipeline = this.pipeline();
             commands.forEach(cmd => pipeline[cmd[0]](...cmd.slice(1)));
@@ -42,7 +61,8 @@ class ExtendedClusterRedis extends ioredis_1.Cluster {
                 const originalIndex = originalIndices[localIndex];
                 results[originalIndex] = result;
             });
-        }
+        });
+        await Promise.all(promises);
         return results;
     }
     calculateSlot(key) {