import { formatGRT , join , Logger , Metrics , SubgraphDeploymentID , timer , toAddress , } from '@graphprotocol/common-ts' ; import { allocationRewardsPool , Allocation , AllocationStatus , INDEXING_RULE_GLOBAL , IndexingRuleAttributes , RewardsPool , indexerError , IndexerErrorCode , POIDisputeAttributes , } from '@graphprotocol/indexer-common' ; import { BlockPointer , NetworkSubgraph } from '@graphprotocol/indexer-common' ; import { Indexer } from './indexer' ; import { AgentConfig } from './types' ; import { Network } from './network' ; import { BigNumber , utils } from 'ethers' ; import PQueue from 'p-queue' ; import pMap from 'p-map' ; import pFilter from 'p-filter' ; import { ReceiptCollector } from './query-fees' ; const allocationInList = ( list , allocation ) : boolean => list . find ( item => item . id === allocation . id ) !== undefined ; const deploymentInList = ( list , deployment ) : boolean => list . find ( item => item . bytes32 === deployment . bytes32 ) !== undefined ; const uniqueDeploymentsOnly = ( value , index , array ) : boolean => array . findIndex ( v => value . bytes32 === v . bytes32 ) === index ; const uniqueDeployments = ( deployments ) : SubgraphDeploymentID [ ] => deployments . filter ( uniqueDeploymentsOnly ) ; const deploymentIDSet = ( deployments ) : Set < string > => new Set ( deployments . map ( id => id . display . bytes32 ) ) ; class Agent { logger : Logger ; metrics : Metrics ; indexer : Indexer ; network : Network ; networkSubgraph : NetworkSubgraph ; allocateOnNetworkSubgraph : boolean ; registerIndexer : boolean ; offchainSubgraphs : SubgraphDeploymentID [ ] ; receiptCollector : ReceiptCollector ; constructor ( logger , metrics , indexer , network , networkSubgraph , allocateOnNetworkSubgraph , registerIndexer , offchainSubgraphs , receiptCollector ) { this . logger = logger . child ( { component : 'Agent' } ) ; this . metrics = metrics ; this . indexer = indexer ; this . network = network ; this . networkSubgraph = networkSubgraph ; this . allocateOnNetworkSubgraph = allocateOnNetworkSubgraph ; this . registerIndexer = registerIndexer ; this . offchainSubgraphs = offchainSubgraphs ; this . receiptCollector = receiptCollector ; } async start ( ) { this . logger . info ( `Connect to Graph node(s)` ) ; await this . indexer . connect ( ) ; this . logger . info ( `Connected to Graph node(s)` ) ; await this . indexer . ensureGlobalIndexingRule ( ) ; if ( this . registerIndexer ) { await this . network . register ( ) ; } const currentEpoch = timer ( 600000 ) . tryMap ( ( ) => this . network . contracts . epochManager . currentEpoch ( ) , { onError : err => this . logger . warn ( `Failed to fetch current epoch` , { err } ) , } ) ; const currentEpochStartBlock = currentEpoch . tryMap ( async ( ) => { const startBlockNumber = await this . network . contracts . epochManager . currentEpochBlock ( ) ; const startBlock = await this . network . ethereum . getBlock ( startBlockNumber . toNumber ( ) ) ; return { number : startBlock . number , hash : startBlock . hash , } ; } , { onError : err => this . logger . warn ( `Failed to fetch start block of current epoch` , { err , } ) , } ) ; const channelDisputeEpochs = timer ( 600000 ) . tryMap ( ( ) => this . network . contracts . staking . channelDisputeEpochs ( ) , { onError : err => this . logger . warn ( `Failed to fetch channel dispute epochs` , { err } ) , } ) ; const maxAllocationEpochs = timer ( 600000 ) . tryMap ( ( ) => this . network . contracts . staking . maxAllocationEpochs ( ) , { onError : err => this . logger . warn ( `Failed to fetch max allocation epochs` , { err } ) , } ) ; const indexingRules = timer ( 60000 ) . tryMap ( ( ) => this . indexer . indexingRules ( true ) , { onError : ( ) => this . logger . warn ( `Failed to obtain indexing rules, trying again later` ) , } ) ; const activeDeployments = timer ( 60000 ) . tryMap ( ( ) => this . indexer . subgraphDeployments ( ) , { onError : ( ) => this . logger . warn ( `Failed to obtain active deployments, trying again later` ) , } ) ; const targetDeployments = timer ( 120000 ) . tryMap ( async ( ) => { const rules = await indexingRules . value ( ) ; return rules . length === 0 ? [ ] : await this . network . subgraphDeploymentsWorthIndexing ( rules ) ; } , { onError : ( ) => this . logger . warn ( `Failed to obtain target deployments, trying again later` ) , } ) ; const activeAllocations = timer ( 120000 ) . tryMap ( ( ) => this . network . allocations ( AllocationStatus . Active ) , { onError : ( ) => this . logger . warn ( `Failed to obtain active allocations, trying again later` ) , } ) ; const recentlyClosedAllocations = join ( { activeAllocations , currentEpoch , } ) . tryMap ( ( { activeAllocations , currentEpoch } ) => this . network . recentlyClosedAllocations ( currentEpoch . toNumber ( ) , 1 ) , { onError : ( ) => this . logger . warn ( `Failed to obtain active allocations, trying again later` ) , } ) ; const claimableAllocations = join ( { currentEpoch , channelDisputeEpochs , } ) . tryMap ( ( { currentEpoch , channelDisputeEpochs } ) => this . network . claimableAllocations ( currentEpoch . toNumber ( ) - channelDisputeEpochs ) , { onError : ( ) => this . logger . warn ( `Failed to obtain claimable allocations, trying again later` ) , } ) ; this . logger . info ( `Waiting for network data before reconciling every 120s` ) ; const disputableAllocations = join ( { currentEpoch , activeDeployments , } ) . tryMap ( ( { currentEpoch , activeDeployments } ) => this . network . disputableAllocations ( currentEpoch , activeDeployments , 0 ) , { onError : ( ) => this . logger . warn ( `Failed to fetch disputable allocations, trying again later` ) , } ) ; join ( { ticker : timer ( 120000 ) , paused : this . network . paused , isOperator : this . network . isOperator , currentEpoch , currentEpochStartBlock , maxAllocationEpochs , indexingRules , activeDeployments , targetDeployments , activeAllocations , recentlyClosedAllocations , claimableAllocations , disputableAllocations , } ) . pipe ( async ( { paused , isOperator , currentEpoch , currentEpochStartBlock , maxAllocationEpochs , indexingRules , activeDeployments , targetDeployments , activeAllocations , recentlyClosedAllocations , claimableAllocations , disputableAllocations , } ) => { this . logger . info ( `Reconcile with the network` ) ; if ( paused ) { return this . logger . info ( `The network is currently paused, not doing anything until it resumes` ) ; } if ( ! isOperator ) { return this . logger . error ( `Not authorized as an operator for the indexer` , { err : indexerError ( IndexerErrorCode . IE034 ) , indexer : toAddress ( this . network . indexerAddress ) , operator : toAddress ( this . network . wallet . address ) , } ) ; } try { await this . claimRebateRewards ( claimableAllocations ) ; } catch ( err ) { this . logger . warn ( `Failed to claim rebate rewards` , { err } ) ; } try { const disputableEpoch = currentEpoch . toNumber ( ) - this . network . poiDisputableEpochs ; await this . identifyPotentialDisputes ( disputableAllocations , disputableEpoch ) ; } catch ( err ) { this . logger . warn ( `Failed PoI dispute monitoring` , { err } ) ; } try { await this . reconcileDeployments ( activeDeployments , targetDeployments , [ ... recentlyClosedAllocations , ... activeAllocations ] ) ; await this . reconcileAllocations ( activeAllocations , targetDeployments , indexingRules , currentEpoch . toNumber ( ) , currentEpochStartBlock , maxAllocationEpochs ) ; } catch ( err ) { this . logger . warn ( `Failed to reconcile indexer and network` , { err : indexerError ( IndexerErrorCode . IE005 , err ) , } ) ; } } ) ; return this ; } async claimRebateRewards ( allocations ) { this . logger . info ( `Claim rebate rewards` , { claimable : allocations . map ( allocation => ( { id : allocation . id , deployment : allocation . subgraphDeployment . id . display , createdAtEpoch : allocation . createdAtEpoch , amount : allocation . queryFeeRebates , } ) ) , } ) ; if ( allocations . length > 0 ) { await this . network . claimMany ( allocations ) ; } } async identifyPotentialDisputes ( disputableAllocations , disputableEpoch ) { const alreadyProcessed = ( await this . indexer . fetchPOIDisputes ( 'potential' , disputableEpoch ) ) . concat ( await this . indexer . fetchPOIDisputes ( 'valid' , disputableEpoch ) ) ; const newDisputableAllocations = disputableAllocations . filter ( allocation => ! alreadyProcessed . find ( dispute => dispute . allocationID == allocation . id ) ) ; if ( newDisputableAllocations . length == 0 ) { this . logger . debug ( 'No new disputable allocations to process for potential disputes' ) ; return ; } this . logger . debug ( `Found new allocations onchain for subgraphs we have indexed. Let's compare PoIs to identify any potential indexing disputes` ) ; const uniqueRewardsPools = await Promise . all ( [ ... new Set ( newDisputableAllocations . map ( allocation => allocationRewardsPool ( allocation ) ) ) , ] . filter ( pool => pool . closedAtEpochStartBlockHash ) . map ( async ( pool ) => { const closedAtEpochStartBlock = await this . network . ethereum . getBlock ( pool . closedAtEpochStartBlockHash ! ) ; const previousEpochStartBlock = await this . network . ethereum . getBlock ( pool . previousEpochStartBlockHash ! ) ; pool . closedAtEpochStartBlockNumber = closedAtEpochStartBlock . number ; pool . referencePOI = await this . indexer . proofOfIndexing ( pool . subgraphDeployment , { number : closedAtEpochStartBlock . number , hash : closedAtEpochStartBlock . hash , } , pool . allocationIndexer ) ; pool . previousEpochStartBlockHash = previousEpochStartBlock . hash ; pool . previousEpochStartBlockNumber = previousEpochStartBlock . number ; pool . referencePreviousPOI = await this . indexer . proofOfIndexing ( pool . subgraphDeployment , { number : previousEpochStartBlock . number , hash : previousEpochStartBlock . hash , } , pool . allocationIndexer ) ; return pool ; } ) ) ; const disputes = newDisputableAllocations . map ( ( allocation ) => { const rewardsPool = uniqueRewardsPools . find ( pool => pool . subgraphDeployment == allocation . subgraphDeployment . id && pool . closedAtEpoch == allocation . closedAtEpoch ) ; if ( ! rewardsPool ) { throw Error ( `No rewards pool found for deployment ${ allocation . subgraphDeployment . id }` ) ; } let status = rewardsPool ! . referencePOI == allocation . poi || rewardsPool ! . referencePreviousPOI == allocation . poi ? 'valid' : 'potential' ; if ( status === 'potential' && ( ! rewardsPool . referencePOI || ! rewardsPool . referencePreviousPOI ) ) { status = 'reference_unavailable' ; } return { allocationID : allocation . id , subgraphDeploymentID : allocation . subgraphDeployment . id . ipfsHash , allocationIndexer : allocation . indexer , allocationAmount : allocation . allocatedTokens . toString ( ) , allocationProof : allocation . poi ! , closedEpoch : allocation . closedAtEpoch , closedEpochReferenceProof : rewardsPool ! . referencePOI , closedEpochStartBlockHash : allocation . closedAtEpochStartBlockHash ! , closedEpochStartBlockNumber : rewardsPool ! . closedAtEpochStartBlockNumber ! , previousEpochReferenceProof : rewardsPool ! . referencePreviousPOI , previousEpochStartBlockHash : rewardsPool ! . previousEpochStartBlockHash ! , previousEpochStartBlockNumber : rewardsPool ! . previousEpochStartBlockNumber ! , status , } as POIDisputeAttributes ; } ) ; const potentialDisputes = disputes . filter ( dispute => dispute . status == 'potential' ) . length ; const stored = await this . indexer . storePoiDisputes ( disputes ) ; this . logger . info ( `Disputable allocations' PoIs validated` , { potentialDisputes : potentialDisputes , validAllocations : stored . length - potentialDisputes , } ) ; } async reconcileDeployments ( activeDeployments , targetDeployments , eligibleAllocations ) { activeDeployments = uniqueDeployments ( activeDeployments ) ; targetDeployments = uniqueDeployments ( targetDeployments ) ; const eligibleAllocationDeployments = uniqueDeployments ( eligibleAllocations . map ( allocation => allocation . subgraphDeployment . id ) ) ; if ( this . networkSubgraph . deployment ) { if ( ! deploymentInList ( targetDeployments , this . networkSubgraph . deployment . id ) ) { targetDeployments . push ( this . networkSubgraph . deployment . id ) ; } } for ( const offchainSubgraph of this . offchainSubgraphs ) { if ( ! deploymentInList ( targetDeployments , offchainSubgraph ) ) { targetDeployments . push ( offchainSubgraph ) ; } } if ( deploymentIDSet ( activeDeployments ) != deploymentIDSet ( targetDeployments ) ) { this . logger . info ( 'Reconcile deployments' , { active : activeDeployments . map ( id => id . display ) , target : targetDeployments . map ( id => id . display ) , } ) ; } const deploy = targetDeployments . filter ( deployment => ! deploymentInList ( activeDeployments , deployment ) ) ; const remove = activeDeployments . filter ( deployment => ! deploymentInList ( targetDeployments , deployment ) && ! deploymentInList ( eligibleAllocationDeployments , deployment ) ) ; this . logger . info ( 'Deployment changes' , { deploy : deploy . map ( id => id . display ) , remove : remove . map ( id => id . display ) , } ) ; const queue = new PQueue ( { concurrency : 10 } ) ; await queue . addAll ( deploy . map ( deployment => async ( ) => { const name = `indexer-agent/${ deployment . ipfsHash . slice ( - 10 ) }` ; this . logger . info ( `Index subgraph deployment` , { name , deployment : deployment . display , } ) ; this . indexer . ensure ( name , deployment ) ; } ) ) ; await queue . addAll ( remove . map ( deployment => async ( ) => this . indexer . remove ( deployment ) ) ) ; await queue . onIdle ( ) ; } async reconcileAllocations ( activeAllocations , targetDeployments , rules , currentEpoch , currentEpochStartBlock , maxAllocationEpochs ) { const allocationLifetime = Math . max ( 1 , maxAllocationEpochs - 1 ) ; this . logger . info ( `Reconcile allocations` , { currentEpoch , maxAllocationEpochs , allocationLifetime , targetDeployments , active : activeAllocations . map ( allocation => ( { id : allocation . id , deployment : allocation . subgraphDeployment . id . display , createdAtEpoch : allocation . createdAtEpoch , } ) ) , } ) ; const deployments = uniqueDeployments ( [ ... targetDeployments , ... activeAllocations . map ( allocation => allocation . subgraphDeployment . id ) , ] ) ; if ( ! this . allocateOnNetworkSubgraph && this . networkSubgraph . deployment ?. id . bytes32 ) { const networkSubgraphDeploymentId = this . networkSubgraph . deployment . id ; targetDeployments = targetDeployments . filter ( deployment => deployment . bytes32 !== networkSubgraphDeploymentId . bytes32 ) ; } this . logger . debug ( `Deployments to reconcile allocations for` , { number : deployments . length , deployments : deployments . map ( deployment => deployment . display ) , } ) ; await pMap ( deployments , async ( deployment ) => { await this . reconcileDeploymentAllocations ( deployment , activeAllocations . filter ( allocation => allocation . subgraphDeployment . id . bytes32 === deployment . bytes32 ) , deploymentInList ( targetDeployments , deployment ) , rules . find ( rule => rule . deployment === deployment . bytes32 ) || rules . find ( rule => rule . deployment === INDEXING_RULE_GLOBAL ) , currentEpoch , currentEpochStartBlock , maxAllocationEpochs ) ; } , { concurrency : 1 } ) ; } async reconcileDeploymentAllocations ( deployment , activeAllocations , worthAllocating , rule , epoch , epochStartBlock , maxAllocationEpochs ) { const logger = this . logger . child ( { deployment : deployment . display , epoch , } ) ; const desiredAllocationAmount = rule ?. allocationAmount ? BigNumber . from ( rule . allocationAmount ) : this . indexer . defaultAllocationAmount ; const desiredNumberOfAllocations = 1 ; const activeAllocationAmount = activeAllocations . reduce ( ( sum , allocation ) => sum . add ( allocation . allocatedTokens ) , BigNumber . from ( '0' ) ) ; if ( desiredAllocationAmount !== activeAllocationAmount || desiredNumberOfAllocations !== activeAllocations . length ) { logger . info ( `Reconcile deployment allocations for deployment '${ deployment . ipfsHash }'` , { desiredAllocationAmount : formatGRT ( desiredAllocationAmount ) , totalActiveAllocationAmount : formatGRT ( activeAllocationAmount ) , desiredNumberOfAllocations , activeNumberOfAllocations : activeAllocations . length , activeAllocations : activeAllocations . map ( allocation => ( { id : allocation . id , createdAtEpoch : allocation . createdAtEpoch , amount : formatGRT ( allocation . allocatedTokens ) , } ) ) , } ) ; } if ( ! worthAllocating ) { logger . info ( `Deployment is not (or no longer) worth allocating towards, close allocation if it is from a previous epoch` , { eligibleForClose : activeAllocations . filter ( allocation => allocation . createdAtEpoch < epoch ) . map ( allocation => allocation . id ) , } ) ; if ( activeAllocations . length > 0 ) { await pMap ( activeAllocations . filter ( allocation => allocation . createdAtEpoch < epoch ) , async ( allocation ) => { await this . closeAllocation ( epochStartBlock , allocation ) ; } , { concurrency : 1 } ) ; } return ; } if ( activeAllocations . length === 0 ) { logger . info ( `No active allocation for deployment, creating one now` , { allocationAmount : formatGRT ( desiredAllocationAmount ) , } ) ; const allocationsCreated = await this . network . allocate ( deployment , desiredAllocationAmount , activeAllocations ) ; if ( allocationsCreated ) { await this . receiptCollector . rememberAllocations ( [ allocationsCreated ] ) ; } return ; } const allocationsToRemove = activeAllocations . length - desiredNumberOfAllocations ; if ( allocationsToRemove > 0 ) { logger . info ( `Close allocations to maintain desired number of allocations` , { desiredNumberOfAllocations , activeAllocations : activeAllocations . length , allocationsToRemove : allocationsToRemove , allocationAmount : formatGRT ( desiredAllocationAmount ) , } ) ; await pMap ( activeAllocations . filter ( allocation => allocation . createdAtEpoch < epoch ) . sort ( ( a , b ) => a . createdAtEpoch - b . closedAtEpoch ) . splice ( 0 , allocationsToRemove ) , async ( allocation ) => { await this . closeAllocation ( epochStartBlock , allocation ) ; } , { concurrency : 1 } ) ; } const lifetime = Math . max ( 1 , maxAllocationEpochs - 1 ) ; let expiredAllocations = activeAllocations . filter ( allocation => epoch >= allocation . createdAtEpoch + lifetime ) ; expiredAllocations = await pFilter ( expiredAllocations , async ( allocation ) => { try { const onChainAllocation = await this . network . contracts . staking . getAllocation ( allocation . id ) ; return onChainAllocation . closedAtEpoch . eq ( '0' ) ; } catch ( err ) { this . logger . warn ( `Failed to cross-check allocation state with contracts; assuming it needs to be closed` , { deployment : deployment . display , allocation : allocation . id , err : indexerError ( IndexerErrorCode . IE006 , err ) , } ) ; return true ; } } ) ; if ( expiredAllocations . length > 0 ) { logger . info ( `Reallocating expired allocations` , { number : expiredAllocations . length , expiredAllocations : expiredAllocations . map ( allocation => allocation . id ) , } ) ; for ( let i = 0 ; i <= activeAllocations . length - 1 ; i ++ ) { const oldAllocation = activeAllocations [ i ] ; if ( allocationInList ( expiredAllocations , oldAllocation ) ) { const { newAllocation , reallocated } = await this . reallocate ( epochStartBlock , oldAllocation , desiredAllocationAmount , activeAllocations ) ; if ( reallocated ) { activeAllocations [ i ] = newAllocation as Allocation ; } } } } } private async closeAllocation ( epochStartBlock , allocation ) { const poi = await this . indexer . proofOfIndexing ( allocation . subgraphDeployment . id , epochStartBlock , this . indexer . indexerAddress ) ; if ( poi === undefined || poi === null || poi === utils . hexlify ( Array ( 32 ) . fill ( 0 ) ) ) { const indexingStatus = await this . indexer . indexingStatus ( allocation . subgraphDeployment . id ) ; const fatalError = indexingStatus . fatalError ; if ( ! fatalError ) { this . logger . error ( `Received a null or zero POI for deployment, no fatal errors` , { deployment : allocation . subgraphDeployment . id . display , allocation : allocation . id , block : indexingStatus . chains [ 0 ] . latestBlock , } ) ; } else { const latestValidPoi = await this . indexer . proofOfIndexing ( allocation . subgraphDeployment . id , indexingStatus . chains [ 0 ] . latestBlock , this . indexer . indexerAddress ) ; this . logger . error ( `Received a null or zero POI for deployment` , { deployment : allocation . subgraphDeployment . id . display , allocation : allocation . id , fatalError , latestValidPoi , } ) ; } return { closed : false , collectingQueryFees : false } ; } const closed = await this . network . close ( allocation , poi ) ; const collectingQueryFees = await this . receiptCollector . collectReceipts ( allocation ) ; return { closed , collectingQueryFees } ; } private async reallocate ( epochStartBlock , existingAllocation , allocationAmount , activeAllocations ) { const poi = await this . indexer . proofOfIndexing ( existingAllocation . subgraphDeployment . id , epochStartBlock , this . indexer . indexerAddress ) ; if ( poi === undefined || poi === null || poi === utils . hexlify ( Array ( 32 ) . fill ( 0 ) ) ) { this . logger . error ( `Received a null or zero POI for deployment` , { deployment : existingAllocation . subgraphDeployment . id . display , allocation : existingAllocation . id , epochStartBlock , } ) ; return { reallocated : false , collectingQueryFees : false , newAllocation : undefined , } ; } const newAllocation = await this . network . closeAndAllocate ( existingAllocation , poi , existingAllocation . subgraphDeployment . id , allocationAmount , activeAllocations ) ; const collectingQueryFees = await this . receiptCollector . collectReceipts ( existingAllocation ) ; return { reallocated : newAllocation !== undefined , collectingQueryFees , newAllocation , } ; } } export const startAgent = async ( config ) : Promise < Agent > => { const agent = new Agent ( config . logger , config . metrics , config . indexer , config . network , config . networkSubgraph , config . allocateOnNetworkSubgraph , config . registerIndexer , config . offchainSubgraphs , config . receiptCollector ) ; return await agent . start ( ) ; } ;