CHIA-2900 Fast forward mempool optimization (#19713)

Fast forward mempool optimization.

benchmarks/coin_store.py

@@ -28,13 +28,13 @@ def make_coin() -> Coin:
     return Coin(rand_hash(), rand_hash(), uint64(1))
 
 
-def make_coins(num: int) -> tuple[list[tuple[bytes32, Coin]], list[bytes32]]:
-    additions: list[tuple[bytes32, Coin]] = []
+def make_coins(num: int) -> tuple[list[tuple[bytes32, Coin, bool]], list[bytes32]]:
+    additions: list[tuple[bytes32, Coin, bool]] = []
     hashes: list[bytes32] = []
     for i in range(num):
         c = make_coin()
         coin_id = c.name()
-        additions.append((coin_id, c))
+        additions.append((coin_id, c, False))
         hashes.append(coin_id)
 
     return additions, hashes
@@ -146,7 +146,7 @@ async def run_new_block_benchmark(version: int) -> None:
             # add one new coins
             c = make_coin()
             coin_id = c.name()
-            additions.append((coin_id, c))
+            additions.append((coin_id, c, False))
             total_add += 1
 
             farmer_coin, pool_coin = rewards(uint32(height))

chia/_tests/blockchain/test_blockchain.py

@@ -1,12 +1,13 @@
 from __future__ import annotations
 
 import asyncio
+import copy
 import logging
 import random
 import time
 from collections.abc import AsyncIterator, Awaitable
 from contextlib import asynccontextmanager
-from dataclasses import replace
+from dataclasses import dataclass, replace
 from typing import Optional
 
 import pytest
@@ -20,6 +21,8 @@ from chia_rs import (
     InfusedChallengeChainSubSlot,
     MerkleSet,
     SpendBundle,
+    SpendBundleConditions,
+    SpendConditions,
     TransactionsInfo,
     UnfinishedBlock,
     is_canonical_serialization,
@@ -38,7 +41,7 @@ from chia._tests.conftest import ConsensusMode
 from chia._tests.util.blockchain import create_blockchain
 from chia._tests.util.get_name_puzzle_conditions import get_name_puzzle_conditions
 from chia.consensus.augmented_chain import AugmentedBlockchain
-from chia.consensus.block_body_validation import ForkInfo
+from chia.consensus.block_body_validation import ForkAdd, ForkInfo
 from chia.consensus.block_header_validation import validate_finished_header_block
 from chia.consensus.block_rewards import calculate_base_farmer_reward
 from chia.consensus.blockchain import AddBlockResult, Blockchain
@@ -4235,3 +4238,170 @@ async def test_get_header_blocks_in_range_tx_filter_non_tx_block(empty_blockchai
     blocks_with_filter = await b.get_header_blocks_in_range(0, 42, tx_filter=True)
     empty_tx_filter = b"\x00"
     assert blocks_with_filter[non_tx_block.header_hash].transactions_filter == empty_tx_filter
+
+
+@dataclass(frozen=True)
+class ForkInfoTestSetup:
+    fork_info: ForkInfo
+    initial_additions_since_fork: dict[bytes32, ForkAdd]
+    test_block: FullBlock
+    coin: Coin
+    child_coin: Coin
+
+    @classmethod
+    def create(cls, same_ph_as_parent: bool, same_amount_as_parent: bool) -> ForkInfoTestSetup:
+        from chia._tests.util.network_protocol_data import full_block as test_block
+
+        unrelated_coin = Coin(bytes32([0] * 32), bytes32([1] * 32), uint64(42))
+        # We add this initial state with an unrelated addition, to create a
+        # difference between the `rollback` state and the completely empty
+        # `reset` state.
+        initial_additions_since_fork = {
+            unrelated_coin.name(): ForkAdd(
+                coin=unrelated_coin,
+                confirmed_height=uint32(1),
+                timestamp=uint64(0),
+                hint=None,
+                is_coinbase=False,
+                same_as_parent=False,
+            )
+        }
+        fork_info = ForkInfo(
+            test_block.height - 1,
+            test_block.height - 1,
+            test_block.prev_header_hash,
+            additions_since_fork=copy.copy(initial_additions_since_fork),
+        )
+        puzzle_hash = bytes32([2] * 32)
+        amount = uint64(1337)
+        coin = Coin(bytes32([3] * 32), puzzle_hash, amount)
+        child_coin_ph = puzzle_hash if same_ph_as_parent else bytes32([4] * 32)
+        child_coin_amount = amount if same_amount_as_parent else uint64(0)
+        child_coin = Coin(coin.name(), child_coin_ph, child_coin_amount)
+        return cls(
+            fork_info=fork_info,
+            initial_additions_since_fork=initial_additions_since_fork,
+            test_block=test_block,
+            coin=coin,
+            child_coin=child_coin,
+        )
+
+    def check_additions(self, expected_same_parent_additions: set[bytes32]) -> None:
+        assert all(
+            a in self.fork_info.additions_since_fork and self.fork_info.additions_since_fork[a].same_as_parent
+            for a in expected_same_parent_additions
+        )
+        remaining_additions = set(self.fork_info.additions_since_fork) - expected_same_parent_additions
+        assert not any(self.fork_info.additions_since_fork[a].same_as_parent for a in remaining_additions)
+
+
+@pytest.mark.parametrize("same_ph_as_parent", [True, False])
+@pytest.mark.parametrize("same_amount_as_parent", [True, False])
+@pytest.mark.parametrize("rollback", [True, False])
+@pytest.mark.parametrize("reset", [True, False])
+@pytest.mark.anyio
+async def test_include_spends_same_as_parent(
+    same_ph_as_parent: bool, same_amount_as_parent: bool, rollback: bool, reset: bool
+) -> None:
+    """
+    Tests that `ForkInfo` properly tracks same-as-parent created coins.
+    A created coin is tracked as such if its puzzle hash and amount match the
+    parent. We're covering here `include_spends`, `rollback` and `reset` in the
+    context of same-as-parent coins.
+    """
+    test_setup = ForkInfoTestSetup.create(same_ph_as_parent, same_amount_as_parent)
+    # Now let's prepare the test spend bundle conditions
+    create_coin = [(test_setup.child_coin.puzzle_hash, test_setup.child_coin.amount, None)]
+    conds = SpendBundleConditions(
+        [
+            SpendConditions(
+                test_setup.coin.name(),
+                test_setup.coin.parent_coin_info,
+                test_setup.coin.puzzle_hash,
+                test_setup.coin.amount,
+                None,
+                None,
+                None,
+                None,
+                None,
+                None,
+                create_coin,
+                [],
+                [],
+                [],
+                [],
+                [],
+                [],
+                [],
+                0,
+            )
+        ],
+        0,
+        0,
+        0,
+        None,
+        None,
+        [],
+        0,
+        0,
+        0,
+        True,
+        0,
+        0,
+    )
+    # Now let's run the test
+    test_setup.fork_info.include_spends(conds, test_setup.test_block, test_setup.test_block.header_hash)
+    # Let's make sure the results are as expected
+    expected_same_parent_additions = (
+        {test_setup.child_coin.name()} if same_ph_as_parent and same_amount_as_parent else set()
+    )
+    test_setup.check_additions(expected_same_parent_additions)
+    if rollback:
+        # Now we rollback before the spend that belongs to the test conditions
+        test_setup.fork_info.rollback(test_setup.test_block.prev_header_hash, test_setup.test_block.height - 1)
+        # That should leave only the initial additions we started with, which
+        # are unrelated to the test conditions. We added this initial state to
+        # create a difference between `rollback` state and the completely empty
+        # `reset` state.
+        assert test_setup.fork_info.additions_since_fork == test_setup.initial_additions_since_fork
+    if reset:
+        # Now we reset to a test height and header hash
+        test_setup.fork_info.reset(1, bytes32([0] * 32))
+        # That should leave this empty
+        assert test_setup.fork_info.additions_since_fork == {}
+
+
+@pytest.mark.parametrize("same_ph_as_parent", [True, False])
+@pytest.mark.parametrize("same_amount_as_parent", [True, False])
+@pytest.mark.parametrize("rollback", [True, False])
+@pytest.mark.parametrize("reset", [True, False])
+@pytest.mark.anyio
+async def test_include_block_same_as_parent_coins(
+    same_ph_as_parent: bool, same_amount_as_parent: bool, rollback: bool, reset: bool
+) -> None:
+    """
+    Tests that `ForkInfo` properly tracks same-as-parent created coins.
+    A created coin is tracked as such if its puzzle hash and amount match the
+    parent. We're covering here `include_block`, `rollback` and `reset` in the
+    context of such coins.
+    """
+    test_setup = ForkInfoTestSetup.create(same_ph_as_parent, same_amount_as_parent)
+    # Now let's run the test
+    test_setup.fork_info.include_block(
+        [(test_setup.child_coin, None)], [test_setup.coin], test_setup.test_block, test_setup.test_block.header_hash
+    )
+    # Let's make sure the results are as expected
+    expected_same_as_parent_additions = (
+        {test_setup.child_coin.name()} if same_ph_as_parent and same_amount_as_parent else set()
+    )
+    test_setup.check_additions(expected_same_as_parent_additions)
+    if rollback:
+        # Now we rollback before the spend that belongs to the test conditions
+        test_setup.fork_info.rollback(test_setup.test_block.prev_header_hash, test_setup.test_block.height - 1)
+        # That should leave only the initial additions we started with
+        assert test_setup.fork_info.additions_since_fork == test_setup.initial_additions_since_fork
+    if reset:
+        # Now we reset to a test height and header hash
+        test_setup.fork_info.reset(1, bytes32([0] * 32))
+        # That should leave this empty
+        assert test_setup.fork_info.additions_since_fork == {}

chia/_tests/core/full_node/stores/test_coin_store.py

@@ -5,6 +5,7 @@ from dataclasses import dataclass
 from pathlib import Path
 from typing import Optional
 
+import aiosqlite
 import pytest
 from chia_rs import CoinState, FullBlock, additions_and_removals, get_flags_for_height_and_constants
 from chia_rs.sized_bytes import bytes32
@@ -28,6 +29,7 @@ from chia.types.blockchain_format.coin import Coin
 from chia.types.coin_record import CoinRecord
 from chia.types.mempool_item import UnspentLineageInfo
 from chia.util.casts import int_to_bytes
+from chia.util.db_wrapper import DBWrapper2
 from chia.util.hash import std_hash
 
 constants = test_constants
@@ -105,7 +107,7 @@ async def test_basic_coin_store(db_version: int, softfork_height: uint32, bt: Bl
                     bytes(block.transactions_generator), [], flags, bt.constants
                 )
                 tx_removals = [removal.name() for removal in removals]
-                tx_additions = [addition for addition, _ in additions]
+                tx_additions = [(addition.name(), addition, False) for addition, _ in additions]
             else:
                 tx_removals, tx_additions = [], []
 
@@ -117,7 +119,7 @@ async def test_basic_coin_store(db_version: int, softfork_height: uint32, bt: Bl
                 block.height,
                 block.foliage_transaction_block.timestamp,
                 reward_coins,
-                [(a.name(), a) for a in tx_additions],
+                tx_additions,
                 tx_removals,
             )
 
@@ -127,7 +129,7 @@ async def test_basic_coin_store(db_version: int, softfork_height: uint32, bt: Bl
                         block.height,
                         block.foliage_transaction_block.timestamp,
                         reward_coins,
-                        [(a.name(), a) for a in tx_additions],
+                        tx_additions,
                         tx_removals,
                     )
 
@@ -145,16 +147,16 @@ async def test_basic_coin_store(db_version: int, softfork_height: uint32, bt: Bl
                 assert record is not None
                 assert record.spent
                 all_records.add(record)
-            for coin in tx_additions:
+            for coin_id, coin, _ in tx_additions:
                 # Check that the added coins are added
-                record = await coin_store.get_coin_record(coin.name())
+                record = await coin_store.get_coin_record(coin_id)
                 assert record is not None
                 assert not record.spent
                 assert coin == record.coin
                 all_records.add(record)
 
             db_records = await coin_store.get_coin_records(
-                [c.name() for c in list(should_be_included_prev) + tx_additions] + tx_removals
+                [c.name() for c in should_be_included_prev] + [coin_id for coin_id, _, _ in tx_additions] + tx_removals
             )
             assert len(db_records) == len(should_be_included_prev) + len(tx_removals) + len(tx_additions)
             assert len(db_records) == len(all_records)
@@ -738,7 +740,7 @@ class UnspentLineageInfoTestItem:
     puzzlehash: bytes
     amount: int
     parent_id: bytes
-    is_spent: bool = False
+    spent_index: int = 0
 
 
 @dataclass
@@ -758,7 +760,7 @@ class UnspentLineageInfoCase:
             UnspentLineageInfoTestItem(TEST_COIN_ID, TEST_PUZZLEHASH, TEST_AMOUNT, TEST_PARENT_ID),
             UnspentLineageInfoTestItem(b"2" * 32, b"2" * 32, 2, b"1" * 32),
             UnspentLineageInfoTestItem(b"3" * 32, b"3" * 32, 3, b"2" * 32),
-            UnspentLineageInfoTestItem(TEST_PARENT_ID, b"4" * 32, TEST_AMOUNT, TEST_PARENT_PARENT_ID, is_spent=True),
+            UnspentLineageInfoTestItem(TEST_PARENT_ID, b"4" * 32, TEST_AMOUNT, TEST_PARENT_PARENT_ID, spent_index=1),
         ],
         expected_success=False,
     ),
@@ -773,7 +775,7 @@ class UnspentLineageInfoCase:
                 TEST_PUZZLEHASH,
                 TEST_PARENT_DIFFERENT_AMOUNT,
                 TEST_PARENT_PARENT_ID,
-                is_spent=True,
+                spent_index=1,
             ),
         ],
         parent_with_diff_amount=True,
