chia-blockchain/chia/full_node/mempool_manager.py

from __future__ import annotations

import asyncio
import contextlib
import logging
import time
from collections.abc import AsyncIterator, Awaitable, Callable, Collection
from concurrent.futures import Executor, ThreadPoolExecutor
from dataclasses import dataclass, field
from types import TracebackType
from typing import TypeVar

from chia_rs import (
    ELIGIBLE_FOR_DEDUP,
    ELIGIBLE_FOR_FF,
    MEMPOOL_MODE,
    BLSCache,
    CoinRecord,
    ConsensusConstants,
    SpendBundle,
    SpendBundleConditions,
    check_time_locks,
    get_flags_for_height_and_constants,
    supports_fast_forward,
    validate_clvm_and_signature,
)
from chia_rs.sized_bytes import bytes32
from chia_rs.sized_ints import uint32, uint64
from chiabip158 import PyBIP158
from typing_extensions import Self

from chia.consensus.block_record import BlockRecordProtocol
from chia.full_node.bitcoin_fee_estimator import create_bitcoin_fee_estimator
from chia.full_node.fee_estimation import FeeBlockInfo, MempoolInfo, MempoolItemInfo
from chia.full_node.fee_estimator_interface import FeeEstimatorInterface
from chia.full_node.mempool import MEMPOOL_ITEM_FEE_LIMIT, Mempool, MempoolRemoveInfo, MempoolRemoveReason
from chia.full_node.pending_tx_cache import ConflictTxCache, PendingTxCache
from chia.types.blockchain_format.coin import Coin
from chia.types.clvm_cost import CLVMCost
from chia.types.fee_rate import FeeRate
from chia.types.generator_types import NewBlockGenerator
from chia.types.mempool_inclusion_status import MempoolInclusionStatus
from chia.types.mempool_item import BundleCoinSpend, MempoolItem, UnspentLineageInfo
from chia.util.db_wrapper import SQLITE_INT_MAX
from chia.util.errors import Err, ValidationError
from chia.util.inline_executor import InlineExecutor

log = logging.getLogger(__name__)

# mempool items replacing existing ones must increase the total fee at least by
# this amount. 0.00001 XCH
MEMPOOL_MIN_FEE_INCREASE = uint64(10000000)


@dataclass
class TimelockConditions:
    assert_height: uint32 = uint32(0)
    assert_seconds: uint64 = uint64(0)
    assert_before_height: uint32 | None = None
    assert_before_seconds: uint64 | None = None


@dataclass
class LineageInfoCache:
    _fun: Callable[[bytes32], Awaitable[UnspentLineageInfo | None]]
    _cache: dict[bytes32, UnspentLineageInfo | None] = field(default_factory=dict)

    async def get_unspent_lineage_info(self, puzzle_hash: bytes32) -> UnspentLineageInfo | None:
        # we rely on KeyError to distinguish between a stored
        # None value and a missing entry
        try:
            return self._cache[puzzle_hash]
        except KeyError:
            pass

        ret = await self._fun(puzzle_hash)
        self._cache[puzzle_hash] = ret
        return ret


def compute_assert_height(
    removal_coin_records: dict[bytes32, CoinRecord],
    conds: SpendBundleConditions,
) -> TimelockConditions:
    """
    Computes the most restrictive height- and seconds assertion in the spend bundle.
    Relative heights and times are resolved using the confirmed heights and
    timestamps from the coin records.
    """

    ret = TimelockConditions()
    ret.assert_height = uint32(conds.height_absolute)
    ret.assert_seconds = uint64(conds.seconds_absolute)
    ret.assert_before_height = (
        uint32(conds.before_height_absolute) if conds.before_height_absolute is not None else None
    )
    ret.assert_before_seconds = (
        uint64(conds.before_seconds_absolute) if conds.before_seconds_absolute is not None else None
    )

    for spend in conds.spends:
        if spend.height_relative is not None:
            h = uint32(removal_coin_records[bytes32(spend.coin_id)].confirmed_block_index + spend.height_relative)
            ret.assert_height = max(ret.assert_height, h)

        if spend.seconds_relative is not None:
            s = uint64(removal_coin_records[bytes32(spend.coin_id)].timestamp + spend.seconds_relative)
            ret.assert_seconds = max(ret.assert_seconds, s)

        if spend.before_height_relative is not None:
            h = uint32(
                removal_coin_records[bytes32(spend.coin_id)].confirmed_block_index + spend.before_height_relative
            )
            if ret.assert_before_height is not None:
                ret.assert_before_height = min(ret.assert_before_height, h)
            else:
                ret.assert_before_height = h

        if spend.before_seconds_relative is not None:
            s = uint64(removal_coin_records[bytes32(spend.coin_id)].timestamp + spend.before_seconds_relative)
            if ret.assert_before_seconds is not None:
                ret.assert_before_seconds = min(ret.assert_before_seconds, s)
            else:
                ret.assert_before_seconds = s

    return ret


@dataclass
class SpendBundleAddInfo:
    cost: uint64 | None
    status: MempoolInclusionStatus
    removals: list[MempoolRemoveInfo]
    error: Err | None


@dataclass
class NewPeakInfo:
    spend_bundle_ids: list[bytes32]
    removals: list[MempoolRemoveInfo]


# For block overhead cost calculation
QUOTE_BYTES = 2
QUOTE_EXECUTION_COST = 20


