Merge pull request #137 from drift-labs/master

cache priority fees

package.json

@@ -6,7 +6,7 @@
   "license": "Apache-2.0",
   "dependencies": {
     "@coral-xyz/anchor": "^0.29.0",
-    "@drift-labs/sdk": "2.76.0-beta.6",
+    "@drift-labs/sdk": "2.76.0-beta.7",
     "@opentelemetry/api": "^1.1.0",
     "@opentelemetry/auto-instrumentations-node": "^0.31.1",
     "@opentelemetry/exporter-prometheus": "^0.31.0",
@@ -56,6 +56,7 @@
     "server-lite": "ts-node src/serverLite.ts",
     "dlob-publish": "ts-node src/publishers/dlobPublisher.ts",
     "trades-publish": "ts-node src/publishers/tradesPublisher.ts",
+    "fees-publish": "ts-node src/publishers/priorityFeesPublisher.ts",
     "ws-manager": "ts-node src/wsConnectionManager.ts",
     "ws-manager:inspect": "yarn build && node --inspect ./lib/wsConnectionManager.js",
     "dev:inspect": "yarn build && node --inspect ./lib/index.js",

src/index.ts

@@ -9,9 +9,6 @@ import { Commitment, Connection, Keypair, PublicKey } from '@solana/web3.js';
 import {
 	BulkAccountLoader,
 	DLOBNode,
-	DLOBOrder,
-	DLOBOrders,
-	DLOBOrdersCoder,
 	DLOBSubscriber,
 	DriftClient,
 	DriftClientSubscriptionConfig,
@@ -299,8 +296,6 @@ const main = async (): Promise<void> => {
 
 	const dlobProvider = getDLOBProviderFromOrderSubscriber(orderSubscriber);
 
-	const dlobCoder = DLOBOrdersCoder.create();
-
 	await driftClient.subscribe();
 	driftClient.eventEmitter.on('error', (e) => {
 		logger.info('clearing house error');
@@ -461,237 +456,24 @@ const main = async (): Promise<void> => {
 	app.get('/startup', handleStartup);
 	app.get('/', handleHealthCheck(2 * WS_FALLBACK_FETCH_INTERVAL, dlobProvider));
 
-	if (FEATURE_FLAGS.ENABLE_ORDERS_ENDPOINTS) {
+	app.get('/priorityFees', async (req, res, next) => {
-		app.get('/orders/json/raw', async (_req, res, next) => {
 		try {
-				// object with userAccount key and orders object serialized
+			const { marketIndex, marketType } = req.query;
-				const orders: Array<any> = [];
-				const oracles: Array<any> = [];
-				const slot = dlobProvider.getSlot();
 
-				for (const market of driftClient.getPerpMarketAccounts()) {
+			const fees = await redisClients[
-					const oracle = driftClient.getOracleDataForPerpMarket(
+				parseInt(process.env.HELIUS_REDIS_HOST_INDEX) ?? 0
-						market.marketIndex
+			].client.get(`priorityFees_${marketType}_${marketIndex}`);
-					);
+			if (fees) {
-					oracles.push({
+				res.status(200).json(JSON.parse(fees));
-						marketIndex: market.marketIndex,
-						...oracle,
-					});
-				}
-
-				for (const {
-					userAccount,
-					publicKey,
-				} of dlobProvider.getUserAccounts()) {
-					for (const order of userAccount.orders) {
-						if (isVariant(order.status, 'init')) {
-							continue;
-						}
-
-						orders.push({
-							user: publicKey.toBase58(),
-							order: order,
-						});
-					}
-				}
-
-				// respond with orders
-				res.writeHead(200);
-				res.end(
-					JSON.stringify({
-						slot,
-						oracles,
-						orders,
-					})
-				);
-			} catch (e) {
-				next(e);
-			}
-		});
-
-		app.get('/orders/json', async (_req, res, next) => {
-			try {
-				// object with userAccount key and orders object serialized
-				const slot = dlobProvider.getSlot();
-				const orders: Array<any> = [];
-				const oracles: Array<any> = [];
-				for (const market of driftClient.getPerpMarketAccounts()) {
-					const oracle = driftClient.getOracleDataForPerpMarket(
-						market.marketIndex
-					);
-					const oracleHuman = {
-						marketIndex: market.marketIndex,
-						price: oracle.price.toString(),
-						slot: oracle.slot.toString(),
-						confidence: oracle.confidence.toString(),
-						hasSufficientNumberOfDataPoints:
-							oracle.hasSufficientNumberOfDataPoints,
-						maxPrice: oracle.maxPrice,
-					};
-					if (oracle.twap) {
-						oracleHuman['twap'] = oracle.twap.toString();
-					}
-					if (oracle.twapConfidence) {
-						oracleHuman['twapConfidence'] = oracle.twapConfidence.toString();
-					}
-					oracles.push(oracleHuman);
-				}
-				for (const {
-					userAccount,
-					publicKey,
-				} of dlobProvider.getUserAccounts()) {
-					for (const order of userAccount.orders) {
-						if (isVariant(order.status, 'init')) {
-							continue;
-						}
-
-						const orderHuman = {
-							status: getVariant(order.status),
-							orderType: getVariant(order.orderType),
-							marketType: getVariant(order.marketType),
-							slot: order.slot.toString(),
-							orderId: order.orderId,
-							userOrderId: order.userOrderId,
-							marketIndex: order.marketIndex,
-							price: order.price.toString(),
-							baseAssetAmount: order.baseAssetAmount.toString(),
-							baseAssetAmountFilled: order.baseAssetAmountFilled.toString(),
-							quoteAssetAmountFilled: order.quoteAssetAmountFilled.toString(),
-							direction: getVariant(order.direction),
-							reduceOnly: order.reduceOnly,
-							triggerPrice: order.triggerPrice.toString(),
-							triggerCondition: getVariant(order.triggerCondition),
-							existingPositionDirection: getVariant(
-								order.existingPositionDirection
-							),
-							postOnly: order.postOnly,
-							immediateOrCancel: order.immediateOrCancel,
-							oraclePriceOffset: order.oraclePriceOffset,
-							auctionDuration: order.auctionDuration,
-							auctionStartPrice: order.auctionStartPrice.toString(),
-							auctionEndPrice: order.auctionEndPrice.toString(),
-							maxTs: order.maxTs.toString(),
-						};
-						if (order.quoteAssetAmountFilled) {
-							orderHuman['quoteAssetAmount'] =
-								order.quoteAssetAmountFilled.toString();
-							orderHuman['quoteAssetAmountFilled'] =
-								order.quoteAssetAmountFilled.toString();
-						}
-
-						orders.push({
-							user: publicKey.toBase58(),
-							order: orderHuman,
-						});
-					}
-				}
-
-				// respond with orders
-				res.writeHead(200);
-				res.end(
-					JSON.stringify({
-						slot,
-						oracles,
-						orders,
-					})
-				);
-			} catch (err) {
-				next(err);
-			}
-		});
-
-		app.get('/orders/idl', async (_req, res, next) => {
-			try {
-				const dlobOrders: DLOBOrders = [];
-
-				for (const {
-					userAccount,
-					publicKey,
-				} of dlobProvider.getUserAccounts()) {
-					for (const order of userAccount.orders) {
-						if (isVariant(order.status, 'init')) {
-							continue;
-						}
-
-						dlobOrders.push({
-							user: publicKey,
-							order,
-						} as DLOBOrder);
-					}
-				}
-
-				res.writeHead(200);
-				res.end(dlobCoder.encode(dlobOrders));
-			} catch (err) {
-				next(err);
-			}
-		});
-
-		app.get('/orders/idlWithSlot', async (req, res, next) => {
-			try {
-				const { marketName, marketIndex, marketType } = req.query;
-				const { normedMarketType, normedMarketIndex, error } =
-					validateDlobQuery(
-						driftClient,
-						driftEnv,
-						marketType as string,
-						marketIndex as string,
-						marketName as string
-					);
-				const useFilter =
-					marketName !== undefined ||
-					marketIndex !== undefined ||
-					marketType !== undefined;
-
-				if (useFilter) {
-					if (
-						error ||
-						normedMarketType === undefined ||
-						normedMarketIndex === undefined
-					) {
-						res.status(400).send(error);
 				return;
+			} else {
+				res.writeHead(404);
+				res.end('Not found');
 			}
-				}
-
-				const dlobOrders: DLOBOrders = [];
-
-				for (const {
-					userAccount,
-					publicKey,
-				} of dlobProvider.getUserAccounts()) {
-					for (const order of userAccount.orders) {
-						if (isVariant(order.status, 'init')) {
-							continue;
-						}
-
-						if (useFilter) {
-							if (
-								getVariant(order.marketType) !== getVariant(normedMarketType) ||
-								order.marketIndex !== normedMarketIndex
-							) {
-								continue;
-							}
-						}
-
-						dlobOrders.push({
-							user: publicKey,
-							order,
-						} as DLOBOrder);
-					}
-				}
-
-				res.end(
-					JSON.stringify({
-						slot: dlobProvider.getSlot(),
-						data: dlobCoder.encode(dlobOrders).toString('base64'),
-					})
-				);
 		} catch (err) {
 			next(err);
 		}
 	});
-	}
 
 	app.get('/topMakers', async (req, res, next) => {
 		try {

src/publishers/priorityFeesPublisher.ts

@@ -0,0 +1,247 @@
+import { program } from 'commander';
+
+import { Connection, Commitment, Keypair } from '@solana/web3.js';
+
+import {
+	DriftClient,
+	initialize,
+	DriftEnv,
+	Wallet,
+	BulkAccountLoader,
+	getMarketsAndOraclesForSubscription,
+} from '@drift-labs/sdk';
+
+import { logger, setLogLevel } from '../utils/logger';
+import { sleep } from '../utils/utils';
+import express from 'express';
+// import { handleHealthCheck } from '../core/metrics';
+import { RedisClient } from '../utils/redisClient';
+
+require('dotenv').config();
+const stateCommitment: Commitment = 'confirmed';
+const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
+const commitHash = process.env.COMMIT;
+const REDIS_HOST = process.env.REDIS_HOST || 'localhost';
+const REDIS_PORT = process.env.REDIS_PORT || '6379';
+const REDIS_PASSWORD = process.env.REDIS_PASSWORD;
+
+// Set up express for health checks
+const app = express();
+
+//@ts-ignore
+const sdkConfig = initialize({ env: process.env.ENV });
+let driftClient: DriftClient;
+
+const opts = program.opts();
+setLogLevel(opts.debug ? 'debug' : 'info');
+
+const token = process.env.TOKEN;
+const endpoint = token
+	? process.env.ENDPOINT + `/${token}`
+	: process.env.ENDPOINT;
+const wsEndpoint = process.env.WS_ENDPOINT;
+const FEE_POLLING_FREQUENCY =
+	parseInt(process.env.FEE_POLLING_FREQUENCY) || 5000;
+
+if (!endpoint.includes('helius')) {
+	throw new Error('We use helius for fee publisher fellas');
+}
+
+logger.info(`RPC endpoint: ${endpoint}`);
+logger.info(`WS endpoint:  ${wsEndpoint}`);
+logger.info(`DriftEnv:     ${driftEnv}`);
+logger.info(`Commit:       ${commitHash}`);
+
+class PriorityFeeSubscriber {
+	endpoint: string;
+	perpMarketPubkeys: { marketIndex: number; pubkey: string }[];
+	spotMarketPubkeys: { marketIndex: number; pubkeys: string[] }[];
+	redisClient: RedisClient;
+	frequencyMs: number;
+
+	constructor(config: {
+		endpoint: string;
+		redisClient: RedisClient;
+		perpMarketPubkeys: { marketIndex: number; pubkey: string }[];
+		spotMarketPubkeys: { marketIndex: number; pubkeys: string[] }[];
+		frequencyMs?: number;
+	}) {
+		this.endpoint = config.endpoint;
+		this.perpMarketPubkeys = config.perpMarketPubkeys;
+		this.spotMarketPubkeys = config.spotMarketPubkeys;
+		this.redisClient = config.redisClient;
+		this.frequencyMs = config.frequencyMs ?? FEE_POLLING_FREQUENCY;
+	}
+
+	async subscribe() {
+		await this.fetchAndPushPriorityFees();
+		setInterval(async () => {
+			await this.fetchAndPushPriorityFees();
+		}, this.frequencyMs);
+	}
+
+	async fetchAndPushPriorityFees() {
+		const [resultPerp, resultSpot] = await Promise.all([
+			fetch(this.endpoint, {
+				method: 'POST',
+				headers: {
+					'Content-Type': 'application/json',
+				},
+				body: JSON.stringify(
+					this.perpMarketPubkeys.map((xx) => {
+						return {
+							jsonrpc: '2.0',
+							id: xx.marketIndex.toString(),
+							method: 'getPriorityFeeEstimate',
+							params: [
+								{
+									accountKeys: [xx.pubkey],
+									options: {
+										includeAllPriorityFeeLevels: true,
+									},
+								},
+							],
+						};
+					})
+				),
+			}),
+			fetch(this.endpoint, {
+				method: 'POST',
+				headers: {
+					'Content-Type': 'application/json',
+				},
+				body: JSON.stringify(
+					this.spotMarketPubkeys.map((xx) => {
+						return {
+							jsonrpc: '2.0',
+							id: (100 + xx.marketIndex).toString(),
+							method: 'getPriorityFeeEstimate',
+							params: [
+								{
+									accountKeys: xx.pubkeys,
+									options: {
+										includeAllPriorityFeeLevels: true,
+									},
+								},
+							],
+						};
+					})
+				),
+			}),
+		]);
+
+		const [dataPerp, dataSpot] = await Promise.all([
+			resultPerp.json(),
+			resultSpot.json(),
+		]);
+
+		dataPerp.forEach((result: any) => {
+			const marketIndex = parseInt(result['id']);
+			this.redisClient.client.publish(
+				`priorityFees_perp_${marketIndex}`,
+				JSON.stringify(result.result['priorityFeeLevels'])
+			);
+			this.redisClient.client.set(
+				`priorityFees_perp_${marketIndex}`,
+				JSON.stringify(result.result['priorityFeeLevels'])
+			);
+		});
+
+		dataSpot.forEach((result: any) => {
+			const marketIndex = parseInt(result['id']) - 100;
+			this.redisClient.client.publish(
+				`priorityFees_spot_${marketIndex}`,
+				JSON.stringify(result.result['priorityFeeLevels'])
+			);
+			this.redisClient.client.set(
+				`priorityFees_spot_${marketIndex}`,
+				JSON.stringify(result.result['priorityFeeLevels'])
+			);
+		});
+	}
+}
+
+const main = async () => {
+	const connection = new Connection(endpoint, {
+		wsEndpoint: wsEndpoint,
+		commitment: stateCommitment,
+	});
+
+	const { perpMarketIndexes, spotMarketIndexes, oracleInfos } =
+		getMarketsAndOraclesForSubscription(sdkConfig.ENV);
+
+	const driftClient = new DriftClient({
+		connection,
+		wallet: new Wallet(new Keypair()),
+		perpMarketIndexes,
+		spotMarketIndexes,
+		oracleInfos,
+		accountSubscription: {
+			type: 'polling',
+			accountLoader: new BulkAccountLoader(connection, stateCommitment, 0),
+		},
+	});
+	await driftClient.subscribe();
+
+	const perpMarketPubkeys = driftClient.getPerpMarketAccounts().map((acct) => {
+		return { marketIndex: acct.marketIndex, pubkey: acct.pubkey.toString() };
+	});
+
+	const usdcMarket = driftClient.getSpotMarketAccount(0).pubkey.toString();
+	const spotMarketPubkeys: { marketIndex: number; pubkeys: string[] }[] = [];
+	for (const market of sdkConfig.SPOT_MARKETS) {
+		const pubkeysForMarket = [usdcMarket];
+
+		const driftMarket = driftClient.getSpotMarketAccount(market.marketIndex);
+		pubkeysForMarket.push(driftMarket.pubkey.toString());
+
+		if (market.serumMarket) {
+			pubkeysForMarket.push(market.serumMarket.toString());
+		}
+
+		if (market.phoenixMarket) {
+			pubkeysForMarket.push(market.phoenixMarket.toString());
+		}
+
+		spotMarketPubkeys.push({
+			marketIndex: market.marketIndex,
+			pubkeys: pubkeysForMarket,
+		});
+	}
+
+	const redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD);
+	await redisClient.connect();
+
+	const priorityFeeSubscriber = new PriorityFeeSubscriber({
+		endpoint,
+		perpMarketPubkeys,
+		spotMarketPubkeys,
+		redisClient,
+	});
+
+	await priorityFeeSubscriber.subscribe();
+	const server = app.listen(8080);
+
+	// Default keepalive is 5s, since the AWS ALB timeout is 60 seconds, clients
+	// sometimes get 502s.
+	// https://shuheikagawa.com/blog/2019/04/25/keep-alive-timeout/
+	// https://stackoverflow.com/a/68922692
+	server.keepAliveTimeout = 61 * 1000;
+	server.headersTimeout = 65 * 1000;
+
+	console.log('Priority fee publisher Publishing Messages');
+};
+
+async function recursiveTryCatch(f: () => void) {
+	try {
+		await f();
+	} catch (e) {
+		console.error(e);
+		await sleep(15000);
+		await recursiveTryCatch(f);
+	}
+}
+
+recursiveTryCatch(() => main());
+
+export { sdkConfig, endpoint, wsEndpoint, driftEnv, commitHash, driftClient };

src/wsConnectionManager.ts

@@ -65,11 +65,13 @@ const getRedisChannelFromMessage = (message: any): string => {
 		throw new Error('Bad market specified');
 	}
 
-	switch (channel) {
+	switch (channel.toLowerCase()) {
 		case 'trades':
 			return `trades_${marketType}_${marketIndex}`;
 		case 'orderbook':
 			return `orderbook_${marketType}_${marketIndex}`;
+		case 'priorityfees':
+			return `priorityFees_${marketType}_${marketIndex}`;
 		case undefined:
 		default:
 			throw new Error('Bad channel specified');
@@ -145,17 +147,14 @@ async function main() {
 								JSON.stringify({
 									channel: requestChannel,
 									error:
-										'Error subscribing to channel with data: ' +
+										'Error subscribing to channel'
-										JSON.stringify(parsedMessage),
 								})
 							);
 						} else {
 							ws.close(
 								1003,
 								JSON.stringify({
-									error:
+									error: 'Error subscribing to channel'
-										'Error subscribing to channel with data: ' +
-										JSON.stringify(parsedMessage),
 								})
 							);
 						}
@@ -221,8 +220,7 @@ async function main() {
 								JSON.stringify({
 									channel: requestChannel,
 									error:
-										'Error unsubscribing from channel with data: ' +
+										'Error unsubscribing from channel'
-										JSON.stringify(parsedMessage),
 								})
 							);
 						} else {
@@ -230,8 +228,7 @@ async function main() {
 								1003,
 								JSON.stringify({
 									error:
-										'Error unsubscribing from channel with data: ' +
+										'Error unsubscribing from channel'
-										JSON.stringify(parsedMessage),
 								})
 							);
 						}

yarn.lock

@@ -115,10 +115,10 @@
     enabled "2.0.x"
     kuler "^2.0.0"
 
-"@drift-labs/sdk@2.76.0-beta.6":
+"@drift-labs/sdk@2.76.0-beta.7":
-  version "2.76.0-beta.6"
+  version "2.76.0-beta.7"
-  resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.76.0-beta.6.tgz#c1f882f52cef4ef33f1ab9c5e346067cd3a325d0"
+  resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.76.0-beta.7.tgz#aa910d2db9446ed825636da1c9984441eb5daa3c"
-  integrity sha512-0H5MD88lrCL2r4l6dBe8a9kaFrDeiHZfuxGq2Nf0O99V5UU57SJ236uM1OIhumQua4kptL6Mv+snKIZji6NgXQ==
+  integrity sha512-nO8eVXQ3hWXJsKm++MRcyUZEXbwUjLLTVpEHvSCOE5Kaqc+z21y+Aqq+44PRMEljqNAE1ouf0Qe6hTsoaqhkZw==
   dependencies:
     "@coral-xyz/anchor" "0.28.1-beta.2"
     "@ellipsis-labs/phoenix-sdk" "^1.4.2"