chia-blockchain/chia/full_node/mempool.py

from __future__ import annotations

import logging
import sqlite3
from collections.abc import Iterator
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from time import monotonic

from chia_rs import (
    DONT_VALIDATE_SIGNATURE,
    AugSchemeMPL,
    BlockBuilder,
    Coin,
    CoinSpend,
    ConsensusConstants,
    G2Element,
    SpendBundle,
    get_flags_for_height_and_constants,
    run_block_generator2,
    solution_generator_backrefs,
)
from chia_rs.sized_bytes import bytes32
from chia_rs.sized_ints import uint32, uint64

from chia.consensus.default_constants import DEFAULT_CONSTANTS
from chia.full_node.eligible_coin_spends import (
    IdenticalSpendDedup,
    SingletonFastForward,
    SkipDedup,
)
from chia.full_node.fee_estimation import FeeMempoolInfo, MempoolInfo, MempoolItemInfo
from chia.full_node.fee_estimator_interface import FeeEstimatorInterface
from chia.types.blockchain_format.serialized_program import SerializedProgram
from chia.types.clvm_cost import CLVMCost
from chia.types.generator_types import NewBlockGenerator
from chia.types.internal_mempool_item import InternalMempoolItem
from chia.types.mempool_item import MempoolItem
from chia.util.batches import to_batches
from chia.util.db_wrapper import SQLITE_MAX_VARIABLE_NUMBER
from chia.util.errors import Err

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 = 10

# Threshold after which we stop including mempool items with fast-forward or
# dedup spends during the creation of a block generator. 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
MEMPOOL_ITEM_FEE_LIMIT = 2**50


@dataclass
class MempoolRemoveInfo:
    items: dict[bytes32, InternalMempoolItem]
    reason: MempoolRemoveReason


@dataclass
class MempoolAddInfo:
    removals: list[MempoolRemoveInfo]
    error: Err | None


class MempoolRemoveReason(Enum):
    CONFLICT = 1
    BLOCK_INCLUSION = 2
    POOL_FULL = 3
    EXPIRED = 4


