fix: cancel stale allocation closes, let deliberate ones stick

packages/indexer-common/src/actions.ts

@@ -256,6 +256,10 @@ export enum ActionType {
   RESIZE = 'resize',
 }
 
+// Source stamped on every action the agent's own reconcile loop queues; other
+// producers (CLI, management API) choose their own source string.
+export const RECONCILE_ACTION_SOURCE = 'indexerAgent'
+
 export enum ActionStatus {
   QUEUED = 'queued',
   APPROVED = 'approved',

packages/indexer-common/src/indexer-management/__tests__/rule-preservation.test.ts

@@ -1,7 +1,21 @@
 import { SubgraphDeploymentID } from '@graphprotocol/common-ts'
-import { findRuleRequestingAllocation } from '../allocations'
-import { IndexingDecisionBasis, IndexingRuleAttributes } from '../models'
-import { SubgraphIdentifierType } from '../../subgraphs'
+import {
+  findRuleRequestingAllocation,
+  isOptOutReason,
+  staleQueuedCloses,
+} from '../allocations'
+import {
+  IndexingDecisionBasis,
+  IndexingRuleAttributes,
+  INDEXING_RULE_GLOBAL,
+} from '../models'
+import {
+  ActivationCriteria,
+  AllocationDecision,
+  SubgraphIdentifierType,
+} from '../../subgraphs'
+import { ActionStatus, ActionType, RECONCILE_ACTION_SOURCE } from '../../actions'
+import { Action } from '../models/action'
 
 // Pure-function coverage for the guard that stops confirmUnallocate stamping a
 // `never` rule over an operator rule that currently requests allocation.
@@ -72,4 +86,141 @@ describe('findRuleRequestingAllocation', () => {
     ).toBeUndefined()
     expect(findRuleRequestingAllocation([], deployment)).toBeUndefined()
   })
