[IND-402] Cache and send order updates for stateful orders.

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

@@ -19,6 +19,7 @@ import {
   IndexerOrder,
   OrderPlaceV1_OrderPlacementStatus,
   StatefulOrderEventV1,
+  OrderUpdateV1,
 } from '@dydxprotocol-indexer/v4-protos';
 import { KafkaMessage } from 'kafkajs';
 import { onMessage } from '../../../src/lib/on-message';
@@ -45,6 +46,10 @@ 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', () => {
   beforeAll(async () => {
@@ -64,6 +69,7 @@ describe('statefulOrderPlacementHandler', () => {
 
   afterEach(async () => {
     await dbHelpers.clearData();
+    await redis.deleteAllAsync(redisClient);
     jest.clearAllMocks();
   });
 
@@ -138,15 +144,48 @@ describe('statefulOrderPlacementHandler', () => {
 
   it.each([
     // TODO(IND-334): Remove after deprecating StatefulOrderPlacementEvent
-    ['stateful order placement', defaultStatefulOrderEvent],
-    ['stateful long term order placement', defaultStatefulOrderLongTermEvent],
+    [
+      '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,
+      },
+    ],
   ])('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);
@@ -185,6 +224,25 @@ 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,14 +6,17 @@ 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';
 
@@ -60,18 +63,34 @@ export class StatefulOrderPlacementHandler extends
       this.generateTimingStatsOptions('upsert_order'),
     );
 
+    const kafakEvents: ConsolidatedKafkaEvent[] = [];
+
     const offChainUpdate: OffChainUpdateV1 = OffChainUpdateV1.fromPartial({
       orderPlace: {
         order,
         placementStatus: OrderPlaceV1_OrderPlacementStatus.ORDER_PLACEMENT_STATUS_OPENED,
       },
     });
+    kafakEvents.push(this.generateConsolidatedVulcanKafkaEvent(
+      getOrderIdHash(order.orderId!),
+      offChainUpdate,
+    ));
 
-    return [
-      this.generateConsolidatedVulcanKafkaEvent(
+    const pendingOrderUpdate: OrderUpdateV1 | undefined = await StatefulOrderUpdatesCache
+      .removeStatefulOrderUpdate(
+        OrderTable.orderIdToUuid(order.orderId!),
+        Date.now(),
+        redisClient,
+      );
+    if (pendingOrderUpdate !== undefined) {
+      kafakEvents.push(this.generateConsolidatedVulcanKafkaEvent(
         getOrderIdHash(order.orderId!),
-        offChainUpdate,
-      ),
-    ];
+        OffChainUpdateV1.fromPartial({
+          orderUpdate: pendingOrderUpdate,
+        }),
+      ));
+    }
+
+    return kafakEvents;
   }
 }

indexer/services/roundtable/__tests__/tasks/remove-old-order-updates.test.ts

@@ -0,0 +1,125 @@
+import { stats } from '@dydxprotocol-indexer/base';
+import {
+  StatefulOrderUpdateInfo,
+  StatefulOrderUpdatesCache,
+  redis,
+  redisTestConstants,
+} from '@dydxprotocol-indexer/redis';
+import config from '../../src/config';
+import removeOldOrderUpdatesTask from '../../src/tasks/remove-old-order-updates';
+import { redisClient } from '../../src/helpers/redis';
+
+describe('remove-old-order-updates', () => {
+  const fakeTime: Date = new Date(2023, 9, 25, 0, 0, 0, 0);
+
+  beforeAll(() => {
+    jest.useFakeTimers().setSystemTime(fakeTime);
+  });
+
+  afterAll(() => {
+    jest.resetAllMocks();
+    jest.useRealTimers();
+  });
+
+  beforeEach(() => {
+    jest.spyOn(stats, 'gauge');
+    jest.clearAllMocks();
+  });
+
+  afterEach(async () => {
+    await redis.deleteAllAsync(redisClient);
+    jest.clearAllMocks();
+  });
+
+  it('succeeds with no cached order updates', async () => {
+    await removeOldOrderUpdatesTask();
+    expect(stats.gauge).toHaveBeenCalledWith(
+      `${config.SERVICE_NAME}.remove_old_order_updates.num_removed`,
+      0,
+    );
+  });
+
+  it('succeeds with no old cached order updates', async () => {
+    await StatefulOrderUpdatesCache.addStatefulOrderUpdate(
+      redisTestConstants.defaultOrderUuidGoodTilBlockTime,
+      redisTestConstants.orderUpdate.orderUpdate,
+      fakeTime.getTime() - 1,
+      redisClient,
+    );
+    const existingUpdates: StatefulOrderUpdateInfo[] = await StatefulOrderUpdatesCache
+      .getOldOrderUpdates(
+        fakeTime.getTime(),
+        redisClient,
+      );
+    expect(existingUpdates).toHaveLength(1);
+
+    await removeOldOrderUpdatesTask();
+    expect(stats.gauge).toHaveBeenCalledWith(
+      `${config.SERVICE_NAME}.remove_old_order_updates.num_removed`,
+      0,
+    );
+
+    const updatesAfterTask: StatefulOrderUpdateInfo[] = await StatefulOrderUpdatesCache
+      .getOldOrderUpdates(
+        fakeTime.getTime(),
+        redisClient,
+      );
+    expect(updatesAfterTask).toEqual(existingUpdates);
+  });
+
+  it('succeeds with no old cached order updates', async () => {
+    await StatefulOrderUpdatesCache.addStatefulOrderUpdate(
+      redisTestConstants.defaultOrderUuidGoodTilBlockTime,
+      redisTestConstants.orderUpdate.orderUpdate,
+      fakeTime.getTime() - 1,
+      redisClient,
+    );
+    const existingUpdates: StatefulOrderUpdateInfo[] = await StatefulOrderUpdatesCache
+      .getOldOrderUpdates(
+        fakeTime.getTime(),
+        redisClient,
+      );
+    expect(existingUpdates).toHaveLength(1);
+
+    await removeOldOrderUpdatesTask();
+    expect(stats.gauge).toHaveBeenCalledWith(
+      `${config.SERVICE_NAME}.remove_old_order_updates.num_removed`,
+      0,
+    );
+
+    const updatesAfterTask: StatefulOrderUpdateInfo[] = await StatefulOrderUpdatesCache
+      .getOldOrderUpdates(
+        fakeTime.getTime(),
+        redisClient,
+      );
+    expect(updatesAfterTask).toEqual(existingUpdates);
+  });
+
+  it('succeeds removing old cached order update', async () => {
+    await StatefulOrderUpdatesCache.addStatefulOrderUpdate(
+      redisTestConstants.defaultOrderUuidGoodTilBlockTime,
+      redisTestConstants.orderUpdate.orderUpdate,
+      fakeTime.getTime() - config.OLD_CACHED_ORDER_UPDATES_WINDOW_MS,
+      redisClient,
+    );
+    const existingUpdates: StatefulOrderUpdateInfo[] = await StatefulOrderUpdatesCache
+      .getOldOrderUpdates(
+        fakeTime.getTime(),
+        redisClient,
+      );
+    expect(existingUpdates).toHaveLength(1);
+
+    await removeOldOrderUpdatesTask();
+    expect(stats.gauge).toHaveBeenCalledWith(
+      `${config.SERVICE_NAME}.remove_old_order_updates.num_removed`,
+      1,
+    );
+
+    const updatesAfterTask: StatefulOrderUpdateInfo[] = await StatefulOrderUpdatesCache
+      .getOldOrderUpdates(
+        fakeTime.getTime(),
+        redisClient,
+      );
+    expect(updatesAfterTask).toHaveLength(0);
+  });
+});

indexer/services/roundtable/src/config.ts

@@ -119,6 +119,9 @@ export const configSchema = {
   MAX_COMPLIANCE_DATA_QUERY_PER_LOOP: parseInteger({ default: 100 }),
   COMPLIANCE_PROVIDER_QUERY_BATCH_SIZE: parseInteger({ default: 100 }),
   COMPLIANCE_PROVIDER_QUERY_DELAY_MS: parseInteger({ default: ONE_SECOND_IN_MILLISECONDS }),
+
+  // Remove old cached order updates
+  OLD_CACHED_ORDER_UPDATES_WINDOW_MS: parseInteger({ default: 30 * ONE_SECOND_IN_MILLISECONDS }),
 };
 
 export default parseSchema(configSchema);

indexer/services/roundtable/src/tasks/remove-old-order-updates.ts

@@ -0,0 +1,68 @@
+import {
+  logger,
+  stats,
+} from '@dydxprotocol-indexer/base';
+import {
+  StatefulOrderUpdateInfo,
+  StatefulOrderUpdatesCache,
+} from '@dydxprotocol-indexer/redis';
+import { OrderUpdateV1 } from '@dydxprotocol-indexer/v4-protos';
+
+import config from '../config';
+import { redisClient } from '../helpers/redis';
+
+/**
+ * This task removes any old cached stateful orer updates from the StatefulOrderUpdates cache
+ */
+export default async function runTask(): Promise<void> {
+  const start: number = Date.now();
+
+  try {
+    const oldUpdateCutoff: number = Date.now() - config.OLD_CACHED_ORDER_UPDATES_WINDOW_MS;
+
+    const oldUpdateInfo: StatefulOrderUpdateInfo[] = await StatefulOrderUpdatesCache
+      .getOldOrderUpdates(
+        oldUpdateCutoff,
+        redisClient,
+      );
+    const removedUpdates: OrderUpdateV1[] = (await Promise.all(
+      oldUpdateInfo.map(
+        (updateInfo: StatefulOrderUpdateInfo): Promise<OrderUpdateV1 | undefined> => {
+          return StatefulOrderUpdatesCache.removeStatefulOrderUpdate(
+            updateInfo.orderId,
+            updateInfo.timestamp,
+            redisClient,
+          );
+        },
+      ),
+    )).filter(
+      (removedUpdate: OrderUpdateV1 | undefined): removedUpdate is OrderUpdateV1 => {
+        if (removedUpdate !== undefined) {
+          logger.info({
+            at: 'remove-old-order-updates#runTask',
+            message: 'Removed old stateful order update',
+            removedUpdate,
+          });
+          return true;
+        }
+        return false;
+      },
+    );
+
+    stats.gauge(
+      `${config.SERVICE_NAME}.remove_old_order_updates.num_removed`,
+      removedUpdates.length,
+    );
+  } catch (error) {
+    logger.error({
+      at: 'remove-old-order-updates#runTas',
+      message: 'Error occurred in task to remove old stateful order updates',
+      error,
+    });
+  } finally {
+    stats.timing(
+      `${config.SERVICE_NAME}.remove_old_order_updates`,
+      Date.now() - start,
+    );
+  }
+}