Merge branch 'master' into mainnet-beta

package.json

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

src/dlob-subscriber/DLOBSubscriberIO.ts

@@ -2,24 +2,17 @@ import {
 	BN,
 	DLOBSubscriber,
 	DLOBSubscriptionConfig,
-	DevnetPerpMarkets,
-	DevnetSpotMarkets,
 	L2OrderBookGenerator,
-	MainnetPerpMarkets,
-	MainnetSpotMarkets,
 	MarketType,
 	groupL2,
 	isVariant,
 } from '@drift-labs/sdk';
-import { driftEnv } from '../publishers/dlobPublisher';
 import { RedisClient } from '../utils/redisClient';
 import {
 	SubscriberLookup,
 	addOracletoResponse,
 	l2WithBNToStrings,
 } from '../utils/utils';
-import { logger } from '../utils/logger';
-import { webhookMessage } from '../utils/webhook';
 
 type wsMarketL2Args = {
 	marketIndex: number;
@@ -33,6 +26,11 @@ type wsMarketL2Args = {
 	updateOnChange?: boolean;
 };
 
+export type wsMarketInfo = {
+	marketIndex: number;
+	marketName: string;
+};
+
 export class DLOBSubscriberIO extends DLOBSubscriber {
 	public marketL2Args: wsMarketL2Args[] = [];
 	public lastSeenL2Formatted: Map<MarketType, Map<number, any>>;
@@ -41,6 +39,8 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 	constructor(
 		config: DLOBSubscriptionConfig & {
 			redisClient: RedisClient;
+			perpMarketInfos: wsMarketInfo[];
+			spotMarketInfos: wsMarketInfo[];
 			spotMarketSubscribers: SubscriberLookup;
 		}
 	) {
@@ -52,17 +52,11 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 		this.lastSeenL2Formatted.set(MarketType.SPOT, new Map());
 		this.lastSeenL2Formatted.set(MarketType.PERP, new Map());
 
-		// Add all active markets to the market L2Args
-		const perpMarkets =
-			driftEnv === 'devnet' ? DevnetPerpMarkets : MainnetPerpMarkets;
-		const spotMarkets =
-			driftEnv === 'devnet' ? DevnetSpotMarkets : MainnetSpotMarkets;
-
-		for (const market of perpMarkets) {
+		for (const market of config.perpMarketInfos) {
 			this.marketL2Args.push({
 				marketIndex: market.marketIndex,
 				marketType: MarketType.PERP,
-				marketName: market.symbol,
+				marketName: market.marketName,
 				depth: -1,
 				numVammOrders: 100,
 				includeVamm: true,
@@ -70,11 +64,11 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 				fallbackL2Generators: [],
 			});
 		}
-		for (const market of spotMarkets) {
+		for (const market of config.spotMarketInfos) {
 			this.marketL2Args.push({
 				marketIndex: market.marketIndex,
 				marketType: MarketType.SPOT,
-				marketName: market.symbol,
+				marketName: market.marketName,
 				depth: -1,
 				includeVamm: false,
 				updateOnChange: true,
@@ -92,7 +86,7 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 			try {
 				this.getL2AndSendMsg(l2Args);
 			} catch (error) {
-				logger.error(error);
+				console.error(error);
 				console.log(`Error getting L2 ${l2Args.marketName}`);
 			}
 		}
@@ -145,19 +139,6 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 			asks: l2Formatted.asks.slice(0, 5),
 		});
 
-		// START HACK FOR DEBUGGING FAILURE WEBSOCKET CONNECTIONS
-		const orderSubscriberSlot = this.slotSource.getSlot();
-		const dlobOracleSlot = parseInt(l2Formatted['oracleData']['slot']);
-		const slotDiff = Math.abs(orderSubscriberSlot - dlobOracleSlot);
-		if (slotDiff > 50) {
-			const msg = `DlobPublisher bad slot in ${l2Formatted['marketType']} market ${l2Formatted['marketName']}! abs(oracleSlot (${dlobOracleSlot}) - orderSubscriberSlot (${orderSubscriberSlot})) = ${slotDiff} > 50`;
-			logger.error(msg);
-			webhookMessage(msg).then(() => {
-				process.exit(1);
-			});
-		}
-		// END HACK FOR DEBUGGING FAILURE WEBSOCKET CONNECTIONS
-
 		this.redisClient.client.publish(
 			`orderbook_${marketType}_${l2Args.marketIndex}`,
 			JSON.stringify(l2Formatted)

src/publishers/dlobPublisher.ts

@@ -14,6 +14,9 @@ import {
 	DriftClientSubscriptionConfig,
 	SlotSubscriber,
 	isVariant,
+	OracleInfo,
+	PerpMarketConfig,
+	SpotMarketConfig,
 } from '@drift-labs/sdk';
 
 import { logger, setLogLevel } from '../utils/logger';
@@ -21,9 +24,13 @@ import {
 	SubscriberLookup,
 	getPhoenixSubscriber,
 	getSerumSubscriber,
+	parsePositiveIntArray,
 	sleep,
 } from '../utils/utils';
-import { DLOBSubscriberIO } from '../dlob-subscriber/DLOBSubscriberIO';
+import {
+	DLOBSubscriberIO,
+	wsMarketInfo,
+} from '../dlob-subscriber/DLOBSubscriberIO';
 import { RedisClient } from '../utils/redisClient';
 import {
 	DLOBProvider,
@@ -64,6 +71,16 @@ const ORDERBOOK_UPDATE_INTERVAL =
 	parseInt(process.env.ORDERBOOK_UPDATE_INTERVAL) || 1000;
 const WS_FALLBACK_FETCH_INTERVAL = 10_000;
 
+// comma separated list of perp market indexes to load: i.e. 0,1,2,3
+const PERP_MARKETS_TO_LOAD = process.env.PERP_MARKETS_TO_LOAD
+	? parsePositiveIntArray(process.env.PERP_MARKETS_TO_LOAD)
+	: undefined;
+
+// comma separated list of spot market indexes to load: i.e. 0,1,2,3
+const SPOT_MARKETS_TO_LOAD = process.env.SPOT_MARKETS_TO_LOAD
+	? parsePositiveIntArray(process.env.SPOT_MARKETS_TO_LOAD)
+	: undefined;
+
 logger.info(`RPC endpoint: ${endpoint}`);
 logger.info(`WS endpoint:  ${wsEndpoint}`);
 logger.info(`DriftEnv:     ${driftEnv}`);
@@ -71,22 +88,117 @@ logger.info(`Commit:       ${commitHash}`);
 
 let MARKET_SUBSCRIBERS: SubscriberLookup = {};
 
+const getMarketsAndOraclesToLoad = (
+	sdkConfig: any
+): {
+	perpMarketInfos: wsMarketInfo[];
+	spotMarketInfos: wsMarketInfo[];
+	oracleInfos?: OracleInfo[];
+} => {
+	const oracleInfos: OracleInfo[] = [];
+	const oraclesTracked = new Set();
+	const perpMarketInfos: wsMarketInfo[] = [];
+	const spotMarketInfos: wsMarketInfo[] = [];
+
+	// only watch all markets if neither env vars are specified
+	const noMarketsSpecified = !PERP_MARKETS_TO_LOAD && !SPOT_MARKETS_TO_LOAD;
+
+	let perpIndexes = PERP_MARKETS_TO_LOAD;
+	if (!perpIndexes) {
+		if (noMarketsSpecified) {
+			perpIndexes = sdkConfig.PERP_MARKETS.map((m) => m.marketIndex);
+		} else {
+			perpIndexes = [];
+		}
+	}
+	let spotIndexes = SPOT_MARKETS_TO_LOAD;
+	if (!spotIndexes) {
+		if (noMarketsSpecified) {
+			spotIndexes = sdkConfig.SPOT_MARKETS.map((m) => m.marketIndex);
+		} else {
+			spotIndexes = [];
+		}
+	}
+
+	if (perpIndexes.length > 0) {
+		for (const idx of perpIndexes) {
+			const perpMarketConfig = sdkConfig.PERP_MARKETS[idx] as PerpMarketConfig;
+			if (!perpMarketConfig) {
+				throw new Error(`Perp market config for ${idx} not found`);
+			}
+			const oracleKey = perpMarketConfig.oracle.toBase58();
+			if (!oraclesTracked.has(oracleKey)) {
+				logger.info(`Tracking oracle ${oracleKey} for perp market ${idx}`);
+				oracleInfos.push({
+					publicKey: perpMarketConfig.oracle,
+					source: perpMarketConfig.oracleSource,
+				});
+				oraclesTracked.add(oracleKey);
+			}
+			perpMarketInfos.push({
+				marketIndex: perpMarketConfig.marketIndex,
+				marketName: perpMarketConfig.symbol,
+			});
+		}
+		logger.info(
+			`DlobPublisher tracking perp markets: ${JSON.stringify(perpMarketInfos)}`
+		);
+	}
+
+	if (spotIndexes.length > 0) {
+		for (const idx of spotIndexes) {
+			const spotMarketConfig = sdkConfig.SPOT_MARKETS[idx] as SpotMarketConfig;
+			if (!spotMarketConfig) {
+				throw new Error(`Spot market config for ${idx} not found`);
+			}
+			const oracleKey = spotMarketConfig.oracle.toBase58();
+			if (!oraclesTracked.has(oracleKey)) {
+				logger.info(`Tracking oracle ${oracleKey} for spot market ${idx}`);
+				oracleInfos.push({
+					publicKey: spotMarketConfig.oracle,
+					source: spotMarketConfig.oracleSource,
+				});
+				oraclesTracked.add(oracleKey);
+			}
+			spotMarketInfos.push({
+				marketIndex: spotMarketConfig.marketIndex,
+				marketName: spotMarketConfig.symbol,
+			});
+		}
+		logger.info(
+			`DlobPublisher tracking spot markets: ${JSON.stringify(spotMarketInfos)}`
+		);
+	}
+
+	return {
+		perpMarketInfos,
+		spotMarketInfos,
+		oracleInfos,
+	};
+};
+
 const initializeAllMarketSubscribers = async (driftClient: DriftClient) => {
 	const markets: SubscriberLookup = {};
 
-	for (const market of sdkConfig.SPOT_MARKETS) {
+	for (const market of driftClient.getSpotMarketAccounts()) {
 		markets[market.marketIndex] = {
 			phoenix: undefined,
 			serum: undefined,
 		};
+		const marketConfig = sdkConfig.SPOT_MARKETS[market.marketIndex];
 
-		if (market.phoenixMarket) {
+		if (marketConfig.phoenixMarket) {
 			const phoenixConfigAccount =
-				await driftClient.getPhoenixV1FulfillmentConfig(market.phoenixMarket);
+				await driftClient.getPhoenixV1FulfillmentConfig(
+					marketConfig.phoenixMarket
+				);
 			if (isVariant(phoenixConfigAccount.status, 'enabled')) {
+				logger.info(
+					`Loading phoenix subscriber for spot market ${market.marketIndex}`
+				);
 				const phoenixSubscriber = getPhoenixSubscriber(
 					driftClient,
-					market,
+					marketConfig,
 					sdkConfig
 				);
 				await phoenixSubscriber.subscribe();
@@ -103,14 +215,17 @@ const initializeAllMarketSubscribers = async (driftClient: DriftClient) => {
 			}
 		}
 
-		if (market.serumMarket) {
+		if (marketConfig.serumMarket) {
 			const serumConfigAccount = await driftClient.getSerumV3FulfillmentConfig(
-				market.serumMarket
+				marketConfig.serumMarket
 			);
 			if (isVariant(serumConfigAccount.status, 'enabled')) {
+				logger.info(
+					`Loading serum subscriber for spot market ${market.marketIndex}`
+				);
 				const serumSubscriber = getSerumSubscriber(
 					driftClient,
-					market,
+					marketConfig,
 					sdkConfig
 				);
 				await serumSubscriber.subscribe();
@@ -177,12 +292,17 @@ const main = async () => {
 		};
 	}
 
+	const { perpMarketInfos, spotMarketInfos, oracleInfos } =
+		getMarketsAndOraclesToLoad(sdkConfig);
 	driftClient = new DriftClient({
 		connection,
 		wallet,
 		programID: clearingHousePublicKey,
 		accountSubscription,
 		env: driftEnv,
+		perpMarketIndexes: perpMarketInfos.map((m) => m.marketIndex),
+		spotMarketIndexes: spotMarketInfos.map((m) => m.marketIndex),
+		oracleInfos,
 	});
 
 	const lamportsBalance = await connection.getBalance(wallet.publicKey);
@@ -271,6 +391,8 @@ const main = async () => {
 		updateFrequency: ORDERBOOK_UPDATE_INTERVAL,
 		redisClient,
 		spotMarketSubscribers: MARKET_SUBSCRIBERS,
+		perpMarketInfos,
+		spotMarketInfos,
 	});
 	await dlobSubscriber.subscribe();
 	if (useWebsocket && !FEATURE_FLAGS.DISABLE_GPA_REFRESH) {

src/utils/utils.ts

@@ -39,6 +39,17 @@ export function sleep(ms: number): Promise<void> {
 	return new Promise((resolve) => setTimeout(resolve, ms));
 }
 
+export function parsePositiveIntArray(
+	intArray: string,
+	separator = ','
+): number[] {
+	return intArray
+		.split(separator)
+		.map((s) => s.trim())
+		.map((s) => parseInt(s))
+		.filter((n) => !isNaN(n) && n > 0);
+}
+
 export const getOracleForMarket = (
 	driftClient: DriftClient,
 	marketType: MarketType,

yarn.lock

@@ -115,10 +115,10 @@
     enabled "2.0.x"
     kuler "^2.0.0"
 
-"@drift-labs/sdk@2.52.0-beta.14":
-  version "2.52.0-beta.14"
-  resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.52.0-beta.14.tgz#93423eee5f3186df9d3a9a18c762c3b0693b6d3c"
-  integrity sha512-3lg8JuPBrWTlYPZr1JhRFWwYaTv92BArlYuATj9Hz4McMH70/PWAHgU2gZPXtzr97VIygGhAYjei0LH8j6yXhw==
+"@drift-labs/sdk@2.53.0-beta.3":
+  version "2.53.0-beta.3"
+  resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.53.0-beta.3.tgz#4ce6aeec21959df24341fc3f7eca5fff6ce8c6d6"
+  integrity sha512-ZG/urHFdTZ1ikc3F9tdRGSvsylCb2woV3eCVsRfxZCgmpDaK09pfCLDGtzkiROlZ0VVe5ECPJinbi0NyK2nNnQ==
   dependencies:
     "@coral-xyz/anchor" "0.28.1-beta.2"
     "@ellipsis-labs/phoenix-sdk" "^1.4.2"