Merge pull request #43 from drift-labs/master

use redis config and ws improvements

package.json

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

src/core/metrics.ts

@@ -40,6 +40,8 @@ enum METRIC_TYPES {
 	gpa_fetch_duration = 'gpa_fetch_duration',
 	last_ws_message_received_ts = 'last_ws_message_received_ts',
 	account_updates_count = 'account_updates_count',
+	cache_hit_count = 'cache_hit_count',
+	cache_miss_count = 'cache_miss_count',
 	current_system_ts = 'current_system_ts',
 	health_status = 'health_status',
 }
@@ -126,6 +128,10 @@ lastWsReceivedTsGauge.addCallback((obs: ObservableResult) => {
 	obs.observe(lastWsMsgReceivedTs, {});
 });
 
+const cacheHitCounter = meter.createCounter(METRIC_TYPES.cache_hit_count, {
+	description: 'Total redis cache hits',
+});
+
 const accountUpdatesCounter = meter.createCounter(
 	METRIC_TYPES.account_updates_count,
 	{
@@ -238,4 +244,5 @@ export {
 	handleHealthCheck,
 	setLastReceivedWsMsgTs,
 	accountUpdatesCounter,
+	cacheHitCounter,
 };

src/dlob-subscriber/DLOBSubscriberIO.ts

@@ -18,6 +18,7 @@ import {
 	addOracletoResponse,
 	l2WithBNToStrings,
 } from '../utils/utils';
+import { logger } from '../utils/logger';
 
 type wsMarketL2Args = {
 	marketIndex: number;
@@ -62,8 +63,8 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 				marketType: MarketType.PERP,
 				marketName: market.symbol,
 				depth: -1,
-				includeVamm: true,
 				numVammOrders: 100,
+				includeVamm: true,
 				updateOnChange: true,
 				fallbackL2Generators: [],
 			});
@@ -87,7 +88,12 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 	override async updateDLOB(): Promise<void> {
 		await super.updateDLOB();
 		for (const l2Args of this.marketL2Args) {
-			this.getL2AndSendMsg(l2Args);
+			try {
+				this.getL2AndSendMsg(l2Args);
+			} catch (error) {
+				logger.error(error);
+				console.log(`Error getting L2 ${l2Args.marketName}`);
+			}
 		}
 	}
 
@@ -124,13 +130,39 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 			l2Args.marketType,
 			l2Args.marketIndex
 		);
+
+		const l2Formatted_depth100 = Object.assign({}, l2Formatted, {
+			bids: l2Formatted.bids.slice(0, 100),
+			asks: l2Formatted.asks.slice(0, 100),
+		});
+		const l2Formatted_depth20 = Object.assign({}, l2Formatted, {
+			bids: l2Formatted.bids.slice(0, 20),
+			asks: l2Formatted.asks.slice(0, 20),
+		});
+		const l2Formatted_depth5 = Object.assign({}, l2Formatted, {
+			bids: l2Formatted.bids.slice(0, 5),
+			asks: l2Formatted.asks.slice(0, 5),
+		});
+
 		this.redisClient.client.publish(
 			`orderbook_${marketType}_${l2Args.marketIndex}`,
 			JSON.stringify(l2Formatted)
 		);
 		this.redisClient.client.set(
 			`last_update_orderbook_${marketType}_${l2Args.marketIndex}`,
-			JSON.stringify(l2Formatted)
+			JSON.stringify(l2Formatted_depth100)
+		);
+		this.redisClient.client.set(
+			`last_update_orderbook_${marketType}_${l2Args.marketIndex}_depth_100`,
+			JSON.stringify(l2Formatted_depth100)
+		);
+		this.redisClient.client.set(
+			`last_update_orderbook_${marketType}_${l2Args.marketIndex}_depth_20`,
+			JSON.stringify(l2Formatted_depth20)
+		);
+		this.redisClient.client.set(
+			`last_update_orderbook_${marketType}_${l2Args.marketIndex}_depth_5`,
+			JSON.stringify(l2Formatted_depth5)
 		);
 	}
 }

src/index.ts

@@ -28,8 +28,6 @@ import {
 	initialize,
 	isVariant,
 	OrderSubscriber,
-	UserAccount,
-	Order,
 } from '@drift-labs/sdk';
 
 import { logger, setLogLevel } from './utils/logger';
@@ -39,6 +37,7 @@ import {
 	gpaFetchDurationHistogram,
 	handleHealthCheck,
 	accountUpdatesCounter,
+	cacheHitCounter,
 	setLastReceivedWsMsgTs,
 } from './core/metrics';
 import { handleResponseTime } from './core/middleware';
