Indexer e2e latency round 2 (backport #1314) (#1558)

Co-authored-by: Jonathan Fung <121899091+jonfung-dydx@users.noreply.github.com>
Co-authored-by: Will Liu <will@dydx.exchange>

indexer/packages/v4-protos/src/index.ts

@@ -16,3 +16,4 @@ export * from './codegen/google/protobuf/timestamp';
 export * from './codegen/dydxprotocol/indexer/protocol/v1/clob';
 export * from './codegen/dydxprotocol/indexer/protocol/v1/subaccount';
 export * from './codegen/dydxprotocol/indexer/shared/removal_reason';
+export * from './utils';

indexer/packages/v4-protos/src/utils.ts

@@ -0,0 +1,12 @@
+import { Timestamp } from './codegen/google/protobuf/timestamp';
+
+export const MILLIS_IN_NANOS: number = 1_000_000;
+export const SECONDS_IN_MILLIS: number = 1_000;
+export function protoTimestampToDate(
+  protoTime: Timestamp,
+): Date {
+  const timeInMillis: number = Number(protoTime.seconds) * SECONDS_IN_MILLIS +
+    Math.floor(protoTime.nanos / MILLIS_IN_NANOS);
+
+  return new Date(timeInMillis);
+}

indexer/services/ender/__tests__/helpers/indexer-proto-helpers.ts

@@ -51,6 +51,7 @@ import {
   PerpetualMarketCreateEventV1,
   PerpetualMarketCreateEventV2,
   DeleveragingEventV1,
+  protoTimestampToDate,
 } from '@dydxprotocol-indexer/v4-protos';
 import {
   PerpetualMarketType,
@@ -64,7 +65,6 @@ import {
   generatePerpetualMarketMessage,
   generatePerpetualPositionsContents,
 } from '../../src/helpers/kafka-helper';
-import { protoTimestampToDate } from '../../src/lib/helper';
 import { DydxIndexerSubtypes, VulcanMessage } from '../../src/lib/types';
 
 // TX Hash is SHA256, so is of length 64 hexadecimal without the '0x'.

indexer/services/ender/src/lib/helper.ts

@@ -8,7 +8,6 @@ import {
 import {
   IndexerTendermintEvent,
   IndexerTendermintEvent_BlockEvent,
-  Timestamp,
   OrderFillEventV1,
   MarketEventV1,
   SubaccountUpdateEventV1,
@@ -32,10 +31,6 @@ import Big from 'big.js';
 import _ from 'lodash';
 import { DateTime } from 'luxon';
 
-import {
-  MILLIS_IN_NANOS,
-  SECONDS_IN_MILLIS,
-} from '../constants';
 import {
   AnnotatedSubaccountMessage,
   DydxIndexerSubtypes,
@@ -73,15 +68,6 @@ export function convertToSubaccountMessage(
   return subaccountMessage;
 }
 
-export function protoTimestampToDate(
-  protoTime: Timestamp,
-): Date {
-  const timeInMillis: number = Number(protoTime.seconds) * SECONDS_IN_MILLIS +
-    Math.floor(protoTime.nanos / MILLIS_IN_NANOS);
-
-  return new Date(timeInMillis);
-}
-
 export function dateToDateTime(
   protoTime: Date,
 ): DateTime {

indexer/services/socks/src/lib/message-forwarder.ts

@@ -87,9 +87,10 @@ export class MessageForwarder {
   }
 
   public onMessage(topic: string, message: KafkaMessage): void {
+    const start: number = Date.now();
     stats.timing(
       `${config.SERVICE_NAME}.message_time_in_queue`,
-      Date.now() - Number(message.timestamp),
+      start - Number(message.timestamp),
       config.MESSAGE_FORWARDER_STATSD_SAMPLE_RATE,
       {
         topic,
@@ -184,10 +185,10 @@ export class MessageForwarder {
 
     if (subscriptions.length > 0) {
       if (message.channel !== Channel.V4_ORDERBOOK ||
-          (
-            // Don't log orderbook messages unless enabled
-            message.channel === Channel.V4_ORDERBOOK && config.ENABLE_ORDERBOOK_LOGS
-          )
+        (
+          // Don't log orderbook messages unless enabled
+          message.channel === Channel.V4_ORDERBOOK && config.ENABLE_ORDERBOOK_LOGS
+        )
       ) {
         logger.debug({
           at: 'message-forwarder#forwardMessage',
@@ -200,7 +201,7 @@ export class MessageForwarder {
 
     // Buffer messages if the subscription is for batched messages
     if (this.subscriptions.batchedSubscriptions[message.channel] &&
-       this.subscriptions.batchedSubscriptions[message.channel][message.id]) {
+      this.subscriptions.batchedSubscriptions[message.channel][message.id]) {
       const bufferKey: string = this.getMessageBufferKey(
         message.channel,
         message.id,

indexer/services/vulcan/__tests__/handlers/order-place-handler.test.ts

@@ -62,6 +62,7 @@ import { expectCanceledOrderStatus, expectOpenOrderIds, handleInitialOrderPlace
 import { expectOffchainUpdateMessage, expectWebsocketOrderbookMessage, expectWebsocketSubaccountMessage } from '../helpers/websocket-helpers';
 import { OrderbookSide } from '../../src/lib/types';
 import { getOrderIdHash, isLongTermOrder, isStatefulOrder } from '@dydxprotocol-indexer/v4-proto-parser';
+import { defaultKafkaHeaders } from '../helpers/constants';
 import config from '../../src/config';
 
 jest.mock('@dydxprotocol-indexer/base', () => ({
@@ -196,6 +197,12 @@ describe('order-place-handler', () => {
     const replacementMessageIoc: KafkaMessage = createKafkaMessage(
       Buffer.from(Uint8Array.from(OffChainUpdateV1.encode(replacementUpdateIoc).finish())),
     );
+    [replacementMessage, replacementMessageGoodTilBlockTime, replacementMessageConditional,
+      replacementMessageFok, replacementMessageIoc].forEach((message) => {
+      // eslint-disable-next-line no-param-reassign
+      message.headers = defaultKafkaHeaders;
+    });
+
     const dbDefaultOrder: OrderFromDatabase = {
       ...testConstants.defaultOrder,
       id: testConstants.defaultOrderId,
@@ -1225,7 +1232,11 @@ function expectWebsocketMessagesSent(
       version: SUBACCOUNTS_WEBSOCKET_MESSAGE_VERSION,
     });
 
-    expectWebsocketSubaccountMessage(producerSendSpy.mock.calls[callIndex][0], subaccountMessage);
+    expectWebsocketSubaccountMessage(
+      producerSendSpy.mock.calls[callIndex][0],
+      subaccountMessage,
+      defaultKafkaHeaders,
+    );
     callIndex += 1;
   }