@@ -796,7 +798,7 @@ class UnspentLineageInfoCase:
             UnspentLineageInfoTestItem(b"2" * 32, TEST_PUZZLEHASH, TEST_AMOUNT, TEST_PARENT_ID),
             UnspentLineageInfoTestItem(b"3" * 32, b"3" * 32, 3, b"2" * 32),
             UnspentLineageInfoTestItem(
-                TEST_PARENT_ID, TEST_PUZZLEHASH, TEST_AMOUNT, TEST_PARENT_PARENT_ID, is_spent=True
+                TEST_PARENT_ID, TEST_PUZZLEHASH, TEST_AMOUNT, TEST_PARENT_PARENT_ID, spent_index=1
             ),
         ],
         expected_success=False,
@@ -804,11 +806,11 @@ class UnspentLineageInfoCase:
     UnspentLineageInfoCase(
         id="Unspent with parent that has same puzzlehash and amount",
         items=[
-            UnspentLineageInfoTestItem(TEST_COIN_ID, TEST_PUZZLEHASH, TEST_AMOUNT, TEST_PARENT_ID),
+            UnspentLineageInfoTestItem(TEST_COIN_ID, TEST_PUZZLEHASH, TEST_AMOUNT, TEST_PARENT_ID, spent_index=-1),
             UnspentLineageInfoTestItem(b"2" * 32, b"2" * 32, 2, b"1" * 32),
             UnspentLineageInfoTestItem(b"3" * 32, b"3" * 32, 3, b"2" * 32),
             UnspentLineageInfoTestItem(
-                TEST_PARENT_ID, TEST_PUZZLEHASH, TEST_AMOUNT, TEST_PARENT_PARENT_ID, is_spent=True
+                TEST_PARENT_ID, TEST_PUZZLEHASH, TEST_AMOUNT, TEST_PARENT_PARENT_ID, spent_index=1
             ),
         ],
         expected_success=True,
@@ -833,7 +835,7 @@ async def test_get_unspent_lineage_info_for_puzzle_hash(case: UnspentLineageInfo
                 (
                     item.coin_id,
                     0,
-                    1 if item.is_spent else 0,
+                    item.spent_index,
                     0,
                     item.puzzlehash,
                     item.parent_id,
@@ -890,3 +892,98 @@ async def test_add_coin_records_to_db() -> None:
             resulting_record = await coin_store.get_coin_record(record.coin.name())
             assert resulting_record is not None
             assert resulting_record == record
+
+
+async def get_spent_index(conn: aiosqlite.Connection, coin_name: bytes32) -> int:
+    cursor = await conn.execute("SELECT spent_index FROM coin_record WHERE coin_name = ?", (coin_name,))
+    row = await cursor.fetchone()
+    assert row is not None
+    return int(row[0])
+
+
+@pytest.mark.anyio
+async def test_new_block_tx_additions() -> None:
+    """
+    Covers properly adding coin records for normal unspent coins and potential
+    fast forward singleton unspent coins. That means giving them spent index 0
+    and -1 respectively.
+    """
+    async with DBConnection(2) as db_wrapper:
+        coin_store = await CoinStore.create(db_wrapper)
+        normal_coin = Coin(bytes32([0] * 32), bytes32([0] * 32), uint64(1))
+        normal_coin_id = normal_coin.name()
+        same_as_parent_coin = Coin(bytes32([0] * 32), bytes32([0] * 32), uint64(1337))
+        same_as_parent_coin_id = same_as_parent_coin.name()
+        await coin_store.new_block(
+            height=uint32(0),
+            timestamp=uint64(1),
+            included_reward_coins=[],
+            tx_additions=[
+                (normal_coin_id, normal_coin, False),
+                (same_as_parent_coin_id, same_as_parent_coin, True),
+            ],
+            tx_removals=[],
+        )
+        async with db_wrapper.reader_no_transaction() as conn:
+            # Normal coin should have spent_index 0
+            assert await get_spent_index(conn, normal_coin_id) == 0
+            # Potential ff singleton should have spent_index -1
+            assert await get_spent_index(conn, same_as_parent_coin_id) == -1
+
+
+@pytest.mark.anyio
+async def test_rollback_to_block_spent_index_update() -> None:
+    """
+    Covers properly marking coins as unspent on rollback. Reward coins and
+    normal coins get `spent_index` set to `0`, potential ff singleton ones get
+    `spent_index` set to `-1`.
+    """
+
+    async def insert_coins(db_wrapper: DBWrapper2, coins: list[tuple[Coin, int, bool]]) -> None:
+        values_to_insert = [
+            (
+                coin.name(),
+                0,
+                spent_index,
+                int(coinbase),
+                coin.puzzle_hash,
+                coin.parent_coin_info,
+                coin.amount.stream_to_bytes(),
+                0,
+            )
+            for coin, spent_index, coinbase in coins
+        ]
+        async with db_wrapper.writer() as conn:
+            await conn.executemany("INSERT INTO coin_record VALUES (?, ?, ?, ?, ?, ?, ?, ?)", values_to_insert)
+
+    async with DBConnection(2) as db_wrapper:
+        coin_store = await CoinStore.create(db_wrapper)
+        # Let's set things up for roll back. All coins are confirmed at height
+        # 0, parent coin gets spent at height 2 and the other test coins get
+        # spent at height 3.
+        parent_coin = Coin(bytes32([0] * 32), bytes32([1] * 32), uint64(1337))
+        parent_coin_id = parent_coin.name()
+        normal_child = Coin(parent_coin_id, bytes32([2] * 32), uint64(42))
+        same_as_parent_child = Coin(parent_coin_id, parent_coin.puzzle_hash, parent_coin.amount)
+        reward_coin = Coin(bytes32([0] * 32), bytes32([0] * 32), uint64(1))
+        await insert_coins(
+            db_wrapper,
+            # List of (coin, spent_index, coinbase) values
+            [
+                (parent_coin, 2, False),
+                (normal_child, 3, False),
+                (same_as_parent_child, 3, False),
+                (reward_coin, 3, True),
+            ],
+        )
+        # Let's roll back
+        await coin_store.rollback_to_block(2)
+        async with db_wrapper.reader_no_transaction() as conn:
+            # Parent should still be spent
+            assert await get_spent_index(conn, parent_coin_id) == 2
+            # Normal child should be unspent with spent_index 0
+            assert await get_spent_index(conn, normal_child.name()) == 0
+            # Same for the reward coin
+            assert await get_spent_index(conn, reward_coin.name()) == 0
+            # The potential ff singleton child should be marked with -1
+            assert await get_spent_index(conn, same_as_parent_child.name()) == -1

chia/_tests/util/spend_sim.py

@@ -219,7 +219,7 @@ class SpendSim:
         coins = set()
         async with self.db_wrapper.reader_no_transaction() as conn:
             cursor = await conn.execute(
-                "SELECT puzzle_hash,coin_parent,amount from coin_record WHERE coinbase=0 AND spent_index==0 ",
+                "SELECT puzzle_hash,coin_parent,amount from coin_record WHERE coinbase=0 AND spent_index <= 0 ",
             )
             rows = await cursor.fetchall()
 
@@ -268,6 +268,7 @@ class SpendSim:
                 if result is not None:
                     bundle, additions = result
                     generator_bundle = bundle
+                    spent_coins: dict[bytes32, Coin] = {}
                     spent_coins_ids = []
                     for spend in generator_bundle.coin_spends:
                         hint_dict, _ = compute_spend_hints_and_additions(spend)
@@ -277,9 +278,15 @@ class SpendSim:
                             if hint_obj.hint is not None:
                                 hints.append((coin_name, bytes(hint_obj.hint)))
                         await self.hint_store.add_hints(hints)
-                        spent_coins_ids.append(spend.coin.name())
+                        spend_id = spend.coin.name()
+                        spent_coins[spend_id] = spend.coin
+                        spent_coins_ids.append(spend_id)
                         tx_removals.append(spend.coin)
-                    tx_additions = [(addition.name(), addition) for addition in additions]
+                    for child in additions:
+                        parent = spent_coins.get(child.parent_coin_info)
+                        assert parent is not None
+                        same_as_parent = child.puzzle_hash == parent.puzzle_hash and child.amount == parent.amount
+                        tx_additions.append((child.name(), child, same_as_parent))
         await self.coin_store.new_block(
             height=uint32(self.block_height + 1),
             timestamp=self.timestamp,
@@ -299,7 +306,7 @@ class SpendSim:
         await self.new_peak(spent_coins_ids)
 
         # return some debugging data
-        return [a for _, a in tx_additions], tx_removals
+        return [a for _, a, _ in tx_additions], tx_removals
 
     def get_height(self) -> uint32:
         return self.block_height

chia/_tests/wallet/test_new_wallet_protocol.py

@@ -828,7 +828,7 @@ async def raw_mpu_setup(one_node: OneNode, self_hostname: str, no_capability: bo
     reward_1 = Coin(std_hash(b"reward 1"), std_hash(b"reward puzzle hash"), uint64(1000))
     reward_2 = Coin(std_hash(b"reward 2"), std_hash(b"reward puzzle hash"), uint64(2000))
     await simulator.full_node.coin_store.new_block(
-        uint32(2), uint64(10000), [reward_1, reward_2], [(coin.name(), coin) for coin, _ in new_coins], []
+        uint32(2), uint64(10000), [reward_1, reward_2], [(coin.name(), coin, False) for coin, _ in new_coins], []
     )
     await simulator.full_node.hint_store.add_hints([(coin.name(), hint) for coin, hint in new_coins])
 
@@ -856,7 +856,7 @@ async def make_coin(full_node: FullNode) -> tuple[Coin, bytes32]:
     reward_1 = Coin(std_hash(b"reward 1"), std_hash(b"reward puzzle hash"), uint64(3000))
     reward_2 = Coin(std_hash(b"reward 2"), std_hash(b"reward puzzle hash"), uint64(4000))
     await full_node.coin_store.new_block(
-        uint32(height + 1), uint64(200000), [reward_1, reward_2], [(coin.name(), coin)], []
+        uint32(height + 1), uint64(200000), [reward_1, reward_2], [(coin.name(), coin, False)], []
     )
     await full_node.hint_store.add_hints([(coin.name(), hint)])
 

chia/consensus/block_body_validation.py

@@ -49,6 +49,8 @@ class ForkAdd:
     timestamp: uint64
     hint: Optional[bytes]
     is_coinbase: bool
+    # This means matching parent puzzle hash and amount
+    same_as_parent: bool
 
 
 @dataclass(frozen=True)
@@ -103,7 +105,9 @@ class ForkInfo:
             timestamp = block.foliage_transaction_block.timestamp
             coin_id = coin.name()
             assert coin_id not in self.additions_since_fork
-            self.additions_since_fork[coin_id] = ForkAdd(coin, block.height, timestamp, None, True)
+            self.additions_since_fork[coin_id] = ForkAdd(
+                coin, block.height, timestamp, hint=None, is_coinbase=True, same_as_parent=False
+            )
 
     def include_spends(self, conds: Optional[SpendBundleConditions], block: FullBlock, header_hash: bytes32) -> None:
         self.update_fork_peak(block, header_hash)
@@ -111,10 +115,14 @@ class ForkInfo:
             assert block.foliage_transaction_block is not None
             timestamp = block.foliage_transaction_block.timestamp
             for spend in conds.spends:
-                self.removals_since_fork[bytes32(spend.coin_id)] = ForkRem(bytes32(spend.puzzle_hash), block.height)
+                spend_coin_id = bytes32(spend.coin_id)
+                self.removals_since_fork[spend_coin_id] = ForkRem(bytes32(spend.puzzle_hash), block.height)
                 for puzzle_hash, amount, hint in spend.create_coin:
-                    coin = Coin(bytes32(spend.coin_id), bytes32(puzzle_hash), uint64(amount))
-                    self.additions_since_fork[coin.name()] = ForkAdd(coin, block.height, timestamp, hint, False)
+                    coin = Coin(spend_coin_id, bytes32(puzzle_hash), uint64(amount))
+                    same_as_parent = coin.puzzle_hash == spend.puzzle_hash and amount == spend.coin_amount
+                    self.additions_since_fork[coin.name()] = ForkAdd(
+                        coin, block.height, timestamp, hint=hint, is_coinbase=False, same_as_parent=same_as_parent
+                    )
         self.include_reward_coins(block)
 
     def include_block(
@@ -127,10 +135,18 @@ class ForkInfo:
         self.update_fork_peak(block, header_hash)
         if block.foliage_transaction_block is not None:
             timestamp = block.foliage_transaction_block.timestamp
+            spent_coins: dict[bytes32, Coin] = {}
             for spend in removals:
-                self.removals_since_fork[bytes32(spend.name())] = ForkRem(bytes32(spend.puzzle_hash), block.height)
+                spend_id = bytes32(spend.name())
+                spent_coins[spend_id] = spend
+                self.removals_since_fork[spend_id] = ForkRem(bytes32(spend.puzzle_hash), block.height)
             for coin, hint in additions:
-                self.additions_since_fork[coin.name()] = ForkAdd(coin, block.height, timestamp, hint, False)
+                parent = spent_coins.get(coin.parent_coin_info)
+                assert parent is not None
+                same_as_parent = coin.puzzle_hash == parent.puzzle_hash and coin.amount == parent.amount
+                self.additions_since_fork[coin.name()] = ForkAdd(
+                    coin, block.height, timestamp, hint=hint, is_coinbase=False, same_as_parent=same_as_parent
+                )
         self.include_reward_coins(block)
 
     def rollback(self, header_hash: bytes32, height: int) -> None:

chia/consensus/blockchain.py

@@ -564,7 +564,7 @@ class Blockchain:
                 if fork_add.confirmed_height == height and fork_add.is_coinbase
             ]
             tx_additions = [
-                (coin_id, fork_add.coin)
+                (coin_id, fork_add.coin, fork_add.same_as_parent)
                 for coin_id, fork_add in fork_info.additions_since_fork.items()
                 if fork_add.confirmed_height == height and not fork_add.is_coinbase
             ]

chia/consensus/coin_store_protocol.py

@@ -23,7 +23,7 @@ class CoinStoreProtocol(Protocol):
         height: uint32,
         timestamp: uint64,
         included_reward_coins: Collection[Coin],
-        tx_additions: Collection[tuple[bytes32, Coin]],
+        tx_additions: Collection[tuple[bytes32, Coin, bool]],
         tx_removals: list[bytes32],
     ) -> None:
         """

chia/full_node/coin_store.py

@@ -66,11 +66,28 @@ class CoinStore:
             log.info("DB: Creating index coin_parent_index")
             await conn.execute("CREATE INDEX IF NOT EXISTS coin_parent_index on coin_record(coin_parent)")
 
+            async with conn.execute("SELECT 1 FROM coin_record LIMIT 1") as cursor:
+                is_new_db = await cursor.fetchone() is None
+            if is_new_db:
+                log.info("DB: Creating index coin_record_ph_ff_unspent_idx")
+                # This partial index optimizes fast forward singleton latest
+                # unspent queries. We're only adding it to new DBs to avoid
+                # complex migrations that affect the huge coin records table.
+                # The performance benefit outweighs the cost of this partial
+                # index as it only includes rows where spent_index is -1.
+                await conn.execute(
+                    """
+                    CREATE INDEX IF NOT EXISTS coin_record_ph_ff_unspent_idx
+                        ON coin_record(puzzle_hash, spent_index)
+                        WHERE spent_index = -1
+                    """
+                )
+
         return self
 
     async def num_unspent(self) -> int:
         async with self.db_wrapper.reader_no_transaction() as conn:
-            async with conn.execute("SELECT COUNT(*) FROM coin_record WHERE spent_index=0") as cursor:
+            async with conn.execute("SELECT COUNT(*) FROM coin_record WHERE spent_index <= 0") as cursor:
                 row = await cursor.fetchone()
         if row is not None:
             count: int = row[0]
@@ -82,7 +99,7 @@ class CoinStore:
         height: uint32,
         timestamp: uint64,
         included_reward_coins: Collection[Coin],
-        tx_additions: Collection[tuple[bytes32, Coin]],
+        tx_additions: Collection[tuple[bytes32, Coin, bool]],
         tx_removals: list[bytes32],
     ) -> None:
         """
@@ -93,14 +110,14 @@ class CoinStore:
 
         db_values_to_insert = []
 
-        for coin_id, coin in tx_additions:
+        for coin_id, coin, same_as_parent in tx_additions:
             db_values_to_insert.append(
                 (
                     coin_id,
                     # confirmed_index
                     height,
                     # spent_index
-                    0,
+                    -1 if same_as_parent else 0,
                     # coinbase
                     0,
                     coin.puzzle_hash,
@@ -155,7 +172,8 @@ class CoinStore:
                 row = await cursor.fetchone()
                 if row is not None:
                     coin = self.row_to_coin(row)
-                    return CoinRecord(coin, row[0], row[1], row[2], row[6])
+                    spent_index = uint32(0) if row[1] <= 0 else uint32(row[1])
+                    return CoinRecord(coin, row[0], spent_index, row[2], row[6])
         return None
 
     async def get_coin_records(self, names: Collection[bytes32]) -> list[CoinRecord]:
@@ -180,7 +198,8 @@ class CoinStore:
             for cursor in cursors:
                 for row in await cursor.fetchall():
                     coin = self.row_to_coin(row)
-                    record = CoinRecord(coin, row[0], row[1], row[2], row[6])
+                    spent_index = uint32(0) if row[1] <= 0 else uint32(row[1])
+                    record = CoinRecord(coin, row[0], spent_index, row[2], row[6])
                     coins.append(record)
 
         return coins
@@ -196,7 +215,8 @@ class CoinStore:
                 coins = []
                 for row in rows:
                     coin = self.row_to_coin(row)
-                    coins.append(CoinRecord(coin, row[0], row[1], row[2], row[6]))
+                    spent_index = uint32(0) if row[1] <= 0 else uint32(row[1])
+                    coins.append(CoinRecord(coin, row[0], spent_index, row[2], row[6]))
                 return coins
 
     async def get_coins_removed_at_height(self, height: uint32) -> list[CoinRecord]:
@@ -211,7 +231,7 @@ class CoinStore:
             ) as cursor:
                 coins = []
                 for row in await cursor.fetchall():
-                    if row[1] != 0:
+                    if row[1] > 0:
                         coin = self.row_to_coin(row)
                         coin_record = CoinRecord(coin, row[0], row[1], row[2], row[6])
                         coins.append(coin_record)
@@ -232,12 +252,13 @@ class CoinStore:
                 f"SELECT confirmed_index, spent_index, coinbase, puzzle_hash, "
                 f"coin_parent, amount, timestamp FROM coin_record INDEXED BY coin_puzzle_hash WHERE puzzle_hash=? "
                 f"AND confirmed_index>=? AND confirmed_index<? "
-                f"{'' if include_spent_coins else 'AND spent_index=0'}",
+                f"{'' if include_spent_coins else 'AND spent_index <= 0'}",
                 (puzzle_hash, start_height, end_height),
             ) as cursor:
                 for row in await cursor.fetchall():
                     coin = self.row_to_coin(row)
-                    coins.add(CoinRecord(coin, row[0], row[1], row[2], row[6]))
+                    spent_index = uint32(0) if row[1] <= 0 else uint32(row[1])
+                    coins.add(CoinRecord(coin, row[0], spent_index, row[2], row[6]))
                 return list(coins)
 
     async def get_coin_records_by_puzzle_hashes(
@@ -260,12 +281,13 @@ class CoinStore:
                 f"coin_parent, amount, timestamp FROM coin_record INDEXED BY coin_puzzle_hash "
                 f"WHERE puzzle_hash in ({'?,' * (len(puzzle_hashes) - 1)}?) "
                 f"AND confirmed_index>=? AND confirmed_index<? "
-                f"{'' if include_spent_coins else 'AND spent_index=0'}",
+                f"{'' if include_spent_coins else 'AND spent_index <= 0'}",
                 (*puzzle_hashes_db, start_height, end_height),
             ) as cursor:
                 for row in await cursor.fetchall():
                     coin = self.row_to_coin(row)
-                    coins.add(CoinRecord(coin, row[0], row[1], row[2], row[6]))
+                    spent_index = uint32(0) if row[1] <= 0 else uint32(row[1])
+                    coins.add(CoinRecord(coin, row[0], spent_index, row[2], row[6]))
                 return list(coins)
 
     async def get_coin_records_by_names(
@@ -286,12 +308,13 @@ class CoinStore:
                 f"coin_parent, amount, timestamp FROM coin_record INDEXED BY sqlite_autoindex_coin_record_1 "
                 f"WHERE coin_name in ({'?,' * (len(names) - 1)}?) "
                 f"AND confirmed_index>=? AND confirmed_index<? "
-                f"{'' if include_spent_coins else 'AND spent_index=0'}",
+                f"{'' if include_spent_coins else 'AND spent_index <= 0'}",
                 [*names, start_height, end_height],
             ) as cursor:
                 for row in await cursor.fetchall():
                     coin = self.row_to_coin(row)
-                    coins.add(CoinRecord(coin, row[0], row[1], row[2], row[6]))
+                    spent_index = uint32(0) if row[1] <= 0 else uint32(row[1])
+                    coins.add(CoinRecord(coin, row[0], spent_index, row[2], row[6]))
 
         return list(coins)
 
@@ -301,7 +324,7 @@ class CoinStore:
     def row_to_coin_state(self, row: sqlite3.Row) -> CoinState:
         coin = self.row_to_coin(row)
         spent_h = None
-        if row[1] != 0:
+        if row[1] > 0:
             spent_h = row[1]
         return CoinState(coin, spent_h, row[0])
 
@@ -325,7 +348,7 @@ class CoinStore:
                     f"coin_parent, amount, timestamp FROM coin_record INDEXED BY coin_puzzle_hash "
                     f"WHERE puzzle_hash in ({'?,' * (len(batch.entries) - 1)}?) "
                     f"AND (confirmed_index>=? OR spent_index>=?)"
-                    f"{'' if include_spent_coins else 'AND spent_index=0'}"
+                    f"{'' if include_spent_coins else ' AND spent_index <= 0'}"
                     " LIMIT ?",
                     (*puzzle_hashes_db, min_height, min_height, max_items - len(coins)),
                 ) as cursor:
@@ -356,12 +379,13 @@ class CoinStore:
                     f"SELECT confirmed_index, spent_index, coinbase, puzzle_hash, coin_parent, amount, timestamp "
                     f"FROM coin_record WHERE coin_parent in ({'?,' * (len(batch.entries) - 1)}?) "
                     f"AND confirmed_index>=? AND confirmed_index<? "
-                    f"{'' if include_spent_coins else 'AND spent_index=0'}",
+                    f"{'' if include_spent_coins else 'AND spent_index <= 0'}",
                     (*parent_ids_db, start_height, end_height),
                 ) as cursor:
                     async for row in cursor:
                         coin = self.row_to_coin(row)
-                        coins.add(CoinRecord(coin, row[0], row[1], row[2], row[6]))
+                        spent_index = uint32(0) if row[1] <= 0 else uint32(row[1])
+                        coins.add(CoinRecord(coin, row[0], spent_index, row[2], row[6]))
 
         return list(coins)
 
@@ -390,7 +414,7 @@ class CoinStore:
                     f"SELECT confirmed_index, spent_index, coinbase, puzzle_hash, coin_parent, amount, timestamp "
                     f"FROM coin_record WHERE coin_name in ({'?,' * (len(batch.entries) - 1)}?) "
                     f"AND (confirmed_index>=? OR spent_index>=?) {max_height_sql}"
-                    f"{'' if include_spent_coins else 'AND spent_index=0'}"
+                    f"{'' if include_spent_coins else 'AND spent_index <= 0'}"
                     " LIMIT ?",
                     (*coin_ids_db, min_height, min_height, max_items - len(coins)),
                 ) as cursor:
@@ -435,7 +459,7 @@ class CoinStore:
             puzzle_hash_count = len(puzzle_hashes_db)
 
             require_spent = "spent_index>0"
-            require_unspent = "spent_index=0"
+            require_unspent = "spent_index <= 0"
             amount_filter = "AND amount>=? " if min_amount > 0 else ""
 
             if include_spent and include_unspent:
@@ -534,7 +558,8 @@ class CoinStore:
             )
             for row in rows:
                 coin = self.row_to_coin(row)
