server lite uses new rotation

This commit is contained in:
Nour Alharithi
2024-12-03 14:33:21 -08:00
parent cb8dcc7ec2
commit 669d4a48fd

View File

@@ -10,7 +10,6 @@ import {
DriftEnv, DriftEnv,
SlotSubscriber, SlotSubscriber,
initialize, initialize,
isVariant,
MarketType, MarketType,
getVariant, getVariant,
} from '@drift-labs/sdk'; } from '@drift-labs/sdk';
@@ -26,7 +25,7 @@ import {
runtimeSpecsGauge, runtimeSpecsGauge,
} from './core/metrics'; } from './core/metrics';
import { handleResponseTime } from './core/middleware'; import { handleResponseTime } from './core/middleware';
import { errorHandler, sleep } from './utils/utils'; import { errorHandler, selectMostRecentBySlot, sleep } from './utils/utils';
import { setGlobalDispatcher, Agent } from 'undici'; import { setGlobalDispatcher, Agent } from 'undici';
setGlobalDispatcher( setGlobalDispatcher(
@@ -60,8 +59,7 @@ const serverPort = process.env.PORT || 6969;
export const ORDERBOOK_UPDATE_INTERVAL = 1000; export const ORDERBOOK_UPDATE_INTERVAL = 1000;
const WS_FALLBACK_FETCH_INTERVAL = ORDERBOOK_UPDATE_INTERVAL * 60; const WS_FALLBACK_FETCH_INTERVAL = ORDERBOOK_UPDATE_INTERVAL * 60;
const SLOT_STALENESS_TOLERANCE = const SLOT_STALENESS_TOLERANCE =
parseInt(process.env.SLOT_STALENESS_TOLERANCE) || 35; parseInt(process.env.SLOT_STALENESS_TOLERANCE) || 100000;
const ROTATION_COOLDOWN = parseInt(process.env.ROTATION_COOLDOWN) || 5000;
const logFormat = const logFormat =
':remote-addr - :remote-user [:date[clf]] ":method :url HTTP/:http-version" :status :res[content-length] ":referrer" ":user-agent" :req[x-forwarded-for]'; ':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<void> => {
// Handle redis client initialization and rotation maps // Handle redis client initialization and rotation maps
const redisClients: Array<RedisClient> = []; const redisClients: Array<RedisClient> = [];
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'); logger.info('Connecting to redis');
for (let i = 0; i < REDIS_CLIENTS.length; i++) { for (let i = 0; i < REDIS_CLIENTS.length; i++) {
redisClients.push(new RedisClient({ prefix: REDIS_CLIENTS[i] })); redisClients.push(new RedisClient({ prefix: REDIS_CLIENTS[i] }));
} }
for (let i = 0; i < sdkConfig.SPOT_MARKETS.length; i++) { const fetchFromRedis = async (
spotMarketRedisMap.set(sdkConfig.SPOT_MARKETS[i].marketIndex, { key: string,
client: redisClients[0], selectionCriteria: (responses: any) => any
clientIndex: 0, ): Promise<JSON> => {
lastRotationTime: 0, const redisResponses = await Promise.all(
lock: false, redisClients.map((client) => client.getRaw(key))
});
}
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(); return selectionCriteria(redisResponses);
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 handleStartup = async (_req, res, _next) => { const handleStartup = async (_req, res, _next) => {
if (slotSubscriber.currentSlot && !redisClients.some((c) => !c.connected)) { if (slotSubscriber.currentSlot && !redisClients.some((c) => !c.connected)) {
@@ -253,32 +173,24 @@ const main = async (): Promise<void> => {
const normedMarketIndex = parseInt(marketIndex as string); const normedMarketIndex = parseInt(marketIndex as string);
const normedMarketType = isSpot ? MarketType.SPOT : MarketType.PERP; const normedMarketType = isSpot ? MarketType.SPOT : MarketType.PERP;
const redisClient = ( const redisL3 = await fetchFromRedis(
isSpot ? spotMarketRedisMap : perpMarketRedisMap
).get(normedMarketIndex).client;
const redisL3 = await redisClient.getRaw(
`last_update_orderbook_l3_${getVariant( `last_update_orderbook_l3_${getVariant(
normedMarketType normedMarketType
)}_${normedMarketIndex}` )}_${normedMarketIndex}`,
selectMostRecentBySlot
); );
if ( if (
redisL3 && redisL3 &&
slotSubscriber.getSlot() - parseInt(JSON.parse(redisL3).slot) < slotSubscriber.getSlot() - redisL3['slot'] < SLOT_STALENESS_TOLERANCE
SLOT_STALENESS_TOLERANCE
) { ) {
cacheHitCounter.add(1, { cacheHitCounter.add(1, {
miss: false, miss: false,
path: req.baseUrl + req.path, path: req.baseUrl + req.path,
}); });
res.writeHead(200); res.writeHead(200);
res.end(redisL3); res.end(JSON.stringify(redisL3));
return; return;
} else { } else {
if (redisL3 && redisClients.length > 1) {
if (canRotate(normedMarketType, normedMarketIndex)) {
rotateClient(normedMarketType, normedMarketIndex);
}
}
cacheHitCounter.add(1, { cacheHitCounter.add(1, {
miss: true, miss: true,
path: req.baseUrl + req.path, path: req.baseUrl + req.path,