[IND-402] Have vulcan re-queue cached order updates instead of ender.

indexer/services/ender/__tests__/handlers/stateful-order/stateful-order-placement-handler.test.ts

@@ -19,7 +19,6 @@ import {
   IndexerOrder,
   OrderPlaceV1_OrderPlacementStatus,
   StatefulOrderEventV1,
-  OrderUpdateV1,
 } from '@dydxprotocol-indexer/v4-protos';
 import { KafkaMessage } from 'kafkajs';
 import { onMessage } from '../../../src/lib/on-message';
@@ -46,9 +45,6 @@ import { STATEFUL_ORDER_ORDER_FILL_EVENT_TYPE } from '../../../src/constants';
 import { producer } from '@dydxprotocol-indexer/kafka';
 import { ORDER_FLAG_LONG_TERM } from '@dydxprotocol-indexer/v4-proto-parser';
 import { createPostgresFunctions } from '../../../src/helpers/postgres/postgres-functions';
-import {
-  redis, redisTestConstants, StatefulOrderUpdateInfo, StatefulOrderUpdatesCache,
-} from '@dydxprotocol-indexer/redis';
 import { redisClient } from '../../../src/helpers/redis/redis-controller';
 
 describe('statefulOrderPlacementHandler', () => {
@@ -69,7 +65,6 @@ describe('statefulOrderPlacementHandler', () => {
 
   afterEach(async () => {
     await dbHelpers.clearData();
-    await redis.deleteAllAsync(redisClient);
     jest.clearAllMocks();
   });
 
@@ -144,48 +139,15 @@ describe('statefulOrderPlacementHandler', () => {
 
   it.each([
     // TODO(IND-334): Remove after deprecating StatefulOrderPlacementEvent
-    [
-      'stateful order placement and no cached update',
-      defaultStatefulOrderEvent,
-      undefined,
-    ],
-    [
-      'stateful long term order placement and no cached update',
-      defaultStatefulOrderLongTermEvent,
-      undefined,
-    ],
-    [
-      'stateful order placement and cached update',
-      defaultStatefulOrderEvent,
-      {
-        ...redisTestConstants.orderUpdate.orderUpdate,
-        orderId: defaultOrder.orderId,
-      },
-    ],
-    [
-      'stateful long term order placement and cached update',
-      defaultStatefulOrderLongTermEvent,
-      {
-        ...redisTestConstants.orderUpdate.orderUpdate,
-        orderId: defaultOrder.orderId,
-      },
-    ],
+    ['stateful order placement', defaultStatefulOrderEvent],
+    ['stateful long term order placement', defaultStatefulOrderLongTermEvent],
   ])('successfully places order with %s', async (
     _name: string,
     statefulOrderEvent: StatefulOrderEventV1,
-    cachedOrderUpdate: OrderUpdateV1 | undefined,
   ) => {
     const kafkaMessage: KafkaMessage = createKafkaMessageFromStatefulOrderEvent(
       statefulOrderEvent,
     );
-    if (cachedOrderUpdate !== undefined) {
-      await StatefulOrderUpdatesCache.addStatefulOrderUpdate(
-        orderId,
-        cachedOrderUpdate,
-        Date.now(),
-        redisClient,
-      );
-    }
 
     await onMessage(kafkaMessage);
     const order: OrderFromDatabase | undefined = await OrderTable.findById(orderId);
@@ -224,25 +186,6 @@ describe('statefulOrderPlacementHandler', () => {
       orderId: defaultOrder.orderId!,
       offchainUpdate: expectedOffchainUpdate,
     });
-
-    // If there was a cached order update, expect the cache to be empty and a corresponding
-    // off-chain update to have been sent to the Kafka producer
-    if (cachedOrderUpdate !== undefined) {
-      const orderUpdates: StatefulOrderUpdateInfo[] = await StatefulOrderUpdatesCache
-        .getOldOrderUpdates(
-          Date.now(),
-          redisClient,
-        );
-      expect(orderUpdates).toHaveLength(0);
-
-      expectVulcanKafkaMessage({
-        producerSendMock,
-        orderId: defaultOrder.orderId!,
-        offchainUpdate: {
-          orderUpdate: cachedOrderUpdate,
-        },
-      });
-    }
   });
 
   it.each([

indexer/services/ender/src/handlers/stateful-order/stateful-order-placement-handler.ts

@@ -6,17 +6,14 @@ import {
   perpetualMarketRefresher,
   OrderStatus,
 } from '@dydxprotocol-indexer/postgres';
-import { StatefulOrderUpdatesCache } from '@dydxprotocol-indexer/redis';
 import { getOrderIdHash } from '@dydxprotocol-indexer/v4-proto-parser';
 import {
   OrderPlaceV1_OrderPlacementStatus,
   OffChainUpdateV1,
   IndexerOrder,
   StatefulOrderEventV1,
-  OrderUpdateV1,
 } from '@dydxprotocol-indexer/v4-protos';
 
-import { redisClient } from '../../helpers/redis/redis-controller';
 import { ConsolidatedKafkaEvent } from '../../lib/types';
 import { AbstractStatefulOrderHandler } from '../abstract-stateful-order-handler';
 
@@ -76,21 +73,6 @@ export class StatefulOrderPlacementHandler extends
       offChainUpdate,
     ));
 
-    const pendingOrderUpdate: OrderUpdateV1 | undefined = await StatefulOrderUpdatesCache
-      .removeStatefulOrderUpdate(
-        OrderTable.orderIdToUuid(order.orderId!),
-        Date.now(),
-        redisClient,
-      );
-    if (pendingOrderUpdate !== undefined) {
-      kafakEvents.push(this.generateConsolidatedVulcanKafkaEvent(
-        getOrderIdHash(order.orderId!),
-        OffChainUpdateV1.fromPartial({
-          orderUpdate: pendingOrderUpdate,
-        }),
-      ));
-    }
-
     return kafakEvents;
   }
 }

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

@@ -40,6 +40,7 @@ import {
   SubaccountOrderIdsCache,
   CanceledOrdersCache,
   updateOrder,
+  StatefulOrderUpdatesCache,
 } from '@dydxprotocol-indexer/redis';
 
 import {
@@ -50,21 +51,28 @@ import {
   RedisOrder,
   SubaccountId,
   SubaccountMessage,
+  OrderUpdateV1,
 } from '@dydxprotocol-indexer/v4-protos';
 import { KafkaMessage } from 'kafkajs';
 import Long from 'long';
 import { convertToRedisOrder, getTriggerPrice } from '../../src/handlers/helpers';
 import { redisClient, redisClient as client } from '../../src/helpers/redis/redis-controller';
 import { onMessage } from '../../src/lib/on-message';
 import { expectCanceledOrdersCacheEmpty, expectOpenOrderIds, handleInitialOrderPlace } from '../helpers/helpers';
-import { expectWebsocketOrderbookMessage, expectWebsocketSubaccountMessage } from '../helpers/websocket-helpers';
+import { expectOffchainUpdateMessage, expectWebsocketOrderbookMessage, expectWebsocketSubaccountMessage } from '../helpers/websocket-helpers';
 import { OrderbookSide } from '../../src/lib/types';
+import { getOrderIdHash } from '@dydxprotocol-indexer/v4-proto-parser';
 
 jest.mock('@dydxprotocol-indexer/base', () => ({
   ...jest.requireActual('@dydxprotocol-indexer/base'),
   wrapBackgroundTask: jest.fn(),
 }));
 
+interface OffchainUpdateRecord {
+  key: Buffer,
+  offchainUpdate: OffChainUpdateV1
+}
+
 describe('order-place-handler', () => {
   beforeAll(() => {
     jest.useFakeTimers();
@@ -640,28 +648,70 @@ describe('order-place-handler', () => {
 
     it.each([
       [
-        'good-til-block-time',
+        'good-til-block-time and no cached update',
         redisTestConstants.defaultOrderGoodTilBlockTime,
         redisTestConstants.defaultRedisOrderGoodTilBlockTime,
         redisTestConstants.defaultOrderUuidGoodTilBlockTime,
         dbOrderGoodTilBlockTime,
+        undefined,
       ],
       [
-        'conditional',
+        'conditional and no cached update',
         redisTestConstants.defaultConditionalOrder,
         redisTestConstants.defaultRedisOrderConditional,
         redisTestConstants.defaultOrderUuidConditional,
         dbConditionalOrder,
+        undefined,
+      ],
+      [
+        'good-til-block-time and cached update',
+        redisTestConstants.defaultOrderGoodTilBlockTime,
+        redisTestConstants.defaultRedisOrderGoodTilBlockTime,
+        redisTestConstants.defaultOrderUuidGoodTilBlockTime,
+        dbOrderGoodTilBlockTime,
+        {
+          ...redisTestConstants.orderUpdate.orderUpdate,
+          orderId: redisTestConstants.defaultOrderIdGoodTilBlockTime,
+        },
+      ],
+      [
+        'conditional and cached update',
+        redisTestConstants.defaultConditionalOrder,
+        redisTestConstants.defaultRedisOrderConditional,
+        redisTestConstants.defaultOrderUuidConditional,
+        dbConditionalOrder,
+        {
+          ...redisTestConstants.orderUpdate.orderUpdate,
+          orderId: redisTestConstants.defaultOrderIdConditional,
+        },
       ],
     ])('handles order place with OPEN placement status, does not exist initially (with %s)', async (
       _name: string,
       orderToPlace: IndexerOrder,
       expectedRedisOrder: RedisOrder,
       expectedOrderUuid: string,
       placedOrder: OrderFromDatabase,
+      cachedOrderUpdate: OrderUpdateV1 | undefined,
     ) => {
       synchronizeWrapBackgroundTask(wrapBackgroundTask);
       const producerSendSpy: jest.SpyInstance = jest.spyOn(producer, 'send').mockReturnThis();
+
+      let expectedOffchainUpdate: OffchainUpdateRecord | undefined;
+      if (cachedOrderUpdate !== undefined) {
+        await StatefulOrderUpdatesCache.addStatefulOrderUpdate(
+          expectedOrderUuid,
+          cachedOrderUpdate,
+          Date.now(),
+          redisClient,
+        );
+        expectedOffchainUpdate = {
+          key: getOrderIdHash(orderToPlace.orderId!),
+          offchainUpdate: {
+            orderUpdate: cachedOrderUpdate,
+          },
+        };
+      }
+
       await handleInitialOrderPlace({
         ...redisTestConstants.orderPlace,
         orderPlace: {
@@ -670,6 +720,7 @@ describe('order-place-handler', () => {
             OrderPlaceV1_OrderPlacementStatus.ORDER_PLACEMENT_STATUS_OPENED,
         },
       });
+
       expectWebsocketMessagesSent(
         producerSendSpy,
         expectedRedisOrder,
@@ -678,7 +729,10 @@ describe('order-place-handler', () => {
         APIOrderStatusEnum.OPEN,
         // Subaccount message should be sent for stateful order if status is OPEN
         true,
+        undefined,
+        expectedOffchainUpdate,
       );
+
       await checkOrderPlace(
         expectedOrderUuid,
         redisTestConstants.defaultSubaccountUuid,
@@ -962,6 +1016,7 @@ function expectWebsocketMessagesSent(
   placementStatus: APIOrderStatus,
   expectSubaccountMessage: boolean,
   expectedOrderbookMessage?: OrderbookMessage,
+  expectedOffchainUpdate?: OffchainUpdateRecord,
 ): void {
   jest.runOnlyPendingTimers();
   // expect one subaccount update message being sent
@@ -972,8 +1027,22 @@ function expectWebsocketMessagesSent(
   if (expectedOrderbookMessage !== undefined) {
     numMessages += 1;
   }
+  if (expectedOffchainUpdate !== undefined) {
+    numMessages += 1;
+  }
   expect(producerSendSpy).toHaveBeenCalledTimes(numMessages);
 
+  let callIndex: number = 0;
+
+  if (expectedOffchainUpdate) {
+    expectOffchainUpdateMessage(
+      producerSendSpy.mock.calls[callIndex][0],
+      expectedOffchainUpdate.key,
+      expectedOffchainUpdate.offchainUpdate,
+    );
+    callIndex += 1;
+  }
+
   if (expectSubaccountMessage) {
     const orderTIF: TimeInForce = protocolTranslations.protocolOrderTIFToTIF(
       redisOrder.order!.timeInForce,
@@ -1016,12 +1085,11 @@ function expectWebsocketMessagesSent(
       version: SUBACCOUNTS_WEBSOCKET_MESSAGE_VERSION,
     });
 
-    expectWebsocketSubaccountMessage(producerSendSpy.mock.calls[0][0], subaccountMessage);
+    expectWebsocketSubaccountMessage(producerSendSpy.mock.calls[callIndex][0], subaccountMessage);
+    callIndex += 1;
   }
 
   if (expectedOrderbookMessage !== undefined) {
-    // If there is no subaccount message, the orderbook message should be the first message
-    const callIndex: number = expectSubaccountMessage ? 1 : 0;
     expectWebsocketOrderbookMessage(
       producerSendSpy.mock.calls[callIndex][0],
       expectedOrderbookMessage,

indexer/services/vulcan/__tests__/helpers/websocket-helpers.ts

@@ -1,5 +1,5 @@
 import { KafkaTopics } from '@dydxprotocol-indexer/kafka';
-import { OrderbookMessage, SubaccountMessage } from '@dydxprotocol-indexer/v4-protos';
+import { OffChainUpdateV1, OrderbookMessage, SubaccountMessage } from '@dydxprotocol-indexer/v4-protos';
 import { ProducerRecord } from 'kafkajs';
 
 export function expectWebsocketSubaccountMessage(
@@ -29,3 +29,20 @@ export function expectWebsocketOrderbookMessage(
   );
   expect(orderbookMessage).toEqual(expectedOrderbookMessage);
 }
+
+export function expectOffchainUpdateMessage(
+  offchainUpdateProducerRecord: ProducerRecord,
+  expectedKey: Buffer,
+  expectedOffchainUpdate: OffChainUpdateV1,
+): void {
+  expect(offchainUpdateProducerRecord.topic).toEqual(KafkaTopics.TO_VULCAN);
+  const offchainUpdateMessageValueBinary: Uint8Array = new Uint8Array(
+    offchainUpdateProducerRecord.messages[0].value as Buffer,
+  );
+  const key: Buffer = offchainUpdateProducerRecord.messages[0].key as Buffer;
+  const offchainUpdate: OffChainUpdateV1 = OffChainUpdateV1.decode(
+    offchainUpdateMessageValueBinary,
+  );
+  expect(offchainUpdate).toEqual(expectedOffchainUpdate);
+  expect(key).toEqual(expectedKey);
+}