Use client pool for redis connections (#1247)

* feat(socket|redis): create a redis client pool for setex operations

* feat: use client pool for leave-room and bump pool size

* chore: remove refreshSlotsCache and error logs

Author: ecp4224

packages/sdk-socket-server-next/package.json

@@ -47,6 +47,7 @@
     "dotenv": "^16.3.1",
     "express": "^4.18.2",
     "express-rate-limit": "^7.1.5",
+    "generic-pool": "^3.9.0",
     "helmet": "^5.1.1",
     "ioredis": "^5.3.2",
     "logform": "^2.6.0",

packages/sdk-socket-server-next/src/analytics-api.ts

@@ -25,6 +25,7 @@ import {
   incrementAnalyticsEvents,
   incrementRedisCacheOperation,
 } from './metrics';
+import genericPool from "generic-pool";
 
 const logger = getLogger();
 
@@ -54,8 +55,6 @@ if (redisNodes.length === 0) {
   process.exit(1);
 }
 
-let redisClient: Cluster | Redis | undefined;
-
 export const getRedisOptions = (
   isTls: boolean,
   password: string | undefined,
@@ -79,15 +78,6 @@ export const getRedisOptions = (
       const targetErrors = [/MOVED/, /READONLY/, /ETIMEDOUT/];
 
       logger.error('Redis reconnect error:', error);
-      if (error.message.includes('MOVED') && redisClient instanceof Cluster) {
-        logger.error('Refreshing Redis Cluster slots cache');
-        try {
-          redisClient?.refreshSlotsCache();
-        } catch (error) {
-          logger.error('Error refreshing Redis Cluster slots cache:', error);
-        }
-      }
-
       return targetErrors.some((targetError) =>
         targetError.test(error.message),
       );
@@ -99,83 +89,97 @@ export const getRedisOptions = (
   return options;
 };
 
-export const getRedisClient = () => {
-  if (!redisClient) {
-    if (redisCluster) {
-      logger.info('Connecting to Redis Cluster...');
+export const buildRedisClient = (usePipelining: boolean = true) => {
+  let newRedisClient: Cluster | Redis | undefined;
 
-      const redisOptions = getRedisOptions(
-        redisTLS,
-        process.env.REDIS_PASSWORD,
-      );
-      const redisClusterOptions: ClusterOptions = {
-        dnsLookup: (address, callback) => callback(null, address),
-        scaleReads: 'slave',
-        slotsRefreshTimeout: 5000,
-        showFriendlyErrorStack: true,
-        slotsRefreshInterval: 2000,
-        clusterRetryStrategy: (times) => Math.min(times * 30, 1000),
-        enableAutoPipelining: true,
-        redisOptions,
-      };
+  if (redisCluster) {
+    logger.info('Connecting to Redis Cluster...');
 
-      logger.debug(
-        'Redis Cluster options:',
-        JSON.stringify(redisClusterOptions, null, 2),
-      );
+    const redisOptions = getRedisOptions(
+      redisTLS,
+      process.env.REDIS_PASSWORD,
+    );
+    const redisClusterOptions: ClusterOptions = {
+      dnsLookup: (address, callback) => callback(null, address),
+      scaleReads: 'slave',
+      slotsRefreshTimeout: 5000,
+      showFriendlyErrorStack: true,
+      slotsRefreshInterval: 2000,
+      clusterRetryStrategy: (times) => Math.min(times * 30, 1000),
+      enableAutoPipelining: usePipelining,
+      redisOptions,
+    };
 
-      redisClient = new Cluster(redisNodes, redisClusterOptions);
-    } else {
-      logger.info('Connecting to single Redis node');
-      redisClient = new Redis(redisNodes[0]);
-    }
+    logger.debug(
+      'Redis Cluster options:',
+      JSON.stringify(redisClusterOptions, null, 2),
+    );
+
+    newRedisClient = new Cluster(redisNodes, redisClusterOptions);
+  } else {
+    logger.info('Connecting to single Redis node');
+    newRedisClient = new Redis(redisNodes[0]);
   }
 
-  redisClient.on('ready', () => {
+  newRedisClient.on('ready', () => {
     logger.info('Redis ready');
-
-    if (redisClient instanceof Cluster) {
-      logger.error('Refreshing Redis Cluster slots cache');
-      try {
-        redisClient?.refreshSlotsCache();
-      } catch (error) {
-        logger.error('Error refreshing Redis Cluster slots cache:', error);
-      }
-    }
   });
 
-  redisClient.on('error', (error) => {
+  newRedisClient.on('error', (error) => {
     logger.error('Redis error:', error);
   });
 
-  redisClient.on('connect', () => {
+  newRedisClient.on('connect', () => {
     logger.info('Connected to Redis Cluster successfully');
   });
 
-  redisClient.on('close', () => {
+  newRedisClient.on('close', () => {
     logger.info('Disconnected from Redis Cluster');
   });
 
-  redisClient.on('reconnecting', () => {
+  newRedisClient.on('reconnecting', () => {
     logger.info('Reconnecting to Redis Cluster');
   });
 
-  redisClient.on('end', () => {
+  newRedisClient.on('end', () => {
     logger.info('Redis Cluster connection ended');
   });
 
-  redisClient.on('wait', () => {
+  newRedisClient.on('wait', () => {
     logger.info('Redis Cluster waiting for connection');
   });
 
-  redisClient.on('select', (node) => {
+  newRedisClient.on('select', (node) => {
     logger.info('Redis Cluster selected node:', node);
   });
 
+  return newRedisClient;
+}
+
+const redisFactory = {
+  create: () => {
+    return Promise.resolve(buildRedisClient(false));
+  },
+  destroy: (client: Cluster | Redis) => {
+    return Promise.resolve(client.disconnect());
+  },
+};
+
+let redisClient: Cluster | Redis | undefined;
+
+export const getGlobalRedisClient = () => {
+  if (!redisClient) {
+    redisClient = buildRedisClient();
+  }
+
   return redisClient;
 };
 
-export const pubClient = getRedisClient();
+export const pubClient = getGlobalRedisClient();
+export const pubClientPool = genericPool.createPool(redisFactory, {
+  max: 35,
+  min: 15,
+});
 
 const app = express();
 

packages/sdk-socket-server-next/src/protocol/handleChannelRejected.ts

@@ -1,5 +1,5 @@
 import { Server, Socket } from 'socket.io';
-import { pubClient } from '../analytics-api';
+import { pubClient, pubClientPool } from '../analytics-api';
 import { config } from '../config';
 import { getLogger } from '../logger';
 import { ChannelConfig } from './handleJoinChannel';
@@ -78,13 +78,17 @@ export const handleChannelRejected = async (
     },
   );
 
+  const client = await pubClientPool.acquire();
+
   // Update redis channel config to inform dApp of rejection
-  await pubClient.setex(
+  await client.setex(
     channelConfigKey,
     config.rejectedChannelExpiry,
     JSON.stringify(channelConfig),
   );
 
+  await pubClientPool.release(client);
+
   // Also broadcast to dapp if it is connected
   socket.broadcast.to(channelId).emit(`rejected-${channelId}`, { channelId });
 

packages/sdk-socket-server-next/src/protocol/handleJoinChannel.ts

@@ -1,7 +1,7 @@
 // protocol/handleJoinChannel.ts
 import { Server, Socket } from 'socket.io';
 import { validate } from 'uuid';
-import { pubClient } from '../analytics-api';
+import { pubClient, pubClientPool } from '../analytics-api';
 import { MAX_CLIENTS_PER_ROOM, config, isDevelopment } from '../config';
 import { getLogger } from '../logger';
 import { rateLimiter } from '../rate-limiter';
@@ -182,11 +182,15 @@ export const handleJoinChannel = async ({
           JSON.stringify(channelConfig),
         );
 
-        await pubClient.setex(
+        const client = await pubClientPool.acquire();
+
+        await client.setex(
           channelConfigKey,
           config.channelExpiry,
           JSON.stringify(channelConfig),
         ); // 1 week expiration
+
+        await pubClientPool.release(client);
       }
     }
 

packages/sdk-socket-server-next/src/redis-check.ts

@@ -2,7 +2,7 @@
 import dotenv from 'dotenv';
 // Dotenv must be loaded before importing local files
 dotenv.config();
-import { getRedisClient } from './analytics-api';
+import { getGlobalRedisClient } from './analytics-api';
 
 import { createLogger } from './logger';
 
@@ -34,7 +34,7 @@ if (redisNodes.length === 0) {
 async function testRedisOperations() {
   try {
     // Connect to Redis
-    const cluster = getRedisClient();
+    const cluster = getGlobalRedisClient();
     logger.info('Connected to Redis Cluster successfully');
 
     // Set a key in Redis

packages/sdk-socket-server-next/src/socket-config.ts

@@ -6,7 +6,7 @@ import { createAdapter } from '@socket.io/redis-adapter';
 
 import { Server, Socket } from 'socket.io';
 import { validate } from 'uuid';
-import { pubClient } from './analytics-api';
+import { pubClient, pubClientPool } from './analytics-api';
 import { getLogger } from './logger';
 import { ACKParams, handleAck } from './protocol/handleAck';
 import {
@@ -95,12 +95,14 @@ export const configureSocketServer = async (
       // Ignore invalid room IDs
       return;
     }
+  
+    const client = await pubClientPool.acquire();
 
     // Force keys into the same hash slot in Redis Cluster, using a hash tag (a substring enclosed in curly braces {})
     const channelOccupancyKey = `channel_occupancy:{${roomId}}`;
 
     // Decrement the number of clients in the room
-    const channelOccupancy = await pubClient.incrby(channelOccupancyKey, -1);
+    const channelOccupancy = await client.incrby(channelOccupancyKey, -1);
 
     logger.debug(
       `'leave-room' socket ${socketId} has left room ${roomId} --> channelOccupancy=${channelOccupancy}`,
@@ -112,14 +114,16 @@ export const configureSocketServer = async (
       const channelOccupancyKey = `channel_occupancy:{${roomId}}`;
 
       // remove from redis
-      await pubClient.del(channelOccupancyKey);
+      await client.del(channelOccupancyKey);
     } else {
       logger.info(
         `'leave-room' Room ${roomId} kept alive with ${channelOccupancy} clients`,
       );
       // Inform the room of the disconnection
       io.to(roomId).emit(`clients_disconnected-${roomId}`);
     }
+
+    await pubClientPool.release(client);
   });
 
   io.on('connection', (socket: Socket) => {

yarn.lock

@@ -11613,6 +11613,7 @@ __metadata:
     eslint-plugin-prettier: ^3.4.0
     express: ^4.18.2
     express-rate-limit: ^7.1.5
+    generic-pool: ^3.9.0
     helmet: ^5.1.1
     ioredis: ^5.3.2
     jest: ^29.6.4
@@ -32806,7 +32807,7 @@ __metadata:
   languageName: node
   linkType: hard
 
-"generic-pool@npm:3.9.0":
+"generic-pool@npm:3.9.0, generic-pool@npm:^3.9.0":
   version: 3.9.0
   resolution: "generic-pool@npm:3.9.0"
   checksum: 3d89e9b2018d2e3bbf44fec78c76b2b7d56d6a484237aa9daf6ff6eedb14b0899dadd703b5d810219baab2eb28e5128fb18b29e91e602deb2eccac14492d8ca8