common: fix errors when there're a lot of ravs pending

packages/indexer-common/src/allocations/__tests__/tap.test.ts

@@ -281,6 +281,55 @@ describe('TAP', () => {
     ])
   })
 
+  test('should mark ravs as redeemed via `markRavsInTransactionsAsRedeemed`', async () => {
+    const nowSecs = Math.floor(Date.now() / 1000)
+    const transactions = {
+      transactions: [
+        {
+          id: 'test',
+          allocationID: ALLOCATION_ID_2.toString().toLowerCase().replace('0x', ''),
+          timestamp: nowSecs,
+          sender: {
+            id: SENDER_ADDRESS_3.toString().toLowerCase().replace('0x', ''),
+          },
+        },
+      ],
+      _meta: {
+        block: {
+          timestamp: nowSecs,
+          hash: 'test',
+        },
+      },
+    }
+
+    const rav2 = {
+      allocationId: ALLOCATION_ID_2,
+      last: true,
+      final: false,
+      timestampNs: 1709067401177959664n,
+      valueAggregate: 20000000000000n,
+      signature: SIGNATURE,
+      senderAddress: SENDER_ADDRESS_3,
+      redeemedAt: null,
+    }
+    await queryFeeModels.receiptAggregateVouchers.create(rav2)
+    const ravs = await tapCollector['pendingRAVs']()
+    await tapCollector['markRavsInTransactionsAsRedeemed'](transactions, ravs)
+    const redeemedRavs = await queryFeeModels.receiptAggregateVouchers.findAll({
+      where: {
+        last: true,
+        final: false,
+        redeemedAt: {
+          [Op.ne]: null,
+        },
+      },
+    })
+    // Expect redeemed rav to be returned here
+    expect(redeemedRavs).toEqual([
+      expect.objectContaining({ ...rav2, redeemedAt: new Date(nowSecs * 1000) }),
+    ])
+  })
+
   test('should mark ravs as final via `markRavsAsFinal`', async () => {
     // we have a redeemed non-final rav in our database
     const nowSecs = Math.floor(Date.now() / 1000)

packages/indexer-common/src/allocations/tap-collector.ts

@@ -328,38 +328,24 @@ export class TapCollector {
   private async pendingRAVs(): Promise<ReceiptAggregateVoucher[]> {
     return await this.models.receiptAggregateVouchers.findAll({
       where: { last: true, final: false },
+      limit: 100,
     })
   }
 
   private async filterAndUpdateRavs(
     ravsLastNotFinal: ReceiptAggregateVoucher[],
   ): Promise<ReceiptAggregateVoucher[]> {
+    // look for all transactions for that includes senderaddress[] and allocations[]
     const tapSubgraphResponse = await this.findTransactionsForRavs(ravsLastNotFinal)
 
-    const redeemedRavsNotOnOurDatabase = tapSubgraphResponse.transactions.filter(
-      (tx) =>
-        !ravsLastNotFinal.find(
-          (rav) =>
-            toAddress(rav.senderAddress) === toAddress(tx.sender.id) &&
-            toAddress(rav.allocationId) === toAddress(tx.allocationID),
-        ),
-    )
-
-    // for each transaction that is not redeemed on our database
-    // but was redeemed on the blockchain, update it to redeemed
-    if (redeemedRavsNotOnOurDatabase.length > 0) {
-      for (const rav of redeemedRavsNotOnOurDatabase) {
-        await this.markRavAsRedeemed(
-          toAddress(rav.allocationID),
-          toAddress(rav.sender.id),
-          rav.timestamp,
-        )
-      }
-    }
+    // check for redeemed ravs in tx list but not marked as redeemed in our database
+    this.markRavsInTransactionsAsRedeemed(tapSubgraphResponse, ravsLastNotFinal)
 
     // Filter unfinalized RAVS fetched from DB, keeping RAVs that have not yet been redeemed on-chain
     const nonRedeemedRavs = ravsLastNotFinal
+      // get all ravs that were marked as redeemed in our database
       .filter((rav) => !!rav.redeemedAt)
+      // get all ravs that wasn't possible to find the transaction
       .filter(
         (rav) =>
           !tapSubgraphResponse.transactions.find(
@@ -388,13 +374,55 @@ export class TapCollector {
     })
   }
 
+  public async markRavsInTransactionsAsRedeemed(
+    tapSubgraphResponse: TapSubgraphResponse,
+    ravsLastNotFinal: ReceiptAggregateVoucher[],
+  ) {
+    // get a list of transactions for ravs marked as not redeemed in our database
+    const redeemedRavsNotOnOurDatabase = tapSubgraphResponse.transactions
+      // get only the transactions that exists, this prevents errors marking as redeemed
+      // transactions for different senders with the same allocation id
+      .filter((tx) => {
+        // check if exists in the ravsLastNotFinal list
+        return !!ravsLastNotFinal.find(
+          (rav) =>
+            // rav has the same sender address as tx
+            toAddress(rav.senderAddress) === toAddress(tx.sender.id) &&
+            // rav has the same allocation id as tx
+            toAddress(rav.allocationId) === toAddress(tx.allocationID) &&
+            // rav was marked as not redeemed in the db
+            !rav.redeemedAt,
+        )
+      })
+
+    // for each transaction that is not redeemed on our database
+    // but was redeemed on the blockchain, update it to redeemed
+    if (redeemedRavsNotOnOurDatabase.length > 0) {
+      for (const rav of redeemedRavsNotOnOurDatabase) {
+        await this.markRavAsRedeemed(
+          toAddress(rav.allocationID),
+          toAddress(rav.sender.id),
+          rav.timestamp,
+        )
+      }
+    }
+  }
+
   public async findTransactionsForRavs(
     ravs: ReceiptAggregateVoucher[],
   ): Promise<TapSubgraphResponse> {
     let meta: TapMeta | undefined = undefined
     let lastId = ''
     const transactions: TapTransaction[] = []
 
+    const unfinalizedRavsAllocationIds = [
+      ...new Set(ravs.map((value) => toAddress(value.allocationId).toLowerCase())),
+    ]
+
+    const senderAddresses = [
+      ...new Set(ravs.map((value) => toAddress(value.senderAddress).toLowerCase())),
+    ]
+
     for (;;) {
       let block: { hash: string } | undefined = undefined
       if (meta?.block?.hash) {
@@ -444,12 +472,8 @@ export class TapCollector {
             lastId,
             pageSize: PAGE_SIZE,
             block,
-            unfinalizedRavsAllocationIds: ravs.map((value) =>
-              toAddress(value.allocationId).toLowerCase(),
-            ),
-            senderAddresses: ravs.map((value) =>
-              toAddress(value.senderAddress).toLowerCase(),
-            ),
+            unfinalizedRavsAllocationIds,
+            senderAddresses,
           },
         )
 
@@ -568,7 +592,7 @@ export class TapCollector {
         logger.error(`Failed to redeem RAV`, {
           err: indexerError(IndexerErrorCode.IE055, err),
         })
-        return
+        continue
       }
       stopTimer()
     }
@@ -671,7 +695,7 @@ export class TapCollector {
     // https://github.com/sequelize/sequelize/issues/7664 (bug been open for 7 years no fix yet or ever)
     const query = `
             UPDATE scalar_tap_ravs
-            SET redeemed_at = ${timestamp ? timestamp : 'NOW()'}
+            SET redeemed_at = ${timestamp ? `to_timestamp(${timestamp})` : 'NOW()'}
             WHERE allocation_id = '${allocationId
               .toString()
               .toLowerCase()