Merge pull request #96 from drift-labs/cache-l3

cache l3 and add metric

src/core/metrics.ts

@@ -37,6 +37,7 @@ enum METRIC_TYPES {
 	cache_miss_count = 'cache_miss_count',
 	current_system_ts = 'current_system_ts',
 	health_status = 'health_status',
+	incoming_requests_count = 'incoming_requests_count',
 }
 
 export enum HEALTH_STATUS {
@@ -116,6 +117,13 @@ const cacheHitCounter = meter.createCounter(METRIC_TYPES.cache_hit_count, {
 	description: 'Total redis cache hits',
 });
 
+const incomingRequestsCounter = meter.createCounter(
+	METRIC_TYPES.incoming_requests_count,
+	{
+		description: 'Total incoming requests',
+	}
+);
+
 const accountUpdatesCounter = meter.createCounter(
 	METRIC_TYPES.account_updates_count,
 	{
@@ -225,6 +233,7 @@ export {
 	endpointResponseTimeHistogram,
 	gpaFetchDurationHistogram,
 	responseStatusCounter,
+	incomingRequestsCounter,
 	handleHealthCheck,
 	setLastReceivedWsMsgTs,
 	accountUpdatesCounter,

src/dlob-subscriber/DLOBSubscriberIO.ts

@@ -15,7 +15,7 @@ import {
 	l2WithBNToStrings,
 } from '../utils/utils';
 
-type wsMarketL2Args = {
+type wsMarketArgs = {
 	marketIndex: number;
 	marketType: MarketType;
 	marketName: string;
@@ -36,7 +36,7 @@ export type wsMarketInfo = {
 };
 
 export class DLOBSubscriberIO extends DLOBSubscriber {
-	public marketL2Args: wsMarketL2Args[] = [];
+	public marketArgs: wsMarketArgs[] = [];
 	public lastSeenL2Formatted: Map<MarketType, Map<number, any>>;
 	redisClient: RedisClient;
 	public killSwitchSlotDiffThreshold: number;
@@ -73,7 +73,7 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 			);
 			const includeVamm = !isVariant(perpMarket.status, 'ammPaused');
 
-			this.marketL2Args.push({
+			this.marketArgs.push({
 				marketIndex: market.marketIndex,
 				marketType: MarketType.PERP,
 				marketName: market.marketName,
@@ -85,7 +85,7 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 			});
 		}
 		for (const market of config.spotMarketInfos) {
-			this.marketL2Args.push({
+			this.marketArgs.push({
 				marketIndex: market.marketIndex,
 				marketType: MarketType.SPOT,
 				marketName: market.marketName,
@@ -102,70 +102,75 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 
 	override async updateDLOB(): Promise<void> {
 		await super.updateDLOB();
-		for (const l2Args of this.marketL2Args) {
+		for (const marketArgs of this.marketArgs) {
 			try {
-				this.getL2AndSendMsg(l2Args);
+				this.getL2AndSendMsg(marketArgs);
+				this.getL3AndSendMsg(marketArgs);
 			} catch (error) {
 				console.error(error);
-				console.log(`Error getting L2 ${l2Args.marketName}`);
+				console.log(`Error getting L2 ${marketArgs.marketName}`);
 			}
 		}
 	}
 
-	getL2AndSendMsg(l2Args: wsMarketL2Args): void {
+	getL2AndSendMsg(marketArgs: wsMarketArgs): void {
-		const grouping = l2Args.grouping;
+		const grouping = marketArgs.grouping;
-		const { marketName, ...l2FuncArgs } = l2Args;
+		const { marketName, ...l2FuncArgs } = marketArgs;
 		const l2 = this.getL2(l2FuncArgs);
 		const slot = l2.slot;
 		const lastMarketSlotAndTime = this.lastMarketSlotMap
-			.get(l2Args.marketType)
+			.get(marketArgs.marketType)
-			.get(l2Args.marketIndex);
+			.get(marketArgs.marketIndex);
 		if (!lastMarketSlotAndTime) {
 			this.lastMarketSlotMap
-				.get(l2Args.marketType)
+				.get(marketArgs.marketType)
-				.set(l2Args.marketIndex, { slot, ts: Date.now() });
+				.set(marketArgs.marketIndex, { slot, ts: Date.now() });
 		}
 
 		if (slot) {
 			delete l2.slot;
 		}
-		const marketType = isVariant(l2Args.marketType, 'perp') ? 'perp' : 'spot';
+		const marketType = isVariant(marketArgs.marketType, 'perp')
+			? 'perp'
+			: 'spot';
 		let l2Formatted: any;
 		if (grouping) {
 			const groupingBN = new BN(grouping);
-			l2Formatted = l2WithBNToStrings(groupL2(l2, groupingBN, l2Args.depth));
+			l2Formatted = l2WithBNToStrings(
+				groupL2(l2, groupingBN, marketArgs.depth)
+			);
 		} else {
 			l2Formatted = l2WithBNToStrings(l2);
 		}
 
-		if (l2Args.updateOnChange) {
+		if (marketArgs.updateOnChange) {
 			if (
 				this.lastSeenL2Formatted
-					.get(l2Args.marketType)
+					.get(marketArgs.marketType)
-					?.get(l2Args.marketIndex) === JSON.stringify(l2Formatted)
+					?.get(marketArgs.marketIndex) === JSON.stringify(l2Formatted)
 			)
 				return;
 		}
 
 		this.lastSeenL2Formatted
-			.get(l2Args.marketType)
+			.get(marketArgs.marketType)
-			?.set(l2Args.marketIndex, JSON.stringify(l2Formatted));
+			?.set(marketArgs.marketIndex, JSON.stringify(l2Formatted));
 		l2Formatted['marketName'] = marketName?.toUpperCase();
 		l2Formatted['marketType'] = marketType?.toLowerCase();
-		l2Formatted['marketIndex'] = l2Args.marketIndex;
+		l2Formatted['marketIndex'] = marketArgs.marketIndex;
 		l2Formatted['ts'] = Date.now();
 		l2Formatted['slot'] = slot;
 		addOracletoResponse(
 			l2Formatted,
 			this.driftClient,
-			l2Args.marketType,
+			marketArgs.marketType,
-			l2Args.marketIndex
+			marketArgs.marketIndex
 		);
 		addMarketSlotToResponse(
 			l2Formatted,
 			this.driftClient,
-			l2Args.marketType,
+			marketArgs.marketType,
-			l2Args.marketIndex
+			marketArgs.marketIndex
 		);
 
 		// Check if slot diffs are too large for oracle
@@ -200,9 +205,11 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 			l2Formatted['marketSlot'] !== lastMarketSlotAndTime.slot
 		) {
 			console.log(
-				`Updating market slot for ${l2Args.marketName} with slot ${l2Formatted['marketSlot']}`
+				`Updating market slot for ${marketArgs.marketName} with slot ${l2Formatted['marketSlot']}`
 			);
-			this.lastMarketSlotMap.get(l2Args.marketType).set(l2Args.marketIndex, {
+			this.lastMarketSlotMap
+				.get(marketArgs.marketType)
+				.set(marketArgs.marketIndex, {
 					slot: l2Formatted['marketSlot'],
 					ts: Date.now(),
 				});
@@ -222,24 +229,121 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 		});
 
 		this.redisClient.client.publish(
-			`orderbook_${marketType}_${l2Args.marketIndex}`,
+			`orderbook_${marketType}_${marketArgs.marketIndex}`,
 			JSON.stringify(l2Formatted)
 		);
 		this.redisClient.client.set(
-			`last_update_orderbook_${marketType}_${l2Args.marketIndex}`,
+			`last_update_orderbook_${marketType}_${marketArgs.marketIndex}`,
 			JSON.stringify(l2Formatted_depth100)
 		);
 		this.redisClient.client.set(
-			`last_update_orderbook_${marketType}_${l2Args.marketIndex}_depth_100`,
+			`last_update_orderbook_${marketType}_${marketArgs.marketIndex}_depth_100`,
 			JSON.stringify(l2Formatted_depth100)
 		);
 		this.redisClient.client.set(
-			`last_update_orderbook_${marketType}_${l2Args.marketIndex}_depth_20`,
+			`last_update_orderbook_${marketType}_${marketArgs.marketIndex}_depth_20`,
 			JSON.stringify(l2Formatted_depth20)
 		);
 		this.redisClient.client.set(
-			`last_update_orderbook_${marketType}_${l2Args.marketIndex}_depth_5`,
+			`last_update_orderbook_${marketType}_${marketArgs.marketIndex}_depth_5`,
 			JSON.stringify(l2Formatted_depth5)
 		);
 	}
+
+	getL3AndSendMsg(marketArgs: wsMarketArgs): void {
+		const { marketName, ...l2FuncArgs } = marketArgs;
+		const l3 = this.getL3(l2FuncArgs);
+		const slot = l3.slot;
+		const lastMarketSlotAndTime = this.lastMarketSlotMap
+			.get(marketArgs.marketType)
+			.get(marketArgs.marketIndex);
+		if (!lastMarketSlotAndTime) {
+			this.lastMarketSlotMap
+				.get(marketArgs.marketType)
+				.set(marketArgs.marketIndex, { slot, ts: Date.now() });
+		}
+
+		if (slot) {
+			delete l3.slot;
+		}
+		const marketType = isVariant(marketArgs.marketType, 'perp')
+			? 'perp'
+			: 'spot';
+
+		for (const key of Object.keys(l3)) {
+			for (const idx in l3[key]) {
+				const level = l3[key][idx];
+				l3[key][idx] = {
+					...level,
+					price: level.price.toString(),
+					size: level.size.toString(),
+				};
+			}
+		}
+
+		l3['marketName'] = marketName?.toUpperCase();
+		l3['marketType'] = marketType?.toLowerCase();
+		l3['marketIndex'] = marketArgs.marketIndex;
+		l3['ts'] = Date.now();
+		l3['slot'] = slot;
+		addOracletoResponse(
+			l3,
+			this.driftClient,
+			marketArgs.marketType,
+			marketArgs.marketIndex
+		);
+		addMarketSlotToResponse(
+			l3,
+			this.driftClient,
+			marketArgs.marketType,
+			marketArgs.marketIndex
+		);
+
+		// Check if slot diffs are too large for oracle
+		if (
+			Math.abs(slot - parseInt(l3['oracleData']['slot'])) >
+			this.killSwitchSlotDiffThreshold
+		) {
+			console.log(`Killing process due to slot diffs for market ${marketName}: 
+				dlobProvider slot: ${slot}
+				oracle slot: ${l3['oracleData']['slot']}
+			`);
+			process.exit(1);
+		}
+
+		// Check if times and slots are too different for market
+		const MAKRET_STALENESS_THRESHOLD =
+			marketType === 'perp'
+				? PERP_MAKRET_STALENESS_THRESHOLD
+				: SPOT_MAKRET_STALENESS_THRESHOLD;
+		if (
+			lastMarketSlotAndTime &&
+			l3['marketSlot'] === lastMarketSlotAndTime.slot &&
+			Date.now() - lastMarketSlotAndTime.ts > MAKRET_STALENESS_THRESHOLD
+		) {
+			console.log(`Killing process due to same slot for market ${marketName} after > ${MAKRET_STALENESS_THRESHOLD}ms: 
+				dlobProvider slot: ${slot}
+				market slot: ${l3['marketSlot']}
+			`);
+			process.exit(1);
+		} else if (
+			lastMarketSlotAndTime &&
+			l3['marketSlot'] !== lastMarketSlotAndTime.slot
+		) {
+			console.log(
+				`Updating market slot for ${marketArgs.marketName} with slot ${l3['marketSlot']}`
+			);
+			this.lastMarketSlotMap
+				.get(marketArgs.marketType)
+				.set(marketArgs.marketIndex, {
+					slot: l3['marketSlot'],
+					ts: Date.now(),
+				});
+		}
+
+		this.redisClient.client.set(
+			`last_update_orderbook_l3_${marketType}_${marketArgs.marketIndex}`,
+			JSON.stringify(l3)
+		);
+	}
 }

src/index.ts

@@ -36,6 +36,7 @@ import {
 	accountUpdatesCounter,
 	cacheHitCounter,
 	setLastReceivedWsMsgTs,
+	incomingRequestsCounter,
 	runtimeSpecsGauge,
 } from './core/metrics';
 import { handleResponseTime } from './core/middleware';
@@ -57,26 +58,29 @@ import { RedisClient } from './utils/redisClient';
 require('dotenv').config();
 
 // Reading in Redis env vars
-const REDIS_HOSTS_ENV = (process.env.REDIS_HOSTS as string) || 'localhost';
+const REDIS_HOSTS = process.env.REDIS_HOSTS?.replace(/^\[|\]$/g, '')
-const REDIS_HOSTS = REDIS_HOSTS_ENV.includes(',')
+	.split(',')
-	? REDIS_HOSTS_ENV.trim()
+	.map((host) => host.trim()) || ['localhost'];
-			.replace(/^\[|\]$/g, '')
+const REDIS_PORTS = process.env.REDIS_PORTS?.replace(/^\[|\]$/g, '')
-			.split(/\s*,\s*/)
+	.split(',')
-	: [REDIS_HOSTS_ENV];
+	.map((port) => parseInt(port.trim(), 10)) || [6379];
+const REDIS_PASSWORDS_ENV = process.env.REDIS_PASSWORDS || "['']";
 
-const REDIS_PASSWORDS_ENV = (process.env.REDIS_PASSWORDS as string) || '';
+let REDIS_PASSWORDS;
-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';
+if (REDIS_PASSWORDS_ENV.trim() === "['']") {
-const REDIS_PORTS = REDIS_PORTS_ENV.includes(',')
+	REDIS_PASSWORDS = [undefined];
-	? REDIS_PORTS_ENV.trim()
+} else {
-			.replace(/^\[|\]$/g, '')
+	REDIS_PASSWORDS = REDIS_PASSWORDS_ENV.replace(/^\[|\]$/g, '')
 		.split(/\s*,\s*/)
-	: [REDIS_PORTS_ENV];
+		.map((pwd) => pwd.replace(/(^'|'$)/g, '').trim())
+		.map((pwd) => (pwd === '' ? undefined : pwd));
+}
+
+console.log('Redis Hosts:', REDIS_HOSTS);
+console.log('Redis Ports:', REDIS_PORTS);
+console.log('Redis Passwords:', REDIS_PASSWORDS);
+
 if (
 	REDIS_PORTS.length !== REDIS_PASSWORDS.length ||
 	REDIS_PORTS.length !== REDIS_HOSTS.length
@@ -142,6 +146,7 @@ app.use((req, _res, next) => {
 			req.url = '/';
 		}
 	}
+	incomingRequestsCounter.add(1);
 	next();
 });
 
@@ -363,7 +368,7 @@ const main = async (): Promise<void> => {
 			redisClients.push(
 				new RedisClient(
 					REDIS_HOSTS[i],
-					REDIS_PORTS[i],
+					REDIS_PORTS[i].toString(),
 					REDIS_PASSWORDS[i] || undefined
 				)
 			);
@@ -1182,6 +1187,32 @@ const main = async (): Promise<void> => {
 				return;
 			}
 
+			const marketTypeStr = getVariant(normedMarketType);
+			if (useRedis) {
+				const redisClient = (
+					marketTypeStr === 'spot' ? spotMarketRedisMap : perpMarketRedisMap
+				).get(normedMarketIndex).client;
+				const redisL3 = await redisClient.client.get(
+					`last_update_orderbook_l3_${marketTypeStr}_${normedMarketIndex}`
+				);
+				if (
+					redisL3 &&
+					dlobProvider.getSlot() - parseInt(JSON.parse(redisL3).slot) <
+						SLOT_STALENESS_TOLERANCE
+				) {
+					cacheHitCounter.add(1, {
+						miss: false,
+					});
+					res.writeHead(200);
+					res.end(redisL3);
+					return;
+				} else {
+					cacheHitCounter.add(1, {
+						miss: true,
+					});
+				}
+			}
+
 			const l3 = dlobSubscriber.getL3({
 				marketIndex: normedMarketIndex,
 				marketType: normedMarketType,