chore: add randomness

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);

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<string, ThrottleInfo> = new Map();
 	let isProcessing = false;
 
 	setInterval(async () => {
@@ -89,11 +93,61 @@ export const OffloadQueue = () => {
 		}
 	};
 
-	const addMessage = (data: any, eventType: EventType = 'DLOBSnapshot') => {
+	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: {
+					...data,
 					eventType,
+				},
 			});
+			return;
+		}
+
+		const marketKey = `${data.marketType}_${data.marketIndex}`;
+		if (shouldSendThrottled(marketKey, currentTime)) {
+			queue.push({
+				data: {
+					...data,
+					eventType,
+				},
+			});
+		}
 	};
 
 	return {