[IA 01] - fix: IndexingAgreement.collect() on CanceledByPayer

packages/horizon/contracts/interfaces/IRecurringCollector.sol

@@ -413,4 +413,11 @@ interface IRecurringCollector is IAuthorizable, IPaymentsCollector {
      * @return The AgreementData struct containing the agreement's data.
      */
     function getAgreement(bytes16 agreementId) external view returns (AgreementData memory);
+
+    /**
+     * @notice Checks if an agreement is collectable.
+     * @param agreement The agreement data
+     * @return The boolean indicating if the agreement is collectable
+     */
+    function isCollectable(AgreementData memory agreement) external view returns (bool);
 }

packages/horizon/contracts/payments/collectors/RecurringCollector.sol

@@ -249,6 +249,11 @@ contract RecurringCollector is EIP712, GraphDirectory, Authorizable, IRecurringC
         return _getAgreement(agreementId);
     }
 
+    /// @inheritdoc IRecurringCollector
+    function isCollectable(AgreementData memory agreement) external pure returns (bool) {
+        return _isCollectable(agreement);
+    }
+
     /**
      * @notice Decodes the collect data.
      * @param data The encoded collect parameters.
@@ -270,7 +275,7 @@ contract RecurringCollector is EIP712, GraphDirectory, Authorizable, IRecurringC
     function _collect(CollectParams memory _params) private returns (uint256) {
         AgreementData storage agreement = _getAgreementStorage(_params.agreementId);
         require(
-            agreement.state == AgreementState.Accepted || agreement.state == AgreementState.CanceledByPayer,
+            _isCollectable(agreement),
             RecurringCollectorAgreementIncorrectState(_params.agreementId, agreement.state)
         );
 
@@ -537,4 +542,13 @@ contract RecurringCollector is EIP712, GraphDirectory, Authorizable, IRecurringC
     function _agreementCollectionStartAt(AgreementData memory _agreement) private pure returns (uint256) {
         return _agreement.lastCollectionAt > 0 ? _agreement.lastCollectionAt : _agreement.acceptedAt;
     }
+
+    /**
+     * @notice Requires that the agreement is collectable.
+     * @param _agreement The agreement data
+     * @return The boolean indicating if the agreement is collectable
+     */
+    function _isCollectable(AgreementData memory _agreement) private pure returns (bool) {
+        return _agreement.state == AgreementState.Accepted || _agreement.state == AgreementState.CanceledByPayer;
+    }
 }

packages/subgraph-service/contracts/libraries/IndexingAgreement.sol

