indexer-common,-agent: Various robustness and UX improvements

- Do not add action to queue if identical action was recently successful
- Do not add action to queue if identical action failed within last
15 minutes
- Use retries for epoch subgraph queries (5 attempts)
- Improve clarity of code to add offchain subgraphs to targetDeployments
- Improve ActionManager.fetchActions
- Improve code clarity and logging on failed indexing status query

Author: fordN

packages/indexer-agent/src/agent.ts

@@ -310,25 +310,28 @@ class Agent {
       indexingRules,
       networkDeploymentAllocationDecisions,
     }).tryMap(
-      async target => {
-        const rules = await indexingRules.value()
+      async ({ indexingRules, networkDeploymentAllocationDecisions }) => {
+        const rules = indexingRules
         const targetDeploymentIDs = new Set(
-          target.networkDeploymentAllocationDecisions
+          networkDeploymentAllocationDecisions
             .filter(decision => decision.toAllocate === true)
             .map(decision => decision.deployment),
         )
 
         // add offchain subgraphs to the deployment list
+        // from rules
         rules
           .filter(
             rule => rule?.decisionBasis === IndexingDecisionBasis.OFFCHAIN,
           )
           .map(rule => {
             targetDeploymentIDs.add(new SubgraphDeploymentID(rule.identifier))
           })
-        return rules.length === 0
-          ? []
-          : [...targetDeploymentIDs, ...this.offchainSubgraphs]
+        // from startup args
+        this.offchainSubgraphs.map(deployment => {
+          targetDeploymentIDs.add(deployment)
+        })
+        return [...targetDeploymentIDs]
       },
       {
         onError: error =>
@@ -505,9 +508,12 @@ class Agent {
             maxAllocationEpochs,
           )
         } catch (err) {
-          this.logger.warn(`Failed to reconcile indexer and network`, {
-            err: indexerError(IndexerErrorCode.IE005, err),
-          })
+          this.logger.warn(
+            `Exited early while reconciling deployments/allocations`,
+            {
+              err: indexerError(IndexerErrorCode.IE005, err),
+            },
+          )
         }
       },
     )

packages/indexer-agent/src/indexer.ts

