Start processing aggregate trading rewards from first reward (#959)

* Start processing aggregate trading rewards from first reward * lint
dydxprotocol · Jan 11, 2024 · c42ab34 · c42ab34
1 parent b4fd6fe
commit c42ab34
Show file tree

Hide file tree

Showing 2 changed files with 55 additions and 11 deletions.
diff --git a/indexer/services/roundtable/__tests__/tasks/aggregate-trading-rewards.test.ts b/indexer/services/roundtable/__tests__/tasks/aggregate-trading-rewards.test.ts
@@ -121,7 +121,7 @@ describe('aggregate-trading-rewards', () => {
       TradingRewardAggregationPeriod.DAILY,
       TradingRewardAggregationPeriod.WEEKLY,
       TradingRewardAggregationPeriod.MONTHLY,
-    ])('Successfully returns interval if cache is empty and no aggregations', async (
+    ])('Successfully returns interval if cache is empty no aggregations, and no trading rewards', async (
       period: TradingRewardAggregationPeriod,
     ) => {
       const firstBlockTime: DateTime = DateTime.fromISO(
@@ -141,7 +141,7 @@ describe('aggregate-trading-rewards', () => {
     });
 
     it(
-      'Successfully returns interval when cache is empty and no',
+      'Successfully returns interval when cache is empty and no aggregation for the period and trading reward',
       async () => {
         await TradingRewardAggregationTable.create({
           ...defaultMonthlyTradingRewardAggregation,
@@ -166,6 +166,35 @@ describe('aggregate-trading-rewards', () => {
       },
     );
 
+    it(
+      'Successfully returns interval when cache is empty and no aggregation and trading reward exists',
+      async () => {
+        await Promise.all([
+          TradingRewardTable.create({
+            address: testConstants.defaultAddress,
+            blockTime: startedAt.plus({ minutes: 2, seconds: 20 }).toISO(), // random constants
+            blockHeight: '1',
+            amount: '10',
+          }),
+          createBlockWithTime(startedAt.plus({ hours: 10 })),
+        ]);
+        const aggregateTradingReward: AggregateTradingReward = new AggregateTradingReward(
+          TradingRewardAggregationPeriod.MONTHLY,
+        );
+        const interval:
+        Interval = await aggregateTradingReward.getTradingRewardDataToProcessInterval();
+
+        expect(interval).toEqual(Interval.fromDateTimes(
+          startedAt.plus({ minutes: 2, seconds: 20 }),
+          startedAt.plus({
+            minutes: 2,
+            seconds: 20,
+            milliseconds: config.AGGREGATE_TRADING_REWARDS_MAX_INTERVAL_SIZE_MS,
+          }),
+        ));
+      },
+    );
+
     it(
       'Successfully returns interval when cache is empty and a complete aggregations exist',
       async () => {

diff --git a/indexer/services/roundtable/src/tasks/aggregate-trading-rewards.ts b/indexer/services/roundtable/src/tasks/aggregate-trading-rewards.ts
@@ -10,6 +10,7 @@ import {
   BlockTable,
   IsoString,
   IsolationLevel,
+  Ordering,
   TradingRewardAggregationColumns,
   TradingRewardAggregationCreateObject,
   TradingRewardAggregationFromDatabase,
@@ -156,7 +157,7 @@ export class AggregateTradingReward {
     if (processedTime === null) {
       logger.info({
         at: 'aggregate-trading-rewards#getTradingRewardDataToProcessInterval',
-        message: 'Resetting AggregateTradingRewardsProcessedCache',
+        message: 'AggregateTradingRewardsProcessedCache is empty',
       });
       const nextStartTime: DateTime = await this.getNextIntervalStartWhenCacheEmpty();
       await this.setProcessedTime(
@@ -172,9 +173,12 @@ export class AggregateTradingReward {
 
   /**
    * Returns the start time of the next interval to process if the
-   * AggregateTradingRewardProcessedCache is empty. If there is a most recent complete aggregation
-   * for this period, returns the end time of the most recent aggregation, otherwise returns the
-   * start time of the first block in the database.
+   * AggregateTradingRewardProcessedCache is empty.
+   * - If there is a most recent complete aggregation for this period,
+   * returns the end time of the most recent aggregation.
+   * - If there is a trading reward in the database,
+   * returns the block time of the trading reward.
+   * - Otherwise returns the start time of the first block in the database.
    */
   private async getNextIntervalStartWhenCacheEmpty(): Promise<DateTime> {
     const latestAggregation:
@@ -186,12 +190,23 @@ export class AggregateTradingReward {
       return DateTime.fromISO(latestAggregation.endedAt!, UTC_OPTIONS);
     }
 
-    // Since we were able to find the latest block, we assume we can find the first block
-    const firstBlock: BlockFromDatabase[] = await BlockTable.findAll({
-      blockHeight: ['1'],
+    const firstTradingReward: TradingRewardFromDatabase[] = await TradingRewardTable.findAll({
       limit: 1,
-    }, []);
-    return DateTime.fromISO(firstBlock[0].time, UTC_OPTIONS);
+    }, [], { orderBy: [[TradingRewardColumns.blockTime, Ordering.ASC]] });
+    if (firstTradingReward.length === 0) {
+      logger.info({
+        at: 'aggregate-trading-rewards#getNextIntervalStartWhenCacheEmpty',
+        message: 'No trading rewards in database, relying on first block time to pick first start time',
+      });
+      // Since we were able to find the latest block, we assume we can find the first block
+      const firstBlock: BlockFromDatabase[] = await BlockTable.findAll({
+        blockHeight: ['1'],
+        limit: 1,
+      }, []);
+      return DateTime.fromISO(firstBlock[0].time, UTC_OPTIONS);
+    }
+
+    return DateTime.fromISO(firstTradingReward[0].blockTime, UTC_OPTIONS);
   }
 
   /**