class Mempool:
    _db_conn: sqlite3.Connection
    # it's expensive to serialize and deserialize G2Element, so we keep those in
    # this separate dictionary
    _items: dict[bytes32, InternalMempoolItem]

    # the most recent block height and timestamp that we know of
    _block_height: uint32
    _timestamp: uint64

    _total_fee: int
    _total_cost: int

    def __init__(self, mempool_info: MempoolInfo, fee_estimator: FeeEstimatorInterface):
        self._db_conn = sqlite3.connect(":memory:")
        self._items = {}
        self._block_height = uint32(0)
        self._timestamp = uint64(0)
        self._total_fee = 0
        self._total_cost = 0

        with self._db_conn:
            # name means SpendBundle hash
            # assert_height may be NIL
            # the seq field indicates the order of items being added to the
            # mempool. It's used as a tie-breaker for items with the same fee
            # rate
            # TODO: In the future, for the "fee_per_cost" field, opt for
            # "GENERATED ALWAYS AS (CAST(fee AS REAL) / cost) VIRTUAL"
            self._db_conn.execute(
                """CREATE TABLE tx(
                name BLOB,
                cost INT NOT NULL,
                fee INT NOT NULL,
                assert_height INT,
                assert_before_height INT,
                assert_before_seconds INT,
                fee_per_cost REAL,
                seq INTEGER PRIMARY KEY AUTOINCREMENT)
                """
            )
            self._db_conn.execute("CREATE INDEX name_idx ON tx(name)")
            self._db_conn.execute("CREATE INDEX feerate ON tx(fee_per_cost)")
            self._db_conn.execute(
                "CREATE INDEX assert_before ON tx(assert_before_height, assert_before_seconds) "
                "WHERE assert_before_height IS NOT NULL OR assert_before_seconds IS NOT NULL"
            )

            # This table maps coin IDs to spend bundles hashes
            self._db_conn.execute(
                """CREATE TABLE spends(
                coin_id BLOB NOT NULL,
                tx BLOB NOT NULL,
                UNIQUE(coin_id, tx))
                """
            )
            self._db_conn.execute("CREATE INDEX spend_by_coin ON spends(coin_id)")
            self._db_conn.execute("CREATE INDEX spend_by_bundle ON spends(tx)")

        self.mempool_info: MempoolInfo = mempool_info
        self.fee_estimator: FeeEstimatorInterface = fee_estimator

    def close(self) -> None:
        self._db_conn.close()

    def _row_to_item(self, row: sqlite3.Row) -> MempoolItem:
        name = bytes32(row[0])
        fee = int(row[2])
        assert_height = row[3]
        assert_before_height = row[4]
        assert_before_seconds = row[5]
        item = self._items[name]

        return MempoolItem(
            item.aggregated_signature,
            uint64(fee),
            item.conds,
            name,
            uint32(item.height_added_to_mempool),
            assert_height,
            assert_before_height,
            assert_before_seconds,
            bundle_coin_spends=item.bundle_coin_spends,
        )

    def total_mempool_fees(self) -> int:
        return self._total_fee

    def total_mempool_cost(self) -> CLVMCost:
        return CLVMCost(uint64(self._total_cost))

    def all_items(self) -> Iterator[MempoolItem]:
        with self._db_conn:
            cursor = self._db_conn.execute("SELECT * FROM tx")
            for row in cursor:
                yield self._row_to_item(row)

    def all_item_ids(self) -> list[bytes32]:
        with self._db_conn:
            cursor = self._db_conn.execute("SELECT name FROM tx")
            return [bytes32(row[0]) for row in cursor]

    def items_with_coin_ids(self, coin_ids: set[bytes32]) -> list[bytes32]:
        """
        Returns a list of transaction ids that spend or create any coins with the provided coin ids.
        This iterates over the internal items instead of using a query.
        """

        transaction_ids: list[bytes32] = []

        for transaction_id, item in self._items.items():
            conds = item.conds
            assert conds is not None

            for spend in conds.spends:
                if spend.coin_id in coin_ids:
                    transaction_ids.append(transaction_id)
                    break

                for puzzle_hash, amount, _memo in spend.create_coin:
                    if Coin(spend.coin_id, puzzle_hash, uint64(amount)).name() in coin_ids:
                        transaction_ids.append(transaction_id)
                        break
                else:
                    continue

                break

        return transaction_ids

    def items_with_puzzle_hashes(self, puzzle_hashes: set[bytes32], include_hints: bool) -> list[bytes32]:
        """
        Returns a list of transaction ids that spend or create any coins
        with the provided puzzle hashes (or hints, if enabled).
        This iterates over the internal items instead of using a query.
        """

        transaction_ids: list[bytes32] = []

        for transaction_id, item in self._items.items():
            conds = item.conds
            assert conds is not None

            for spend in conds.spends:
                if spend.puzzle_hash in puzzle_hashes:
                    transaction_ids.append(transaction_id)
                    break

                for puzzle_hash, _amount, memo in spend.create_coin:
                    if puzzle_hash in puzzle_hashes or (include_hints and memo is not None and memo in puzzle_hashes):
                        transaction_ids.append(transaction_id)
                        break
                else:
                    continue

                break

        return transaction_ids

    # TODO: move "process_mempool_items()" into this class in order to do this a
    # bit more efficiently
    def items_by_feerate(self) -> Iterator[MempoolItem]:
        cursor = self._db_conn.execute("SELECT * FROM tx ORDER BY fee_per_cost DESC, seq ASC")
        for row in cursor:
            yield self._row_to_item(row)

    def size(self) -> int:
        cursor = self._db_conn.execute("SELECT COUNT(name) FROM tx")
        row = cursor.fetchone()
        return int(row[0])

    def get_item_by_id(self, item_id: bytes32) -> MempoolItem | None:
        with self._db_conn:
            cursor = self._db_conn.execute("SELECT * FROM tx WHERE name=?", (item_id,))
            row = cursor.fetchone()
            return None if row is None else self._row_to_item(row)

    # TODO: we need a bulk lookup function like this too
    def get_items_by_coin_id(self, spent_coin_id: bytes32) -> Iterator[MempoolItem]:
        cursor = self._db_conn.execute(
            """
            SELECT *
            FROM tx
            WHERE name IN (
                SELECT tx
                FROM spends
                WHERE coin_id = ?
            )
            """,
            (spent_coin_id,),
        )
        for row in cursor:
            yield self._row_to_item(row)

    def get_items_by_coin_ids(self, spent_coin_ids: list[bytes32]) -> list[MempoolItem]:
        items: list[MempoolItem] = []
        for batch in to_batches(spent_coin_ids, SQLITE_MAX_VARIABLE_NUMBER):
            args = ",".join(["?"] * len(batch.entries))
            cursor = self._db_conn.execute(
                f"SELECT * FROM tx WHERE name IN (SELECT tx FROM spends WHERE coin_id IN ({args}))",
                tuple(batch.entries),
            )
            items.extend(self._row_to_item(row) for row in cursor)
        return items

    def get_min_fee_rate(self, cost: int) -> float | None:
        """
        Gets the minimum fpc rate that a transaction with specified cost will need in order to get included.
        """

        if not self.at_full_capacity(cost):
            return 0

        # TODO: make MempoolItem.cost be CLVMCost
        current_cost = self._total_cost

        # Iterates through all spends in increasing fee per cost
        with self._db_conn:
            cursor = self._db_conn.execute("SELECT cost,fee_per_cost FROM tx ORDER BY fee_per_cost ASC, seq DESC")

            item_cost: int
            fee_per_cost: float
            for item_cost, fee_per_cost in cursor:
                current_cost -= item_cost
                # Removing one at a time, until our transaction of size cost fits
                if current_cost + cost <= self.mempool_info.max_size_in_cost:
                    return fee_per_cost

            log.info(
                f"Transaction with cost {cost} does not fit in mempool of max cost {self.mempool_info.max_size_in_cost}"
            )
            return None

    def new_tx_block(self, block_height: uint32, timestamp: uint64) -> MempoolRemoveInfo:
        """
        Remove all items that became invalid because of this new height and
        timestamp. (we don't know about which coins were spent in this new block
        here, so those are handled separately)
        """
        with self._db_conn:
            cursor = self._db_conn.execute(
                "SELECT name FROM tx WHERE assert_before_seconds <= ? OR assert_before_height <= ?",
                (timestamp, block_height),
            )
            to_remove = [bytes32(row[0]) for row in cursor]

        self._block_height = block_height
        self._timestamp = timestamp

        return self.remove_from_pool(to_remove, MempoolRemoveReason.EXPIRED)

    def remove_from_pool(self, items: list[bytes32], reason: MempoolRemoveReason) -> MempoolRemoveInfo:
        """
        Removes an item from the mempool.
        """
        if items == []:
            return MempoolRemoveInfo({}, reason)

        removed_items: list[MempoolItemInfo] = []
        if reason != MempoolRemoveReason.BLOCK_INCLUSION:
            for batch in to_batches(items, SQLITE_MAX_VARIABLE_NUMBER):
                args = ",".join(["?"] * len(batch.entries))
                with self._db_conn:
                    cursor = self._db_conn.execute(
                        f"SELECT name, cost, fee FROM tx WHERE name in ({args})", batch.entries
                    )
                    for row in cursor:
                        name = bytes32(row[0])
                        internal_item = self._items[name]
                        item = MempoolItemInfo(int(row[1]), int(row[2]), internal_item.height_added_to_mempool)
                        removed_items.append(item)

        removed_internal_items = {name: self._items.pop(name) for name in items}

        for batch in to_batches(items, SQLITE_MAX_VARIABLE_NUMBER):
            args = ",".join(["?"] * len(batch.entries))
            with self._db_conn:
                cursor = self._db_conn.execute(
                    f"SELECT SUM(cost), SUM(fee) FROM tx WHERE name in ({args})", batch.entries
                )
                cost_to_remove, fee_to_remove = cursor.fetchone()

                self._db_conn.execute(f"DELETE FROM tx WHERE name in ({args})", batch.entries)
                self._db_conn.execute(f"DELETE FROM spends WHERE tx in ({args})", batch.entries)

            self._total_cost -= cost_to_remove
            self._total_fee -= fee_to_remove
            assert self._total_cost >= 0
            assert self._total_fee >= 0

        if reason != MempoolRemoveReason.BLOCK_INCLUSION:
            info = FeeMempoolInfo(
                self.mempool_info, self.total_mempool_cost(), self.total_mempool_fees(), datetime.now()
            )
            for iteminfo in removed_items:
                self.fee_estimator.remove_mempool_item(info, iteminfo)

        return MempoolRemoveInfo(removed_internal_items, reason)

    def add_to_pool(self, item: MempoolItem) -> MempoolAddInfo:
        """
        Adds an item to the mempool by kicking out transactions (if it doesn't fit), in order of increasing fee per cost
        """

        assert item.fee < MEMPOOL_ITEM_FEE_LIMIT
        assert item.conds is not None
        assert item.cost <= self.mempool_info.max_block_clvm_cost

        removals: list[MempoolRemoveInfo] = []

        # we have certain limits on transactions that will expire soon
        # (in the next 15 minutes)
        block_cutoff = self._block_height + 48
        time_cutoff = self._timestamp + 900
        if (item.assert_before_height is not None and item.assert_before_height < block_cutoff) or (
            item.assert_before_seconds is not None and item.assert_before_seconds < time_cutoff
        ):
            # this lists only transactions that expire soon, in order of
            # lowest fee rate along with the cumulative cost of such
            # transactions counting from highest to lowest fee rate
            cursor = self._db_conn.execute(
                """
                SELECT name,
                    fee_per_cost,
                    SUM(cost) OVER (ORDER BY fee_per_cost DESC, seq ASC) AS cumulative_cost
                FROM tx
                WHERE assert_before_seconds IS NOT NULL AND assert_before_seconds < ?
                    OR assert_before_height IS NOT NULL AND assert_before_height < ?
                ORDER BY cumulative_cost DESC
                """,
                (time_cutoff, block_cutoff),
            )
            to_remove: list[bytes32] = []
            for row in cursor:
                name, fee_per_cost, cumulative_cost = row

                # there's space for us, stop pruning
                if cumulative_cost + item.cost <= self.mempool_info.max_block_clvm_cost:
                    break

                # we can't evict any more transactions, abort (and don't
                # evict what we put aside in "to_remove" list)
                if fee_per_cost > item.fee_per_cost:
                    return MempoolAddInfo([], Err.INVALID_FEE_LOW_FEE)
                to_remove.append(name)

            removals.append(self.remove_from_pool(to_remove, MempoolRemoveReason.EXPIRED))

            # if we don't find any entries, it's OK to add this entry

        if self._total_cost + item.cost > self.mempool_info.max_size_in_cost:
            # pick the items with the lowest fee per cost to remove
            cursor = self._db_conn.execute(
                """SELECT name FROM tx
                WHERE name NOT IN (
                    SELECT name FROM (
                        SELECT name,
                        SUM(cost) OVER (ORDER BY fee_per_cost DESC, seq ASC) AS total_cost
                        FROM tx) AS tx_with_cost
                    WHERE total_cost <= ?)
                """,
                (self.mempool_info.max_size_in_cost - item.cost,),
            )
            to_remove = [bytes32(row[0]) for row in cursor]
            removals.append(self.remove_from_pool(to_remove, MempoolRemoveReason.POOL_FULL))

        with self._db_conn as conn:
            # TODO: In the future, for the "fee_per_cost" field, opt for
            # "GENERATED ALWAYS AS (CAST(fee AS REAL) / cost) VIRTUAL"
            conn.execute(
                "INSERT INTO "
                "tx(name,cost,fee,assert_height,assert_before_height,assert_before_seconds,fee_per_cost) "
                "VALUES(?, ?, ?, ?, ?, ?, ?)",
                (
                    item.name,
                    item.cost,
                    item.fee,
                    item.assert_height,
                    item.assert_before_height,
                    item.assert_before_seconds,
                    item.fee / item.cost,
                ),
            )
            all_coin_spends = []
            # item.name is a property
            # only compute its name once (the spend bundle name)
            item_name = item.name
            for coin_id, bcs in item.bundle_coin_spends.items():
                # any FF spend should be indexed by its latest singleton coin
                # ID, this way we'll find it when the singleton is spent
                if bcs.latest_singleton_lineage is not None:
                    all_coin_spends.append((bcs.latest_singleton_lineage.coin_id, item_name))
                else:
                    all_coin_spends.append((coin_id, item_name))
            conn.executemany("INSERT OR IGNORE INTO spends VALUES(?, ?)", all_coin_spends)

        self._items[item_name] = InternalMempoolItem(
            item.aggregated_signature, item.conds, item.height_added_to_mempool, item.bundle_coin_spends
        )
        self._total_cost += item.cost
        self._total_fee += item.fee

        info = FeeMempoolInfo(self.mempool_info, self.total_mempool_cost(), self.total_mempool_fees(), datetime.now())
        self.fee_estimator.add_mempool_item(info, MempoolItemInfo(item.cost, item.fee, item.height_added_to_mempool))
        return MempoolAddInfo(removals, None)

    # each tuple holds new_coin_id, current_coin_id, mempool item name
    def update_spend_index(self, spends_to_update: list[tuple[bytes32, bytes32, bytes32]]) -> None:
        with self._db_conn as conn:
            conn.executemany("UPDATE OR REPLACE spends SET coin_id=? WHERE coin_id=? AND tx=?", spends_to_update)

    def at_full_capacity(self, cost: int) -> bool:
        """
        Checks whether the mempool is at full capacity and cannot accept a transaction with size cost.
        """

        return self._total_cost + cost > self.mempool_info.max_size_in_cost

    def create_block_generator(
        self,
        constants: ConsensusConstants,
        prev_tx_height: uint32,
        timeout: float,
    ) -> NewBlockGenerator | None:
        """
        prev_tx_height is needed in case we fast-forward a transaction and we
        need to re-run its puzzle.
        """

        mempool_bundle = self.create_bundle_from_mempool_items(constants, prev_tx_height, timeout)
        if mempool_bundle is None:
            return None

        spend_bundle, additions = mempool_bundle
        removals = spend_bundle.removals()
        log.info(f"Add rem: {len(additions)} {len(removals)}")

        # since the hard fork has activated, block generators are
        # allowed to be serialized with CLVM back-references. We can do that
        # unconditionally.
        start_time = monotonic()
        spends = [(cs.coin, bytes(cs.puzzle_reveal), bytes(cs.solution)) for cs in spend_bundle.coin_spends]
        block_program = solution_generator_backrefs(spends)

        duration = monotonic() - start_time
        log.log(
            logging.INFO if duration < 1 else logging.WARNING,
            f"serializing block generator took {duration:0.2f} seconds "
            f"spends: {len(removals)} additions: {len(additions)}",
        )

        flags = get_flags_for_height_and_constants(prev_tx_height, constants) | DONT_VALIDATE_SIGNATURE

        err, conds = run_block_generator2(
            block_program,
            [],
            constants.MAX_BLOCK_COST_CLVM,
            flags,
            spend_bundle.aggregated_signature,
            None,
            constants,
        )

        # this should not happen. This is essentially an assertion failure
        if err is not None:  # pragma: no cover
            log.error(
                f"Failed to compute block cost during farming: {err} "
                f"prev-tx-height: {prev_tx_height} "
                f"generator: {bytes(block_program).hex()}"
            )
            return None

        assert conds is not None
        assert conds.cost > 0

        return NewBlockGenerator(
            SerializedProgram.from_bytes(block_program),
            [],
            [],
            spend_bundle.aggregated_signature,
            additions,
            removals,
            uint64(conds.cost),
        )

    def create_bundle_from_mempool_items(
        self, constants: ConsensusConstants, prev_tx_height: uint32, timeout: float = 1.0
    ) -> tuple[SpendBundle, list[Coin]] | None:
        cost_sum = 0  # Checks that total cost does not exceed block maximum
        fee_sum = 0  # Checks that total fees don't exceed 64 bits
        processed_spend_bundles = 0
        additions: list[Coin] = []
        # This contains a map of coin ID to a coin spend solution and its
        # isolated cost. We reconstruct it for every bundle we create from
        # mempool items because we deduplicate on the first coin spend solution
        # that comes with the highest fee rate item, and that can change across
        # calls.
        dedup_coin_spends = IdenticalSpendDedup()
        # This contains a map of fast forward eligible singleton puzzle hash to
        # the most recent unspent singleton data, to allow chaining fast forward
        # singleton spends.
        singleton_ff = SingletonFastForward()
        coin_spends: list[CoinSpend] = []
        sigs: list[G2Element] = []
        log.info(f"Starting to make block, max cost: {self.mempool_info.max_block_clvm_cost}")
        bundle_creation_start = monotonic()
        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])
            item = self._items[name]

            current_time = monotonic()
            if current_time - bundle_creation_start >= timeout:
                log.info(f"exiting early, already spent {current_time - bundle_creation_start:0.2f} s")
                break
            try:
                assert item.conds is not None
                cost = item.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.
                    if any(
                        sd.eligible_for_dedup or sd.supports_fast_forward for sd in item.bundle_coin_spends.values()
                    ):
                        log.info(f"Skipping transaction with dedup or FF spends {name}")
                        continue

                    unique_coin_spends = []
                    unique_additions = []
                    for spend_data in item.bundle_coin_spends.values():
                        unique_coin_spends.append(spend_data.coin_spend)
                        unique_additions.extend(spend_data.additions)
                    cost_saving = 0
                else:
                    bundle_coin_spends = singleton_ff.process_fast_forward_spends(
                        mempool_item=item, prev_tx_height=prev_tx_height, constants=constants
                    )
                    unique_coin_spends, cost_saving, unique_additions = dedup_coin_spends.get_deduplication_info(
                        bundle_coin_spends=bundle_coin_spends
                    )
                item_cost = cost - cost_saving
                log.info(
                    "Cumulative cost: %d, fee per cost: %0.4f, item cost: %d", cost_sum, fee / item_cost, item_cost
                )
                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.aggregated_signature)
                cost_sum = new_cost_sum
                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 SkipDedup as e:
                log.info(f"{e}")
                continue
            except Exception as e:
                log.info(f"Exception while checking a mempool item for deduplication: {e}")
                skipped_items += 1
                continue
        if coin_spends == []:
            return None
        log.info(
            f"Cumulative cost of block (real cost should be less) {cost_sum}. Proportion "
            f"full: {cost_sum / self.mempool_info.max_block_clvm_cost}"
        )
        aggregated_signature = AugSchemeMPL.aggregate(sigs)
        agg = SpendBundle(coin_spends, aggregated_signature)
        bundle_creation_end = monotonic()
        duration = bundle_creation_end - bundle_creation_start
        log.log(
            logging.INFO if duration < 1 else logging.WARNING,
            f"create_bundle_from_mempool_items took {duration:0.4f} seconds",
        )
        return agg, additions

    def create_block_generator2(
        self, constants: ConsensusConstants, prev_tx_height: uint32, timeout: float
    ) -> NewBlockGenerator | None:
        fee_sum = 0  # Checks that total fees don't exceed 64 bits
        additions: list[Coin] = []
        removals: list[Coin] = []

        dedup_coin_spends = IdenticalSpendDedup()
        singleton_ff = SingletonFastForward()
        log.info(f"Starting to make block, max cost: {self.mempool_info.max_block_clvm_cost}")
        generator_creation_start = monotonic()
        cursor = self._db_conn.execute("SELECT name, fee FROM tx ORDER BY fee_per_cost DESC, seq ASC")
        builder = BlockBuilder()
        skipped_items = 0
        # the total (estimated) cost of the transactions added so far
        block_cost = 0
        added_spends = 0

        batch_transactions: list[SpendBundle] = []
        batch_additions: list[Coin] = []
        batch_spends = 0
        # this cost only includes conditions and execution cost, not byte-cost
        batch_cost = 0

        for row in cursor:
            current_time = monotonic()
            if current_time - generator_creation_start >= timeout:
                log.info(f"exiting early, already spent {current_time - generator_creation_start:0.2f} s")
                break

            name = bytes32(row[0])
            fee = int(row[1])
            item = self._items[name]
            try:
                assert item.conds is not None
                cost = item.conds.condition_cost + item.conds.execution_cost
                bundle_coin_spends = singleton_ff.process_fast_forward_spends(
                    mempool_item=item, prev_tx_height=prev_tx_height, constants=constants
                )
                unique_coin_spends, cost_saving, unique_additions = dedup_coin_spends.get_deduplication_info(
                    bundle_coin_spends=bundle_coin_spends
                )
                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

                # if adding item would make us exceed the block cost, commit the
                # batch we've built up first, to see if more space may be freed
                # up by the compression
                if block_cost + item.conds.cost - cost_saving > constants.MAX_BLOCK_COST_CLVM:
                    added, done = builder.add_spend_bundles(batch_transactions, uint64(batch_cost), constants)

                    block_cost = builder.cost()
                    if added:
                        added_spends += batch_spends
                        additions.extend(batch_additions)
                        removals.extend([cs.coin for sb in batch_transactions for cs in sb.coin_spends])
                        log.info(
                            f"adding TX batch, additions: {len(batch_additions)} removals: {batch_spends} "
                            f"cost: {batch_cost} total cost: {block_cost}"
                        )
                    else:
                        log.info(f"Skipping transaction batch cumulative cost: {block_cost} batch cost: {batch_cost}")
                        skipped_items += 1

                    batch_cost = 0
                    batch_transactions = []
                    batch_additions = []
                    batch_spends = 0
                    if done:
                        break

                batch_cost += cost - cost_saving
                batch_transactions.append(SpendBundle(unique_coin_spends, item.aggregated_signature))
                batch_spends += len(unique_coin_spends)
                batch_additions.extend(unique_additions)
                fee_sum = new_fee_sum
                block_cost += item.conds.cost - cost_saving
            except SkipDedup as e:
                log.info(f"{e}")
                continue
            except Exception as e:
                log.info(f"Exception while checking a mempool item for deduplication: {e}")
                skipped_items += 1
                continue

        if len(batch_transactions) > 0:
            added, _ = builder.add_spend_bundles(batch_transactions, uint64(batch_cost), constants)
            log.info(f"trying to add residual batch: {len(batch_transactions)} batch cost: {batch_cost} added: {added}")

            if added:
                added_spends += batch_spends
                additions.extend(batch_additions)
                removals.extend([cs.coin for sb in batch_transactions for cs in sb.coin_spends])
                log.info(
                    f"adding TX batch, additions: {len(batch_additions)} removals: {batch_spends} "
                    f"cost: {batch_cost} total cost: {block_cost}"
                )

        if removals == []:
            return None

        generator_creation_end = monotonic()
        duration = generator_creation_end - generator_creation_start
        block_program, signature, cost = builder.finalize(constants)
        log.log(
            logging.INFO if duration < 2 else logging.WARNING,
            f"create_block_generator2() took {duration:0.4f} seconds. "
            f"block cost: {cost} spends: {added_spends} additions: {len(additions)}",
        )

        return NewBlockGenerator(
            SerializedProgram.from_bytes(block_program),
            [],
            [],
            signature,
            additions,
            removals,
            uint64(cost),
        )