From 3ffb6e7330f3ccb9306a29af269817c413bf1949 Mon Sep 17 00:00:00 2001 From: Nour Alharithi Date: Tue, 26 Dec 2023 13:07:40 -0800 Subject: [PATCH] rotate redis connecitons --- src/index.ts | 132 +++++++++++++++++++++++++++++++++++++++----- src/serverLite.ts | 138 ++++++++++++++++++++++++++++++++++++++++++---- 2 files changed, 245 insertions(+), 25 deletions(-) diff --git a/src/index.ts b/src/index.ts index 9b1bc2b..a6737ea 100644 --- a/src/index.ts +++ b/src/index.ts @@ -57,23 +57,32 @@ import { RedisClient } from './utils/redisClient'; require('dotenv').config(); // Reading in Redis env vars -const REDIS_HOSTS_ENV = process.env.REDIS_HOSTS as string || 'localhost'; +const REDIS_HOSTS_ENV = (process.env.REDIS_HOSTS as string) || 'localhost'; const REDIS_HOSTS = REDIS_HOSTS_ENV.includes(',') ? REDIS_HOSTS_ENV.trim() .replace(/^\[|\]$/g, '') .split(/\s*,\s*/) : [REDIS_HOSTS_ENV]; -const REDIS_PASSWORDS_ENV = process.env.REDIS_HOSTS as string; +const REDIS_PASSWORDS_ENV = (process.env.REDIS_PASSWORDS as string) || ''; const REDIS_PASSWORDS = REDIS_PASSWORDS_ENV.includes(',') ? REDIS_PASSWORDS_ENV.trim() .replace(/^\[|\]$/g, '') .split(/\s*,\s*/) : [REDIS_PASSWORDS_ENV]; -if (REDIS_HOSTS.length !== REDIS_PASSWORDS.length) { - throw('REDIS_HOSTS and REDIS_PASSWORDS must be the same length'); + +const REDIS_PORTS_ENV = (process.env.REDIS_PORTS as string) || '6379'; +const REDIS_PORTS = REDIS_PORTS_ENV.includes(',') + ? REDIS_PORTS_ENV.trim() + .replace(/^\[|\]$/g, '') + .split(/\s*,\s*/) + : [REDIS_PORTS_ENV]; +if ( + REDIS_PORTS.length !== REDIS_PASSWORDS.length || + REDIS_PORTS.length !== REDIS_HOSTS.length +) { + throw 'REDIS_HOSTS and REDIS_PASSWORDS and REDIS_PORTS must be the same length'; } -const REDIS_PORT = '6379'; const driftEnv = (process.env.ENV || 'devnet') as DriftEnv; const commitHash = process.env.COMMIT; @@ -84,6 +93,8 @@ const stateCommitment: Commitment = 'processed'; const serverPort = process.env.PORT || 6969; export const ORDERBOOK_UPDATE_INTERVAL = 1000; const WS_FALLBACK_FETCH_INTERVAL = ORDERBOOK_UPDATE_INTERVAL * 10; +const SLOT_STALENESS_TOLERANCE = + parseInt(process.env.SLOT_STALENESS_TOLERANCE) || 20; const useWebsocket = process.env.USE_WEBSOCKET?.toLowerCase() === 'true'; const rateLimitCallsPerSecond = process.env.RATE_LIMIT_CALLS_PER_SECOND ? parseInt(process.env.RATE_LIMIT_CALLS_PER_SECOND) @@ -335,18 +346,39 @@ const main = async () => { `DLOBSubscriber initialized in ${Date.now() - initDlobSubscriberStart} ms` ); - let redisClient: RedisClient; const redisClients: Array = []; - let redisClientMap: + const spotMarketRedisMap: Map< + number, + { client: RedisClient; clientIndex: number } + > = new Map(); + const perpMarketRedisMap: Map< + number, + { client: RedisClient; clientIndex: number } + > = new Map(); if (useRedis) { logger.info('Connecting to redis'); for (let i = 0; i < REDIS_HOSTS.length; i++) { redisClients.push( - new RedisClient(REDIS_HOSTS[i], REDIS_PORT, REDIS_PASSWORDS[i]) + new RedisClient( + REDIS_HOSTS[i], + REDIS_PORTS[i], + REDIS_PASSWORDS[i] || undefined + ) ); await redisClients[i].connect(); } - redisClient = redisClients[0]; + for (let i = 0; i < sdkConfig.SPOT_MARKETS.length; i++) { + spotMarketRedisMap.set(sdkConfig.SPOT_MARKETS[i].marketIndex, { + client: redisClients[0], + clientIndex: 0, + }); + } + for (let i = 0; i < sdkConfig.PERP_MARKETS.length; i++) { + perpMarketRedisMap.set(sdkConfig.PERP_MARKETS[i].marketIndex, { + client: redisClients[0], + clientIndex: 0, + }); + } } logger.info(`Initializing all market subscribers...`); @@ -757,6 +789,7 @@ const main = async () => { !grouping ) { let redisL2: string; + const redisClient = perpMarketRedisMap.get(normedMarketIndex).client; if (parseInt(adjustedDepth as string) === 5) { redisL2 = await redisClient.client.get( `last_update_orderbook_perp_${normedMarketIndex}_depth_5` @@ -772,9 +805,22 @@ const main = async () => { } if ( redisL2 && - dlobProvider.getSlot() - parseInt(JSON.parse(redisL2).slot) < 10 - ) + dlobProvider.getSlot() - parseInt(JSON.parse(redisL2).slot) < + SLOT_STALENESS_TOLERANCE + ) { l2Formatted = redisL2; + } else { + if (redisL2 && redisClients.length > 1) { + const nextClientIndex = + (perpMarketRedisMap.get(normedMarketIndex).clientIndex + 1) % + redisClients.length; + perpMarketRedisMap.set(normedMarketIndex, { + client: redisClients[nextClientIndex], + clientIndex: nextClientIndex, + }); + console.log('Rotated redis client to index ', nextClientIndex); + } + } } else if ( isSpot && `${includeSerum}`?.toLowerCase() === 'true' && @@ -783,6 +829,7 @@ const main = async () => { !grouping ) { let redisL2: string; + const redisClient = spotMarketRedisMap.get(normedMarketIndex).client; if (parseInt(adjustedDepth as string) === 5) { redisL2 = await redisClient.client.get( `last_update_orderbook_spot_${normedMarketIndex}_depth_5` @@ -798,9 +845,22 @@ const main = async () => { } if ( redisL2 && - dlobProvider.getSlot() - parseInt(JSON.parse(redisL2).slot) < 10 - ) + dlobProvider.getSlot() - parseInt(JSON.parse(redisL2).slot) < + SLOT_STALENESS_TOLERANCE + ) { l2Formatted = redisL2; + } else { + if (redisL2 && redisClients.length > 1) { + const nextClientIndex = + (spotMarketRedisMap.get(normedMarketIndex).clientIndex + 1) % + redisClients.length; + spotMarketRedisMap.set(normedMarketIndex, { + client: redisClients[nextClientIndex], + clientIndex: nextClientIndex, + }); + console.log('Rotated redis client to index ', nextClientIndex); + } + } } if (l2Formatted) { @@ -939,6 +999,8 @@ const main = async () => { !normedParam['grouping'] ) { let redisL2: string; + const redisClient = + perpMarketRedisMap.get(normedMarketIndex).client; if (parseInt(adjustedDepth as string) === 5) { redisL2 = await redisClient.client.get( `last_update_orderbook_perp_${normedMarketIndex}_depth_5` @@ -954,8 +1016,27 @@ const main = async () => { } if (redisL2) { const parsedRedisL2 = JSON.parse(redisL2); - if (dlobProvider.getSlot() - parseInt(parsedRedisL2.slot) < 10) + if ( + dlobProvider.getSlot() - parseInt(parsedRedisL2.slot) < + SLOT_STALENESS_TOLERANCE + ) { l2Formatted = parsedRedisL2; + } else { + if (redisClients.length > 1) { + const nextClientIndex = + (perpMarketRedisMap.get(normedMarketIndex).clientIndex + + 1) % + redisClients.length; + perpMarketRedisMap.set(normedMarketIndex, { + client: redisClients[nextClientIndex], + clientIndex: nextClientIndex, + }); + console.log( + 'Rotated redis client to index ', + nextClientIndex + ); + } + } } } else if ( isSpot && @@ -964,6 +1045,8 @@ const main = async () => { !normedParam['grouping'] ) { let redisL2: string; + const redisClient = + spotMarketRedisMap.get(normedMarketIndex).client; if (parseInt(adjustedDepth as string) === 5) { redisL2 = await redisClient.client.get( `last_update_orderbook_spot_${normedMarketIndex}_depth_5` @@ -979,8 +1062,27 @@ const main = async () => { } if (redisL2) { const parsedRedisL2 = JSON.parse(redisL2); - if (dlobProvider.getSlot() - parseInt(parsedRedisL2.slot) < 10) + if ( + dlobProvider.getSlot() - parseInt(parsedRedisL2.slot) < + SLOT_STALENESS_TOLERANCE + ) { l2Formatted = parsedRedisL2; + } else { + if (redisClients.length > 1) { + const nextClientIndex = + (spotMarketRedisMap.get(normedMarketIndex).clientIndex + + 1) % + redisClients.length; + spotMarketRedisMap.set(normedMarketIndex, { + client: redisClients[nextClientIndex], + clientIndex: nextClientIndex, + }); + console.log( + 'Rotated redis client to index ', + nextClientIndex + ); + } + } } } diff --git a/src/serverLite.ts b/src/serverLite.ts index 187b8ed..aeea93d 100644 --- a/src/serverLite.ts +++ b/src/serverLite.ts @@ -30,9 +30,33 @@ import { 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; +// Reading in Redis env vars +const REDIS_HOSTS_ENV = (process.env.REDIS_HOSTS as string) || 'localhost'; +const REDIS_HOSTS = REDIS_HOSTS_ENV.includes(',') + ? REDIS_HOSTS_ENV.trim() + .replace(/^\[|\]$/g, '') + .split(/\s*,\s*/) + : [REDIS_HOSTS_ENV]; + +const REDIS_PASSWORDS_ENV = (process.env.REDIS_PASSWORDS as string) || ''; +const REDIS_PASSWORDS = REDIS_PASSWORDS_ENV.includes(',') + ? REDIS_PASSWORDS_ENV.trim() + .replace(/^\[|\]$/g, '') + .split(/\s*,\s*/) + : [REDIS_PASSWORDS_ENV]; + +const REDIS_PORTS_ENV = (process.env.REDIS_PORTS as string) || '6379'; +const REDIS_PORTS = REDIS_PORTS_ENV.includes(',') + ? REDIS_PORTS_ENV.trim() + .replace(/^\[|\]$/g, '') + .split(/\s*,\s*/) + : [REDIS_PORTS_ENV]; +if ( + REDIS_PORTS.length !== REDIS_PASSWORDS.length || + REDIS_PORTS.length !== REDIS_HOSTS.length +) { + throw 'REDIS_HOSTS and REDIS_PASSWORDS and REDIS_PORTS must be the same length'; +} const driftEnv = (process.env.ENV || 'devnet') as DriftEnv; const commitHash = process.env.COMMIT; @@ -132,10 +156,38 @@ logger.info(`Commit: ${commitHash}`); const main = async () => { // Redis connect - logger.info('Connecting to redis'); - const redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD); - await redisClient.connect(); - + const redisClients: Array = []; + const spotMarketRedisMap: Map< + number, + { client: RedisClient; clientIndex: number } + > = new Map(); + const perpMarketRedisMap: Map< + number, + { client: RedisClient; clientIndex: number } + > = new Map(); + for (let i = 0; i < REDIS_HOSTS.length; i++) { + redisClients.push( + new RedisClient( + REDIS_HOSTS[i], + REDIS_PORTS[i], + REDIS_PASSWORDS[i] || undefined + ) + ); + await redisClients[i].connect(); + } + for (let i = 0; i < sdkConfig.SPOT_MARKETS.length; i++) { + spotMarketRedisMap.set(sdkConfig.SPOT_MARKETS[i].marketIndex, { + client: redisClients[0], + clientIndex: 0, + }); + } + for (let i = 0; i < sdkConfig.PERP_MARKETS.length; i++) { + perpMarketRedisMap.set(sdkConfig.PERP_MARKETS[i].marketIndex, { + client: redisClients[0], + clientIndex: 0, + }); + } + // Slot subscriber for source const connection = new Connection(endpoint, { commitment: stateCommitment, @@ -154,7 +206,13 @@ const main = async () => { ); const handleStartup = async (_req, res, _next) => { - if (redisClient.connected) { + let healthy= false; + for (const redisClient of redisClients) { + if (redisClient.connected) { + healthy = true; + } + } + if (healthy) { res.writeHead(200); res.end('OK'); } else { @@ -205,6 +263,7 @@ const main = async () => { !grouping ) { let redisL2: string; + const redisClient = perpMarketRedisMap.get(normedMarketIndex).client; if (parseInt(adjustedDepth as string) === 5) { redisL2 = await redisClient.client.get( `last_update_orderbook_perp_${normedMarketIndex}_depth_5` @@ -223,6 +282,17 @@ const main = async () => { slotSubscriber.getSlot() - parseInt(JSON.parse(redisL2).slot); if (slotDiff < SLOT_STALENESS_TOLERANCE) { l2Formatted = redisL2; + } else { + if (redisClients.length > 1) { + const nextClientIndex = + (perpMarketRedisMap.get(normedMarketIndex).clientIndex + 1) % + redisClients.length; + perpMarketRedisMap.set(normedMarketIndex, { + client: redisClients[nextClientIndex], + clientIndex: nextClientIndex, + }); + console.log('Rotated redis client to index ', nextClientIndex); + } } } } else if ( @@ -233,6 +303,7 @@ const main = async () => { !grouping ) { let redisL2: string; + const redisClient = spotMarketRedisMap.get(normedMarketIndex).client; if (parseInt(adjustedDepth as string) === 5) { redisL2 = await redisClient.client.get( `last_update_orderbook_spot_${normedMarketIndex}_depth_5` @@ -251,6 +322,17 @@ const main = async () => { slotSubscriber.getSlot() - parseInt(JSON.parse(redisL2).slot); if (slotDiff < SLOT_STALENESS_TOLERANCE) { l2Formatted = redisL2; + } else { + if (redisClients.length > 1) { + const nextClientIndex = + (spotMarketRedisMap.get(normedMarketIndex).clientIndex + 1) % + redisClients.length; + spotMarketRedisMap.set(normedMarketIndex, { + client: redisClients[nextClientIndex], + clientIndex: nextClientIndex, + }); + console.log('Rotated redis client to index ', nextClientIndex); + } } } } @@ -341,6 +423,8 @@ const main = async () => { !normedParam['grouping'] ) { let redisL2: string; + const redisClient = + perpMarketRedisMap.get(normedMarketIndex).client; if (parseInt(adjustedDepth as string) === 5) { redisL2 = await redisClient.client.get( `last_update_orderbook_perp_${normedMarketIndex}_depth_5` @@ -359,8 +443,24 @@ const main = async () => { if ( slotSubscriber.getSlot() - parseInt(parsedRedisL2.slot) < SLOT_STALENESS_TOLERANCE - ) + ) { l2Formatted = parsedRedisL2; + } else { + if (redisClients.length > 1) { + const nextClientIndex = + (perpMarketRedisMap.get(normedMarketIndex).clientIndex + + 1) % + redisClients.length; + perpMarketRedisMap.set(normedMarketIndex, { + client: redisClients[nextClientIndex], + clientIndex: nextClientIndex, + }); + console.log( + 'Rotated redis client to index ', + nextClientIndex + ); + } + } } } else if ( isSpot && @@ -369,6 +469,8 @@ const main = async () => { !normedParam['grouping'] ) { let redisL2: string; + const redisClient = + spotMarketRedisMap.get(normedMarketIndex).client; if (parseInt(adjustedDepth as string) === 5) { redisL2 = await redisClient.client.get( `last_update_orderbook_spot_${normedMarketIndex}_depth_5` @@ -387,8 +489,24 @@ const main = async () => { if ( slotSubscriber.getSlot() - parseInt(parsedRedisL2.slot) < SLOT_STALENESS_TOLERANCE - ) + ) { l2Formatted = parsedRedisL2; + } else { + if (redisClients.length > 1) { + const nextClientIndex = + (spotMarketRedisMap.get(normedMarketIndex).clientIndex + + 1) % + redisClients.length; + spotMarketRedisMap.set(normedMarketIndex, { + client: redisClients[nextClientIndex], + clientIndex: nextClientIndex, + }); + console.log( + 'Rotated redis client to index ', + nextClientIndex + ); + } + } } if (l2Formatted) {