From 4883a1ee294989fd3af86f3a604c8627bf004633 Mon Sep 17 00:00:00 2001 From: Nour Alharithi Date: Thu, 4 Apr 2024 12:00:35 -0700 Subject: [PATCH 1/4] cache priority fees --- package.json | 1 + src/index.ts | 250 ++---------------------- src/publishers/priorityFeesPublisher.ts | 238 ++++++++++++++++++++++ src/wsConnectionManager.ts | 4 +- 4 files changed, 258 insertions(+), 235 deletions(-) create mode 100644 src/publishers/priorityFeesPublisher.ts diff --git a/package.json b/package.json index f06d731..3473866 100644 --- a/package.json +++ b/package.json @@ -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", diff --git a/src/index.ts b/src/index.ts index 489fa49..a1bbb41 100644 --- a/src/index.ts +++ b/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 => { 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 => { app.get('/startup', handleStartup); app.get('/', handleHealthCheck(2 * WS_FALLBACK_FETCH_INTERVAL, dlobProvider)); - if (FEATURE_FLAGS.ENABLE_ORDERS_ENDPOINTS) { - app.get('/orders/json/raw', async (_req, res, next) => { - try { - // object with userAccount key and orders object serialized - const orders: Array = []; - const oracles: Array = []; - const slot = dlobProvider.getSlot(); + app.get('/priorityFees', async (req, res, next) => { + try { + const { marketIndex, marketType } = req.query; - for (const market of driftClient.getPerpMarketAccounts()) { - const oracle = driftClient.getOracleDataForPerpMarket( - market.marketIndex - ); - oracles.push({ - 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); + const fees = await redisClients[ + parseInt(process.env.HELIUS_REDIS_HOST_INDEX) ?? 0 + ].client.get(`priorityFees_${marketType}_${marketIndex}`); + if (fees) { + res.status(200).json(JSON.parse(fees)); + return; + } else { + res.writeHead(404); + res.end('Not found'); } - }); - - app.get('/orders/json', async (_req, res, next) => { - try { - // object with userAccount key and orders object serialized - const slot = dlobProvider.getSlot(); - const orders: Array = []; - const oracles: Array = []; - 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; - } - } - - 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); - } - }); - } + } catch (err) { + next(err); + } + }); app.get('/topMakers', async (req, res, next) => { try { diff --git a/src/publishers/priorityFeesPublisher.ts b/src/publishers/priorityFeesPublisher.ts new file mode 100644 index 0000000..5fb53d6 --- /dev/null +++ b/src/publishers/priorityFeesPublisher.ts @@ -0,0 +1,238 @@ +import { program } from 'commander'; + +import { Connection, Commitment, Keypair } from '@solana/web3.js'; + +import { + DriftClient, + initialize, + DriftEnv, + Wallet, + BulkAccountLoader, + isVariant, + 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; + +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; pubkey: string }[]; + redisClient: RedisClient; + frequencyMs: number; + + constructor(config: { + endpoint: string; + redisClient: RedisClient; + perpMarketPubkeys: { marketIndex: number; pubkey: string }[]; + spotMarketPubkeys: { marketIndex: number; pubkey: string }[]; + frequencyMs?: number; + }) { + this.endpoint = config.endpoint; + this.perpMarketPubkeys = config.perpMarketPubkeys; + this.spotMarketPubkeys = config.spotMarketPubkeys; + this.redisClient = config.redisClient; + this.frequencyMs = config.frequencyMs ?? 5000; + } + + 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: '1', + 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: '1', + method: 'getPriorityFeeEstimate', + params: [ + { + accountKeys: [xx.pubkey], + options: { + includeAllPriorityFeeLevels: true, + }, + }, + ], + }; + }) + ), + }), + ]); + + const [dataPerp, dataSpot] = await Promise.all([ + resultPerp.json(), + resultSpot.json(), + ]); + + dataPerp.forEach((result: any, index: number) => { + const marketIndex = this.perpMarketPubkeys[index].marketIndex; + 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, index: number) => { + const marketIndex = this.perpMarketPubkeys[index].marketIndex; + 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 spotMarketPubkeys = []; + for (const market of sdkConfig.SPOT_MARKETS) { + if (market.serumMarket) { + const serumConfigAccount = await driftClient.getSerumV3FulfillmentConfig( + market.serumMarket + ); + if (isVariant(serumConfigAccount.status, 'enabled')) { + spotMarketPubkeys.push({ + marketIndex: market.marketIndex, + pubkey: market.serumMarket.toString(), + }); + } + } + } + + 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 }; diff --git a/src/wsConnectionManager.ts b/src/wsConnectionManager.ts index 2f56720..e43c2d1 100644 --- a/src/wsConnectionManager.ts +++ b/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'); From 486401444c5d2e9cb35e43be2dd9a80a831a51d0 Mon Sep 17 00:00:00 2001 From: Nour Alharithi Date: Thu, 4 Apr 2024 12:06:07 -0700 Subject: [PATCH 2/4] make frequency configurable --- src/publishers/priorityFeesPublisher.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/publishers/priorityFeesPublisher.ts b/src/publishers/priorityFeesPublisher.ts index 5fb53d6..f035f34 100644 --- a/src/publishers/priorityFeesPublisher.ts +++ b/src/publishers/priorityFeesPublisher.ts @@ -41,6 +41,7 @@ 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'); @@ -69,7 +70,7 @@ class PriorityFeeSubscriber { this.perpMarketPubkeys = config.perpMarketPubkeys; this.spotMarketPubkeys = config.spotMarketPubkeys; this.redisClient = config.redisClient; - this.frequencyMs = config.frequencyMs ?? 5000; + this.frequencyMs = config.frequencyMs ?? FEE_POLLING_FREQUENCY; } async subscribe() { From ecf6c61ace88a2d5dadc5ce4c6fa98d54592e150 Mon Sep 17 00:00:00 2001 From: Nour Alharithi Date: Thu, 4 Apr 2024 12:32:07 -0700 Subject: [PATCH 3/4] better id rpc requests --- src/publishers/priorityFeesPublisher.ts | 28 ++++++++++++++++++------- 1 file changed, 21 insertions(+), 7 deletions(-) diff --git a/src/publishers/priorityFeesPublisher.ts b/src/publishers/priorityFeesPublisher.ts index f035f34..5becf86 100644 --- a/src/publishers/priorityFeesPublisher.ts +++ b/src/publishers/priorityFeesPublisher.ts @@ -41,7 +41,8 @@ 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; +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'); @@ -91,7 +92,7 @@ class PriorityFeeSubscriber { this.perpMarketPubkeys.map((xx) => { return { jsonrpc: '2.0', - id: '1', + id: xx.marketIndex.toString(), method: 'getPriorityFeeEstimate', params: [ { @@ -114,7 +115,7 @@ class PriorityFeeSubscriber { this.spotMarketPubkeys.map((xx) => { return { jsonrpc: '2.0', - id: '1', + id: (100 + xx.marketIndex).toString(), method: 'getPriorityFeeEstimate', params: [ { @@ -135,8 +136,8 @@ class PriorityFeeSubscriber { resultSpot.json(), ]); - dataPerp.forEach((result: any, index: number) => { - const marketIndex = this.perpMarketPubkeys[index].marketIndex; + dataPerp.forEach((result: any) => { + const marketIndex = parseInt(result['id']); this.redisClient.client.publish( `priorityFees_perp_${marketIndex}`, JSON.stringify(result.result['priorityFeeLevels']) @@ -147,8 +148,8 @@ class PriorityFeeSubscriber { ); }); - dataSpot.forEach((result: any, index: number) => { - const marketIndex = this.perpMarketPubkeys[index].marketIndex; + dataSpot.forEach((result: any) => { + const marketIndex = parseInt(result['id']) - 100; this.redisClient.client.publish( `priorityFees_spot_${marketIndex}`, JSON.stringify(result.result['priorityFeeLevels']) @@ -197,6 +198,19 @@ const main = async () => { marketIndex: market.marketIndex, pubkey: market.serumMarket.toString(), }); + } else { + if (market.phoenixMarket) { + const phoneixConfigAccount = + await driftClient.getPhoenixV1FulfillmentConfig( + market.phoenixMarket + ); + if (isVariant(phoneixConfigAccount.status, 'enabled')) { + spotMarketPubkeys.push({ + marketIndex: market.marketIndex, + pubkey: market.phoenixMarket.toString(), + }); + } + } } } } From 3b808dd254e01de40f989b08b94221256ba98c28 Mon Sep 17 00:00:00 2001 From: Nour Alharithi Date: Thu, 4 Apr 2024 13:26:14 -0700 Subject: [PATCH 4/4] more comprehensive spot market inclusion --- src/publishers/priorityFeesPublisher.ts | 48 +++++++++++-------------- 1 file changed, 21 insertions(+), 27 deletions(-) diff --git a/src/publishers/priorityFeesPublisher.ts b/src/publishers/priorityFeesPublisher.ts index 5becf86..2ebd136 100644 --- a/src/publishers/priorityFeesPublisher.ts +++ b/src/publishers/priorityFeesPublisher.ts @@ -8,7 +8,6 @@ import { DriftEnv, Wallet, BulkAccountLoader, - isVariant, getMarketsAndOraclesForSubscription, } from '@drift-labs/sdk'; @@ -56,7 +55,7 @@ logger.info(`Commit: ${commitHash}`); class PriorityFeeSubscriber { endpoint: string; perpMarketPubkeys: { marketIndex: number; pubkey: string }[]; - spotMarketPubkeys: { marketIndex: number; pubkey: string }[]; + spotMarketPubkeys: { marketIndex: number; pubkeys: string[] }[]; redisClient: RedisClient; frequencyMs: number; @@ -64,7 +63,7 @@ class PriorityFeeSubscriber { endpoint: string; redisClient: RedisClient; perpMarketPubkeys: { marketIndex: number; pubkey: string }[]; - spotMarketPubkeys: { marketIndex: number; pubkey: string }[]; + spotMarketPubkeys: { marketIndex: number; pubkeys: string[] }[]; frequencyMs?: number; }) { this.endpoint = config.endpoint; @@ -119,7 +118,7 @@ class PriorityFeeSubscriber { method: 'getPriorityFeeEstimate', params: [ { - accountKeys: [xx.pubkey], + accountKeys: xx.pubkeys, options: { includeAllPriorityFeeLevels: true, }, @@ -187,32 +186,27 @@ const main = async () => { const perpMarketPubkeys = driftClient.getPerpMarketAccounts().map((acct) => { return { marketIndex: acct.marketIndex, pubkey: acct.pubkey.toString() }; }); - const spotMarketPubkeys = []; + + 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) { - const serumConfigAccount = await driftClient.getSerumV3FulfillmentConfig( - market.serumMarket - ); - if (isVariant(serumConfigAccount.status, 'enabled')) { - spotMarketPubkeys.push({ - marketIndex: market.marketIndex, - pubkey: market.serumMarket.toString(), - }); - } else { - if (market.phoenixMarket) { - const phoneixConfigAccount = - await driftClient.getPhoenixV1FulfillmentConfig( - market.phoenixMarket - ); - if (isVariant(phoneixConfigAccount.status, 'enabled')) { - spotMarketPubkeys.push({ - marketIndex: market.marketIndex, - pubkey: market.phoenixMarket.toString(), - }); - } - } - } + 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);