diff --git a/src/dlob-subscriber/DLOBSubscriberIO.ts b/src/dlob-subscriber/DLOBSubscriberIO.ts index c598b52..9075139 100644 --- a/src/dlob-subscriber/DLOBSubscriberIO.ts +++ b/src/dlob-subscriber/DLOBSubscriberIO.ts @@ -78,7 +78,7 @@ export class DLOBSubscriberIO extends DLOBSubscriber { env: DriftEnv; redisClient: RedisClient; indicativeQuotesRedisClient?: RedisClient; - enableOffloadQueue?: boolean + enableOffloadQueue?: boolean; perpMarketInfos: wsMarketInfo[]; spotMarketInfos: wsMarketInfo[]; spotMarketSubscribers: SubscriberLookup; @@ -404,11 +404,13 @@ export class DLOBSubscriberIO extends DLOBSubscriber { if (this.offloadQueue) { try { - this.offloadQueue.addMessage(Object.assign({}, l2Formatted, { - bids: l2Formatted.bids.slice(0, 20), - asks: l2Formatted.asks.slice(0, 20), - ts: Math.floor(new Date().getTime() / 1000), - })); + this.offloadQueue.addMessage( + Object.assign({}, l2Formatted, { + bids: l2Formatted.bids.slice(0, 20), + asks: l2Formatted.asks.slice(0, 20), + ts: Math.floor(new Date().getTime() / 1000), + }) + ); } catch (error) { logger.error('Error adding message to offload queue:', error); } @@ -450,6 +452,7 @@ export class DLOBSubscriberIO extends DLOBSubscriber { } getL3AndSendMsg(marketArgs: wsMarketArgs): void { + const shouldSendMessage = Math.random() < 1 / 60; const { marketName, ...l2FuncArgs } = marketArgs; const l3 = this.getL3(l2FuncArgs); const slot = l3.slot; @@ -498,6 +501,19 @@ export class DLOBSubscriberIO extends DLOBSubscriber { marketArgs.marketIndex ); + if (shouldSendMessage && this.offloadQueue) { + try { + this.offloadQueue.addMessage( + Object.assign({}, l3, { + ts: Math.floor(new Date().getTime() / 1000), + }), + 'DLOBL3Snapshot' + ); + } catch (error) { + logger.error('Error adding message to offload queue:', error); + } + } + this.redisClient.set( `last_update_orderbook_l3_${marketType}_${marketArgs.marketIndex}${ this.indicativeQuotesRedisClient ? '_indicative' : '' diff --git a/src/utils/offload.ts b/src/utils/offload.ts index 36ab36c..aeb0750 100644 --- a/src/utils/offload.ts +++ b/src/utils/offload.ts @@ -6,8 +6,11 @@ import { import { ConfiguredRetryStrategy } from '@aws-sdk/util-retry'; import { logger } from '../utils/logger'; +type EventType = 'DLOBSnapshot' | 'DLOBL3Snapshot'; + interface QueueMessage { data: any; + eventType: EventType; } const kinesis = new KinesisClient({ @@ -17,7 +20,7 @@ const kinesis = new KinesisClient({ ), }); -const kinesisStream = process.env.KINESIS_STREAM_NAME +const kinesisStream = process.env.KINESIS_STREAM_NAME; const batchInterval = process.env.KINESIS_BATCH_INTERVAL ? parseInt(process.env.KINESIS_BATCH_INTERVAL) : 10000; @@ -47,7 +50,6 @@ export const OffloadQueue = () => { JSON.stringify({ ...msg.data, source: 'dlob', - eventType: 'DLOBSnapshot', }) ), PartitionKey: Date.now().toString(), @@ -87,9 +89,10 @@ export const OffloadQueue = () => { } }; - const addMessage = (data: any): void => { + const addMessage = (data: any, eventType: EventType = 'DLOBSnapshot') => { queue.push({ data, + eventType, }); };