diff --git a/src/dlob-subscriber/DLOBSubscriberIO.ts b/src/dlob-subscriber/DLOBSubscriberIO.ts index 9075139..0f7f52e 100644 --- a/src/dlob-subscriber/DLOBSubscriberIO.ts +++ b/src/dlob-subscriber/DLOBSubscriberIO.ts @@ -452,7 +452,6 @@ 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; @@ -501,13 +500,13 @@ export class DLOBSubscriberIO extends DLOBSubscriber { marketArgs.marketIndex ); - if (shouldSendMessage && this.offloadQueue) { + if (this.offloadQueue) { try { this.offloadQueue.addMessage( Object.assign({}, l3, { ts: Math.floor(new Date().getTime() / 1000), }), - 'DLOBL3Snapshot' + { eventType: 'DLOBL3Snapshot', throttle: true } ); } catch (error) { logger.error('Error adding message to offload queue:', error); diff --git a/src/utils/offload.ts b/src/utils/offload.ts index aeb0750..090a5af 100644 --- a/src/utils/offload.ts +++ b/src/utils/offload.ts @@ -10,7 +10,10 @@ type EventType = 'DLOBSnapshot' | 'DLOBL3Snapshot'; interface QueueMessage { data: any; - eventType: EventType; +} + +interface ThrottleInfo { + lastSentTime: number; } const kinesis = new KinesisClient({ @@ -27,6 +30,7 @@ const batchInterval = process.env.KINESIS_BATCH_INTERVAL export const OffloadQueue = () => { const queue: QueueMessage[] = []; + const throttleMap: Map = new Map(); let isProcessing = false; setInterval(async () => { @@ -89,11 +93,61 @@ export const OffloadQueue = () => { } }; - const addMessage = (data: any, eventType: EventType = 'DLOBSnapshot') => { - queue.push({ - data, - eventType, - }); + const shouldSendThrottled = ( + marketKey: string, + currentTime: number + ): boolean => { + const throttleInfo = throttleMap.get(marketKey); + const lastSent = throttleInfo?.lastSentTime || 0; + const timeSinceLastSent = currentTime - lastSent; + + if (timeSinceLastSent >= 60000) { + throttleMap.set(marketKey, { lastSentTime: currentTime }); + return true; + } + + if (timeSinceLastSent >= 45000) { + const timeInWindow = timeSinceLastSent - 45000; + const probability = Math.min(timeInWindow / 22500, 0.8); + + if (Math.random() < probability) { + throttleMap.set(marketKey, { lastSentTime: currentTime }); + return true; + } + } + + return false; + }; + + const addMessage = ( + data: any, + options: { + eventType?: EventType; + throttle?: boolean; + } = {} + ) => { + const { eventType = 'DLOBSnapshot', throttle = true } = options; + const currentTime = Date.now(); + + if (!throttle) { + queue.push({ + data: { + ...data, + eventType, + }, + }); + return; + } + + const marketKey = `${data.marketType}_${data.marketIndex}`; + if (shouldSendThrottled(marketKey, currentTime)) { + queue.push({ + data: { + ...data, + eventType, + }, + }); + } }; return {