From 861cddfeda9e9aee5f1c196b58c1fa993978651e Mon Sep 17 00:00:00 2001 From: Miguel de Elias Date: Tue, 31 Mar 2026 16:21:44 -0300 Subject: [PATCH 1/3] feat: implement on-chain cancel for DIPs agreements --- .../src/indexer-management/allocations.ts | 9 - .../resolvers/allocations.ts | 11 - .../src/indexing-fees/__tests__/dips.test.ts | 235 ++++++++++++++++++ .../indexer-common/src/indexing-fees/dips.ts | 117 +++++++++ 4 files changed, 352 insertions(+), 20 deletions(-) diff --git a/packages/indexer-common/src/indexer-management/allocations.ts b/packages/indexer-common/src/indexer-management/allocations.ts index 79716f515..ae0ff7e13 100644 --- a/packages/indexer-common/src/indexer-management/allocations.ts +++ b/packages/indexer-common/src/indexer-management/allocations.ts @@ -1214,15 +1214,6 @@ export class AllocationManager { await upsertIndexingRule(logger, this.models, neverIndexingRule) - if (this.dipsManager) { - await this.dipsManager.tryCancelAgreement(allocationID) - await this.dipsManager.tryUpdateAgreementAllocation( - allocation.subgraphDeployment.id.toString(), - toAddress(allocationID), - null, - ) - } - return { actionID, type: 'unallocate', diff --git a/packages/indexer-common/src/indexer-management/resolvers/allocations.ts b/packages/indexer-common/src/indexer-management/resolvers/allocations.ts index 87faa6eeb..1186e4626 100644 --- a/packages/indexer-common/src/indexer-management/resolvers/allocations.ts +++ b/packages/indexer-common/src/indexer-management/resolvers/allocations.ts @@ -1762,17 +1762,6 @@ export default { await models.IndexingRule.upsert(offchainIndexingRule) - const allocationManager = - actionManager?.allocationManagers[network.specification.networkIdentifier] - if (allocationManager?.dipsManager) { - await allocationManager.dipsManager.tryCancelAgreement(allocation) - await allocationManager.dipsManager.tryUpdateAgreementAllocation( - allocationData.subgraphDeployment.id.toString(), - toAddress(allocation), - null, - ) - } - // Since upsert succeeded, we _must_ have a rule const updatedRule = await models.IndexingRule.findOne({ where: { identifier: offchainIndexingRule.identifier }, diff --git a/packages/indexer-common/src/indexing-fees/__tests__/dips.test.ts b/packages/indexer-common/src/indexing-fees/__tests__/dips.test.ts index 636c515bf..f557d21e1 100644 --- a/packages/indexer-common/src/indexing-fees/__tests__/dips.test.ts +++ b/packages/indexer-common/src/indexing-fees/__tests__/dips.test.ts @@ -17,6 +17,7 @@ import { IndexerManagementClient, MultiNetworks, } from '@graphprotocol/indexer-common' +import type { SubgraphIndexingAgreement } from '../agreement-monitor' import { connectDatabase, createLogger, @@ -458,6 +459,240 @@ describe('DipsManager', () => { expect(deployments).toHaveLength(1) expect(deployments[0].ipfsHash).toBe(testDeploymentId) }) + + describe('cancelAgreement', () => { + const mockAgreement: SubgraphIndexingAgreement = { + id: '0x123e4567e89b12d3a456426614174000', + allocationId: '0xabcd47df40c29949a75a6693c77834c00b8ad626', + subgraphDeploymentId: 'QmTZ8ejXJxRo7vDBS4uwqBeGoxLSWbhaA7oXa1RvxunLy7', + state: 1, // Accepted + lastCollectionAt: '0', + endsAt: '9999999999', + maxInitialTokens: '1000', + maxOngoingTokensPerSecond: '100', + tokensPerSecond: '10', + tokensPerEntityPerSecond: '1', + minSecondsPerCollection: 60, + maxSecondsPerCollection: 300, + canceledAt: '0', + } + + beforeEach(() => { + // Track the agreement so we can verify cleanup + dipsManager.collectionTracker.track(mockAgreement.id, { + lastCollectedAt: 0, + minSecondsPerCollection: 60, + maxSecondsPerCollection: 300, + }) + }) + + test('successful cancel + final collect attempt', async () => { + const mockReceipt = { hash: '0xcancel123' } + const mockCollectReceipt = { hash: '0xcollect456' } + + // Mock cancel transaction + network.transactionManager.executeTransaction = jest + .fn() + .mockResolvedValueOnce(mockReceipt) // cancel + .mockResolvedValueOnce(mockCollectReceipt) // collect + + // Mock block number and graph node methods for collect + network.networkProvider.getBlockNumber = jest.fn().mockResolvedValue(100) + graphNode.entityCount = jest.fn().mockResolvedValue([250000]) + graphNode.subgraphFeatures = jest + .fn() + .mockResolvedValue({ network: 'mainnet' }) + graphNode.blockHashFromNumber = jest + .fn() + .mockResolvedValue('0xblockhash') + graphNode.proofOfIndexing = jest + .fn() + .mockResolvedValue( + '0x0000000000000000000000000000000000000000000000000000000000000001', + ) + + const result = await dipsManager.cancelAgreement( + mockAgreement.id, + mockAgreement, + ) + + expect(result).toBe(true) + // executeTransaction called twice: once for cancel, once for collect + expect(network.transactionManager.executeTransaction).toHaveBeenCalledTimes(2) + // Tracker should be cleaned up (untracked = ready) + expect( + dipsManager.collectionTracker.isReadyForCollection(mockAgreement.id, 0), + ).toBe(true) + }) + + test('cancel fails returns false, no collect attempted', async () => { + // Mock cancel transaction failure + network.transactionManager.executeTransaction = jest + .fn() + .mockRejectedValueOnce(new Error('cancel tx reverted')) + + const result = await dipsManager.cancelAgreement( + mockAgreement.id, + mockAgreement, + ) + + expect(result).toBe(false) + // executeTransaction called only once (for cancel) + expect(network.transactionManager.executeTransaction).toHaveBeenCalledTimes(1) + }) + + test('cancel succeeds but collect fails returns true, tracker still cleaned up', async () => { + const mockReceipt = { hash: '0xcancel123' } + + // Mock cancel succeeds + network.transactionManager.executeTransaction = jest + .fn() + .mockResolvedValueOnce(mockReceipt) // cancel succeeds + .mockRejectedValueOnce(new Error('collect failed')) // collect fails + + // Mock block number and graph node methods + network.networkProvider.getBlockNumber = jest.fn().mockResolvedValue(100) + graphNode.entityCount = jest.fn().mockResolvedValue([250000]) + graphNode.subgraphFeatures = jest + .fn() + .mockResolvedValue({ network: 'mainnet' }) + graphNode.blockHashFromNumber = jest + .fn() + .mockResolvedValue('0xblockhash') + graphNode.proofOfIndexing = jest + .fn() + .mockResolvedValue( + '0x0000000000000000000000000000000000000000000000000000000000000001', + ) + + const result = await dipsManager.cancelAgreement( + mockAgreement.id, + mockAgreement, + ) + + expect(result).toBe(true) + // Tracker should be cleaned up even though collect failed + expect( + dipsManager.collectionTracker.isReadyForCollection(mockAgreement.id, 0), + ).toBe(true) + }) + }) + + describe('cleanupFinishedAgreement', () => { + const baseAgreement: SubgraphIndexingAgreement = { + id: '0x123e4567e89b12d3a456426614174000', + allocationId: '0xabcd47df40c29949a75a6693c77834c00b8ad626', + subgraphDeploymentId: 'QmTZ8ejXJxRo7vDBS4uwqBeGoxLSWbhaA7oXa1RvxunLy7', + state: 1, + lastCollectionAt: '0', + endsAt: '9999999999', + maxInitialTokens: '1000', + maxOngoingTokensPerSecond: '100', + tokensPerSecond: '10', + tokensPerEntityPerSecond: '1', + minSecondsPerCollection: 60, + maxSecondsPerCollection: 300, + canceledAt: '0', + } + + beforeEach(() => { + dipsManager.collectionTracker.track(baseAgreement.id, { + lastCollectedAt: 0, + minSecondsPerCollection: 60, + maxSecondsPerCollection: 300, + }) + }) + + test('removes payer-cancelled agreement from tracker after collection', () => { + const removeSpy = jest.spyOn(dipsManager.collectionTracker, 'remove') + const agreement = { ...baseAgreement, state: 3 } + + const result = dipsManager.cleanupFinishedAgreement(agreement, 1000, logger) + + expect(result).toBe(true) + expect(removeSpy).toHaveBeenCalledWith(agreement.id) + }) + + test('removes expired agreement from tracker after collection', () => { + const removeSpy = jest.spyOn(dipsManager.collectionTracker, 'remove') + const nowSeconds = 2000 + const agreement = { ...baseAgreement, state: 1, endsAt: '1000' } + + const result = dipsManager.cleanupFinishedAgreement( + agreement, + nowSeconds, + logger, + ) + + expect(result).toBe(true) + expect(removeSpy).toHaveBeenCalledWith(agreement.id) + }) + + test('does not remove active agreement from tracker', () => { + const removeSpy = jest.spyOn(dipsManager.collectionTracker, 'remove') + const nowSeconds = 1000 + const agreement = { ...baseAgreement, state: 1, endsAt: '9999999999' } + + const result = dipsManager.cleanupFinishedAgreement( + agreement, + nowSeconds, + logger, + ) + + expect(result).toBe(false) + expect(removeSpy).not.toHaveBeenCalled() + }) + }) + + describe('cancelBlocklistedAgreements', () => { + const mockAgreement: SubgraphIndexingAgreement = { + id: '0x123e4567e89b12d3a456426614174000', + allocationId: '0xabcd47df40c29949a75a6693c77834c00b8ad626', + subgraphDeploymentId: 'QmTZ8ejXJxRo7vDBS4uwqBeGoxLSWbhaA7oXa1RvxunLy7', + state: 1, + lastCollectionAt: '0', + endsAt: '9999999999', + maxInitialTokens: '1000', + maxOngoingTokensPerSecond: '100', + tokensPerSecond: '10', + tokensPerEntityPerSecond: '1', + minSecondsPerCollection: 60, + maxSecondsPerCollection: 300, + canceledAt: '0', + } + + test('cancels agreements with NEVER rule for their deployment', async () => { + // Create a NEVER rule for the test deployment + await managementModels.IndexingRule.create({ + identifier: testDeploymentId, + identifierType: SubgraphIdentifierType.DEPLOYMENT, + decisionBasis: IndexingDecisionBasis.NEVER, + requireSupported: true, + safety: true, + protocolNetwork: 'eip155:421614', + allocationAmount: '0', + }) + + const cancelSpy = jest + .spyOn(dipsManager, 'cancelAgreement') + .mockResolvedValue(true) + + await dipsManager.cancelBlocklistedAgreements([mockAgreement]) + + expect(cancelSpy).toHaveBeenCalledTimes(1) + expect(cancelSpy).toHaveBeenCalledWith(mockAgreement.id, mockAgreement) + }) + + test('does not cancel agreements without NEVER rule', async () => { + const cancelSpy = jest + .spyOn(dipsManager, 'cancelAgreement') + .mockResolvedValue(true) + + await dipsManager.cancelBlocklistedAgreements([mockAgreement]) + + expect(cancelSpy).not.toHaveBeenCalled() + }) + }) }) }) diff --git a/packages/indexer-common/src/indexing-fees/dips.ts b/packages/indexer-common/src/indexing-fees/dips.ts index f585b1466..ab4304fd5 100644 --- a/packages/indexer-common/src/indexing-fees/dips.ts +++ b/packages/indexer-common/src/indexing-fees/dips.ts @@ -481,6 +481,99 @@ export class DipsManager { } } + async cancelAgreement( + agreementId: string, + agreement: SubgraphIndexingAgreement, + ): Promise { + const logger = this.logger.child({ + function: 'cancelAgreement', + agreementId, + }) + + // Step 1: Cancel on-chain + const indexerAddress = this.network.specification.indexerOptions.address + try { + const receipt = await this.network.transactionManager.executeTransaction( + async () => + this.network.contracts.SubgraphService.cancelIndexingAgreement.estimateGas( + indexerAddress, + agreementId, + ), + async (gasLimit) => + this.network.contracts.SubgraphService.cancelIndexingAgreement( + indexerAddress, + agreementId, + { gasLimit }, + ), + logger.child({ function: 'SubgraphService.cancelIndexingAgreement' }), + ) + + if (receipt === 'paused' || receipt === 'unauthorized') { + logger.warn('Cannot cancel: network paused or unauthorized') + return false + } + + logger.info('Successfully cancelled agreement on-chain', { + txHash: receipt.hash, + }) + } catch (err) { + const errorMsg = err instanceof Error ? err.message : String(err) + logger.error('Failed to cancel agreement on-chain', { error: errorMsg }) + return false + } + + // Step 2: Best-effort final collection + try { + const blockNumber = await this.network.networkProvider.getBlockNumber() + await this.tryCollectAgreement(agreement, blockNumber, logger) + logger.info('Final collection succeeded after cancel') + } catch (err) { + const errorMsg = err instanceof Error ? err.message : String(err) + logger.warn('Final collection after cancel failed, fees may be lost', { + error: errorMsg, + }) + } + + // Step 3: Cleanup + this.collectionTracker.remove(agreementId) + + return true + } + + async cancelBlocklistedAgreements( + agreements: SubgraphIndexingAgreement[], + ): Promise { + const logger = this.logger.child({ + function: 'cancelBlocklistedAgreements', + }) + + const allDeploymentRules = await this.models.IndexingRule.findAll({ + where: { + identifierType: SubgraphIdentifierType.DEPLOYMENT, + }, + }) + + for (const agreement of agreements) { + const subgraphDeploymentID = new SubgraphDeploymentID( + agreement.subgraphDeploymentId, + ) + const blocklistedRule = allDeploymentRules.find( + (rule) => + new SubgraphDeploymentID(rule.identifier).bytes32 === + subgraphDeploymentID.bytes32 && + rule.decisionBasis === IndexingDecisionBasis.NEVER, + ) + + if (blocklistedRule) { + logger.info('Cancelling blocklisted agreement', { + agreementId: agreement.id, + deployment: subgraphDeploymentID.display, + }) + await this.cancelAgreement(agreement.id, agreement) + } + } + } + async collectAgreementPayments(): Promise { const logger = this.logger.child({ function: 'collectAgreementPayments' }) const indexerAddress = this.network.specification.indexerOptions.address @@ -495,6 +588,9 @@ export class DipsManager { return } + // Cancel any agreements whose deployments are blocklisted + await this.cancelBlocklistedAgreements(agreements) + // Use chain timestamp for consistency with contract timing and subgraph data const blockNumber = await this.network.networkProvider.getBlockNumber() const block = await this.network.networkProvider.getBlock(blockNumber) @@ -527,6 +623,7 @@ export class DipsManager { try { await this.tryCollectAgreement(agreement, blockNumber, logger) this.collectionTracker.updateAfterCollection(agreement.id, nowSeconds) + this.cleanupFinishedAgreement(agreement, nowSeconds, logger) } catch (err) { if (this.isDeterministicError(err)) { const parsedError = tryParseCustomError(err) @@ -547,6 +644,26 @@ export class DipsManager { } } + cleanupFinishedAgreement( + agreement: SubgraphIndexingAgreement, + nowSeconds: number, + logger: Logger, + ): boolean { + const isCancelledByPayer = agreement.state === 3 + const isExpired = + Number(agreement.endsAt) > 0 && Number(agreement.endsAt) < nowSeconds + + if (isCancelledByPayer || isExpired) { + logger.info('Agreement finished, removing from collection tracker', { + agreementId: agreement.id, + reason: isCancelledByPayer ? 'payer-cancelled' : 'expired', + }) + this.collectionTracker.remove(agreement.id) + return true + } + return false + } + private async tryCollectAgreement( agreement: SubgraphIndexingAgreement, blockNumber: number, From e7e08d07d4b9a0b580b5595788bc4d48d03f8333 Mon Sep 17 00:00:00 2001 From: Miguel de Elias Date: Tue, 28 Apr 2026 13:04:00 -0300 Subject: [PATCH 2/3] fix: check offchain decision basis for dips agreements --- .../src/indexing-fees/__tests__/dips.test.ts | 23 ++++++++++++- .../indexer-common/src/indexing-fees/dips.ts | 32 ++++++++++--------- 2 files changed, 39 insertions(+), 16 deletions(-) diff --git a/packages/indexer-common/src/indexing-fees/__tests__/dips.test.ts b/packages/indexer-common/src/indexing-fees/__tests__/dips.test.ts index f557d21e1..847a91881 100644 --- a/packages/indexer-common/src/indexing-fees/__tests__/dips.test.ts +++ b/packages/indexer-common/src/indexing-fees/__tests__/dips.test.ts @@ -683,7 +683,28 @@ describe('DipsManager', () => { expect(cancelSpy).toHaveBeenCalledWith(mockAgreement.id, mockAgreement) }) - test('does not cancel agreements without NEVER rule', async () => { + test('cancels agreements with OFFCHAIN rule for their deployment', async () => { + await managementModels.IndexingRule.create({ + identifier: testDeploymentId, + identifierType: SubgraphIdentifierType.DEPLOYMENT, + decisionBasis: IndexingDecisionBasis.OFFCHAIN, + requireSupported: true, + safety: true, + protocolNetwork: 'eip155:421614', + allocationAmount: '0', + }) + + const cancelSpy = jest + .spyOn(dipsManager, 'cancelAgreement') + .mockResolvedValue(true) + + await dipsManager.cancelBlocklistedAgreements([mockAgreement]) + + expect(cancelSpy).toHaveBeenCalledTimes(1) + expect(cancelSpy).toHaveBeenCalledWith(mockAgreement.id, mockAgreement) + }) + + test('does not cancel agreements without NEVER or OFFCHAIN rule', async () => { const cancelSpy = jest .spyOn(dipsManager, 'cancelAgreement') .mockResolvedValue(true) diff --git a/packages/indexer-common/src/indexing-fees/dips.ts b/packages/indexer-common/src/indexing-fees/dips.ts index ab4304fd5..25e308692 100644 --- a/packages/indexer-common/src/indexing-fees/dips.ts +++ b/packages/indexer-common/src/indexing-fees/dips.ts @@ -214,11 +214,8 @@ export class DipsManager { identifierType: SubgraphIdentifierType.DEPLOYMENT, }, }) - const blocklistedRule = allDeploymentRules.find( - (rule) => - new SubgraphDeploymentID(rule.identifier).bytes32 === - subgraphDeploymentID.bytes32 && - rule.decisionBasis === IndexingDecisionBasis.NEVER, + const blocklistedRule = allDeploymentRules.find((rule) => + this.isOnChainOptOutRule(rule, subgraphDeploymentID), ) if (blocklistedRule) { @@ -557,11 +554,8 @@ export class DipsManager { const subgraphDeploymentID = new SubgraphDeploymentID( agreement.subgraphDeploymentId, ) - const blocklistedRule = allDeploymentRules.find( - (rule) => - new SubgraphDeploymentID(rule.identifier).bytes32 === - subgraphDeploymentID.bytes32 && - rule.decisionBasis === IndexingDecisionBasis.NEVER, + const blocklistedRule = allDeploymentRules.find((rule) => + this.isOnChainOptOutRule(rule, subgraphDeploymentID), ) if (blocklistedRule) { @@ -764,6 +758,17 @@ export class DipsManager { return typedError?.code === 'CALL_EXCEPTION' } + private isOnChainOptOutRule( + rule: IndexingRuleAttributes, + deploymentId: SubgraphDeploymentID, + ): boolean { + return ( + new SubgraphDeploymentID(rule.identifier).bytes32 === deploymentId.bytes32 && + (rule.decisionBasis === IndexingDecisionBasis.NEVER || + rule.decisionBasis === IndexingDecisionBasis.OFFCHAIN) + ) + } + private async cleanupDipsRule( consumer: PendingRcaConsumer, proposal: DecodedRcaProposal, @@ -824,11 +829,8 @@ export class DipsManager { identifierType: SubgraphIdentifierType.DEPLOYMENT, }, }) - const blocklistedRule = allDeploymentRules.find( - (rule) => - new SubgraphDeploymentID(rule.identifier).bytes32 === - subgraphDeploymentID.bytes32 && - rule.decisionBasis === IndexingDecisionBasis.NEVER, + const blocklistedRule = allDeploymentRules.find((rule) => + this.isOnChainOptOutRule(rule, subgraphDeploymentID), ) if (blocklistedRule) { this.logger.info( From 400deb7543e0156c5b5167bf1d6ea5f8a247940a Mon Sep 17 00:00:00 2001 From: Maikol <86025070+Maikol@users.noreply.github.com> Date: Tue, 28 Apr 2026 16:37:32 -0300 Subject: [PATCH 3/3] feat: migrate to new indexing payments subgraph (#1201) --- packages/indexer-agent/src/__tests__/agent.ts | 105 ++++++++++++++++++ packages/indexer-agent/src/agent.ts | 32 ++++++ packages/indexer-agent/src/commands/start.ts | 24 ++++ .../src/indexer-management/monitor.ts | 33 +++--- .../resolvers/allocations.ts | 2 +- .../__tests__/agreement-monitor.test.ts | 15 +-- .../__tests__/collect-agreements.test.ts | 14 ++- .../src/indexing-fees/__tests__/dips.test.ts | 59 ++++------ .../src/indexing-fees/agreement-monitor.ts | 14 ++- .../indexer-common/src/indexing-fees/dips.ts | 10 +- .../src/network-specification.ts | 1 + packages/indexer-common/src/network.ts | 39 +++++++ 12 files changed, 273 insertions(+), 75 deletions(-) diff --git a/packages/indexer-agent/src/__tests__/agent.ts b/packages/indexer-agent/src/__tests__/agent.ts index 2d3c181b0..1d19b86e2 100644 --- a/packages/indexer-agent/src/__tests__/agent.ts +++ b/packages/indexer-agent/src/__tests__/agent.ts @@ -1,5 +1,6 @@ import { Agent, + addIndexingPaymentsSubgraphToTarget, convertSubgraphBasedRulesToDeploymentBased, consolidateAllocationDecisions, resolveTargetDeployments, @@ -472,3 +473,107 @@ describe('reconcileDeploymentAllocationAction', () => { expect(operator.presentPOIForAllocations).not.toHaveBeenCalled() }) }) + +describe('addIndexingPaymentsSubgraphToTarget function', () => { + const paymentsDeployment = new SubgraphDeploymentID( + 'QmddSbatCN1XmufoBm1bBwPx4L3FtuMAMHUNNBSPzrgL2a', + ) + + it('pushes the deployment when enableDips is true and deployment is defined', () => { + const target: SubgraphDeploymentID[] = [] + addIndexingPaymentsSubgraphToTarget(true, paymentsDeployment, target) + expect(target.map(d => d.bytes32)).toContain(paymentsDeployment.bytes32) + }) + + it('does nothing when enableDips is false', () => { + const target: SubgraphDeploymentID[] = [] + addIndexingPaymentsSubgraphToTarget(false, paymentsDeployment, target) + expect(target).toHaveLength(0) + }) + + it('does nothing when deployment is undefined', () => { + const target: SubgraphDeploymentID[] = [] + addIndexingPaymentsSubgraphToTarget(true, undefined, target) + expect(target).toHaveLength(0) + }) + + it('does not duplicate an already-present deployment', () => { + const target: SubgraphDeploymentID[] = [paymentsDeployment] + addIndexingPaymentsSubgraphToTarget(true, paymentsDeployment, target) + expect(target).toHaveLength(1) + }) +}) + +describe('reconcileDeployments indexing-payments carve-out wiring', () => { + const paymentsDeployment = new SubgraphDeploymentID( + 'QmddSbatCN1XmufoBm1bBwPx4L3FtuMAMHUNNBSPzrgL2a', + ) + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const mockLogger: any = { + child: jest.fn().mockReturnThis(), + info: jest.fn(), + debug: jest.fn(), + warn: jest.fn(), + error: jest.fn(), + trace: jest.fn(), + } + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + function createAgentUnderTest(network: any) { + const agent = Object.create(Agent.prototype) + agent.logger = mockLogger + agent.offchainSubgraphs = [] + agent.multiNetworks = { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + map: async (fn: any) => Promise.all([fn({ network })]), + } + agent.graphNode = { + subgraphDeploymentsAssignments: jest.fn().mockResolvedValue([]), + ensure: jest.fn().mockResolvedValue(undefined), + pause: jest.fn().mockResolvedValue(undefined), + } + return agent + } + + it('schedules the indexing-payments deployment for indexing when DIPS is enabled', async () => { + const agent = createAgentUnderTest({ + networkSubgraph: { deployment: undefined }, + specification: { indexerOptions: { enableDips: true } }, + indexingPaymentsSubgraph: { deployment: { id: paymentsDeployment } }, + }) + + await agent.reconcileDeployments([], [], []) + + const ensureCalls = agent.graphNode.ensure.mock.calls + const ensuredDeployments = ensureCalls.map( + // eslint-disable-next-line @typescript-eslint/no-explicit-any + (call: any[]) => (call[1] as SubgraphDeploymentID).bytes32, + ) + expect(ensuredDeployments).toContain(paymentsDeployment.bytes32) + }) + + it('does not schedule the indexing-payments deployment when DIPS is disabled', async () => { + const agent = createAgentUnderTest({ + networkSubgraph: { deployment: undefined }, + specification: { indexerOptions: { enableDips: false } }, + indexingPaymentsSubgraph: { deployment: { id: paymentsDeployment } }, + }) + + await agent.reconcileDeployments([], [], []) + + expect(agent.graphNode.ensure).not.toHaveBeenCalled() + }) + + it('does not schedule the indexing-payments deployment when deployment is undefined', async () => { + const agent = createAgentUnderTest({ + networkSubgraph: { deployment: undefined }, + specification: { indexerOptions: { enableDips: true } }, + indexingPaymentsSubgraph: { deployment: undefined }, + }) + + await agent.reconcileDeployments([], [], []) + + expect(agent.graphNode.ensure).not.toHaveBeenCalled() + }) +}) diff --git a/packages/indexer-agent/src/agent.ts b/packages/indexer-agent/src/agent.ts index b4b5b98d1..24ff8a63b 100644 --- a/packages/indexer-agent/src/agent.ts +++ b/packages/indexer-agent/src/agent.ts @@ -55,6 +55,16 @@ const deploymentInList = ( ): boolean => list.find(item => item.bytes32 === deployment.bytes32) !== undefined +export function addIndexingPaymentsSubgraphToTarget( + enableDips: boolean, + deployment: SubgraphDeploymentID | undefined, + targetDeployments: SubgraphDeploymentID[], +): void { + if (!enableDips || !deployment) return + if (deploymentInList(targetDeployments, deployment)) return + targetDeployments.push(deployment) +} + const deploymentRuleInList = ( list: IndexingRuleAttributes[], deployment: SubgraphDeploymentID, @@ -906,6 +916,17 @@ export class Agent { } }) + // ---------------------------------------------------------------------------------------- + // Ensure the indexing-payments subgraph is always indexed when DIPS is enabled + // ---------------------------------------------------------------------------------------- + await this.multiNetworks.map(async ({ network }) => { + addIndexingPaymentsSubgraphToTarget( + network.specification.indexerOptions.enableDips, + network.indexingPaymentsSubgraph?.deployment?.id, + targetDeployments, + ) + }) + // ---------------------------------------------------------------------------------------- // Inspect Deployments and Networks // ---------------------------------------------------------------------------------------- @@ -1328,6 +1349,17 @@ export class Agent { network.specification.networkIdentifier, ) } + // Indexing payments subgraph (DIPS) + if ( + network.specification.indexerOptions.enableDips && + network.specification.subgraphs.indexingPaymentsSubgraph?.deployment !== + undefined + ) { + await this.ensureSubgraphIndexing( + network.specification.subgraphs.indexingPaymentsSubgraph.deployment, + network.specification.networkIdentifier, + ) + } } } diff --git a/packages/indexer-agent/src/commands/start.ts b/packages/indexer-agent/src/commands/start.ts index 98d8ce0a8..6d28caf53 100644 --- a/packages/indexer-agent/src/commands/start.ts +++ b/packages/indexer-agent/src/commands/start.ts @@ -198,6 +198,19 @@ export const start = { type: 'string', group: 'TAP Subgraph', }) + .option('indexing-payments-subgraph-deployment', { + description: + 'Indexing payments subgraph deployment (for local hosting)', + array: false, + type: 'string', + group: 'Indexing Fees ("DIPs")', + }) + .option('indexing-payments-subgraph-endpoint', { + description: 'Endpoint to query the indexing payments subgraph from', + array: false, + type: 'string', + group: 'Indexing Fees ("DIPs")', + }) .option('allocate-on-network-subgraph', { description: 'Whether to allocate to the network subgraph', type: 'boolean', @@ -434,6 +447,13 @@ export const start = { if (argv['enable-dips'] && !argv['dipper-endpoint']) { return 'Invalid --dipper-endpoint provided. Must be provided when --enable-dips is true.' } + if ( + argv['enable-dips'] && + !argv['indexing-payments-subgraph-endpoint'] && + !argv['indexing-payments-subgraph-deployment'] + ) { + return 'At least one of --indexing-payments-subgraph-endpoint and --indexing-payments-subgraph-deployment must be provided when --enable-dips is true.' + } return true }) }, @@ -507,6 +527,10 @@ export async function createNetworkSpecification( deployment: argv.tapSubgraphDeployment, url: argv.tapSubgraphEndpoint, }, + indexingPaymentsSubgraph: { + deployment: argv.indexingPaymentsSubgraphDeployment, + url: argv.indexingPaymentsSubgraphEndpoint, + }, } const networkProvider = { diff --git a/packages/indexer-common/src/indexer-management/monitor.ts b/packages/indexer-common/src/indexer-management/monitor.ts index f9475a2e3..6de5f7e2f 100644 --- a/packages/indexer-common/src/indexer-management/monitor.ts +++ b/packages/indexer-common/src/indexer-management/monitor.ts @@ -55,27 +55,28 @@ export class NetworkMonitor { private networkSubgraph: SubgraphClient, private ethereum: Provider, private epochSubgraph: SubgraphClient, + private indexingPaymentsSubgraph?: SubgraphClient, ) {} async hasActiveDipsAgreement(allocationId: string): Promise { - try { - const result = await this.networkSubgraph.checkedQuery( - gql` - query indexingAgreements($allocationId: Bytes!) { - indexingAgreements( - where: { allocationId: $allocationId, state_not: 0 } - first: 1 - ) { - id - } - } - `, - { allocationId: allocationId.toLowerCase() }, - ) - return (result.data?.indexingAgreements?.length ?? 0) > 0 - } catch { + // No DIPS subgraph configured → no agreement can exist + if (!this.indexingPaymentsSubgraph) { return false } + const result = await this.indexingPaymentsSubgraph.checkedQuery( + gql` + query indexingAgreements($allocationId: Bytes!) { + indexingAgreements( + where: { allocationId: $allocationId, state_in: [Accepted, CanceledByPayer] } + first: 1 + ) { + id + } + } + `, + { allocationId: allocationId.toLowerCase() }, + ) + return (result.data?.indexingAgreements?.length ?? 0) > 0 } poiDisputeMonitoringEnabled(): boolean { diff --git a/packages/indexer-common/src/indexer-management/resolvers/allocations.ts b/packages/indexer-common/src/indexer-management/resolvers/allocations.ts index 1186e4626..e81d21cfd 100644 --- a/packages/indexer-common/src/indexer-management/resolvers/allocations.ts +++ b/packages/indexer-common/src/indexer-management/resolvers/allocations.ts @@ -1686,7 +1686,7 @@ export default { force: boolean protocolNetwork: string }, - { logger, models, multiNetworks, actionManager }: IndexerManagementResolverContext, + { logger, models, multiNetworks }: IndexerManagementResolverContext, ): Promise => { logger.debug('Execute closeAllocation() mutation', { allocationID: allocation, diff --git a/packages/indexer-common/src/indexing-fees/__tests__/agreement-monitor.test.ts b/packages/indexer-common/src/indexing-fees/__tests__/agreement-monitor.test.ts index f1788d15c..1911bbe42 100644 --- a/packages/indexer-common/src/indexing-fees/__tests__/agreement-monitor.test.ts +++ b/packages/indexer-common/src/indexing-fees/__tests__/agreement-monitor.test.ts @@ -1,8 +1,5 @@ -/* eslint-disable @typescript-eslint/no-explicit-any,@typescript-eslint/no-unused-vars */ -import { - fetchCollectableAgreements, - SubgraphIndexingAgreement, -} from '../agreement-monitor' +/* eslint-disable @typescript-eslint/no-explicit-any */ +import { fetchCollectableAgreements } from '../agreement-monitor' const mockQuery = jest.fn() const mockNetworkSubgraph = { query: mockQuery } as any @@ -22,7 +19,7 @@ describe('fetchCollectableAgreements', () => { id: '0x00000000000000000000000000000001', allocationId: '0xaaaa', subgraphDeploymentId: '0xbbbb', - state: 1, + state: 'Accepted', lastCollectionAt: '1000', endsAt: '9999999999', maxInitialTokens: '1000000', @@ -41,7 +38,7 @@ describe('fetchCollectableAgreements', () => { expect(result).toHaveLength(1) expect(result[0].id).toBe('0x00000000000000000000000000000001') - expect(result[0].state).toBe(1) + expect(result[0].state).toBe('Accepted') expect(mockQuery).toHaveBeenCalledTimes(1) }) @@ -61,7 +58,7 @@ describe('fetchCollectableAgreements', () => { id: `0x${i.toString(16).padStart(32, '0')}`, allocationId: '0xaaaa', subgraphDeploymentId: '0xbbbb', - state: 1, + state: 'Accepted', lastCollectionAt: '1000', endsAt: '9999999999', maxInitialTokens: '1000000', @@ -78,7 +75,7 @@ describe('fetchCollectableAgreements', () => { id: '0x' + 'f'.repeat(32), allocationId: '0xaaaa', subgraphDeploymentId: '0xbbbb', - state: 1, + state: 'Accepted', lastCollectionAt: '1000', endsAt: '9999999999', maxInitialTokens: '1000000', diff --git a/packages/indexer-common/src/indexing-fees/__tests__/collect-agreements.test.ts b/packages/indexer-common/src/indexing-fees/__tests__/collect-agreements.test.ts index 94c8d3caf..e84a436cb 100644 --- a/packages/indexer-common/src/indexing-fees/__tests__/collect-agreements.test.ts +++ b/packages/indexer-common/src/indexing-fees/__tests__/collect-agreements.test.ts @@ -1,6 +1,7 @@ /* eslint-disable @typescript-eslint/no-explicit-any */ import { Logger } from '@graphprotocol/common-ts' import { DipsManager } from '../dips' +import type { SubgraphIndexingAgreement } from '../agreement-monitor' const logger = { child: jest.fn().mockReturnThis(), @@ -40,6 +41,7 @@ const mockGraphNode = { const mockNetwork = { contracts: mockContracts, networkSubgraph: mockNetworkSubgraph, + indexingPaymentsSubgraph: mockNetworkSubgraph, transactionManager: mockTransactionManager, specification: { indexerOptions: { @@ -55,20 +57,26 @@ const mockNetwork = { }, } as any -const mockModels = {} as any +const mockModels = { + IndexingRule: { + findAll: jest.fn().mockResolvedValue([]), + }, +} as any function createDipsManager(): DipsManager { return new DipsManager(logger, mockModels, mockNetwork, mockGraphNode, null) } // Helper: agreement that was last collected long ago (ready to collect) -function makeReadyAgreement(id = '0x00000000000000000000000000000001') { +function makeReadyAgreement( + id = '0x00000000000000000000000000000001', +): SubgraphIndexingAgreement { return { id, allocationId: '0xaaaa', subgraphDeploymentId: '0xbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb', - state: 1, + state: 'Accepted', lastCollectionAt: '0', // never collected → always ready endsAt: '9999999999', maxInitialTokens: '1000000', diff --git a/packages/indexer-common/src/indexing-fees/__tests__/dips.test.ts b/packages/indexer-common/src/indexing-fees/__tests__/dips.test.ts index 847a91881..5d954d73c 100644 --- a/packages/indexer-common/src/indexing-fees/__tests__/dips.test.ts +++ b/packages/indexer-common/src/indexing-fees/__tests__/dips.test.ts @@ -465,7 +465,7 @@ describe('DipsManager', () => { id: '0x123e4567e89b12d3a456426614174000', allocationId: '0xabcd47df40c29949a75a6693c77834c00b8ad626', subgraphDeploymentId: 'QmTZ8ejXJxRo7vDBS4uwqBeGoxLSWbhaA7oXa1RvxunLy7', - state: 1, // Accepted + state: 'Accepted', lastCollectionAt: '0', endsAt: '9999999999', maxInitialTokens: '1000', @@ -499,22 +499,15 @@ describe('DipsManager', () => { // Mock block number and graph node methods for collect network.networkProvider.getBlockNumber = jest.fn().mockResolvedValue(100) graphNode.entityCount = jest.fn().mockResolvedValue([250000]) - graphNode.subgraphFeatures = jest - .fn() - .mockResolvedValue({ network: 'mainnet' }) - graphNode.blockHashFromNumber = jest - .fn() - .mockResolvedValue('0xblockhash') + graphNode.subgraphFeatures = jest.fn().mockResolvedValue({ network: 'mainnet' }) + graphNode.blockHashFromNumber = jest.fn().mockResolvedValue('0xblockhash') graphNode.proofOfIndexing = jest .fn() .mockResolvedValue( '0x0000000000000000000000000000000000000000000000000000000000000001', ) - const result = await dipsManager.cancelAgreement( - mockAgreement.id, - mockAgreement, - ) + const result = await dipsManager.cancelAgreement(mockAgreement.id, mockAgreement) expect(result).toBe(true) // executeTransaction called twice: once for cancel, once for collect @@ -531,10 +524,7 @@ describe('DipsManager', () => { .fn() .mockRejectedValueOnce(new Error('cancel tx reverted')) - const result = await dipsManager.cancelAgreement( - mockAgreement.id, - mockAgreement, - ) + const result = await dipsManager.cancelAgreement(mockAgreement.id, mockAgreement) expect(result).toBe(false) // executeTransaction called only once (for cancel) @@ -553,22 +543,15 @@ describe('DipsManager', () => { // Mock block number and graph node methods network.networkProvider.getBlockNumber = jest.fn().mockResolvedValue(100) graphNode.entityCount = jest.fn().mockResolvedValue([250000]) - graphNode.subgraphFeatures = jest - .fn() - .mockResolvedValue({ network: 'mainnet' }) - graphNode.blockHashFromNumber = jest - .fn() - .mockResolvedValue('0xblockhash') + graphNode.subgraphFeatures = jest.fn().mockResolvedValue({ network: 'mainnet' }) + graphNode.blockHashFromNumber = jest.fn().mockResolvedValue('0xblockhash') graphNode.proofOfIndexing = jest .fn() .mockResolvedValue( '0x0000000000000000000000000000000000000000000000000000000000000001', ) - const result = await dipsManager.cancelAgreement( - mockAgreement.id, - mockAgreement, - ) + const result = await dipsManager.cancelAgreement(mockAgreement.id, mockAgreement) expect(result).toBe(true) // Tracker should be cleaned up even though collect failed @@ -583,7 +566,7 @@ describe('DipsManager', () => { id: '0x123e4567e89b12d3a456426614174000', allocationId: '0xabcd47df40c29949a75a6693c77834c00b8ad626', subgraphDeploymentId: 'QmTZ8ejXJxRo7vDBS4uwqBeGoxLSWbhaA7oXa1RvxunLy7', - state: 1, + state: 'Accepted', lastCollectionAt: '0', endsAt: '9999999999', maxInitialTokens: '1000', @@ -605,7 +588,7 @@ describe('DipsManager', () => { test('removes payer-cancelled agreement from tracker after collection', () => { const removeSpy = jest.spyOn(dipsManager.collectionTracker, 'remove') - const agreement = { ...baseAgreement, state: 3 } + const agreement = { ...baseAgreement, state: 'CanceledByPayer' as const } const result = dipsManager.cleanupFinishedAgreement(agreement, 1000, logger) @@ -616,13 +599,9 @@ describe('DipsManager', () => { test('removes expired agreement from tracker after collection', () => { const removeSpy = jest.spyOn(dipsManager.collectionTracker, 'remove') const nowSeconds = 2000 - const agreement = { ...baseAgreement, state: 1, endsAt: '1000' } + const agreement = { ...baseAgreement, state: 'Accepted' as const, endsAt: '1000' } - const result = dipsManager.cleanupFinishedAgreement( - agreement, - nowSeconds, - logger, - ) + const result = dipsManager.cleanupFinishedAgreement(agreement, nowSeconds, logger) expect(result).toBe(true) expect(removeSpy).toHaveBeenCalledWith(agreement.id) @@ -631,13 +610,13 @@ describe('DipsManager', () => { test('does not remove active agreement from tracker', () => { const removeSpy = jest.spyOn(dipsManager.collectionTracker, 'remove') const nowSeconds = 1000 - const agreement = { ...baseAgreement, state: 1, endsAt: '9999999999' } + const agreement = { + ...baseAgreement, + state: 'Accepted' as const, + endsAt: '9999999999', + } - const result = dipsManager.cleanupFinishedAgreement( - agreement, - nowSeconds, - logger, - ) + const result = dipsManager.cleanupFinishedAgreement(agreement, nowSeconds, logger) expect(result).toBe(false) expect(removeSpy).not.toHaveBeenCalled() @@ -649,7 +628,7 @@ describe('DipsManager', () => { id: '0x123e4567e89b12d3a456426614174000', allocationId: '0xabcd47df40c29949a75a6693c77834c00b8ad626', subgraphDeploymentId: 'QmTZ8ejXJxRo7vDBS4uwqBeGoxLSWbhaA7oXa1RvxunLy7', - state: 1, + state: 'Accepted', lastCollectionAt: '0', endsAt: '9999999999', maxInitialTokens: '1000', diff --git a/packages/indexer-common/src/indexing-fees/agreement-monitor.ts b/packages/indexer-common/src/indexing-fees/agreement-monitor.ts index 413256d5a..e34466d04 100644 --- a/packages/indexer-common/src/indexing-fees/agreement-monitor.ts +++ b/packages/indexer-common/src/indexing-fees/agreement-monitor.ts @@ -1,11 +1,17 @@ import gql from 'graphql-tag' import { SubgraphClient } from '../subgraph-client' +export type AgreementState = + | 'NotAccepted' + | 'Accepted' + | 'CanceledByServiceProvider' + | 'CanceledByPayer' + export interface SubgraphIndexingAgreement { id: string allocationId: string subgraphDeploymentId: string - state: number + state: AgreementState lastCollectionAt: string endsAt: string maxInitialTokens: string @@ -20,7 +26,7 @@ export interface SubgraphIndexingAgreement { const INDEXING_AGREEMENTS_QUERY = gql` query indexingAgreements($indexer: String!, $lastId: String!) { indexingAgreements( - where: { serviceProvider: $indexer, state_in: [1, 3], id_gt: $lastId } + where: { indexer: $indexer, state_in: [Accepted, CanceledByPayer], id_gt: $lastId } orderBy: id orderDirection: asc first: 1000 @@ -43,14 +49,14 @@ const INDEXING_AGREEMENTS_QUERY = gql` ` export async function fetchCollectableAgreements( - networkSubgraph: SubgraphClient, + subgraphClient: SubgraphClient, indexerAddress: string, ): Promise { const all: SubgraphIndexingAgreement[] = [] let lastId = '' for (;;) { - const result = await networkSubgraph.query(INDEXING_AGREEMENTS_QUERY, { + const result = await subgraphClient.query(INDEXING_AGREEMENTS_QUERY, { indexer: indexerAddress.toLowerCase(), lastId, }) diff --git a/packages/indexer-common/src/indexing-fees/dips.ts b/packages/indexer-common/src/indexing-fees/dips.ts index 25e308692..61861c387 100644 --- a/packages/indexer-common/src/indexing-fees/dips.ts +++ b/packages/indexer-common/src/indexing-fees/dips.ts @@ -572,8 +572,14 @@ export class DipsManager { const logger = this.logger.child({ function: 'collectAgreementPayments' }) const indexerAddress = this.network.specification.indexerOptions.address + if (!this.network.indexingPaymentsSubgraph) { + logger.warn( + 'Indexing payments subgraph not configured, skipping agreement collection', + ) + return + } const agreements = await fetchCollectableAgreements( - this.network.networkSubgraph, + this.network.indexingPaymentsSubgraph, indexerAddress, ) @@ -643,7 +649,7 @@ export class DipsManager { nowSeconds: number, logger: Logger, ): boolean { - const isCancelledByPayer = agreement.state === 3 + const isCancelledByPayer = agreement.state === 'CanceledByPayer' const isExpired = Number(agreement.endsAt) > 0 && Number(agreement.endsAt) < nowSeconds diff --git a/packages/indexer-common/src/network-specification.ts b/packages/indexer-common/src/network-specification.ts index 813d92314..d5b662f0f 100644 --- a/packages/indexer-common/src/network-specification.ts +++ b/packages/indexer-common/src/network-specification.ts @@ -153,6 +153,7 @@ export const ProtocolSubgraphs = z networkSubgraph: Subgraph, epochSubgraph: Subgraph, tapSubgraph: OptionalSubgraph, + indexingPaymentsSubgraph: OptionalSubgraph, }) .strict() // TODO: Ensure the `url` property is always defined until Epoch Subgraph diff --git a/packages/indexer-common/src/network.ts b/packages/indexer-common/src/network.ts index 8b35b3378..336abb93c 100644 --- a/packages/indexer-common/src/network.ts +++ b/packages/indexer-common/src/network.ts @@ -61,6 +61,7 @@ export class Network { tapCollector: TapCollector | undefined graphTallyCollector: GraphTallyCollector | undefined dipsCollector: DipsCollector | undefined + indexingPaymentsSubgraph: SubgraphClient | undefined specification: spec.NetworkSpecification paused: Eventual isOperator: Eventual @@ -84,6 +85,7 @@ export class Network { queryFeeModels: QueryFeeModels, managementModels: IndexerManagementModels, dipsCollector: DipsCollector | undefined, + indexingPaymentsSubgraph: SubgraphClient | undefined, ) { this.logger = logger this.contracts = contracts @@ -101,6 +103,7 @@ export class Network { this.queryFeeModels = queryFeeModels this.managementModels = managementModels this.dipsCollector = dipsCollector + this.indexingPaymentsSubgraph = indexingPaymentsSubgraph } static async create( @@ -187,6 +190,40 @@ export class Network { }) } + // * ----------------------------------------------------------------------- + // * Indexing Payments Subgraph + // * ----------------------------------------------------------------------- + let indexingPaymentsSubgraph: SubgraphClient | undefined = undefined + if (specification.subgraphs.indexingPaymentsSubgraph) { + const indexingPaymentsSubgraphDeploymentId = specification.subgraphs + .indexingPaymentsSubgraph.deployment + ? new SubgraphDeploymentID( + specification.subgraphs.indexingPaymentsSubgraph.deployment, + ) + : undefined + const indexingPaymentsFreshnessChecker = new SubgraphFreshnessChecker( + 'IndexingPayments Subgraph', + networkProvider, + specification.subgraphs.maxBlockDistance, + specification.subgraphs.freshnessSleepMilliseconds, + logger.child({ component: 'FreshnessChecker' }), + Infinity, + ) + indexingPaymentsSubgraph = await SubgraphClient.create({ + name: 'IndexingPaymentsSubgraph', + logger, + deployment: + indexingPaymentsSubgraphDeploymentId !== undefined + ? { + graphNode, + deployment: indexingPaymentsSubgraphDeploymentId, + } + : undefined, + endpoint: specification.subgraphs.indexingPaymentsSubgraph!.url, + subgraphFreshnessChecker: indexingPaymentsFreshnessChecker, + }) + } + // * ----------------------------------------------------------------------- // * Contracts // * ----------------------------------------------------------------------- @@ -257,6 +294,7 @@ export class Network { networkSubgraph, networkProvider, epochSubgraph, + indexingPaymentsSubgraph, ) // * ----------------------------------------------------------------------- @@ -389,6 +427,7 @@ export class Network { queryFeeModels, managementModels, dipsCollector, + indexingPaymentsSubgraph, ) }