feat: check escrow balance before rav redeem

packages/indexer-common/src/allocations/__tests__/escrow-accounts.test.ts

@@ -0,0 +1,67 @@
+import { Address, toAddress } from '@graphprotocol/common-ts'
+import { EscrowAccountResponse, EscrowAccounts } from '../escrow-accounts'
+
+const timeout = 30000
+
+const SENDER_ADDRESS_1 = toAddress('ffcf8fdee72ac11b5c542428b35eef5769c409f0')
+const SENDER_ADDRESS_2 = toAddress('dead47df40c29949a75a6693c77834c00b8ad624')
+const SENDER_ADDRESS_3 = toAddress('6aea8894b5ab5a36cdc2d8be9290046801dd5fed')
+
+describe('EscrowAccounts', () => {
+  test(
+    'fromResponse should create correctly EscrowAccount',
+    () => {
+      const response: EscrowAccountResponse = {
+        escrowAccounts: [
+          {
+            sender: {
+              id: SENDER_ADDRESS_1,
+            },
+            balance: '1000',
+          },
+          {
+            sender: {
+              id: SENDER_ADDRESS_2,
+            },
+            balance: '2000',
+          },
+        ],
+      }
+
+      const escrowAccounts = EscrowAccounts.fromResponse(response)
+
+      expect(escrowAccounts.getBalanceForSender(SENDER_ADDRESS_1)).toEqual(1000n)
+      expect(escrowAccounts.getBalanceForSender(SENDER_ADDRESS_2)).toEqual(2000n)
+      expect(() => escrowAccounts.getBalanceForSender(SENDER_ADDRESS_3)).toThrowError()
+    },
+    timeout,
+  )
+  test('test subtractSenderBalance', () => {
+    const balances = new Map<Address, bigint>()
+    balances.set(SENDER_ADDRESS_1, 1000n)
+    balances.set(SENDER_ADDRESS_2, 1000n)
+    balances.set(SENDER_ADDRESS_3, 1000n)
+    const escrowAccounts = new EscrowAccounts(balances)
+
+    expect(escrowAccounts.getBalanceForSender(SENDER_ADDRESS_1)).toEqual(1000n)
+
+    escrowAccounts.subtractSenderBalance(SENDER_ADDRESS_1, 100n)
+    expect(escrowAccounts.getBalanceForSender(SENDER_ADDRESS_1)).toEqual(900n)
+
+    escrowAccounts.subtractSenderBalance(SENDER_ADDRESS_1, 100n)
+    expect(escrowAccounts.getBalanceForSender(SENDER_ADDRESS_1)).toEqual(800n)
+
+    escrowAccounts.subtractSenderBalance(SENDER_ADDRESS_1, 600n)
+    expect(escrowAccounts.getBalanceForSender(SENDER_ADDRESS_1)).toEqual(200n)
+
+    expect(() =>
+      escrowAccounts.subtractSenderBalance(SENDER_ADDRESS_1, 400n),
+    ).toThrowError()
+
+    escrowAccounts.subtractSenderBalance(SENDER_ADDRESS_1, 200n)
+
+    expect(escrowAccounts.getBalanceForSender(SENDER_ADDRESS_1)).toEqual(0n)
+    expect(escrowAccounts.getBalanceForSender(SENDER_ADDRESS_2)).toEqual(1000n)
+    expect(escrowAccounts.getBalanceForSender(SENDER_ADDRESS_3)).toEqual(1000n)
+  })
+})

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

