Merge branch 'master' into staging

package.json

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

src/dlobProvider.ts

@@ -13,6 +13,7 @@ export type DLOBProvider = {
 	getUserAccount(publicKey: PublicKey): UserAccount | undefined;
 	size(): number;
 	fetch(): Promise<void>;
+	getSlot(): number;
 };
 
 export function getDLOBProviderFromUserMap(userMap: UserMap): DLOBProvider {
@@ -43,6 +44,9 @@ export function getDLOBProviderFromUserMap(userMap: UserMap): DLOBProvider {
 		fetch: () => {
 			return userMap.sync();
 		},
+		getSlot: () => {
+			return userMap.getSlot();
+		},
 	};
 }
 
@@ -82,6 +86,9 @@ export function getDLOBProviderFromOrderSubscriber(
 		fetch() {
 			return orderSubscriber.fetch();
 		},
+		getSlot: () => {
+			return orderSubscriber.getSlot();
+		},
 	};
 }
 
@@ -121,5 +128,8 @@ export function getDLOBProviderFromGrpcOrderSubscriber(
 		fetch() {
 			return orderSubscriber.fetch();
 		},
+		getSlot: () => {
+			return orderSubscriber.getSlot();
+		},
 	};
 }

src/index.ts

@@ -18,9 +18,7 @@ import {
 	DriftClient,
 	DriftClientSubscriptionConfig,
 	DriftEnv,
-	SlotSource,
 	SlotSubscriber,
-	UserMap,
 	Wallet,
 	getUserStatsAccountPublicKey,
 	getVariant,
@@ -34,7 +32,6 @@ import { logger, setLogLevel } from './utils/logger';
 
 import * as http from 'http';
 import {
-	gpaFetchDurationHistogram,
 	handleHealthCheck,
 	accountUpdatesCounter,
 	cacheHitCounter,
@@ -54,11 +51,7 @@ import {
 	validateDlobQuery,
 } from './utils/utils';
 import FEATURE_FLAGS from './utils/featureFlags';
-import {
+import { getDLOBProviderFromOrderSubscriber } from './dlobProvider';
-	DLOBProvider,
-	getDLOBProviderFromOrderSubscriber,
-	getDLOBProviderFromUserMap,
-} from './dlobProvider';
 import { RedisClient } from './utils/redisClient';
 
 require('dotenv').config();
@@ -77,8 +70,6 @@ const serverPort = process.env.PORT || 6969;
 export const ORDERBOOK_UPDATE_INTERVAL = 1000;
 const WS_FALLBACK_FETCH_INTERVAL = ORDERBOOK_UPDATE_INTERVAL * 10;
 const useWebsocket = process.env.USE_WEBSOCKET?.toLowerCase() === 'true';
-const useOrderSubscriber =
-	process.env.USE_ORDER_SUBSCRIBER?.toLowerCase() === 'true';
 const rateLimitCallsPerSecond = process.env.RATE_LIMIT_CALLS_PER_SECOND
 	? parseInt(process.env.RATE_LIMIT_CALLS_PER_SECOND)
 	: 1;
@@ -155,7 +146,6 @@ const wsEndpoint = process.env.WS_ENDPOINT;
 logger.info(`RPC endpoint:       ${endpoint}`);
 logger.info(`WS endpoint:        ${wsEndpoint}`);
 logger.info(`useWebsocket:       ${useWebsocket}`);
-logger.info(`useOrderSubscriber: ${useOrderSubscriber}`);
 logger.info(`DriftEnv:           ${driftEnv}`);
 logger.info(`Commit:             ${commitHash}`);
 
@@ -236,7 +226,6 @@ const main = async () => {
 	let slotSubscriber: SlotSubscriber | undefined;
 
 	let accountSubscription: DriftClientSubscriptionConfig;
-	let slotSource: SlotSource;
 
 	if (!useWebsocket) {
 		bulkAccountLoader = new BulkAccountLoader(
@@ -249,10 +238,6 @@ const main = async () => {
 			type: 'polling',
 			accountLoader: bulkAccountLoader,
 		};
-
-		slotSource = {
-			getSlot: () => bulkAccountLoader!.getSlot(),
-		};
 	} else {
 		accountSubscription = {
 			type: 'websocket',
@@ -260,10 +245,6 @@ const main = async () => {
 		};
 		slotSubscriber = new SlotSubscriber(connection);
 		await slotSubscriber.subscribe();
-
-		slotSource = {
-			getSlot: () => slotSubscriber!.getSlot(),
-		};
 	}
 
 	driftClient = new DriftClient({
@@ -274,57 +255,38 @@ const main = async () => {
 		env: driftEnv,
 	});
 
-	let dlobProvider: DLOBProvider;
+	let subscriptionConfig;
-	if (useOrderSubscriber) {
+	if (useWebsocket) {
-		let subscriptionConfig;
+		subscriptionConfig = {
-		if (useWebsocket) {
+			type: 'websocket',
-			subscriptionConfig = {
+			commitment: stateCommitment,
-				type: 'websocket',
+			resyncIntervalMs: WS_FALLBACK_FETCH_INTERVAL,
-				commitment: stateCommitment,
-			};
-		} else {
-			subscriptionConfig = {
-				type: 'polling',
-				frequency: ORDERBOOK_UPDATE_INTERVAL,
-				commitment: stateCommitment,
-			};
-		}
-
-		let updatesReceivedTotal = 0;
-		const orderSubscriber = new OrderSubscriber({
-			driftClient,
-			subscriptionConfig,
-		});
-		orderSubscriber.eventEmitter.on(
-			'updateReceived',
-			(_pubkey: PublicKey, _slot: number, _dataType: 'raw' | 'decoded') => {
-				setLastReceivedWsMsgTs(Date.now());
-				// eslint-disable-next-line @typescript-eslint/no-unused-vars
-				updatesReceivedTotal++;
-				accountUpdatesCounter.add(1);
-			}
-		);
-
-		dlobProvider = getDLOBProviderFromOrderSubscriber(orderSubscriber);
-
-		slotSource = {
-			getSlot: () => orderSubscriber.getSlot(),
 		};
 	} else {
-		const userMap = new UserMap({
+		subscriptionConfig = {
-			driftClient,
+			type: 'polling',
-			subscriptionConfig: {
+			frequency: ORDERBOOK_UPDATE_INTERVAL,
-				type: 'websocket',
+			commitment: stateCommitment,
-				resubTimeoutMs: 30_000,
+		};
-				commitment: stateCommitment,
-			},
-			skipInitialLoad: false,
-			includeIdle: false,
-		});
-
-		dlobProvider = getDLOBProviderFromUserMap(userMap);
 	}
 
+	let updatesReceivedTotal = 0;
+	const orderSubscriber = new OrderSubscriber({
+		driftClient,
+		subscriptionConfig,
+	});
+	orderSubscriber.eventEmitter.on(
+		'updateReceived',
+		(_pubkey: PublicKey, _slot: number, _dataType: 'raw' | 'decoded') => {
+			setLastReceivedWsMsgTs(Date.now());
+			// eslint-disable-next-line @typescript-eslint/no-unused-vars
+			updatesReceivedTotal++;
+			accountUpdatesCounter.add(1);
+		}
+	);
+
+	const dlobProvider = getDLOBProviderFromOrderSubscriber(orderSubscriber);
+
 	const dlobCoder = DLOBOrdersCoder.create();
 
 	await driftClient.subscribe();
@@ -344,34 +306,13 @@ const main = async () => {
 	logger.info(
 		`GPA refresh?: ${useWebsocket && !FEATURE_FLAGS.DISABLE_GPA_REFRESH}`
 	);
-	if (useWebsocket && !FEATURE_FLAGS.DISABLE_GPA_REFRESH) {
-		const recursiveFetch = (delay = WS_FALLBACK_FETCH_INTERVAL) => {
-			setTimeout(() => {
-				const startFetch = Date.now();
-				dlobProvider
-					.fetch()
-					.then(() => {
-						gpaFetchDurationHistogram.record(Date.now() - startFetch);
-					})
-					.catch((e) => {
-						logger.error('Failed to fetch GPA');
-						console.log(e);
-					})
-					.finally(() => {
-						// eslint-disable-next-line @typescript-eslint/no-unused-vars
-						recursiveFetch();
-					});
-			}, delay);
-		};
-		recursiveFetch();
-	}
 
 	logger.info(`Initializing DLOBSubscriber...`);
 	const initDlobSubscriberStart = Date.now();
 	const dlobSubscriber = new DLOBSubscriber({
 		driftClient,
 		dlobSource: dlobProvider,
-		slotSource,
+		slotSource: dlobProvider,
 		updateFrequency: ORDERBOOK_UPDATE_INTERVAL,
 	});
 	await dlobSubscriber.subscribe();
@@ -407,10 +348,10 @@ const main = async () => {
 
 	app.get(
 		'/health',
-		handleHealthCheck(2 * WS_FALLBACK_FETCH_INTERVAL, slotSource)
+		handleHealthCheck(2 * WS_FALLBACK_FETCH_INTERVAL, dlobProvider)
 	);
 	app.get('/startup', handleStartup);
-	app.get('/', handleHealthCheck(2 * WS_FALLBACK_FETCH_INTERVAL, slotSource));
+	app.get('/', handleHealthCheck(2 * WS_FALLBACK_FETCH_INTERVAL, dlobProvider));
 
 	if (FEATURE_FLAGS.ENABLE_ORDERS_ENDPOINTS) {
 		app.get('/orders/json/raw', async (_req, res, next) => {
@@ -418,7 +359,7 @@ const main = async () => {
 				// object with userAccount key and orders object serialized
 				const orders: Array<any> = [];
 				const oracles: Array<any> = [];
-				const slot = slotSource.getSlot();
+				const slot = dlobProvider.getSlot();
 
 				for (const market of driftClient.getPerpMarketAccounts()) {
 					const oracle = driftClient.getOracleDataForPerpMarket(
@@ -463,7 +404,7 @@ const main = async () => {
 		app.get('/orders/json', async (_req, res, next) => {
 			try {
 				// object with userAccount key and orders object serialized
-				const slot = slotSource.getSlot();
+				const slot = dlobProvider.getSlot();
 				const orders: Array<any> = [];
 				const oracles: Array<any> = [];
 				for (const market of driftClient.getPerpMarketAccounts()) {
@@ -633,7 +574,7 @@ const main = async () => {
 
 				res.end(
 					JSON.stringify({
-						slot: slotSource.getSlot(),
+						slot: dlobProvider.getSlot(),
 						data: dlobCoder.encode(dlobOrders).toString('base64'),
 					})
 				);
@@ -724,7 +665,7 @@ const main = async () => {
 						.getDLOB()
 						.getRestingLimitBids(
 							normedMarketIndex,
-							slotSource.getSlot(),
+							dlobProvider.getSlot(),
 							normedMarketType,
 							oracle
 						)
@@ -735,7 +676,7 @@ const main = async () => {
 						.getDLOB()
 						.getRestingLimitAsks(
 							normedMarketIndex,
-							slotSource.getSlot(),
+							dlobProvider.getSlot(),
 							normedMarketType,
 							oracle
 						)
@@ -809,7 +750,7 @@ const main = async () => {
 					}
 					if (
 						redisL2 &&
-						slotSource.getSlot() - parseInt(JSON.parse(redisL2).slot) < 10
+						dlobProvider.getSlot() - parseInt(JSON.parse(redisL2).slot) < 10
 					)
 						l2Formatted = redisL2;
 				} else if (
@@ -835,7 +776,7 @@ const main = async () => {
 					}
 					if (
 						redisL2 &&
-						slotSource.getSlot() - parseInt(JSON.parse(redisL2).slot) < 10
+						dlobProvider.getSlot() - parseInt(JSON.parse(redisL2).slot) < 10
 					)
 						l2Formatted = redisL2;
 				}
@@ -991,7 +932,7 @@ const main = async () => {
 							}
 							if (redisL2) {
 								const parsedRedisL2 = JSON.parse(redisL2);
-								if (slotSource.getSlot() - parseInt(parsedRedisL2.slot) < 10)
+								if (dlobProvider.getSlot() - parseInt(parsedRedisL2.slot) < 10)
 									l2Formatted = parsedRedisL2;
 							}
 						} else if (
@@ -1016,7 +957,7 @@ const main = async () => {
 							}
 							if (redisL2) {
 								const parsedRedisL2 = JSON.parse(redisL2);
-								if (slotSource.getSlot() - parseInt(parsedRedisL2.slot) < 10)
+								if (dlobProvider.getSlot() - parseInt(parsedRedisL2.slot) < 10)
 									l2Formatted = parsedRedisL2;
 							}
 						}

yarn.lock

@@ -115,10 +115,10 @@
     enabled "2.0.x"
     kuler "^2.0.0"
 
-"@drift-labs/sdk@2.52.0-beta.7":
+"@drift-labs/sdk@2.52.0-beta.8":
-  version "2.52.0-beta.7"
+  version "2.52.0-beta.8"
-  resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.52.0-beta.7.tgz#9fb2cb24d140e0d66b27e6bd24ae2268a6d8a3b5"
+  resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.52.0-beta.8.tgz#87a44cb244affdf6c5a5e551cd16e33b9d38fd9c"
-  integrity sha512-12I84O5EGU78fPWza+DMfqD23Zo2xMtsJbwyz6nrSKd840rGvI8KCVGnK7hv18dI/3uVmvdp79Mlg60LHieK0g==
+  integrity sha512-r0qX2jrwXefw4SRPvJ0VqV1t61m9smmSdTjZcGFeYAj/zcROHfBh4htE1iCTO+E+yHocMELHZzHM1DkdP2aRDQ==
   dependencies:
     "@coral-xyz/anchor" "0.28.1-beta.2"
     "@ellipsis-labs/phoenix-sdk" "^1.4.2"