def is_atom_canonical(clvm_buffer: bytes, offset: int) -> tuple[int, bool]:
    b = clvm_buffer[offset]
    if (b & 0b11000000) == 0b10000000:
        # 6 bits length prefix
        mask = 0b00111111
        prefix_len = 0
        min_value = 1
    elif (b & 0b11100000) == 0b11000000:
        # 5 + 8 bits length prefix
        mask = 0b00011111
        prefix_len = 1
        min_value = 1 << 6
    elif (b & 0b11110000) == 0b11100000:
        # 4 + 8 + 8 bits length prefix
        mask = 0b00001111
        prefix_len = 2
        min_value = 1 << (5 + 8)
    elif (b & 0b11111000) == 0b11110000:
        # 3 + 8 + 8 + 8 bits length prefix
        mask = 0b00000111
        prefix_len = 3
        min_value = 1 << (4 + 8 + 8)
    elif (b & 0b11111100) == 0b11111000:
        # 2 + 8 + 8 + 8 + 8 bits length prefix
        mask = 0b00000011
        prefix_len = 4
        min_value = 1 << (3 + 8 + 8 + 8)
    elif (b & 0b11111110) == 0b11111100:
        # 1 + 8 + 8 + 8 + 8 + 8 bits length prefix
        mask = 0b00000001
        prefix_len = 5
        min_value = 1 << (2 + 8 + 8 + 8 + 8)

    atom_len = b & mask
    for i in range(prefix_len):
        atom_len <<= 8
        offset += 1
        atom_len |= clvm_buffer[offset]

    return 1 + prefix_len + atom_len, atom_len >= min_value


def is_clvm_canonical(clvm_buffer: bytes) -> bool:
    """
    checks whether the CLVM serialization is all canonical representation.
    atoms can be serialized in more than one way by using more bytes than
    necessary to encode the length prefix. This functions ensures that all atoms are
    encoded with the shortest representation. back-references are not allowed
    and will make this function return false
    """
    assert clvm_buffer != b""

    offset = 0
    tokens_left = 1
    while True:
        b = clvm_buffer[offset]

        # pair
        if b == 0xFF:
            tokens_left += 1
            offset += 1
            continue

        # back references cannot be considered canonical, since they may be
        # encoded in many different ways
        if b == 0xFE:
            return False

        # small atom or NIL
        if b <= 0x80:
            tokens_left -= 1
            offset += 1
        else:
            atom_len, canonical = is_atom_canonical(clvm_buffer, offset)
            if not canonical:
                return False
            tokens_left -= 1
            offset += atom_len

        if tokens_left == 0:
            break

    # if there's garbage at the end, it's not canonical
    return offset == len(clvm_buffer)


def check_removals(
    removals: dict[bytes32, CoinRecord],
    bundle_coin_spends: dict[bytes32, BundleCoinSpend],
    *,
    get_items_by_coin_ids: Callable[[list[bytes32]], list[MempoolItem]],
) -> tuple[Err | None, list[MempoolItem]]:
    """
    This function checks for double spends, unknown spends and conflicting transactions in mempool.
    Returns Error (if any), the set of existing MempoolItems with conflicting spends (if any).
    Note that additions are not checked for duplicates, because having duplicate additions requires also
    having duplicate removals.
    """
    conflicts = set()
    for coin_id, coin_bcs in bundle_coin_spends.items():
        # 1. Checks if it's been spent already
        if removals[coin_id].spent and not coin_bcs.supports_fast_forward:
            return Err.DOUBLE_SPEND, []

        # 2. Checks if there's a mempool conflict
        conflicting_items = get_items_by_coin_ids([coin_id])
        for item in conflicting_items:
            if item in conflicts:
                continue
            conflict_bcs = item.bundle_coin_spends.get(coin_id)
            if conflict_bcs is None:
                # Check if this is an item that spends an older ff singleton
                # version with a latest version that matches our coin ID.
                conflict_bcs = next(
                    (
                        bcs
                        for bcs in item.bundle_coin_spends.values()
                        if bcs.latest_singleton_lineage is not None and bcs.latest_singleton_lineage.coin_id == coin_id
                    ),
                    None,
                )
                # We're not expected to get here but let's handle it gracefully
                if conflict_bcs is None:
                    log.warning(f"Coin ID {coin_id} expected but not found in mempool item {item.name}")
                    return Err.INVALID_SPEND_BUNDLE, []
            # if the spend we're adding to the mempool is not DEDUP nor FF, it's
            # just a regular conflict
            if not coin_bcs.supports_fast_forward and not coin_bcs.eligible_for_dedup:
                conflicts.add(item)

            # if the spend we're adding is FF, but there's a conflicting spend
            # that isn't FF, they can't be chained, so that's a conflict
            elif coin_bcs.supports_fast_forward and not conflict_bcs.supports_fast_forward:
                conflicts.add(item)

            # if the spend we're adding is DEDUP, but there's a conflicting spend
            # that isn't DEDUP, we cannot merge them, so that's a conflict
            elif coin_bcs.eligible_for_dedup and not conflict_bcs.eligible_for_dedup:
                conflicts.add(item)

            # if the spend we're adding is DEDUP but the existing spend has a
            # different solution, we cannot merge them, so that's a conflict
            elif coin_bcs.eligible_for_dedup and bytes(coin_bcs.coin_spend.solution) != bytes(
                conflict_bcs.coin_spend.solution
            ):
                conflicts.add(item)

    if len(conflicts) > 0:
        return Err.MEMPOOL_CONFLICT, list(conflicts)
    return None, []


