diff --git a/package.json b/package.json index 7b7245b..7d308f3 100644 --- a/package.json +++ b/package.json @@ -6,7 +6,7 @@ "license": "Apache-2.0", "dependencies": { "@coral-xyz/anchor": "^0.29.0", - "@drift-labs/sdk": "2.49.0-beta.17", + "@drift-labs/sdk": "2.52.0-beta.0", "@opentelemetry/api": "^1.1.0", "@opentelemetry/auto-instrumentations-node": "^0.31.1", "@opentelemetry/exporter-prometheus": "^0.31.0", diff --git a/src/core/metrics.ts b/src/core/metrics.ts index 9675b68..d1683f4 100644 --- a/src/core/metrics.ts +++ b/src/core/metrics.ts @@ -40,6 +40,8 @@ enum METRIC_TYPES { gpa_fetch_duration = 'gpa_fetch_duration', last_ws_message_received_ts = 'last_ws_message_received_ts', account_updates_count = 'account_updates_count', + cache_hit_count = 'cache_hit_count', + cache_miss_count = 'cache_miss_count', current_system_ts = 'current_system_ts', health_status = 'health_status', } @@ -126,6 +128,14 @@ lastWsReceivedTsGauge.addCallback((obs: ObservableResult) => { obs.observe(lastWsMsgReceivedTs, {}); }); +const cacheHitCounter = meter.createCounter(METRIC_TYPES.cache_hit_count, { + description: 'Total redis cache hits', +}); + +const cacheMissCounter = meter.createCounter(METRIC_TYPES.cache_miss_count, { + description: 'Total redis cache misses', +}); + const accountUpdatesCounter = meter.createCounter( METRIC_TYPES.account_updates_count, { @@ -238,4 +248,6 @@ export { handleHealthCheck, setLastReceivedWsMsgTs, accountUpdatesCounter, + cacheHitCounter, + cacheMissCounter, }; diff --git a/src/dlob-subscriber/DLOBSubscriberIO.ts b/src/dlob-subscriber/DLOBSubscriberIO.ts index a7cea28..0a20194 100644 --- a/src/dlob-subscriber/DLOBSubscriberIO.ts +++ b/src/dlob-subscriber/DLOBSubscriberIO.ts @@ -62,8 +62,8 @@ export class DLOBSubscriberIO extends DLOBSubscriber { marketType: MarketType.PERP, marketName: market.symbol, depth: -1, - includeVamm: true, numVammOrders: 100, + includeVamm: true, updateOnChange: true, fallbackL2Generators: [], }); @@ -124,13 +124,39 @@ export class DLOBSubscriberIO extends DLOBSubscriber { l2Args.marketType, l2Args.marketIndex ); + + const l2Formatted_depth100 = Object.assign({}, l2Formatted, { + bids: l2Formatted.bids.slice(0, 100), + asks: l2Formatted.asks.slice(0, 100), + }); + const l2Formatted_depth20 = Object.assign({}, l2Formatted, { + bids: l2Formatted.bids.slice(0, 20), + asks: l2Formatted.asks.slice(0, 20), + }); + const l2Formatted_depth5 = Object.assign({}, l2Formatted, { + bids: l2Formatted.bids.slice(0, 5), + asks: l2Formatted.asks.slice(0, 5), + }); + this.redisClient.client.publish( `orderbook_${marketType}_${l2Args.marketIndex}`, JSON.stringify(l2Formatted) ); this.redisClient.client.set( `last_update_orderbook_${marketType}_${l2Args.marketIndex}`, - JSON.stringify(l2Formatted) + JSON.stringify(l2Formatted_depth100) + ); + this.redisClient.client.set( + `last_update_orderbook_${marketType}_${l2Args.marketIndex}_depth_100`, + JSON.stringify(l2Formatted_depth100) + ); + this.redisClient.client.set( + `last_update_orderbook_${marketType}_${l2Args.marketIndex}_depth_20`, + JSON.stringify(l2Formatted_depth20) + ); + this.redisClient.client.set( + `last_update_orderbook_${marketType}_${l2Args.marketIndex}_depth_5`, + JSON.stringify(l2Formatted_depth5) ); } } diff --git a/src/index.ts b/src/index.ts index a584c0a..02db2bd 100644 --- a/src/index.ts +++ b/src/index.ts @@ -28,8 +28,6 @@ import { initialize, isVariant, OrderSubscriber, - UserAccount, - Order, } from '@drift-labs/sdk'; import { logger, setLogLevel } from './utils/logger'; @@ -39,7 +37,9 @@ import { gpaFetchDurationHistogram, handleHealthCheck, accountUpdatesCounter, + cacheHitCounter, setLastReceivedWsMsgTs, + cacheMissCounter, } from './core/metrics'; import { handleResponseTime } from './core/middleware'; import { @@ -59,8 +59,14 @@ import { getDLOBProviderFromOrderSubscriber, getDLOBProviderFromUserMap, } from './dlobProvider'; +import { RedisClient } from './utils/redisClient'; require('dotenv').config(); + +const REDIS_HOST = process.env.REDIS_HOST || 'localhost'; +const REDIS_PORT = process.env.REDIS_PORT || '6379'; +const REDIS_PASSWORD = process.env.REDIS_PASSWORD; + const driftEnv = (process.env.ENV || 'devnet') as DriftEnv; const commitHash = process.env.COMMIT; //@ts-ignore @@ -77,6 +83,7 @@ const rateLimitCallsPerSecond = process.env.RATE_LIMIT_CALLS_PER_SECOND ? parseInt(process.env.RATE_LIMIT_CALLS_PER_SECOND) : 1; const loadTestAllowed = process.env.ALLOW_LOAD_TEST?.toLowerCase() === 'true'; +const useRedis = process.env.USE_REDIS?.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]'; @@ -264,6 +271,7 @@ const main = async () => { 'updateReceived', (_pubkey: PublicKey, _slot: number, _dataType: 'raw' | 'decoded') => { setLastReceivedWsMsgTs(Date.now()); + // eslint-disable-next-line @typescript-eslint/no-unused-vars updatesReceivedTotal++; accountUpdatesCounter.add(1); } @@ -343,6 +351,13 @@ const main = async () => { `DLOBSubscriber initialized in ${Date.now() - initDlobSubscriberStart} ms` ); + let redisClient: RedisClient; + if (useRedis) { + logger.info('Connecting to redis'); + redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD); + await redisClient.connect(); + } + logger.info(`Initializing all market subscribers...`); const initAllMarketSubscribersStart = Date.now(); MARKET_SUBSCRIBERS = await initializeAllMarketSubscribers(driftClient); @@ -742,6 +757,69 @@ const main = async () => { adjustedDepth = '-1'; } + let l2Formatted: any; + if (useRedis) { + if ( + !isSpot && + `${includeVamm}`.toLowerCase() === 'true' && + `${includeOracle}`.toLowerCase().toLowerCase() === 'true' && + !grouping + ) { + let redisL2: string; + if (parseInt(adjustedDepth as string) === 5) { + redisL2 = await redisClient.client.get( + `last_update_orderbook_perp_${normedMarketIndex}_depth_5` + ); + } else if (parseInt(adjustedDepth as string) === 20) { + redisL2 = await redisClient.client.get( + `last_update_orderbook_perp_${normedMarketIndex}_depth_20` + ); + } else if (parseInt(adjustedDepth as string) === 100) { + redisL2 = await redisClient.client.get( + `last_update_orderbook_perp_${normedMarketIndex}_depth_100` + ); + } + if ( + redisL2 && + slotSource.getSlot() - parseInt(JSON.parse(redisL2).slot) < 10 + ) + l2Formatted = redisL2; + } else if ( + isSpot && + `${includeSerum}`.toLowerCase() === 'true' && + `${includePhoenix}`.toLowerCase() === 'true' && + `${includeOracle}`.toLowerCase() === 'true' && + !grouping + ) { + let redisL2: string; + if (parseInt(adjustedDepth as string) === 5) { + redisL2 = await redisClient.client.get( + `last_update_orderbook_spot_${normedMarketIndex}_depth_5` + ); + } else if (parseInt(adjustedDepth as string) === 20) { + redisL2 = await redisClient.client.get( + `last_update_orderbook_spot_${normedMarketIndex}_depth_20` + ); + } else if (parseInt(adjustedDepth as string) === 100) { + redisL2 = await redisClient.client.get( + `last_update_orderbook_spot_${normedMarketIndex}_depth_100` + ); + } + if ( + redisL2 && + slotSource.getSlot() - parseInt(JSON.parse(redisL2).slot) < 10 + ) + l2Formatted = redisL2; + } + + if (l2Formatted) { + cacheHitCounter.add(1); + res.writeHead(200); + res.end(l2Formatted); + return; + } + } + const l2 = dlobSubscriber.getL2({ marketIndex: normedMarketIndex, marketType: normedMarketType, @@ -783,7 +861,7 @@ const main = async () => { res.end(JSON.stringify(l2Formatted)); } else { // make the BNs into strings - const l2Formatted = l2WithBNToStrings(l2); + l2Formatted = l2WithBNToStrings(l2); if (`${includeOracle}`.toLowerCase() === 'true') { addOracletoResponse( l2Formatted, @@ -792,10 +870,10 @@ const main = async () => { normedMarketIndex ); } - - res.writeHead(200); - res.end(JSON.stringify(l2Formatted)); } + cacheMissCounter.add(1); + res.writeHead(200); + res.end(JSON.stringify(l2Formatted)); } catch (err) { next(err); } @@ -836,86 +914,148 @@ const main = async () => { return; } - const l2s = normedParams.map((normedParam) => { - const { normedMarketType, normedMarketIndex, error } = - validateDlobQuery( - driftClient, - driftEnv, - normedParam['marketType'] as string, - normedParam['marketIndex'] as string, - normedParam['marketName'] as string - ); - if (error) { - res.status(400).send(`Bad Request: ${error}`); - return; - } - - const isSpot = isVariant(normedMarketType, 'spot'); - - let adjustedDepth = normedParam['depth'] ?? '10'; - if (normedParam['grouping'] !== undefined) { - // If grouping is also supplied, we want the entire book depth. - // we will apply depth after grouping - adjustedDepth = '-1'; - } - - const l2 = dlobSubscriber.getL2({ - marketIndex: normedMarketIndex, - marketType: normedMarketType, - depth: parseInt(adjustedDepth as string), - includeVamm: isSpot - ? false - : `${normedParam['includeVamm']}`.toLowerCase() === 'true', - fallbackL2Generators: isSpot - ? [ - `${normedParam['includePhoenix']}`.toLowerCase() === 'true' && - MARKET_SUBSCRIBERS[normedMarketIndex].phoenix, - `${normedParam['includeSerum']}`.toLowerCase() === 'true' && - MARKET_SUBSCRIBERS[normedMarketIndex].serum, - ].filter((a) => !!a) - : [], - }); - - if (normedParam['grouping']) { - const finalDepth = normedParam['depth'] - ? parseInt(normedParam['depth'] as string) - : 10; - if (isNaN(parseInt(normedParam['grouping'] as string))) { - res - .status(400) - .send('Bad Request: grouping must be a number if supplied'); + const l2s = await Promise.all( + normedParams.map(async (normedParam) => { + const { normedMarketType, normedMarketIndex, error } = + validateDlobQuery( + driftClient, + driftEnv, + normedParam['marketType'] as string, + normedParam['marketIndex'] as string, + normedParam['marketName'] as string + ); + if (error) { + res.status(400).send(`Bad Request: ${error}`); return; } - const groupingBN = new BN( - parseInt(normedParam['grouping'] as string) - ); - const l2Formatted = l2WithBNToStrings( - groupL2(l2, groupingBN, finalDepth) - ); - if (`${normedParam['includeOracle']}`.toLowerCase() === 'true') { - addOracletoResponse( - l2Formatted, - driftClient, - normedMarketType, - normedMarketIndex - ); + const isSpot = isVariant(normedMarketType, 'spot'); + + let adjustedDepth = normedParam['depth'] ?? '10'; + if (normedParam['grouping'] !== undefined) { + // If grouping is also supplied, we want the entire book depth. + // we will apply depth after grouping + adjustedDepth = '-1'; } - return l2Formatted; - } else { - // make the BNs into strings - const l2Formatted = l2WithBNToStrings(l2); - if (`${normedParam['includeOracle']}`.toLowerCase() === 'true') { - addOracletoResponse( - l2Formatted, - driftClient, - normedMarketType, - normedMarketIndex - ); + + let l2Formatted: any; + if (useRedis) { + if ( + marketType === 'perp' && + normedParam['includeVamm'].toLowerCase() === 'true' && + normedParam['includeOracle'].toLowerCase() === 'true' && + !normedParam['grouping'] + ) { + let redisL2: string; + if (parseInt(adjustedDepth as string) === 5) { + redisL2 = await redisClient.client.get( + `last_update_orderbook_perp_${normedMarketIndex}_depth_5` + ); + } else if (parseInt(adjustedDepth as string) === 20) { + redisL2 = await redisClient.client.get( + `last_update_orderbook_perp_${normedMarketIndex}_depth_20` + ); + } else if (parseInt(adjustedDepth as string) === 100) { + redisL2 = await redisClient.client.get( + `last_update_orderbook_perp_${normedMarketIndex}_depth_100` + ); + } + if (redisL2) { + const parsedRedisL2 = JSON.parse(redisL2); + if (slotSource.getSlot() - parseInt(parsedRedisL2.slot) < 10) + l2Formatted = parsedRedisL2; + } + } else if ( + marketType === 'spot' && + normedParam['includePhoenix'].toLowerCase() === 'true' && + normedParam['includeSerum'].toLowerCase() === 'true' && + !normedParam['grouping'] + ) { + let redisL2: string; + if (parseInt(adjustedDepth as string) === 5) { + redisL2 = await redisClient.client.get( + `last_update_orderbook_spot_${normedMarketIndex}_depth_5` + ); + } else if (parseInt(adjustedDepth as string) === 20) { + redisL2 = await redisClient.client.get( + `last_update_orderbook_spot_${normedMarketIndex}_depth_20` + ); + } else if (parseInt(adjustedDepth as string) === 100) { + redisL2 = await redisClient.client.get( + `last_update_orderbook_spot_${normedMarketIndex}_depth_100` + ); + } + if (redisL2) { + const parsedRedisL2 = JSON.parse(redisL2); + if (slotSource.getSlot() - parseInt(parsedRedisL2.slot) < 10) + l2Formatted = parsedRedisL2; + } + } + + if (l2Formatted) { + cacheHitCounter.add(1); + return l2Formatted; + } } + + const l2 = dlobSubscriber.getL2({ + marketIndex: normedMarketIndex, + marketType: normedMarketType, + depth: parseInt(adjustedDepth as string), + includeVamm: isSpot + ? false + : `${normedParam['includeVamm']}`.toLowerCase() === 'true', + fallbackL2Generators: isSpot + ? [ + `${normedParam['includePhoenix']}`.toLowerCase() === 'true' && + MARKET_SUBSCRIBERS[normedMarketIndex].phoenix, + `${normedParam['includeSerum']}`.toLowerCase() === 'true' && + MARKET_SUBSCRIBERS[normedMarketIndex].serum, + ].filter((a) => !!a) + : [], + }); + + if (normedParam['grouping']) { + const finalDepth = normedParam['depth'] + ? parseInt(normedParam['depth'] as string) + : 10; + if (isNaN(parseInt(normedParam['grouping'] as string))) { + res + .status(400) + .send('Bad Request: grouping must be a number if supplied'); + return; + } + const groupingBN = new BN( + parseInt(normedParam['grouping'] as string) + ); + + l2Formatted = l2WithBNToStrings( + groupL2(l2, groupingBN, finalDepth) + ); + if (`${normedParam['includeOracle']}`.toLowerCase() === 'true') { + addOracletoResponse( + l2Formatted, + driftClient, + normedMarketType, + normedMarketIndex + ); + } + } else { + // make the BNs into strings + l2Formatted = l2WithBNToStrings(l2); + if (`${normedParam['includeOracle']}`.toLowerCase() === 'true') { + addOracletoResponse( + l2Formatted, + driftClient, + normedMarketType, + normedMarketIndex + ); + } + } + cacheMissCounter.add(1); return l2Formatted; - } - }); + }) + ); res.writeHead(200); res.end(JSON.stringify({ l2s })); diff --git a/src/publishers/dlobPublisher.ts b/src/publishers/dlobPublisher.ts index 7aa15d6..b6049e6 100644 --- a/src/publishers/dlobPublisher.ts +++ b/src/publishers/dlobPublisher.ts @@ -36,8 +36,6 @@ import { GeyserOrderSubscriber } from '../grpc/OrderSubscriberGRPC'; require('dotenv').config(); const stateCommitment: Commitment = 'processed'; -const ORDERBOOK_UPDATE_INTERVAL = 400; -const WS_FALLBACK_FETCH_INTERVAL = ORDERBOOK_UPDATE_INTERVAL * 10; const driftEnv = (process.env.ENV || 'devnet') as DriftEnv; const commitHash = process.env.COMMIT; const REDIS_HOST = process.env.REDIS_HOST || 'localhost'; @@ -62,6 +60,9 @@ const useOrderSubscriber = const useGrpc = process.env.USE_GRPC?.toLowerCase() === 'true'; const useWebsocket = process.env.USE_WEBSOCKET?.toLowerCase() === 'true'; +const ORDERBOOK_UPDATE_INTERVAL = useGrpc ? 500 : 1000; +const WS_FALLBACK_FETCH_INTERVAL = 10_000; + logger.info(`RPC endpoint: ${endpoint}`); logger.info(`WS endpoint: ${wsEndpoint}`); logger.info(`DriftEnv: ${driftEnv}`); diff --git a/yarn.lock b/yarn.lock index 6b0e84c..cc463b0 100644 --- a/yarn.lock +++ b/yarn.lock @@ -115,10 +115,10 @@ enabled "2.0.x" kuler "^2.0.0" -"@drift-labs/sdk@2.49.0-beta.17": - version "2.49.0-beta.17" - resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.49.0-beta.17.tgz#59d39712dcd2d0e0e6905b42c8efb743178277f3" - integrity sha512-0OLJFxDlKxIsQ32MuqVp/oHJFaeCy1IMigLmU92Zlapvjd8euR73VueiF4rGGROwrqO4IIamPgf68cD7D1V2HA== +"@drift-labs/sdk@2.52.0-beta.0": + version "2.52.0-beta.0" + resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.52.0-beta.0.tgz#03da2e192e3f2180238e5e464336e0fc1697c0e0" + integrity sha512-SnuXh06oJFzAvTQZ/dOjcWduhKC8rRMlyTE5Xqb3LfCcEkQ/B+XvxTItCSzAPh33qq0MMhZKjxGdqKezUxnV1w== dependencies: "@coral-xyz/anchor" "0.28.1-beta.2" "@ellipsis-labs/phoenix-sdk" "^1.4.2"