chore: remove reallocate step and add continuous RAV collection for long lived allocation (#1181)

Author: Maikol

packages/indexer-agent/src/__tests__/agent.ts

@@ -1,9 +1,14 @@
 import {
+  Agent,
   convertSubgraphBasedRulesToDeploymentBased,
   consolidateAllocationDecisions,
   resolveTargetDeployments,
 } from '../agent'
 import {
+  ActivationCriteria,
+  Allocation,
+  AllocationDecision,
+  AllocationStatus,
   INDEXING_RULE_GLOBAL,
   IndexingDecisionBasis,
   IndexingRuleAttributes,
@@ -328,3 +333,142 @@ describe('resolveTargetDeployments function', () => {
     )
   })
 })
+
+describe('reconcileDeploymentAllocationAction', () => {
+  const deployment = new SubgraphDeploymentID(
+    'QmXZiV6S13ha6QXq4dmaM3TB4CHcDxBMvGexSNu9Kc28EH',
+  )
+
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  const mockLogger: any = {
+    child: jest.fn().mockReturnThis(),
+    info: jest.fn(),
+    warn: jest.fn(),
+    error: jest.fn(),
+    debug: jest.fn(),
+    trace: jest.fn(),
+  }
+
+  const activeAllocations: Allocation[] = [
+    {
+      id: '0x0000000000000000000000000000000000000001',
+      status: AllocationStatus.ACTIVE,
+      isLegacy: false,
+      subgraphDeployment: {
+        id: deployment,
+        ipfsHash: deployment.ipfsHash,
+      },
+      indexer: '0x0000000000000000000000000000000000000000',
+      allocatedTokens: BigInt(1000),
+      createdAt: 0,
+      createdAtEpoch: 1,
+      createdAtBlockHash: '0x0',
+      closedAt: 0,
+      closedAtEpoch: 0,
+      closedAtEpochStartBlockHash: undefined,
+      previousEpochStartBlockHash: undefined,
+      closedAtBlockHash: '0x0',
+      poi: undefined,
+      queryFeeRebates: undefined,
+      queryFeesCollected: undefined,
+    } as unknown as Allocation,
+  ]
+
+  const decision = new AllocationDecision(
+    deployment,
+    {
+      identifier: deployment.ipfsHash,
+      identifierType: SubgraphIdentifierType.DEPLOYMENT,
+      allocationAmount: '1000',
+      decisionBasis: IndexingDecisionBasis.RULES,
+    } as IndexingRuleAttributes,
+    true,
+    ActivationCriteria.SIGNAL_THRESHOLD,
+    'eip155:42161',
+  )
+
+  function createAgent() {
+    const agent = Object.create(Agent.prototype)
+    agent.logger = mockLogger
+    agent.graphNode = {
+      indexingStatus: jest.fn().mockResolvedValue([
+        {
+          subgraphDeployment: { ipfsHash: deployment.ipfsHash },
+          health: 'healthy',
+        },
+      ]),
+    }
+    agent.identifyExpiringAllocations = jest
+      .fn()
+      .mockResolvedValue([activeAllocations[0]])
+    return agent
+  }
+
+  function createOperator() {
+    return {
+      closeEligibleAllocations: jest.fn(),
+      createAllocation: jest.fn(),
+      refreshExpiredAllocations: jest.fn(),
+      presentPOIForAllocations: jest.fn(),
+    }
+  }
+
+  function createNetwork(isHorizon: boolean) {
+    return {
+      isHorizon: { value: jest.fn().mockResolvedValue(isHorizon) },
+      specification: { networkIdentifier: 'eip155:42161' },
+      networkMonitor: {
+        closedAllocations: jest.fn().mockResolvedValue([]),
+      },
+    }
+  }
+
+  it('should call presentPOIForAllocations instead of refreshExpiredAllocations for Horizon allocations', async () => {
+    const agent = createAgent()
+    const operator = createOperator()
+    const network = createNetwork(true)
+
+    await agent.reconcileDeploymentAllocationAction(
+      decision,
+      activeAllocations,
+      10,
+      { value: jest.fn().mockResolvedValue(28) },
+      network,
+      operator,
+      false,
+    )
+
+    expect(agent.identifyExpiringAllocations).toHaveBeenCalled()
+    expect(operator.refreshExpiredAllocations).not.toHaveBeenCalled()
+    expect(operator.presentPOIForAllocations).toHaveBeenCalledWith(
+      expect.anything(),
+      [activeAllocations[0]],
+      network,
+    )
+  })
+
+  it('should call refreshExpiredAllocations for legacy allocations', async () => {
+    const agent = createAgent()
+    const operator = createOperator()
+    const network = createNetwork(false)
+
+    await agent.reconcileDeploymentAllocationAction(
+      decision,
+      activeAllocations,
+      10,
+      { value: jest.fn().mockResolvedValue(28) },
+      network,
+      operator,
+      false,
+    )
+
+    expect(agent.identifyExpiringAllocations).toHaveBeenCalled()
+    expect(operator.refreshExpiredAllocations).toHaveBeenCalledWith(
+      expect.anything(),
+      decision,
+      [activeAllocations[0]],
+      false,
+    )
+    expect(operator.presentPOIForAllocations).not.toHaveBeenCalled()
+  })
+})

packages/indexer-agent/src/agent.ts