@@ -18,7 +18,6 @@ import {
   SubgraphIdentifierType,
   parseGraphQLIndexingStatus,
   CostModelAttributes,
-  ActionFilter,
   ActionResult,
   ActionItem,
   Action,
@@ -28,6 +27,7 @@ import {
   ActionType,
   Allocation,
   AllocationDecision,
+  ActionFilter,
 } from '@graphprotocol/indexer-common'
 import { CombinedError } from '@urql/core'
 import pMap from 'p-map'

packages/indexer-common/src/actions.ts

@@ -1,6 +1,8 @@
 import { ActionManager, NetworkMonitor } from './indexer-management'
 import { AllocationStatus } from './allocations'
-import { WhereOperators } from 'sequelize'
+import { WhereOperators, WhereOptions } from 'sequelize'
+import { Op } from 'sequelize'
+import { WhereAttributeHashValue } from 'sequelize/types/model'
 
 export interface ActionParamsInput {
   deploymentID?: string
@@ -123,11 +125,25 @@ export const validateActionInputs = async (
 }
 
 export interface ActionFilter {
-  type?: ActionType | undefined
-  status?: ActionStatus | undefined
-  source?: string | undefined
-  reason?: string | undefined
-  updatedAt?: WhereOperators | undefined
+  type?: ActionType
+  status?: ActionStatus
+  source?: string
+  reason?: string
+  updatedAt?: WhereOperators
+}
+
+export const actionFilterToWhereOptions = (filter: ActionFilter): WhereOptions => {
+  const whereOptions = [] as WhereAttributeHashValue<any>[]
+
+  Object.entries(filter).forEach(([key, value]) => {
+    if (value) {
+      const obj: { [key: string]: any } = {}
+      obj[key] = value
+      whereOptions.push(obj)
+    }
+  })
+
+  return whereOptions.length == 0 ? {} : { [Op.and]: whereOptions }
 }
 
 export interface ActionResult {

packages/indexer-common/src/errors.ts

@@ -82,6 +82,7 @@ export enum IndexerErrorCode {
   IE069 = 'IE069',
   IE070 = 'IE070',
   IE071 = 'IE071',
+  IE072 = 'IE072',
 }
 
 export const INDEXER_ERROR_MESSAGES: Record<IndexerErrorCode, string> = {
@@ -157,6 +158,7 @@ export const INDEXER_ERROR_MESSAGES: Record<IndexerErrorCode, string> = {
   IE069: 'Failed to query Epoch Block Oracle Subgraph',
   IE070: 'Failed to query latest valid epoch and block hash',
   IE071: 'Add Epoch subgraph support for non-protocol chains',
+  IE072: 'Failed to execute batch tx (contract: staking)',
 }
 
 export type IndexerErrorCause = unknown

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

@@ -4,11 +4,12 @@ import {
   IndexerManagementModels,
   IndexingDecisionBasis,
   IndexingRuleAttributes,
-  SubgraphIdentifierType,
-  fetchIndexingRules,
-  upsertIndexingRule,
-} from '@graphprotocol/indexer-common'
-import { Sequelize } from 'sequelize'
+} from '../models'
+import { fetchIndexingRules, upsertIndexingRule } from '../rules'
+import { SubgraphIdentifierType } from '../../subgraphs'
+import { ActionManager } from '../actions'
+import { actionFilterToWhereOptions, ActionStatus, ActionType } from '../../actions'
+import { literal, Op, Sequelize } from 'sequelize'
 
 // Make global Jest variable available
 // eslint-disable-next-line @typescript-eslint/no-explicit-any
@@ -29,7 +30,7 @@ describe('Indexing Rules', () => {
   beforeAll(setupModels)
   test('Insert and fetch indexing rule', async () => {
     const logger = createLogger({
-      name: 'POI dispute tests',
+      name: 'Indexing rule helpers tests',
       async: false,
       level: __LOG_LEVEL__ ?? 'error',
     })
@@ -59,3 +60,88 @@ describe('Indexing Rules', () => {
     await expect(fetchIndexingRules(models, false)).resolves.toHaveLength(1)
   })
 })
+
+describe('Actions', () => {
+  beforeAll(setupModels)
+
+  test('Generate where options', async () => {
+    const ActionFilter = {
+      status: ActionStatus.FAILED,
+      type: ActionType.ALLOCATE,
+    }
+    expect(actionFilterToWhereOptions(ActionFilter)).toEqual({
+      [Op.and]: [{ status: 'failed' }, { type: 'allocate' }],
+    })
+
+    const yesterday = literal("NOW() - INTERVAL '1d'")
+    const ActionFilter2 = {
+      status: ActionStatus.FAILED,
+      type: ActionType.ALLOCATE,
+      updatedAt: { [Op.gte]: yesterday },
+    }
+
+    const where = actionFilterToWhereOptions(ActionFilter2)
+    expect(where).toEqual({
+      [Op.and]: [
+        { status: 'failed' },
+        { type: 'allocate' },
+        { updatedAt: { [Op.gte]: yesterday } },
+      ],
+    })
+
+    await expect(
+      models.Action.findAll({
+        where,
+      }),
+    ).resolves.toHaveLength(0)
+  })
+
+  test('Insert and fetch actions', async () => {
+    const action = {
+      status: ActionStatus.FAILED,
+      type: ActionType.ALLOCATE,
+      deploymentID: 'QmQ44hgrWWt3Qf2X9XEX2fPyTbmQbChxwNm5c1t4mhKpGt',
+      amount: '10000',
+      force: false,
+      source: 'indexerAgent',
+      reason: 'indexingRule',
+      priority: 0,
+    }
+
+    await models.Action.upsert(action)
+
+    const filterOptions = {
+      status: ActionStatus.FAILED,
+      type: ActionType.ALLOCATE,
+    }
+
+    const whereOptions = actionFilterToWhereOptions(filterOptions)
+    expect(whereOptions).toEqual({
+      [Op.and]: [{ status: 'failed' }, { type: 'allocate' }],
+    })
+
+    await expect(ActionManager.fetchActions(models, filterOptions)).resolves.toHaveLength(
+      1,
+    )
+
+    await expect(ActionManager.fetchActions(models, filterOptions)).resolves.toHaveLength(
+      1,
+    )
+
+    await expect(
+      ActionManager.fetchActions(models, {
+        status: ActionStatus.FAILED,
+        type: ActionType.ALLOCATE,
+        updatedAt: { [Op.gte]: literal("NOW() - INTERVAL '1d'") },
+      }),
+    ).resolves.toHaveLength(1)
+
+    await expect(
+      ActionManager.fetchActions(models, {
+        status: ActionStatus.FAILED,
+        type: ActionType.ALLOCATE,
+        updatedAt: { [Op.lte]: literal("NOW() - INTERVAL '1d'") },
+      }),
+    ).resolves.toHaveLength(0)
+  })
+})

packages/indexer-common/src/indexer-management/__tests__/resolvers/actions.ts

@@ -828,4 +828,104 @@ describe('Actions', () => {
         .toPromise(),
     ).resolves.toHaveProperty('data.actions', [])
   })
+
+  test('Reject queueing for action that has recently failed', async () => {
+    const failedAction = {
+      status: ActionStatus.FAILED,
+      type: ActionType.ALLOCATE,
+      deploymentID: subgraphDeployment1,
+      amount: '10000',
+      force: false,
+      source: 'indexerAgent',
+      reason: 'indexingRule',
+      priority: 0,
+    } as ActionInput
+
+    const proposedAction = {
+      status: ActionStatus.QUEUED,
+      type: ActionType.ALLOCATE,
+      deploymentID: subgraphDeployment1,
+      amount: '10000',
+      source: 'indexerAgent',
+      reason: 'indexingRule',
+      priority: 0,
+    } as ActionInput
+
+    await managementModels.Action.create(failedAction, {
+      validate: true,
+      returning: true,
+    })
+
+    await expect(
+      client.mutation(QUEUE_ACTIONS_MUTATION, { actions: [proposedAction] }).toPromise(),
+    ).resolves.toHaveProperty(
+      'error',
+      new CombinedError({
+        graphQLErrors: [
+          new GraphQLError(
+            "Recently executed 'allocate' action found in queue targeting 'QmQ44hgrWWt3Qf2X9XEX2fPyTbmQbChxwNm5c1t4mhKpGt', ignoring.",
+          ),
+        ],
+      }),
+    )
+    await expect(
+      client
+        .query(ACTIONS_QUERY, {
+          filter: { source: 'indexerAgent' },
+        })
+        .toPromise(),
+    ).resolves.toHaveProperty('data.actions', [
+      await actionInputToExpected(failedAction, 1),
+    ])
+  })
+
+  test('Reject queueing for action that has recently succeeded', async () => {
+    const successfulAction = {
+      status: ActionStatus.SUCCESS,
+      type: ActionType.ALLOCATE,
+      deploymentID: subgraphDeployment1,
+      amount: '10000',
+      force: false,
+      source: 'indexerAgent',
+      reason: 'indexingRule',
+      priority: 0,
+    } as ActionInput
+
+    const proposedAction = {
+      status: ActionStatus.QUEUED,
+      type: ActionType.ALLOCATE,
+      deploymentID: subgraphDeployment1,
+      amount: '10000',
+      source: 'indexerAgent',
+      reason: 'indexingRule',
+      priority: 0,
+    } as ActionInput
+
+    await managementModels.Action.create(successfulAction, {
+      validate: true,
+      returning: true,
+    })
+
+    await expect(
+      client.mutation(QUEUE_ACTIONS_MUTATION, { actions: [proposedAction] }).toPromise(),
+    ).resolves.toHaveProperty(
+      'error',
+      new CombinedError({
+        graphQLErrors: [
+          new GraphQLError(
+            "Recently executed 'allocate' action found in queue targeting 'QmQ44hgrWWt3Qf2X9XEX2fPyTbmQbChxwNm5c1t4mhKpGt', ignoring.",
+          ),
+        ],
+      }),
+    )
+    await expect(
+      client
+        .query(ACTIONS_QUERY, {
+          filter: { source: 'indexerAgent' },
+        })
+        .toPromise(),
+    ).resolves.toHaveProperty('data.actions', [
+      await actionInputToExpected(successfulAction, 1),
+    ])
+  })
 })