class MempoolManager:
    pool: Executor
    constants: ConsensusConstants
    seen_bundle_hashes: dict[bytes32, bytes32]
    get_coin_records: Callable[[Collection[bytes32]], Awaitable[list[CoinRecord]]]
    get_unspent_lineage_info_for_puzzle_hash: Callable[[bytes32], Awaitable[UnspentLineageInfo | None]]
    nonzero_fee_minimum_fpc: int
    mempool_max_total_cost: int
    # a cache of MempoolItems that conflict with existing items in the pool
    _conflict_cache: ConflictTxCache
    # cache of MempoolItems with height conditions making them not valid yet
    _pending_cache: PendingTxCache
    seen_cache_size: int
    # the peak is only ever set to a transaction block. We use it to validate
    # timelocks against, so it must have a timestamp.
    peak: BlockRecordProtocol | None
    mempool: Mempool
    _worker_queue_size: int
    max_block_clvm_cost: uint64
    max_tx_clvm_cost: uint64

    def __init__(
        self,
        get_coin_records: Callable[[Collection[bytes32]], Awaitable[list[CoinRecord]]],
        get_unspent_lineage_info_for_puzzle_hash: Callable[[bytes32], Awaitable[UnspentLineageInfo | None]],
        consensus_constants: ConsensusConstants,
        *,
        single_threaded: bool = False,
        max_tx_clvm_cost: uint64 | None = None,
    ):
        self.constants: ConsensusConstants = consensus_constants

        # Keep track of seen spend_bundles
        self.seen_bundle_hashes: dict[bytes32, bytes32] = {}

        self.get_coin_records = get_coin_records
        self.get_unspent_lineage_info_for_puzzle_hash = get_unspent_lineage_info_for_puzzle_hash

        # The fee per cost must be above this amount to consider the fee "nonzero", and thus able to kick out other
        # transactions. This prevents spam. This is equivalent to 0.055 XCH per block, or about 0.00005 XCH for two
        # spends.
        self.nonzero_fee_minimum_fpc = 5

        # We need to deduct the block overhead, which consists of the wrapping
        # quote opcode's bytes cost as well as its execution cost.
        BLOCK_OVERHEAD = QUOTE_BYTES * self.constants.COST_PER_BYTE + QUOTE_EXECUTION_COST

        self.max_block_clvm_cost = uint64(self.constants.MAX_BLOCK_COST_CLVM - BLOCK_OVERHEAD)
        self.max_tx_clvm_cost = (
            max_tx_clvm_cost if max_tx_clvm_cost is not None else uint64(self.constants.MAX_BLOCK_COST_CLVM // 2)
        )
        self.mempool_max_total_cost = int(self.constants.MAX_BLOCK_COST_CLVM * self.constants.MEMPOOL_BLOCK_BUFFER)

        # Transactions that were unable to enter mempool, used for retry. (they were invalid)
        self._conflict_cache = ConflictTxCache(self.constants.MAX_BLOCK_COST_CLVM * 1, 1000)
        self._pending_cache = PendingTxCache(self.constants.MAX_BLOCK_COST_CLVM * 1, 1000)
        self.seen_cache_size = 10000
        self._worker_queue_size = 0
        if single_threaded:
            self.pool = InlineExecutor()
        else:
            self.pool = ThreadPoolExecutor(max_workers=2, thread_name_prefix="mempool-")

        # The mempool will correspond to a certain peak
        self.peak: BlockRecordProtocol | None = None
        self.fee_estimator: FeeEstimatorInterface = create_bitcoin_fee_estimator(self.max_block_clvm_cost)
        mempool_info = MempoolInfo(
            CLVMCost(uint64(self.mempool_max_total_cost)),
            FeeRate(uint64(self.nonzero_fee_minimum_fpc)),
            CLVMCost(uint64(self.max_block_clvm_cost)),
        )
        self.mempool: Mempool = Mempool(mempool_info, self.fee_estimator)

    @classmethod
    @contextlib.asynccontextmanager
    async def managed(
        cls,
        get_coin_records: Callable[[Collection[bytes32]], Awaitable[list[CoinRecord]]],
        get_unspent_lineage_info_for_puzzle_hash: Callable[[bytes32], Awaitable[UnspentLineageInfo | None]],
        consensus_constants: ConsensusConstants,
        single_threaded: bool = False,
        max_tx_clvm_cost: uint64 | None = None,
    ) -> AsyncIterator[Self]:
        self = cls(
            get_coin_records,
            get_unspent_lineage_info_for_puzzle_hash,
            consensus_constants,
            single_threaded=single_threaded,
            max_tx_clvm_cost=max_tx_clvm_cost,
        )
        try:
            yield self
        finally:
            self.shut_down()

    def shut_down(self) -> None:
        self.pool.shutdown(wait=True)
        self.mempool.close()

    def __enter__(self) -> Self:
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        traceback: TracebackType | None,
    ) -> None:
        self.shut_down()

    # TODO: remove this, use create_generator() instead
    def create_bundle_from_mempool(self, last_tb_header_hash: bytes32) -> tuple[SpendBundle, list[Coin]] | None:
        """
        Returns aggregated spendbundle that can be used for creating new block,
        additions and removals in that spend_bundle
        """
        if self.peak is None or self.peak.header_hash != last_tb_header_hash:
            return None
        return self.mempool.create_bundle_from_mempool_items(self.constants, self.peak.height)

    def create_block_generator(self, last_tb_header_hash: bytes32, timeout: float) -> NewBlockGenerator | None:
        """
        Returns a block generator program, the aggregate signature and all additions and removals, for a new block
        """
        if self.peak is None or self.peak.header_hash != last_tb_header_hash:
            return None
        return self.mempool.create_block_generator(self.constants, self.peak.height, timeout)

    def create_block_generator2(self, last_tb_header_hash: bytes32, timeout: float) -> NewBlockGenerator | None:
        """
        Returns a block generator program, the aggregate signature and all additions, for a new block
        """
        if self.peak is None or self.peak.header_hash != last_tb_header_hash:
            return None
        return self.mempool.create_block_generator2(self.constants, self.peak.height, timeout)

    def get_filter(self) -> bytes:
        all_transactions: set[bytes32] = set()
        byte_array_list = []
        for key in self.mempool.all_item_ids():
            if key not in all_transactions:
                all_transactions.add(key)
                byte_array_list.append(bytearray(key))

        tx_filter: PyBIP158 = PyBIP158(byte_array_list)
        return bytes(tx_filter.GetEncoded())

    def is_fee_enough(self, fees: uint64, cost: uint64) -> bool:
        """
        Determines whether any of the pools can accept a transaction with a given fees
        and cost.
        """
        if cost == 0:
            return False
        fees_per_cost = fees / cost
        if not self.mempool.at_full_capacity(cost):
            return True
        if fees_per_cost < self.nonzero_fee_minimum_fpc:
            return False
        min_fee_rate = self.mempool.get_min_fee_rate(cost)
        return min_fee_rate is not None and fees_per_cost > min_fee_rate

    def add_and_maybe_pop_seen(self, spend_name: bytes32) -> None:
        self.seen_bundle_hashes[spend_name] = spend_name
        while len(self.seen_bundle_hashes) > self.seen_cache_size:
            first_in = next(iter(self.seen_bundle_hashes.keys()))
            self.seen_bundle_hashes.pop(first_in)

    def seen(self, bundle_hash: bytes32) -> bool:
        """Return true if we saw this spendbundle recently"""
        return bundle_hash in self.seen_bundle_hashes

    def remove_seen(self, bundle_hash: bytes32) -> None:
        if bundle_hash in self.seen_bundle_hashes:
            self.seen_bundle_hashes.pop(bundle_hash)

    async def pre_validate_spendbundle(
        self, spend_bundle: SpendBundle, spend_bundle_id: bytes32 | None = None, bls_cache: BLSCache | None = None
    ) -> SpendBundleConditions:
        """
        Errors are included within the cached_result.
        This runs in another process so we don't block the main thread
        """

        if spend_bundle.coin_spends == []:
            raise ValidationError(Err.INVALID_SPEND_BUNDLE, "Empty SpendBundle")

        assert self.peak is not None

        self._worker_queue_size += 1
        try:
            flags = get_flags_for_height_and_constants(self.peak.height, self.constants)
            sbc, new_cache_entries, duration = await asyncio.get_running_loop().run_in_executor(
                self.pool,
                validate_clvm_and_signature,
                spend_bundle,
                self.max_tx_clvm_cost,
                self.constants,
                flags | MEMPOOL_MODE,
            )
        # validate_clvm_and_signature raises a ValueError with an error code
        except ValueError as e:
            # Convert that to a ValidationError
            if len(e.args) > 1:
                error = Err(e.args[1])
                raise ValidationError(error)
            else:
                raise ValidationError(Err.UNKNOWN)  # pragma: no cover
        finally:
            self._worker_queue_size -= 1

        if sbc.num_atoms > sbc.cost * 60_000_000 / self.constants.MAX_BLOCK_COST_CLVM:
            raise ValidationError(Err.INVALID_SPEND_BUNDLE, "too many atoms")

        if sbc.num_pairs > sbc.cost * 60_000_000 / self.constants.MAX_BLOCK_COST_CLVM:
            raise ValidationError(Err.INVALID_SPEND_BUNDLE, "too many pairs")

        if bls_cache is not None:
            bls_cache.update(new_cache_entries)

        if spend_bundle_id is None:
            spend_bundle_id = spend_bundle.name()

        log.log(
            logging.DEBUG if duration < 2 else logging.WARNING,
            f"pre_validate_spendbundle took {duration:0.4f} seconds "
            f"for {spend_bundle_id} (queue-size: {self._worker_queue_size})",
        )
        return sbc

    async def add_spend_bundle(
        self,
        new_spend: SpendBundle,
        conds: SpendBundleConditions,
        spend_name: bytes32,
        first_added_height: uint32,
        get_coin_records: Callable[[Collection[bytes32]], Awaitable[list[CoinRecord]]] | None = None,
        get_unspent_lineage_info_for_puzzle_hash: Callable[[bytes32], Awaitable[UnspentLineageInfo | None]]
        | None = None,
    ) -> SpendBundleAddInfo:
        """
        Validates and adds to mempool a new_spend with the given
        SpendBundleConditions, and spend_name, and the current mempool. The mempool
        should be locked during this call (blockchain lock). If there are mempool
        conflicts, the conflicting spends might be removed (if the new spend is
        a superset of the previous). Otherwise, the new spend might be
        added to the potential pool.

        Args:
            new_spend: spend bundle to validate and add
            conds: SpendBundleConditions resulting from running the clvm in the spend bundle's coin spends
            spend_name: hash of the spend bundle data, passed in as an optimization

        Returns:
            Optional[uint64]: cost of the entire transaction, None iff status is FAILED
            MempoolInclusionStatus:  SUCCESS (should add to pool), FAILED (cannot add), and PENDING (can add later)
            list[MempoolRemoveInfo]: conflicting mempool items which were removed, if no Err
            Optional[Err]: Err is set iff status is FAILED
        """

        # Skip if already added
        existing_item = self.mempool.get_item_by_id(spend_name)
        if existing_item is not None:
            return SpendBundleAddInfo(existing_item.cost, MempoolInclusionStatus.SUCCESS, [], None)

        if get_coin_records is None:
            get_coin_records = self.get_coin_records

        if get_unspent_lineage_info_for_puzzle_hash is None:
            get_unspent_lineage_info_for_puzzle_hash = self.get_unspent_lineage_info_for_puzzle_hash

        err, item, remove_items = await self.validate_spend_bundle(
            new_spend,
            conds,
            spend_name,
            first_added_height,
            get_coin_records,
            get_unspent_lineage_info_for_puzzle_hash,
        )
        if err is None:
            # No error, immediately add to mempool, after removing conflicting TXs.
            assert item is not None
            conflict = self.mempool.remove_from_pool(remove_items, MempoolRemoveReason.CONFLICT)
            info = self.mempool.add_to_pool(item)
            if info.error is not None:
                return SpendBundleAddInfo(item.cost, MempoolInclusionStatus.FAILED, [], info.error)
            return SpendBundleAddInfo(item.cost, MempoolInclusionStatus.SUCCESS, [*info.removals, conflict], None)
        elif err is Err.MEMPOOL_CONFLICT and item is not None:
            # The transaction has a conflict with another item in the
            # mempool, put it aside and re-try it later
            self._conflict_cache.add(item)
            return SpendBundleAddInfo(item.cost, MempoolInclusionStatus.PENDING, [], err)
        elif item is not None:
            # The transasction has a height assertion and is not yet valid.
            # remember it to try it again later
            self._pending_cache.add(item)
            return SpendBundleAddInfo(item.cost, MempoolInclusionStatus.PENDING, [], err)
        else:
            # Cannot add to the mempool or pending pool.
            return SpendBundleAddInfo(None, MempoolInclusionStatus.FAILED, [], err)

    async def validate_spend_bundle(
        self,
        new_spend: SpendBundle,
        conds: SpendBundleConditions,
        spend_name: bytes32,
        first_added_height: uint32,
        get_coin_records: Callable[[Collection[bytes32]], Awaitable[list[CoinRecord]]],
        get_unspent_lineage_info_for_puzzle_hash: Callable[[bytes32], Awaitable[UnspentLineageInfo | None]],
    ) -> tuple[Err | None, MempoolItem | None, list[bytes32]]:
        """
        Validates new_spend with the given SpendBundleConditions, and
        spend_name, and the current mempool. The mempool should
        be locked during this call (blockchain lock).

        Args:
            new_spend: spend bundle to validate
            conds: result of running the clvm transaction
            spend_name: hash of the spend bundle data, passed in as an optimization
            first_added_height: The block height that `new_spend`  first entered this node's mempool.
                Used to estimate how long a spend has taken to be included on the chain.
                This value could differ node to node. Not preserved across full_node restarts.

        Returns:
            Optional[Err]: Err is set if we cannot add to the mempool, None if we will immediately add to mempool
            Optional[MempoolItem]: the item to add (to mempool or pending pool)
            list[bytes32]: conflicting mempool items to remove, if no Err
        """
        start_time = time.monotonic()
        if self.peak is None:
            return Err.MEMPOOL_NOT_INITIALIZED, None, []

        cost = conds.cost

        removal_names: set[bytes32] = set()
        additions_dict: dict[bytes32, Coin] = {}
        addition_amount: int = 0

        # Map of coin ID to SpendConditions
        spend_conditions = {bytes32(spend.coin_id): spend for spend in conds.spends}

        # if this happens, the SpendBundle doesn't match the
        # SpendBundleConditions.
        assert len(new_spend.coin_spends) == len(spend_conditions)

        bundle_coin_spends: dict[bytes32, BundleCoinSpend] = {}
        for coin_spend in new_spend.coin_spends:
            coin_id = coin_spend.coin.name()
            removal_names.add(coin_id)

            # if this coin_id isn't found, the SpendBundle doesn't match the
            # SpendBundleConditions.
            spend_conds = spend_conditions.pop(coin_id)

            if bool(spend_conds.flags & ELIGIBLE_FOR_DEDUP) and not is_clvm_canonical(bytes(coin_spend.solution)):
                return Err.INVALID_COIN_SOLUTION, None, []

            lineage_info = None
            if bool(spend_conds.flags & ELIGIBLE_FOR_FF) and supports_fast_forward(coin_spend):
                # Make sure the fast forward spend still has a version that is
                # still unspent, because if the singleton has been spent in a
                # non-FF spend, this fast forward spend will never become valid.
                # So treat this as a normal spend, which requires the exact coin
                # to exist and be unspent.
                # Singletons that were created before the optimization of using
                # spent_index will also fail this test, and such spends will
                # fall back to be treated as non-FF spends.
                lineage_info = await get_unspent_lineage_info_for_puzzle_hash(spend_conds.puzzle_hash)

            spend_additions = []
            for puzzle_hash, amount, _ in spend_conds.create_coin:
                child_coin = Coin(coin_id, puzzle_hash, uint64(amount))
                spend_additions.append(child_coin)
                additions_dict[child_coin.name()] = child_coin
                addition_amount += amount

            bundle_coin_spends[coin_id] = BundleCoinSpend(
                coin_spend=coin_spend,
                eligible_for_dedup=bool(spend_conds.flags & ELIGIBLE_FOR_DEDUP),
                additions=spend_additions,
                cost=uint64(spend_conds.condition_cost + spend_conds.execution_cost),
                latest_singleton_lineage=lineage_info,
            )

        # fast forward spends are only allowed when bundled with other, non-FF, spends
        # in order to evict an FF spend, it must be associated with a normal
        # spend that can be included in a block or invalidated some other way
        if all([s.supports_fast_forward for s in bundle_coin_spends.values()]):
            return Err.INVALID_SPEND_BUNDLE, None, []

        removal_record_dict: dict[bytes32, CoinRecord] = {}
        removal_amount: int = 0
        removal_records = await get_coin_records(removal_names)
        for record in removal_records:
            removal_record_dict[record.coin.name()] = record

        for name in removal_names:
            if name not in removal_record_dict and name not in additions_dict:
                return Err.UNKNOWN_UNSPENT, None, []
            if name in additions_dict:
                removal_coin = additions_dict[name]
                # The timestamp and block-height of this coin being spent needs
                # to be consistent with what we use to check time-lock
                # conditions (below). All spends (including ephemeral coins) are
                # spent simultaneously. Ephemeral coins with an
                # ASSERT_SECONDS_RELATIVE 0 condition are still OK to spend in
                # the same block.
                assert self.peak.timestamp is not None
                removal_record = CoinRecord(
                    removal_coin,
                    uint32(self.peak.height + 1),
                    uint32(0),
                    False,
                    self.peak.timestamp,
                )
                removal_record_dict[name] = removal_record
            else:
                removal_record = removal_record_dict[name]
            removal_amount += removal_record.coin.amount

        fees = uint64(removal_amount - addition_amount)

        if cost == 0:
            return Err.UNKNOWN, None, []

        if cost > self.max_tx_clvm_cost:
            return Err.BLOCK_COST_EXCEEDS_MAX, None, []

        # this is not very likely to happen, but it's here to ensure SQLite
        # never runs out of precision in its computation of fees.
        # sqlite's integers are signed int64, so the max value they can
        # represent is 2^63-1
        if fees > MEMPOOL_ITEM_FEE_LIMIT or SQLITE_INT_MAX - self.mempool.total_mempool_fees() <= fees:
            return Err.INVALID_BLOCK_FEE_AMOUNT, None, []

        fees_per_cost: float = fees / cost
        # If pool is at capacity check the fee, if not then accept even without the fee
        if self.mempool.at_full_capacity(cost):
            if fees_per_cost < self.nonzero_fee_minimum_fpc:
                return Err.INVALID_FEE_TOO_CLOSE_TO_ZERO, None, []
            min_fee_rate = self.mempool.get_min_fee_rate(cost)
            if min_fee_rate is None:
                return Err.INVALID_COST_RESULT, None, []
            if fees_per_cost <= min_fee_rate:
                return Err.INVALID_FEE_LOW_FEE, None, []

        # Check removals against UnspentDB + DiffStore + Mempool + SpendBundle
        # Use this information later when constructing a block
        fail_reason, conflicts = check_removals(
            removal_record_dict, bundle_coin_spends, get_items_by_coin_ids=self.mempool.get_items_by_coin_ids
        )

        # If we have a mempool conflict, continue, since we still want to keep around the TX in the pending pool.
        if fail_reason is not None and fail_reason is not Err.MEMPOOL_CONFLICT:
            return fail_reason, None, []

        # Verify conditions, create hash_key list for aggsig check
        for spend in conds.spends:
            coin_record: CoinRecord = removal_record_dict[bytes32(spend.coin_id)]
            # Check that the revealed removal puzzles actually match the puzzle hash
            if spend.puzzle_hash != coin_record.coin.puzzle_hash:
                log.warning("Mempool rejecting transaction because of wrong puzzle_hash")
                log.warning(f"{spend.puzzle_hash.hex()} != {coin_record.coin.puzzle_hash.hex()}")
                return Err.WRONG_PUZZLE_HASH, None, []

        # the height and time we pass in here represent the previous transaction
        # block's height and timestamp. In the mempool, the most recent peak
        # block we've received will be the previous transaction block, from the
        # point-of-view of the next block to be farmed. Therefore we pass in the
        # current peak's height and timestamp
        assert self.peak.timestamp is not None
        tl_error_rust: int | None = check_time_locks(
            removal_record_dict,
            conds,
            self.peak.height,
            self.peak.timestamp,
        )
        tl_error: Err | None = None
        if tl_error_rust is not None:
            tl_error = Err(tl_error_rust)

        timelocks: TimelockConditions = compute_assert_height(removal_record_dict, conds)

        if timelocks.assert_before_height is not None and timelocks.assert_before_height <= timelocks.assert_height:
            # returning None as the "potential" means it failed. We won't store it
            # in the pending cache
            return Err.IMPOSSIBLE_HEIGHT_ABSOLUTE_CONSTRAINTS, None, []  # MempoolInclusionStatus.FAILED
        if timelocks.assert_before_seconds is not None and timelocks.assert_before_seconds <= timelocks.assert_seconds:
            return Err.IMPOSSIBLE_SECONDS_ABSOLUTE_CONSTRAINTS, None, []  # MempoolInclusionStatus.FAILED

        potential = MempoolItem(
            new_spend.aggregated_signature,
            uint64(fees),
            conds,
            spend_name,
            first_added_height,
            timelocks.assert_height,
            timelocks.assert_before_height,
            timelocks.assert_before_seconds,
            bundle_coin_spends,
        )

        if tl_error:
            if tl_error is Err.ASSERT_HEIGHT_ABSOLUTE_FAILED or tl_error is Err.ASSERT_HEIGHT_RELATIVE_FAILED:
                return tl_error, potential, []  # MempoolInclusionStatus.PENDING
            else:
                return tl_error, None, []  # MempoolInclusionStatus.FAILED

        if fail_reason is Err.MEMPOOL_CONFLICT:
            log.debug(f"Replace attempted. number of MempoolItems: {len(conflicts)}")
            if not can_replace(conflicts, potential):
                return Err.MEMPOOL_CONFLICT, potential, []

        duration = time.monotonic() - start_time

        log.log(
            logging.DEBUG if duration < 2 else logging.WARNING,
            f"add_spendbundle {spend_name} took {duration:0.2f} seconds. "
            f"Cost: {cost} ({round(100.0 * cost / self.constants.MAX_BLOCK_COST_CLVM, 3)}% of max block cost)",
        )

        if duration > 2:
            log.warning("validating spend took too long, rejecting")
            return Err.INVALID_SPEND_BUNDLE, None, []

        return None, potential, [item.name for item in conflicts]

    def get_spendbundle(self, bundle_hash: bytes32) -> SpendBundle | None:
        """Returns a full SpendBundle if it's inside one the mempools"""
        item: MempoolItem | None = self.mempool.get_item_by_id(bundle_hash)
        if item is not None:
            return item.to_spend_bundle()
        return None

    def get_mempool_item(self, bundle_hash: bytes32, include_pending: bool = False) -> MempoolItem | None:
        """
        Returns the MempoolItem in the mempool that matches the provided spend bundle hash (id)
        or None if not found.

        If include_pending is specified, also check the PENDING cache.
        """
        item = self.mempool.get_item_by_id(bundle_hash)
        if not item and include_pending:
            # no async lock needed since we're not mutating the pending_cache
            item = self._pending_cache.get(bundle_hash)
        if not item and include_pending:
            item = self._conflict_cache.get(bundle_hash)

        return item

    async def new_peak(self, new_peak: BlockRecordProtocol | None, spent_coins: list[bytes32] | None) -> NewPeakInfo:
        """
        Called when a new peak is available, we try to recreate a mempool for the new tip.
        new_peak should always be the most recent *transaction* block of the chain. Since
        the mempool cannot traverse the chain to find the most recent transaction block,
        we wouldn't be able to detect, and correctly update the mempool, if we saw a
        non-transaction block on a fork. self.peak must always be set to a transaction
        block.
        """
        if new_peak is None:
            return NewPeakInfo([], [])
        # we're only interested in transaction blocks
        if new_peak.is_transaction_block is False:
            return NewPeakInfo([], [])
        if self.peak == new_peak:
            return NewPeakInfo([], [])
        assert new_peak.timestamp is not None
        self.fee_estimator.new_block_height(new_peak.height)
        included_items: list[MempoolItemInfo] = []
        new_peak_start = time.monotonic()

        expired = self.mempool.new_tx_block(new_peak.height, new_peak.timestamp)
        mempool_item_removals: list[MempoolRemoveInfo] = [expired]

        use_optimization: bool = self.peak is not None and new_peak.prev_transaction_block_hash == self.peak.header_hash
        self.peak = new_peak

        lineage_cache = LineageInfoCache(self.get_unspent_lineage_info_for_puzzle_hash)

        if use_optimization and spent_coins is not None:
            # We don't reinitialize a mempool, just kick removed items
            # transactions in the mempool may be spending multiple coins,
            # when looking up transactions by all coin IDs, we're likely to
            # find the same transaction multiple times. We put them in a set
            # to deduplicate
            spendbundle_ids_to_remove: set[bytes32] = set()

            # rebasing a fast forward spend is more expensive than to just
            # evict the item. So, any FF spend we may need to rebase, defer
            # them until after we've gone through all spends
            deferred_ff_items: set[tuple[bytes32, MempoolItem]] = set()

            for spend in spent_coins:
                items = self.mempool.get_items_by_coin_id(spend)
                for item in items:
                    # this is a property, compute it once
                    item_name = item.name

                    # if we've already decided to remove this mempool item
                    # because of some other coin, we don't need to do any more
                    # work
                    if item_name in spendbundle_ids_to_remove:
                        continue

                    bcs = item.bundle_coin_spends.get(spend)
                    if bcs is not None and bcs.latest_singleton_lineage is None:
                        # this is a regular coin spend that's now made it into
                        # a block and we just evict its mempool item
                        included_items.append(MempoolItemInfo(item.cost, item.fee, item.height_added_to_mempool))
                        self.remove_seen(item_name)
                        spendbundle_ids_to_remove.add(item_name)
                        continue

                    deferred_ff_items.add((spend, item))

            # fast forward spends are indexed under the latest singleton coin ID
            # if it's spent, we need to update the index in the mempool. This
            # list lets us perform a bulk update
            # new_coin_id, current_coin_id, mempool item name
            spends_to_update: list[tuple[bytes32, bytes32, bytes32]] = []

            for spend, item in deferred_ff_items:
                item_name = item.spend_bundle_name
                if item_name in spendbundle_ids_to_remove:
                    continue
                # there may be multiple matching spends in the mempool
                # item, for the same singleton
                found_matches = 0
                for bcs in item.bundle_coin_spends.values():
                    if bcs.latest_singleton_lineage is None or bcs.latest_singleton_lineage.coin_id != spend:
                        continue
                    found_matches += 1

                    # TODO: in the future, we could pass this new coin ID
                    # into new_peak() and avoid this DB lookup
                    lineage_info = await lineage_cache.get_unspent_lineage_info(bcs.coin_spend.coin.puzzle_hash)
                    if lineage_info is None:
                        # this singleton no longer has an unspent coin with
                        # this puzzle-hash. FF is not longer available and we
                        # just need to evict this mempool item
                        self.remove_seen(item_name)
                        spendbundle_ids_to_remove.add(item_name)
                        break

                    spends_to_update.append((lineage_info.coin_id, spend, item_name))
                    bcs.latest_singleton_lineage = lineage_info

                if found_matches == 0:  # pragma: no cover
                    # We are not expected to get here. this is all
                    # defensive to get rid of the spend bundle or patch
                    # it up
                    log.warning(
                        f"MempoolItem indexed as spending coin: {spend}, "
                        f"but spend is not found in item: {item_name}. Evicting mempool item"
                    )
                    # we don't expect this to happen, so evict the
                    # item as a precaution
                    spendbundle_ids_to_remove.add(item_name)

            if len(spends_to_update) > 0:
                self.mempool.update_spend_index(spends_to_update)

            mempool_item_removals.append(
                self.mempool.remove_from_pool(list(spendbundle_ids_to_remove), MempoolRemoveReason.BLOCK_INCLUSION)
            )
        else:
            log.warning(
                "updating the mempool using the slow-path. "
                f"peak: {self.peak.header_hash.hex()} "
                f"new-peak-prev: {new_peak.prev_transaction_block_hash} "
                f"coins: {'not set' if spent_coins is None else 'set'}"
            )
            old_pool = self.mempool
            self.mempool = Mempool(old_pool.mempool_info, old_pool.fee_estimator)
            self.seen_bundle_hashes = {}

            # in order to make this a bit quicker, we look-up all the spends in
            # a single query, rather than one at a time.
            coin_records: dict[bytes32, CoinRecord] = {}

            removals: set[bytes32] = set()
            for item in old_pool.all_items():
                removals.update(item.bundle_coin_spends)

            for record in await self.get_coin_records(removals):
                name = record.coin.name()
                coin_records[name] = record

            async def local_get_coin_records(names: Collection[bytes32]) -> list[CoinRecord]:
                ret: list[CoinRecord] = []
                for name in names:
                    r = coin_records.get(name)
                    if r is not None:
                        ret.append(r)
                return ret

            for item in old_pool.all_items():
                info = await self.add_spend_bundle(
                    item.to_spend_bundle(),
                    item.conds,
                    item.spend_bundle_name,
                    item.height_added_to_mempool,
                    local_get_coin_records,
                    lineage_cache.get_unspent_lineage_info,
                )
                # Only add to `seen` if inclusion worked, so it can be resubmitted in case of a reorg
                if info.status == MempoolInclusionStatus.SUCCESS:
                    self.add_and_maybe_pop_seen(item.spend_bundle_name)
                # If the spend bundle was confirmed or conflicting (can no longer be in mempool), it won't be
                # successfully added to the new mempool.
                if info.status == MempoolInclusionStatus.FAILED and info.error == Err.DOUBLE_SPEND:
                    # Item was in mempool, but after the new block it's a double spend.
                    # Item is most likely included in the block.
                    included_items.append(MempoolItemInfo(item.cost, item.fee, item.height_added_to_mempool))

            old_pool.close()

        potential_txs = self._pending_cache.drain(new_peak.height)
        potential_txs.update(self._conflict_cache.drain())
        txs_added = []
        for item in potential_txs.values():
            info = await self.add_spend_bundle(
                item.to_spend_bundle(),
                item.conds,
                item.spend_bundle_name,
                item.height_added_to_mempool,
                self.get_coin_records,
                lineage_cache.get_unspent_lineage_info,
            )
            if info.status == MempoolInclusionStatus.SUCCESS:
                txs_added.append(item.spend_bundle_name)
            mempool_item_removals.extend(info.removals)
        log.info(
            f"Size of mempool: {self.mempool.size()} spends, "
            f"cost: {self.mempool.total_mempool_cost()} "
            f"minimum fee rate (in FPC) to get in for 5M cost tx: {self.mempool.get_min_fee_rate(5000000)}"
        )
        self.mempool.fee_estimator.new_block(FeeBlockInfo(new_peak.height, included_items))
        duration = time.monotonic() - new_peak_start
        log.log(logging.WARNING if duration > 1 else logging.INFO, f"new_peak() took {duration:0.2f} seconds")
        return NewPeakInfo(txs_added, mempool_item_removals)

    def get_items_not_in_filter(self, mempool_filter: PyBIP158, limit: int = 100) -> list[MempoolItem]:
        items: list[MempoolItem] = []

        assert limit > 0

        # Send 100 with the highest fee per cost
        for item in self.mempool.items_by_feerate():
            if len(items) >= limit:
                return items
            if mempool_filter.Match(bytearray(item.spend_bundle_name)):
                continue
            items.append(item)

        return items


