Merge pull request #61 from drift-labs/master

use multiple redis clients

package.json

@@ -6,7 +6,7 @@
   "license": "Apache-2.0",
   "dependencies": {
     "@coral-xyz/anchor": "^0.29.0",
-    "@drift-labs/sdk": "2.53.0-beta.5",
+    "@drift-labs/sdk": "2.53.0-beta.8",
     "@opentelemetry/api": "^1.1.0",
     "@opentelemetry/auto-instrumentations-node": "^0.31.1",
     "@opentelemetry/exporter-prometheus": "^0.31.0",

src/index.ts

@@ -56,9 +56,33 @@ import { RedisClient } from './utils/redisClient';
 
 require('dotenv').config();
 
-const REDIS_HOST = process.env.REDIS_HOST || 'localhost';
+// Reading in Redis env vars
-const REDIS_PORT = process.env.REDIS_PORT || '6379';
+const REDIS_HOSTS_ENV = (process.env.REDIS_HOSTS as string) || 'localhost';
-const REDIS_PASSWORD = process.env.REDIS_PASSWORD;
+const REDIS_HOSTS = REDIS_HOSTS_ENV.includes(',')
+	? REDIS_HOSTS_ENV.trim()
+			.replace(/^\[|\]$/g, '')
+			.split(/\s*,\s*/)
+	: [REDIS_HOSTS_ENV];
+
+const REDIS_PASSWORDS_ENV = (process.env.REDIS_PASSWORDS as string) || '';
+const REDIS_PASSWORDS = REDIS_PASSWORDS_ENV.includes(',')
+	? REDIS_PASSWORDS_ENV.trim()
+			.replace(/^\[|\]$/g, '')
+			.split(/\s*,\s*/)
+	: [REDIS_PASSWORDS_ENV];
+
+const REDIS_PORTS_ENV = (process.env.REDIS_PORTS as string) || '6379';
+const REDIS_PORTS = REDIS_PORTS_ENV.includes(',')
+	? REDIS_PORTS_ENV.trim()
+			.replace(/^\[|\]$/g, '')
+			.split(/\s*,\s*/)
+	: [REDIS_PORTS_ENV];
+if (
+	REDIS_PORTS.length !== REDIS_PASSWORDS.length ||
+	REDIS_PORTS.length !== REDIS_HOSTS.length
+) {
+	throw 'REDIS_HOSTS and REDIS_PASSWORDS and REDIS_PORTS must be the same length';
+}
 
 const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
 const commitHash = process.env.COMMIT;
@@ -69,6 +93,8 @@ const stateCommitment: Commitment = 'processed';
 const serverPort = process.env.PORT || 6969;
 export const ORDERBOOK_UPDATE_INTERVAL = 1000;
 const WS_FALLBACK_FETCH_INTERVAL = ORDERBOOK_UPDATE_INTERVAL * 10;
+const SLOT_STALENESS_TOLERANCE =
+	parseInt(process.env.SLOT_STALENESS_TOLERANCE) || 20;
 const useWebsocket = process.env.USE_WEBSOCKET?.toLowerCase() === 'true';
 const rateLimitCallsPerSecond = process.env.RATE_LIMIT_CALLS_PER_SECOND
 	? parseInt(process.env.RATE_LIMIT_CALLS_PER_SECOND)
@@ -320,11 +346,39 @@ const main = async () => {
 		`DLOBSubscriber initialized in ${Date.now() - initDlobSubscriberStart} ms`
 	);
 
-	let redisClient: RedisClient;
+	const redisClients: Array<RedisClient> = [];
+	const spotMarketRedisMap: Map<
+		number,
+		{ client: RedisClient; clientIndex: number }
+	> = new Map();
+	const perpMarketRedisMap: Map<
+		number,
+		{ client: RedisClient; clientIndex: number }
+	> = new Map();
 	if (useRedis) {
 		logger.info('Connecting to redis');
-		redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD);
+		for (let i = 0; i < REDIS_HOSTS.length; i++) {
-		await redisClient.connect();
+			redisClients.push(
+				new RedisClient(
+					REDIS_HOSTS[i],
+					REDIS_PORTS[i],
+					REDIS_PASSWORDS[i] || undefined
+				)
+			);
+			await redisClients[i].connect();
+		}
+		for (let i = 0; i < sdkConfig.SPOT_MARKETS.length; i++) {
+			spotMarketRedisMap.set(sdkConfig.SPOT_MARKETS[i].marketIndex, {
+				client: redisClients[0],
+				clientIndex: 0,
+			});
+		}
+		for (let i = 0; i < sdkConfig.PERP_MARKETS.length; i++) {
+			perpMarketRedisMap.set(sdkConfig.PERP_MARKETS[i].marketIndex, {
+				client: redisClients[0],
+				clientIndex: 0,
+			});
+		}
 	}
 
 	logger.info(`Initializing all market subscribers...`);
