From 2a6899384c7316c22e7b037e3f2bd20dfad0e113 Mon Sep 17 00:00:00 2001 From: moosecat Date: Tue, 3 Dec 2024 13:47:02 -0800 Subject: [PATCH] Nour/improved rotation (#290) * improved rotation * fix bugs * only parse slot if necessary * add metrics for ws source * add select by recent slot to utils * fix solana web3.js version --------- Co-authored-by: wphan --- drift-common | 2 +- example/wsClient.ts | 13 +- package.json | 2 +- src/core/metrics.ts | 1 + src/index.ts | 362 ++++++++++-------------------- src/publishers/dlobPublisher.ts | 1 - src/publishers/tradesPublisher.ts | 4 +- src/utils/utils.ts | 15 ++ src/wsConnectionManager.ts | 185 ++++++++++----- yarn.lock | 105 +++------ 10 files changed, 307 insertions(+), 383 deletions(-) diff --git a/drift-common b/drift-common index 915ab84..0303c32 160000 --- a/drift-common +++ b/drift-common @@ -1 +1 @@ -Subproject commit 915ab84fe0bf55f88945ae665342dfacb80c528c +Subproject commit 0303c32d038e1ea1c9c21fc5c31f16730eed5fc3 diff --git a/example/wsClient.ts b/example/wsClient.ts index f9151d5..a366c1d 100644 --- a/example/wsClient.ts +++ b/example/wsClient.ts @@ -1,5 +1,6 @@ import WebSocket from 'ws'; -const ws = new WebSocket('wss://dlob.drift.trade/ws'); +// const ws = new WebSocket('wss://dlob.drift.trade/ws'); +const ws = new WebSocket('http://localhost:3000/ws'); import { sleep } from '../src/utils/utils'; ws.on('open', async () => { @@ -12,22 +13,26 @@ ws.on('open', async () => { ws.send(JSON.stringify({ type: 'subscribe', marketType: 'spot', channel: 'orderbook', market: 'SOL' })); await sleep(5000); - ws.send(JSON.stringify({ type: 'unsubscribe', marketType: 'perp', channel: 'orderbook', market: 'SOL-PERP' })); + // ws.send(JSON.stringify({ type: 'unsubscribe', marketType: 'perp', channel: 'orderbook', market: 'SOL-PERP' })); console.log("####################"); // Subscribe to trades data ws.send(JSON.stringify({ type: 'subscribe', marketType: 'perp', channel: 'trades', market: 'SOL-PERP' })); - ws.send(JSON.stringify({ type: 'subscribe', marketType: 'spot', channel: 'trades', market: 'SOL' })); + ws.send(JSON.stringify({ type: 'subscribe', marketType: 'spot', channel: 'trades', market: 'SOL' })); await sleep(5000); - ws.send(JSON.stringify({ type: 'unsubscribe', marketType: 'perp', channel: 'trades', market: 'SOL-PERP' })); + // ws.send(JSON.stringify({ type: 'unsubscribe', marketType: 'perp', channel: 'trades', market: 'SOL-PERP' })); console.log("####################"); }); ws.on('message', (data: WebSocket.Data) => { try { const message = JSON.parse(data.toString()); + if (!message.channel) { + console.log(`Received data: ${JSON.stringify(message)}`); + return; + } console.log(`Received data from channel: ${JSON.stringify(message.channel)}`); // book and trades data is in message.data } catch (e) { diff --git a/package.json b/package.json index a516fc1..6765ccd 100644 --- a/package.json +++ b/package.json @@ -14,7 +14,7 @@ "@opentelemetry/sdk-node": "^0.31.0", "@project-serum/anchor": "^0.19.1-beta.1", "@project-serum/serum": "^0.13.65", - "@solana/web3.js": "^1.73.3", + "@solana/web3.js": "1.95.2", "@triton-one/yellowstone-grpc": "^0.3.0", "@types/redis": "^4.0.11", "@types/ws": "^8.5.8", diff --git a/src/core/metrics.ts b/src/core/metrics.ts index bd6b883..d32294f 100644 --- a/src/core/metrics.ts +++ b/src/core/metrics.ts @@ -8,6 +8,7 @@ import { View, } from '@opentelemetry/sdk-metrics-base'; import { SlotSource } from '@drift-labs/sdk'; +require('dotenv').config(); /** * Creates {count} buckets of size {increment} starting from {start}. Each bucket stores the count of values within its "size". diff --git a/src/index.ts b/src/index.ts index 950daef..a4c9fd6 100644 --- a/src/index.ts +++ b/src/index.ts @@ -17,7 +17,6 @@ import { initialize, isVariant, OrderSubscriber, - MarketType, PhoenixSubscriber, BulkAccountLoader, isOperationPaused, @@ -49,6 +48,7 @@ import { getAccountFromId, getRawAccountFromId, getOpenbookSubscriber, + selectMostRecentBySlot, } from './utils/utils'; import FEATURE_FLAGS from './utils/featureFlags'; import { getDLOBProviderFromOrderSubscriber } from './dlobProvider'; @@ -84,10 +84,8 @@ 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) || 1000; const useWebsocket = process.env.USE_WEBSOCKET?.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]'; @@ -299,107 +297,27 @@ 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(); - if (useRedis) { - 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, - }); - } + logger.info('Connecting to redis'); + for (let i = 0; i < REDIS_CLIENTS.length; i++) { + redisClients.push(new RedisClient({ prefix: REDIS_CLIENTS[i] })); } + 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 userMapClient = new RedisClient({ host: process.env.ELASTICACHE_USERMAP_HOST ?? process.env.ELASTICACHE_HOST, port: process.env.ELASTICACHE_USERMAP_PORT ?? process.env.ELASTICACHE_PORT, prefix: RedisClientPrefix.USER_MAP, }); - 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; - } - } - } - logger.info(`Initializing all market subscribers...`); const initAllMarketSubscribersStart = Date.now(); MARKET_SUBSCRIBERS = await initializeAllMarketSubscribers(driftClient); @@ -555,27 +473,22 @@ const main = async (): Promise => { } let topMakers: string[]; - if (useRedis) { - const redisClient = isVariant(normedMarketType, 'perp') - ? perpMarketRedisMap.get(normedMarketIndex).client - : spotMarketRedisMap.get(normedMarketIndex).client; - const redisResponse = await redisClient.getRaw( - `last_update_orderbook_best_makers_${getVariant( - normedMarketType - )}_${marketIndex}` - ); - if (redisResponse) { - const parsedResponse = JSON.parse(redisResponse); - if ( - parsedResponse && - dlobProvider.getSlot() - parsedResponse.slot < - SLOT_STALENESS_TOLERANCE - ) { - if (side === 'bid') { - topMakers = parsedResponse.bids; - } else { - topMakers = parsedResponse.asks; - } + + const redisResponse = await fetchFromRedis( + `last_update_orderbook_best_makers_${getVariant( + normedMarketType + )}_${marketIndex}`, + selectMostRecentBySlot + ); + if (redisResponse) { + if ( + dlobProvider.getSlot() - redisResponse['slot'] < + SLOT_STALENESS_TOLERANCE + ) { + if (side === 'bid') { + topMakers = redisResponse['bids']; + } else { + topMakers = redisResponse['asks']; } } } @@ -729,54 +642,38 @@ const main = async (): Promise => { const adjustedDepth = depth ?? '100'; let l2Formatted: any; - if (useRedis) { - if (!isSpot && `${includeVamm}`?.toLowerCase() === 'true') { - const redisClient = perpMarketRedisMap.get(normedMarketIndex).client; - const redisL2 = await redisClient.get( - `last_update_orderbook_perp_${normedMarketIndex}` - ); - const depth = Math.min(parseInt(adjustedDepth as string) ?? 1, 100); - redisL2['bids'] = redisL2['bids']?.slice(0, depth); - redisL2['asks'] = redisL2['asks']?.slice(0, depth); + if (!isSpot && `${includeVamm}`?.toLowerCase() === 'true') { + const redisL2 = await fetchFromRedis( + `last_update_orderbook_perp_${normedMarketIndex}`, + selectMostRecentBySlot + ); + const depth = Math.min(parseInt(adjustedDepth as string) ?? 1, 100); + redisL2['bids'] = redisL2['bids']?.slice(0, depth); + redisL2['asks'] = redisL2['asks']?.slice(0, depth); - if ( - redisL2 && - dlobProvider.getSlot() - parseInt(redisL2['slot']) < - SLOT_STALENESS_TOLERANCE - ) { - l2Formatted = JSON.stringify(redisL2); - } else { - if (redisL2 && redisClients.length > 1) { - if (canRotate(normedMarketType, normedMarketIndex)) { - rotateClient(normedMarketType, normedMarketIndex); - } - } - } - } else if ( - isSpot && - `${includePhoenix}`?.toLowerCase() === 'true' && - `${includeOpenbook}`?.toLowerCase() === 'true' + if ( + redisL2 && + dlobProvider.getSlot() - redisL2['slot'] < SLOT_STALENESS_TOLERANCE ) { - const redisClient = spotMarketRedisMap.get(normedMarketIndex).client; - const redisL2 = await redisClient.get( - `last_update_orderbook_spot_${normedMarketIndex}` - ); - const depth = Math.min(parseInt(adjustedDepth as string) ?? 1, 100); - redisL2['bids'] = redisL2['bids']?.slice(0, depth); - redisL2['asks'] = redisL2['asks']?.slice(0, depth); - if ( - redisL2 && - dlobProvider.getSlot() - parseInt(redisL2['slot']) < - SLOT_STALENESS_TOLERANCE - ) { - l2Formatted = JSON.stringify(redisL2); - } else { - if (redisL2 && redisClients.length > 1) { - if (canRotate(normedMarketType, normedMarketIndex)) { - rotateClient(normedMarketType, normedMarketIndex); - } - } - } + l2Formatted = JSON.stringify(redisL2); + } + } else if ( + isSpot && + `${includePhoenix}`?.toLowerCase() === 'true' && + `${includeOpenbook}`?.toLowerCase() === 'true' + ) { + const redisL2 = await fetchFromRedis( + `last_update_orderbook_spot_${normedMarketIndex}`, + selectMostRecentBySlot + ); + const depth = Math.min(parseInt(adjustedDepth as string) ?? 1, 100); + redisL2['bids'] = redisL2['bids']?.slice(0, depth); + redisL2['asks'] = redisL2['asks']?.slice(0, depth); + if ( + redisL2 && + dlobProvider.getSlot() - redisL2['slot'] < SLOT_STALENESS_TOLERANCE + ) { + l2Formatted = JSON.stringify(redisL2); } if (l2Formatted) { @@ -891,65 +788,40 @@ const main = async (): Promise => { const adjustedDepth = normedParam['depth'] ?? '100'; let l2Formatted: any; - if (useRedis) { - if ( - !isSpot && - normedParam['includeVamm']?.toLowerCase() === 'true' - ) { - const redisClient = - perpMarketRedisMap.get(normedMarketIndex).client; - const redisL2 = await redisClient.get( - `last_update_orderbook_perp_${normedMarketIndex}` - ); - const depth = Math.min( - parseInt(adjustedDepth as string) ?? 1, - 100 - ); - redisL2['bids'] = redisL2['bids']?.slice(0, depth); - redisL2['asks'] = redisL2['asks']?.slice(0, depth); - if (redisL2) { - if ( - dlobProvider.getSlot() - parseInt(redisL2['slot']) < - SLOT_STALENESS_TOLERANCE - ) { - l2Formatted = redisL2; - } else { - if (redisClients.length > 1) { - if (canRotate(normedMarketType, normedMarketIndex)) { - rotateClient(normedMarketType, normedMarketIndex); - } - } - } + if (!isSpot && normedParam['includeVamm']?.toLowerCase() === 'true') { + const redisL2 = await fetchFromRedis( + `last_update_orderbook_perp_${normedMarketIndex}`, + selectMostRecentBySlot + ); + const depth = Math.min(parseInt(adjustedDepth as string) ?? 1, 100); + redisL2['bids'] = redisL2['bids']?.slice(0, depth); + redisL2['asks'] = redisL2['asks']?.slice(0, depth); + if (redisL2) { + if ( + dlobProvider.getSlot() - redisL2['slot'] < + SLOT_STALENESS_TOLERANCE + ) { + l2Formatted = redisL2; } - } else if ( - isSpot && - normedParam['includePhoenix']?.toLowerCase() === 'true' && - normedParam['includeOpenbook']?.toLowerCase() === 'true' - ) { - const redisClient = - spotMarketRedisMap.get(normedMarketIndex).client; - const redisL2 = await redisClient.get( - `last_update_orderbook_spot_${normedMarketIndex}` - ); - const depth = Math.min( - parseInt(adjustedDepth as string) ?? 1, - 100 - ); - redisL2['bids'] = redisL2['bids']?.slice(0, depth); - redisL2['asks'] = redisL2['asks']?.slice(0, depth); - if (redisL2) { - if ( - dlobProvider.getSlot() - parseInt(redisL2['slot']) < - SLOT_STALENESS_TOLERANCE - ) { - l2Formatted = redisL2; - } else { - if (redisClients.length > 1) { - if (canRotate(normedMarketType, normedMarketIndex)) { - rotateClient(normedMarketType, normedMarketIndex); - } - } - } + } + } else if ( + isSpot && + normedParam['includePhoenix']?.toLowerCase() === 'true' && + normedParam['includeOpenbook']?.toLowerCase() === 'true' + ) { + const redisL2 = await fetchFromRedis( + `last_update_orderbook_spot_${normedMarketIndex}`, + selectMostRecentBySlot + ); + const depth = Math.min(parseInt(adjustedDepth as string) ?? 1, 100); + redisL2['bids'] = redisL2['bids']?.slice(0, depth); + redisL2['asks'] = redisL2['asks']?.slice(0, depth); + if (redisL2) { + if ( + dlobProvider.getSlot() - redisL2['slot'] < + SLOT_STALENESS_TOLERANCE + ) { + l2Formatted = redisL2; } } @@ -1033,31 +905,27 @@ const main = async (): Promise => { } const marketTypeStr = getVariant(normedMarketType); - if (useRedis) { - const redisClient = ( - marketTypeStr === 'spot' ? spotMarketRedisMap : perpMarketRedisMap - ).get(normedMarketIndex).client; - const redisL3 = await redisClient.getRaw( - `last_update_orderbook_l3_${marketTypeStr}_${normedMarketIndex}` - ); - if ( - redisL3 && - dlobProvider.getSlot() - parseInt(JSON.parse(redisL3).slot) < - SLOT_STALENESS_TOLERANCE - ) { - cacheHitCounter.add(1, { - miss: false, - path: req.baseUrl + req.path, - }); - res.writeHead(200); - res.end(redisL3); - return; - } else { - cacheHitCounter.add(1, { - miss: true, - path: req.baseUrl + req.path, - }); - } + + const redisL3 = await fetchFromRedis( + `last_update_orderbook_l3_${marketTypeStr}_${normedMarketIndex}`, + selectMostRecentBySlot + ); + if ( + redisL3 && + dlobProvider.getSlot() - redisL3['slot'] < SLOT_STALENESS_TOLERANCE + ) { + cacheHitCounter.add(1, { + miss: false, + path: req.baseUrl + req.path, + }); + res.writeHead(200); + res.end(JSON.stringify(redisL3)); + return; + } else { + cacheHitCounter.add(1, { + miss: true, + path: req.baseUrl + req.path, + }); } const l3 = dlobSubscriber.getL3({ diff --git a/src/publishers/dlobPublisher.ts b/src/publishers/dlobPublisher.ts index 12c3a62..05f86a7 100644 --- a/src/publishers/dlobPublisher.ts +++ b/src/publishers/dlobPublisher.ts @@ -20,7 +20,6 @@ import { PhoenixSubscriber, MarketType, OraclePriceData, - OrderSubscriberConfig, } from '@drift-labs/sdk'; import { RedisClient, RedisClientPrefix } from '@drift/common/clients'; diff --git a/src/publishers/tradesPublisher.ts b/src/publishers/tradesPublisher.ts index 3f75af3..1427279 100644 --- a/src/publishers/tradesPublisher.ts +++ b/src/publishers/tradesPublisher.ts @@ -79,7 +79,9 @@ const main = async () => { const redisClient = new RedisClient({ prefix: redisClientPrefix }); await redisClient.connect(); - const slotSubscriber = new SlotSubscriber(connection, {}); + const slotSubscriber = new SlotSubscriber(connection, { + resubTimeoutMs: 10_000, + }); const lamportsBalance = await connection.getBalance(wallet.publicKey); logger.info( diff --git a/src/utils/utils.ts b/src/utils/utils.ts index 82fb44d..2d9dd5e 100644 --- a/src/utils/utils.ts +++ b/src/utils/utils.ts @@ -472,3 +472,18 @@ export type SubscriberLookup = { openbook?: OpenbookV2Subscriber; }; }; + +export const selectMostRecentBySlot = (responses: any[]): any => { + const parsedResponses = responses + .map((response) => { + try { + return JSON.parse(response); + } catch { + return null; + } + }) + .filter((parsed) => parsed && typeof parsed.slot === 'number'); + return parsedResponses.reduce((mostRecent, current) => { + return !mostRecent || current.slot > mostRecent.slot ? current : mostRecent; + }, null); +}; diff --git a/src/wsConnectionManager.ts b/src/wsConnectionManager.ts index bc7c3ef..7563c50 100644 --- a/src/wsConnectionManager.ts +++ b/src/wsConnectionManager.ts @@ -3,8 +3,8 @@ import express from 'express'; import * as http from 'http'; import compression from 'compression'; import { WebSocket, WebSocketServer } from 'ws'; -import { sleep } from './utils/utils'; -import { register, Gauge } from 'prom-client'; +import { sleep, selectMostRecentBySlot } from './utils/utils'; +import { register, Gauge, Counter } from 'prom-client'; import { DriftEnv, PerpMarkets, SpotMarkets } from '@drift-labs/sdk'; import { RedisClient, RedisClientPrefix } from '@drift/common/clients'; @@ -21,6 +21,16 @@ const wsConnectionsGauge = new Gauge({ name: 'websocket_connections', help: 'Number of active WebSocket connections', }); +const wsOrderbookSourceCounter = new Counter({ + name: 'websocket_orderbook_source', + help: 'Number of orderbook messages sent from source', + labelNames: ['source'], +}); +const wsOrderbookSourceLastSlotGauge = new Gauge({ + name: 'websocket_orderbook_source_last_slot', + help: 'Last slot of orderbook messages from a source', + labelNames: ['source'], +}); const server = http.createServer(app); const wss = new WebSocketServer({ @@ -34,15 +44,27 @@ const WS_PORT = process.env.WS_PORT || '3000'; console.log(`WS LISTENER PORT : ${WS_PORT}`); const MAX_BUFFERED_AMOUNT = 300000; -const REDIS_CLIENT = process.env.REDIS_CLIENT || 'DLOB'; -const CHANNEL_PREFIX = RedisClientPrefix[REDIS_CLIENT]; -console.log('Redis Clients:', REDIS_CLIENT); -const CHANNEL_PREFIX_HELIUS = RedisClientPrefix.DLOB_HELIUS; +const envClients = []; +const clients = process.env.REDIS_CLIENT?.trim() + .replace(/^\[|\]$/g, '') + .split(/\s*,\s*/); + +clients?.forEach((client) => envClients.push(RedisClientPrefix[client])); + +const REDIS_CLIENTS = envClients.length + ? envClients + : [RedisClientPrefix.DLOB, RedisClientPrefix.DLOB_HELIUS]; +console.log('Redis Clients:', REDIS_CLIENTS); + +const regexp = new RegExp(REDIS_CLIENTS.join('|'), 'g'); const sanitiseChannelForClient = (channel: string | undefined): string => { - return channel - ?.replace(CHANNEL_PREFIX, '') - ?.replace(CHANNEL_PREFIX_HELIUS, ''); + return channel?.replace(regexp, ''); +}; +const getChannelPrefix = (channel: string | undefined): string | undefined => { + if (!channel) return undefined; + const match = channel.match(regexp); + return match?.[0]; }; const getRedisChannelFromMessage = (message: any): string => { @@ -70,11 +92,11 @@ const getRedisChannelFromMessage = (message: any): string => { switch (channel.toLowerCase()) { case 'trades': - return `${CHANNEL_PREFIX}trades_${marketType}_${marketIndex}`; + return `trades_${marketType}_${marketIndex}`; case 'orderbook': - return `${CHANNEL_PREFIX}orderbook_${marketType}_${marketIndex}`; + return `orderbook_${marketType}_${marketIndex}`; case 'priorityfees': - return `${CHANNEL_PREFIX_HELIUS}priorityFees_${marketType}_${marketIndex}`; + return `priorityFees_${marketType}_${marketIndex}`; case undefined: default: throw new Error('Bad channel specified'); @@ -82,47 +104,71 @@ const getRedisChannelFromMessage = (message: any): string => { }; async function main() { - const redisClient = new RedisClient({}); - const lastMessageRetriever = new RedisClient({ prefix: CHANNEL_PREFIX }); + const subscribedChannelToSlot: Map = new Map(); - await redisClient.connect(); - await lastMessageRetriever.connect(); + const redisClients: Array = []; + for (let i = 0; i < REDIS_CLIENTS.length; i++) { + const redisClient = new RedisClient({ prefix: REDIS_CLIENTS[i] }); + redisClients.push(redisClient); + + redisClient.forceGetClient().on('connect', () => { + subscribedChannels.forEach(async (channel) => { + try { + await redisClient.subscribe(channel); + } catch (error) { + console.error(`Error subscribing to ${channel}:`, error); + } + }); + }); + + redisClient.forceGetClient().on('message', (subscribedChannel, message) => { + const sanitizedChannel = sanitiseChannelForClient(subscribedChannel); + const channelPrefix = getChannelPrefix(sanitizedChannel); + const subscribers = channelSubscribers.get(sanitizedChannel); + if (subscribers) { + if (sanitizedChannel.includes('orderbook')) { + const messageSlot = JSON.parse(message)['slot']; + wsOrderbookSourceLastSlotGauge.set( + { + source: channelPrefix, + }, + messageSlot + ); + + const lastMessageSlot = subscribedChannelToSlot.get(sanitizedChannel); + if (!lastMessageSlot || lastMessageSlot <= messageSlot) { + subscribedChannelToSlot.set(sanitizedChannel, messageSlot); + } else if (lastMessageSlot > messageSlot) { + return; + } + } + subscribers.forEach((ws) => { + if ( + ws.readyState === WebSocket.OPEN && + ws.bufferedAmount < MAX_BUFFERED_AMOUNT + ) { + wsOrderbookSourceCounter.inc({ + source: channelPrefix, + }); + ws.send( + JSON.stringify({ + channel: sanitizedChannel, + data: message, + }) + ); + } + }); + } + }); + + redisClient.forceGetClient().on('error', (error) => { + console.error('Redis client error:', error); + }); + } const channelSubscribers = new Map>(); const subscribedChannels = new Set(); - redisClient.forceGetClient().on('connect', () => { - subscribedChannels.forEach(async (channel) => { - try { - await redisClient.subscribe(channel); - } catch (error) { - console.error(`Error subscribing to ${channel}:`, error); - } - }); - }); - - redisClient.forceGetClient().on('message', (subscribedChannel, message) => { - const subscribers = channelSubscribers.get(subscribedChannel); - if (subscribers) { - subscribers.forEach((ws) => { - if ( - ws.readyState === WebSocket.OPEN && - ws.bufferedAmount < MAX_BUFFERED_AMOUNT - ) - ws.send( - JSON.stringify({ - channel: sanitiseChannelForClient(subscribedChannel), - data: message, - }) - ); - }); - } - }); - - redisClient.forceGetClient().on('error', (error) => { - console.error('Redis client error:', error); - }); - wss.on('connection', (ws: WebSocket) => { console.log('Client connected'); wsConnectionsGauge.inc(); @@ -166,8 +212,12 @@ async function main() { if (!subscribedChannels.has(redisChannel)) { console.log('Trying to subscribe to channel', redisChannel); - redisClient - .subscribe(redisChannel) + redisClients[0] + .subscribe( + redisChannel.includes('priorityFees') + ? `dlob-helius:${redisChannel}` + : `${redisClients[0].getPrefix()}${redisChannel}` + ) .then(() => { subscribedChannels.add(redisChannel); }) @@ -179,6 +229,21 @@ async function main() { ); return; }); + + if (redisChannel.includes('orderbook') && redisClients.length > 1) { + redisClients.slice(1).map((redisClient) => { + redisClient + .subscribe(`${redisClient.getPrefix()}${redisChannel}`) + .catch(() => { + ws.send( + JSON.stringify({ + error: `Error subscribing to channel: ${parsedMessage}`, + }) + ); + return; + }); + }); + } } if (!channelSubscribers.get(redisChannel)) { @@ -194,17 +259,21 @@ async function main() { ); // Fetch and send last message if (redisChannel.includes('orderbook')) { - const lastUpdateChannel = `last_update_${redisChannel}`.replace( - CHANNEL_PREFIX, - '' - ); - const lastMessage = await lastMessageRetriever.getRaw( - lastUpdateChannel + const lastMessages = await Promise.all( + redisClients.map((redisClient) => + redisClient.getRaw( + `last_update_${redisChannel}`.replace( + redisClient.getPrefix(), + '' + ) + ) + ) ); + const lastMessage = selectMostRecentBySlot(lastMessages); if (lastMessage) { ws.send( JSON.stringify({ - channel: lastUpdateChannel, + channel: redisChannel, data: lastMessage, }) ); @@ -278,7 +347,7 @@ async function main() { clearInterval(bufferInterval); channelSubscribers.forEach((subscribers, channel) => { if (subscribers.delete(ws) && subscribers.size === 0) { - redisClient.unsubscribe(channel); + redisClients.map((redisClient) => redisClient.unsubscribe(channel)); channelSubscribers.delete(channel); subscribedChannels.delete(channel); } diff --git a/yarn.lock b/yarn.lock index f8475ad..d7a4a84 100644 --- a/yarn.lock +++ b/yarn.lock @@ -287,13 +287,6 @@ dependencies: regenerator-runtime "^0.13.11" -"@babel/runtime@^7.23.2": - version "7.23.5" - resolved "https://registry.yarnpkg.com/@babel/runtime/-/runtime-7.23.5.tgz#11edb98f8aeec529b82b211028177679144242db" - integrity sha512-NdUTHcPe4C99WxPub+K9l9tK5/lV4UXIoaHSYgzco9BCyjKAAwzdBI+wWtYqHt7LJdbo74ZjRPJgzVweq1sz0w== - dependencies: - regenerator-runtime "^0.14.0" - "@babel/runtime@^7.24.6", "@babel/runtime@^7.24.8": version "7.24.8" resolved "https://registry.yarnpkg.com/@babel/runtime/-/runtime-7.24.8.tgz#5d958c3827b13cc6d05e038c07fb2e5e3420d82e" @@ -460,7 +453,7 @@ kuler "^2.0.0" "@drift-labs/sdk@file:./drift-common/protocol/sdk", "@drift-labs/sdk@file:drift-common/protocol/sdk": - version "2.103.0-beta.8" + version "2.103.0-beta.9" dependencies: "@coral-xyz/anchor" "0.28.0" "@coral-xyz/anchor-30" "npm:@coral-xyz/anchor@0.30.1" @@ -487,7 +480,7 @@ "@drift/common@file:./drift-common/common-ts": version "1.0.0" dependencies: - "@drift-labs/sdk" "file:../../Library/Caches/Yarn/v6/npm-@drift-common-1.0.0-2478b7b0-73c2-48c2-8cdc-077c34db7294-1733168269147/node_modules/@drift/protocol/sdk" + "@drift-labs/sdk" "file:../../Library/Caches/Yarn/v6/npm-@drift-common-1.0.0-736eee55-37f6-4ebd-9a13-95d148c9fb80-1733261412306/node_modules/@drift/protocol/sdk" "@jest/globals" "^29.3.1" "@slack/web-api" "^6.4.0" "@solana/spl-token" "^0.3.8" @@ -891,13 +884,6 @@ dependencies: "@noble/hashes" "1.4.0" -"@noble/curves@^1.2.0": - version "1.2.0" - resolved "https://registry.yarnpkg.com/@noble/curves/-/curves-1.2.0.tgz#92d7e12e4e49b23105a2555c6984d41733d65c35" - integrity sha512-oYclrNgRaM9SsBUBVbb8M6DTV7ZHRTKugureoYEncY5c65HOmRzvSiTE3y5CYaPYJA/GVkrhXEoF0M3Ya9PMnw== - dependencies: - "@noble/hashes" "1.3.2" - "@noble/curves@^1.4.0": version "1.4.0" resolved "https://registry.yarnpkg.com/@noble/curves/-/curves-1.4.0.tgz#f05771ef64da724997f69ee1261b2417a49522d6" @@ -915,11 +901,6 @@ resolved "https://registry.yarnpkg.com/@noble/ed25519/-/ed25519-1.7.3.tgz#57e1677bf6885354b466c38e2b620c62f45a7123" integrity sha512-iR8GBkDt0Q3GyaVcIu7mSsVIqnFbkbRzGLWlvhwunacoLwt4J3swfKhfaM6rN6WY+TBGoYT1GtT1mIh2/jGbRQ== -"@noble/hashes@1.3.2", "@noble/hashes@^1.3.1": - version "1.3.2" - resolved "https://registry.yarnpkg.com/@noble/hashes/-/hashes-1.3.2.tgz#6f26dbc8fbc7205873ce3cee2f690eba0d421b39" - integrity sha512-MVC8EAQp7MvEcm30KWENFjgR+Mkmf+D189XJTkFIlwohU5hcBbn1ZkKq7KVTi2Hme3PMGF390DaL52beVrIihQ== - "@noble/hashes@1.4.0", "@noble/hashes@^1.3.0", "@noble/hashes@^1.4.0": version "1.4.0" resolved "https://registry.yarnpkg.com/@noble/hashes/-/hashes-1.4.0.tgz#45814aa329f30e4fe0ba49426f49dfccdd066426" @@ -930,6 +911,11 @@ resolved "https://registry.yarnpkg.com/@noble/hashes/-/hashes-1.1.3.tgz#360afc77610e0a61f3417e497dcf36862e4f8111" integrity sha512-CE0FCR57H2acVI5UOzIGSSIYxZ6v/HOhDR0Ro9VLyhnzLwx0o8W1mmgaqlEUx4049qJDlIBRztv5k+MM8vbO3A== +"@noble/hashes@^1.3.1": + version "1.3.2" + resolved "https://registry.yarnpkg.com/@noble/hashes/-/hashes-1.3.2.tgz#6f26dbc8fbc7205873ce3cee2f690eba0d421b39" + integrity sha512-MVC8EAQp7MvEcm30KWENFjgR+Mkmf+D189XJTkFIlwohU5hcBbn1ZkKq7KVTi2Hme3PMGF390DaL52beVrIihQ== + "@noble/secp256k1@^1.6.3": version "1.7.0" resolved "https://registry.yarnpkg.com/@noble/secp256k1/-/secp256k1-1.7.0.tgz#d15357f7c227e751d90aa06b05a0e5cf993ba8c1" @@ -1956,6 +1942,27 @@ rpc-websockets "^8.0.1" superstruct "^1.0.4" +"@solana/web3.js@1.95.2", "@solana/web3.js@^1.54.0", "@solana/web3.js@^1.77.3", "@solana/web3.js@^1.93.0", "@solana/web3.js@^1.95.0": + version "1.95.2" + resolved "https://registry.yarnpkg.com/@solana/web3.js/-/web3.js-1.95.2.tgz#6f8a0362fa75886a21550dbec49aad54481463a6" + integrity sha512-SjlHp0G4qhuhkQQc+YXdGkI8EerCqwxvgytMgBpzMUQTafrkNant3e7pgilBGgjy/iM40ICvWBLgASTPMrQU7w== + dependencies: + "@babel/runtime" "^7.24.8" + "@noble/curves" "^1.4.2" + "@noble/hashes" "^1.4.0" + "@solana/buffer-layout" "^4.0.1" + agentkeepalive "^4.5.0" + bigint-buffer "^1.1.5" + bn.js "^5.2.1" + borsh "^0.7.0" + bs58 "^4.0.1" + buffer "6.0.3" + fast-stable-stringify "^1.0.0" + jayson "^4.1.1" + node-fetch "^2.7.0" + rpc-websockets "^9.0.2" + superstruct "^2.0.2" + "@solana/web3.js@^1.17.0", "@solana/web3.js@^1.21.0", "@solana/web3.js@^1.30.2": version "1.67.0" resolved "https://registry.yarnpkg.com/@solana/web3.js/-/web3.js-1.67.0.tgz#bd67742cd9c176889fd1954080173dd7d0c224a2" @@ -1999,48 +2006,6 @@ rpc-websockets "^7.5.1" superstruct "^0.14.2" -"@solana/web3.js@^1.54.0", "@solana/web3.js@^1.77.3", "@solana/web3.js@^1.93.0", "@solana/web3.js@^1.95.0": - version "1.95.2" - resolved "https://registry.yarnpkg.com/@solana/web3.js/-/web3.js-1.95.2.tgz#6f8a0362fa75886a21550dbec49aad54481463a6" - integrity sha512-SjlHp0G4qhuhkQQc+YXdGkI8EerCqwxvgytMgBpzMUQTafrkNant3e7pgilBGgjy/iM40ICvWBLgASTPMrQU7w== - dependencies: - "@babel/runtime" "^7.24.8" - "@noble/curves" "^1.4.2" - "@noble/hashes" "^1.4.0" - "@solana/buffer-layout" "^4.0.1" - agentkeepalive "^4.5.0" - bigint-buffer "^1.1.5" - bn.js "^5.2.1" - borsh "^0.7.0" - bs58 "^4.0.1" - buffer "6.0.3" - fast-stable-stringify "^1.0.0" - jayson "^4.1.1" - node-fetch "^2.7.0" - rpc-websockets "^9.0.2" - superstruct "^2.0.2" - -"@solana/web3.js@^1.73.3": - version "1.87.6" - resolved "https://registry.yarnpkg.com/@solana/web3.js/-/web3.js-1.87.6.tgz#6744cfc5f4fc81e0f58241c0a92648a7320bb3bf" - integrity sha512-LkqsEBgTZztFiccZZXnawWa8qNCATEqE97/d0vIwjTclmVlc8pBpD1DmjfVHtZ1HS5fZorFlVhXfpwnCNDZfyg== - dependencies: - "@babel/runtime" "^7.23.2" - "@noble/curves" "^1.2.0" - "@noble/hashes" "^1.3.1" - "@solana/buffer-layout" "^4.0.0" - agentkeepalive "^4.3.0" - bigint-buffer "^1.1.5" - bn.js "^5.2.1" - borsh "^0.7.0" - bs58 "^4.0.1" - buffer "6.0.3" - fast-stable-stringify "^1.0.0" - jayson "^4.1.0" - node-fetch "^2.6.12" - rpc-websockets "^7.5.1" - superstruct "^0.14.2" - "@solana/web3.js@^1.90.0": version "1.95.1" resolved "https://registry.yarnpkg.com/@solana/web3.js/-/web3.js-1.95.1.tgz#fcbbaf845309ff7ceb8d3726702799e8c27530e8" @@ -5106,13 +5071,6 @@ node-fetch@2, node-fetch@2.6.7: dependencies: whatwg-url "^5.0.0" -node-fetch@^2.6.12, node-fetch@^2.7.0: - version "2.7.0" - resolved "https://registry.yarnpkg.com/node-fetch/-/node-fetch-2.7.0.tgz#d0f0fa6e3e2dc1d27efcd8ad99d550bda94d187d" - integrity sha512-c4FRfUm/dbcWZ7U+1Wq0AwCyFL+3nt2bEw05wfxSz+DWpWsitgmSgYmy2dQdWyKC1694ELPqMs/YzUSNozLt8A== - dependencies: - whatwg-url "^5.0.0" - node-fetch@^2.6.7: version "2.6.9" resolved "https://registry.yarnpkg.com/node-fetch/-/node-fetch-2.6.9.tgz#7c7f744b5cc6eb5fd404e0c7a9fec630a55657e6" @@ -5120,6 +5078,13 @@ node-fetch@^2.6.7: dependencies: whatwg-url "^5.0.0" +node-fetch@^2.7.0: + version "2.7.0" + resolved "https://registry.yarnpkg.com/node-fetch/-/node-fetch-2.7.0.tgz#d0f0fa6e3e2dc1d27efcd8ad99d550bda94d187d" + integrity sha512-c4FRfUm/dbcWZ7U+1Wq0AwCyFL+3nt2bEw05wfxSz+DWpWsitgmSgYmy2dQdWyKC1694ELPqMs/YzUSNozLt8A== + dependencies: + whatwg-url "^5.0.0" + node-gyp-build@^4.3.0: version "4.5.0" resolved "https://registry.yarnpkg.com/node-gyp-build/-/node-gyp-build-4.5.0.tgz#7a64eefa0b21112f89f58379da128ac177f20e40"