From a8e2d3fc3df837200b74943af87dd620646b00c2 Mon Sep 17 00:00:00 2001 From: GitHub Actions <41898282+github-actions[bot]@users.noreply.github.com> Date: Fri, 8 Dec 2023 21:59:18 +0000 Subject: [PATCH 01/10] Bumping sdk dependency to 2.49.0-beta.18 --- package.json | 2 +- yarn.lock | 8 ++++---- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/package.json b/package.json index 7b7245b..f9244e1 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.49.0-beta.17", + "@drift-labs/sdk": "2.49.0-beta.18", "@opentelemetry/api": "^1.1.0", "@opentelemetry/auto-instrumentations-node": "^0.31.1", "@opentelemetry/exporter-prometheus": "^0.31.0", diff --git a/yarn.lock b/yarn.lock index 6b0e84c..e194578 100644 --- a/yarn.lock +++ b/yarn.lock @@ -115,10 +115,10 @@ enabled "2.0.x" kuler "^2.0.0" -"@drift-labs/sdk@2.49.0-beta.17": - version "2.49.0-beta.17" - resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.49.0-beta.17.tgz#59d39712dcd2d0e0e6905b42c8efb743178277f3" - integrity sha512-0OLJFxDlKxIsQ32MuqVp/oHJFaeCy1IMigLmU92Zlapvjd8euR73VueiF4rGGROwrqO4IIamPgf68cD7D1V2HA== +"@drift-labs/sdk@2.49.0-beta.18": + version "2.49.0-beta.18" + resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.49.0-beta.18.tgz#7505e9d0f8883f0622cfa5cc4b67338d22d3919e" + integrity sha512-aYaJLEyLoLvqg8jY0w6y7nXwckEQYC6L4AIpaIwnHPH7+ApsJ/c3GuYRKWdtmOjHvxlCfdBsAibvhtwF+2MHuA== dependencies: "@coral-xyz/anchor" "0.28.1-beta.2" "@ellipsis-labs/phoenix-sdk" "^1.4.2" From 9abc31a32d8db0871bcc4613272a78a02664cd4f Mon Sep 17 00:00:00 2001 From: GitHub Actions <41898282+github-actions[bot]@users.noreply.github.com> Date: Fri, 8 Dec 2023 22:10:37 +0000 Subject: [PATCH 02/10] Bumping sdk dependency to 2.49.0-beta.19 --- package.json | 2 +- yarn.lock | 8 ++++---- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/package.json b/package.json index f9244e1..d88ab93 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.49.0-beta.18", + "@drift-labs/sdk": "2.49.0-beta.19", "@opentelemetry/api": "^1.1.0", "@opentelemetry/auto-instrumentations-node": "^0.31.1", "@opentelemetry/exporter-prometheus": "^0.31.0", diff --git a/yarn.lock b/yarn.lock index e194578..a6a3897 100644 --- a/yarn.lock +++ b/yarn.lock @@ -115,10 +115,10 @@ enabled "2.0.x" kuler "^2.0.0" -"@drift-labs/sdk@2.49.0-beta.18": - version "2.49.0-beta.18" - resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.49.0-beta.18.tgz#7505e9d0f8883f0622cfa5cc4b67338d22d3919e" - integrity sha512-aYaJLEyLoLvqg8jY0w6y7nXwckEQYC6L4AIpaIwnHPH7+ApsJ/c3GuYRKWdtmOjHvxlCfdBsAibvhtwF+2MHuA== +"@drift-labs/sdk@2.49.0-beta.19": + version "2.49.0-beta.19" + resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.49.0-beta.19.tgz#cc1d22ab9c46e76008e869c266ed6734e892a461" + integrity sha512-MibgKjqarVlXdumSUgcCK8X87zcXX+S6PkxjPzsXR+Rkoc1hwOHtP0ANzL3ZmBaxsqK8cpPwTM1a90fww0BGRw== dependencies: "@coral-xyz/anchor" "0.28.1-beta.2" "@ellipsis-labs/phoenix-sdk" "^1.4.2" From 5493351f40e25d85469b459c2a5572e812145836 Mon Sep 17 00:00:00 2001 From: Nour Alharithi Date: Fri, 8 Dec 2023 17:09:50 -0800 Subject: [PATCH 03/10] fetch from redis --- src/dlob-subscriber/DLOBSubscriberIO.ts | 30 ++- src/index.ts | 263 +++++++++++++++++------- src/publishers/dlobPublisher.ts | 5 +- 3 files changed, 214 insertions(+), 84 deletions(-) diff --git a/src/dlob-subscriber/DLOBSubscriberIO.ts b/src/dlob-subscriber/DLOBSubscriberIO.ts index a7cea28..0a20194 100644 --- a/src/dlob-subscriber/DLOBSubscriberIO.ts +++ b/src/dlob-subscriber/DLOBSubscriberIO.ts @@ -62,8 +62,8 @@ export class DLOBSubscriberIO extends DLOBSubscriber { marketType: MarketType.PERP, marketName: market.symbol, depth: -1, - includeVamm: true, numVammOrders: 100, + includeVamm: true, updateOnChange: true, fallbackL2Generators: [], }); @@ -124,13 +124,39 @@ export class DLOBSubscriberIO extends DLOBSubscriber { l2Args.marketType, l2Args.marketIndex ); + + const l2Formatted_depth100 = Object.assign({}, l2Formatted, { + bids: l2Formatted.bids.slice(0, 100), + asks: l2Formatted.asks.slice(0, 100), + }); + const l2Formatted_depth20 = Object.assign({}, l2Formatted, { + bids: l2Formatted.bids.slice(0, 20), + asks: l2Formatted.asks.slice(0, 20), + }); + const l2Formatted_depth5 = Object.assign({}, l2Formatted, { + bids: l2Formatted.bids.slice(0, 5), + asks: l2Formatted.asks.slice(0, 5), + }); + this.redisClient.client.publish( `orderbook_${marketType}_${l2Args.marketIndex}`, JSON.stringify(l2Formatted) ); this.redisClient.client.set( `last_update_orderbook_${marketType}_${l2Args.marketIndex}`, - JSON.stringify(l2Formatted) + JSON.stringify(l2Formatted_depth100) + ); + this.redisClient.client.set( + `last_update_orderbook_${marketType}_${l2Args.marketIndex}_depth_100`, + JSON.stringify(l2Formatted_depth100) + ); + this.redisClient.client.set( + `last_update_orderbook_${marketType}_${l2Args.marketIndex}_depth_20`, + JSON.stringify(l2Formatted_depth20) + ); + this.redisClient.client.set( + `last_update_orderbook_${marketType}_${l2Args.marketIndex}_depth_5`, + JSON.stringify(l2Formatted_depth5) ); } } diff --git a/src/index.ts b/src/index.ts index a584c0a..e364c7d 100644 --- a/src/index.ts +++ b/src/index.ts @@ -28,8 +28,6 @@ import { initialize, isVariant, OrderSubscriber, - UserAccount, - Order, } from '@drift-labs/sdk'; import { logger, setLogLevel } from './utils/logger'; @@ -59,8 +57,14 @@ import { getDLOBProviderFromOrderSubscriber, getDLOBProviderFromUserMap, } from './dlobProvider'; +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; + const driftEnv = (process.env.ENV || 'devnet') as DriftEnv; const commitHash = process.env.COMMIT; //@ts-ignore @@ -343,6 +347,10 @@ const main = async () => { `DLOBSubscriber initialized in ${Date.now() - initDlobSubscriberStart} ms` ); + logger.info('Connecting to redis'); + const redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD); + await redisClient.connect(); + logger.info(`Initializing all market subscribers...`); const initAllMarketSubscribersStart = Date.now(); MARKET_SUBSCRIBERS = await initializeAllMarketSubscribers(driftClient); @@ -742,6 +750,55 @@ const main = async () => { adjustedDepth = '-1'; } + let l2Formatted: any; + if ( + !isSpot && + `${includeVamm}`.toLowerCase() === 'true' && + `${includeOracle}`.toLowerCase().toLowerCase() === 'true' && + !grouping + ) { + if (parseInt(adjustedDepth as string) === 5) { + l2Formatted = await redisClient.client.get( + `last_update_orderbook_perp_${normedMarketIndex}_depth_5` + ); + } else if (parseInt(adjustedDepth as string) === 20) { + l2Formatted = await redisClient.client.get( + `last_update_orderbook_perp_${normedMarketIndex}_depth_20` + ); + } else if (parseInt(adjustedDepth as string) === 100) { + l2Formatted = await redisClient.client.get( + `last_update_orderbook_perp_${normedMarketIndex}_depth_100` + ); + } + } else if ( + isSpot && + `${includeSerum}`.toLowerCase() === 'true' && + `${includePhoenix}`.toLowerCase() === 'true' && + `${includeOracle}`.toLowerCase() === 'true' && + !grouping + ) { + if (parseInt(adjustedDepth as string) === 5) { + l2Formatted = await redisClient.client.get( + `last_update_orderbook_spot_${normedMarketIndex}_depth_5` + ); + } else if (parseInt(adjustedDepth as string) === 20) { + l2Formatted = await redisClient.client.get( + `last_update_orderbook_spot_${normedMarketIndex}_depth_20` + ); + } else if (parseInt(adjustedDepth as string) === 100) { + console.log('100'); + l2Formatted = await redisClient.client.get( + `last_update_orderbook_spot_${normedMarketIndex}_depth_100` + ); + } + } + + if (l2Formatted) { + res.writeHead(200); + res.end(JSON.stringify(l2Formatted)); + return; + } + const l2 = dlobSubscriber.getL2({ marketIndex: normedMarketIndex, marketType: normedMarketType, @@ -783,7 +840,7 @@ const main = async () => { res.end(JSON.stringify(l2Formatted)); } else { // make the BNs into strings - const l2Formatted = l2WithBNToStrings(l2); + l2Formatted = l2WithBNToStrings(l2); if (`${includeOracle}`.toLowerCase() === 'true') { addOracletoResponse( l2Formatted, @@ -792,10 +849,10 @@ const main = async () => { normedMarketIndex ); } - - res.writeHead(200); - res.end(JSON.stringify(l2Formatted)); } + + res.writeHead(200); + res.end(JSON.stringify(l2Formatted)); } catch (err) { next(err); } @@ -836,86 +893,132 @@ const main = async () => { return; } - const l2s = normedParams.map((normedParam) => { - const { normedMarketType, normedMarketIndex, error } = - validateDlobQuery( - driftClient, - driftEnv, - normedParam['marketType'] as string, - normedParam['marketIndex'] as string, - normedParam['marketName'] as string - ); - if (error) { - res.status(400).send(`Bad Request: ${error}`); - return; - } - - const isSpot = isVariant(normedMarketType, 'spot'); - - let adjustedDepth = normedParam['depth'] ?? '10'; - if (normedParam['grouping'] !== undefined) { - // If grouping is also supplied, we want the entire book depth. - // we will apply depth after grouping - adjustedDepth = '-1'; - } - - const l2 = dlobSubscriber.getL2({ - marketIndex: normedMarketIndex, - marketType: normedMarketType, - depth: parseInt(adjustedDepth as string), - includeVamm: isSpot - ? false - : `${normedParam['includeVamm']}`.toLowerCase() === 'true', - fallbackL2Generators: isSpot - ? [ - `${normedParam['includePhoenix']}`.toLowerCase() === 'true' && - MARKET_SUBSCRIBERS[normedMarketIndex].phoenix, - `${normedParam['includeSerum']}`.toLowerCase() === 'true' && - MARKET_SUBSCRIBERS[normedMarketIndex].serum, - ].filter((a) => !!a) - : [], - }); - - if (normedParam['grouping']) { - const finalDepth = normedParam['depth'] - ? parseInt(normedParam['depth'] as string) - : 10; - if (isNaN(parseInt(normedParam['grouping'] as string))) { - res - .status(400) - .send('Bad Request: grouping must be a number if supplied'); + const l2s = await Promise.all( + normedParams.map(async (normedParam) => { + const { normedMarketType, normedMarketIndex, error } = + validateDlobQuery( + driftClient, + driftEnv, + normedParam['marketType'] as string, + normedParam['marketIndex'] as string, + normedParam['marketName'] as string + ); + if (error) { + res.status(400).send(`Bad Request: ${error}`); return; } - const groupingBN = new BN( - parseInt(normedParam['grouping'] as string) - ); - const l2Formatted = l2WithBNToStrings( - groupL2(l2, groupingBN, finalDepth) - ); - if (`${normedParam['includeOracle']}`.toLowerCase() === 'true') { - addOracletoResponse( - l2Formatted, - driftClient, - normedMarketType, - normedMarketIndex + const isSpot = isVariant(normedMarketType, 'spot'); + + let adjustedDepth = normedParam['depth'] ?? '10'; + if (normedParam['grouping'] !== undefined) { + // If grouping is also supplied, we want the entire book depth. + // we will apply depth after grouping + adjustedDepth = '-1'; + } + + let l2Formatted: any; + if ( + marketType === 'perp' && + normedParam['includeVamm'].toLowerCase() === 'true' && + normedParam['includeOracle'].toLowerCase() === 'true' && + !normedParam['grouping'] + ) { + if (parseInt(adjustedDepth as string) === 5) { + l2Formatted = await redisClient.client.get( + `last_update_orderbook_perp_${normedMarketIndex}_depth_5` + ); + } else if (parseInt(adjustedDepth as string) === 20) { + l2Formatted = await redisClient.client.get( + `last_update_orderbook_perp_${normedMarketIndex}_depth_20` + ); + } else if (parseInt(adjustedDepth as string) === 100) { + l2Formatted = await redisClient.client.get( + `last_update_orderbook_perp_${normedMarketIndex}_depth_100` + ); + } + } else if ( + marketType === 'spot' && + normedParam['includePhoenix'].toLowerCase() === 'true' && + normedParam['includeSerum'].toLowerCase() === 'true' && + !normedParam['grouping'] + ) { + if (parseInt(adjustedDepth as string) === 5) { + l2Formatted = await redisClient.client.get( + `last_update_orderbook_spot_${normedMarketIndex}_depth_5` + ); + } else if (parseInt(adjustedDepth as string) === 20) { + l2Formatted = await redisClient.client.get( + `last_update_orderbook_spot_${normedMarketIndex}_depth_20` + ); + } else if (parseInt(adjustedDepth as string) === 100) { + l2Formatted = await redisClient.client.get( + `last_update_orderbook_spot_${normedMarketIndex}_depth_100` + ); + } + } + + if (l2Formatted) { + return l2Formatted; + } + + const l2 = dlobSubscriber.getL2({ + marketIndex: normedMarketIndex, + marketType: normedMarketType, + depth: parseInt(adjustedDepth as string), + includeVamm: isSpot + ? false + : `${normedParam['includeVamm']}`.toLowerCase() === 'true', + fallbackL2Generators: isSpot + ? [ + `${normedParam['includePhoenix']}`.toLowerCase() === 'true' && + MARKET_SUBSCRIBERS[normedMarketIndex].phoenix, + `${normedParam['includeSerum']}`.toLowerCase() === 'true' && + MARKET_SUBSCRIBERS[normedMarketIndex].serum, + ].filter((a) => !!a) + : [], + }); + + if (normedParam['grouping']) { + const finalDepth = normedParam['depth'] + ? parseInt(normedParam['depth'] as string) + : 10; + if (isNaN(parseInt(normedParam['grouping'] as string))) { + res + .status(400) + .send('Bad Request: grouping must be a number if supplied'); + return; + } + const groupingBN = new BN( + parseInt(normedParam['grouping'] as string) ); + + l2Formatted = l2WithBNToStrings( + groupL2(l2, groupingBN, finalDepth) + ); + if (`${normedParam['includeOracle']}`.toLowerCase() === 'true') { + addOracletoResponse( + l2Formatted, + driftClient, + normedMarketType, + normedMarketIndex + ); + } + } else { + // make the BNs into strings + l2Formatted = l2WithBNToStrings(l2); + if (`${normedParam['includeOracle']}`.toLowerCase() === 'true') { + addOracletoResponse( + l2Formatted, + driftClient, + normedMarketType, + normedMarketIndex + ); + } } return l2Formatted; - } else { - // make the BNs into strings - const l2Formatted = l2WithBNToStrings(l2); - if (`${normedParam['includeOracle']}`.toLowerCase() === 'true') { - addOracletoResponse( - l2Formatted, - driftClient, - normedMarketType, - normedMarketIndex - ); - } - return l2Formatted; - } - }); + }) + ); res.writeHead(200); res.end(JSON.stringify({ l2s })); diff --git a/src/publishers/dlobPublisher.ts b/src/publishers/dlobPublisher.ts index 7aa15d6..b6049e6 100644 --- a/src/publishers/dlobPublisher.ts +++ b/src/publishers/dlobPublisher.ts @@ -36,8 +36,6 @@ import { GeyserOrderSubscriber } from '../grpc/OrderSubscriberGRPC'; require('dotenv').config(); const stateCommitment: Commitment = 'processed'; -const ORDERBOOK_UPDATE_INTERVAL = 400; -const WS_FALLBACK_FETCH_INTERVAL = ORDERBOOK_UPDATE_INTERVAL * 10; const driftEnv = (process.env.ENV || 'devnet') as DriftEnv; const commitHash = process.env.COMMIT; const REDIS_HOST = process.env.REDIS_HOST || 'localhost'; @@ -62,6 +60,9 @@ const useOrderSubscriber = const useGrpc = process.env.USE_GRPC?.toLowerCase() === 'true'; const useWebsocket = process.env.USE_WEBSOCKET?.toLowerCase() === 'true'; +const ORDERBOOK_UPDATE_INTERVAL = useGrpc ? 500 : 1000; +const WS_FALLBACK_FETCH_INTERVAL = 10_000; + logger.info(`RPC endpoint: ${endpoint}`); logger.info(`WS endpoint: ${wsEndpoint}`); logger.info(`DriftEnv: ${driftEnv}`); From abc23a01ae4d429a637fe367f0683b3540771d59 Mon Sep 17 00:00:00 2001 From: Nour Alharithi Date: Sat, 9 Dec 2023 16:49:32 -0800 Subject: [PATCH 04/10] slot check before using redis --- src/index.ts | 255 +++++++++++++++++++++++++++++++++------------------ 1 file changed, 168 insertions(+), 87 deletions(-) diff --git a/src/index.ts b/src/index.ts index e364c7d..7e688b2 100644 --- a/src/index.ts +++ b/src/index.ts @@ -81,6 +81,7 @@ const rateLimitCallsPerSecond = process.env.RATE_LIMIT_CALLS_PER_SECOND ? parseInt(process.env.RATE_LIMIT_CALLS_PER_SECOND) : 1; const loadTestAllowed = process.env.ALLOW_LOAD_TEST?.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]'; @@ -268,6 +269,7 @@ const main = async () => { 'updateReceived', (_pubkey: PublicKey, _slot: number, _dataType: 'raw' | 'decoded') => { setLastReceivedWsMsgTs(Date.now()); + // eslint-disable-next-line @typescript-eslint/no-unused-vars updatesReceivedTotal++; accountUpdatesCounter.add(1); } @@ -347,9 +349,12 @@ const main = async () => { `DLOBSubscriber initialized in ${Date.now() - initDlobSubscriberStart} ms` ); - logger.info('Connecting to redis'); - const redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD); - await redisClient.connect(); + let redisClient: RedisClient; + if (useRedis) { + logger.info('Connecting to redis'); + redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD); + await redisClient.connect(); + } logger.info(`Initializing all market subscribers...`); const initAllMarketSubscribersStart = Date.now(); @@ -751,52 +756,93 @@ const main = async () => { } let l2Formatted: any; - if ( - !isSpot && - `${includeVamm}`.toLowerCase() === 'true' && - `${includeOracle}`.toLowerCase().toLowerCase() === 'true' && - !grouping - ) { - if (parseInt(adjustedDepth as string) === 5) { - l2Formatted = await redisClient.client.get( - `last_update_orderbook_perp_${normedMarketIndex}_depth_5` - ); - } else if (parseInt(adjustedDepth as string) === 20) { - l2Formatted = await redisClient.client.get( - `last_update_orderbook_perp_${normedMarketIndex}_depth_20` - ); - } else if (parseInt(adjustedDepth as string) === 100) { - l2Formatted = await redisClient.client.get( - `last_update_orderbook_perp_${normedMarketIndex}_depth_100` - ); + if (useRedis) { + if ( + !isSpot && + `${includeVamm}`.toLowerCase() === 'true' && + `${includeOracle}`.toLowerCase().toLowerCase() === 'true' && + !grouping + ) { + let redisL2: string; + if (parseInt(adjustedDepth as string) === 5) { + redisL2 = await redisClient.client.get( + `last_update_orderbook_perp_${normedMarketIndex}_depth_5` + ); + if ( + Math.abs( + parseInt(JSON.parse(redisL2).slot) - slotSource.getSlot() + ) < 10 + ) + l2Formatted = redisL2; + } else if (parseInt(adjustedDepth as string) === 20) { + redisL2 = await redisClient.client.get( + `last_update_orderbook_perp_${normedMarketIndex}_depth_20` + ); + if ( + Math.abs( + parseInt(JSON.parse(redisL2).slot) - slotSource.getSlot() + ) < 10 + ) + l2Formatted = redisL2; + } else if (parseInt(adjustedDepth as string) === 100) { + redisL2 = await redisClient.client.get( + `last_update_orderbook_perp_${normedMarketIndex}_depth_100` + ); + } + if ( + redisL2 && + Math.abs( + parseInt(JSON.parse(redisL2).slot) - slotSource.getSlot() + ) < 10 + ) + l2Formatted = redisL2; + } else if ( + isSpot && + `${includeSerum}`.toLowerCase() === 'true' && + `${includePhoenix}`.toLowerCase() === 'true' && + `${includeOracle}`.toLowerCase() === 'true' && + !grouping + ) { + let redisL2: string; + if (parseInt(adjustedDepth as string) === 5) { + redisL2 = await redisClient.client.get( + `last_update_orderbook_spot_${normedMarketIndex}_depth_5` + ); + if ( + Math.abs( + parseInt(JSON.parse(redisL2).slot) - slotSource.getSlot() + ) < 10 + ) + l2Formatted = redisL2; + } else if (parseInt(adjustedDepth as string) === 20) { + redisL2 = await redisClient.client.get( + `last_update_orderbook_spot_${normedMarketIndex}_depth_20` + ); + if ( + Math.abs( + parseInt(JSON.parse(redisL2).slot) - slotSource.getSlot() + ) < 10 + ) + l2Formatted = redisL2; + } else if (parseInt(adjustedDepth as string) === 100) { + redisL2 = await redisClient.client.get( + `last_update_orderbook_spot_${normedMarketIndex}_depth_100` + ); + } + if ( + redisL2 && + Math.abs( + parseInt(JSON.parse(redisL2).slot) - slotSource.getSlot() + ) < 10 + ) + l2Formatted = redisL2; } - } else if ( - isSpot && - `${includeSerum}`.toLowerCase() === 'true' && - `${includePhoenix}`.toLowerCase() === 'true' && - `${includeOracle}`.toLowerCase() === 'true' && - !grouping - ) { - if (parseInt(adjustedDepth as string) === 5) { - l2Formatted = await redisClient.client.get( - `last_update_orderbook_spot_${normedMarketIndex}_depth_5` - ); - } else if (parseInt(adjustedDepth as string) === 20) { - l2Formatted = await redisClient.client.get( - `last_update_orderbook_spot_${normedMarketIndex}_depth_20` - ); - } else if (parseInt(adjustedDepth as string) === 100) { - console.log('100'); - l2Formatted = await redisClient.client.get( - `last_update_orderbook_spot_${normedMarketIndex}_depth_100` - ); - } - } - if (l2Formatted) { - res.writeHead(200); - res.end(JSON.stringify(l2Formatted)); - return; + if (l2Formatted) { + res.writeHead(200); + res.end(JSON.stringify(l2Formatted)); + return; + } } const l2 = dlobSubscriber.getL2({ @@ -918,48 +964,83 @@ const main = async () => { } let l2Formatted: any; - if ( - marketType === 'perp' && - normedParam['includeVamm'].toLowerCase() === 'true' && - normedParam['includeOracle'].toLowerCase() === 'true' && - !normedParam['grouping'] - ) { - if (parseInt(adjustedDepth as string) === 5) { - l2Formatted = await redisClient.client.get( - `last_update_orderbook_perp_${normedMarketIndex}_depth_5` - ); - } else if (parseInt(adjustedDepth as string) === 20) { - l2Formatted = await redisClient.client.get( - `last_update_orderbook_perp_${normedMarketIndex}_depth_20` - ); - } else if (parseInt(adjustedDepth as string) === 100) { - l2Formatted = await redisClient.client.get( - `last_update_orderbook_perp_${normedMarketIndex}_depth_100` - ); + if (useRedis) { + if ( + marketType === 'perp' && + normedParam['includeVamm'].toLowerCase() === 'true' && + normedParam['includeOracle'].toLowerCase() === 'true' && + !normedParam['grouping'] + ) { + let redisL2: string; + if (parseInt(adjustedDepth as string) === 5) { + redisL2 = await redisClient.client.get( + `last_update_orderbook_perp_${normedMarketIndex}_depth_5` + ); + if ( + Math.abs( + parseInt(JSON.parse(redisL2).slot) - slotSource.getSlot() + ) < 10 + ) + l2Formatted = redisL2; + } else if (parseInt(adjustedDepth as string) === 20) { + redisL2 = await redisClient.client.get( + `last_update_orderbook_perp_${normedMarketIndex}_depth_20` + ); + if ( + Math.abs( + parseInt(JSON.parse(redisL2).slot) - slotSource.getSlot() + ) < 10 + ) + l2Formatted = redisL2; + } else if (parseInt(adjustedDepth as string) === 100) { + redisL2 = await redisClient.client.get( + `last_update_orderbook_perp_${normedMarketIndex}_depth_100` + ); + } + } else if ( + marketType === 'spot' && + normedParam['includePhoenix'].toLowerCase() === 'true' && + normedParam['includeSerum'].toLowerCase() === 'true' && + !normedParam['grouping'] + ) { + let redisL2: string; + if (parseInt(adjustedDepth as string) === 5) { + redisL2 = await redisClient.client.get( + `last_update_orderbook_spot_${normedMarketIndex}_depth_5` + ); + if ( + Math.abs( + parseInt(JSON.parse(redisL2).slot) - slotSource.getSlot() + ) < 10 + ) + l2Formatted = redisL2; + } else if (parseInt(adjustedDepth as string) === 20) { + redisL2 = await redisClient.client.get( + `last_update_orderbook_spot_${normedMarketIndex}_depth_20` + ); + if ( + Math.abs( + parseInt(JSON.parse(redisL2).slot) - slotSource.getSlot() + ) < 10 + ) + l2Formatted = redisL2; + } else if (parseInt(adjustedDepth as string) === 100) { + redisL2 = await redisClient.client.get( + `last_update_orderbook_spot_${normedMarketIndex}_depth_100` + ); + } + if ( + redisL2 && + Math.abs( + parseInt(JSON.parse(redisL2).slot) - slotSource.getSlot() + ) < 10 + ) + l2Formatted = redisL2; } - } else if ( - marketType === 'spot' && - normedParam['includePhoenix'].toLowerCase() === 'true' && - normedParam['includeSerum'].toLowerCase() === 'true' && - !normedParam['grouping'] - ) { - if (parseInt(adjustedDepth as string) === 5) { - l2Formatted = await redisClient.client.get( - `last_update_orderbook_spot_${normedMarketIndex}_depth_5` - ); - } else if (parseInt(adjustedDepth as string) === 20) { - l2Formatted = await redisClient.client.get( - `last_update_orderbook_spot_${normedMarketIndex}_depth_20` - ); - } else if (parseInt(adjustedDepth as string) === 100) { - l2Formatted = await redisClient.client.get( - `last_update_orderbook_spot_${normedMarketIndex}_depth_100` - ); - } - } - if (l2Formatted) { - return l2Formatted; + if (l2Formatted) { + return l2Formatted; + } } const l2 = dlobSubscriber.getL2({ From 3b5bd2509087e6c1d14b0ab1341ef0fe25390cdc Mon Sep 17 00:00:00 2001 From: GitHub Actions <41898282+github-actions[bot]@users.noreply.github.com> Date: Sun, 10 Dec 2023 02:25:28 +0000 Subject: [PATCH 05/10] Bumping sdk dependency to 2.52.0-beta.0 --- package.json | 2 +- yarn.lock | 8 ++++---- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/package.json b/package.json index d88ab93..7d308f3 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.49.0-beta.19", + "@drift-labs/sdk": "2.52.0-beta.0", "@opentelemetry/api": "^1.1.0", "@opentelemetry/auto-instrumentations-node": "^0.31.1", "@opentelemetry/exporter-prometheus": "^0.31.0", diff --git a/yarn.lock b/yarn.lock index a6a3897..cc463b0 100644 --- a/yarn.lock +++ b/yarn.lock @@ -115,10 +115,10 @@ enabled "2.0.x" kuler "^2.0.0" -"@drift-labs/sdk@2.49.0-beta.19": - version "2.49.0-beta.19" - resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.49.0-beta.19.tgz#cc1d22ab9c46e76008e869c266ed6734e892a461" - integrity sha512-MibgKjqarVlXdumSUgcCK8X87zcXX+S6PkxjPzsXR+Rkoc1hwOHtP0ANzL3ZmBaxsqK8cpPwTM1a90fww0BGRw== +"@drift-labs/sdk@2.52.0-beta.0": + version "2.52.0-beta.0" + resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.52.0-beta.0.tgz#03da2e192e3f2180238e5e464336e0fc1697c0e0" + integrity sha512-SnuXh06oJFzAvTQZ/dOjcWduhKC8rRMlyTE5Xqb3LfCcEkQ/B+XvxTItCSzAPh33qq0MMhZKjxGdqKezUxnV1w== dependencies: "@coral-xyz/anchor" "0.28.1-beta.2" "@ellipsis-labs/phoenix-sdk" "^1.4.2" From ad1ef693fe9e5d34ea595a29aa5cbb97fbc80dd7 Mon Sep 17 00:00:00 2001 From: Nour Alharithi Date: Sun, 10 Dec 2023 14:34:18 -0800 Subject: [PATCH 06/10] metrics counter --- src/core/metrics.ts | 12 ++++++++ src/index.ts | 72 +++++++++------------------------------------ 2 files changed, 26 insertions(+), 58 deletions(-) diff --git a/src/core/metrics.ts b/src/core/metrics.ts index 9675b68..d1683f4 100644 --- a/src/core/metrics.ts +++ b/src/core/metrics.ts @@ -40,6 +40,8 @@ enum METRIC_TYPES { gpa_fetch_duration = 'gpa_fetch_duration', last_ws_message_received_ts = 'last_ws_message_received_ts', account_updates_count = 'account_updates_count', + cache_hit_count = 'cache_hit_count', + cache_miss_count = 'cache_miss_count', current_system_ts = 'current_system_ts', health_status = 'health_status', } @@ -126,6 +128,14 @@ lastWsReceivedTsGauge.addCallback((obs: ObservableResult) => { obs.observe(lastWsMsgReceivedTs, {}); }); +const cacheHitCounter = meter.createCounter(METRIC_TYPES.cache_hit_count, { + description: 'Total redis cache hits', +}); + +const cacheMissCounter = meter.createCounter(METRIC_TYPES.cache_miss_count, { + description: 'Total redis cache misses', +}); + const accountUpdatesCounter = meter.createCounter( METRIC_TYPES.account_updates_count, { @@ -238,4 +248,6 @@ export { handleHealthCheck, setLastReceivedWsMsgTs, accountUpdatesCounter, + cacheHitCounter, + cacheMissCounter, }; diff --git a/src/index.ts b/src/index.ts index 7e688b2..c4031da 100644 --- a/src/index.ts +++ b/src/index.ts @@ -37,7 +37,9 @@ import { gpaFetchDurationHistogram, handleHealthCheck, accountUpdatesCounter, + cacheHitCounter, setLastReceivedWsMsgTs, + cacheMissCounter, } from './core/metrics'; import { handleResponseTime } from './core/middleware'; import { @@ -768,22 +770,10 @@ const main = async () => { redisL2 = await redisClient.client.get( `last_update_orderbook_perp_${normedMarketIndex}_depth_5` ); - if ( - Math.abs( - parseInt(JSON.parse(redisL2).slot) - slotSource.getSlot() - ) < 10 - ) - l2Formatted = redisL2; } else if (parseInt(adjustedDepth as string) === 20) { redisL2 = await redisClient.client.get( `last_update_orderbook_perp_${normedMarketIndex}_depth_20` ); - if ( - Math.abs( - parseInt(JSON.parse(redisL2).slot) - slotSource.getSlot() - ) < 10 - ) - l2Formatted = redisL2; } else if (parseInt(adjustedDepth as string) === 100) { redisL2 = await redisClient.client.get( `last_update_orderbook_perp_${normedMarketIndex}_depth_100` @@ -791,9 +781,7 @@ const main = async () => { } if ( redisL2 && - Math.abs( - parseInt(JSON.parse(redisL2).slot) - slotSource.getSlot() - ) < 10 + slotSource.getSlot() - parseInt(JSON.parse(redisL2).slot) < 10 ) l2Formatted = redisL2; } else if ( @@ -808,22 +796,10 @@ const main = async () => { redisL2 = await redisClient.client.get( `last_update_orderbook_spot_${normedMarketIndex}_depth_5` ); - if ( - Math.abs( - parseInt(JSON.parse(redisL2).slot) - slotSource.getSlot() - ) < 10 - ) - l2Formatted = redisL2; } else if (parseInt(adjustedDepth as string) === 20) { redisL2 = await redisClient.client.get( `last_update_orderbook_spot_${normedMarketIndex}_depth_20` ); - if ( - Math.abs( - parseInt(JSON.parse(redisL2).slot) - slotSource.getSlot() - ) < 10 - ) - l2Formatted = redisL2; } else if (parseInt(adjustedDepth as string) === 100) { redisL2 = await redisClient.client.get( `last_update_orderbook_spot_${normedMarketIndex}_depth_100` @@ -831,14 +807,13 @@ const main = async () => { } if ( redisL2 && - Math.abs( - parseInt(JSON.parse(redisL2).slot) - slotSource.getSlot() - ) < 10 + slotSource.getSlot() - parseInt(JSON.parse(redisL2).slot) < 10 ) l2Formatted = redisL2; } if (l2Formatted) { + cacheHitCounter.add(1); res.writeHead(200); res.end(JSON.stringify(l2Formatted)); return; @@ -896,7 +871,7 @@ const main = async () => { ); } } - + cacheMissCounter.add(1); res.writeHead(200); res.end(JSON.stringify(l2Formatted)); } catch (err) { @@ -976,27 +951,20 @@ const main = async () => { redisL2 = await redisClient.client.get( `last_update_orderbook_perp_${normedMarketIndex}_depth_5` ); - if ( - Math.abs( - parseInt(JSON.parse(redisL2).slot) - slotSource.getSlot() - ) < 10 - ) - l2Formatted = redisL2; } else if (parseInt(adjustedDepth as string) === 20) { redisL2 = await redisClient.client.get( `last_update_orderbook_perp_${normedMarketIndex}_depth_20` ); - if ( - Math.abs( - parseInt(JSON.parse(redisL2).slot) - slotSource.getSlot() - ) < 10 - ) - l2Formatted = redisL2; } else if (parseInt(adjustedDepth as string) === 100) { redisL2 = await redisClient.client.get( `last_update_orderbook_perp_${normedMarketIndex}_depth_100` ); } + if ( + redisL2 && + slotSource.getSlot() - parseInt(JSON.parse(redisL2).slot) < 10 + ) + l2Formatted = redisL2; } else if ( marketType === 'spot' && normedParam['includePhoenix'].toLowerCase() === 'true' && @@ -1008,22 +976,10 @@ const main = async () => { redisL2 = await redisClient.client.get( `last_update_orderbook_spot_${normedMarketIndex}_depth_5` ); - if ( - Math.abs( - parseInt(JSON.parse(redisL2).slot) - slotSource.getSlot() - ) < 10 - ) - l2Formatted = redisL2; } else if (parseInt(adjustedDepth as string) === 20) { redisL2 = await redisClient.client.get( `last_update_orderbook_spot_${normedMarketIndex}_depth_20` ); - if ( - Math.abs( - parseInt(JSON.parse(redisL2).slot) - slotSource.getSlot() - ) < 10 - ) - l2Formatted = redisL2; } else if (parseInt(adjustedDepth as string) === 100) { redisL2 = await redisClient.client.get( `last_update_orderbook_spot_${normedMarketIndex}_depth_100` @@ -1031,14 +987,13 @@ const main = async () => { } if ( redisL2 && - Math.abs( - parseInt(JSON.parse(redisL2).slot) - slotSource.getSlot() - ) < 10 + slotSource.getSlot() - parseInt(JSON.parse(redisL2).slot) < 10 ) l2Formatted = redisL2; } if (l2Formatted) { + cacheHitCounter.add(1); return l2Formatted; } } @@ -1097,6 +1052,7 @@ const main = async () => { ); } } + cacheMissCounter.add(1); return l2Formatted; }) ); From 21c3e8ca8eecbb8479aaee30d45a4d7e6c88babb Mon Sep 17 00:00:00 2001 From: Nour Alharithi Date: Sun, 10 Dec 2023 15:20:01 -0800 Subject: [PATCH 07/10] cache consistent stringify --- src/index.ts | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/src/index.ts b/src/index.ts index c4031da..02db2bd 100644 --- a/src/index.ts +++ b/src/index.ts @@ -815,7 +815,7 @@ const main = async () => { if (l2Formatted) { cacheHitCounter.add(1); res.writeHead(200); - res.end(JSON.stringify(l2Formatted)); + res.end(l2Formatted); return; } } @@ -960,11 +960,11 @@ const main = async () => { `last_update_orderbook_perp_${normedMarketIndex}_depth_100` ); } - if ( - redisL2 && - slotSource.getSlot() - parseInt(JSON.parse(redisL2).slot) < 10 - ) - l2Formatted = redisL2; + if (redisL2) { + const parsedRedisL2 = JSON.parse(redisL2); + if (slotSource.getSlot() - parseInt(parsedRedisL2.slot) < 10) + l2Formatted = parsedRedisL2; + } } else if ( marketType === 'spot' && normedParam['includePhoenix'].toLowerCase() === 'true' && @@ -985,11 +985,11 @@ const main = async () => { `last_update_orderbook_spot_${normedMarketIndex}_depth_100` ); } - if ( - redisL2 && - slotSource.getSlot() - parseInt(JSON.parse(redisL2).slot) < 10 - ) - l2Formatted = redisL2; + if (redisL2) { + const parsedRedisL2 = JSON.parse(redisL2); + if (slotSource.getSlot() - parseInt(parsedRedisL2.slot) < 10) + l2Formatted = parsedRedisL2; + } } if (l2Formatted) { From 27bfef5a1ef0b8e2cfd83cc168365b0e892ff58b Mon Sep 17 00:00:00 2001 From: Nour Alharithi Date: Mon, 11 Dec 2023 06:51:41 -0800 Subject: [PATCH 08/10] try catch for l2 and no ping/pong --- src/dlob-subscriber/DLOBSubscriberIO.ts | 8 +++++++- src/wsConnectionManager.ts | 22 ---------------------- 2 files changed, 7 insertions(+), 23 deletions(-) diff --git a/src/dlob-subscriber/DLOBSubscriberIO.ts b/src/dlob-subscriber/DLOBSubscriberIO.ts index 0a20194..1f6faf1 100644 --- a/src/dlob-subscriber/DLOBSubscriberIO.ts +++ b/src/dlob-subscriber/DLOBSubscriberIO.ts @@ -18,6 +18,7 @@ import { addOracletoResponse, l2WithBNToStrings, } from '../utils/utils'; +import { logger } from '../utils/logger'; type wsMarketL2Args = { marketIndex: number; @@ -87,7 +88,12 @@ export class DLOBSubscriberIO extends DLOBSubscriber { override async updateDLOB(): Promise { await super.updateDLOB(); for (const l2Args of this.marketL2Args) { - this.getL2AndSendMsg(l2Args); + try { + this.getL2AndSendMsg(l2Args); + } catch (error) { + logger.error(error); + console.log(`Error getting L2 ${l2Args.marketName}`); + } } } diff --git a/src/wsConnectionManager.ts b/src/wsConnectionManager.ts index 744b6fd..bb37b5c 100644 --- a/src/wsConnectionManager.ts +++ b/src/wsConnectionManager.ts @@ -233,31 +233,9 @@ async function main() { } }); - // Ping/pong connection timeout - let pongTimeoutId; - let isAlive = true; - const pingIntervalId = setInterval(() => { - isAlive = false; - pongTimeoutId = setTimeout(() => { - if (!isAlive) { - console.log('Disconnecting because of ping/pong timeout'); - ws.terminate(); - } - }, 5000); // 5 seconds to wait for a pong - ws.ping(); - }, 30000); - - // Listen for pong messages - ws.on('pong', () => { - isAlive = true; - clearTimeout(pongTimeoutId); - }); - // Handle disconnection ws.on('close', () => { // Clear any existing intervals and timeouts - clearInterval(pingIntervalId); - clearTimeout(pongTimeoutId); channelSubscribers.forEach((subscribers, channel) => { if (subscribers.delete(ws) && subscribers.size === 0) { redisClient.client.unsubscribe(channel); From 1324465d33572e13a29058229d03c70ff1b7b9f8 Mon Sep 17 00:00:00 2001 From: Nour Alharithi Date: Mon, 11 Dec 2023 07:05:04 -0800 Subject: [PATCH 09/10] resub timeout on publisher ws --- src/publishers/dlobPublisher.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/src/publishers/dlobPublisher.ts b/src/publishers/dlobPublisher.ts index b6049e6..5209823 100644 --- a/src/publishers/dlobPublisher.ts +++ b/src/publishers/dlobPublisher.ts @@ -149,6 +149,7 @@ const main = async () => { accountSubscription = { type: 'websocket', commitment: stateCommitment, + resubTimeoutMs: 30_000, }; slotSubscriber = new SlotSubscriber(connection); await slotSubscriber.subscribe(); From 41728ad1f77194929415376629801ef774c77fdf Mon Sep 17 00:00:00 2001 From: Nour Alharithi Date: Mon, 11 Dec 2023 08:11:39 -0800 Subject: [PATCH 10/10] change cache hit metric --- src/core/metrics.ts | 5 ----- src/index.ts | 17 ++++++++++++----- 2 files changed, 12 insertions(+), 10 deletions(-) diff --git a/src/core/metrics.ts b/src/core/metrics.ts index d1683f4..6b102f7 100644 --- a/src/core/metrics.ts +++ b/src/core/metrics.ts @@ -132,10 +132,6 @@ const cacheHitCounter = meter.createCounter(METRIC_TYPES.cache_hit_count, { description: 'Total redis cache hits', }); -const cacheMissCounter = meter.createCounter(METRIC_TYPES.cache_miss_count, { - description: 'Total redis cache misses', -}); - const accountUpdatesCounter = meter.createCounter( METRIC_TYPES.account_updates_count, { @@ -249,5 +245,4 @@ export { setLastReceivedWsMsgTs, accountUpdatesCounter, cacheHitCounter, - cacheMissCounter, }; diff --git a/src/index.ts b/src/index.ts index 02db2bd..8dd51f0 100644 --- a/src/index.ts +++ b/src/index.ts @@ -39,7 +39,6 @@ import { accountUpdatesCounter, cacheHitCounter, setLastReceivedWsMsgTs, - cacheMissCounter, } from './core/metrics'; import { handleResponseTime } from './core/middleware'; import { @@ -813,7 +812,9 @@ const main = async () => { } if (l2Formatted) { - cacheHitCounter.add(1); + cacheHitCounter.add(1, { + miss: false, + }); res.writeHead(200); res.end(l2Formatted); return; @@ -871,7 +872,9 @@ const main = async () => { ); } } - cacheMissCounter.add(1); + cacheHitCounter.add(1, { + miss: true, + }); res.writeHead(200); res.end(JSON.stringify(l2Formatted)); } catch (err) { @@ -993,7 +996,9 @@ const main = async () => { } if (l2Formatted) { - cacheHitCounter.add(1); + cacheHitCounter.add(1, { + miss: false, + }); return l2Formatted; } } @@ -1052,7 +1057,9 @@ const main = async () => { ); } } - cacheMissCounter.add(1); + cacheHitCounter.add(1, { + miss: true, + }); return l2Formatted; }) );