@@ -59,8 +58,14 @@ import {
 	getDLOBProviderFromOrderSubscriber,
 	getDLOBProviderFromUserMap,
 } from './dlobProvider';
+import { RedisClient } from './utils/redisClient';
 
 require('dotenv').config();
+
+const REDIS_HOST = process.env.REDIS_HOST || 'localhost';
+const REDIS_PORT = process.env.REDIS_PORT || '6379';
+const REDIS_PASSWORD = process.env.REDIS_PASSWORD;
+
 const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
 const commitHash = process.env.COMMIT;
 //@ts-ignore
@@ -77,6 +82,7 @@ const rateLimitCallsPerSecond = process.env.RATE_LIMIT_CALLS_PER_SECOND
 	? parseInt(process.env.RATE_LIMIT_CALLS_PER_SECOND)
 	: 1;
 const loadTestAllowed = process.env.ALLOW_LOAD_TEST?.toLowerCase() === 'true';
+const useRedis = process.env.USE_REDIS?.toLowerCase() === 'true';
 
 const logFormat =
 	':remote-addr - :remote-user [:date[clf]] ":method :url HTTP/:http-version" :status :res[content-length] ":referrer" ":user-agent" :req[x-forwarded-for]';
@@ -264,6 +270,7 @@ const main = async () => {
 			'updateReceived',
 			(_pubkey: PublicKey, _slot: number, _dataType: 'raw' | 'decoded') => {
 				setLastReceivedWsMsgTs(Date.now());
+				// eslint-disable-next-line @typescript-eslint/no-unused-vars
 				updatesReceivedTotal++;
 				accountUpdatesCounter.add(1);
 			}
@@ -343,6 +350,13 @@ const main = async () => {
 		`DLOBSubscriber initialized in ${Date.now() - initDlobSubscriberStart} ms`
 	);
 
+	let redisClient: RedisClient;
+	if (useRedis) {
+		logger.info('Connecting to redis');
+		redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD);
+		await redisClient.connect();
+	}
+
 	logger.info(`Initializing all market subscribers...`);
 	const initAllMarketSubscribersStart = Date.now();
 	MARKET_SUBSCRIBERS = await initializeAllMarketSubscribers(driftClient);
@@ -742,6 +756,71 @@ const main = async () => {
 				adjustedDepth = '-1';
 			}
 
