Merge branch 'master' into staging

package.json

@@ -5,7 +5,7 @@
   "main": "lib/index.js",
   "license": "Apache-2.0",
   "dependencies": {
-    "@drift-labs/sdk": "2.48.0-beta.9",
+    "@drift-labs/sdk": "2.48.0-beta.13",
     "@opentelemetry/api": "^1.1.0",
     "@opentelemetry/auto-instrumentations-node": "^0.31.1",
     "@opentelemetry/exporter-prometheus": "^0.31.0",

src/core/metrics.ts

@@ -134,11 +134,10 @@ const handleHealthCheck = async (req, res, next) => {
 		if (lastHealthCheckState) {
 			res.writeHead(200);
 			res.end('OK');
-		} else {
+			lastHealthCheckPerformed = Date.now();
-			res.writeHead(500);
+			return;
-			res.end(`NOK`);
 		}
-		return;
+		// always check if last check was unhealthy (give it another chance to recover)
 	}
 
 	const { lastSlotReceived, lastSlotReceivedMutex } = getSlotHealthCheckInfo();
@@ -149,7 +148,9 @@ const handleHealthCheck = async (req, res, next) => {
 			lastHealthCheckState = lastSlotReceived > lastHealthCheckSlot;
 			if (!lastHealthCheckState) {
 				logger.error(
-					`Unhealthy: lastSlot: ${lastSlotReceived}, lastHealthCheckSlot: ${lastHealthCheckSlot}`
+					`Unhealthy: lastSlot: ${lastSlotReceived}, lastHealthCheckSlot: ${lastHealthCheckSlot}, timeSinceLastCheck: ${
+						Date.now() - lastHealthCheckPerformed
+					} ms`
 				);
 			}
 

src/dlobProvider.ts

@@ -0,0 +1,77 @@
+import { DLOB, OrderSubscriber, UserAccount, UserMap } from '@drift-labs/sdk';
+import { PublicKey } from '@solana/web3.js';
+
+export type DLOBProvider = {
+	subscribe(): Promise<void>;
+	getDLOB(slot: number): Promise<DLOB>;
+	getUniqueAuthorities(): PublicKey[];
+	getUserAccounts(): Generator<{
+		userAccount: UserAccount;
+		publicKey: PublicKey;
+	}>;
+	getUserAccount(publicKey: PublicKey): UserAccount | undefined;
+	size(): number;
+};
+
+export function getDLOBProviderFromUserMap(userMap: UserMap): DLOBProvider {
+	return {
+		subscribe: async () => {
+			await userMap.subscribe();
+		},
+		getDLOB: async (slot: number) => {
+			return await userMap.getDLOB(slot);
+		},
+		getUniqueAuthorities: () => {
+			return userMap.getUniqueAuthorities();
+		},
+		getUserAccounts: function* () {
+			for (const user of userMap.values()) {
+				yield {
+					userAccount: user.getUserAccount(),
+					publicKey: user.getUserAccountPublicKey(),
+				};
+			}
+		},
+		getUserAccount: (publicKey) => {
+			return userMap.get(publicKey.toString())?.getUserAccount();
+		},
+		size: () => {
+			return userMap.size();
+		},
+	};
+}
+
+export function getDLOBProviderFromOrderSubscriber(
+	orderSubscriber: OrderSubscriber
+): DLOBProvider {
+	return {
+		subscribe: async () => {
+			await orderSubscriber.subscribe();
+		},
+		getDLOB: async (slot: number) => {
+			return await orderSubscriber.getDLOB(slot);
+		},
+		getUniqueAuthorities: () => {
+			const authorities = new Set<PublicKey>();
+			for (const { userAccount } of orderSubscriber.usersAccounts.values()) {
+				authorities.add(userAccount.authority);
+			}
+			return Array.from(authorities.values());
+		},
+		getUserAccounts: function* () {
+			for (const [
+				key,
+				{ userAccount },
+			] of orderSubscriber.usersAccounts.entries()) {
+				yield { userAccount: userAccount, publicKey: new PublicKey(key) };
+			}
+		},
+		getUserAccount: (publicKey) => {
+			return orderSubscriber.usersAccounts.get(publicKey.toString())
+				?.userAccount;
+		},
+		size(): number {
+			return orderSubscriber.usersAccounts.size;
+		},
+	};
+}

src/index.ts

@@ -21,12 +21,13 @@ import {
 	SlotSource,
 	SlotSubscriber,
 	UserMap,
-	UserStatsMap,
 	Wallet,
+	getUserStatsAccountPublicKey,
 	getVariant,
 	groupL2,
 	initialize,
 	isVariant,
+	OrderSubscriber,
 } from '@drift-labs/sdk';
 
 import { logger, setLogLevel } from './utils/logger';
@@ -46,6 +47,11 @@ import {
 	sleep,
 	validateDlobQuery,
 } from './utils/utils';
+import {
+	DLOBProvider,
+	getDLOBProviderFromOrderSubscriber,
+	getDLOBProviderFromUserMap,
+} from './dlobProvider';
 
 require('dotenv').config();
 const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
@@ -57,6 +63,8 @@ const stateCommitment: Commitment = 'processed';
 const serverPort = process.env.PORT || 6969;
 export const ORDERBOOK_UPDATE_INTERVAL = 1000;
 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)
@@ -66,7 +74,7 @@ const loadTestAllowed = process.env.ALLOW_LOAD_TEST?.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]';
 const logHttp = morgan(logFormat, {
-	skip: (_req, res) => res.statusCode < 400,
+	skip: (_req, res) => res.statusCode <= 500,
 });
 
 let driftClient: DriftClient;