+
+  const globalRule = (basis: IndexingDecisionBasis) =>
+    ({
+      identifier: INDEXING_RULE_GLOBAL,
+      identifierType: SubgraphIdentifierType.GROUP,
+      decisionBasis: basis,
+    }) as IndexingRuleAttributes
+
+  it('falls back to a global always rule when no specific rule exists', () => {
+    const found = findRuleRequestingAllocation(
+      [globalRule(IndexingDecisionBasis.ALWAYS)],
+      deployment,
+    )
+    expect(found?.identifier).toBe(INDEXING_RULE_GLOBAL)
+  })
+
+  it('lets a specific opt-out rule override a global always rule', () => {
+    expect(
+      findRuleRequestingAllocation(
+        [
+          rule({ decisionBasis: IndexingDecisionBasis.NEVER }),
+          globalRule(IndexingDecisionBasis.ALWAYS),
+        ],
+        deployment,
+      ),
+    ).toBeUndefined()
+  })
+})
+
+// The reason recorded at queue time is `<identifierType>:<criteria>`; only the
+// never/offchain criteria mark a close that an opt-out rule decided.
+describe('isOptOutReason', () => {
+  it('recognises opt-out criteria regardless of the rule flavour prefix', () => {
+    for (const reason of ['deployment:never', 'deployment:offchain', 'group:offchain']) {
+      expect(isOptOutReason(reason)).toBe(true)
+    }
+  })
+
+  it('rejects allocate-worthy, gate, and manual reasons', () => {
+    for (const reason of [
+      'deployment:always',
+      'deployment:dips',
+      'deployment:min_stake',
+      'deployment:unsupported',
+      'deployment:none',
+      'none:na',
+      'manual',
+    ]) {
+      expect(isOptOutReason(reason)).toBe(false)
+    }
+  })
+
+  // Lock isOptOutReason to the exact format reconcile records: the reason on a
+  // queued close is AllocationDecision.reasonString(), not a hand-written string.
+  it('stays in lockstep with the reason format reconcile records', () => {
+    const deployment = new SubgraphDeploymentID(
+      'QmXZiV6S13ha6QXq4dmaM3TB4CHcDxBMvGexSNu9Kc28EH',
+    )
+    const decision = (basis: IndexingDecisionBasis, criteria: ActivationCriteria) =>
+      new AllocationDecision(
+        deployment,
+        {
+          identifier: deployment.ipfsHash,
+          identifierType: SubgraphIdentifierType.DEPLOYMENT,
+          decisionBasis: basis,
+        } as IndexingRuleAttributes,
+        false,
+        criteria,
+        'eip155:42161',
+      )
+    expect(
+      isOptOutReason(
+        decision(
+          IndexingDecisionBasis.OFFCHAIN,
+          ActivationCriteria.OFFCHAIN,
+        ).reasonString(),
+      ),
+    ).toBe(true)
+    expect(
+      isOptOutReason(
+        decision(IndexingDecisionBasis.NEVER, ActivationCriteria.NEVER).reasonString(),
+      ),
+    ).toBe(true)
+    expect(
+      isOptOutReason(
+        decision(IndexingDecisionBasis.ALWAYS, ActivationCriteria.ALWAYS).reasonString(),
+      ),
+    ).toBe(false)
+  })
+})
+
+describe('staleQueuedCloses', () => {
+  const deployment = new SubgraphDeploymentID(
+    'QmXZiV6S13ha6QXq4dmaM3TB4CHcDxBMvGexSNu9Kc28EH',
+  )
+  const alwaysRule = {
+    identifier: deployment.ipfsHash,
+    identifierType: SubgraphIdentifierType.DEPLOYMENT,
+    decisionBasis: IndexingDecisionBasis.ALWAYS,
+  } as IndexingRuleAttributes
+
+  const close = (overrides: Partial<Action>) =>
+    ({
+      id: 1,
+      type: ActionType.UNALLOCATE,
+      status: ActionStatus.APPROVED,
+      source: RECONCILE_ACTION_SOURCE,
+      reason: 'deployment:offchain',
+      deploymentID: deployment.ipfsHash,
+      ...overrides,
+    }) as Action
+
+  it('matches a reconcile-queued opt-out close whose rule now says allocate', () => {
+    expect(staleQueuedCloses([close({})], [alwaysRule])).toHaveLength(1)
+  })
+
+  it('never matches manual or API closes, other action types, or current closes', () => {
+    expect(
+      staleQueuedCloses([close({ source: 'indexerCLI' })], [alwaysRule]),
+    ).toHaveLength(0)
+    expect(staleQueuedCloses([close({ reason: 'manual' })], [alwaysRule])).toHaveLength(0)
+    expect(
+      staleQueuedCloses([close({ type: ActionType.ALLOCATE })], [alwaysRule]),
+    ).toHaveLength(0)
+    expect(
+      staleQueuedCloses([close({ reason: 'deployment:always' })], [alwaysRule]),
+    ).toHaveLength(0)
+    expect(staleQueuedCloses([close({})], [])).toHaveLength(0)
+  })
+
+  // A deploying close is crash-recovery residue whose transaction may already
+  // be on-chain; cancelling its record would contradict what actually happened.
+  it('never matches a close that is already deploying', () => {
+    expect(
+      staleQueuedCloses([close({ status: ActionStatus.DEPLOYING })], [alwaysRule]),
+    ).toHaveLength(0)
+  })
 })

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

@@ -19,6 +19,7 @@ import {
   OrderDirection,
   GraphNode,
   sequentialTimerMap,
+  staleQueuedCloses,
 } from '@graphprotocol/indexer-common'
 
 import { Order, Transaction } from 'sequelize'
@@ -356,6 +357,46 @@ export class ActionManager {
           return []
         }
 