+			let l2Formatted: any;
+			if (useRedis) {
+				if (
+					!isSpot &&
+					`${includeVamm}`.toLowerCase() === 'true' &&
+					`${includeOracle}`.toLowerCase().toLowerCase() === 'true' &&
+					!grouping
+				) {
+					let redisL2: string;
+					if (parseInt(adjustedDepth as string) === 5) {
+						redisL2 = await redisClient.client.get(
+							`last_update_orderbook_perp_${normedMarketIndex}_depth_5`
+						);
+					} else if (parseInt(adjustedDepth as string) === 20) {
+						redisL2 = await redisClient.client.get(
+							`last_update_orderbook_perp_${normedMarketIndex}_depth_20`
+						);
+					} else if (parseInt(adjustedDepth as string) === 100) {
+						redisL2 = await redisClient.client.get(
+							`last_update_orderbook_perp_${normedMarketIndex}_depth_100`
+						);
+					}
+					if (
+						redisL2 &&
+						slotSource.getSlot() - parseInt(JSON.parse(redisL2).slot) < 10
+					)
+						l2Formatted = redisL2;
+				} else if (
+					isSpot &&
+					`${includeSerum}`.toLowerCase() === 'true' &&
+					`${includePhoenix}`.toLowerCase() === 'true' &&
+					`${includeOracle}`.toLowerCase() === 'true' &&
+					!grouping
+				) {
+					let redisL2: string;
+					if (parseInt(adjustedDepth as string) === 5) {
+						redisL2 = await redisClient.client.get(
+							`last_update_orderbook_spot_${normedMarketIndex}_depth_5`
+						);
+					} else if (parseInt(adjustedDepth as string) === 20) {
+						redisL2 = await redisClient.client.get(
+							`last_update_orderbook_spot_${normedMarketIndex}_depth_20`
+						);
+					} else if (parseInt(adjustedDepth as string) === 100) {
+						redisL2 = await redisClient.client.get(
+							`last_update_orderbook_spot_${normedMarketIndex}_depth_100`
+						);
+					}
+					if (
+						redisL2 &&
+						slotSource.getSlot() - parseInt(JSON.parse(redisL2).slot) < 10
+					)
+						l2Formatted = redisL2;
+				}
+
+				if (l2Formatted) {
+					cacheHitCounter.add(1, {
+						miss: false,
+					});
+					res.writeHead(200);
+					res.end(l2Formatted);
+					return;
+				}
+			}
+
 			const l2 = dlobSubscriber.getL2({
 				marketIndex: normedMarketIndex,
 				marketType: normedMarketType,
@@ -783,7 +862,7 @@ const main = async () => {
 				res.end(JSON.stringify(l2Formatted));
 			} else {
 				// make the BNs into strings
-				const l2Formatted = l2WithBNToStrings(l2);
+				l2Formatted = l2WithBNToStrings(l2);
 				if (`${includeOracle}`.toLowerCase() === 'true') {
 					addOracletoResponse(
 						l2Formatted,
@@ -792,10 +871,12 @@ const main = async () => {
 						normedMarketIndex
 					);
 				}
-
-				res.writeHead(200);
-				res.end(JSON.stringify(l2Formatted));
 			}
+			cacheHitCounter.add(1, {
+				miss: true,
+			});
+			res.writeHead(200);
+			res.end(JSON.stringify(l2Formatted));
 		} catch (err) {
 			next(err);
 		}
@@ -836,86 +917,152 @@ const main = async () => {
 				return;
 			}
 
-			const l2s = normedParams.map((normedParam) => {
+			const l2s = await Promise.all(
-				const { normedMarketType, normedMarketIndex, error } =
+				normedParams.map(async (normedParam) => {
-					validateDlobQuery(
+					const { normedMarketType, normedMarketIndex, error } =
-						driftClient,
+						validateDlobQuery(
-						driftEnv,
+							driftClient,
-						normedParam['marketType'] as string,
+							driftEnv,
-						normedParam['marketIndex'] as string,
+							normedParam['marketType'] as string,
-						normedParam['marketName'] as string
+							normedParam['marketIndex'] as string,
-					);
+							normedParam['marketName'] as string
-				if (error) {
+						);
-					res.status(400).send(`Bad Request: ${error}`);
+					if (error) {
-					return;
+						res.status(400).send(`Bad Request: ${error}`);
-				}
-
-				const isSpot = isVariant(normedMarketType, 'spot');
-
-				let adjustedDepth = normedParam['depth'] ?? '10';
-				if (normedParam['grouping'] !== undefined) {
-					// If grouping is also supplied, we want the entire book depth.
-					// we will apply depth after grouping
-					adjustedDepth = '-1';
-				}
-
-				const l2 = dlobSubscriber.getL2({
-					marketIndex: normedMarketIndex,
-					marketType: normedMarketType,
-					depth: parseInt(adjustedDepth as string),
-					includeVamm: isSpot
-						? false
-						: `${normedParam['includeVamm']}`.toLowerCase() === 'true',
-					fallbackL2Generators: isSpot
-						? [
-								`${normedParam['includePhoenix']}`.toLowerCase() === 'true' &&
-									MARKET_SUBSCRIBERS[normedMarketIndex].phoenix,
-								`${normedParam['includeSerum']}`.toLowerCase() === 'true' &&
-									MARKET_SUBSCRIBERS[normedMarketIndex].serum,
-						  ].filter((a) => !!a)
-						: [],
-				});
-
-				if (normedParam['grouping']) {
-					const finalDepth = normedParam['depth']
-						? parseInt(normedParam['depth'] as string)
-						: 10;
-					if (isNaN(parseInt(normedParam['grouping'] as string))) {
-						res
-							.status(400)
-							.send('Bad Request: grouping must be a number if supplied');
 						return;
 					}
-					const groupingBN = new BN(
-						parseInt(normedParam['grouping'] as string)
-					);
 
-					const l2Formatted = l2WithBNToStrings(
+					const isSpot = isVariant(normedMarketType, 'spot');
-						groupL2(l2, groupingBN, finalDepth)
+
-					);
+					let adjustedDepth = normedParam['depth'] ?? '10';
-					if (`${normedParam['includeOracle']}`.toLowerCase() === 'true') {
+					if (normedParam['grouping'] !== undefined) {
-						addOracletoResponse(
+						// If grouping is also supplied, we want the entire book depth.
-							l2Formatted,
+						// we will apply depth after grouping
-							driftClient,
+						adjustedDepth = '-1';
-							normedMarketType,
-							normedMarketIndex
-						);
 					}
-					return l2Formatted;
+
-				} else {
+					let l2Formatted: any;
-					// make the BNs into strings
+					if (useRedis) {
-					const l2Formatted = l2WithBNToStrings(l2);
+						if (
-					if (`${normedParam['includeOracle']}`.toLowerCase() === 'true') {
+							marketType === 'perp' &&
-						addOracletoResponse(
+							normedParam['includeVamm'].toLowerCase() === 'true' &&
-							l2Formatted,
+							normedParam['includeOracle'].toLowerCase() === 'true' &&
-							driftClient,
+							!normedParam['grouping']
-							normedMarketType,
+						) {
-							normedMarketIndex
+							let redisL2: string;
-						);
+							if (parseInt(adjustedDepth as string) === 5) {
+								redisL2 = await redisClient.client.get(
+									`last_update_orderbook_perp_${normedMarketIndex}_depth_5`
+								);
+							} else if (parseInt(adjustedDepth as string) === 20) {
+								redisL2 = await redisClient.client.get(
+									`last_update_orderbook_perp_${normedMarketIndex}_depth_20`
+								);
+							} else if (parseInt(adjustedDepth as string) === 100) {
+								redisL2 = await redisClient.client.get(
+									`last_update_orderbook_perp_${normedMarketIndex}_depth_100`
+								);
+							}
+							if (redisL2) {
+								const parsedRedisL2 = JSON.parse(redisL2);
+								if (slotSource.getSlot() - parseInt(parsedRedisL2.slot) < 10)
+									l2Formatted = parsedRedisL2;
+							}
+						} else if (
+							marketType === 'spot' &&
+							normedParam['includePhoenix'].toLowerCase() === 'true' &&
+							normedParam['includeSerum'].toLowerCase() === 'true' &&
+							!normedParam['grouping']
+						) {
+							let redisL2: string;
+							if (parseInt(adjustedDepth as string) === 5) {
+								redisL2 = await redisClient.client.get(
+									`last_update_orderbook_spot_${normedMarketIndex}_depth_5`
+								);
+							} else if (parseInt(adjustedDepth as string) === 20) {
+								redisL2 = await redisClient.client.get(
+									`last_update_orderbook_spot_${normedMarketIndex}_depth_20`
+								);
+							} else if (parseInt(adjustedDepth as string) === 100) {
+								redisL2 = await redisClient.client.get(
+									`last_update_orderbook_spot_${normedMarketIndex}_depth_100`
+								);
+							}
+							if (redisL2) {
+								const parsedRedisL2 = JSON.parse(redisL2);
+								if (slotSource.getSlot() - parseInt(parsedRedisL2.slot) < 10)
+									l2Formatted = parsedRedisL2;
+							}
+						}
+
+						if (l2Formatted) {
+							cacheHitCounter.add(1, {
+								miss: false,
+							});
+							return l2Formatted;
+						}
 					}
+
+					const l2 = dlobSubscriber.getL2({
+						marketIndex: normedMarketIndex,
+						marketType: normedMarketType,
+						depth: parseInt(adjustedDepth as string),
+						includeVamm: isSpot
+							? false
+							: `${normedParam['includeVamm']}`.toLowerCase() === 'true',
+						fallbackL2Generators: isSpot
+							? [
+									`${normedParam['includePhoenix']}`.toLowerCase() === 'true' &&
+										MARKET_SUBSCRIBERS[normedMarketIndex].phoenix,
+									`${normedParam['includeSerum']}`.toLowerCase() === 'true' &&
+										MARKET_SUBSCRIBERS[normedMarketIndex].serum,
+							  ].filter((a) => !!a)
+							: [],
+					});
+
+					if (normedParam['grouping']) {
+						const finalDepth = normedParam['depth']
+							? parseInt(normedParam['depth'] as string)
+							: 10;
+						if (isNaN(parseInt(normedParam['grouping'] as string))) {
+							res
+								.status(400)
+								.send('Bad Request: grouping must be a number if supplied');
+							return;
+						}
+						const groupingBN = new BN(
+							parseInt(normedParam['grouping'] as string)
+						);
+
+						l2Formatted = l2WithBNToStrings(
+							groupL2(l2, groupingBN, finalDepth)
+						);
+						if (`${normedParam['includeOracle']}`.toLowerCase() === 'true') {
+							addOracletoResponse(
+								l2Formatted,
+								driftClient,
+								normedMarketType,
+								normedMarketIndex
+							);
+						}
+					} else {
+						// make the BNs into strings
+						l2Formatted = l2WithBNToStrings(l2);
+						if (`${normedParam['includeOracle']}`.toLowerCase() === 'true') {
+							addOracletoResponse(
+								l2Formatted,
+								driftClient,
+								normedMarketType,
+								normedMarketIndex
+							);
+						}
+					}
+					cacheHitCounter.add(1, {
+						miss: true,
+					});
 					return l2Formatted;
-				}
+				})
-			});
+			);
 
 			res.writeHead(200);
 			res.end(JSON.stringify({ l2s }));

src/publishers/dlobPublisher.ts

@@ -36,8 +36,6 @@ import { GeyserOrderSubscriber } from '../grpc/OrderSubscriberGRPC';
 
 require('dotenv').config();
 const stateCommitment: Commitment = 'processed';
-const ORDERBOOK_UPDATE_INTERVAL = 400;
-const WS_FALLBACK_FETCH_INTERVAL = ORDERBOOK_UPDATE_INTERVAL * 10;
 const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
 const commitHash = process.env.COMMIT;
 const REDIS_HOST = process.env.REDIS_HOST || 'localhost';
@@ -62,6 +60,9 @@ const useOrderSubscriber =
 const useGrpc = process.env.USE_GRPC?.toLowerCase() === 'true';
 const useWebsocket = process.env.USE_WEBSOCKET?.toLowerCase() === 'true';
 
+const ORDERBOOK_UPDATE_INTERVAL = useGrpc ? 500 : 1000;
+const WS_FALLBACK_FETCH_INTERVAL = 10_000;
+
 logger.info(`RPC endpoint: ${endpoint}`);
 logger.info(`WS endpoint:  ${wsEndpoint}`);
 logger.info(`DriftEnv:     ${driftEnv}`);
@@ -148,6 +149,7 @@ const main = async () => {
 		accountSubscription = {
 			type: 'websocket',
 			commitment: stateCommitment,
+			resubTimeoutMs: 30_000,
 		};
 		slotSubscriber = new SlotSubscriber(connection);
 		await slotSubscriber.subscribe();

src/wsConnectionManager.ts

@@ -233,31 +233,9 @@ async function main() {
 			}
 		});
 
-		// Ping/pong connection timeout
-		let pongTimeoutId;
-		let isAlive = true;
-		const pingIntervalId = setInterval(() => {
-			isAlive = false;
-			pongTimeoutId = setTimeout(() => {
-				if (!isAlive) {
-					console.log('Disconnecting because of ping/pong timeout');
-					ws.terminate();
-				}
-			}, 5000); // 5 seconds to wait for a pong
-			ws.ping();
-		}, 30000);
-
-		// Listen for pong messages
-		ws.on('pong', () => {
-			isAlive = true;
-			clearTimeout(pongTimeoutId);
-		});
-
 		// Handle disconnection
 		ws.on('close', () => {
 			// Clear any existing intervals and timeouts
-			clearInterval(pingIntervalId);
-			clearTimeout(pongTimeoutId);
 			channelSubscribers.forEach((subscribers, channel) => {
 				if (subscribers.delete(ws) && subscribers.size === 0) {
 					redisClient.client.unsubscribe(channel);

yarn.lock

@@ -115,10 +115,10 @@
     enabled "2.0.x"
     kuler "^2.0.0"
 
-"@drift-labs/sdk@2.49.0-beta.17":
+"@drift-labs/sdk@2.52.0-beta.0":
-  version "2.49.0-beta.17"
+  version "2.52.0-beta.0"
-  resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.49.0-beta.17.tgz#59d39712dcd2d0e0e6905b42c8efb743178277f3"
+  resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.52.0-beta.0.tgz#03da2e192e3f2180238e5e464336e0fc1697c0e0"
-  integrity sha512-0OLJFxDlKxIsQ32MuqVp/oHJFaeCy1IMigLmU92Zlapvjd8euR73VueiF4rGGROwrqO4IIamPgf68cD7D1V2HA==
+  integrity sha512-SnuXh06oJFzAvTQZ/dOjcWduhKC8rRMlyTE5Xqb3LfCcEkQ/B+XvxTItCSzAPh33qq0MMhZKjxGdqKezUxnV1w==
   dependencies:
     "@coral-xyz/anchor" "0.28.1-beta.2"
     "@ellipsis-labs/phoenix-sdk" "^1.4.2"