diff --git a/src/index.ts b/src/index.ts index a4c9fd6..0164064 100644 --- a/src/index.ts +++ b/src/index.ts @@ -305,7 +305,7 @@ const main = async (): Promise => { const fetchFromRedis = async ( key: string, selectionCriteria: (responses: any) => any - ): Promise => { + ): Promise => { const redisResponses = await Promise.all( redisClients.map((client) => client.getRaw(key)) ); @@ -650,7 +650,8 @@ const main = async (): Promise => { const depth = Math.min(parseInt(adjustedDepth as string) ?? 1, 100); redisL2['bids'] = redisL2['bids']?.slice(0, depth); redisL2['asks'] = redisL2['asks']?.slice(0, depth); - + console.log(redisL2['slot']); + console.log(SLOT_STALENESS_TOLERANCE); if ( redisL2 && dlobProvider.getSlot() - redisL2['slot'] < SLOT_STALENESS_TOLERANCE @@ -675,16 +676,16 @@ const main = async (): Promise => { ) { l2Formatted = JSON.stringify(redisL2); } + } - if (l2Formatted) { - cacheHitCounter.add(1, { - miss: false, - path: req.baseUrl + req.path, - }); - res.writeHead(200); - res.end(l2Formatted); - return; - } + if (l2Formatted) { + cacheHitCounter.add(1, { + miss: false, + path: req.baseUrl + req.path, + }); + res.writeHead(200); + res.end(l2Formatted); + return; } let validateIncludeVamm = false; @@ -824,14 +825,14 @@ const main = async (): Promise => { l2Formatted = redisL2; } } + } - if (l2Formatted) { - cacheHitCounter.add(1, { - miss: false, - path: req.baseUrl + req.path, - }); - return l2Formatted; - } + if (l2Formatted) { + cacheHitCounter.add(1, { + miss: false, + path: req.baseUrl + req.path, + }); + return l2Formatted; } let validateIncludeVamm = false; diff --git a/src/serverLite.ts b/src/serverLite.ts index f79044c..7fd921f 100644 --- a/src/serverLite.ts +++ b/src/serverLite.ts @@ -10,7 +10,6 @@ import { DriftEnv, SlotSubscriber, initialize, - isVariant, MarketType, getVariant, } from '@drift-labs/sdk'; @@ -26,7 +25,7 @@ import { runtimeSpecsGauge, } from './core/metrics'; import { handleResponseTime } from './core/middleware'; -import { errorHandler, sleep } from './utils/utils'; +import { errorHandler, selectMostRecentBySlot, sleep } from './utils/utils'; import { setGlobalDispatcher, Agent } from 'undici'; setGlobalDispatcher( @@ -60,8 +59,7 @@ const serverPort = process.env.PORT || 6969; export const ORDERBOOK_UPDATE_INTERVAL = 1000; const WS_FALLBACK_FETCH_INTERVAL = ORDERBOOK_UPDATE_INTERVAL * 60; const SLOT_STALENESS_TOLERANCE = - parseInt(process.env.SLOT_STALENESS_TOLERANCE) || 35; -const ROTATION_COOLDOWN = parseInt(process.env.ROTATION_COOLDOWN) || 5000; + parseInt(process.env.SLOT_STALENESS_TOLERANCE) || 100000; const logFormat = ':remote-addr - :remote-user [:date[clf]] ":method :url HTTP/:http-version" :status :res[content-length] ":referrer" ":user-agent" :req[x-forwarded-for]'; @@ -132,98 +130,20 @@ const main = async (): Promise => { // Handle redis client initialization and rotation maps const redisClients: Array = []; - const spotMarketRedisMap: Map< - number, - { - client: RedisClient; - clientIndex: number; - lastRotationTime: number; - lock: boolean; - } - > = new Map(); - const perpMarketRedisMap: Map< - number, - { - client: RedisClient; - clientIndex: number; - lastRotationTime: number; - lock: boolean; - } - > = new Map(); logger.info('Connecting to redis'); for (let i = 0; i < REDIS_CLIENTS.length; i++) { redisClients.push(new RedisClient({ prefix: REDIS_CLIENTS[i] })); } - for (let i = 0; i < sdkConfig.SPOT_MARKETS.length; i++) { - spotMarketRedisMap.set(sdkConfig.SPOT_MARKETS[i].marketIndex, { - client: redisClients[0], - clientIndex: 0, - lastRotationTime: 0, - lock: false, - }); - } - for (let i = 0; i < sdkConfig.PERP_MARKETS.length; i++) { - perpMarketRedisMap.set(sdkConfig.PERP_MARKETS[i].marketIndex, { - client: redisClients[0], - clientIndex: 0, - lastRotationTime: 0, - lock: false, - }); - } - - function canRotate(marketType: MarketType, marketIndex: number) { - if (isVariant(marketType, 'spot')) { - const state = spotMarketRedisMap.get(marketIndex); - if (state) { - const now = Date.now(); - if (now - state.lastRotationTime > ROTATION_COOLDOWN && !state.lock) { - state.lastRotationTime = now; - return true; - } - } - } else { - const state = perpMarketRedisMap.get(marketIndex); - if (state) { - const now = Date.now(); - if (now - state.lastRotationTime > ROTATION_COOLDOWN && !state.lock) { - state.lastRotationTime = now; - return true; - } - } - } - return false; - } - - function rotateClient(marketType: MarketType, marketIndex: number) { - if (isVariant(marketType, 'spot')) { - const state = spotMarketRedisMap.get(marketIndex); - if (state) { - state.lock = true; - const nextClientIndex = (state.clientIndex + 1) % redisClients.length; - state.client = redisClients[nextClientIndex]; - state.clientIndex = nextClientIndex; - logger.info( - `Rotated redis client to index ${nextClientIndex} for spot market ${marketIndex}` - ); - state.lastRotationTime = Date.now(); - state.lock = false; - } - } else { - const state = perpMarketRedisMap.get(marketIndex); - if (state) { - state.lock = true; - const nextClientIndex = (state.clientIndex + 1) % redisClients.length; - state.client = redisClients[nextClientIndex]; - state.clientIndex = nextClientIndex; - logger.info( - `Rotated redis client to index ${nextClientIndex} for perp market ${marketIndex}` - ); - state.lastRotationTime = Date.now(); - state.lock = false; - } - } - } + const fetchFromRedis = async ( + key: string, + selectionCriteria: (responses: any) => any + ): Promise => { + const redisResponses = await Promise.all( + redisClients.map((client) => client.getRaw(key)) + ); + return selectionCriteria(redisResponses); + }; const handleStartup = async (_req, res, _next) => { if (slotSubscriber.currentSlot && !redisClients.some((c) => !c.connected)) { @@ -253,32 +173,24 @@ const main = async (): Promise => { const normedMarketIndex = parseInt(marketIndex as string); const normedMarketType = isSpot ? MarketType.SPOT : MarketType.PERP; - const redisClient = ( - isSpot ? spotMarketRedisMap : perpMarketRedisMap - ).get(normedMarketIndex).client; - const redisL3 = await redisClient.getRaw( + const redisL3 = await fetchFromRedis( `last_update_orderbook_l3_${getVariant( normedMarketType - )}_${normedMarketIndex}` + )}_${normedMarketIndex}`, + selectMostRecentBySlot ); if ( redisL3 && - slotSubscriber.getSlot() - parseInt(JSON.parse(redisL3).slot) < - SLOT_STALENESS_TOLERANCE + slotSubscriber.getSlot() - redisL3['slot'] < SLOT_STALENESS_TOLERANCE ) { cacheHitCounter.add(1, { miss: false, path: req.baseUrl + req.path, }); res.writeHead(200); - res.end(redisL3); + res.end(JSON.stringify(redisL3)); return; } else { - if (redisL3 && redisClients.length > 1) { - if (canRotate(normedMarketType, normedMarketIndex)) { - rotateClient(normedMarketType, normedMarketIndex); - } - } cacheHitCounter.add(1, { miss: true, path: req.baseUrl + req.path, diff --git a/src/utils/utils.ts b/src/utils/utils.ts index 2d9dd5e..7aa2ab8 100644 --- a/src/utils/utils.ts +++ b/src/utils/utils.ts @@ -473,7 +473,12 @@ export type SubscriberLookup = { }; }; -export const selectMostRecentBySlot = (responses: any[]): any => { +export const selectMostRecentBySlot = ( + responses: any[] +): { + slot: number; + [key: string]: any; +} => { const parsedResponses = responses .map((response) => { try {