+        // A close queued under an opt-out rule can outlive that rule. Cancel it
+        // here, inside the same lock that guards execution, rather than close an
+        // allocation the operator has since asked to keep.
+        try {
+          const currentRules = await this.models.IndexingRule.findAll({
+            where: { protocolNetwork },
+            transaction,
+          })
+          const staleCloses = staleQueuedCloses(approvedAndDeployingActions, currentRules)
+          if (staleCloses.length > 0) {
+            logger.info(
+              `Canceling queued closes whose opt-out rule has since flipped back to allocate`,
+              {
+                canceled: staleCloses.map((action) => ({
+                  id: action.id,
+                  deploymentID: action.deploymentID,
+                  reason: action.reason,
+                })),
+              },
+            )
+            const staleIds = new Set(staleCloses.map((action) => action.id))
+            await this.models.Action.update(
+              { status: ActionStatus.CANCELED },
+              { where: { id: Array.from(staleIds) }, transaction },
+            )
+            approvedAndDeployingActions = approvedAndDeployingActions.filter(
+              (action) => !staleIds.has(action.id),
+            )
+            if (approvedAndDeployingActions.length === 0) {
+              logger.debug('All approved actions were canceled as stale closes')
+              return []
+            }
+          }
+        } catch (error) {
+          // Fail open: a veto failure must not block the batch. A stale close
+          // may then execute, but the post-close rule guard still prevents the
+          // never-stamp, so the allocation is recreated rather than blocklisted.
+          logger.warn('Failed to check queued closes against current rules', { error })
+        }
+
         // Limit batch size to prevent multicall failures when there are many allocations
         const maxBatchSize =
           network.specification.indexerOptions.autoAllocationMaxBatchSize
@@ -381,6 +422,12 @@ export class ActionManager {
       },
     )
 
