Explore looking beyond mempool items that hit the maximum cost limit (#17346)

Explore looking beyond mempool items that hit the maximum cost/fee limits.

chia/full_node/mempool.py

@@ -25,6 +25,20 @@ from chia.util.misc import to_batches
 
 log = logging.getLogger(__name__)
 
+# Maximum number of mempool items that can be skipped (not considered) during
+# the creation of a block bundle. An item is skipped if it won't fit in the
+# block we're trying to create.
+MAX_SKIPPED_ITEMS = 20
+
+# Threshold after which we stop including mempool items with eligible spends
+# during the creation of a block bundle. We do that to avoid spending too much
+# time on potentially expensive items.
+PRIORITY_TX_THRESHOLD = 3
+
+# Typical cost of a standard XCH spend. It's used as a heuristic to help
+# determine how close to the block size limit we're willing to go.
+MIN_COST_THRESHOLD = 6_000_000
+
 # We impose a limit on the fee a single transaction can pay in order to have the
 # sum of all fees in the mempool be less than 2^63. That's the limit of sqlite's
 # integers, which we rely on for computing fee per cost as well as the fee sum
@@ -395,6 +409,7 @@ class Mempool:
         log.info(f"Starting to make block, max cost: {self.mempool_info.max_block_clvm_cost}")
         with self._db_conn:
             cursor = self._db_conn.execute("SELECT name, fee FROM tx ORDER BY fee_per_cost DESC, seq ASC")
+        skipped_items = 0
         for row in cursor:
             name = bytes32(row[0])
             fee = int(row[1])
@@ -403,22 +418,58 @@ class Mempool:
                 continue
             try:
                 cost = uint64(0 if item.npc_result.conds is None else item.npc_result.conds.cost)
+                if skipped_items >= PRIORITY_TX_THRESHOLD:
+                    # If we've encountered `PRIORITY_TX_THRESHOLD` number of
+                    # transactions that don't fit in the remaining block size,
+                    # we want to keep looking for smaller transactions that
+                    # might fit, but we also want to avoid spending too much
+                    # time on potentially expensive ones, hence this shortcut.
+                    unique_coin_spends = []
+                    unique_additions = []
+                    for spend_data in item.bundle_coin_spends.values():
+                        if spend_data.eligible_for_dedup:
+                            raise Exception(f"Skipping transaction with eligible coin(s): {name.hex()}")
+                        unique_coin_spends.append(spend_data.coin_spend)
+                        unique_additions.extend(spend_data.additions)
+                    cost_saving = 0
+                else:
                     unique_coin_spends, cost_saving, unique_additions = eligible_coin_spends.get_deduplication_info(
                         bundle_coin_spends=item.bundle_coin_spends, max_cost=cost
                     )
                 item_cost = cost - cost_saving
-                log.info("Cumulative cost: %d, fee per cost: %0.4f", cost_sum, fee / item_cost)
+                log.info(
-                if (
+                    "Cumulative cost: %d, fee per cost: %0.4f, item cost: %d", cost_sum, fee / item_cost, item_cost
-                    item_cost + cost_sum > self.mempool_info.max_block_clvm_cost
+                )
-                    or fee + fee_sum > DEFAULT_CONSTANTS.MAX_COIN_AMOUNT
+                new_fee_sum = fee_sum + fee
-                ):
+                if new_fee_sum > DEFAULT_CONSTANTS.MAX_COIN_AMOUNT:
+                    # Such a fee is very unlikely to happen but we're defensively
+                    # accounting for it
+                    break  # pragma: no cover
+                new_cost_sum = cost_sum + item_cost
+                if new_cost_sum > self.mempool_info.max_block_clvm_cost:
+                    # Let's skip this item
+                    log.info(
+                        "Skipping mempool item. Cumulative cost %d exceeds maximum block cost %d",
+                        new_cost_sum,
+                        self.mempool_info.max_block_clvm_cost,
+                    )
+                    skipped_items += 1
+                    if skipped_items < MAX_SKIPPED_ITEMS:
+                        continue
+                    # Let's stop taking more items if we skipped `MAX_SKIPPED_ITEMS`
                     break
                 coin_spends.extend(unique_coin_spends)
                 additions.extend(unique_additions)
                 sigs.append(item.spend_bundle.aggregated_signature)
-                cost_sum += item_cost
+                cost_sum = new_cost_sum
-                fee_sum += fee
+                fee_sum = new_fee_sum
                 processed_spend_bundles += 1
+                # Let's stop taking more items if we don't have enough cost left
+                # for at least `MIN_COST_THRESHOLD` because that would mean we're
+                # getting very close to the limit anyway and *probably* won't
+                # find transactions small enough to fit at this point
+                if self.mempool_info.max_block_clvm_cost - cost_sum < MIN_COST_THRESHOLD:
+                    break
             except Exception as e:
                 log.debug(f"Exception while checking a mempool item for deduplication: {e}")
                 continue

tests/core/mempool/test_mempool_manager.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
-from dataclasses import dataclass
+import dataclasses
+import logging
 from typing import Any, Awaitable, Callable, Collection, Dict, List, Optional, Set, Tuple
 
 import pytest
@@ -11,6 +12,7 @@ from chia.consensus.constants import ConsensusConstants
 from chia.consensus.cost_calculator import NPCResult
 from chia.consensus.default_constants import DEFAULT_CONSTANTS
 from chia.full_node.bundle_tools import simple_solution_generator
+from chia.full_node.mempool import MAX_SKIPPED_ITEMS, PRIORITY_TX_THRESHOLD
 from chia.full_node.mempool_check_conditions import get_name_puzzle_conditions, mempool_check_time_locks
 from chia.full_node.mempool_manager import (
     MEMPOOL_MIN_FEE_INCREASE,
@@ -68,7 +70,7 @@ TEST_COIN_RECORD3 = CoinRecord(TEST_COIN3, uint32(0), uint32(0), False, TEST_TIM
 TEST_HEIGHT = uint32(5)
 
 
-@dataclass(frozen=True)
+@dataclasses.dataclass(frozen=True)
 class TestBlockRecord:
     """
     This is a subset of BlockRecord that the mempool manager uses for peak.
@@ -133,7 +135,9 @@ async def instantiate_mempool_manager(
     return mempool_manager
 
 
-async def setup_mempool_with_coins(*, coin_amounts: List[int]) -> Tuple[MempoolManager, List[Coin]]:
+async def setup_mempool_with_coins(
+    *, coin_amounts: List[int], max_block_clvm_cost: Optional[int] = None
+) -> Tuple[MempoolManager, List[Coin]]:
     coins = []
     test_coin_records = {}
     for amount in coin_amounts:
@@ -149,7 +153,11 @@ async def setup_mempool_with_coins(*, coin_amounts: List[int]) -> Tuple[MempoolM
                 ret.append(r)
         return ret
 
-    mempool_manager = await instantiate_mempool_manager(get_coin_records)
+    if max_block_clvm_cost is not None:
+        constants = dataclasses.replace(DEFAULT_CONSTANTS, MAX_BLOCK_COST_CLVM=max_block_clvm_cost)
+    else:
+        constants = DEFAULT_CONSTANTS
+    mempool_manager = await instantiate_mempool_manager(get_coin_records, constants=constants)
     return (mempool_manager, coins)
 
 
@@ -976,37 +984,82 @@ async def test_create_bundle_from_mempool(reverse_tx_order: bool) -> None:
     assert len([s for s in low_rate_spends if s in result[0].coin_spends]) == 0
 
 
+@pytest.mark.parametrize("num_skipped_items", [PRIORITY_TX_THRESHOLD, MAX_SKIPPED_ITEMS])
 @pytest.mark.anyio
-async def test_create_bundle_from_mempool_on_max_cost() -> None:
+async def test_create_bundle_from_mempool_on_max_cost(num_skipped_items: int, caplog: pytest.LogCaptureFixture) -> None:
     # This test exercises the path where an item's inclusion would exceed the
     # maximum cumulative cost, so it gets skipped as a result
+
+    # NOTE:
+    # 1. After PRIORITY_TX_THRESHOLD, we skip items with eligible coins.
+    # 2. After skipping MAX_SKIPPED_ITEMS, we stop processing further items.
+
     async def make_and_send_big_cost_sb(coin: Coin) -> None:
         conditions = []
         g1 = G1Element()
-        for _ in range(2436):
+        for _ in range(120):
             conditions.append([ConditionOpcode.AGG_SIG_UNSAFE, g1, IDENTITY_PUZZLE_HASH])
-        conditions.append([ConditionOpcode.CREATE_COIN, IDENTITY_PUZZLE_HASH, coin.amount - 1])
+        conditions.append([ConditionOpcode.CREATE_COIN, IDENTITY_PUZZLE_HASH, coin.amount - 10_000_000])
         # Create a spend bundle with a big enough cost that gets it close to the limit
         _, _, res = await generate_and_add_spendbundle(mempool_manager, conditions, coin)
         assert res[1] == MempoolInclusionStatus.SUCCESS
 
-    mempool_manager, coins = await setup_mempool_with_coins(coin_amounts=[1000000000, 1000000001])
+    mempool_manager, coins = await setup_mempool_with_coins(
-    # Create a spend bundle with a big enough cost that gets it close to the limit
+        coin_amounts=list(range(1_000_000_000, 1_000_000_030)), max_block_clvm_cost=550_000_000
-    await make_and_send_big_cost_sb(coins[0])
+    )
-    # Create a second spend bundle with a relatively smaller cost.
+    # Create the spend bundles with a big enough cost that they get close to the limit
-    # Combined with the first spend bundle, we'd exceed the maximum block clvm cost
+    for i in range(num_skipped_items):
-    conditions = [[ConditionOpcode.CREATE_COIN, IDENTITY_PUZZLE_HASH, coins[1].amount - 2]]
+        await make_and_send_big_cost_sb(coins[i])
-    sb2, _, res = await generate_and_add_spendbundle(mempool_manager, conditions, coins[1])
+
+    # Create a spend bundle with a relatively smaller cost.
+    # Combined with a big cost spend bundle, we'd exceed the maximum block clvm cost
+    sb2_coin = coins[num_skipped_items]
+    conditions = [[ConditionOpcode.CREATE_COIN, IDENTITY_PUZZLE_HASH, sb2_coin.amount - 200_000]]
+    sb2, _, res = await generate_and_add_spendbundle(mempool_manager, conditions, sb2_coin)
     assert res[1] == MempoolInclusionStatus.SUCCESS
+    sb2_addition = Coin(sb2_coin.name(), IDENTITY_PUZZLE_HASH, sb2_coin.amount - 200_000)
+    # Create 4 extra spend bundles with smaller FPC and smaller costs
+    extra_sbs = []
+    extra_additions = []
+    for i in range(num_skipped_items + 1, num_skipped_items + 5):
+        conditions = [[ConditionOpcode.CREATE_COIN, IDENTITY_PUZZLE_HASH, coins[i].amount - 30_000]]
+        # Make the first of these without eligible coins
+        if i == num_skipped_items + 1:
+            conditions.append([ConditionOpcode.AGG_SIG_UNSAFE, G1Element(), IDENTITY_PUZZLE_HASH])
+        sb, _, res = await generate_and_add_spendbundle(mempool_manager, conditions, coins[i])
+        extra_sbs.append(sb)
+        coin = Coin(coins[i].name(), IDENTITY_PUZZLE_HASH, coins[i].amount - 30_000)
+        extra_additions.append(coin)
+        assert res[1] == MempoolInclusionStatus.SUCCESS
+
     assert mempool_manager.peak is not None
+    caplog.set_level(logging.DEBUG)
     result = mempool_manager.create_bundle_from_mempool(mempool_manager.peak.header_hash)
     assert result is not None
     agg, additions = result
-    # The second spend bundle has a higher FPC so it should get picked first
+    skipped_due_to_eligible_coins = sum(
-    assert agg == sb2
+        1
-    # The first spend bundle hits the maximum block clvm cost and gets skipped
+        for line in caplog.text.split("\n")
-    assert additions == [Coin(coins[1].name(), IDENTITY_PUZZLE_HASH, coins[1].amount - 2)]
+        if "DEBUG Exception while checking a mempool item for deduplication: Skipping transaction with eligible coin(s)"
-    assert agg.removals() == [coins[1]]
+        in line
+    )
+    if num_skipped_items == PRIORITY_TX_THRESHOLD:
+        # We skipped enough big cost items to reach `PRIORITY_TX_THRESHOLD`,
+        # so the first from the extra 4 (the one without eligible coins) went in,
+        # and the other 3 were skipped (they have eligible coins)
+        assert skipped_due_to_eligible_coins == 3
+        assert agg == SpendBundle.aggregate([sb2, extra_sbs[0]])
+        assert additions == [sb2_addition, extra_additions[0]]
+        assert agg.removals() == [sb2_coin, coins[num_skipped_items + 1]]
+    elif num_skipped_items == MAX_SKIPPED_ITEMS:
+        # We skipped enough big cost items to trigger `MAX_SKIPPED_ITEMS` so
+        # we didn't process any of the extra items
+        assert skipped_due_to_eligible_coins == 0
+        assert agg == SpendBundle.aggregate([sb2])
+        assert additions == [sb2_addition]
+        assert agg.removals() == [sb2_coin]
+    else:
+        raise ValueError("num_skipped_items must be PRIORITY_TX_THRESHOLD or MAX_SKIPPED_ITEMS")  # pragma: no cover
 
 
 @pytest.mark.parametrize(