diff --git a/Dockerfile b/Dockerfile index db035ec..fb0e939 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,25 +1,29 @@ -FROM public.ecr.aws/docker/library/node:22 AS builder +FROM public.ecr.aws/docker/library/node:24 AS builder RUN npm install -g typescript@5.4.5 +RUN yarn config set registry https://registry.npmjs.org WORKDIR /app COPY . . -WORKDIR /app/drift-common/protocol/sdk +RUN rm -rf /app/dlob-server/drift-common && \ + cp -a /app/_tmp/trade-drift-dlob-main-20260313/drift-common /app/dlob-server/drift-common + +WORKDIR /app/dlob-server/drift-common/protocol/sdk RUN yarn && yarn build -WORKDIR /app/drift-common/common-ts +WORKDIR /app/dlob-server/drift-common/common-ts RUN yarn && yarn build -WORKDIR /app +WORKDIR /app/dlob-server RUN yarn && yarn build -FROM public.ecr.aws/docker/library/node:22-alpine +FROM public.ecr.aws/docker/library/node:24-alpine RUN apk add python3 make g++ --virtual .build &&\ npm install -C /lib bigint-buffer @triton-one/yellowstone-grpc@1.3.0 helius-laserstream@0.1.8 rpc-websockets@7.5.1 &&\ apk del .build -COPY --from=builder /app/lib/ ./lib/ +COPY --from=builder /app/dlob-server/lib/ ./lib/ ENV NODE_ENV=production EXPOSE 9464 diff --git a/package.json b/package.json index 440a8af..8cac28f 100644 --- a/package.json +++ b/package.json @@ -32,6 +32,7 @@ "express": "^4.18.2", "ioredis": "^5.4.1", "morgan": "^1.10.0", + "pg": "^8.12.0", "redis": "^4.6.10", "response-time": "^2.3.2", "rxjs": "^7.8.1", diff --git a/src/athena/testRedisPolling.ts b/src/athena/testRedisPolling.ts index c52906d..61698e2 100644 --- a/src/athena/testRedisPolling.ts +++ b/src/athena/testRedisPolling.ts @@ -1,4 +1,4 @@ -import { RedisClient, RedisClientPrefix } from '@drift/common/clients'; +import { RedisClient, HOT_REDIS_KEY_PREFIX } from '../utils/redisClients'; import { logger, setLogLevel } from '../utils/logger'; setLogLevel('info'); @@ -194,10 +194,10 @@ const main = async () => { logger.info(`📡 Connecting to Redis:`); logger.info(` Host: ${redisHost}`); logger.info(` Port: ${redisPort}`); - logger.info(` Prefix: ${RedisClientPrefix.DLOB}`); + logger.info(` Prefix: ${HOT_REDIS_KEY_PREFIX}`); const redisClient = new RedisClient({ - prefix: RedisClientPrefix.DLOB, + prefix: HOT_REDIS_KEY_PREFIX, }); try { diff --git a/src/dlob-subscriber/OrderSubscriberFiltered.ts b/src/dlob-subscriber/OrderSubscriberFiltered.ts index a69cfca..2efe673 100644 --- a/src/dlob-subscriber/OrderSubscriberFiltered.ts +++ b/src/dlob-subscriber/OrderSubscriberFiltered.ts @@ -3,6 +3,7 @@ import { OrderSubscriberConfig, UserAccount, } from '@drift-labs/sdk'; +import { ReconnectingGrpcOrderSubscription } from './ReconnectingGrpcOrderSubscription'; export class OrderSubscriberFiltered extends OrderSubscriber { public ignoreList: string[]; @@ -14,6 +15,20 @@ export class OrderSubscriberFiltered extends OrderSubscriber { ) { super(config); this.ignoreList = config.ignoreList ?? []; + + if (config.subscriptionConfig.type === 'grpc') { + (this as any).subscription = new ReconnectingGrpcOrderSubscription({ + orderSubscriber: this, + grpcConfigs: config.subscriptionConfig.grpcConfigs, + skipInitialLoad: config.subscriptionConfig.skipInitialLoad, + resubOpts: { + resubTimeoutMs: config.subscriptionConfig.resubTimeoutMs, + logResubMessages: config.subscriptionConfig.logResubMessages, + }, + resyncIntervalMs: config.subscriptionConfig.resyncIntervalMs, + decoded: config.decodeData, + }); + } } override async tryUpdateUserAccount( diff --git a/src/dlob-subscriber/ReconnectingGrpcOrderSubscription.ts b/src/dlob-subscriber/ReconnectingGrpcOrderSubscription.ts new file mode 100644 index 0000000..a7bc312 --- /dev/null +++ b/src/dlob-subscriber/ReconnectingGrpcOrderSubscription.ts @@ -0,0 +1,614 @@ +import type { OrderSubscriber, UserAccount } from '@drift-labs/sdk'; +import { Context, MemcmpFilter, PublicKey } from '@solana/web3.js'; +import { Buffer } from 'buffer'; +import { + CommitmentLevel, + createClient, + type Client, + type ClientDuplexStream, + type SubscribeRequest, + type SubscribeUpdate, +} from '@drift-labs/sdk/lib/node/isomorphic/grpc'; +import type { + GrpcConfigs, + ResubOpts, +} from '@drift-labs/sdk/lib/node/accounts/types'; +import { WebSocketProgramAccountSubscriber } from '@drift-labs/sdk/lib/node/accounts/webSocketProgramAccountSubscriber'; +import { LaserstreamProgramAccountSubscriber } from '@drift-labs/sdk/lib/node/accounts/laserProgramAccountSubscriber'; +import { getNonIdleUserFilter, getUserFilter } from '@drift-labs/sdk'; +import bs58 from 'bs58'; +import { logger } from '../utils/logger'; + +type StreamFailureEvent = 'error' | 'end' | 'close'; + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +function getReconnectDelayMs(): number { + const raw = process.env.GRPC_RECONNECT_DELAY_MS; + const parsed = raw ? parseInt(raw, 10) : 1000; + return Number.isFinite(parsed) && parsed >= 0 ? parsed : 1000; +} + +function formatError(error: unknown): string { + if (error instanceof Error) { + return `${error.name}: ${error.message}`; + } + return String(error); +} + +function isExpectedGrpcTeardownError(error: unknown): boolean { + if (!error || typeof error !== 'object') { + return false; + } + + const maybeGrpcError = error as { + code?: unknown; + details?: unknown; + message?: unknown; + }; + + const details = + typeof maybeGrpcError.details === 'string' ? maybeGrpcError.details : ''; + const message = + typeof maybeGrpcError.message === 'string' ? maybeGrpcError.message : ''; + + return ( + maybeGrpcError.code === 1 || + details.includes('Cancelled on client') || + message.includes('Cancelled on client') + ); +} + +class ReconnectingGrpcProgramAccountSubscriber< + T, +> extends WebSocketProgramAccountSubscriber { + private client: Client; + private stream?: ClientDuplexStream; + private commitmentLevel: CommitmentLevel; + private streamFailureHandled = false; + private readonly onStreamFailure?: ( + event: StreamFailureEvent, + error?: unknown + ) => void; + + private dataHandler?: (chunk: SubscribeUpdate) => void; + private errorHandler?: (error: Error) => void; + private endHandler?: () => void; + private closeHandler?: () => void; + + private constructor( + client: Client, + commitmentLevel: CommitmentLevel, + subscriptionName: string, + accountDiscriminator: string, + program: any, + decodeBufferFn: (accountName: string, ix: Buffer) => T, + options: { filters: MemcmpFilter[] } = { + filters: [], + }, + resubOpts?: ResubOpts, + onStreamFailure?: (event: StreamFailureEvent, error?: unknown) => void + ) { + super( + subscriptionName, + accountDiscriminator, + program, + decodeBufferFn, + options, + resubOpts + ); + this.client = client; + this.commitmentLevel = commitmentLevel; + this.onStreamFailure = onStreamFailure; + } + + public static async create( + grpcConfigs: GrpcConfigs, + subscriptionName: string, + accountDiscriminator: string, + program: any, + decodeBufferFn: (accountName: string, ix: Buffer) => U, + options: { filters: MemcmpFilter[] } = { + filters: [], + }, + resubOpts?: ResubOpts, + onStreamFailure?: (event: StreamFailureEvent, error?: unknown) => void + ): Promise> { + const client = await createClient( + grpcConfigs.endpoint, + grpcConfigs.token, + grpcConfigs.channelOptions ?? {} + ); + const commitmentLevel = + // @ts-ignore runtime enum value is valid, TS types are incomplete here + grpcConfigs.commitmentLevel ?? CommitmentLevel.CONFIRMED; + + return new ReconnectingGrpcProgramAccountSubscriber( + client, + commitmentLevel, + subscriptionName, + accountDiscriminator, + program, + decodeBufferFn, + options, + resubOpts, + onStreamFailure + ); + } + + override async subscribe( + onChange: ( + accountId: PublicKey, + data: T, + context: Context, + buffer: Buffer + ) => void + ): Promise { + if (this.listenerId != null || this.isUnsubscribing) { + return; + } + + this.onChange = onChange; + this.streamFailureHandled = false; + this.stream = + (await this.client.subscribe()) as unknown as typeof this.stream; + + const filters = this.options.filters.map((filter) => { + return { + memcmp: { + offset: filter.memcmp.offset.toString(), + bytes: bs58.decode(filter.memcmp.bytes), + }, + }; + }); + + const request: SubscribeRequest = { + slots: {}, + accounts: { + drift: { + account: [], + owner: [this.program.programId.toBase58()], + filters, + }, + }, + transactions: {}, + blocks: {}, + blocksMeta: {}, + accountsDataSlice: [], + commitment: this.commitmentLevel, + entry: {}, + transactionsStatus: {}, + }; + + this.dataHandler = (chunk: SubscribeUpdate) => { + if (!chunk.account) { + return; + } + + const slot = Number(chunk.account.slot); + const accountInfo = { + owner: new PublicKey(chunk.account.account.owner), + lamports: Number(chunk.account.account.lamports), + data: Buffer.from(chunk.account.account.data), + executable: chunk.account.account.executable, + rentEpoch: Number(chunk.account.account.rentEpoch), + }; + + if (this.resubOpts?.resubTimeoutMs) { + this.receivingData = true; + clearTimeout(this.timeoutId); + this.handleRpcResponse( + { slot }, + { + accountId: new PublicKey(chunk.account.account.pubkey), + accountInfo, + } + ); + this.setTimeout(); + } else { + this.handleRpcResponse( + { slot }, + { + accountId: new PublicKey(chunk.account.account.pubkey), + accountInfo, + } + ); + } + }; + this.errorHandler = (error: Error) => { + this.handleStreamFailure('error', error); + }; + this.endHandler = () => { + this.handleStreamFailure('end'); + }; + this.closeHandler = () => { + this.handleStreamFailure('close'); + }; + + this.stream.on('data', this.dataHandler); + this.stream.on('error', this.errorHandler); + this.stream.on('end', this.endHandler); + this.stream.on('close', this.closeHandler); + + try { + await new Promise((resolve, reject) => { + this.stream!.write(request, (err) => { + if (err === null || err === undefined) { + this.listenerId = 1; + if (this.resubOpts?.resubTimeoutMs) { + this.receivingData = true; + this.setTimeout(); + } + resolve(); + } else { + reject(err); + } + }); + }); + } catch (reason) { + this.cleanupStream(); + console.error(reason); + throw reason; + } + } + + override async unsubscribe(onResub = false): Promise { + if (!onResub && this.resubOpts) { + this.resubOpts.resubTimeoutMs = undefined; + } + this.isUnsubscribing = true; + clearTimeout(this.timeoutId); + this.timeoutId = undefined; + + try { + if (this.listenerId != null && this.stream) { + const request: SubscribeRequest = { + slots: {}, + accounts: {}, + transactions: {}, + blocks: {}, + blocksMeta: {}, + accountsDataSlice: [], + entry: {}, + transactionsStatus: {}, + }; + + await new Promise((resolve, reject) => { + this.stream!.write(request, (err) => { + if (err === null || err === undefined) { + resolve(); + } else { + reject(err); + } + }); + }); + } + } catch (reason) { + console.error(reason); + } finally { + this.listenerId = undefined; + this.receivingData = false; + this.cleanupStream(); + this.isUnsubscribing = false; + } + } + + private handleStreamFailure( + event: StreamFailureEvent, + error?: unknown + ): void { + if (this.isUnsubscribing || this.streamFailureHandled) { + return; + } + + this.streamFailureHandled = true; + this.listenerId = undefined; + this.receivingData = false; + clearTimeout(this.timeoutId); + this.timeoutId = undefined; + this.cleanupStream(); + this.onStreamFailure?.(event, error); + } + + private cleanupStream(): void { + const stream = this.stream; + this.stream = undefined; + + if (!stream) { + return; + } + + if (this.dataHandler) { + stream.removeListener('data', this.dataHandler); + this.dataHandler = undefined; + } + if (this.errorHandler) { + stream.removeListener('error', this.errorHandler); + this.errorHandler = undefined; + } + if (this.endHandler) { + stream.removeListener('end', this.endHandler); + this.endHandler = undefined; + } + if (this.closeHandler) { + stream.removeListener('close', this.closeHandler); + this.closeHandler = undefined; + } + + stream.once('error', (error) => { + if (isExpectedGrpcTeardownError(error)) { + logger.debug( + `Ignoring expected gRPC stream error during teardown: ${formatError( + error + )}` + ); + return; + } + + logger.warn( + `Ignoring unexpected gRPC stream error during teardown: ${formatError( + error + )}` + ); + }); + + try { + if (typeof (stream as any).cancel === 'function') { + (stream as any).cancel(); + } + } catch { + // Ignore stream teardown errors while recovering. + } + } +} + +export class ReconnectingGrpcOrderSubscription { + private orderSubscriber: OrderSubscriber; + private skipInitialLoad: boolean; + private resubOpts?: ResubOpts; + private resyncIntervalMs?: number; + private subscriber?: + | ReconnectingGrpcProgramAccountSubscriber + | LaserstreamProgramAccountSubscriber; + private resyncTimeoutId?: ReturnType; + private decoded?: boolean; + private grpcConfigs: GrpcConfigs; + private reconnectPromise?: Promise; + private stopped = false; + private reconnectDelayMs = getReconnectDelayMs(); + + constructor({ + grpcConfigs, + orderSubscriber, + skipInitialLoad = false, + resubOpts, + resyncIntervalMs, + decoded = true, + }: { + grpcConfigs: GrpcConfigs; + orderSubscriber: OrderSubscriber; + skipInitialLoad?: boolean; + resubOpts?: ResubOpts; + resyncIntervalMs?: number; + decoded?: boolean; + }) { + this.orderSubscriber = orderSubscriber; + this.skipInitialLoad = skipInitialLoad; + this.resubOpts = resubOpts; + this.resyncIntervalMs = resyncIntervalMs; + this.decoded = decoded; + this.grpcConfigs = grpcConfigs; + } + + public async subscribe(): Promise { + return this.subscribeInternal(true); + } + + private async subscribeInternal(resetStopped: boolean): Promise { + if (resetStopped) { + this.stopped = false; + } + if (this.subscriber) { + return; + } + + const nextSubscriber = await this.createSubscriber(); + this.subscriber = nextSubscriber; + + try { + await nextSubscriber.subscribe( + ( + accountId: PublicKey, + account: UserAccount, + context: Context, + buffer: Buffer + ) => { + const userKey = accountId.toBase58(); + if (this.decoded ?? true) { + this.orderSubscriber.tryUpdateUserAccount( + userKey, + 'decoded', + account, + context.slot + ); + } else { + this.orderSubscriber.tryUpdateUserAccount( + userKey, + 'buffer', + buffer, + context.slot + ); + } + } + ); + + if (!this.skipInitialLoad) { + await this.orderSubscriber.fetch(); + } + + this.startResyncLoop(); + } catch (error) { + this.subscriber = undefined; + try { + await nextSubscriber.unsubscribe(true); + } catch { + // Best effort cleanup before retry. + } + throw error; + } + } + + public async unsubscribe(): Promise { + this.stopped = true; + this.clearResyncLoop(); + + const currentSubscriber = this.subscriber; + this.subscriber = undefined; + if (!currentSubscriber) { + return; + } + + try { + await currentSubscriber.unsubscribe(); + } catch (error) { + logger.warn( + `OrderSubscriber gRPC unsubscribe failed during shutdown: ${formatError( + error + )}` + ); + } + } + + private async createSubscriber() { + if (this.grpcConfigs.client === 'laser') { + return LaserstreamProgramAccountSubscriber.create( + this.grpcConfigs, + 'OrderSubscriber', + 'User', + this.orderSubscriber.driftClient.program, + this.orderSubscriber.decodeFn, + { + filters: [getUserFilter(), getNonIdleUserFilter()], + }, + this.resubOpts + ); + } + + return ReconnectingGrpcProgramAccountSubscriber.create( + this.grpcConfigs, + 'OrderSubscriber', + 'User', + this.orderSubscriber.driftClient.program, + this.orderSubscriber.decodeFn, + { + filters: [getUserFilter(), getNonIdleUserFilter()], + }, + this.resubOpts, + (event, error) => { + void this.handleStreamFailure(event, error); + } + ); + } + + private async handleStreamFailure( + event: StreamFailureEvent, + error?: unknown + ): Promise { + if (this.stopped) { + return; + } + if (this.reconnectPromise) { + return this.reconnectPromise; + } + + const errorSuffix = error + ? `: ${formatError(error)}` + : ''; + logger.warn( + `OrderSubscriber gRPC stream ${event}${errorSuffix}; reconnecting` + ); + + this.reconnectPromise = this.reconnectLoop().finally(() => { + this.reconnectPromise = undefined; + }); + return this.reconnectPromise; + } + + private async reconnectLoop(): Promise { + const currentSubscriber = this.subscriber; + this.subscriber = undefined; + this.clearResyncLoop(); + + if (currentSubscriber) { + try { + await currentSubscriber.unsubscribe(true); + } catch (error) { + logger.warn( + `OrderSubscriber gRPC cleanup failed before reconnect: ${formatError( + error + )}` + ); + } + } + + while (!this.stopped) { + if (this.reconnectDelayMs > 0) { + await sleep(this.reconnectDelayMs); + } + if (this.stopped) { + return; + } + + try { + await this.subscribeInternal(false); + logger.info('OrderSubscriber gRPC reconnect succeeded'); + return; + } catch (error) { + logger.error( + `OrderSubscriber gRPC reconnect failed: ${formatError(error)}` + ); + } + } + } + + private startResyncLoop(): void { + if (!this.resyncIntervalMs) { + return; + } + + this.clearResyncLoop(); + + const recursiveResync = () => { + this.resyncTimeoutId = setTimeout(() => { + if (this.stopped) { + return; + } + + this.orderSubscriber + .fetch() + .catch((error) => { + logger.error( + `Failed to resync OrderSubscriber: ${formatError(error)}` + ); + }) + .finally(() => { + if (!this.resyncTimeoutId || this.stopped) { + return; + } + recursiveResync(); + }); + }, this.resyncIntervalMs); + }; + + recursiveResync(); + } + + private clearResyncLoop(): void { + if (this.resyncTimeoutId !== undefined) { + clearTimeout(this.resyncTimeoutId); + this.resyncTimeoutId = undefined; + } + } +} diff --git a/src/index.ts b/src/index.ts index 4f4943f..976f69e 100644 --- a/src/index.ts +++ b/src/index.ts @@ -24,8 +24,6 @@ import { AssetType, MarketType, } from '@drift-labs/sdk'; -import { RedisClient, RedisClientPrefix } from '@drift/common/clients'; - import { logger, setLogLevel } from './utils/logger'; import * as http from 'http'; @@ -53,6 +51,13 @@ import { HermesClient } from '@pythnetwork/hermes-client'; import { COMMON_UI_UTILS, ENUM_UTILS } from '@drift/common'; import { AuctionParamArgs } from './utils/types'; import { TakerFillVsOracleBpsRedisResult } from './athena/repositories/fillQualityAnalytics'; +import { + RedisClient, + RedisClientPrefix, + HOT_REDIS_KEY_PREFIX, + ALL_REDIS_KEY_PREFIX, + parseRedisClients, +} from './utils/redisClients'; setGlobalDispatcher( new Agent({ @@ -62,16 +67,7 @@ setGlobalDispatcher( require('dotenv').config(); -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]; +const REDIS_CLIENTS = parseRedisClients(process.env.REDIS_CLIENT); console.log('Redis Clients:', REDIS_CLIENTS); const driftEnv = (process.env.ENV || 'devnet') as DriftEnv; @@ -286,7 +282,7 @@ const main = async (): Promise => { .find((client) => { return ( client.forceGetClient().options.keyPrefix === - RedisClientPrefix.DLOB_HELIUS + ALL_REDIS_KEY_PREFIX ); }) .getRaw(`priorityFees_${marketType}_${marketIndex}`); @@ -332,7 +328,7 @@ const main = async (): Promise => { .find( (client) => client.forceGetClient().options.keyPrefix === - RedisClientPrefix.DLOB_HELIUS + ALL_REDIS_KEY_PREFIX ) .getRaw( `priorityFees_${normedParam['marketType']}_${normedParam['marketIndex']}` @@ -523,7 +519,7 @@ const main = async (): Promise => { const redisClient = redisClients.find( (client) => - client.forceGetClient().options.keyPrefix === RedisClientPrefix.DLOB + client.forceGetClient().options.keyPrefix === HOT_REDIS_KEY_PREFIX ); const redisResponseGainers = await redisClient.getRaw( diff --git a/src/publishers/dlobPublisher.ts b/src/publishers/dlobPublisher.ts index cf91ae0..b4a8042 100644 --- a/src/publishers/dlobPublisher.ts +++ b/src/publishers/dlobPublisher.ts @@ -19,8 +19,6 @@ import { ONE, WebSocketAccountSubscriberV2, } from '@drift-labs/sdk'; -import { RedisClient, RedisClientPrefix } from '@drift/common/clients'; - import { logger, setLogLevel } from '../utils/logger'; import { SubscriberLookup, @@ -47,10 +45,28 @@ import { FillQualityAnalyticsRepository, TakerFillVsOracleBpsRedisResult, } from '../athena/repositories/fillQualityAnalytics'; +import { + RedisClient, + HOT_REDIS_CLIENT, + getRedisClientPrefix, +} from '../utils/redisClients'; + +function envInt(name: string, fallback: number): number { + const raw = process.env[name]; + if (raw == null) return fallback; + const parsed = parseInt(raw, 10); + return Number.isFinite(parsed) ? parsed : fallback; +} setGlobalDispatcher( new Agent({ connections: 200, + // Solana RPC calls like getProgramAccounts can legitimately take >30s during + // startup (esp. on self-hosted nodes). Default undici timeouts are too tight + // and cause startup loops (UND_ERR_HEADERS_TIMEOUT). + connectTimeout: envInt('FETCH_CONNECT_TIMEOUT_MS', 10_000), + headersTimeout: envInt('FETCH_HEADERS_TIMEOUT_MS', 180_000), + bodyTimeout: envInt('FETCH_BODY_TIMEOUT_MS', 180_000), }) ); @@ -62,7 +78,8 @@ const metricsPort = process.env.METRICS_PORT ? parseInt(process.env.METRICS_PORT) : 9464; -const REDIS_CLIENT = process.env.REDIS_CLIENT || 'DLOB'; +const REDIS_CLIENT = process.env.REDIS_CLIENT || HOT_REDIS_CLIENT; +const REDIS_PREFIX = getRedisClientPrefix(REDIS_CLIENT); // Set up express for health checks const app = express(); @@ -171,7 +188,11 @@ const ignoreList = process.env.IGNORE_LIST?.split(',') || [ logger.info(`RPC endpoint: ${endpoint}`); logger.info(`WS endpoint: ${wsEndpoint}`); logger.info(`GRPC endpoint: ${grpcEndpoint}`); -logger.info(`GRPC Token: ${token}`); +logger.info( + `GRPC Token: ${ + token ? `${token.slice(0, 4)}…${token.slice(-4)}` : '(missing)' + }` +); logger.info( `useOrderSubscriber: ${useOrderSubscriber}, useWebsocket: ${useWebsocket}, useGrpc: ${useGrpc}` ); @@ -380,7 +401,7 @@ const main = async () => { const clearingHousePublicKey = new PublicKey(sdkConfig.DRIFT_PROGRAM_ID); const redisClient = new RedisClient({ - prefix: RedisClientPrefix[REDIS_CLIENT], + prefix: REDIS_PREFIX, }); await redisClient.connect(); @@ -473,6 +494,9 @@ const main = async () => { perpMarketIndexes: perpMarketInfos.map((m) => m.marketIndex), spotMarketIndexes: spotMarketInfos.map((m) => m.marketIndex), oracleInfos, + // Publisher doesn't need to load/subscribe to any user accounts. + // Avoids expensive RPC lookups that can time out on self-hosted nodes. + skipLoadUsers: true, }); const lamportsBalance = await connection.getBalance(wallet.publicKey); @@ -638,21 +662,21 @@ const main = async () => { marketType: 'perp', marketName: market.marketName, redisClient: REDIS_CLIENT, - redisPrefix: RedisClientPrefix[REDIS_CLIENT], + redisPrefix: REDIS_PREFIX, }); oracleSlotGauge.setLatestValue(oracleDataAndSlot.slot.toNumber(), { marketIndex: market.marketIndex, marketType: 'perp', marketName: market.marketName, redisClient: REDIS_CLIENT, - redisPrefix: RedisClientPrefix[REDIS_CLIENT], + redisPrefix: REDIS_PREFIX, }); marketSlotGauge.setLatestValue(marketAccount.slot, { marketIndex: market.marketIndex, marketType: 'perp', marketName: market.marketName, redisClient: REDIS_CLIENT, - redisPrefix: RedisClientPrefix[REDIS_CLIENT], + redisPrefix: REDIS_PREFIX, }); }); spotMarketInfos.forEach((market) => { @@ -668,21 +692,21 @@ const main = async () => { marketType: 'spot', marketName: market.marketName, redisClient: REDIS_CLIENT, - redisPrefix: RedisClientPrefix[REDIS_CLIENT], + redisPrefix: REDIS_PREFIX, }); oracleSlotGauge.setLatestValue(oracleDataAndSlot.slot.toNumber(), { marketIndex: market.marketIndex, marketType: 'spot', marketName: market.marketName, redisClient: REDIS_CLIENT, - redisPrefix: RedisClientPrefix[REDIS_CLIENT], + redisPrefix: REDIS_PREFIX, }); marketSlotGauge.setLatestValue(marketAccount.slot, { marketIndex: market.marketIndex, marketType: 'spot', marketName: market.marketName, redisClient: REDIS_CLIENT, - redisPrefix: RedisClientPrefix[REDIS_CLIENT], + redisPrefix: REDIS_PREFIX, }); }); }, 10_000); diff --git a/src/publishers/priorityFeesPublisher.ts b/src/publishers/priorityFeesPublisher.ts index 3410b41..b6bddba 100644 --- a/src/publishers/priorityFeesPublisher.ts +++ b/src/publishers/priorityFeesPublisher.ts @@ -10,12 +10,11 @@ import { BulkAccountLoader, getMarketsAndOraclesForSubscription, } from '@drift-labs/sdk'; -import { RedisClient, RedisClientPrefix } from '@drift/common/clients'; - import { logger, setLogLevel } from '../utils/logger'; import { sleep } from '../utils/utils'; import express from 'express'; import { setGlobalDispatcher, Agent } from 'undici'; +import { RedisClient, ALL_REDIS_KEY_PREFIX } from '../utils/redisClients'; setGlobalDispatcher( new Agent({ @@ -27,7 +26,7 @@ require('dotenv').config(); const stateCommitment: Commitment = 'confirmed'; const driftEnv = (process.env.ENV || 'devnet') as DriftEnv; const commitHash = process.env.COMMIT; -const redisClientPrefix = RedisClientPrefix.DLOB_HELIUS; +const redisClientPrefix = ALL_REDIS_KEY_PREFIX; // Set up express for health checks const app = express(); diff --git a/src/publishers/tradesPublisher.ts b/src/publishers/tradesPublisher.ts index ba62d18..9f23f71 100644 --- a/src/publishers/tradesPublisher.ts +++ b/src/publishers/tradesPublisher.ts @@ -20,12 +20,15 @@ import { OrderActionRecord, Event, } from '@drift-labs/sdk'; -import { RedisClient, RedisClientPrefix } from '@drift/common/clients'; - import { logger, setLogLevel } from '../utils/logger'; import { sleep } from '../utils/utils'; import { fromEvent, filter, map } from 'rxjs'; import { setGlobalDispatcher, Agent } from 'undici'; +import { + RedisClient, + HOT_REDIS_CLIENT, + getRedisClientPrefix, +} from '../utils/redisClients'; setGlobalDispatcher( new Agent({ @@ -36,9 +39,9 @@ setGlobalDispatcher( require('dotenv').config(); const driftEnv = (process.env.ENV || 'devnet') as DriftEnv; const commitHash = process.env.COMMIT; -const REDIS_CLIENT = process.env.REDIS_CLIENT || 'DLOB'; +const REDIS_CLIENT = process.env.REDIS_CLIENT || HOT_REDIS_CLIENT; console.log('Redis Clients:', REDIS_CLIENT); -const redisClientPrefix = RedisClientPrefix[REDIS_CLIENT]; +const redisClientPrefix = getRedisClientPrefix(REDIS_CLIENT); //@ts-ignore const sdkConfig = initialize({ env: process.env.ENV }); diff --git a/src/scripts/dlobAllRedisToPostgresDerivedWriter.ts b/src/scripts/dlobAllRedisToPostgresDerivedWriter.ts new file mode 100644 index 0000000..adaa8a2 --- /dev/null +++ b/src/scripts/dlobAllRedisToPostgresDerivedWriter.ts @@ -0,0 +1,284 @@ +import Redis from 'ioredis'; +import { Pool } from 'pg'; +import process from 'node:process'; + +import { + deriveNormalizedRow, + deriveRawSnapshotRow, + envInt, + sleep, +} from './dlobHotPgCommon'; + +const cfg = { + source: String(process.env.DLOB_SOURCE || 'mevnode_bot_all_derived'), + redisHost: String(process.env.REDIS_HOST || process.env.ELASTICACHE_HOST || 'dlob-redis'), + redisPort: envInt('REDIS_PORT', envInt('ELASTICACHE_PORT', 6379)), + redisKeyPrefix: String(process.env.REDIS_KEY_PREFIX || 'dlob-all:'), + pollMs: envInt('DLOB_POLL_MS', 1000), + depthLevels: envInt('NORMALIZED_DEPTH', 10), + pricePrecision: envInt('PRICE_PRECISION', 1_000_000), + basePrecision: envInt('BASE_PRECISION', 1_000_000_000), + pgHost: String(process.env.PGHOST || 'postgres'), + pgPort: envInt('PGPORT', 5432), + pgUser: String(process.env.PGUSER || 'admin'), + pgPassword: String(process.env.PGPASSWORD || ''), + pgDatabase: String(process.env.PGDATABASE || 'crypto'), +}; + +const redis = new Redis({ + host: cfg.redisHost, + port: cfg.redisPort, + maxRetriesPerRequest: null, + enableReadyCheck: false, +}); + +const pool = new Pool({ + host: cfg.pgHost, + port: cfg.pgPort, + user: cfg.pgUser, + password: cfg.pgPassword, + database: cfg.pgDatabase, + max: 4, +}); + +const seen = new Map(); + +const insertDerivedTsSql = ` +INSERT INTO public.dlob_all_derived_ts ( + event_ts, + source, + market_type, + market_index, + market_name, + is_indicative, + ts_ms, + slot, + market_slot, + mark_price, + oracle_price, + best_bid_price, + best_ask_price, + mid_price, + spread_quote, + spread_bps, + depth_levels, + bid_levels, + ask_levels, + top_bid_size, + top_ask_size, + top_bid_notional, + top_ask_notional, + depth_bid_base, + depth_ask_base, + depth_bid_quote, + depth_ask_quote, + imbalance, + bids_norm, + asks_norm, + raw_payload_hash +) VALUES ( + to_timestamp($1::double precision / 1000.0), + $2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25,$26,$27,$28,$29,$30,$31 +) +ON CONFLICT DO NOTHING +`; + +const upsertDerivedLatestSql = ` +INSERT INTO public.dlob_all_derived_latest ( + source, + market_type, + market_index, + market_name, + is_indicative, + ts_ms, + slot, + market_slot, + mark_price, + oracle_price, + best_bid_price, + best_ask_price, + mid_price, + spread_quote, + spread_bps, + depth_levels, + bid_levels, + ask_levels, + top_bid_size, + top_ask_size, + top_bid_notional, + top_ask_notional, + depth_bid_base, + depth_ask_base, + depth_bid_quote, + depth_ask_quote, + imbalance, + bids_norm, + asks_norm, + raw_payload_hash, + updated_at +) VALUES ( + $1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25,$26,$27,$28,$29,$30,now() +) +ON CONFLICT (source, market_type, market_index, is_indicative) +DO UPDATE SET + market_name = EXCLUDED.market_name, + ts_ms = EXCLUDED.ts_ms, + slot = EXCLUDED.slot, + market_slot = EXCLUDED.market_slot, + mark_price = EXCLUDED.mark_price, + oracle_price = EXCLUDED.oracle_price, + best_bid_price = EXCLUDED.best_bid_price, + best_ask_price = EXCLUDED.best_ask_price, + mid_price = EXCLUDED.mid_price, + spread_quote = EXCLUDED.spread_quote, + spread_bps = EXCLUDED.spread_bps, + depth_levels = EXCLUDED.depth_levels, + bid_levels = EXCLUDED.bid_levels, + ask_levels = EXCLUDED.ask_levels, + top_bid_size = EXCLUDED.top_bid_size, + top_ask_size = EXCLUDED.top_ask_size, + top_bid_notional = EXCLUDED.top_bid_notional, + top_ask_notional = EXCLUDED.top_ask_notional, + depth_bid_base = EXCLUDED.depth_bid_base, + depth_ask_base = EXCLUDED.depth_ask_base, + depth_bid_quote = EXCLUDED.depth_bid_quote, + depth_ask_quote = EXCLUDED.depth_ask_quote, + imbalance = EXCLUDED.imbalance, + bids_norm = EXCLUDED.bids_norm, + asks_norm = EXCLUDED.asks_norm, + raw_payload_hash = EXCLUDED.raw_payload_hash, + updated_at = EXCLUDED.updated_at +`; + +async function syncOnce(): Promise { + const keys = (await redis.keys(`${cfg.redisKeyPrefix}*`)).sort(); + if (keys.length === 0) return; + + const client = await pool.connect(); + let writes = 0; + + try { + await client.query('BEGIN'); + for (const key of keys) { + const payloadText = await redis.get(key); + if (!payloadText) continue; + + const raw = deriveRawSnapshotRow({ + source: cfg.source, + redisKey: key, + payloadText, + }); + if (raw == null) continue; + + const normalized = deriveNormalizedRow(raw, { + source: cfg.source, + pricePrecision: cfg.pricePrecision, + basePrecision: cfg.basePrecision, + depthLevels: cfg.depthLevels, + }); + if (normalized == null) continue; + + const token = `${normalized.tsMs}:${normalized.rawPayloadHash}`; + if (seen.get(key) === token) continue; + + await client.query(insertDerivedTsSql, [ + normalized.tsMs, + normalized.source, + normalized.marketType, + normalized.marketIndex, + normalized.marketName, + normalized.isIndicative, + normalized.tsMs, + normalized.slot, + normalized.marketSlot, + normalized.markPrice, + normalized.oraclePrice, + normalized.bestBidPrice, + normalized.bestAskPrice, + normalized.midPrice, + normalized.spreadQuote, + normalized.spreadBps, + normalized.depthLevels, + normalized.bidLevels, + normalized.askLevels, + normalized.topBidSize, + normalized.topAskSize, + normalized.topBidNotional, + normalized.topAskNotional, + normalized.depthBidBase, + normalized.depthAskBase, + normalized.depthBidQuote, + normalized.depthAskQuote, + normalized.imbalance, + JSON.stringify(normalized.bidsNorm), + JSON.stringify(normalized.asksNorm), + normalized.rawPayloadHash, + ]); + + await client.query(upsertDerivedLatestSql, [ + normalized.source, + normalized.marketType, + normalized.marketIndex, + normalized.marketName, + normalized.isIndicative, + normalized.tsMs, + normalized.slot, + normalized.marketSlot, + normalized.markPrice, + normalized.oraclePrice, + normalized.bestBidPrice, + normalized.bestAskPrice, + normalized.midPrice, + normalized.spreadQuote, + normalized.spreadBps, + normalized.depthLevels, + normalized.bidLevels, + normalized.askLevels, + normalized.topBidSize, + normalized.topAskSize, + normalized.topBidNotional, + normalized.topAskNotional, + normalized.depthBidBase, + normalized.depthAskBase, + normalized.depthBidQuote, + normalized.depthAskQuote, + normalized.imbalance, + JSON.stringify(normalized.bidsNorm), + JSON.stringify(normalized.asksNorm), + normalized.rawPayloadHash, + ]); + + seen.set(key, token); + writes += 1; + } + await client.query('COMMIT'); + if (writes > 0) { + console.log( + `[dlob-all-redis-to-postgres-derived-writer] wrote ${writes} normalized snapshot(s) from prefix ${cfg.redisKeyPrefix}` + ); + } + } catch (error) { + await client.query('ROLLBACK'); + console.error('[dlob-all-redis-to-postgres-derived-writer] sync failed', error); + } finally { + client.release(); + } +} + +async function main(): Promise { + console.log( + `[dlob-all-redis-to-postgres-derived-writer] source=${cfg.source} redis=${cfg.redisHost}:${cfg.redisPort} prefix=${cfg.redisKeyPrefix} pg=${cfg.pgHost}:${cfg.pgPort}/${cfg.pgDatabase}` + ); + await redis.ping(); + await pool.query('select 1'); + + for (;;) { + await syncOnce(); + await sleep(cfg.pollMs); + } +} + +main().catch((error) => { + console.error('[dlob-all-redis-to-postgres-derived-writer] fatal', error); + process.exit(1); +}); diff --git a/src/scripts/dlobHotPgCommon.ts b/src/scripts/dlobHotPgCommon.ts new file mode 100644 index 0000000..4dbf803 --- /dev/null +++ b/src/scripts/dlobHotPgCommon.ts @@ -0,0 +1,321 @@ +import crypto from 'node:crypto'; + +export type SnapshotKind = 'orderbook_l2' | 'orderbook_l3' | 'best_makers'; + +export type RawSnapshotRow = { + source: string; + redisKey: string; + snapshotKind: SnapshotKind; + marketType: string; + marketIndex: number | null; + marketName: string; + isIndicative: boolean; + tsMs: number; + slot: number | null; + marketSlot: number | null; + payloadHash: string; + markPriceRaw: string | null; + oraclePriceRaw: string | null; + bestBidPriceRaw: string | null; + bestAskPriceRaw: string | null; + spreadPctRaw: string | null; + spreadQuoteRaw: string | null; + oracleData: unknown | null; + bids: unknown[] | null; + asks: unknown[] | null; + payload: Record; +}; + +export type DerivedSnapshotRow = { + source: string; + marketType: string; + marketIndex: number | null; + marketName: string; + isIndicative: boolean; + tsMs: number; + slot: number | null; + marketSlot: number | null; + markPrice: number | null; + oraclePrice: number | null; + bestBidPrice: number | null; + bestAskPrice: number | null; + midPrice: number | null; + spreadQuote: number | null; + spreadBps: number | null; + depthLevels: number; + bidLevels: number; + askLevels: number; + topBidSize: number | null; + topAskSize: number | null; + topBidNotional: number | null; + topAskNotional: number | null; + depthBidBase: number; + depthAskBase: number; + depthBidQuote: number; + depthAskQuote: number; + imbalance: number | null; + bidsNorm: Array>; + asksNorm: Array>; + rawPayloadHash: string; +}; + +export function envInt(name: string, fallback: number): number { + const raw = process.env[name]; + if (raw == null || raw === '') return fallback; + const parsed = Number.parseInt(raw, 10); + return Number.isFinite(parsed) ? parsed : fallback; +} + +export function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +export function toOptionalInt(value: unknown): number | null { + if (value == null) return null; + const parsed = Number.parseInt(String(value), 10); + return Number.isFinite(parsed) ? parsed : null; +} + +export function toOptionalString(value: unknown): string | null { + if (value == null) return null; + const text = String(value).trim(); + return text.length > 0 ? text : null; +} + +export function toScaledNumber(value: unknown, scale: number): number | null { + const text = toOptionalString(value); + if (text == null) return null; + const parsed = Number(text); + if (!Number.isFinite(parsed)) return null; + return parsed / scale; +} + +export function parseRedisKey(redisKey: string): { + snapshotKind: SnapshotKind; + marketType: string; + marketIndex: number | null; + isIndicative: boolean; +} | null { + const key = redisKey.includes(':') ? redisKey.slice(redisKey.indexOf(':') + 1) : redisKey; + const isIndicative = key.endsWith('_indicative'); + const baseKey = isIndicative ? key.slice(0, -'_indicative'.length) : key; + + let snapshotKind: SnapshotKind; + if (baseKey.includes('best_makers')) { + snapshotKind = 'best_makers'; + } else if (baseKey.includes('orderbook_l3_')) { + snapshotKind = 'orderbook_l3'; + } else if (baseKey.includes('orderbook_')) { + snapshotKind = 'orderbook_l2'; + } else { + return null; + } + + const match = baseKey.match(/_(perp|spot)_(\d+)$/); + if (!match) return null; + + return { + snapshotKind, + marketType: match[1], + marketIndex: Number.parseInt(match[2], 10), + isIndicative, + }; +} + +export function hashPayload(payloadText: string): string { + return crypto.createHash('sha1').update(payloadText).digest('hex'); +} + +export function pickTsMs(payload: Record): number { + const direct = toOptionalInt(payload.ts); + if (direct != null) return direct; + const updated = toOptionalInt(payload.updatedAtTs); + if (updated != null) return updated; + return Date.now(); +} + +export function deriveRawSnapshotRow(input: { + source: string; + redisKey: string; + payloadText: string; +}): RawSnapshotRow | null { + const meta = parseRedisKey(input.redisKey); + if (meta == null) return null; + + let payload: Record; + try { + payload = JSON.parse(input.payloadText); + } catch { + return null; + } + + const bids = Array.isArray(payload.bids) ? payload.bids : []; + const asks = Array.isArray(payload.asks) ? payload.asks : []; + const marketName = + toOptionalString(payload.marketName) ?? + `${meta.marketType.toUpperCase()}-${meta.marketIndex ?? 'unknown'}`; + + return { + source: input.source, + redisKey: input.redisKey, + snapshotKind: meta.snapshotKind, + marketType: toOptionalString(payload.marketType) ?? meta.marketType, + marketIndex: + typeof payload.marketIndex === 'number' + ? payload.marketIndex + : meta.marketIndex, + marketName, + isIndicative: meta.isIndicative, + tsMs: pickTsMs(payload), + slot: toOptionalInt(payload.slot), + marketSlot: toOptionalInt(payload.marketSlot), + payloadHash: hashPayload(input.payloadText), + markPriceRaw: toOptionalString(payload.markPrice), + oraclePriceRaw: toOptionalString( + (payload.oracleData as Record | undefined)?.price ?? payload.oracle + ), + bestBidPriceRaw: toOptionalString(payload.bestBidPrice), + bestAskPriceRaw: toOptionalString(payload.bestAskPrice), + spreadPctRaw: toOptionalString(payload.spreadPct), + spreadQuoteRaw: toOptionalString(payload.spreadQuote), + oracleData: + payload.oracleData && typeof payload.oracleData === 'object' + ? payload.oracleData + : null, + bids, + asks, + payload, + }; +} + +function normalizeLevels( + levels: unknown[], + limit: number, + pricePrecision: number, + basePrecision: number +): Array> { + return levels + .slice(0, limit) + .map((level) => { + const item = level as Record; + const price = toScaledNumber(item.price, pricePrecision); + const size = toScaledNumber(item.size, basePrecision); + if (price == null || size == null) return null; + return { + price, + sizeBase: size, + notional: price * size, + sources: + item.sources && typeof item.sources === 'object' ? item.sources : null, + }; + }) + .filter((item): item is Record => item != null); +} + +export function deriveNormalizedRow( + raw: RawSnapshotRow, + config: { + source: string; + pricePrecision: number; + basePrecision: number; + depthLevels: number; + } +): DerivedSnapshotRow | null { + if (raw.snapshotKind !== 'orderbook_l2') return null; + + const bidsNorm = normalizeLevels( + raw.bids ?? [], + config.depthLevels, + config.pricePrecision, + config.basePrecision + ); + const asksNorm = normalizeLevels( + raw.asks ?? [], + config.depthLevels, + config.pricePrecision, + config.basePrecision + ); + + const markPrice = toScaledNumber(raw.markPriceRaw, config.pricePrecision); + const oraclePrice = toScaledNumber(raw.oraclePriceRaw, config.pricePrecision); + const bestBidPrice = toScaledNumber(raw.bestBidPriceRaw, config.pricePrecision); + const bestAskPrice = toScaledNumber(raw.bestAskPriceRaw, config.pricePrecision); + + const midPrice = + bestBidPrice != null && bestAskPrice != null + ? (bestBidPrice + bestAskPrice) / 2 + : null; + const spreadQuote = + bestBidPrice != null && bestAskPrice != null + ? bestAskPrice - bestBidPrice + : null; + const spreadBps = + spreadQuote != null && midPrice != null && midPrice > 0 + ? (spreadQuote / midPrice) * 10_000 + : null; + + const bidLevels = bidsNorm.length; + const askLevels = asksNorm.length; + + const topBidSize = + bidLevels > 0 ? Number((bidsNorm[0].sizeBase as number | null) ?? null) : null; + const topAskSize = + askLevels > 0 ? Number((asksNorm[0].sizeBase as number | null) ?? null) : null; + const topBidNotional = + bidLevels > 0 ? Number((bidsNorm[0].notional as number | null) ?? null) : null; + const topAskNotional = + askLevels > 0 ? Number((asksNorm[0].notional as number | null) ?? null) : null; + + const depthBidBase = bidsNorm.reduce( + (sum, level) => sum + Number(level.sizeBase ?? 0), + 0 + ); + const depthAskBase = asksNorm.reduce( + (sum, level) => sum + Number(level.sizeBase ?? 0), + 0 + ); + const depthBidQuote = bidsNorm.reduce( + (sum, level) => sum + Number(level.notional ?? 0), + 0 + ); + const depthAskQuote = asksNorm.reduce( + (sum, level) => sum + Number(level.notional ?? 0), + 0 + ); + const denom = depthBidQuote + depthAskQuote; + const imbalance = + denom > 0 ? (depthBidQuote - depthAskQuote) / denom : null; + + return { + source: config.source, + marketType: raw.marketType, + marketIndex: raw.marketIndex, + marketName: raw.marketName, + isIndicative: raw.isIndicative, + tsMs: raw.tsMs, + slot: raw.slot, + marketSlot: raw.marketSlot, + markPrice, + oraclePrice, + bestBidPrice, + bestAskPrice, + midPrice, + spreadQuote, + spreadBps, + depthLevels: config.depthLevels, + bidLevels, + askLevels, + topBidSize, + topAskSize, + topBidNotional, + topAskNotional, + depthBidBase, + depthAskBase, + depthBidQuote, + depthAskQuote, + imbalance, + bidsNorm, + asksNorm, + rawPayloadHash: raw.payloadHash, + }; +} diff --git a/src/scripts/dlobHotPostgresToPostgresDerivedWriter.ts b/src/scripts/dlobHotPostgresToPostgresDerivedWriter.ts new file mode 100644 index 0000000..9d71d52 --- /dev/null +++ b/src/scripts/dlobHotPostgresToPostgresDerivedWriter.ts @@ -0,0 +1,343 @@ +import { Pool } from 'pg'; +import process from 'node:process'; + +import { + deriveNormalizedRow, + envInt, + sleep, + type RawSnapshotRow, +} from './dlobHotPgCommon'; + +const cfg = { + source: String(process.env.DLOB_SOURCE || 'mevnode_bot_hot_derived'), + rawSource: String(process.env.RAW_SOURCE || 'mevnode_bot_hot_raw'), + pollMs: envInt('DLOB_POLL_MS', 5000), + depthLevels: envInt('NORMALIZED_DEPTH', 10), + pricePrecision: envInt('PRICE_PRECISION', 1_000_000), + basePrecision: envInt('BASE_PRECISION', 1_000_000_000), + pgHost: String(process.env.PGHOST || 'postgres'), + pgPort: envInt('PGPORT', 5432), + pgUser: String(process.env.PGUSER || 'admin'), + pgPassword: String(process.env.PGPASSWORD || ''), + pgDatabase: String(process.env.PGDATABASE || 'crypto'), +}; + +const pool = new Pool({ + host: cfg.pgHost, + port: cfg.pgPort, + user: cfg.pgUser, + password: cfg.pgPassword, + database: cfg.pgDatabase, + max: 4, +}); + +const seen = new Map(); +let syncQueue: Promise = Promise.resolve(); + +const selectRawLatestSql = ` +SELECT + source, + redis_key, + snapshot_kind, + market_type, + market_index, + market_name, + is_indicative, + ts_ms, + slot, + market_slot, + payload_hash, + mark_price_raw, + oracle_price_raw, + best_bid_price_raw, + best_ask_price_raw, + spread_pct_raw, + spread_quote_raw, + oracle_data, + bids, + asks, + payload +FROM public.dlob_hot_snapshot_latest +WHERE source = $1 + AND snapshot_kind = 'orderbook_l2' +ORDER BY market_index NULLS LAST, is_indicative ASC +`; + +const insertDerivedTsSql = ` +INSERT INTO public.dlob_hot_derived_ts ( + event_ts, + source, + market_type, + market_index, + market_name, + is_indicative, + ts_ms, + slot, + market_slot, + mark_price, + oracle_price, + best_bid_price, + best_ask_price, + mid_price, + spread_quote, + spread_bps, + depth_levels, + bid_levels, + ask_levels, + top_bid_size, + top_ask_size, + top_bid_notional, + top_ask_notional, + depth_bid_base, + depth_ask_base, + depth_bid_quote, + depth_ask_quote, + imbalance, + bids_norm, + asks_norm, + raw_payload_hash +) VALUES ( + to_timestamp($1::double precision / 1000.0), + $2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25,$26,$27,$28,$29,$30,$31 +) +ON CONFLICT DO NOTHING +`; + +const upsertDerivedLatestSql = ` +INSERT INTO public.dlob_hot_derived_latest ( + source, + market_type, + market_index, + market_name, + is_indicative, + ts_ms, + slot, + market_slot, + mark_price, + oracle_price, + best_bid_price, + best_ask_price, + mid_price, + spread_quote, + spread_bps, + depth_levels, + bid_levels, + ask_levels, + top_bid_size, + top_ask_size, + top_bid_notional, + top_ask_notional, + depth_bid_base, + depth_ask_base, + depth_bid_quote, + depth_ask_quote, + imbalance, + bids_norm, + asks_norm, + raw_payload_hash, + updated_at +) VALUES ( + $1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25,$26,$27,$28,$29,$30,now() +) +ON CONFLICT (source, market_type, market_index, is_indicative) +DO UPDATE SET + market_name = EXCLUDED.market_name, + ts_ms = EXCLUDED.ts_ms, + slot = EXCLUDED.slot, + market_slot = EXCLUDED.market_slot, + mark_price = EXCLUDED.mark_price, + oracle_price = EXCLUDED.oracle_price, + best_bid_price = EXCLUDED.best_bid_price, + best_ask_price = EXCLUDED.best_ask_price, + mid_price = EXCLUDED.mid_price, + spread_quote = EXCLUDED.spread_quote, + spread_bps = EXCLUDED.spread_bps, + depth_levels = EXCLUDED.depth_levels, + bid_levels = EXCLUDED.bid_levels, + ask_levels = EXCLUDED.ask_levels, + top_bid_size = EXCLUDED.top_bid_size, + top_ask_size = EXCLUDED.top_ask_size, + top_bid_notional = EXCLUDED.top_bid_notional, + top_ask_notional = EXCLUDED.top_ask_notional, + depth_bid_base = EXCLUDED.depth_bid_base, + depth_ask_base = EXCLUDED.depth_ask_base, + depth_bid_quote = EXCLUDED.depth_bid_quote, + depth_ask_quote = EXCLUDED.depth_ask_quote, + imbalance = EXCLUDED.imbalance, + bids_norm = EXCLUDED.bids_norm, + asks_norm = EXCLUDED.asks_norm, + raw_payload_hash = EXCLUDED.raw_payload_hash, + updated_at = EXCLUDED.updated_at +`; + +function rowToRawSnapshot(row: Record): RawSnapshotRow { + return { + source: String(row.source), + redisKey: String(row.redis_key), + snapshotKind: 'orderbook_l2', + marketType: String(row.market_type), + marketIndex: + typeof row.market_index === 'number' ? row.market_index : null, + marketName: String(row.market_name), + isIndicative: Boolean(row.is_indicative), + tsMs: Number(row.ts_ms), + slot: typeof row.slot === 'number' ? row.slot : null, + marketSlot: typeof row.market_slot === 'number' ? row.market_slot : null, + payloadHash: String(row.payload_hash), + markPriceRaw: row.mark_price_raw ? String(row.mark_price_raw) : null, + oraclePriceRaw: row.oracle_price_raw ? String(row.oracle_price_raw) : null, + bestBidPriceRaw: row.best_bid_price_raw ? String(row.best_bid_price_raw) : null, + bestAskPriceRaw: row.best_ask_price_raw ? String(row.best_ask_price_raw) : null, + spreadPctRaw: row.spread_pct_raw ? String(row.spread_pct_raw) : null, + spreadQuoteRaw: row.spread_quote_raw ? String(row.spread_quote_raw) : null, + oracleData: row.oracle_data ?? null, + bids: Array.isArray(row.bids) ? row.bids : null, + asks: Array.isArray(row.asks) ? row.asks : null, + payload: (row.payload as Record) ?? {}, + }; +} + +async function syncOnce(): Promise { + const client = await pool.connect(); + let writes = 0; + + try { + const { rows } = await client.query(selectRawLatestSql, [cfg.rawSource]); + await client.query('BEGIN'); + + for (const dbRow of rows as Array>) { + const raw = rowToRawSnapshot(dbRow); + const normalized = deriveNormalizedRow(raw, { + source: cfg.source, + pricePrecision: cfg.pricePrecision, + basePrecision: cfg.basePrecision, + depthLevels: cfg.depthLevels, + }); + if (normalized == null) continue; + + const token = `${normalized.tsMs}:${normalized.rawPayloadHash}`; + const dedupeKey = `${normalized.marketType}:${normalized.marketIndex}:${normalized.isIndicative}`; + if (seen.get(dedupeKey) === token) continue; + + await client.query(insertDerivedTsSql, [ + normalized.tsMs, + normalized.source, + normalized.marketType, + normalized.marketIndex, + normalized.marketName, + normalized.isIndicative, + normalized.tsMs, + normalized.slot, + normalized.marketSlot, + normalized.markPrice, + normalized.oraclePrice, + normalized.bestBidPrice, + normalized.bestAskPrice, + normalized.midPrice, + normalized.spreadQuote, + normalized.spreadBps, + normalized.depthLevels, + normalized.bidLevels, + normalized.askLevels, + normalized.topBidSize, + normalized.topAskSize, + normalized.topBidNotional, + normalized.topAskNotional, + normalized.depthBidBase, + normalized.depthAskBase, + normalized.depthBidQuote, + normalized.depthAskQuote, + normalized.imbalance, + JSON.stringify(normalized.bidsNorm), + JSON.stringify(normalized.asksNorm), + normalized.rawPayloadHash, + ]); + + await client.query(upsertDerivedLatestSql, [ + normalized.source, + normalized.marketType, + normalized.marketIndex, + normalized.marketName, + normalized.isIndicative, + normalized.tsMs, + normalized.slot, + normalized.marketSlot, + normalized.markPrice, + normalized.oraclePrice, + normalized.bestBidPrice, + normalized.bestAskPrice, + normalized.midPrice, + normalized.spreadQuote, + normalized.spreadBps, + normalized.depthLevels, + normalized.bidLevels, + normalized.askLevels, + normalized.topBidSize, + normalized.topAskSize, + normalized.topBidNotional, + normalized.topAskNotional, + normalized.depthBidBase, + normalized.depthAskBase, + normalized.depthBidQuote, + normalized.depthAskQuote, + normalized.imbalance, + JSON.stringify(normalized.bidsNorm), + JSON.stringify(normalized.asksNorm), + normalized.rawPayloadHash, + ]); + + seen.set(dedupeKey, token); + writes += 1; + } + + await client.query('COMMIT'); + if (writes > 0) { + console.log( + `[dlob-hot-postgres-to-postgres-derived-writer] wrote ${writes} normalized snapshot(s) from raw source ${cfg.rawSource}` + ); + } + } catch (error) { + await client.query('ROLLBACK'); + console.error('[dlob-hot-postgres-to-postgres-derived-writer] sync failed', error); + } finally { + client.release(); + } +} + +function enqueueSync(reason: string): void { + syncQueue = syncQueue + .then(async () => { + await syncOnce(); + }) + .catch((error) => { + console.error( + `[dlob-hot-postgres-to-postgres-derived-writer] queued sync failed (${reason})`, + error + ); + }); +} + +async function main(): Promise { + console.log( + `[dlob-hot-postgres-to-postgres-derived-writer] source=${cfg.source} rawSource=${cfg.rawSource} pg=${cfg.pgHost}:${cfg.pgPort}/${cfg.pgDatabase}` + ); + await pool.query('select 1'); + const listener = await pool.connect(); + await listener.query('LISTEN dlob_hot_raw_updates'); + listener.on('notification', () => { + enqueueSync('notify:dlob_hot_raw_updates'); + }); + + await syncOnce(); + + for (;;) { + await sleep(cfg.pollMs); + enqueueSync('fallback-poll'); + await syncQueue; + } +} + +main().catch((error) => { + console.error('[dlob-hot-postgres-to-postgres-derived-writer] fatal', error); + process.exit(1); +}); diff --git a/src/scripts/dlobHotRedisToPostgresRawWriter.ts b/src/scripts/dlobHotRedisToPostgresRawWriter.ts new file mode 100644 index 0000000..3ed17cf --- /dev/null +++ b/src/scripts/dlobHotRedisToPostgresRawWriter.ts @@ -0,0 +1,280 @@ +import Redis from 'ioredis'; +import { Pool } from 'pg'; +import process from 'node:process'; + +import { + deriveRawSnapshotRow, + envInt, + sleep, +} from './dlobHotPgCommon'; + +const cfg = { + source: String(process.env.DLOB_SOURCE || 'mevnode_bot_hot_raw'), + redisHost: String(process.env.REDIS_HOST || process.env.ELASTICACHE_HOST || 'dlob-redis'), + redisPort: envInt('REDIS_PORT', envInt('ELASTICACHE_PORT', 6379)), + redisKeyPrefix: String(process.env.REDIS_KEY_PREFIX || 'dlob-hot:'), + pollMs: envInt('DLOB_POLL_MS', 5000), + pgHost: String(process.env.PGHOST || 'postgres'), + pgPort: envInt('PGPORT', 5432), + pgUser: String(process.env.PGUSER || 'admin'), + pgPassword: String(process.env.PGPASSWORD || ''), + pgDatabase: String(process.env.PGDATABASE || 'crypto'), +}; + +const redis = new Redis({ + host: cfg.redisHost, + port: cfg.redisPort, + maxRetriesPerRequest: null, + enableReadyCheck: false, +}); + +const pool = new Pool({ + host: cfg.pgHost, + port: cfg.pgPort, + user: cfg.pgUser, + password: cfg.pgPassword, + database: cfg.pgDatabase, + max: 4, +}); +const subscriber = new Redis({ + host: cfg.redisHost, + port: cfg.redisPort, + maxRetriesPerRequest: null, + enableReadyCheck: false, +}); + +const seen = new Map(); +let writeQueue: Promise = Promise.resolve(); + +const insertRawTsSql = ` +INSERT INTO public.dlob_hot_snapshot_ts ( + event_ts, + source, + redis_key, + snapshot_kind, + market_type, + market_index, + market_name, + is_indicative, + ts_ms, + slot, + market_slot, + payload_hash, + mark_price_raw, + oracle_price_raw, + best_bid_price_raw, + best_ask_price_raw, + spread_pct_raw, + spread_quote_raw, + oracle_data, + bids, + asks, + payload +) VALUES ( + to_timestamp($1::double precision / 1000.0), + $2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22 +) +ON CONFLICT DO NOTHING +`; + +const upsertRawLatestSql = ` +INSERT INTO public.dlob_hot_snapshot_latest ( + source, + redis_key, + snapshot_kind, + market_type, + market_index, + market_name, + is_indicative, + ts_ms, + slot, + market_slot, + payload_hash, + mark_price_raw, + oracle_price_raw, + best_bid_price_raw, + best_ask_price_raw, + spread_pct_raw, + spread_quote_raw, + oracle_data, + bids, + asks, + payload, + updated_at +) VALUES ( + $1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,now() +) +ON CONFLICT (source, market_type, market_index, snapshot_kind, is_indicative) +DO UPDATE SET + redis_key = EXCLUDED.redis_key, + market_name = EXCLUDED.market_name, + ts_ms = EXCLUDED.ts_ms, + slot = EXCLUDED.slot, + market_slot = EXCLUDED.market_slot, + payload_hash = EXCLUDED.payload_hash, + mark_price_raw = EXCLUDED.mark_price_raw, + oracle_price_raw = EXCLUDED.oracle_price_raw, + best_bid_price_raw = EXCLUDED.best_bid_price_raw, + best_ask_price_raw = EXCLUDED.best_ask_price_raw, + spread_pct_raw = EXCLUDED.spread_pct_raw, + spread_quote_raw = EXCLUDED.spread_quote_raw, + oracle_data = EXCLUDED.oracle_data, + bids = EXCLUDED.bids, + asks = EXCLUDED.asks, + payload = EXCLUDED.payload, + updated_at = EXCLUDED.updated_at +`; + +function canonicalRedisKey(redisKey: string): string { + if (!redisKey.startsWith(cfg.redisKeyPrefix)) return redisKey; + const suffix = redisKey.slice(cfg.redisKeyPrefix.length); + if (suffix.startsWith('last_update_')) return redisKey; + return `${cfg.redisKeyPrefix}last_update_${suffix}`; +} + +async function persistRows(rows: Array>, reason: string): Promise { + const filteredRows = rows.filter((row): row is NonNullable => row != null); + if (filteredRows.length === 0) return; + const client = await pool.connect(); + let writes = 0; + + try { + await client.query('BEGIN'); + for (const row of filteredRows) { + const token = `${row.tsMs}:${row.payloadHash}`; + if (seen.get(row.redisKey) === token) continue; + + await client.query(insertRawTsSql, [ + row.tsMs, + row.source, + row.redisKey, + row.snapshotKind, + row.marketType, + row.marketIndex, + row.marketName, + row.isIndicative, + row.tsMs, + row.slot, + row.marketSlot, + row.payloadHash, + row.markPriceRaw, + row.oraclePriceRaw, + row.bestBidPriceRaw, + row.bestAskPriceRaw, + row.spreadPctRaw, + row.spreadQuoteRaw, + row.oracleData ? JSON.stringify(row.oracleData) : null, + row.bids ? JSON.stringify(row.bids) : null, + row.asks ? JSON.stringify(row.asks) : null, + JSON.stringify(row.payload), + ]); + + await client.query(upsertRawLatestSql, [ + row.source, + row.redisKey, + row.snapshotKind, + row.marketType, + row.marketIndex, + row.marketName, + row.isIndicative, + row.tsMs, + row.slot, + row.marketSlot, + row.payloadHash, + row.markPriceRaw, + row.oraclePriceRaw, + row.bestBidPriceRaw, + row.bestAskPriceRaw, + row.spreadPctRaw, + row.spreadQuoteRaw, + row.oracleData ? JSON.stringify(row.oracleData) : null, + row.bids ? JSON.stringify(row.bids) : null, + row.asks ? JSON.stringify(row.asks) : null, + JSON.stringify(row.payload), + ]); + + seen.set(row.redisKey, token); + writes += 1; + } + if (writes > 0) { + await client.query(`select pg_notify('dlob_hot_raw_updates', $1)`, [ + String(Date.now()), + ]); + } + await client.query('COMMIT'); + if (writes > 0) { + console.log( + `[dlob-hot-redis-to-postgres-raw-writer] wrote ${writes} snapshot(s) via ${reason}` + ); + } + } catch (error) { + await client.query('ROLLBACK'); + console.error('[dlob-hot-redis-to-postgres-raw-writer] sync failed', error); + } finally { + client.release(); + } +} + +async function syncOnce(): Promise { + const keys = (await redis.keys(`${cfg.redisKeyPrefix}*`)).sort(); + if (keys.length === 0) return; + + const rows: Array> = []; + for (const key of keys) { + const payloadText = await redis.get(key); + if (!payloadText) continue; + rows.push( + deriveRawSnapshotRow({ + source: cfg.source, + redisKey: key, + payloadText, + }) + ); + } + + await persistRows(rows, `full-sync:${cfg.redisKeyPrefix}`); +} + +function enqueueWrite(task: () => Promise): void { + writeQueue = writeQueue + .then(task) + .catch((error) => { + console.error('[dlob-hot-redis-to-postgres-raw-writer] queued task failed', error); + }); +} + +function handlePublishedSnapshot(channel: string, payloadText: string): void { + enqueueWrite(async () => { + const row = deriveRawSnapshotRow({ + source: cfg.source, + redisKey: canonicalRedisKey(channel), + payloadText, + }); + await persistRows([row], `redis-publish:${channel}`); + }); +} + +async function main(): Promise { + console.log( + `[dlob-hot-redis-to-postgres-raw-writer] source=${cfg.source} redis=${cfg.redisHost}:${cfg.redisPort} prefix=${cfg.redisKeyPrefix} pg=${cfg.pgHost}:${cfg.pgPort}/${cfg.pgDatabase}` + ); + await redis.ping(); + await pool.query('select 1'); + await subscriber.psubscribe(`${cfg.redisKeyPrefix}orderbook_*`); + subscriber.on('pmessage', (_pattern, channel, message) => { + handlePublishedSnapshot(channel, message); + }); + + await syncOnce(); + + for (;;) { + await sleep(cfg.pollMs); + enqueueWrite(syncOnce); + await writeQueue; + } +} + +main().catch((error) => { + console.error('[dlob-hot-redis-to-postgres-raw-writer] fatal', error); + process.exit(1); +}); diff --git a/src/scripts/publishUnsettledPnlUsers.ts b/src/scripts/publishUnsettledPnlUsers.ts index 5b9c714..4f15467 100644 --- a/src/scripts/publishUnsettledPnlUsers.ts +++ b/src/scripts/publishUnsettledPnlUsers.ts @@ -1,4 +1,3 @@ -import { RedisClient, RedisClientPrefix } from '@drift/common/clients'; import { sleep } from '@drift/common'; import { BigNum, @@ -17,6 +16,11 @@ import { import { Connection, Keypair } from '@solana/web3.js'; import { logger } from '../utils/logger'; import Bottleneck from 'bottleneck'; +import { + RedisClient, + RedisClientPrefix, + HOT_REDIS_KEY_PREFIX, +} from '../utils/redisClients'; const dotenv = require('dotenv'); dotenv.config(); @@ -61,7 +65,7 @@ const userMapRedisClient = new RedisClient({ }); const dlobRedisClient = new RedisClient({ - prefix: RedisClientPrefix.DLOB, + prefix: HOT_REDIS_KEY_PREFIX, }); const main = async () => { diff --git a/src/serverLite.ts b/src/serverLite.ts index b25e80a..3c6a550 100644 --- a/src/serverLite.ts +++ b/src/serverLite.ts @@ -12,9 +12,8 @@ import { initialize, MarketType, getVariant, + isVariant, } from '@drift-labs/sdk'; -import { RedisClient, RedisClientPrefix } from '@drift/common/clients'; - import { logger, setLogLevel } from './utils/logger'; import * as http from 'http'; @@ -22,6 +21,7 @@ import { handleHealthCheck } from './core/middleware'; import { errorHandler, selectMostRecentBySlot, sleep } from './utils/utils'; import { setGlobalDispatcher, Agent } from 'undici'; import { Metrics } from './core/metricsV2'; +import { RedisClient, parseRedisClients } from './utils/redisClients'; setGlobalDispatcher( new Agent({ @@ -32,16 +32,7 @@ setGlobalDispatcher( require('dotenv').config(); // Reading in Redis env vars -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]; +const REDIS_CLIENTS = parseRedisClients(process.env.REDIS_CLIENT); console.log('Redis Clients:', REDIS_CLIENTS); const driftEnv = (process.env.ENV || 'devnet') as DriftEnv; @@ -203,6 +194,112 @@ const main = async (): Promise => { } }); + app.get('/l2', async (req, res, next) => { + try { + const { marketName, marketIndex, marketType, depth, includeIndicative } = + req.query; + + let normedMarketType: MarketType | undefined; + let normedMarketIndex: number | undefined; + + if (marketName != null && String(marketName).trim() !== '') { + const name = String(marketName).toUpperCase().trim(); + const perp = sdkConfig?.PERP_MARKETS?.find((m) => m?.symbol === name); + if (perp) { + normedMarketType = MarketType.PERP; + normedMarketIndex = perp.marketIndex; + } else { + const spot = sdkConfig?.SPOT_MARKETS?.find((m) => m?.symbol === name); + if (spot) { + normedMarketType = MarketType.SPOT; + normedMarketIndex = spot.marketIndex; + } + } + + if (normedMarketType == null || normedMarketIndex == null) { + res.status(400).send('Bad Request: unrecognized marketName'); + return; + } + } else { + if (marketIndex == null || marketType == null) { + res + .status(400) + .send( + 'Bad Request: (marketName) or (marketIndex and marketType) must be supplied' + ); + return; + } + + const mt = String(marketType).toLowerCase().trim(); + const idx = parseInt(String(marketIndex), 10); + if (!Number.isFinite(idx) || idx < 0) { + res.status(400).send('Bad Request: invalid marketIndex'); + return; + } + + if (mt === 'perp') { + if (!sdkConfig?.PERP_MARKETS?.[idx]) { + res.status(400).send('Bad Request: invalid marketIndex'); + return; + } + normedMarketType = MarketType.PERP; + normedMarketIndex = idx; + } else if (mt === 'spot') { + if (!sdkConfig?.SPOT_MARKETS?.[idx]) { + res.status(400).send('Bad Request: invalid marketIndex'); + return; + } + normedMarketType = MarketType.SPOT; + normedMarketIndex = idx; + } else { + res + .status(400) + .send('Bad Request: marketType must be either "spot" or "perp"'); + return; + } + } + + const includeIndicativeStr = + (includeIndicative as string)?.toLowerCase() === 'true'; + const depthToUse = Math.min(parseInt(String(depth ?? '100'), 10) || 1, 100); + const indicativeSuffix = includeIndicativeStr ? '_indicative' : ''; + + const redisL2 = await fetchFromRedis( + `last_update_orderbook_${getVariant(normedMarketType)}_${normedMarketIndex}${indicativeSuffix}`, + selectMostRecentBySlot + ); + + if ( + redisL2 && + slotSubscriber.getSlot() - redisL2['slot'] < SLOT_STALENESS_TOLERANCE + ) { + redisL2['bids'] = redisL2['bids']?.slice(0, depthToUse); + redisL2['asks'] = redisL2['asks']?.slice(0, depthToUse); + cacheHitCounter.add(1, { + miss: false, + path: req.baseUrl + req.path, + marketIndex: normedMarketIndex, + marketType: isVariant(normedMarketType, 'spot') ? 'spot' : 'perp', + }); + res.writeHead(200); + res.end(JSON.stringify(redisL2)); + return; + } else { + cacheHitCounter.add(1, { + miss: true, + path: req.baseUrl + req.path, + marketIndex: normedMarketIndex, + marketType: isVariant(normedMarketType, 'spot') ? 'spot' : 'perp', + }); + res.writeHead(500); + res.end(JSON.stringify({ error: 'No cached L2 found' })); + return; + } + } catch (err) { + next(err); + } + }); + server.listen(serverPort, () => { logger.info( `DLOB server lite listening on port http://localhost:${serverPort}` diff --git a/src/utils/redisClients.ts b/src/utils/redisClients.ts new file mode 100644 index 0000000..06ab08b --- /dev/null +++ b/src/utils/redisClients.ts @@ -0,0 +1,47 @@ +import { + RedisClient, + RedisClientPrefix as DriftRedisClientPrefix, +} from '@drift/common/clients'; + +export { RedisClient }; + +export const HOT_REDIS_CLIENT = 'DLOB_HOT'; +export const ALL_REDIS_CLIENT = 'DLOB_ALL'; + +export const HOT_REDIS_KEY_PREFIX = 'dlob-hot:'; +export const ALL_REDIS_KEY_PREFIX = 'dlob-all:'; + +export const RedisClientPrefix = { + USER_MAP: DriftRedisClientPrefix.USER_MAP, + DLOB_HOT: HOT_REDIS_KEY_PREFIX, + DLOB_ALL: ALL_REDIS_KEY_PREFIX, +} as const; + +export function getRedisClientPrefix(clientName?: string): string { + switch ((clientName || HOT_REDIS_CLIENT).trim()) { + case 'DLOB': + case HOT_REDIS_CLIENT: + return HOT_REDIS_KEY_PREFIX; + case 'DLOB_HELIUS': + case ALL_REDIS_CLIENT: + return ALL_REDIS_KEY_PREFIX; + case 'USER_MAP': + return RedisClientPrefix.USER_MAP; + default: + throw new Error(`Unknown REDIS_CLIENT: ${clientName}`); + } +} + +export function parseRedisClients(clientNames?: string): string[] { + const prefixes = + clientNames + ?.trim() + .replace(/^\[|\]$/g, '') + .split(/\s*,\s*/) + .filter(Boolean) + .map((client) => getRedisClientPrefix(client)) ?? []; + + return prefixes.length + ? prefixes + : [HOT_REDIS_KEY_PREFIX, ALL_REDIS_KEY_PREFIX]; +} diff --git a/src/wsConnectionManager.ts b/src/wsConnectionManager.ts index 7bf2172..64201d2 100644 --- a/src/wsConnectionManager.ts +++ b/src/wsConnectionManager.ts @@ -5,8 +5,12 @@ import compression from 'compression'; import { WebSocket, WebSocketServer } from 'ws'; import { sleep, selectMostRecentBySlot, GROUPING_OPTIONS } from './utils/utils'; import { DriftEnv, PerpMarkets, SpotMarkets } from '@drift-labs/sdk'; -import { RedisClient, RedisClientPrefix } from '@drift/common/clients'; import { Metrics, GaugeValue } from './core/metricsV2'; +import { + RedisClient, + ALL_REDIS_KEY_PREFIX, + parseRedisClients, +} from './utils/redisClients'; // Stream selector with hysteresis to prevent flip-flopping between sources const STREAM_SWITCH_THRESHOLD_MS = 10_000; // 10 seconds @@ -211,16 +215,7 @@ console.log(`WS LISTENER PORT : ${WS_PORT}`); const MAX_BUFFERED_AMOUNT = 300000; -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]; +const REDIS_CLIENTS = parseRedisClients(process.env.REDIS_CLIENT); console.log('Redis Clients:', REDIS_CLIENTS); const regexp = new RegExp(REDIS_CLIENTS.join('|'), 'g'); @@ -438,7 +433,7 @@ async function main() { redisClients[0] .subscribe( redisChannel.includes('priorityFees') - ? `dlob-helius:${redisChannel}` + ? `${ALL_REDIS_KEY_PREFIX}${redisChannel}` : `${redisClients[0].getPrefix()}${redisChannel}` ) .then(() => {