@@ -196,6 +204,7 @@ const main = async () => {
 			type: 'polling',
 			accountLoader: bulkAccountLoader,
 		};
+
 		slotSource = {
 			getSlot: () => bulkAccountLoader!.getSlot(),
 		};
@@ -206,6 +215,7 @@ const main = async () => {
 		};
 		slotSubscriber = new SlotSubscriber(connection);
 		await slotSubscriber.subscribe();
+
 		slotSource = {
 			getSlot: () => slotSubscriber!.getSlot(),
 		};
@@ -219,6 +229,40 @@ const main = async () => {
 		env: driftEnv,
 	});
 
+	let dlobProvider: DLOBProvider;
+	if (useOrderSubscriber) {
+		let subscriptionConfig;
+		if (useWebsocket) {
+			subscriptionConfig = {
+				type: 'websocket',
+			};
+		} else {
+			subscriptionConfig = {
+				type: 'polling',
+				frequency: ORDERBOOK_UPDATE_INTERVAL,
+			};
+		}
+
+		const orderSubscriber = new OrderSubscriber({
+			driftClient,
+			subscriptionConfig,
+		});
+
+		dlobProvider = getDLOBProviderFromOrderSubscriber(orderSubscriber);
+
+		slotSource = {
+			getSlot: () => orderSubscriber.getSlot(),
+		};
+	} else {
+		const userMap = new UserMap(
+			driftClient,
+			driftClient.userAccountSubscriptionConfig,
+			false
+		);
+
+		dlobProvider = getDLOBProviderFromUserMap(userMap);
+	}
+
 	const dlobCoder = DLOBOrdersCoder.create();
 
 	await driftClient.subscribe();
@@ -231,27 +275,19 @@ const main = async () => {
 		lastSlotReceived = slotSource.getSlot();
 	}, ORDERBOOK_UPDATE_INTERVAL);
 
