diff --git a/packages/indexer-common/src/actions.ts b/packages/indexer-common/src/actions.ts index 98d1e9e06..37f18476c 100644 --- a/packages/indexer-common/src/actions.ts +++ b/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', diff --git a/packages/indexer-common/src/indexer-management/__tests__/rule-preservation.test.ts b/packages/indexer-common/src/indexer-management/__tests__/rule-preservation.test.ts index ab122439e..5973a3b8b 100644 --- a/packages/indexer-common/src/indexer-management/__tests__/rule-preservation.test.ts +++ b/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 `:`; 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) => + ({ + 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) + }) }) diff --git a/packages/indexer-common/src/indexer-management/actions.ts b/packages/indexer-common/src/indexer-management/actions.ts index 9d40787c6..318b690e6 100644 --- a/packages/indexer-common/src/indexer-management/actions.ts +++ b/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, diff --git a/packages/indexer-common/src/indexer-management/allocations.ts b/packages/indexer-common/src/indexer-management/allocations.ts index eb7e10ff4..7a7a2e56d 100644 --- a/packages/indexer-common/src/indexer-management/allocations.ts +++ b/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 `:`; +// 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 { 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 { diff --git a/packages/indexer-common/src/indexer-management/resolvers/actions.ts b/packages/indexer-common/src/indexer-management/resolvers/actions.ts index 5ce0d26c0..01577b82a 100644 --- a/packages/indexer-common/src/indexer-management/resolvers/actions.ts +++ b/packages/indexer-common/src/indexer-management/resolvers/actions.ts @@ -14,6 +14,7 @@ import { Network, NetworkMapped, OrderDirection, + RECONCILE_ACTION_SOURCE, validateActionInputs, validateNetworkIdentifier, } from '@graphprotocol/indexer-common' @@ -37,7 +38,7 @@ async function executeQueueOperation( // Check for previously failed conflicting actions const conflictingActions = recentlyAttemptedActions.filter(function (recentAction) { const areEqual = compareActions(recentAction, action) - const fromAgent = action.source === 'indexerAgent' + const fromAgent = action.source === RECONCILE_ACTION_SOURCE return areEqual && fromAgent }) if (conflictingActions.length > 0) { diff --git a/packages/indexer-common/src/operator.ts b/packages/indexer-common/src/operator.ts index 0e90614e1..98f46d836 100644 --- a/packages/indexer-common/src/operator.ts +++ b/packages/indexer-common/src/operator.ts @@ -4,6 +4,7 @@ import { ActionResult, ActionStatus, ActionType, + RECONCILE_ACTION_SOURCE, Allocation, AllocationDecision, AllocationManagementMode, @@ -293,7 +294,7 @@ export class Operator { ...action.params, status, type: action.type, - source: 'indexerAgent', + source: RECONCILE_ACTION_SOURCE, reason: action.reason, priority: 0, protocolNetwork: action.protocolNetwork,