T = TypeVar("T", uint32, uint64)


def optional_min(a: T | None, b: T | None) -> T | None:
    return min((v for v in [a, b] if v is not None), default=None)


def optional_max(a: T | None, b: T | None) -> T | None:
    return max((v for v in [a, b] if v is not None), default=None)


def can_replace(conflicting_items: list[MempoolItem], new_item: MempoolItem) -> bool:
    """
    This function implements the mempool replacement rules. Given a Mempool item
    we're attempting to insert into the mempool (new_item) and the set of existing
    mempool items that conflict with it, this function answers the question whether
    the existing items can be replaced by the new one.
    """

    conflicting_fees = 0
    conflicting_cost = 0
    assert_height: uint32 | None = None
    assert_before_height: uint32 | None = None
    assert_before_seconds: uint64 | None = None
    # we don't allow replacing mempool items with new ones that remove
    # eligibility for dedup and fast-forward. Doing so could be abused by
    # denying such spends from operating as intended
    # collect all coins that are eligible for dedup and FF in the existing items
    existing_ff_spends: set[bytes32] = set()
    existing_dedup_spends: set[bytes32] = set()

    for item in conflicting_items:
        conflicting_fees += item.fee
        conflicting_cost += item.cost

        # All coins spent in all conflicting items must also be spent in the new item. (superset rule). This is
        # important because otherwise there exists an attack. A user spends coin A. An attacker replaces the
        # bundle with AB with a higher fee. An attacker then replaces the bundle with just B with a higher
        # fee than AB therefore kicking out A altogether. The better way to solve this would be to keep a cache
        # of booted transactions like A, and retry them after they get removed from mempool due to a conflict.
        for coin_id, bcs in item.bundle_coin_spends.items():
            if coin_id not in new_item.bundle_coin_spends:
                log.debug("Rejecting conflicting tx as it does not spend conflicting coin %s", coin_id)
                return False
            if bcs.supports_fast_forward:
                existing_ff_spends.add(bytes32(coin_id))
            if bcs.eligible_for_dedup:
                existing_dedup_spends.add(bytes32(coin_id))

        assert_height = optional_max(assert_height, item.assert_height)
        assert_before_height = optional_min(assert_before_height, item.assert_before_height)
        assert_before_seconds = optional_min(assert_before_seconds, item.assert_before_seconds)

    # New item must have higher fee per cost
    conflicting_fees_per_cost = conflicting_fees / conflicting_cost
    if new_item.fee_per_cost <= conflicting_fees_per_cost:
        log.debug(
            f"Rejecting conflicting tx due to not increasing fees per cost "
            f"({new_item.fee_per_cost} <= {conflicting_fees_per_cost})"
        )
        return False

    # New item must increase the total fee at least by a certain amount
    fee_increase = new_item.fee - conflicting_fees
    if fee_increase < MEMPOOL_MIN_FEE_INCREASE:
        log.debug(f"Rejecting conflicting tx due to low fee increase ({fee_increase})")
        return False

    # New item may not have a different effective height/time lock (time-lock rule)
    if new_item.assert_height != assert_height:
        log.debug(
            "Rejecting conflicting tx due to changing ASSERT_HEIGHT constraints %s -> %s",
            assert_height,
            new_item.assert_height,
        )
        return False

    if new_item.assert_before_height != assert_before_height:
        log.debug(
            "Rejecting conflicting tx due to changing ASSERT_BEFORE_HEIGHT constraints %s -> %s",
            assert_before_height,
            new_item.assert_before_height,
        )
        return False

    if new_item.assert_before_seconds != assert_before_seconds:
        log.debug(
            "Rejecting conflicting tx due to changing ASSERT_BEFORE_SECONDS constraints %s -> %s",
            assert_before_seconds,
            new_item.assert_before_seconds,
        )
        return False

    if len(existing_ff_spends) > 0 or len(existing_dedup_spends) > 0:
        for coin_id, bcs in new_item.bundle_coin_spends.items():
            if not bcs.supports_fast_forward and coin_id in existing_ff_spends:
                log.debug("Rejecting conflicting tx due to changing ELIGIBLE_FOR_FF of coin spend %s", coin_id)
                return False

            if not bcs.eligible_for_dedup and coin_id in existing_dedup_spends:
                log.debug("Rejecting conflicting tx due to changing ELIGIBLE_FOR_DEDUP of coin spend %s", coin_id)
                return False

    log.info(f"Replacing conflicting tx in mempool. New tx fee: {new_item.fee}, old tx fees: {conflicting_fees}")
    return True