@@ -2,9 +2,11 @@ import {
   defineQueryFeeModels,
   GraphNode,
   Network,
+  EscrowAccounts,
   QueryFeeModels,
   TapSubgraphResponse,
   TapCollector,
+  Allocation,
 } from '@graphprotocol/indexer-common'
 import {
   Address,
@@ -502,6 +504,36 @@ describe('TAP', () => {
     },
     timeout,
   )
+
+  test('test `submitRAVs` with escrow account lower on balance', async () => {
+    // mock redeemRav to not call the blockchain
+    const redeemRavFunc = jest
+      .spyOn(tapCollector, 'redeemRav')
+      .mockImplementation(jest.fn())
+
+    // mock fromResponse to return the correct escrow account
+    // eslint-disable-next-line @typescript-eslint/no-unused-vars
+    jest.spyOn(EscrowAccounts, 'fromResponse').mockImplementation((_) => {
+      const balances = new Map<Address, bigint>()
+      balances.set(SENDER_ADDRESS_1, 40000000000000n)
+      return new EscrowAccounts(balances)
+    })
+
+    const [first] = await queryFeeModels.receiptAggregateVouchers.findAll()
+    const rav = first.getSignedRAV()
+
+    const ravWithAllocation = {
+      rav,
+      allocation: {} as Allocation,
+      sender: first.senderAddress,
+    }
+    const ravs = [ravWithAllocation, ravWithAllocation, ravWithAllocation]
+    // submit 3 ravs
+    await tapCollector['submitRAVs'](ravs)
+    // expect to be able to redeem only 2 of them
+    // because of the balance
+    expect(redeemRavFunc).toBeCalledTimes(2)
+  })
 })
 
 function createLastNonFinalRav(

packages/indexer-common/src/allocations/escrow-accounts.ts

@@ -0,0 +1,67 @@
+import { Address, toAddress } from '@graphprotocol/common-ts'
+import { TAPSubgraph } from '../tap-subgraph'
+import gql from 'graphql-tag'
+
+type U256 = bigint
+
+export type EscrowAccountResponse = {
+  escrowAccounts: {
+    balance: string
+    sender: {
+      id: string
+    }
+  }[]
+}
+
+export class EscrowAccounts {
+  constructor(private sendersBalances: Map<Address, U256>) {}
+
+  getBalanceForSender(sender: Address): U256 {
+    const balance = this.sendersBalances.get(sender)
+    if (balance === undefined) {
+      throw new Error(`No balance found for sender: ${sender}`)
+    }
+    return balance
+  }
+
+  subtractSenderBalance(sender: Address, ravValue: U256) {
+    const balance = this.getBalanceForSender(sender)
+    if (balance < ravValue) {
+      throw new Error(`Negative balances are not allowed`)
+    }
+    const newBalance = balance - ravValue
+    this.sendersBalances.set(sender, newBalance)
+  }
+
+  static fromResponse(response: EscrowAccountResponse): EscrowAccounts {
+    const sendersBalances = new Map<Address, U256>()
+    response.escrowAccounts.forEach((account) => {
+      sendersBalances.set(toAddress(account.sender.id), BigInt(account.balance))
+    })
+
+    return new EscrowAccounts(sendersBalances)
+  }
+}
+
+export const getEscrowAccounts = async (
+  tapSubgraph: TAPSubgraph,
+  indexer: Address,
+): Promise<EscrowAccounts> => {
+  const result = await tapSubgraph.query<EscrowAccountResponse>(
+    gql`
+      query EscrowAccountQuery($indexer: ID!) {
+        escrowAccounts(where: { receiver_: { id: $indexer } }) {
+          balance
+          sender {
+            id
+          }
+        }
+      }
+    `,
+    { indexer },
+  )
+  if (!result.data) {
+    throw `There was an error while querying Tap Subgraph. Errors: ${result.error}`
+  }
+  return EscrowAccounts.fromResponse(result.data)
+}

packages/indexer-common/src/allocations/index.ts

@@ -1,3 +1,4 @@
+export * from './escrow-accounts'
 export * from './keys'
 export * from './query-fees'
 export * from './tap-collector'

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

@@ -29,6 +29,7 @@ import pReduce from 'p-reduce'
 import { TAPSubgraph } from '../tap-subgraph'
 import { NetworkSubgraph, QueryResult } from '../network-subgraph'
 import gql from 'graphql-tag'
+import { getEscrowAccounts } from './escrow-accounts'
 
 // every 15 minutes
 const RAV_CHECK_INTERVAL_MS = 900_000
@@ -60,7 +61,7 @@ interface ValidRavs {
   eligible: RavWithAllocation[]
 }
 
-interface RavWithAllocation {
+export interface RavWithAllocation {
   rav: SignedRAV
   allocation: Allocation
   sender: Address
@@ -109,6 +110,7 @@ export class TapCollector {
   declare tapSubgraph: TAPSubgraph
   declare networkSubgraph: NetworkSubgraph
   declare finalityTime: number
+  declare indexerAddress: Address
 
   // eslint-disable-next-line @typescript-eslint/no-empty-function -- Private constructor to prevent direct instantiation
   private constructor() {}
@@ -138,10 +140,11 @@ export class TapCollector {
     collector.tapSubgraph = tapSubgraph
     collector.networkSubgraph = networkSubgraph
 
-    const { voucherRedemptionThreshold, finalityTime } =
+    const { voucherRedemptionThreshold, finalityTime, address } =
       networkSpecification.indexerOptions
     collector.ravRedemptionThreshold = voucherRedemptionThreshold
     collector.finalityTime = finalityTime
+    collector.indexerAddress = address
 
     collector.logger.info(`RAV processing is initiated`)
     collector.startRAVProcessing()
@@ -526,63 +529,40 @@ export class TapCollector {
       function: 'submitRAVs()',
       ravsToSubmit: signedRavs.length,
     })
-    const escrow = this.tapContracts
 
     logger.info(`Redeem last RAVs on chain individually`, {
       signedRavs,
     })
+    const escrowAccounts = await getEscrowAccounts(this.tapSubgraph, this.indexerAddress)
 
     // Redeem RAV one-by-one as no plual version available
     for (const { rav: signedRav, allocation, sender } of signedRavs) {
       const { rav } = signedRav
+
+      // verify escrow balances
+      const ravValue = BigInt(rav.valueAggregate.toString())
+      const senderBalance = escrowAccounts.getBalanceForSender(sender)
+      if (senderBalance < ravValue) {
+        this.logger.warn(
+          'RAV was not sent to the blockchain \
+          because its value aggregate is lower than escrow balance.',
+          {
+            rav,
+            sender,
+            senderBalance,
+          },
+        )
+        continue
+      }
+
       const stopTimer = this.metrics.ravsRedeemDuration.startTimer({
         allocation: rav.allocationId,
       })
       try {
-        const proof = await tapAllocationIdProof(
-          allocationSigner(this.transactionManager.wallet, allocation),
-          parseInt(this.protocolNetwork.split(':')[1]),
-          sender,
-          toAddress(rav.allocationId),
-          toAddress(escrow.escrow.address),
-        )
-        this.logger.debug(`Computed allocationIdProof`, {
-          allocationId: rav.allocationId,
-          proof,
-        })
-        // Submit the signed RAV on chain
-        const txReceipt = await this.transactionManager.executeTransaction(
-          () => escrow.escrow.estimateGas.redeem(signedRav, proof),
-          (gasLimit) =>
-            escrow.escrow.redeem(signedRav, proof, {
-              gasLimit,
-            }),
-          logger.child({ function: 'redeem' }),
-        )
-
-        // get tx receipt and post process
-        if (txReceipt === 'paused' || txReceipt === 'unauthorized') {
-          this.metrics.ravRedeemsInvalid.inc({ allocation: rav.allocationId })
-          return
-        }
-        this.metrics.ravCollectedFees.set(
-          { allocation: rav.allocationId },
-          parseFloat(rav.valueAggregate.toString()),
-        )
-
-        try {
-          await this.markRavAsRedeemed(toAddress(rav.allocationId), sender)
-          logger.info(
-            `Updated receipt aggregate vouchers table with redeemed_at for allocation ${rav.allocationId} and sender ${sender}`,
-          )
-        } catch (err) {
-          logger.warn(
-            `Failed to update receipt aggregate voucher table with redeemed_at for allocation ${rav.allocationId}`,
-            {
-              err,
-            },
-          )
-        }
+        await this.redeemRav(logger, allocation, sender, signedRav)
+        // subtract from the escrow account
+        // THIS IS A MUT OPERATION
+        escrowAccounts.subtractSenderBalance(sender, ravValue)
       } catch (err) {
         this.metrics.ravRedeemsFailed.inc({ allocation: rav.allocationId })
         logger.error(`Failed to redeem RAV`, {
@@ -625,6 +605,63 @@ export class TapCollector {
     )
   }
 
+  public async redeemRav(
+    logger: Logger,
+    allocation: Allocation,
+    sender: Address,
+    signedRav: SignedRAV,
+  ) {
+    const { rav } = signedRav
+
+    const escrow = this.tapContracts
+
+    const proof = await tapAllocationIdProof(
+      allocationSigner(this.transactionManager.wallet, allocation),
+      parseInt(this.protocolNetwork.split(':')[1]),
+      sender,
+      toAddress(rav.allocationId),
+      toAddress(escrow.escrow.address),
+    )
+    this.logger.debug(`Computed allocationIdProof`, {
+      allocationId: rav.allocationId,
+      proof,
+    })
+    // Submit the signed RAV on chain
+    const txReceipt = await this.transactionManager.executeTransaction(
+      () => escrow.escrow.estimateGas.redeem(signedRav, proof),
+      (gasLimit) =>
+        escrow.escrow.redeem(signedRav, proof, {
+          gasLimit,
+        }),
+      logger.child({ function: 'redeem' }),
+    )
+
+    // get tx receipt and post process
+    if (txReceipt === 'paused' || txReceipt === 'unauthorized') {
+      this.metrics.ravRedeemsInvalid.inc({ allocation: rav.allocationId })
+      return
+    }
+
+    this.metrics.ravCollectedFees.set(
+      { allocation: rav.allocationId },
+      parseFloat(rav.valueAggregate.toString()),
+    )
+
+    try {
+      await this.markRavAsRedeemed(toAddress(rav.allocationId), sender)
+      logger.info(
+        `Updated receipt aggregate vouchers table with redeemed_at for allocation ${rav.allocationId} and sender ${sender}`,
+      )
+    } catch (err) {
+      logger.warn(
+        `Failed to update receipt aggregate voucher table with redeemed_at for allocation ${rav.allocationId}`,
+        {
+          err,
+        },
+      )
+    }
+  }
+
   private async markRavAsRedeemed(
     allocationId: Address,
     senderAddress: Address,