chore: force executing DIPs actions in manual and oversight mode

Author: pcarranzav

docs/action-queue.md

@@ -8,8 +8,8 @@ The action execution worker will only grab items from the action queue to execut
 
 ## Allocation management modes:
 - `auto`: The indexer-agent will act similarly to the legacy paradigm. When it identifies allocation actions it will add them to the queue with ActionStatus = `approved`; the execution worker process will pick up the approved actions within 30 seconds and execute them.
-- `manual`: The indexer-agent will not add any items to the action queue in this mode. It will spin up an indexer-management server which can be interacted with manually or integrated with 3rd party tools to add actions to the action queue and execute them.
-- `oversight`: The indexer-agent will add run its reconciliation loop to make allocation decisions and when actions are identified it will queue them. These actions will then require approval before they can be executed.
+- `manual`: The indexer-agent will not add any items to the action queue in this mode. It will spin up an indexer-management server which can be interacted with manually or integrated with 3rd party tools to add actions to the action queue and execute them. An exception to this is indexing agreements (DIPs), for which actions will be queued and executed even in this mode.
+- `oversight`: The indexer-agent will add run its reconciliation loop to make allocation decisions and when actions are identified it will queue them. These actions will then require approval before they can be executed. An exception to this is indexing agreements (DIPs), for which actions will be queued as approved and executed even in this mode.
 
 ## Actions CLI
 The indexer-cli provides an `actions` module for manually working with the action queue. It uses the #Graphql API hosted by the indexer management server to interact with the actions queue.

packages/indexer-agent/src/agent.ts