+    // An empty batch (nothing approved, or everything canceled as stale) has
+    // nothing to execute; executeBatch treats it as an error, so return early.
+    if (prioritizedActions.length === 0) {
+      return []
+    }
+
     try {
       logger.debug('Executing batch action', {
         prioritizedActions,

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

@@ -7,7 +7,9 @@ import {
 import {
   Action,
   ActionFailure,
+  ActionStatus,
   ActionType,
+  ActivationCriteria,
   Allocation,
   AllocationResult,
   AllocationStatus,
@@ -20,6 +22,7 @@ import {
   IndexerError,
   IndexerErrorCode,
   IndexerManagementModels,
+  INDEXING_RULE_GLOBAL,
   IndexingDecisionBasis,
   IndexingRuleAttributes,
   IndexingStatus,
@@ -28,6 +31,7 @@ import {
   preprocessRules,
   Network,
   PresentPOIResult,
+  RECONCILE_ACTION_SOURCE,
   ResizeAllocationResult,
   SubgraphIdentifierType,
   SubgraphStatus,
@@ -140,19 +144,50 @@ export function encodeCollectData(allocationId: string, poiData: POIData): strin
   return encodeCollectIndexingRewardsData(allocationId, poiData.poi, encodedPOIMetadata)
 }
 
-// A queued close can outlive the rule that justified it. Find a deployment
-// rule that currently asks for allocation (operator `always` or DIPs-managed);
-// stamping `never` over it would also blocklist the deployment for DIPs.
+// A queued close can outlive the rule that justified it. Find the effective
+// rule (deployment-specific first, else global) when it currently asks for
+// allocation; a `never` stamp over it would also blocklist DIPs proposals.
 export function findRuleRequestingAllocation(
   rules: IndexingRuleAttributes[],
   deployment: SubgraphDeploymentID,
 ): IndexingRuleAttributes | undefined {
-  return rules.find(
-    (rule) =>
-      rule.identifierType === SubgraphIdentifierType.DEPLOYMENT &&
-      new SubgraphDeploymentID(rule.identifier).bytes32 === deployment.bytes32 &&
-      (rule.decisionBasis === IndexingDecisionBasis.ALWAYS ||
-        rule.decisionBasis === IndexingDecisionBasis.DIPS),
+  const effective =
+    rules.find(
+      (rule) =>
+        rule.identifierType === SubgraphIdentifierType.DEPLOYMENT &&
+        new SubgraphDeploymentID(rule.identifier).bytes32 === deployment.bytes32,
+    ) ?? rules.find((rule) => rule.identifier === INDEXING_RULE_GLOBAL)
+  return effective?.decisionBasis === IndexingDecisionBasis.ALWAYS ||
+    effective?.decisionBasis === IndexingDecisionBasis.DIPS
+    ? effective
+    : undefined
+}
+
+// Reconcile records each queued action's reason as `<identifierType>:<criteria>`;
+// only the never/offchain criteria mark a close that an opt-out rule decided.
+export function isOptOutReason(reason: string): boolean {
+  const criteria = reason.split(':').pop()
+  return criteria === ActivationCriteria.NEVER || criteria === ActivationCriteria.OFFCHAIN
+}
+
+// Closes that reconcile queued under an opt-out rule which has since flipped
+// back to allocate: safe to cancel while still approved. A deploying close may
+// already be on-chain; manual and API closes carry other sources.
+export function staleQueuedCloses(
+  actions: Action[],
+  rules: IndexingRuleAttributes[],
+): Action[] {
+  return actions.filter(
+    (action) =>
+      action.type === ActionType.UNALLOCATE &&
+      action.status === ActionStatus.APPROVED &&
+      action.source === RECONCILE_ACTION_SOURCE &&
+      isOptOutReason(action.reason) &&
+      !!action.deploymentID &&
+      findRuleRequestingAllocation(
+        rules,
+        new SubgraphDeploymentID(action.deploymentID),
+      ) !== undefined,
   )
 }
 
@@ -462,6 +497,8 @@ export class AllocationManager {
           // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
           action.allocationID!,
           receipt,
+          action.source,
+          action.reason,
         )
       case ActionType.PRESENT_POI:
         return await this.confirmPresentPOI(
@@ -867,6 +904,8 @@ export class AllocationManager {
     actionID: number,
     allocationID: string,
     receipt: TransactionReceipt | 'paused' | 'unauthorized',
+    source: string,
+    reason: string,
   ): Promise<CloseAllocationResult> {
     const logger = this.logger.child({ action: actionID })
     logger.info(`Confirming unallocate transaction`)
@@ -927,22 +966,26 @@ export class AllocationManager {
     })
     const allocation = await this.network.networkMonitor.allocation(allocationID)
 
-    // The rule that justified this close may have changed while the action
-    // sat in the queue: an operator flipping the deployment back to `always`
-    // (or DIPs taking it over) must not be overwritten with `never`.
-    const currentRules = await this.models.IndexingRule.findAll({
-      where: { protocolNetwork: this.network.specification.networkIdentifier },
-    })
-    const allocateRule = findRuleRequestingAllocation(
-      currentRules,
-      allocation.subgraphDeployment.id,
-    )
+    // Agent-queued closes re-check the live rule: a stale opt-out close and a
+    // health-check close under a live allocate rule must not stamp `never`.
+    // Manual and API closes stamp unconditionally, so a hand-queued close sticks.
+    let allocateRule: IndexingRuleAttributes | undefined
+    if (source === RECONCILE_ACTION_SOURCE) {
+      const currentRules = await this.models.IndexingRule.findAll({
+        where: { protocolNetwork: this.network.specification.networkIdentifier },
+      })
+      allocateRule = findRuleRequestingAllocation(
+        currentRules,
+        allocation.subgraphDeployment.id,
+      )
+    }
     if (allocateRule) {
       logger.info(
         `Keeping the current indexing rule: it requests allocation, so skipping the never-rule stamp after this close`,
         {
           deployment: allocation.subgraphDeployment.id.ipfsHash,
           currentDecisionBasis: allocateRule.decisionBasis,
+          queuedReason: reason,
         },
       )
     } else {