fix: indexer collects final payment after payer-side cancel

packages/indexer-common/src/indexing-fees/__tests__/dips.test.ts

@@ -670,6 +670,62 @@ describe('DipsManager', () => {
           dipsManager.collectionTracker.isReadyForCollection(mockAgreement.id, 0),
         ).toBe(true)
       })
+
+      test('CanceledByPayer skips on-chain cancel and only runs final collect', async () => {
+        const canceledByPayer = { ...mockAgreement, state: 'CanceledByPayer' as const }
+        const mockCollectReceipt = { hash: '0xcollect456' }
+
+        network.transactionManager.executeTransaction = jest
+          .fn()
+          .mockResolvedValueOnce(mockCollectReceipt) // only collect, no cancel
+
+        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(
+          canceledByPayer.id,
+          canceledByPayer,
+        )
+
+        expect(result).toBe(true)
+        expect(network.transactionManager.executeTransaction).toHaveBeenCalledTimes(1)
+      })
+
+      test('CanceledByPayer with failing final collect still returns true', async () => {
+        const canceledByPayer = { ...mockAgreement, state: 'CanceledByPayer' as const }
+
+        network.transactionManager.executeTransaction = jest
+          .fn()
+          .mockRejectedValueOnce(new Error('collect failed'))
+
+        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(
+          canceledByPayer.id,
+          canceledByPayer,
+        )
+
+        expect(result).toBe(true)
+        expect(network.transactionManager.executeTransaction).toHaveBeenCalledTimes(1)
+        expect(
+          dipsManager.collectionTracker.isReadyForCollection(canceledByPayer.id, 0),
+        ).toBe(true)
+      })
     })
 
     describe('cleanupFinishedAgreement', () => {
@@ -803,6 +859,36 @@ describe('DipsManager', () => {
 
         expect(cancelSpy).not.toHaveBeenCalled()
       })
+
+      test('skips CanceledByPayer agreements even when a NEVER rule exists', async () => {
+        // Reproduces the bug where the payer canceled on-chain first
+        // (via dipper) and the agent's own NEVER rule for the closed
+        // allocation would otherwise route the agreement back through
+        // cancelAgreement, which would attempt a redundant on-chain
+        // cancel and skip the final collection.
+        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)
+
+        const canceledByPayer: SubgraphIndexingAgreement = {
+          ...mockAgreement,
+          state: 'CanceledByPayer',
+        }
+
+        await dipsManager.cancelBlocklistedAgreements([canceledByPayer])
+
+        expect(cancelSpy).not.toHaveBeenCalled()
+      })
     })
   })
 })

packages/indexer-common/src/indexing-fees/dips.ts

@@ -463,6 +463,8 @@ export class DipsManager {
     }
   }
 
+  // Returns true once the final-collect step ran (whether we canceled on-chain
+  // or the payer already had); false only when our own on-chain cancel failed.
   async cancelAgreement(
     agreementId: string,
     agreement: SubgraphIndexingAgreement,
@@ -472,36 +474,43 @@ export class DipsManager {
       agreementId,
     })
 
-    // Step 1: Cancel on-chain
+    // Step 1: Cancel on-chain (skipped if payer already canceled — a second
+    // cancel reverts on `InvalidAgreementState` and would skip the final collect).
     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 (agreement.state === 'CanceledByPayer') {
+      logger.info(
+        'Payer already canceled on-chain; skipping cancel, proceeding to final collection',
       )
+    } else {
+      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')
+        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
       }
-
-      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
@@ -511,7 +520,8 @@ export class DipsManager {
       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', {
+      logger.error('Final collection after cancel failed, fees may be lost', {
+        deployment: agreement.subgraphDeploymentId,
         error: errorMsg,
       })
     }
@@ -536,6 +546,12 @@ export class DipsManager {
     })
 
     for (const agreement of agreements) {
+      // Already-canceled agreements need a final collect, not another cancel —
+      // the regular collection loop handles them. cancelAgreement also guards
+      // this state internally as defense-in-depth.
+      if (agreement.state === 'CanceledByPayer') {
+        continue
+      }
       const subgraphDeploymentID = new SubgraphDeploymentID(
         agreement.subgraphDeploymentId,
       )