@@ -852,6 +852,7 @@ export class Agent {
     maxAllocationEpochs: number,
     network: Network,
     operator: Operator,
+    forceAction: boolean = false,
   ): Promise<void> {
     const logger = this.logger.child({
       deployment: deploymentAllocationDecision.deployment.ipfsHash,
@@ -873,6 +874,7 @@ export class Agent {
           logger,
           deploymentAllocationDecision,
           activeDeploymentAllocations,
+          forceAction,
         )
       case true: {
         // If no active allocations and subgraph health passes safety check, create one
@@ -909,6 +911,7 @@ export class Agent {
               logger,
               deploymentAllocationDecision,
               mostRecentlyClosedAllocation,
+              forceAction,
             )
           }
         } else if (activeDeploymentAllocations.length > 0) {
@@ -917,6 +920,7 @@ export class Agent {
               logger,
               deploymentAllocationDecision,
               activeDeploymentAllocations,
+              forceAction,
             )
           } else {
             // Refresh any expiring allocations
@@ -933,6 +937,7 @@ export class Agent {
                 logger,
                 deploymentAllocationDecision,
                 expiringAllocations,
+                forceAction,
               )
             }
           }
@@ -952,21 +957,37 @@ export class Agent {
     // --------------------------------------------------------------------------------
     const { network, operator } = this.networkAndOperator
     let validatedAllocationDecisions = [...allocationDecisions]
+    let dipsDeployments: SubgraphDeploymentID[] = []
+    if (network.specification.indexerOptions.enableDips) {
+      if (!operator.dipsManager) {
+        throw new Error('DipsManager is not available')
+      }
+      dipsDeployments = await operator.dipsManager.getActiveDipsDeployments()
+    }
 
     if (
       network.specification.indexerOptions.allocationManagementMode ===
       AllocationManagementMode.MANUAL
     ) {
-      this.logger.trace(
-        `Skipping allocation reconciliation since AllocationManagementMode = 'manual'`,
-        {
-          protocolNetwork: network.specification.networkIdentifier,
-          targetDeployments: allocationDecisions
-            .filter(decision => decision.toAllocate)
-            .map(decision => decision.deployment.ipfsHash),
-        },
-      )
-      validatedAllocationDecisions = [] as AllocationDecision[]
+      if (network.specification.indexerOptions.enableDips) {
+        this.logger.warn(
+          `Allocation management is manual, but DIPs is enabled. Reconciling DIPs allocations anyways.`,
+        )
+        validatedAllocationDecisions = validatedAllocationDecisions.filter(
+          decision => dipsDeployments.includes(decision.deployment),
+        )
+      } else {
+        this.logger.trace(
+          `Skipping allocation reconciliation since AllocationManagementMode = 'manual'`,
+          {
+            protocolNetwork: network.specification.networkIdentifier,
+            targetDeployments: allocationDecisions
+              .filter(decision => decision.toAllocate)
+              .map(decision => decision.deployment.ipfsHash),
+          },
+        )
+        validatedAllocationDecisions = [] as AllocationDecision[]
+      }
     } else {
       const networkSubgraphDeployment = network.networkSubgraph.deployment
       if (
@@ -1028,6 +1049,7 @@ export class Agent {
         maxAllocationEpochs,
         network,
         operator,
+        dipsDeployments.includes(decision.deployment), // Force actions if this is a DIPs deployment
       ),
     )
     return

packages/indexer-common/src/operator.ts

@@ -263,16 +263,26 @@ export class Operator {
     return result.data.actions
   }
 
-  async queueAction(action: ActionItem): Promise<Action[]> {
+  async queueAction(action: ActionItem, forceAction: boolean = false): Promise<Action[]> {
     let status = ActionStatus.QUEUED
     switch (this.specification.indexerOptions.allocationManagementMode) {
       case AllocationManagementMode.MANUAL:
-        throw Error(`Cannot queue actions when AllocationManagementMode = 'MANUAL'`)
+        if (forceAction) {
+          status = ActionStatus.APPROVED
+        } else {
+          throw Error(`Cannot queue actions when AllocationManagementMode = 'MANUAL'`)
+        }
+        break
       case AllocationManagementMode.AUTO:
         status = ActionStatus.APPROVED
         break
       case AllocationManagementMode.OVERSIGHT:
-        status = ActionStatus.QUEUED
+        if (forceAction) {
+          status = ActionStatus.APPROVED
+        } else {
+          status = ActionStatus.QUEUED
+        }
+        break
     }
 
     const actionInput = {
@@ -341,6 +351,7 @@ export class Operator {
     logger: Logger,
     deploymentAllocationDecision: AllocationDecision,
     mostRecentlyClosedAllocation: Allocation | undefined,
+    forceAction: boolean = false,
   ): Promise<void> {
     const desiredAllocationAmount = deploymentAllocationDecision.ruleMatch.rule
       ?.allocationAmount
@@ -369,15 +380,18 @@ export class Operator {
     }
 
     // Send AllocateAction to the queue
-    await this.queueAction({
-      params: {
-        deploymentID: deploymentAllocationDecision.deployment.ipfsHash,
-        amount: formatGRT(desiredAllocationAmount),
+    await this.queueAction(
+      {
+        params: {
+          deploymentID: deploymentAllocationDecision.deployment.ipfsHash,
+          amount: formatGRT(desiredAllocationAmount),
+        },
+        type: ActionType.ALLOCATE,
+        reason: deploymentAllocationDecision.reasonString(),
+        protocolNetwork: deploymentAllocationDecision.protocolNetwork,
       },
-      type: ActionType.ALLOCATE,
-      reason: deploymentAllocationDecision.reasonString(),
-      protocolNetwork: deploymentAllocationDecision.protocolNetwork,
-    })
+      forceAction,
+    )
 
     return
   }
@@ -386,6 +400,7 @@ export class Operator {
     logger: Logger,
     deploymentAllocationDecision: AllocationDecision,
     activeDeploymentAllocations: Allocation[],
+    forceAction: boolean = false,
   ): Promise<void> {
     const activeDeploymentAllocationsEligibleForClose = activeDeploymentAllocations.map(
       (allocation) => allocation.id,
@@ -405,17 +420,20 @@ export class Operator {
         activeDeploymentAllocationsEligibleForClose,
         async (allocation) => {
           // Send unallocate action to the queue
-          await this.queueAction({
-            params: {
-              allocationID: allocation,
-              deploymentID: deploymentAllocationDecision.deployment.ipfsHash,
-              poi: undefined,
-              force: false,
-            },
-            type: ActionType.UNALLOCATE,
-            reason: deploymentAllocationDecision.reasonString(),
-            protocolNetwork: deploymentAllocationDecision.protocolNetwork,
-          } as ActionItem)
+          await this.queueAction(
+            {
+              params: {
+                allocationID: allocation,
+                deploymentID: deploymentAllocationDecision.deployment.ipfsHash,
+                poi: undefined,
+                force: false,
+              },
+              type: ActionType.UNALLOCATE,
+              reason: deploymentAllocationDecision.reasonString(),
+              protocolNetwork: deploymentAllocationDecision.protocolNetwork,
+            } as ActionItem,
+            forceAction,
+          )
         },
         { concurrency: 1 },
       )
@@ -426,6 +444,7 @@ export class Operator {
     logger: Logger,
     deploymentAllocationDecision: AllocationDecision,
     expiredAllocations: Allocation[],
+    forceAction: boolean = false,
   ): Promise<void> {
     if (deploymentAllocationDecision.ruleMatch.rule?.autoRenewal) {
       logger.info(`Reallocating expired allocations`, {
@@ -442,16 +461,19 @@ export class Operator {
       await pMap(
         expiredAllocations,
         async (allocation) => {
-          await this.queueAction({
-            params: {
-              allocationID: allocation.id,
-              deploymentID: deploymentAllocationDecision.deployment.ipfsHash,
-              amount: formatGRT(desiredAllocationAmount),
+          await this.queueAction(
+            {
+              params: {
+                allocationID: allocation.id,
+                deploymentID: deploymentAllocationDecision.deployment.ipfsHash,
+                amount: formatGRT(desiredAllocationAmount),
+              },
+              type: ActionType.REALLOCATE,
+              reason: `${deploymentAllocationDecision.reasonString()}:allocationExpiring`, // Need to update to include 'ExpiringSoon'
+              protocolNetwork: deploymentAllocationDecision.protocolNetwork,
             },
-            type: ActionType.REALLOCATE,
-            reason: `${deploymentAllocationDecision.reasonString()}:allocationExpiring`, // Need to update to include 'ExpiringSoon'
-            protocolNetwork: deploymentAllocationDecision.protocolNetwork,
-          })
+            forceAction,
+          )
         },
         {
           stopOnError: false,