diff --git a/package.json b/package.json index 8e7e9ba..729e823 100644 --- a/package.json +++ b/package.json @@ -5,7 +5,7 @@ "main": "lib/index.js", "license": "Apache-2.0", "dependencies": { - "@drift-labs/sdk": "2.48.0-beta.9", + "@drift-labs/sdk": "2.48.0-beta.13", "@opentelemetry/api": "^1.1.0", "@opentelemetry/auto-instrumentations-node": "^0.31.1", "@opentelemetry/exporter-prometheus": "^0.31.0", diff --git a/src/core/metrics.ts b/src/core/metrics.ts index 344661a..99e9de0 100644 --- a/src/core/metrics.ts +++ b/src/core/metrics.ts @@ -134,11 +134,10 @@ const handleHealthCheck = async (req, res, next) => { if (lastHealthCheckState) { res.writeHead(200); res.end('OK'); - } else { - res.writeHead(500); - res.end(`NOK`); + lastHealthCheckPerformed = Date.now(); + return; } - return; + // always check if last check was unhealthy (give it another chance to recover) } const { lastSlotReceived, lastSlotReceivedMutex } = getSlotHealthCheckInfo(); @@ -149,7 +148,9 @@ const handleHealthCheck = async (req, res, next) => { lastHealthCheckState = lastSlotReceived > lastHealthCheckSlot; if (!lastHealthCheckState) { logger.error( - `Unhealthy: lastSlot: ${lastSlotReceived}, lastHealthCheckSlot: ${lastHealthCheckSlot}` + `Unhealthy: lastSlot: ${lastSlotReceived}, lastHealthCheckSlot: ${lastHealthCheckSlot}, timeSinceLastCheck: ${ + Date.now() - lastHealthCheckPerformed + } ms` ); } diff --git a/src/dlobProvider.ts b/src/dlobProvider.ts new file mode 100644 index 0000000..1470543 --- /dev/null +++ b/src/dlobProvider.ts @@ -0,0 +1,77 @@ +import { DLOB, OrderSubscriber, UserAccount, UserMap } from '@drift-labs/sdk'; +import { PublicKey } from '@solana/web3.js'; + +export type DLOBProvider = { + subscribe(): Promise; + getDLOB(slot: number): Promise; + getUniqueAuthorities(): PublicKey[]; + getUserAccounts(): Generator<{ + userAccount: UserAccount; + publicKey: PublicKey; + }>; + getUserAccount(publicKey: PublicKey): UserAccount | undefined; + size(): number; +}; + +export function getDLOBProviderFromUserMap(userMap: UserMap): DLOBProvider { + return { + subscribe: async () => { + await userMap.subscribe(); + }, + getDLOB: async (slot: number) => { + return await userMap.getDLOB(slot); + }, + getUniqueAuthorities: () => { + return userMap.getUniqueAuthorities(); + }, + getUserAccounts: function* () { + for (const user of userMap.values()) { + yield { + userAccount: user.getUserAccount(), + publicKey: user.getUserAccountPublicKey(), + }; + } + }, + getUserAccount: (publicKey) => { + return userMap.get(publicKey.toString())?.getUserAccount(); + }, + size: () => { + return userMap.size(); + }, + }; +} + +export function getDLOBProviderFromOrderSubscriber( + orderSubscriber: OrderSubscriber +): DLOBProvider { + return { + subscribe: async () => { + await orderSubscriber.subscribe(); + }, + getDLOB: async (slot: number) => { + return await orderSubscriber.getDLOB(slot); + }, + getUniqueAuthorities: () => { + const authorities = new Set(); + for (const { userAccount } of orderSubscriber.usersAccounts.values()) { + authorities.add(userAccount.authority); + } + return Array.from(authorities.values()); + }, + getUserAccounts: function* () { + for (const [ + key, + { userAccount }, + ] of orderSubscriber.usersAccounts.entries()) { + yield { userAccount: userAccount, publicKey: new PublicKey(key) }; + } + }, + getUserAccount: (publicKey) => { + return orderSubscriber.usersAccounts.get(publicKey.toString()) + ?.userAccount; + }, + size(): number { + return orderSubscriber.usersAccounts.size; + }, + }; +} diff --git a/src/index.ts b/src/index.ts index e327da4..888b6a1 100644 --- a/src/index.ts +++ b/src/index.ts @@ -21,12 +21,13 @@ import { SlotSource, SlotSubscriber, UserMap, - UserStatsMap, Wallet, + getUserStatsAccountPublicKey, getVariant, groupL2, initialize, isVariant, + OrderSubscriber, } from '@drift-labs/sdk'; import { logger, setLogLevel } from './utils/logger'; @@ -46,6 +47,11 @@ import { sleep, validateDlobQuery, } from './utils/utils'; +import { + DLOBProvider, + getDLOBProviderFromOrderSubscriber, + getDLOBProviderFromUserMap, +} from './dlobProvider'; require('dotenv').config(); const driftEnv = (process.env.ENV || 'devnet') as DriftEnv; @@ -57,6 +63,8 @@ const stateCommitment: Commitment = 'processed'; const serverPort = process.env.PORT || 6969; export const ORDERBOOK_UPDATE_INTERVAL = 1000; const useWebsocket = process.env.USE_WEBSOCKET?.toLowerCase() === 'true'; +const useOrderSubscriber = + process.env.USE_ORDER_SUBSCRIBER?.toLowerCase() === 'true'; const rateLimitCallsPerSecond = process.env.RATE_LIMIT_CALLS_PER_SECOND ? parseInt(process.env.RATE_LIMIT_CALLS_PER_SECOND) @@ -66,7 +74,7 @@ const loadTestAllowed = process.env.ALLOW_LOAD_TEST?.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]'; const logHttp = morgan(logFormat, { - skip: (_req, res) => res.statusCode < 400, + skip: (_req, res) => res.statusCode <= 500, }); let driftClient: DriftClient; @@ -196,6 +204,7 @@ const main = async () => { type: 'polling', accountLoader: bulkAccountLoader, }; + slotSource = { getSlot: () => bulkAccountLoader!.getSlot(), }; @@ -206,6 +215,7 @@ const main = async () => { }; slotSubscriber = new SlotSubscriber(connection); await slotSubscriber.subscribe(); + slotSource = { getSlot: () => slotSubscriber!.getSlot(), }; @@ -219,6 +229,40 @@ const main = async () => { env: driftEnv, }); + let dlobProvider: DLOBProvider; + if (useOrderSubscriber) { + let subscriptionConfig; + if (useWebsocket) { + subscriptionConfig = { + type: 'websocket', + }; + } else { + subscriptionConfig = { + type: 'polling', + frequency: ORDERBOOK_UPDATE_INTERVAL, + }; + } + + const orderSubscriber = new OrderSubscriber({ + driftClient, + subscriptionConfig, + }); + + dlobProvider = getDLOBProviderFromOrderSubscriber(orderSubscriber); + + slotSource = { + getSlot: () => orderSubscriber.getSlot(), + }; + } else { + const userMap = new UserMap( + driftClient, + driftClient.userAccountSubscriptionConfig, + false + ); + + dlobProvider = getDLOBProviderFromUserMap(userMap); + } + const dlobCoder = DLOBOrdersCoder.create(); await driftClient.subscribe(); @@ -231,27 +275,19 @@ const main = async () => { lastSlotReceived = slotSource.getSlot(); }, ORDERBOOK_UPDATE_INTERVAL); - const userStatsMap = new UserStatsMap(driftClient); - - logger.info(`Initializing userMap...`); - const initUserMapStart = Date.now(); - const userMap = new UserMap( - driftClient, - driftClient.userAccountSubscriptionConfig, - false, - async (authorities) => { - await userStatsMap.sync(authorities); - }, - { hasOpenOrders: true } + logger.info(`Initializing DLOB Provider...`); + const initDLOBProviderStart = Date.now(); + await dlobProvider.subscribe(); + logger.info( + `dlob provider initialized in ${Date.now() - initDLOBProviderStart} ms` ); - await userMap.subscribe(); - logger.info(`userMap initialized in ${Date.now() - initUserMapStart} ms`); + logger.info(`dlob provider size ${dlobProvider.size()}`); logger.info(`Initializing DLOBSubscriber...`); const initDlobSubscriberStart = Date.now(); const dlobSubscriber = new DLOBSubscriber({ driftClient, - dlobSource: userMap, + dlobSource: dlobProvider, slotSource, updateFrequency: ORDERBOOK_UPDATE_INTERVAL, }); @@ -263,11 +299,7 @@ const main = async () => { MARKET_SUBSCRIBERS = await initializeAllMarketSubscribers(driftClient); const handleStartup = async (_req, res, _next) => { - if ( - driftClient.isSubscribed && - userMap.size() > 0 && - userStatsMap.size() > 0 - ) { + if (driftClient.isSubscribed && dlobProvider.size() > 0) { res.writeHead(200); res.end('OK'); } else { @@ -297,16 +329,14 @@ const main = async () => { }); } - for (const user of userMap.values()) { - const userAccount = user.getUserAccount(); - + for (const { userAccount, publicKey } of dlobProvider.getUserAccounts()) { for (const order of userAccount.orders) { if (isVariant(order.status, 'init')) { continue; } orders.push({ - user: user.getUserAccountPublicKey().toBase58(), + user: publicKey.toBase58(), order: order, }); } @@ -352,9 +382,7 @@ const main = async () => { } oracles.push(oracleHuman); } - for (const user of userMap.values()) { - const userAccount = user.getUserAccount(); - + for (const { userAccount, publicKey } of dlobProvider.getUserAccounts()) { for (const order of userAccount.orders) { if (isVariant(order.status, 'init')) { continue; @@ -392,7 +420,7 @@ const main = async () => { } orders.push({ - user: user.getUserAccountPublicKey().toBase58(), + user: publicKey.toBase58(), order: orderHuman, }); } @@ -416,16 +444,14 @@ const main = async () => { try { const dlobOrders: DLOBOrders = []; - for (const user of userMap.values()) { - const userAccount = user.getUserAccount(); - + for (const { userAccount, publicKey } of dlobProvider.getUserAccounts()) { for (const order of userAccount.orders) { if (isVariant(order.status, 'init')) { continue; } dlobOrders.push({ - user: user.getUserAccountPublicKey(), + user: publicKey, order, } as DLOBOrder); } @@ -466,9 +492,7 @@ const main = async () => { const dlobOrders: DLOBOrders = []; - for (const user of userMap.values()) { - const userAccount = user.getUserAccount(); - + for (const { userAccount, publicKey } of dlobProvider.getUserAccounts()) { for (const order of userAccount.orders) { if (isVariant(order.status, 'init')) { continue; @@ -484,7 +508,7 @@ const main = async () => { } dlobOrders.push({ - user: user.getUserAccountPublicKey(), + user: publicKey, order, } as DLOBOrder); } @@ -555,14 +579,15 @@ const main = async () => { continue; } else { if (`${includeUserStats}`.toLowerCase() === 'true') { - const userAccount = side.userAccount.toBase58(); - await userMap.mustGet(userAccount); - const userStats = await userStatsMap.mustGet( - userMap.getUserAuthority(userAccount)!.toBase58() + const userAccount = dlobProvider.getUserAccount( + side.userAccount ); topMakers.add([ userAccount, - userStats.userStatsAccountPublicKey.toBase58(), + getUserStatsAccountPublicKey( + driftClient.program.programId, + userAccount.authority + ), ]); } else { topMakers.add(side.userAccount.toBase58()); diff --git a/src/publishers/dlobPublisher.ts b/src/publishers/dlobPublisher.ts index a598f30..9ad03d7 100644 --- a/src/publishers/dlobPublisher.ts +++ b/src/publishers/dlobPublisher.ts @@ -6,16 +6,22 @@ import { DriftClient, initialize, DriftEnv, - SlotSubscriber, UserMap, Wallet, BulkAccountLoader, + OrderSubscriber, + SlotSource, } from '@drift-labs/sdk'; import { logger, setLogLevel } from '../utils/logger'; import { sleep } from '../utils/utils'; import { DLOBSubscriberIO } from '../dlob-subscriber/DLOBSubscriberIO'; import { RedisClient } from '../utils/redisClient'; +import { + DLOBProvider, + getDLOBProviderFromOrderSubscriber, + getDLOBProviderFromUserMap, +} from '../dlobProvider'; require('dotenv').config(); const driftEnv = (process.env.ENV || 'devnet') as DriftEnv; @@ -37,6 +43,9 @@ setLogLevel(opts.debug ? 'debug' : 'info'); const endpoint = process.env.ENDPOINT; const wsEndpoint = process.env.WS_ENDPOINT; +const useOrderSubscriber = + process.env.USE_ORDER_SUBSCRIBER?.toLowerCase() === 'true'; + logger.info(`RPC endpoint: ${endpoint}`); logger.info(`WS endpoint: ${wsEndpoint}`); logger.info(`DriftEnv: ${driftEnv}`); @@ -68,8 +77,6 @@ const main = async () => { env: driftEnv, }); - const slotSubscriber = new SlotSubscriber(connection, {}); - const lamportsBalance = await connection.getBalance(wallet.publicKey); logger.info( `DriftClient ProgramId: ${driftClient.program.programId.toBase58()}` @@ -83,22 +90,45 @@ const main = async () => { logger.error(e); }); - await slotSubscriber.subscribe(); + let slotSource: SlotSource; + let dlobProvider: DLOBProvider; + if (useOrderSubscriber) { + const orderSubscriber = new OrderSubscriber({ + driftClient, + subscriptionConfig: { + type: 'polling', + frequency: ORDERBOOK_UPDATE_INTERVAL, + }, + }); - const userMap = new UserMap( - driftClient, - driftClient.userAccountSubscriptionConfig, - false - ); - await userMap.subscribe(); + dlobProvider = getDLOBProviderFromOrderSubscriber(orderSubscriber); + + slotSource = { + getSlot: () => orderSubscriber.getSlot(), + }; + } else { + const userMap = new UserMap( + driftClient, + driftClient.userAccountSubscriptionConfig, + false + ); + + dlobProvider = getDLOBProviderFromUserMap(userMap); + + slotSource = { + getSlot: () => bulkAccountLoader.getSlot(), + }; + } + + await dlobProvider.subscribe(); const redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD); await redisClient.connect(); const dlobSubscriber = new DLOBSubscriberIO({ driftClient, - dlobSource: userMap, - slotSource: slotSubscriber, + dlobSource: dlobProvider, + slotSource, updateFrequency: ORDERBOOK_UPDATE_INTERVAL, redisClient, }); diff --git a/yarn.lock b/yarn.lock index fe86aa4..8f7a265 100644 --- a/yarn.lock +++ b/yarn.lock @@ -80,10 +80,10 @@ enabled "2.0.x" kuler "^2.0.0" -"@drift-labs/sdk@2.48.0-beta.9": - version "2.48.0-beta.9" - resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.48.0-beta.9.tgz#2d0c99f58387ff3c5e45859ae4fe088339c8dd55" - integrity sha512-y0wrq680nv7Uwvcg9jOrjZPXxoae9yRg8M9B4bQgDFeNJmZNmcEmnLhWQ+CbPi8u/O0glhOdtKUk4UF9eSPXAg== +"@drift-labs/sdk@2.48.0-beta.13": + version "2.48.0-beta.13" + resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.48.0-beta.13.tgz#0ad1cfd4daac97e1859fe617a749a2676e462b63" + integrity sha512-ogtY38Db44PVZOr8WwNHijnwVMBLFvIYNxmlE9BPehdEgIQ9PYT9QXalanhQ1Ruc/Cs5DbMIT3XvjtLyfFzEUg== dependencies: "@coral-xyz/anchor" "0.28.1-beta.2" "@ellipsis-labs/phoenix-sdk" "^1.4.2"