@@ -1130,7 +1130,6 @@ export class Agent {
               forceAction,
             )
           } else {
-            // Refresh any expiring allocations
             const expiringAllocations = await this.identifyExpiringAllocations(
               logger,
               activeDeploymentAllocations,
@@ -1140,12 +1139,22 @@ export class Agent {
               network,
             )
             if (expiringAllocations.length > 0) {
-              await operator.refreshExpiredAllocations(
-                logger,
-                deploymentAllocationDecision,
-                expiringAllocations,
-                forceAction,
-              )
+              if (isHorizon) {
+                // Horizon allocations don't need the close/reopen cycle.
+                // Indexing rewards are collected via presentPOI instead.
+                await operator.presentPOIForAllocations(
+                  logger,
+                  expiringAllocations,
+                  network,
+                )
+              } else {
+                await operator.refreshExpiredAllocations(
+                  logger,
+                  deploymentAllocationDecision,
+                  expiringAllocations,
+                  forceAction,
+                )
+              }
             }
           }
         }

packages/indexer-agent/src/commands/start.ts

@@ -288,6 +288,19 @@ export const start = {
         default: 50,
         group: 'Query Fees',
       })
+      .option('rav-collection-interval', {
+        description:
+          'Minimum time in seconds between periodic RAV collections per active allocation',
+        type: 'number',
+        default: 14400,
+        group: 'Query Fees',
+      })
+      .option('rav-check-interval', {
+        description: 'How often the RAV processing loop runs, in seconds',
+        type: 'number',
+        default: 900,
+        group: 'Query Fees',
+      })
       .option('horizon-address-book', {
         description: 'Graph Horizon contracts address book file path',
         type: 'string',
@@ -455,6 +468,8 @@ export async function createNetworkSpecification(
     enableDips: argv.enableDips,
     dipperEndpoint: argv.dipperEndpoint,
     dipsAllocationAmount: argv.dipsAllocationAmount,
+    ravCollectionInterval: argv.ravCollectionInterval,
+    ravCheckInterval: argv.ravCheckInterval,
     dipsEpochsMargin: argv.dipsEpochsMargin,
   }
 

packages/indexer-common/src/__tests__/dips-agreement-protection.test.ts

@@ -0,0 +1,107 @@
+/* eslint-disable @typescript-eslint/no-explicit-any */
+import { ActionInput, ActionStatus, ActionType, validateActionInputs } from '../actions'
+import { AllocationStatus } from '../allocations'
+
+const mockAllocation = {
+  status: AllocationStatus.ACTIVE,
+  subgraphDeployment: { id: { ipfsHash: 'QmTest' } },
+}
+
+const createMockNetworkMonitor = (hasAgreement: boolean) => ({
+  hasActiveDipsAgreement: jest.fn().mockResolvedValue(hasAgreement),
+  allocation: jest.fn().mockResolvedValue(mockAllocation),
+  subgraphDeployment: jest.fn().mockResolvedValue({}),
+})
+
+const createMockLogger = () => ({
+  warn: jest.fn(),
+  info: jest.fn(),
+  debug: jest.fn(),
+  error: jest.fn(),
+  child: jest.fn().mockReturnThis(),
+  trace: jest.fn(),
+})
+
+const baseAction: ActionInput = {
+  type: ActionType.UNALLOCATE,
+  deploymentID: 'QmTest',
+  allocationID: '0x1234567890123456789012345678901234567890',
+  source: 'test',
+  reason: 'test',
+  status: ActionStatus.QUEUED,
+  priority: 0,
+  protocolNetwork: 'eip155:421614',
+  force: false,
+  isLegacy: false,
+}
+
+describe('validateActionInputs DIPS agreement protection', () => {
+  it('should reject UNALLOCATE with active DIPS agreement when force is not set', async () => {
+    const monitor = createMockNetworkMonitor(true)
+    const logger = createMockLogger()
+
+    await expect(
+      validateActionInputs([baseAction], monitor as any, logger as any),
+    ).rejects.toThrow(/active DIPS agreement/)
+  })
+
+  it('should allow UNALLOCATE with active DIPS agreement when force is true', async () => {
+    const monitor = createMockNetworkMonitor(true)
+    const logger = createMockLogger()
+
+    const action = { ...baseAction, force: true }
+
+    await expect(
+      validateActionInputs([action], monitor as any, logger as any),
+    ).resolves.toBeUndefined()
+
+    expect(logger.warn).toHaveBeenCalledWith(
+      'Force-closing allocation with active DIPS agreement',
+      expect.objectContaining({ allocationId: action.allocationID }),
+    )
+  })
+
+  it('should allow UNALLOCATE with no active DIPS agreement', async () => {
+    const monitor = createMockNetworkMonitor(false)
+    const logger = createMockLogger()
+
+    await expect(
+      validateActionInputs([baseAction], monitor as any, logger as any),
+    ).resolves.toBeUndefined()
+  })
+
+  it('should not check agreement for ALLOCATE actions', async () => {
+    const monitor = createMockNetworkMonitor(true)
+    const logger = createMockLogger()
+
+    const action: ActionInput = {
+      ...baseAction,
+      type: ActionType.ALLOCATE,
+      amount: '10000',
+      allocationID: undefined,
+    }
+
+    await expect(
+      validateActionInputs([action], monitor as any, logger as any),
+    ).resolves.toBeUndefined()
+
+    expect(monitor.hasActiveDipsAgreement).not.toHaveBeenCalled()
+  })
+
+  it('should not check DIPS agreement for REALLOCATE actions', async () => {
+    const monitor = createMockNetworkMonitor(true)
+    const logger = createMockLogger()
+
+    const action: ActionInput = {
+      ...baseAction,
+      type: ActionType.REALLOCATE,
+      amount: '10000',
+    }
+
+    // REALLOCATE still validates but doesn't check DIPS agreements
+    // (REALLOCATE itself is deprecated and will be removed)
+    await expect(
+      validateActionInputs([action], monitor as any, logger as any),
+    ).resolves.not.toThrow()
+  })
+})