-                record = CoinRecord(coin, uint32(0), row[1], row[2], uint64(0))
+                spent_index = uint32(0) if row[1] <= 0 else uint32(row[1])
+                record = CoinRecord(coin, uint32(0), spent_index, row[2], uint64(0))
                 coin_name = bytes32(row[7])
                 coin_changes[coin_name] = record
 
@@ -554,7 +579,32 @@ class CoinStore:
                 if coin_name not in coin_changes:
                     coin_changes[coin_name] = record
 
-            await conn.execute("UPDATE coin_record SET spent_index=0 WHERE spent_index>?", (block_index,))
+            # If the coin to update is not a reward coin and its parent is
+            # spent and has the same puzzle hash and amount, we set its
+            # spent_index to -1 as a potential fast forward singleton unspent
+            # otherwise we set it to 0 as a normal unspent.
+            await conn.execute(
+                """
+                UPDATE coin_record INDEXED BY coin_spent_index
+                SET spent_index = CASE
+                    WHEN
+                        coinbase = 0 AND
+                        EXISTS (
+                            SELECT 1
+                            FROM coin_record AS parent INDEXED BY sqlite_autoindex_coin_record_1
+                            WHERE
+                                parent.coin_name = coin_record.coin_parent AND
+                                parent.puzzle_hash = coin_record.puzzle_hash AND
+                                parent.amount = coin_record.amount AND
+                                parent.spent_index > 0
+                        )
+                    THEN -1
+                    ELSE 0
+                END
+                WHERE spent_index > ?
+                """,
+                (block_index,),
+            )
         return coin_changes
 
     # Update coin_record to be spent in DB
@@ -571,7 +621,7 @@ class CoinStore:
                 ret: Cursor = await conn.execute(
                     f"UPDATE coin_record INDEXED BY sqlite_autoindex_coin_record_1 "
                     f"SET spent_index={index} "
-                    f"WHERE spent_index=0 "
+                    f"WHERE spent_index <= 0 "
                     f"AND coin_name IN ({name_params})",
                     batch.entries,
                 )
@@ -588,9 +638,9 @@ class CoinStore:
                 "SELECT unspent.coin_name, "
                 "unspent.coin_parent, "
                 "parent.coin_parent "
-                "FROM coin_record AS unspent INDEXED BY coin_puzzle_hash "
+                "FROM coin_record AS unspent "
                 "LEFT JOIN coin_record AS parent ON unspent.coin_parent = parent.coin_name "
-                "WHERE unspent.spent_index = 0 "
+                "WHERE unspent.spent_index = -1 "
                 "AND parent.spent_index > 0 "
                 "AND unspent.puzzle_hash = ? "
                 "AND parent.puzzle_hash = unspent.puzzle_hash "