chore: collect all dips receipts periodically

Author: pcarranzav

packages/indexer-agent/src/__tests__/indexer.ts

@@ -146,6 +146,7 @@ const setup = async () => {
   const network = await Network.create(
     logger,
     networkSpecification,
+    models,
     queryFeeModels,
     graphNode,
     metrics,

packages/indexer-agent/src/commands/start.ts

@@ -612,6 +612,7 @@ export async function run(
   const network = await Network.create(
     logger,
     networkSpecification,
+    managementModels,
     queryFeeModels,
     graphNode,
     metrics,

packages/indexer-cli/src/__tests__/util.ts

@@ -87,6 +87,7 @@ export const setup = async () => {
   const network = await Network.create(
     logger,
     testNetworkSpecification,
+    models,
     queryFeeModels,
     graphNode,
     metrics,

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

@@ -7,6 +7,7 @@ import {
   TapSubgraphResponse,
   TapCollector,
   Allocation,
+  defineIndexerManagementModels,
 } from '@graphprotocol/indexer-common'
 import {
   Address,
@@ -43,6 +44,7 @@ const setup = async () => {
   // Clearing the registry prevents duplicate metric registration in the default registry.
   metrics.registry.clear()
   sequelize = await connectDatabase(__DATABASE__)
+  const models = defineIndexerManagementModels(sequelize)
   queryFeeModels = defineQueryFeeModels(sequelize)
   sequelize = await sequelize.sync({ force: true })
 
@@ -56,6 +58,7 @@ const setup = async () => {
   const network = await Network.create(
     logger,
     testNetworkSpecification,
+    models,
     queryFeeModels,
     graphNode,
     metrics,

packages/indexer-common/src/allocations/__tests__/validate-queries.test.ts

@@ -1,4 +1,5 @@
 import {
+  defineIndexerManagementModels,
   defineQueryFeeModels,
   GraphNode,
   Network,
@@ -36,6 +37,7 @@ const setup = async () => {
   // Clearing the registry prevents duplicate metric registration in the default registry.
   metrics.registry.clear()
   sequelize = await connectDatabase(__DATABASE__)
+  const models = defineIndexerManagementModels(sequelize)
   queryFeeModels = defineQueryFeeModels(sequelize)
   sequelize = await sequelize.sync({ force: true })
 
@@ -49,6 +51,7 @@ const setup = async () => {
   const network = await Network.create(
     logger,
     testNetworkSpecification,
+    models,
     queryFeeModels,
     graphNode,
     metrics,

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

@@ -61,6 +61,7 @@ const setup = async () => {
   const network = await Network.create(
     logger,
     testNetworkSpecification,
+    managementModels,
     queryFeeModels,
     graphNode,
     metrics,

packages/indexer-common/src/indexer-management/__tests__/util.ts

@@ -56,6 +56,7 @@ export const createTestManagementClient = async (
   const network = await Network.create(
     logger,
     networkSpecification,
+    managementModels,
     queryFeeModels,
     graphNode,
     metrics,

packages/indexer-common/src/indexing-fees/dips.ts

@@ -7,7 +7,10 @@ import {
   IndexingDecisionBasis,
   IndexingRuleAttributes,
   Network,
+  QueryFeeModels,
+  sequentialTimerMap,
   SubgraphIdentifierType,
+  TapCollector,
   upsertIndexingRule,
 } from '@graphprotocol/indexer-common'
 import { Op } from 'sequelize'
@@ -23,6 +26,10 @@ import {
   GatewayDipsServiceClientImpl,
 } from '@graphprotocol/dips-proto/generated/gateway'
 import { IndexingAgreement } from '../indexer-management/models/indexing-agreement'
+import { NetworkSpecification } from '../network-specification'
+import { Wallet } from 'ethers'
+
+const DIPS_COLLECTION_INTERVAL = 60_000
 
 export class DipsManager {
   private gatewayDipsServiceClient: GatewayDipsServiceClientImpl
@@ -86,9 +93,109 @@ export class DipsManager {
       await agreement.save()
     }
   }
+  async ensureAgreementRules() {
+    if (!this.parent) {
+      this.logger.error(
+        'DipsManager has no parent AllocationManager, cannot ensure agreement rules',
+      )
+      return
+    }
+    // Get all the indexing agreements that are not cancelled
+    const indexingAgreements = await this.models.IndexingAgreement.findAll({
+      where: {
+        cancelled_at: null,
+      },
+    })
+    // For each agreement, check that there is an indexing rule to always
+    // allocate to the agreement's subgraphDeploymentId, and if not, create one
+    for (const agreement of indexingAgreements) {
+      const subgraphDeploymentID = new SubgraphDeploymentID(
+        agreement.subgraph_deployment_id,
+      )
+      // If there is not yet an indexingRule that deems this deployment worth allocating to, make one
+      if (!(await this.parent.matchingRuleExists(this.logger, subgraphDeploymentID))) {
+        this.logger.debug(`Creating indexing rule for agreement ${agreement.id}`)
+        const indexingRule = {
+          identifier: agreement.subgraph_deployment_id,
+          allocationAmount: formatGRT(
+            this.network.specification.indexerOptions.dipsAllocationAmount,
+          ),
+          identifierType: SubgraphIdentifierType.DEPLOYMENT,
+          decisionBasis: IndexingDecisionBasis.ALWAYS,
+          protocolNetwork: this.network.specification.networkIdentifier,
+          autoRenewal: true,
+          allocationLifetime: Math.max(
+            Number(agreement.min_epochs_per_collection),
+            Number(agreement.max_epochs_per_collection) -
+              this.network.specification.indexerOptions.dipsEpochsMargin,
+          ),
+        } as Partial<IndexingRuleAttributes>
+
+        await upsertIndexingRule(this.logger, this.models, indexingRule)
+      }
+    }
+  }
+}
+
+export class DipsCollector {
+  private gatewayDipsServiceClient: GatewayDipsServiceClientImpl
+  constructor(
+    private logger: Logger,
+    private managementModels: IndexerManagementModels,
+    private queryFeeModels: QueryFeeModels,
+    private specification: NetworkSpecification,
+    private tapCollector: TapCollector,
+    private wallet: Wallet,
+  ) {
+    if (!this.specification.indexerOptions.dipperEndpoint) {
+      throw new Error('dipperEndpoint is not set')
+    }
+    this.gatewayDipsServiceClient = createGatewayDipsServiceClient(
+      this.specification.indexerOptions.dipperEndpoint,
+    )
+  }
+
+  static create(
+    logger: Logger,
+    managementModels: IndexerManagementModels,
+    queryFeeModels: QueryFeeModels,
+    specification: NetworkSpecification,
+    tapCollector: TapCollector,
+    wallet: Wallet,
+  ) {
+    const collector = new DipsCollector(
+      logger,
+      managementModels,
+      queryFeeModels,
+      specification,
+      tapCollector,
+      wallet,
+    )
+    collector.startCollectionLoop()
+    return collector
+  }
+
+  startCollectionLoop() {
+    sequentialTimerMap(
+      {
+        logger: this.logger,
+        milliseconds: DIPS_COLLECTION_INTERVAL,
+      },
+      async () => {
+        this.logger.debug('Running DIPS payment collection loop')
+        await this.collectAllPayments()
+      },
+      {
+        onError: (err) => {
+          this.logger.error('Failed to collect DIPS payments', { err })
+        },
+      },
+    )
+  }
+
   // Collect payments for all outstanding agreements
   async collectAllPayments() {
-    const outstandingAgreements = await this.models.IndexingAgreement.findAll({
+    const outstandingAgreements = await this.managementModels.IndexingAgreement.findAll({
       where: {
         last_payment_collected_at: null,
         last_allocation_id: {
@@ -110,27 +217,27 @@ export class DipsManager {
       agreement.id,
       agreement.last_allocation_id,
       entityCount,
-      this.network.wallet,
+      this.wallet,
     )
     try {
       const response = await this.gatewayDipsServiceClient.CollectPayment({
         version: 1,
         signedCollection: collection,
       })
       if (response.status === CollectPaymentStatus.ACCEPT) {
-        if (!this.network.tapCollector) {
+        if (!this.tapCollector) {
           throw new Error('TapCollector not initialized')
         }
         // Store the tap receipt in the database
         const tapReceipt = decodeTapReceipt(
           response.tapReceipt,
-          this.network.tapCollector?.tapContracts.tapVerifier.address,
+          this.tapCollector?.tapContracts.tapVerifier.address,
         )
         // TODO: check that the signer of the TAP receipt is a signer
         // on the corresponding escrow account for the payer (sender) of the
         // indexing agreement
         const escrowSender = await getEscrowSenderForSigner(
-          this.network.tapCollector?.tapSubgraph,
+          this.tapCollector?.tapSubgraph,
           tapReceipt.signer_address,
         )
         if (escrowSender !== agreement.payer) {
@@ -139,9 +246,9 @@ export class DipsManager {
             'Signer of TAP receipt is not a signer on the indexing agreement',
           )
         }
-        await this.network.queryFeeModels.scalarTapReceipts.create(tapReceipt)
+        await this.queryFeeModels.scalarTapReceipts.create(tapReceipt)
         // Mark the agreement as having had a payment collected
-        await this.models.IndexingAgreement.update(
+        await this.managementModels.IndexingAgreement.update(
           {
             last_payment_collected_at: new Date(),
           },
@@ -160,46 +267,4 @@ export class DipsManager {
       })
     }
   }
-  async ensureAgreementRules() {
-    if (!this.parent) {
-      this.logger.error(
-        'DipsManager has no parent AllocationManager, cannot ensure agreement rules',
-      )
-      return
-    }
-    // Get all the indexing agreements that are not cancelled
-    const indexingAgreements = await this.models.IndexingAgreement.findAll({
-      where: {
-        cancelled_at: null,
-      },
-    })
-    // For each agreement, check that there is an indexing rule to always
-    // allocate to the agreement's subgraphDeploymentId, and if not, create one
-    for (const agreement of indexingAgreements) {
-      const subgraphDeploymentID = new SubgraphDeploymentID(
-        agreement.subgraph_deployment_id,
-      )
-      // If there is not yet an indexingRule that deems this deployment worth allocating to, make one
-      if (!(await this.parent.matchingRuleExists(this.logger, subgraphDeploymentID))) {
-        this.logger.debug(`Creating indexing rule for agreement ${agreement.id}`)
-        const indexingRule = {
-          identifier: agreement.subgraph_deployment_id,
-          allocationAmount: formatGRT(
-            this.network.specification.indexerOptions.dipsAllocationAmount,
-          ),
-          identifierType: SubgraphIdentifierType.DEPLOYMENT,
-          decisionBasis: IndexingDecisionBasis.ALWAYS,
-          protocolNetwork: this.network.specification.networkIdentifier,
-          autoRenewal: true,
-          allocationLifetime: Math.max(
-            Number(agreement.min_epochs_per_collection),
-            Number(agreement.max_epochs_per_collection) -
-              this.network.specification.indexerOptions.dipsEpochsMargin,
-          ),
-        } as Partial<IndexingRuleAttributes>
-
-        await upsertIndexingRule(this.logger, this.models, indexingRule)
-      }
-    }
-  }
 }

packages/indexer-common/src/network.ts

@@ -29,12 +29,14 @@ import {
   AllocationReceiptCollector,
   SubgraphFreshnessChecker,
   monitorEligibleAllocations,
+  IndexerManagementModels,
 } from '.'
 import { resolveChainId } from './indexer-management'
 import { monitorEthBalance } from './utils'
 import { QueryFeeModels } from './query-fees'
 import { readFileSync } from 'fs'
 import { TapCollector } from './allocations/tap-collector'
+import { DipsCollector } from './indexing-fees/dips'
 
 export class Network {
   logger: Logger
@@ -49,10 +51,12 @@ export class Network {
   receiptCollector: AllocationReceiptCollector | undefined
 
   tapCollector: TapCollector | undefined
+  dipsCollector: DipsCollector | undefined
   specification: spec.NetworkSpecification
   paused: Eventual<boolean>
   isOperator: Eventual<boolean>
   queryFeeModels: QueryFeeModels
+  managementModels: IndexerManagementModels
   private constructor(
     logger: Logger,
     contracts: NetworkContracts,
@@ -67,6 +71,8 @@ export class Network {
     paused: Eventual<boolean>,
     isOperator: Eventual<boolean>,
     queryFeeModels: QueryFeeModels,
+    managementModels: IndexerManagementModels,
+    dipsCollector: DipsCollector | undefined,
   ) {
     this.logger = logger
     this.contracts = contracts
@@ -81,11 +87,14 @@ export class Network {
     this.paused = paused
     this.isOperator = isOperator
     this.queryFeeModels = queryFeeModels
+    this.managementModels = managementModels
+    this.dipsCollector = dipsCollector
   }
 
   static async create(
     parentLogger: Logger,
     specification: spec.NetworkSpecification,
+    managementModels: IndexerManagementModels,
     queryFeeModels: QueryFeeModels,
     graphNode: GraphNode,
     metrics: Metrics,
@@ -331,6 +340,23 @@ export class Network {
         Tap Subgraph: ${!!tapSubgraph}.`)
     }
 
+    let dipsCollector: DipsCollector | undefined = undefined
+    if (specification.indexerOptions.enableDips) {
+      if (!tapCollector) {
+        throw new Error(
+          'TapCollector is not initialized, cannot initialize DipsCollector',
+        )
+      }
+      dipsCollector = DipsCollector.create(
+        logger,
+        managementModels,
+        queryFeeModels,
+        specification,
+        tapCollector,
+        wallet,
+      )
+    }
+
     // --------------------------------------------------------------------------------
     // * Network
     // --------------------------------------------------------------------------------
@@ -348,6 +374,8 @@ export class Network {
       paused,
       isOperator,
       queryFeeModels,
+      managementModels,
+      dipsCollector,
     )
   }