@@ -735,6 +789,7 @@ const main = async () => {
 					!grouping
 				) {
 					let redisL2: string;
+					const redisClient = perpMarketRedisMap.get(normedMarketIndex).client;
 					if (parseInt(adjustedDepth as string) === 5) {
 						redisL2 = await redisClient.client.get(
 							`last_update_orderbook_perp_${normedMarketIndex}_depth_5`
@@ -750,9 +805,24 @@ const main = async () => {
 					}
 					if (
 						redisL2 &&
-						dlobProvider.getSlot() - parseInt(JSON.parse(redisL2).slot) < 10
+						dlobProvider.getSlot() - parseInt(JSON.parse(redisL2).slot) <
-					)
+							SLOT_STALENESS_TOLERANCE
+					) {
 						l2Formatted = redisL2;
+					} else {
+						if (redisL2 && redisClients.length > 1) {
+							const nextClientIndex =
+								(perpMarketRedisMap.get(normedMarketIndex).clientIndex + 1) %
+								redisClients.length;
+							perpMarketRedisMap.set(normedMarketIndex, {
+								client: redisClients[nextClientIndex],
+								clientIndex: nextClientIndex,
+							});
+							console.log(
+								`Rotated redis client to index ${nextClientIndex} for perp market ${normedMarketIndex}`
+							);
+						}
+					}
 				} else if (
 					isSpot &&
 					`${includeSerum}`?.toLowerCase() === 'true' &&
@@ -761,6 +831,7 @@ const main = async () => {
 					!grouping
 				) {
 					let redisL2: string;
+					const redisClient = spotMarketRedisMap.get(normedMarketIndex).client;
 					if (parseInt(adjustedDepth as string) === 5) {
 						redisL2 = await redisClient.client.get(
 							`last_update_orderbook_spot_${normedMarketIndex}_depth_5`
@@ -776,9 +847,24 @@ const main = async () => {
 					}
 					if (
 						redisL2 &&
-						dlobProvider.getSlot() - parseInt(JSON.parse(redisL2).slot) < 10
+						dlobProvider.getSlot() - parseInt(JSON.parse(redisL2).slot) <
-					)
+							SLOT_STALENESS_TOLERANCE
+					) {
 						l2Formatted = redisL2;
+					} else {
+						if (redisL2 && redisClients.length > 1) {
+							const nextClientIndex =
+								(spotMarketRedisMap.get(normedMarketIndex).clientIndex + 1) %
+								redisClients.length;
+							spotMarketRedisMap.set(normedMarketIndex, {
+								client: redisClients[nextClientIndex],
+								clientIndex: nextClientIndex,
+							});
+							console.log(
+								`Rotated redis client to index ${nextClientIndex} for spot market ${normedMarketIndex}`
+							);
+						}
+					}
 				}
 
 				if (l2Formatted) {
@@ -917,6 +1003,8 @@ const main = async () => {
 							!normedParam['grouping']
 						) {
 							let redisL2: string;
+							const redisClient =
+								perpMarketRedisMap.get(normedMarketIndex).client;
 							if (parseInt(adjustedDepth as string) === 5) {
 								redisL2 = await redisClient.client.get(
 									`last_update_orderbook_perp_${normedMarketIndex}_depth_5`
@@ -932,8 +1020,26 @@ const main = async () => {
 							}
 							if (redisL2) {
 								const parsedRedisL2 = JSON.parse(redisL2);
-								if (dlobProvider.getSlot() - parseInt(parsedRedisL2.slot) < 10)
+								if (
+									dlobProvider.getSlot() - parseInt(parsedRedisL2.slot) <
+									SLOT_STALENESS_TOLERANCE
+								) {
 									l2Formatted = parsedRedisL2;
+								} else {
+									if (redisClients.length > 1) {
+										const nextClientIndex =
+											(perpMarketRedisMap.get(normedMarketIndex).clientIndex +
+												1) %
+											redisClients.length;
+										perpMarketRedisMap.set(normedMarketIndex, {
+											client: redisClients[nextClientIndex],
+											clientIndex: nextClientIndex,
+										});
+										console.log(
+											`Rotated redis client to index ${nextClientIndex} for perp market ${normedMarketIndex}`
+										);
+									}
+								}
 							}
 						} else if (
 							isSpot &&
@@ -942,6 +1048,8 @@ const main = async () => {
 							!normedParam['grouping']
 						) {
 							let redisL2: string;
+							const redisClient =
+								spotMarketRedisMap.get(normedMarketIndex).client;
 							if (parseInt(adjustedDepth as string) === 5) {
 								redisL2 = await redisClient.client.get(
 									`last_update_orderbook_spot_${normedMarketIndex}_depth_5`
@@ -957,8 +1065,26 @@ const main = async () => {
 							}
 							if (redisL2) {
 								const parsedRedisL2 = JSON.parse(redisL2);
-								if (dlobProvider.getSlot() - parseInt(parsedRedisL2.slot) < 10)
+								if (
+									dlobProvider.getSlot() - parseInt(parsedRedisL2.slot) <
+									SLOT_STALENESS_TOLERANCE
+								) {
 									l2Formatted = parsedRedisL2;
+								} else {
+									if (redisClients.length > 1) {
+										const nextClientIndex =
+											(spotMarketRedisMap.get(normedMarketIndex).clientIndex +
+												1) %
+											redisClients.length;
+										spotMarketRedisMap.set(normedMarketIndex, {
+											client: redisClients[nextClientIndex],
+											clientIndex: nextClientIndex,
+										});
+										console.log(
+											`Rotated redis client to index ${nextClientIndex} for spot market ${normedMarketIndex}`
+										);
+									}
+								}
 							}
 						}
 

src/publishers/dlobPublisher.ts

@@ -324,7 +324,8 @@ const main = async () => {
 	const initAllMarketSubscribersStart = Date.now();
 	MARKET_SUBSCRIBERS = await initializeAllMarketSubscribers(driftClient);
 	logger.info(
-		`All market subscribers initialized in ${Date.now() - initAllMarketSubscribersStart
+		`All market subscribers initialized in ${
+			Date.now() - initAllMarketSubscribersStart
 		} ms`
 	);
 

src/serverLite.ts

@@ -30,9 +30,33 @@ import {
 
 require('dotenv').config();
 
-const REDIS_HOST = process.env.REDIS_HOST || 'localhost';
+// Reading in Redis env vars
-const REDIS_PORT = process.env.REDIS_PORT || '6379';
+const REDIS_HOSTS_ENV = (process.env.REDIS_HOSTS as string) || 'localhost';
-const REDIS_PASSWORD = process.env.REDIS_PASSWORD;
+const REDIS_HOSTS = REDIS_HOSTS_ENV.includes(',')
+	? REDIS_HOSTS_ENV.trim()
+			.replace(/^\[|\]$/g, '')
+			.split(/\s*,\s*/)
+	: [REDIS_HOSTS_ENV];
+
+const REDIS_PASSWORDS_ENV = (process.env.REDIS_PASSWORDS as string) || '';
+const REDIS_PASSWORDS = REDIS_PASSWORDS_ENV.includes(',')
+	? REDIS_PASSWORDS_ENV.trim()
+			.replace(/^\[|\]$/g, '')
+			.split(/\s*,\s*/)
+	: [REDIS_PASSWORDS_ENV];
+
+const REDIS_PORTS_ENV = (process.env.REDIS_PORTS as string) || '6379';
+const REDIS_PORTS = REDIS_PORTS_ENV.includes(',')
+	? REDIS_PORTS_ENV.trim()
+			.replace(/^\[|\]$/g, '')
+			.split(/\s*,\s*/)
+	: [REDIS_PORTS_ENV];
+if (
+	REDIS_PORTS.length !== REDIS_PASSWORDS.length ||
+	REDIS_PORTS.length !== REDIS_HOSTS.length
+) {
+	throw 'REDIS_HOSTS and REDIS_PASSWORDS and REDIS_PORTS must be the same length';
+}
 
 const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
 const commitHash = process.env.COMMIT;
@@ -132,9 +156,37 @@ logger.info(`Commit:             ${commitHash}`);
 
 const main = async () => {
 	// Redis connect
-	logger.info('Connecting to redis');
+	const redisClients: Array<RedisClient> = [];
-	const redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD);
+	const spotMarketRedisMap: Map<
-	await redisClient.connect();
+		number,
+		{ client: RedisClient; clientIndex: number }
+	> = new Map();
+	const perpMarketRedisMap: Map<
+		number,
+		{ client: RedisClient; clientIndex: number }
+	> = new Map();
+	for (let i = 0; i < REDIS_HOSTS.length; i++) {
+		redisClients.push(
+			new RedisClient(
+				REDIS_HOSTS[i],
+				REDIS_PORTS[i],
+				REDIS_PASSWORDS[i] || undefined
+			)
+		);
+		await redisClients[i].connect();
+	}
+	for (let i = 0; i < sdkConfig.SPOT_MARKETS.length; i++) {
+		spotMarketRedisMap.set(sdkConfig.SPOT_MARKETS[i].marketIndex, {
+			client: redisClients[0],
+			clientIndex: 0,
+		});
+	}
+	for (let i = 0; i < sdkConfig.PERP_MARKETS.length; i++) {
+		perpMarketRedisMap.set(sdkConfig.PERP_MARKETS[i].marketIndex, {
+			client: redisClients[0],
+			clientIndex: 0,
+		});
+	}
 
 	// Slot subscriber for source
 	const connection = new Connection(endpoint, {
@@ -154,7 +206,13 @@ const main = async () => {
 	);
 
 	const handleStartup = async (_req, res, _next) => {
-		if (redisClient.connected) {
+		let healthy = false;
+		for (const redisClient of redisClients) {
+			if (redisClient.connected) {
+				healthy = true;
+			}
+		}
+		if (healthy) {
 			res.writeHead(200);
 			res.end('OK');
 		} else {
@@ -205,6 +263,7 @@ const main = async () => {
 				!grouping
 			) {
 				let redisL2: string;
+				const redisClient = perpMarketRedisMap.get(normedMarketIndex).client;
 				if (parseInt(adjustedDepth as string) === 5) {
 					redisL2 = await redisClient.client.get(
 						`last_update_orderbook_perp_${normedMarketIndex}_depth_5`
@@ -223,6 +282,19 @@ const main = async () => {
 						slotSubscriber.getSlot() - parseInt(JSON.parse(redisL2).slot);
 					if (slotDiff < SLOT_STALENESS_TOLERANCE) {
 						l2Formatted = redisL2;
+					} else {
+						if (redisClients.length > 1) {
+							const nextClientIndex =
+								(perpMarketRedisMap.get(normedMarketIndex).clientIndex + 1) %
+								redisClients.length;
+							perpMarketRedisMap.set(normedMarketIndex, {
+								client: redisClients[nextClientIndex],
+								clientIndex: nextClientIndex,
+							});
+							console.log(
+								`Rotated redis client to index ${nextClientIndex} for perp market ${normedMarketIndex}`
+							);
+						}
 					}
 				}
 			} else if (
@@ -233,6 +305,7 @@ const main = async () => {
 				!grouping
 			) {
 				let redisL2: string;
+				const redisClient = spotMarketRedisMap.get(normedMarketIndex).client;
 				if (parseInt(adjustedDepth as string) === 5) {
 					redisL2 = await redisClient.client.get(
 						`last_update_orderbook_spot_${normedMarketIndex}_depth_5`
@@ -251,6 +324,19 @@ const main = async () => {
 						slotSubscriber.getSlot() - parseInt(JSON.parse(redisL2).slot);
 					if (slotDiff < SLOT_STALENESS_TOLERANCE) {
 						l2Formatted = redisL2;
+					} else {
+						if (redisClients.length > 1) {
+							const nextClientIndex =
+								(spotMarketRedisMap.get(normedMarketIndex).clientIndex + 1) %
+								redisClients.length;
+							spotMarketRedisMap.set(normedMarketIndex, {
+								client: redisClients[nextClientIndex],
+								clientIndex: nextClientIndex,
+							});
+							console.log(
+								`Rotated redis client to index ${nextClientIndex} for spot market ${normedMarketIndex}`
+							);
+						}
 					}
 				}
 			}
@@ -341,6 +427,8 @@ const main = async () => {
 						!normedParam['grouping']
 					) {
 						let redisL2: string;
+						const redisClient =
+							perpMarketRedisMap.get(normedMarketIndex).client;
 						if (parseInt(adjustedDepth as string) === 5) {
 							redisL2 = await redisClient.client.get(
 								`last_update_orderbook_perp_${normedMarketIndex}_depth_5`
@@ -359,8 +447,23 @@ const main = async () => {
 							if (
 								slotSubscriber.getSlot() - parseInt(parsedRedisL2.slot) <
 								SLOT_STALENESS_TOLERANCE
-							)
+							) {
 								l2Formatted = parsedRedisL2;
+							} else {
+								if (redisClients.length > 1) {
+									const nextClientIndex =
+										(perpMarketRedisMap.get(normedMarketIndex).clientIndex +
+											1) %
+										redisClients.length;
+									perpMarketRedisMap.set(normedMarketIndex, {
+										client: redisClients[nextClientIndex],
+										clientIndex: nextClientIndex,
+									});
+									console.log(
+										`Rotated redis client to index ${nextClientIndex} for perp market ${normedMarketIndex}`
+									);
+								}
+							}
 						}
 					} else if (
 						isSpot &&
@@ -369,6 +472,8 @@ const main = async () => {
 						!normedParam['grouping']
 					) {
 						let redisL2: string;
+						const redisClient =
+							spotMarketRedisMap.get(normedMarketIndex).client;
 						if (parseInt(adjustedDepth as string) === 5) {
 							redisL2 = await redisClient.client.get(
 								`last_update_orderbook_spot_${normedMarketIndex}_depth_5`
@@ -387,8 +492,23 @@ const main = async () => {
 							if (
 								slotSubscriber.getSlot() - parseInt(parsedRedisL2.slot) <
 								SLOT_STALENESS_TOLERANCE
-							)
+							) {
 								l2Formatted = parsedRedisL2;
+							} else {
+								if (redisClients.length > 1) {
+									const nextClientIndex =
+										(spotMarketRedisMap.get(normedMarketIndex).clientIndex +
+											1) %
+										redisClients.length;
+									spotMarketRedisMap.set(normedMarketIndex, {
+										client: redisClients[nextClientIndex],
+										clientIndex: nextClientIndex,
+									});
+									console.log(
+										`Rotated redis client to index ${nextClientIndex} for spot market ${normedMarketIndex}`
+									);
+								}
+							}
 						}
 
 						if (l2Formatted) {

yarn.lock

@@ -115,10 +115,10 @@
     enabled "2.0.x"
     kuler "^2.0.0"
 
-"@drift-labs/sdk@2.53.0-beta.5":
+"@drift-labs/sdk@2.53.0-beta.8":
-  version "2.53.0-beta.5"
+  version "2.53.0-beta.8"
-  resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.53.0-beta.5.tgz#2cbd77105e3248886a345cb07a67fd60407eae3a"
+  resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.53.0-beta.8.tgz#bc12747a8f2c4fa26b979fa6a8202a3bd3f52b2b"
-  integrity sha512-EAW+eqSkAe490IblsC0/hFWeYFLU9wZ71yq1nw6Aemq0RC6ygsN/eaQiVMOO1v8MQ5ar/ef6n27FFHO4k5UOkg==
+  integrity sha512-R2OPzgWjEVrJMeSwXa/UiquM6c72iOnFdoHbXJ4JJPd/KJDvghjOIXshF+7ttf80Uwswrox0sEhCJOYcYZOMIQ==
   dependencies:
     "@coral-xyz/anchor" "0.28.1-beta.2"
     "@ellipsis-labs/phoenix-sdk" "^1.4.2"