feat(pulse): keeper payment (#2617)

* feat: keeper payment

* fix: let gasbenchmark test contract accept payment

* fix: use checks-effects-interactions for the pyth and keeper fees

target_chains/ethereum/contracts/contracts/pulse/Scheduler.sol

@@ -250,6 +250,8 @@ abstract contract Scheduler is IScheduler, SchedulerState {
         bytes[] calldata updateData,
         bytes32[] calldata priceIds
     ) external override {
+        uint256 startGas = gasleft();
+
         SubscriptionStatus storage status = _state.subscriptionStatuses[
             subscriptionId
         ];
@@ -261,9 +263,12 @@ abstract contract Scheduler is IScheduler, SchedulerState {
             revert InactiveSubscription();
         }
 
-        // Verify price IDs match subscription
+        // Verify price IDs match subscription length
         if (priceIds.length != params.priceIds.length) {
-            revert InvalidPriceIdsLength(priceIds[0], params.priceIds[0]);
+            revert InvalidPriceIdsLength(
+                priceIds.length,
+                params.priceIds.length
+            );
         }
 
         // Keepers must provide priceIds in the exact same order as defined in the subscription
@@ -277,7 +282,7 @@ abstract contract Scheduler is IScheduler, SchedulerState {
         IPyth pyth = IPyth(_state.pyth);
         uint256 pythFee = pyth.getUpdateFee(updateData);
 
-        // Check if subscription has enough balance
+        // If we don't have enough balance, revert
         if (status.balanceInWei < pythFee) {
             revert InsufficientBalance();
         }
@@ -285,19 +290,19 @@ abstract contract Scheduler is IScheduler, SchedulerState {
         // Parse the price feed updates with an acceptable timestamp range of [-1h, +10s] from now.
         // We will validate the trigger conditions ourselves.
         uint64 curTime = SafeCast.toUint64(block.timestamp);
-        uint64 maxPublishTime = curTime + FUTURE_TIMESTAMP_MAX_VALIDITY_PERIOD;
-        uint64 minPublishTime = curTime > PAST_TIMESTAMP_MAX_VALIDITY_PERIOD
-            ? curTime - PAST_TIMESTAMP_MAX_VALIDITY_PERIOD
-            : 0;
         (
             PythStructs.PriceFeed[] memory priceFeeds,
             uint64[] memory slots
         ) = pyth.parsePriceFeedUpdatesWithSlots{value: pythFee}(
                 updateData,
                 priceIds,
-                minPublishTime,
-                maxPublishTime
+                curTime > PAST_TIMESTAMP_MAX_VALIDITY_PERIOD
+                    ? curTime - PAST_TIMESTAMP_MAX_VALIDITY_PERIOD
+                    : 0,
+                curTime + FUTURE_TIMESTAMP_MAX_VALIDITY_PERIOD
             );
+        status.balanceInWei -= pythFee;
+        status.totalSpent += pythFee;
 
         // Verify all price feeds have the same Pythnet slot.
         // All feeds in a subscription must be updated at the same time.
@@ -312,36 +317,21 @@ abstract contract Scheduler is IScheduler, SchedulerState {
         // is more recent than latest stored update's. Reverts if not.
         _validateShouldUpdatePrices(subscriptionId, params, status, priceFeeds);
 
-        // Store the price updates, update status, and emit event
-        _storePriceUpdatesAndStatus(
-            subscriptionId,
-            status,
-            priceFeeds,
-            pythFee
-        );
-    }
-
-    /**
-     * @notice Stores the price updates, updates subscription status, and emits event.
-     */
-    function _storePriceUpdatesAndStatus(
-        uint256 subscriptionId,
-        SubscriptionStatus storage status,
-        PythStructs.PriceFeed[] memory priceFeeds,
-        uint256 pythFee
-    ) internal {
-        // Store the price updates
+        // Update status and store the updates
+        uint256 latestPublishTime = 0; // Use the most recent publish time from the validated feeds
         for (uint8 i = 0; i < priceFeeds.length; i++) {
-            _state.priceUpdates[subscriptionId][priceFeeds[i].id] = priceFeeds[
-                i
-            ];
+            if (priceFeeds[i].price.publishTime > latestPublishTime) {
+                latestPublishTime = priceFeeds[i].price.publishTime;
+            }
         }
-        status.priceLastUpdatedAt = priceFeeds[0].price.publishTime;
-        status.balanceInWei -= pythFee;
-        status.totalUpdates += 1;
-        status.totalSpent += pythFee;
+        status.priceLastUpdatedAt = latestPublishTime;
+        status.totalUpdates += priceFeeds.length;
+
+        _storePriceUpdates(subscriptionId, priceFeeds);
 
-        emit PricesUpdated(subscriptionId, priceFeeds[0].price.publishTime);
+        _processFeesAndPayKeeper(status, startGas, priceIds.length);
+
+        emit PricesUpdated(subscriptionId, latestPublishTime);
     }
 
     /**
@@ -737,4 +727,53 @@ abstract contract Scheduler is IScheduler, SchedulerState {
             _state.activeSubscriptionIndex[subscriptionId] = 0;
         }
     }
+
+    /**
+     * @notice Internal function to store the parsed price feeds.
+     * @param subscriptionId The ID of the subscription.
+     * @param priceFeeds The array of price feeds to store.
+     */
+    function _storePriceUpdates(
+        uint256 subscriptionId,
+        PythStructs.PriceFeed[] memory priceFeeds
+    ) internal {
+        for (uint8 i = 0; i < priceFeeds.length; i++) {
+            _state.priceUpdates[subscriptionId][priceFeeds[i].id] = priceFeeds[
+                i
+            ];
+        }
+    }
+
+    /**
+     * @notice Internal function to calculate total fees, deduct from balance, and pay the keeper.
+     * @dev This function sends funds to `msg.sender`, so be sure that this is being called by a keeper.
+     * @dev Note that the Pyth fee is already paid in the parsePriceFeedUpdatesWithSlots call.
+     * @param status Storage reference to the subscription's status.
+     * @param startGas Gas remaining at the start of the parent function call.
+     * @param numPriceIds Number of price IDs being updated.
+     */
+    function _processFeesAndPayKeeper(
+        SubscriptionStatus storage status,
+        uint256 startGas,
+        uint256 numPriceIds
+    ) internal {
+        // Calculate fee components
+        uint256 gasCost = (startGas - gasleft() + GAS_OVERHEAD) * tx.gasprice;
+        uint256 keeperSpecificFee = uint256(_state.singleUpdateKeeperFeeInWei) *
+            numPriceIds;
+        uint256 totalKeeperFee = gasCost + keeperSpecificFee;
+
+        // Check balance
+        if (status.balanceInWei < totalKeeperFee) {
+            revert InsufficientBalance();
+        }
+
+        // Pay keeper and update status if successful
+        (bool sent, ) = msg.sender.call{value: totalKeeperFee}("");
+        if (!sent) {
+            revert KeeperPaymentFailed();
+        }
+        status.balanceInWei -= totalKeeperFee;
+        status.totalSpent += totalKeeperFee;
+    }
 }

target_chains/ethereum/contracts/contracts/pulse/SchedulerErrors.sol

@@ -12,7 +12,7 @@ error CannotUpdatePermanentSubscription();
 
 // Price feed errors
 error InvalidPriceId(bytes32 providedPriceId, bytes32 expectedPriceId);
-error InvalidPriceIdsLength(bytes32 providedLength, bytes32 expectedLength);
+error InvalidPriceIdsLength(uint256 providedLength, uint256 expectedLength);
 error EmptyPriceIds();
 error TooManyPriceIds(uint256 provided, uint256 maximum);
 error DuplicatePriceId(bytes32 priceId);
@@ -29,3 +29,6 @@ error TimestampOlderThanLastUpdate(
 // Whitelist errors
 error TooManyWhitelistedReaders(uint256 provided, uint256 maximum);
 error DuplicateWhitelistAddress(address addr);
+
+// Payment errors
+error KeeperPaymentFailed();

target_chains/ethereum/contracts/contracts/pulse/SchedulerState.sol

@@ -17,6 +17,9 @@ contract SchedulerState {
     /// Maximum time in the future (relative to current block timestamp)
     /// for which a price update timestamp is considered valid
     uint64 public constant FUTURE_TIMESTAMP_MAX_VALIDITY_PERIOD = 10 seconds;
+    /// Fixed gas overhead component used in keeper fee calculation.
+    /// This is a rough estimate of the tx overhead for a keeper to call updatePriceFeeds.
+    uint256 public constant GAS_OVERHEAD = 30000;
 
     struct State {
         /// Monotonically increasing counter for subscription IDs

target_chains/ethereum/contracts/forge-test/PulseScheduler.t.sol

@@ -11,6 +11,8 @@ import "../contracts/pulse/SchedulerState.sol";
 import "../contracts/pulse/SchedulerEvents.sol";
 import "../contracts/pulse/SchedulerErrors.sol";
 import "./utils/PulseSchedulerTestUtils.t.sol";
+import "@pythnetwork/pyth-sdk-solidity/IPyth.sol";
+
 contract MockReader {
     address private _scheduler;
 
@@ -76,7 +78,7 @@ contract SchedulerTest is Test, SchedulerEvents, PulseSchedulerTestUtils {
         pusher = address(4);
 
         uint128 minBalancePerFeed = 10 ** 16; // 0.01 ether
-        uint128 keeperFee = 10 ** 15; // 0.001 ether
+        uint128 keeperFee = 10 ** 14; // 0.0001 ether
 
         SchedulerUpgradeable _scheduler = new SchedulerUpgradeable();
         proxy = new ERC1967Proxy(address(_scheduler), "");
@@ -800,7 +802,7 @@ contract SchedulerTest is Test, SchedulerEvents, PulseSchedulerTestUtils {
         );
     }
 
-    function testUpdatePriceFeedsWorks() public {
+    function testUpdatePriceFeedsUpdatesPricesCorrectly() public {
         // --- First Update ---
         // Add a subscription and funds
         uint256 subscriptionId = addTestSubscription(
@@ -828,7 +830,6 @@ contract SchedulerTest is Test, SchedulerEvents, PulseSchedulerTestUtils {
         emit PricesUpdated(subscriptionId, publishTime1);
         vm.prank(pusher);
 
-        vm.breakpoint("a");
         scheduler.updatePriceFeeds(subscriptionId, updateData1, priceIds);
 
         // Verify first update
@@ -841,8 +842,8 @@ contract SchedulerTest is Test, SchedulerEvents, PulseSchedulerTestUtils {
         );
         assertEq(
             status1.totalUpdates,
-            1,
-            "Total updates should be 1 after first update"
+            priceIds.length,
+            "Total updates should be equal to the number of price feeds"
         );
         assertTrue(
             status1.totalSpent > 0,
@@ -880,7 +881,6 @@ contract SchedulerTest is Test, SchedulerEvents, PulseSchedulerTestUtils {
         emit PricesUpdated(subscriptionId, publishTime2);
         vm.prank(pusher);
 
-        vm.breakpoint("b");
         scheduler.updatePriceFeeds(subscriptionId, updateData2, priceIds);
 
         // Verify second update
@@ -893,8 +893,8 @@ contract SchedulerTest is Test, SchedulerEvents, PulseSchedulerTestUtils {
         );
         assertEq(
             status2.totalUpdates,
-            2,
-            "Total updates should be 2 after second update"
+            priceIds.length * 2,
+            "Total updates should be equal to the number of price feeds * 2 (first + second update)"
         );
         assertTrue(
             status2.totalSpent > spentAfterFirst,
@@ -911,6 +911,148 @@ contract SchedulerTest is Test, SchedulerEvents, PulseSchedulerTestUtils {
         );
     }
 
+    function testUpdatePriceFeedsPaysKeeperCorrectly() public {
+        // Set gas price
+        uint256 gasPrice = 0.1 gwei;
+        vm.txGasPrice(gasPrice);
+
+        // Add subscription and funds
+        uint256 subscriptionId = addTestSubscription(
+            scheduler,
+            address(reader)
+        );
+
+        // Prepare update data
+        (SchedulerState.SubscriptionParams memory params, ) = scheduler
+            .getSubscription(subscriptionId);
+        (
+            PythStructs.PriceFeed[] memory priceFeeds,
+            uint64[] memory slots
+        ) = createMockPriceFeedsWithSlots(
+                SafeCast.toUint64(block.timestamp),
+                params.priceIds.length
+            );
+
+        uint256 mockPythFee = MOCK_PYTH_FEE_PER_FEED * params.priceIds.length;
+        mockParsePriceFeedUpdatesWithSlots(pyth, priceFeeds, slots);
+        bytes[] memory updateData = createMockUpdateData(priceFeeds);
+
+        // Get state before
+        uint256 pusherBalanceBefore = pusher.balance;
+        (, SchedulerState.SubscriptionStatus memory statusBefore) = scheduler
+            .getSubscription(subscriptionId);
+        console.log(
+            "Subscription balance before update:",
+            vm.toString(statusBefore.balanceInWei)
+        );
+
+        // Perform update
+        vm.prank(pusher);
+        scheduler.updatePriceFeeds(subscriptionId, updateData, params.priceIds);
+
+        // Get state after
+        (, SchedulerState.SubscriptionStatus memory statusAfter) = scheduler
+            .getSubscription(subscriptionId);
+
+        // Calculate total fee deducted from subscription
+        uint256 totalFeeDeducted = statusBefore.balanceInWei -
+            statusAfter.balanceInWei;
+
+        // Calculate minimum keeper fee (overhead + feed-specific fee)
+        // The real cost is more because of the gas used in the updatePriceFeeds function
+        uint256 minKeeperFee = (scheduler.GAS_OVERHEAD() * gasPrice) +
+            (uint256(scheduler.getSingleUpdateKeeperFeeInWei()) *
+                params.priceIds.length);
+
+        assertGt(
+            totalFeeDeducted,
+            minKeeperFee + mockPythFee,
+            "Total fee deducted should be greater than the sum of keeper fee and Pyth fee (since gas usage of updatePriceFeeds is not accounted for)"
+        );
+        assertEq(
+            statusAfter.totalSpent,
+            statusBefore.totalSpent + totalFeeDeducted,
+            "Total spent should increase by the total fee deducted"
+        );
+        assertEq(
+            pusher.balance,
+            pusherBalanceBefore + totalFeeDeducted - mockPythFee,
+            "Pusher balance should increase by the keeper fee"
+        );
+
+        // This assertion is self-evident based on the calculations above, but keeping it for clarity
+        assertEq(
+            statusAfter.balanceInWei,
+            statusBefore.balanceInWei - totalFeeDeducted,
+            "Subscription balance should decrease by the total fee deducted"
+        );
+    }
+
+    function testUpdatePriceFeedsRevertsInsufficientBalanceForKeeperFee()
+        public
+    {
+        // Set gas price
+        uint256 gasPrice = 0.5 gwei;
+        vm.txGasPrice(gasPrice);
+
+        // Mock the minimum balance for the subscription to be
+        // zero so that we can test the keeper fee
+        vm.mockCall(
+            address(scheduler),
+            abi.encodeWithSelector(Scheduler.getMinimumBalance.selector),
+            abi.encode(0)
+        );
+
+        // Add subscription
+        uint256 subscriptionId = addTestSubscription(
+            scheduler,
+            address(reader)
+        );
+        bytes32[] memory priceIds = createPriceIds();
+
+        // Prepare update data and get Pyth fee
+        uint64 publishTime = SafeCast.toUint64(block.timestamp);
+        PythStructs.PriceFeed[] memory priceFeeds;
+        uint64[] memory slots;
+        (priceFeeds, slots) = createMockPriceFeedsWithSlots(
+            publishTime,
+            priceIds.length
+        );
+        uint256 mockPythFee = MOCK_PYTH_FEE_PER_FEED * priceIds.length;
+        mockParsePriceFeedUpdatesWithSlots(pyth, priceFeeds, slots);
+        bytes[] memory updateData = createMockUpdateData(priceFeeds);
+
+        // Calculate minimum keeper fee (overhead + feed-specific fee)
+        // The real cost is more because of the gas used in the updatePriceFeeds function
+        uint256 minKeeperFee = (scheduler.GAS_OVERHEAD() * gasPrice) +
+            (uint256(scheduler.getSingleUpdateKeeperFeeInWei()) *
+                priceIds.length);
+
+        // Fund subscription without enough for Pyth fee + keeper fee
+        // It won't be enough because of the gas cost of updatePriceFeeds
+        uint256 fundAmount = mockPythFee + minKeeperFee;
+        scheduler.addFunds{value: fundAmount}(subscriptionId);
+
+        // Get and print the subscription balance before attempting the update
+        (, SchedulerState.SubscriptionStatus memory status) = scheduler
+            .getSubscription(subscriptionId);
+        console.log(
+            "Subscription balance before update:",
+            vm.toString(status.balanceInWei)
+        );
+        console.log("Required Pyth fee:", vm.toString(mockPythFee));
+        console.log("Minimum keeper fee:", vm.toString(minKeeperFee));
+        console.log(
+            "Total minimum required:",
+            vm.toString(mockPythFee + minKeeperFee)
+        );
+
+        // Expect revert due to insufficient balance for total fee
+        vm.expectRevert(abi.encodeWithSelector(InsufficientBalance.selector));
+        vm.prank(pusher);
+        scheduler.updatePriceFeeds(subscriptionId, updateData, priceIds);
+    }
+
     function testUpdatePriceFeedsRevertsOnHeartbeatUpdateConditionNotMet()
         public
     {

target_chains/ethereum/contracts/forge-test/PulseSchedulerGasBenchmark.t.sol

@@ -227,4 +227,7 @@ contract PulseSchedulerGasBenchmark is Test, PulseSchedulerTestUtils {
     function testGetActiveSubscriptions1000() public {
         _runGetActiveSubscriptionsBenchmark(1000);
     }
+
+    // Allow the contract to receive Ether (for keeper payments during tests)
+    receive() external payable {}
 }