@@ -254,6 +254,12 @@ library IndexingAgreement {
      */
     error IndexingAgreementNotActive(bytes16 agreementId);
 
+    /**
+     * @notice Thrown when the agreement is not collectable
+     * @param agreementId The agreement ID
+     */
+    error IndexingAgreementNotCollectable(bytes16 agreementId);
+
     /**
      * @notice Thrown when trying to interact with an agreement not owned by the indexer
      * @param agreementId The agreement ID
@@ -517,7 +523,7 @@ library IndexingAgreement {
             wrapper.agreement.allocationId,
             wrapper.collectorAgreement.serviceProvider
         );
-        require(_isActive(wrapper), IndexingAgreementNotActive(params.agreementId));
+        require(_isCollectable(wrapper), IndexingAgreementNotCollectable(params.agreementId));
 
         require(
             wrapper.agreement.version == IndexingAgreementVersion.V1,
@@ -692,17 +698,37 @@ library IndexingAgreement {
     /**
      * @notice Checks if the agreement is active
      * Requirements:
+     * - The indexing agreement is valid
      * - The underlying collector agreement has been accepted
-     * - The underlying collector agreement's data service is this contract
-     * - The indexing agreement has been accepted and has a valid allocation ID
      * @param wrapper The agreement wrapper containing the indexing agreement and collector agreement data
      * @return True if the agreement is active, false otherwise
      **/
     function _isActive(AgreementWrapper memory wrapper) private view returns (bool) {
-        return
-            wrapper.collectorAgreement.dataService == address(this) &&
-            wrapper.collectorAgreement.state == IRecurringCollector.AgreementState.Accepted &&
-            wrapper.agreement.allocationId != address(0);
+        return _isValid(wrapper) && wrapper.collectorAgreement.state == IRecurringCollector.AgreementState.Accepted;
+    }
+
+    /**
+     * @notice Checks if the agreement is collectable
+     * Requirements:
+     * - The indexing agreement is valid
+     * - The underlying collector agreement is collectable
+     * @param wrapper The agreement wrapper containing the indexing agreement and collector agreement data
+     * @return True if the agreement is collectable, false otherwise
+     **/
+    function _isCollectable(AgreementWrapper memory wrapper) private view returns (bool) {
+        return _isValid(wrapper) && _directory().recurringCollector().isCollectable(wrapper.collectorAgreement);
+    }
+
+    /**
+     * @notice Checks if the agreement is valid
+     * Requirements:
+     * - The underlying collector agreement's data service is this contract
+     * - The indexing agreement has been accepted and has a valid allocation ID
+     * @param wrapper The agreement wrapper containing the indexing agreement and collector agreement data
+     * @return True if the agreement is valid, false otherwise
+     **/
+    function _isValid(AgreementWrapper memory wrapper) private view returns (bool) {
+        return wrapper.collectorAgreement.dataService == address(this) && wrapper.agreement.allocationId != address(0);
     }
 
     /**

packages/subgraph-service/test/unit/subgraphService/indexing-agreement/integration.t.sol

@@ -18,6 +18,13 @@ contract SubgraphServiceIndexingAgreementIntegrationTest is SubgraphServiceIndex
         uint256 indexerTokensLocked;
     }
 
+    struct ExpectedTokens {
+        uint256 expectedTotalTokensCollected;
+        uint256 expectedTokensLocked;
+        uint256 expectedProtocolTokensBurnt;
+        uint256 expectedIndexerTokensCollected;
+    }
+
     /*
      * TESTS
      */
@@ -27,81 +34,164 @@ contract SubgraphServiceIndexingAgreementIntegrationTest is SubgraphServiceIndex
         Seed memory seed,
         uint256 fuzzyTokensCollected
     ) public {
-        uint256 expectedTotalTokensCollected = bound(fuzzyTokensCollected, 1000, 1_000_000);
-        uint256 expectedTokensLocked = stakeToFeesRatio * expectedTotalTokensCollected;
-        uint256 expectedProtocolTokensBurnt = expectedTotalTokensCollected.mulPPMRoundUp(
-            graphPayments.PROTOCOL_PAYMENT_CUT()
-        );
-        uint256 expectedIndexerTokensCollected = expectedTotalTokensCollected - expectedProtocolTokensBurnt;
-
+        // Setup
+        ExpectedTokens memory expectedTokens = _newExptectedTokens(fuzzyTokensCollected);
         Context storage ctx = _newCtx(seed);
         IndexerState memory indexerState = _withIndexer(ctx);
-        _addTokensToProvision(indexerState, expectedTokensLocked);
+        _addTokensToProvision(indexerState, expectedTokens.expectedTokensLocked);
         IRecurringCollector.RecurringCollectionAgreement memory rca = _recurringCollectorHelper.sensibleRCA(
             ctx.ctxInternal.seed.rca
         );
-        uint256 agreementTokensPerSecond = 1;
-        rca.deadline = uint64(block.timestamp); // accept now
-        rca.endsAt = type(uint64).max; // no expiration
-        rca.maxInitialTokens = 0; // no initial payment
-        rca.maxOngoingTokensPerSecond = type(uint32).max; // unlimited tokens per second
-        rca.minSecondsPerCollection = 1; // 1 second between collections
-        rca.maxSecondsPerCollection = type(uint32).max; // no maximum time between collections
-        rca.serviceProvider = indexerState.addr; // service provider is the indexer
-        rca.dataService = address(subgraphService); // data service is the subgraph service
-        rca.metadata = _encodeAcceptIndexingAgreementMetadataV1(
-            indexerState.subgraphDeploymentId,
-            IndexingAgreement.IndexingAgreementTermsV1({
-                tokensPerSecond: agreementTokensPerSecond,
-                tokensPerEntityPerSecond: 0 // no payment for entities
-            })
-        );
+        _sharedSetup(ctx, rca, indexerState, expectedTokens);
 
-        _setupPayerWithEscrow(rca.payer, ctx.payer.signerPrivateKey, indexerState.addr, expectedTotalTokensCollected);
+        TestState memory beforeCollect = _getState(rca.payer, indexerState.addr);
 
+        // Collect
         resetPrank(indexerState.addr);
-        // Set the payments destination to the indexer address
-        subgraphService.setPaymentsDestination(indexerState.addr);
-        // Accept the Indexing Agreement
-        subgraphService.acceptIndexingAgreement(
-            indexerState.allocationId,
-            _recurringCollectorHelper.generateSignedRCA(rca, ctx.payer.signerPrivateKey)
+        uint256 tokensCollected = subgraphService.collect(
+            indexerState.addr,
+            IGraphPayments.PaymentTypes.IndexingFee,
+            _encodeCollectDataV1(
+                rca.agreementId,
+                1,
+                keccak256(abi.encodePacked("poi")),
+                epochManager.currentEpochBlock(),
+                bytes("")
+            )
         );
-        // Skip ahead to collection point
-        skip(expectedTotalTokensCollected / agreementTokensPerSecond);
-        // vm.assume(block.timestamp < type(uint64).max);
+
+        TestState memory afterCollect = _getState(rca.payer, indexerState.addr);
+        _sharedAssert(beforeCollect, afterCollect, expectedTokens, tokensCollected);
+    }
+
+    function test_SubgraphService_CollectIndexingFee_WhenCanceledByPayer_Integration(
+        Seed memory seed,
+        uint256 fuzzyTokensCollected
+    ) public {
+        // Setup
+        ExpectedTokens memory expectedTokens = _newExptectedTokens(fuzzyTokensCollected);
+        Context storage ctx = _newCtx(seed);
+        IndexerState memory indexerState = _withIndexer(ctx);
+        IRecurringCollector.RecurringCollectionAgreement memory rca = _recurringCollectorHelper.sensibleRCA(
+            ctx.ctxInternal.seed.rca
+        );
+        _sharedSetup(ctx, rca, indexerState, expectedTokens);
+
+        // Cancel the indexing agreement by the payer
+        resetPrank(ctx.payer.signer);
+        subgraphService.cancelIndexingAgreementByPayer(rca.agreementId);
+
         TestState memory beforeCollect = _getState(rca.payer, indexerState.addr);
-        bytes16 agreementId = rca.agreementId;
+
+        // Collect
+        resetPrank(indexerState.addr);
         uint256 tokensCollected = subgraphService.collect(
             indexerState.addr,
             IGraphPayments.PaymentTypes.IndexingFee,
             _encodeCollectDataV1(
-                agreementId,
+                rca.agreementId,
                 1,
                 keccak256(abi.encodePacked("poi")),
                 epochManager.currentEpochBlock(),
                 bytes("")
             )
         );
+
         TestState memory afterCollect = _getState(rca.payer, indexerState.addr);
-        uint256 indexerTokensCollected = afterCollect.indexerBalance - beforeCollect.indexerBalance;
-        uint256 protocolTokensBurnt = tokensCollected - indexerTokensCollected;
+        _sharedAssert(beforeCollect, afterCollect, expectedTokens, tokensCollected);
+    }
+
+    /* solhint-enable graph/func-name-mixedcase */
+
+    function _sharedSetup(
+        Context storage _ctx,
+        IRecurringCollector.RecurringCollectionAgreement memory _rca,
+        IndexerState memory _indexerState,
+        ExpectedTokens memory _expectedTokens
+    ) internal {
+        _addTokensToProvision(_indexerState, _expectedTokens.expectedTokensLocked);
+
+        IndexingAgreement.IndexingAgreementTermsV1 memory terms = IndexingAgreement.IndexingAgreementTermsV1({
+            tokensPerSecond: 1,
+            tokensPerEntityPerSecond: 0 // no payment for entities
+        });
+        _rca.deadline = uint64(block.timestamp); // accept now
+        _rca.endsAt = type(uint64).max; // no expiration
+        _rca.maxInitialTokens = 0; // no initial payment
+        _rca.maxOngoingTokensPerSecond = type(uint32).max; // unlimited tokens per second
+        _rca.minSecondsPerCollection = 1; // 1 second between collections
+        _rca.maxSecondsPerCollection = type(uint32).max; // no maximum time between collections
+        _rca.serviceProvider = _indexerState.addr; // service provider is the indexer
+        _rca.dataService = address(subgraphService); // data service is the subgraph service
+        _rca.metadata = _encodeAcceptIndexingAgreementMetadataV1(_indexerState.subgraphDeploymentId, terms);
+
+        _setupPayerWithEscrow(
+            _rca.payer,
+            _ctx.payer.signerPrivateKey,
+            _indexerState.addr,
+            _expectedTokens.expectedTotalTokensCollected
+        );
+
+        resetPrank(_indexerState.addr);
+        // Set the payments destination to the indexer address
+        subgraphService.setPaymentsDestination(_indexerState.addr);
+
+        // Accept the Indexing Agreement
+        subgraphService.acceptIndexingAgreement(
+            _indexerState.allocationId,
+            _recurringCollectorHelper.generateSignedRCA(_rca, _ctx.payer.signerPrivateKey)
+        );
+
+        // Skip ahead to collection point
+        skip(_expectedTokens.expectedTotalTokensCollected / terms.tokensPerSecond);
+    }
+
+    function _newExptectedTokens(uint256 _fuzzyTokensCollected) internal view returns (ExpectedTokens memory) {
+        uint256 expectedTotalTokensCollected = bound(_fuzzyTokensCollected, 1000, 1_000_000);
+        uint256 expectedTokensLocked = stakeToFeesRatio * expectedTotalTokensCollected;
+        uint256 expectedProtocolTokensBurnt = expectedTotalTokensCollected.mulPPMRoundUp(
+            graphPayments.PROTOCOL_PAYMENT_CUT()
+        );
+        uint256 expectedIndexerTokensCollected = expectedTotalTokensCollected - expectedProtocolTokensBurnt;
+        return
+            ExpectedTokens({
+                expectedTotalTokensCollected: expectedTotalTokensCollected,
+                expectedTokensLocked: expectedTokensLocked,
+                expectedProtocolTokensBurnt: expectedProtocolTokensBurnt,
+                expectedIndexerTokensCollected: expectedIndexerTokensCollected
+            });
+    }
+
+    function _sharedAssert(
+        TestState memory _beforeCollect,
+        TestState memory _afterCollect,
+        ExpectedTokens memory _expectedTokens,
+        uint256 _tokensCollected
+    ) internal pure {
+        uint256 indexerTokensCollected = _afterCollect.indexerBalance - _beforeCollect.indexerBalance;
+        assertEq(_expectedTokens.expectedTotalTokensCollected, _tokensCollected, "Total tokens collected should match");
         assertEq(
-            afterCollect.escrowBalance,
-            beforeCollect.escrowBalance - tokensCollected,
-            "Escrow balance should be reduced by the amount collected"
+            _expectedTokens.expectedProtocolTokensBurnt,
+            _tokensCollected - indexerTokensCollected,
+            "Protocol tokens burnt should match"
         );
-        assertEq(tokensCollected, expectedTotalTokensCollected, "Total tokens collected should match");
-        assertEq(expectedProtocolTokensBurnt, protocolTokensBurnt, "Protocol tokens burnt should match");
-        assertEq(indexerTokensCollected, expectedIndexerTokensCollected, "Indexer tokens collected should match");
         assertEq(
-            afterCollect.indexerTokensLocked,
-            beforeCollect.indexerTokensLocked + expectedTokensLocked,
-            "Locked tokens should match"
+            _expectedTokens.expectedIndexerTokensCollected,
+            indexerTokensCollected,
+            "Indexer tokens collected should match"
+        );
+        assertEq(
+            _afterCollect.escrowBalance,
+            _beforeCollect.escrowBalance - _expectedTokens.expectedTotalTokensCollected,
+            "_Escrow balance should be reduced by the amount collected"
         );
-    }
 
-    /* solhint-enable graph/func-name-mixedcase */
+        assertEq(
+            _afterCollect.indexerTokensLocked,
+            _beforeCollect.indexerTokensLocked + _expectedTokens.expectedTokensLocked,
+            "_Locked tokens should match"
+        );
+    }
 
     function _addTokensToProvision(IndexerState memory _indexerState, uint256 _tokensToAddToProvision) private {
         deal({ token: address(token), to: _indexerState.addr, give: _tokensToAddToProvision });