-	const userStatsMap = new UserStatsMap(driftClient);
+	logger.info(`Initializing DLOB Provider...`);
-
+	const initDLOBProviderStart = Date.now();
-	logger.info(`Initializing userMap...`);
+	await dlobProvider.subscribe();
-	const initUserMapStart = Date.now();
+	logger.info(
-	const userMap = new UserMap(
+		`dlob provider initialized in ${Date.now() - initDLOBProviderStart} ms`
-		driftClient,
-		driftClient.userAccountSubscriptionConfig,
-		false,
-		async (authorities) => {
-			await userStatsMap.sync(authorities);
-		},
-		{ hasOpenOrders: true }
 	);
-	await userMap.subscribe();
+	logger.info(`dlob provider size ${dlobProvider.size()}`);
-	logger.info(`userMap initialized in ${Date.now() - initUserMapStart} ms`);
 
 	logger.info(`Initializing DLOBSubscriber...`);
 	const initDlobSubscriberStart = Date.now();
 	const dlobSubscriber = new DLOBSubscriber({
 		driftClient,
-		dlobSource: userMap,
+		dlobSource: dlobProvider,
 		slotSource,
 		updateFrequency: ORDERBOOK_UPDATE_INTERVAL,
 	});
@@ -263,11 +299,7 @@ const main = async () => {
 	MARKET_SUBSCRIBERS = await initializeAllMarketSubscribers(driftClient);
 
 	const handleStartup = async (_req, res, _next) => {
-		if (
+		if (driftClient.isSubscribed && dlobProvider.size() > 0) {
-			driftClient.isSubscribed &&
-			userMap.size() > 0 &&
-			userStatsMap.size() > 0
-		) {
 			res.writeHead(200);
 			res.end('OK');
 		} else {
@@ -297,16 +329,14 @@ const main = async () => {
 				});
 			}
 
-			for (const user of userMap.values()) {
+			for (const { userAccount, publicKey } of dlobProvider.getUserAccounts()) {
-				const userAccount = user.getUserAccount();
-
 				for (const order of userAccount.orders) {
 					if (isVariant(order.status, 'init')) {
 						continue;
 					}
 
 					orders.push({
-						user: user.getUserAccountPublicKey().toBase58(),
+						user: publicKey.toBase58(),
 						order: order,
 					});
 				}
@@ -352,9 +382,7 @@ const main = async () => {
 				}
 				oracles.push(oracleHuman);
 			}
-			for (const user of userMap.values()) {
+			for (const { userAccount, publicKey } of dlobProvider.getUserAccounts()) {
-				const userAccount = user.getUserAccount();
-
 				for (const order of userAccount.orders) {
 					if (isVariant(order.status, 'init')) {
 						continue;
@@ -392,7 +420,7 @@ const main = async () => {
 					}
 
 					orders.push({
-						user: user.getUserAccountPublicKey().toBase58(),
+						user: publicKey.toBase58(),
 						order: orderHuman,
 					});
 				}
@@ -416,16 +444,14 @@ const main = async () => {
 		try {
 			const dlobOrders: DLOBOrders = [];
 
-			for (const user of userMap.values()) {
+			for (const { userAccount, publicKey } of dlobProvider.getUserAccounts()) {
-				const userAccount = user.getUserAccount();
-
 				for (const order of userAccount.orders) {
 					if (isVariant(order.status, 'init')) {
 						continue;
 					}
 
 					dlobOrders.push({
-						user: user.getUserAccountPublicKey(),
+						user: publicKey,
 						order,
 					} as DLOBOrder);
 				}
@@ -466,9 +492,7 @@ const main = async () => {
 
 			const dlobOrders: DLOBOrders = [];
 
-			for (const user of userMap.values()) {
+			for (const { userAccount, publicKey } of dlobProvider.getUserAccounts()) {
-				const userAccount = user.getUserAccount();
-
 				for (const order of userAccount.orders) {
 					if (isVariant(order.status, 'init')) {
 						continue;
@@ -484,7 +508,7 @@ const main = async () => {
 					}
 
 					dlobOrders.push({
-						user: user.getUserAccountPublicKey(),
+						user: publicKey,
 						order,
 					} as DLOBOrder);
 				}
@@ -555,14 +579,15 @@ const main = async () => {
 							continue;
 						} else {
 							if (`${includeUserStats}`.toLowerCase() === 'true') {
-								const userAccount = side.userAccount.toBase58();
+								const userAccount = dlobProvider.getUserAccount(
-								await userMap.mustGet(userAccount);
+									side.userAccount
-								const userStats = await userStatsMap.mustGet(
-									userMap.getUserAuthority(userAccount)!.toBase58()
 								);
 								topMakers.add([
 									userAccount,
-									userStats.userStatsAccountPublicKey.toBase58(),
+									getUserStatsAccountPublicKey(
+										driftClient.program.programId,
+										userAccount.authority
+									),
 								]);
 							} else {
 								topMakers.add(side.userAccount.toBase58());

src/publishers/dlobPublisher.ts

@@ -6,16 +6,22 @@ import {
 	DriftClient,
 	initialize,
 	DriftEnv,
-	SlotSubscriber,
 	UserMap,
 	Wallet,
 	BulkAccountLoader,
+	OrderSubscriber,
+	SlotSource,
 } from '@drift-labs/sdk';
 
 import { logger, setLogLevel } from '../utils/logger';
 import { sleep } from '../utils/utils';
 import { DLOBSubscriberIO } from '../dlob-subscriber/DLOBSubscriberIO';
 import { RedisClient } from '../utils/redisClient';
+import {
+	DLOBProvider,
+	getDLOBProviderFromOrderSubscriber,
+	getDLOBProviderFromUserMap,
+} from '../dlobProvider';
 
 require('dotenv').config();
 const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
@@ -37,6 +43,9 @@ setLogLevel(opts.debug ? 'debug' : 'info');
 
 const endpoint = process.env.ENDPOINT;
 const wsEndpoint = process.env.WS_ENDPOINT;
+const useOrderSubscriber =
+	process.env.USE_ORDER_SUBSCRIBER?.toLowerCase() === 'true';
+
 logger.info(`RPC endpoint: ${endpoint}`);
 logger.info(`WS endpoint:  ${wsEndpoint}`);
 logger.info(`DriftEnv:     ${driftEnv}`);
@@ -68,8 +77,6 @@ const main = async () => {
 		env: driftEnv,
 	});
 
-	const slotSubscriber = new SlotSubscriber(connection, {});
-
 	const lamportsBalance = await connection.getBalance(wallet.publicKey);
 	logger.info(
 		`DriftClient ProgramId: ${driftClient.program.programId.toBase58()}`
@@ -83,22 +90,45 @@ const main = async () => {
 		logger.error(e);
 	});
 
-	await slotSubscriber.subscribe();
+	let slotSource: SlotSource;
+	let dlobProvider: DLOBProvider;
+	if (useOrderSubscriber) {
+		const orderSubscriber = new OrderSubscriber({
+			driftClient,
+			subscriptionConfig: {
+				type: 'polling',
+				frequency: ORDERBOOK_UPDATE_INTERVAL,
+			},
+		});
 
-	const userMap = new UserMap(
+		dlobProvider = getDLOBProviderFromOrderSubscriber(orderSubscriber);
-		driftClient,
+
-		driftClient.userAccountSubscriptionConfig,
+		slotSource = {
-		false
+			getSlot: () => orderSubscriber.getSlot(),
-	);
+		};
-	await userMap.subscribe();
+	} else {
+		const userMap = new UserMap(
+			driftClient,
+			driftClient.userAccountSubscriptionConfig,
+			false
+		);
+
+		dlobProvider = getDLOBProviderFromUserMap(userMap);
+
+		slotSource = {
+			getSlot: () => bulkAccountLoader.getSlot(),
+		};
+	}
+
+	await dlobProvider.subscribe();
 
 	const redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD);
 	await redisClient.connect();
 
 	const dlobSubscriber = new DLOBSubscriberIO({
 		driftClient,
-		dlobSource: userMap,
+		dlobSource: dlobProvider,
-		slotSource: slotSubscriber,
+		slotSource,
 		updateFrequency: ORDERBOOK_UPDATE_INTERVAL,
 		redisClient,
 	});

yarn.lock

@@ -80,10 +80,10 @@
     enabled "2.0.x"
     kuler "^2.0.0"
 
-"@drift-labs/sdk@2.48.0-beta.9":
+"@drift-labs/sdk@2.48.0-beta.13":
-  version "2.48.0-beta.9"
+  version "2.48.0-beta.13"
-  resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.48.0-beta.9.tgz#2d0c99f58387ff3c5e45859ae4fe088339c8dd55"
+  resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.48.0-beta.13.tgz#0ad1cfd4daac97e1859fe617a749a2676e462b63"
-  integrity sha512-y0wrq680nv7Uwvcg9jOrjZPXxoae9yRg8M9B4bQgDFeNJmZNmcEmnLhWQ+CbPi8u/O0glhOdtKUk4UF9eSPXAg==
+  integrity sha512-ogtY38Db44PVZOr8WwNHijnwVMBLFvIYNxmlE9BPehdEgIQ9PYT9QXalanhQ1Ruc/Cs5DbMIT3XvjtLyfFzEUg==
   dependencies:
     "@coral-xyz/anchor" "0.28.1-beta.2"
     "@ellipsis-labs/phoenix-sdk" "^1.4.2"