diff --git a/package.json b/package.json index 7ae52d9..7697e4a 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.53.0-beta.5", + "@drift-labs/sdk": "2.53.0-beta.8", "@opentelemetry/api": "^1.1.0", "@opentelemetry/auto-instrumentations-node": "^0.31.1", "@opentelemetry/exporter-prometheus": "^0.31.0", diff --git a/src/index.ts b/src/index.ts index aa7fc59..931aef3 100644 --- a/src/index.ts +++ b/src/index.ts @@ -56,9 +56,33 @@ 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; +// 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; @@ -69,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) @@ -320,11 +346,39 @@ const main = async () => { `DLOBSubscriber initialized in ${Date.now() - initDlobSubscriberStart} ms` ); - let redisClient: RedisClient; + const redisClients: Array = []; + 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'); - redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD); - await redisClient.connect(); + 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, + }); + } } logger.info(`Initializing all market subscribers...`); @@ -735,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` @@ -750,9 +805,24 @@ 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} for perp market ${normedMarketIndex}` + ); + } + } } else if ( isSpot && `${includeSerum}`?.toLowerCase() === 'true' && @@ -761,6 +831,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` @@ -776,9 +847,24 @@ 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} for spot market ${normedMarketIndex}` + ); + } + } } if (l2Formatted) { @@ -917,6 +1003,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` @@ -932,8 +1020,26 @@ 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} for perp market ${normedMarketIndex}` + ); + } + } } } else if ( isSpot && @@ -942,6 +1048,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` @@ -957,8 +1065,26 @@ 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} for spot market ${normedMarketIndex}` + ); + } + } } } diff --git a/src/publishers/dlobPublisher.ts b/src/publishers/dlobPublisher.ts index f54e46b..10d9504 100644 --- a/src/publishers/dlobPublisher.ts +++ b/src/publishers/dlobPublisher.ts @@ -324,7 +324,8 @@ const main = async () => { const initAllMarketSubscribersStart = Date.now(); MARKET_SUBSCRIBERS = await initializeAllMarketSubscribers(driftClient); logger.info( - `All market subscribers initialized in ${Date.now() - initAllMarketSubscribersStart + `All market subscribers initialized in ${ + Date.now() - initAllMarketSubscribersStart } ms` ); diff --git a/src/serverLite.ts b/src/serverLite.ts index 187b8ed..f607797 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,9 +156,37 @@ 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, { @@ -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,19 @@ 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} for perp market ${normedMarketIndex}` + ); + } } } } else if ( @@ -233,6 +305,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 +324,19 @@ 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} for spot market ${normedMarketIndex}` + ); + } } } } @@ -341,6 +427,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 +447,23 @@ 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} for perp market ${normedMarketIndex}` + ); + } + } } } else if ( isSpot && @@ -369,6 +472,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 +492,23 @@ 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} for spot market ${normedMarketIndex}` + ); + } + } } if (l2Formatted) { diff --git a/yarn.lock b/yarn.lock index 55b5cfd..cbb5f50 100644 --- a/yarn.lock +++ b/yarn.lock @@ -115,10 +115,10 @@ enabled "2.0.x" kuler "^2.0.0" -"@drift-labs/sdk@2.53.0-beta.5": - version "2.53.0-beta.5" - resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.53.0-beta.5.tgz#2cbd77105e3248886a345cb07a67fd60407eae3a" - integrity sha512-EAW+eqSkAe490IblsC0/hFWeYFLU9wZ71yq1nw6Aemq0RC6ygsN/eaQiVMOO1v8MQ5ar/ef6n27FFHO4k5UOkg== +"@drift-labs/sdk@2.53.0-beta.8": + version "2.53.0-beta.8" + resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.53.0-beta.8.tgz#bc12747a8f2c4fa26b979fa6a8202a3bd3f52b2b" + integrity sha512-R2OPzgWjEVrJMeSwXa/UiquM6c72iOnFdoHbXJ4JJPd/KJDvghjOIXshF+7ttf80Uwswrox0sEhCJOYcYZOMIQ== dependencies: "@coral-xyz/anchor" "0.28.1-beta.2" "@ellipsis-labs/phoenix-sdk" "^1.4.2"