diff --git a/chia/full_node/full_node.py b/chia/full_node/full_node.py
index 2e2ff0703fce..1036345cbd3e 100644
--- a/chia/full_node/full_node.py
+++ b/chia/full_node/full_node.py
@@ -30,6 +30,7 @@
 )
 
 from chia_rs import AugSchemeMPL
+from packaging.version import Version
 
 from chia.consensus.block_body_validation import ForkInfo
 from chia.consensus.block_creation import unfinished_block_to_full_block
@@ -1896,10 +1897,11 @@ async def add_unfinished_block(
             return None
 
         block_hash = bytes32(block.reward_chain_block.get_hash())
+        foliage_tx_hash = block.foliage.foliage_transaction_block_hash
 
-        # This searched for the trunk hash (unfinished reward hash). If we have already added a block with the same
-        # hash, return
-        if self.full_node_store.get_unfinished_block(block_hash) is not None:
+        # If we have already added the block with this reward block hash and
+        # foliage hash, return
+        if self.full_node_store.get_unfinished_block2(block_hash, foliage_tx_hash)[0] is not None:
             return None
 
         peak: Optional[BlockRecord] = self.blockchain.get_peak()
@@ -1986,7 +1988,7 @@ async def add_unfinished_block(
         assert validate_result.required_iters is not None
 
         # Perform another check, in case we have already concurrently added the same unfinished block
-        if self.full_node_store.get_unfinished_block(block_hash) is not None:
+        if self.full_node_store.get_unfinished_block2(block_hash, foliage_tx_hash)[0] is not None:
             return None
 
         if block.prev_header_hash == self.constants.GENESIS_CHALLENGE:
@@ -2065,12 +2067,27 @@ async def add_unfinished_block(
         timelord_msg = make_msg(ProtocolMessageTypes.new_unfinished_block_timelord, timelord_request)
         await self.server.send_to_all([timelord_msg], NodeType.TIMELORD)
 
+        # create two versions of the NewUnfinishedBlock message, one to be sent
+        # to newer clients and one for older clients
         full_node_request = full_node_protocol.NewUnfinishedBlock(block.reward_chain_block.get_hash())
         msg = make_msg(ProtocolMessageTypes.new_unfinished_block, full_node_request)
-        if peer is not None:
-            await self.server.send_to_all([msg], NodeType.FULL_NODE, peer.peer_node_id)
-        else:
-            await self.server.send_to_all([msg], NodeType.FULL_NODE)
+
+        full_node_request2 = full_node_protocol.NewUnfinishedBlock2(
+            block.reward_chain_block.get_hash(), block.foliage.foliage_transaction_block_hash
+        )
+        msg2 = make_msg(ProtocolMessageTypes.new_unfinished_block2, full_node_request2)
+
+        def old_clients(conn: WSChiaConnection) -> bool:
+            # don't send this to peers with new clients
+            return conn.protocol_version <= Version("0.0.35")
+
+        def new_clients(conn: WSChiaConnection) -> bool:
+            # don't send this to peers with old clients
+            return conn.protocol_version > Version("0.0.35")
+
+        peer_id: Optional[bytes32] = None if peer is None else peer.peer_node_id
+        await self.server.send_to_all_if([msg], NodeType.FULL_NODE, old_clients, peer_id)
+        await self.server.send_to_all_if([msg2], NodeType.FULL_NODE, new_clients, peer_id)
 
         self._state_changed("unfinished_block")
 
diff --git a/chia/full_node/full_node_api.py b/chia/full_node/full_node_api.py
index ad8cd451584b..f5fea84107c1 100644
--- a/chia/full_node/full_node_api.py
+++ b/chia/full_node/full_node_api.py
@@ -430,6 +430,12 @@ async def new_unfinished_block(
         if block_hash in self.full_node.full_node_store.requesting_unfinished_blocks:
             return None
 
+        # if we've already learned about an unfinished block with this reward
+        # hash via the v2 protocol, and we've requested it. Assume it's the same
+        # block
+        if block_hash in self.full_node.full_node_store.requesting_unfinished_blocks2:
+            return None
+
         msg = make_msg(
             ProtocolMessageTypes.request_unfinished_block,
             full_node_protocol.RequestUnfinishedBlock(block_hash),
@@ -462,6 +468,70 @@ async def request_unfinished_block(
             return msg
         return None
 
+    @api_request()
+    async def new_unfinished_block2(
+        self, new_unfinished_block: full_node_protocol.NewUnfinishedBlock2
+    ) -> Optional[Message]:
+        # Ignore if syncing
+        if self.full_node.sync_store.get_sync_mode():
+            return None
+        block_hash = new_unfinished_block.unfinished_reward_hash
+        foliage_hash = new_unfinished_block.foliage_hash
+        entry, count, have_better = self.full_node.full_node_store.get_unfinished_block2(block_hash, foliage_hash)
+
+        if entry is not None:
+            return None
+
+        if have_better:
+            self.log.info(
+                f"Already have a better Unfinished Block with partial hash {block_hash.hex()} ignoring this one"
+            )
+            return None
+
+        max_duplicate_unfinished_blocks = self.full_node.config.get("max_duplicate_unfinished_blocks", 3)
+        if count > max_duplicate_unfinished_blocks:
+            self.log.info(
+                f"Already have {count} Unfinished Blocks with partial hash {block_hash.hex()} ignoring another one"
+            )
+            return None
+
+        # This prevents us from downloading the same block from many peers
+        if self.full_node.full_node_store.is_requesting_unfinished_block(block_hash, foliage_hash):
+            return None
+
+        msg = make_msg(
+            ProtocolMessageTypes.request_unfinished_block2,
+            full_node_protocol.RequestUnfinishedBlock2(block_hash, foliage_hash),
+        )
+        self.full_node.full_node_store.mark_requesting_unfinished_block(block_hash, foliage_hash)
+
+        # However, we want to eventually download from other peers, if this peer does not respond
+        # Todo: keep track of who it was
+        async def eventually_clear() -> None:
+            await asyncio.sleep(5)
+            self.full_node.full_node_store.remove_requesting_unfinished_block(block_hash, foliage_hash)
+
+        asyncio.create_task(eventually_clear())
+
+        return msg
+
+    @api_request(reply_types=[ProtocolMessageTypes.respond_unfinished_block])
+    async def request_unfinished_block2(
+        self, request_unfinished_block: full_node_protocol.RequestUnfinishedBlock2
+    ) -> Optional[Message]:
+        unfinished_block: Optional[UnfinishedBlock]
+        unfinished_block, _, _ = self.full_node.full_node_store.get_unfinished_block2(
+            request_unfinished_block.unfinished_reward_hash,
+            request_unfinished_block.foliage_hash,
+        )
+        if unfinished_block is not None:
+            msg = make_msg(
+                ProtocolMessageTypes.respond_unfinished_block,
+                full_node_protocol.RespondUnfinishedBlock(unfinished_block),
+            )
+            return msg
+        return None
+
     @api_request(peer_required=True, bytes_required=True)
     async def respond_unfinished_block(
         self,
diff --git a/chia/full_node/full_node_store.py b/chia/full_node/full_node_store.py
index 0fe6950afb11..10c280cb8e48 100644
--- a/chia/full_node/full_node_store.py
+++ b/chia/full_node/full_node_store.py
@@ -38,6 +38,35 @@ class FullNodeStorePeakResult(Streamable):
     new_infusion_points: List[timelord_protocol.NewInfusionPointVDF]
 
 
+@dataclasses.dataclass
+class UnfinishedBlockEntry:
+    unfinished_block: UnfinishedBlock
+    result: PreValidationResult
+    height: uint32
+
+
+def find_best_block(
+    result: Dict[Optional[bytes32], UnfinishedBlockEntry]
+) -> Tuple[Optional[bytes32], Optional[UnfinishedBlock]]:
+    """
+    Given a collection of UnfinishedBlocks (all with the same reward block
+    hash), return the "best" one. i.e. the one with the smallest foliage hash.
+    """
+    if len(result) == 0:
+        return None, None
+
+    all_blocks = list(result.items())
+    if len(all_blocks) == 1:
+        return all_blocks[0][0], all_blocks[0][1].unfinished_block
+
+    # if there are unfinished blocks with foliage (i.e. not None) we prefer
+    # those, so drop the first element
+    all_blocks = [e for e in all_blocks if e[0] is not None]
+    all_blocks = sorted(all_blocks)
+
+    return all_blocks[0][0], all_blocks[0][1].unfinished_block
+
+
 class FullNodeStore:
     constants: ConsensusConstants
 
@@ -52,7 +81,10 @@ class FullNodeStore:
     seen_unfinished_blocks: Dict[bytes32, None]
 
     # Unfinished blocks, keyed from reward hash
-    unfinished_blocks: Dict[bytes32, Tuple[uint32, UnfinishedBlock, PreValidationResult]]
+    # There may be multiple different unfinished blocks with the same partial
+    # hash (reward chain block hash). They are stored under their partial hash
+    # though. The inner dictionary uses the foliage hash as the key
+    unfinished_blocks: Dict[bytes32, Dict[Optional[bytes32], UnfinishedBlockEntry]]
 
     # Finished slots and sps from the peak's slot onwards
     # We store all 32 SPs for each slot, starting as 32 Nones and filling them as we go
@@ -82,6 +114,10 @@ class FullNodeStore:
     # Partial hashes of unfinished blocks we are requesting
     requesting_unfinished_blocks: Set[bytes32]
 
+    # with the updated protocol for UnfinishedBlocks, when we request a block
+    # with a specific foliage hash, we add the outstanding request to this dict
+    requesting_unfinished_blocks2: Dict[bytes32, Set[Optional[bytes32]]]
+
     previous_generator: Optional[CompressorArg]
     pending_tx_request: Dict[bytes32, bytes32]  # tx_id: peer_id
     peers_with_tx: Dict[bytes32, Set[bytes32]]  # tx_id: Set[peer_ids}
@@ -103,6 +139,7 @@ def __init__(self, constants: ConsensusConstants):
         self.recent_signage_points = LRUCache(500)
         self.recent_eos = LRUCache(50)
         self.requesting_unfinished_blocks = set()
+        self.requesting_unfinished_blocks2 = {}
         self.previous_generator = None
         self.future_cache_key_times = {}
         self.constants = constants
@@ -115,6 +152,22 @@ def __init__(self, constants: ConsensusConstants):
         self.serialized_wp_message_tip = None
         self.max_seen_unfinished_blocks = 1000
 
+    def is_requesting_unfinished_block(self, reward_block_hash: bytes32, foliage_hash: Optional[bytes32]) -> bool:
+        ents = self.requesting_unfinished_blocks2.get(reward_block_hash)
+        return ents is not None and foliage_hash in ents
+
+    def mark_requesting_unfinished_block(self, reward_block_hash: bytes32, foliage_hash: Optional[bytes32]) -> None:
+        ents = self.requesting_unfinished_blocks2.setdefault(reward_block_hash, set())
+        ents.add(foliage_hash)
+
+    def remove_requesting_unfinished_block(self, reward_block_hash: bytes32, foliage_hash: Optional[bytes32]) -> None:
+        ents = self.requesting_unfinished_blocks2.get(reward_block_hash)
+        if ents is None:
+            return
+        ents.discard(foliage_hash)
+        if len(ents) == 0:
+            del self.requesting_unfinished_blocks2[reward_block_hash]
+
     def add_candidate_block(
         self, quality_string: bytes32, height: uint32, unfinished_block: UnfinishedBlock, backup: bool = False
     ) -> None:
@@ -164,32 +217,99 @@ def seen_unfinished_block(self, object_hash: bytes32) -> bool:
     def add_unfinished_block(
         self, height: uint32, unfinished_block: UnfinishedBlock, result: PreValidationResult
     ) -> None:
-        self.unfinished_blocks[unfinished_block.partial_hash] = (height, unfinished_block, result)
+        partial_hash = unfinished_block.partial_hash
+        entry = self.unfinished_blocks.setdefault(partial_hash, {})
+        entry[unfinished_block.foliage.foliage_transaction_block_hash] = UnfinishedBlockEntry(
+            unfinished_block, result, height
+        )
 
     def get_unfinished_block(self, unfinished_reward_hash: bytes32) -> Optional[UnfinishedBlock]:
         result = self.unfinished_blocks.get(unfinished_reward_hash, None)
         if result is None:
             return None
-        return result[1]
+        # The old API doesn't distinguish between duplicate UnfinishedBlocks,
+        # return the *best* UnfinishedBlock. This is the path taken when the
+        # timelord sends us an infusion point with this specific reward block
+        # hash. We pick one of the unfinished blocks based on an arbitrary but
+        # deterministic property.
+        # this sorts the UnfinishedBlocks by the foliage hash, and picks the
+        # smallest hash
+        foliage_hash, block = find_best_block(result)
+        return block
+
+    def get_unfinished_block2(
+        self, unfinished_reward_hash: bytes32, unfinished_foliage_hash: Optional[bytes32]
+    ) -> Tuple[Optional[UnfinishedBlock], int, bool]:
+        """
+        Looks up an UnfinishedBlock by its reward block hash and foliage hash.
+        If the foliage hash is None (e.g. it's not a transaction block), we fall
+        back to the original function that looks up unfinished blocks just by
+        their reward block hash.
+        Returns:
+            1. the (optional) UnfinishedBlock
+            2. the number of other candidate blocks we know of with the same
+               reward block hash
+            3. whether we already have a "better" UnfinishedBlock candidate than
+               this
+        """
+        result = self.unfinished_blocks.get(unfinished_reward_hash, None)
+        if result is None:
+            return None, 0, False
+        if unfinished_foliage_hash is None:
+            return self.get_unfinished_block(unfinished_reward_hash), len(result), False
+
+        foliage_hash, block = find_best_block(result)
+        has_better: bool = foliage_hash is not None and foliage_hash < unfinished_foliage_hash
+
+        entry = result.get(unfinished_foliage_hash)
+
+        if entry is None:
+            return None, len(result), has_better
+        else:
+            return entry.unfinished_block, len(result), has_better
 
     def get_unfinished_block_result(self, unfinished_reward_hash: bytes32) -> Optional[PreValidationResult]:
         result = self.unfinished_blocks.get(unfinished_reward_hash, None)
         if result is None:
             return None
-        return result[2]
+        return next(iter(result.values())).result
+
+    def get_unfinished_block_result2(
+        self, unfinished_reward_hash: bytes32, unfinished_foliage_hash: Optional[bytes32]
+    ) -> Optional[PreValidationResult]:
+        result = self.unfinished_blocks.get(unfinished_reward_hash, None)
+        if result is None:
+            return None
+        if unfinished_foliage_hash is None:
+            return next(iter(result.values())).result
+        else:
+            entry = result.get(unfinished_foliage_hash)
+            return None if entry is None else entry.result
 
     # returns all unfinished blocks for the specified height
     def get_unfinished_blocks(self, height: uint32) -> List[UnfinishedBlock]:
-        return [block for ub_height, block, _ in self.unfinished_blocks.values() if ub_height == height]
+        ret: List[UnfinishedBlock] = []
+        for entry in self.unfinished_blocks.values():
+            for ube in entry.values():
+                if ube.height == height:
+                    ret.append(ube.unfinished_block)
+        return ret
 
     def clear_unfinished_blocks_below(self, height: uint32) -> None:
-        del_keys: List[bytes32] = []
-        for partial_reward_hash, (unf_height, unfinished_block, _) in self.unfinished_blocks.items():
-            if unf_height < height:
-                del_keys.append(partial_reward_hash)
-        for del_key in del_keys:
-            del self.unfinished_blocks[del_key]
-
+        del_partial: List[bytes32] = []
+        for partial_hash, entry in self.unfinished_blocks.items():
+            del_foliage: List[Optional[bytes32]] = []
+            for foliage_hash, ube in entry.items():
+                if ube.height < height:
+                    del_foliage.append(foliage_hash)
+            for fh in del_foliage:
+                del entry[fh]
+            if len(entry) == 0:
+                del_partial.append(partial_hash)
+        for ph in del_partial:
+            del self.unfinished_blocks[ph]
+
+    # TODO: this should be removed. It's only used by a test
     def remove_unfinished_block(self, partial_reward_hash: bytes32) -> None:
         if partial_reward_hash in self.unfinished_blocks:
             del self.unfinished_blocks[partial_reward_hash]
diff --git a/chia/protocols/full_node_protocol.py b/chia/protocols/full_node_protocol.py
index c9baf051f03c..4a3607dfcb81 100644
--- a/chia/protocols/full_node_protocol.py
+++ b/chia/protocols/full_node_protocol.py
@@ -118,6 +118,20 @@ class RequestUnfinishedBlock(Streamable):
     unfinished_reward_hash: bytes32
 
 
+@streamable
+@dataclass(frozen=True)
+class NewUnfinishedBlock2(Streamable):
+    unfinished_reward_hash: bytes32
+    foliage_hash: Optional[bytes32]
+
+
+@streamable
+@dataclass(frozen=True)
+class RequestUnfinishedBlock2(Streamable):
+    unfinished_reward_hash: bytes32
+    foliage_hash: Optional[bytes32]
+
+
 @streamable
 @dataclass(frozen=True)
 class RespondUnfinishedBlock(Streamable):
diff --git a/chia/protocols/protocol_message_types.py b/chia/protocols/protocol_message_types.py
index 06aef536b50c..4be6c4af7321 100644
--- a/chia/protocols/protocol_message_types.py
+++ b/chia/protocols/protocol_message_types.py
@@ -114,4 +114,8 @@ class ProtocolMessageTypes(Enum):
     request_fee_estimates = 89
     respond_fee_estimates = 90
 
+    # new Full Node protocol messages
+    new_unfinished_block2 = 92
+    request_unfinished_block2 = 93
+
     error = 255
diff --git a/chia/protocols/protocol_state_machine.py b/chia/protocols/protocol_state_machine.py
index 09893ce8b2bb..482e7a59eb1d 100644
--- a/chia/protocols/protocol_state_machine.py
+++ b/chia/protocols/protocol_state_machine.py
@@ -8,6 +8,7 @@
     pmt.new_peak,
     pmt.new_transaction,
     pmt.new_unfinished_block,
+    pmt.new_unfinished_block2,
     pmt.new_signage_point_or_end_of_sub_slot,
     pmt.request_mempool_transactions,
     pmt.new_compact_vdf,
@@ -30,6 +31,7 @@
     pmt.request_block: [pmt.respond_block, pmt.reject_block],
     pmt.request_blocks: [pmt.respond_blocks, pmt.reject_blocks],
     pmt.request_unfinished_block: [pmt.respond_unfinished_block],
+    pmt.request_unfinished_block2: [pmt.respond_unfinished_block],
     pmt.request_block_header: [pmt.respond_block_header, pmt.reject_header_request],
     pmt.request_removals: [pmt.respond_removals, pmt.reject_removals_request],
     pmt.request_additions: [pmt.respond_additions, pmt.reject_additions_request],
diff --git a/chia/protocols/shared_protocol.py b/chia/protocols/shared_protocol.py
index 3616ae827639..36d495d65541 100644
--- a/chia/protocols/shared_protocol.py
+++ b/chia/protocols/shared_protocol.py
@@ -7,7 +7,7 @@
 from chia.util.ints import int16, uint8, uint16
 from chia.util.streamable import Streamable, streamable
 
-protocol_version = "0.0.35"
+protocol_version = "0.0.36"
 
 
 """
diff --git a/chia/server/rate_limit_numbers.py b/chia/server/rate_limit_numbers.py
index c448ff546188..1746b6b44048 100644
--- a/chia/server/rate_limit_numbers.py
+++ b/chia/server/rate_limit_numbers.py
@@ -101,6 +101,8 @@ def compose_rate_limits(old_rate_limits: Dict[str, Any], new_rate_limits: Dict[s
             ProtocolMessageTypes.respond_block: RLSettings(200, 2 * 1024 * 1024, 10 * 2 * 1024 * 1024),
             ProtocolMessageTypes.new_unfinished_block: RLSettings(200, 100),
             ProtocolMessageTypes.request_unfinished_block: RLSettings(200, 100),
+            ProtocolMessageTypes.new_unfinished_block2: RLSettings(200, 100),
+            ProtocolMessageTypes.request_unfinished_block2: RLSettings(200, 100),
             ProtocolMessageTypes.respond_unfinished_block: RLSettings(200, 2 * 1024 * 1024, 10 * 2 * 1024 * 1024),
             ProtocolMessageTypes.new_signage_point_or_end_of_sub_slot: RLSettings(200, 200),
             ProtocolMessageTypes.request_signage_point_or_end_of_sub_slot: RLSettings(200, 200),
diff --git a/chia/server/server.py b/chia/server/server.py
index f597fa725892..fe70ee355d7d 100644
--- a/chia/server/server.py
+++ b/chia/server/server.py
@@ -585,6 +585,19 @@ async def send_to_all(
                 for message in messages:
                     await connection.send_message(message)
 
+    async def send_to_all_if(
+        self,
+        messages: List[Message],
+        node_type: NodeType,
+        predicate: Callable[[WSChiaConnection], bool],
+        exclude: Optional[bytes32] = None,
+    ) -> None:
+        await self.validate_broadcast_message_type(messages, node_type)
+        for _, connection in self.all_connections.items():
+            if connection.connection_type is node_type and connection.peer_node_id != exclude and predicate(connection):
+                for message in messages:
+                    await connection.send_message(message)
+
     async def send_to_specific(self, messages: List[Message], node_id: bytes32) -> None:
         if node_id in self.all_connections:
             connection = self.all_connections[node_id]
diff --git a/chia/simulator/block_tools.py b/chia/simulator/block_tools.py
index a2f092642dbe..a61db6ec15a3 100644
--- a/chia/simulator/block_tools.py
+++ b/chia/simulator/block_tools.py
@@ -2050,8 +2050,10 @@ def create_block_tools(
     return bt
 
 
-def make_unfinished_block(block: FullBlock, constants: ConsensusConstants) -> UnfinishedBlock:
-    if is_overflow_block(constants, uint8(block.reward_chain_block.signage_point_index)):
+def make_unfinished_block(
+    block: FullBlock, constants: ConsensusConstants, *, force_overflow: bool = False
+) -> UnfinishedBlock:
+    if force_overflow or is_overflow_block(constants, uint8(block.reward_chain_block.signage_point_index)):
         finished_ss = block.finished_sub_slots[:-1]
     else:
         finished_ss = block.finished_sub_slots
diff --git a/chia/util/initial-config.yaml b/chia/util/initial-config.yaml
index fa351891832e..e86b3840c7a0 100644
--- a/chia/util/initial-config.yaml
+++ b/chia/util/initial-config.yaml
@@ -379,6 +379,14 @@ full_node:
 
   multiprocessing_start_method: default
 
+  # The maximum number of UnfinishedBlocks we accept (and forward) with the
+  # same reward hash (but different foliage hashes). Traditionally this was
+  # effectively 1, meaning whichever UnfinishedBlock we saw first was the only
+  # one we forwarded. In 2.2.0 we relaxed the protocol to allow some
+  # duplicates be forwarded, in order to allow the timelords to,
+  # deterministically, pick which one to infuse
+  max_duplicate_unfinished_blocks: 3
+
   # If True, starts an RPC server at the following port
   start_rpc_server: True
   rpc_port: 8555
diff --git a/install-timelord.sh b/install-timelord.sh
old mode 100644
new mode 100755
diff --git a/tests/core/full_node/stores/test_full_node_store.py b/tests/core/full_node/stores/test_full_node_store.py
index 010e18319ad9..62a3eddd7bc9 100644
--- a/tests/core/full_node/stores/test_full_node_store.py
+++ b/tests/core/full_node/stores/test_full_node_store.py
@@ -3,21 +3,22 @@
 import dataclasses
 import logging
 import random
-from typing import AsyncIterator, List, Optional
+from typing import AsyncIterator, Dict, List, Optional
 
 import pytest
 
 from chia.consensus.blockchain import AddBlockResult, Blockchain
 from chia.consensus.constants import ConsensusConstants
+from chia.consensus.default_constants import DEFAULT_CONSTANTS
 from chia.consensus.difficulty_adjustment import get_next_sub_slot_iters_and_difficulty
 from chia.consensus.find_fork_point import find_fork_point_in_chain
 from chia.consensus.multiprocess_validation import PreValidationResult
 from chia.consensus.pot_iterations import is_overflow_block
-from chia.full_node.full_node_store import FullNodeStore
+from chia.full_node.full_node_store import FullNodeStore, UnfinishedBlockEntry, find_best_block
 from chia.full_node.signage_point import SignagePoint
 from chia.protocols import timelord_protocol
 from chia.protocols.timelord_protocol import NewInfusionPointVDF
-from chia.simulator.block_tools import BlockTools, create_block_tools_async, get_signage_point
+from chia.simulator.block_tools import BlockTools, create_block_tools_async, get_signage_point, make_unfinished_block
 from chia.simulator.keyring import TempKeyring
 from chia.types.blockchain_format.sized_bytes import bytes32
 from chia.types.full_block import FullBlock
@@ -25,6 +26,7 @@
 from chia.util.block_cache import BlockCache
 from chia.util.hash import std_hash
 from chia.util.ints import uint8, uint32, uint64, uint128
+from chia.util.recursive_replace import recursive_replace
 from tests.blockchain.blockchain_test_utils import _validate_and_add_block, _validate_and_add_block_no_error
 from tests.util.blockchain import create_blockchain
 
@@ -61,6 +63,89 @@ async def empty_blockchain_with_original_constants(
         yield bc1
 
 
+@pytest.mark.anyio
+@pytest.mark.parametrize("num_duplicates", [0, 1, 3, 10])
+@pytest.mark.parametrize("include_none", [True, False])
+async def test_unfinished_block_rank(
+    empty_blockchain: Blockchain,
+    custom_block_tools: BlockTools,
+    seeded_random: random.Random,
+    num_duplicates: int,
+    include_none: bool,
+) -> None:
+    blocks = custom_block_tools.get_consecutive_blocks(
+        1,
+        guarantee_transaction_block=True,
+    )
+
+    assert blocks[-1].is_transaction_block()
+    store = FullNodeStore(custom_block_tools.constants)
+    unf: UnfinishedBlock = make_unfinished_block(blocks[-1], custom_block_tools.constants)
+
+    # create variants of the unfinished block, where all we do is to change
+    # the foliage_transaction_block_hash. As if they all had different foliage,
+    # but the same reward block hash (i.e. the same proof-of-space)
+    unfinished: List[UnfinishedBlock] = [
+        recursive_replace(unf, "foliage.foliage_transaction_block_hash", bytes32([idx + 4] * 32))
+        for idx in range(num_duplicates)
+    ]
+
+    if include_none:
+        unfinished.append(recursive_replace(unf, "foliage.foliage_transaction_block_hash", None))
+
+    # shuffle them to ensure the order we add them to the store isn't relevant
+    seeded_random.shuffle(unfinished)
+    for new_unf in unfinished:
+        store.add_unfinished_block(
+            uint32(2), new_unf, PreValidationResult(None, uint64(123532), None, False, uint32(0))
+        )
+
+    # now ask for "the" unfinished block given the proof-of-space.
+    # the FullNodeStore should return the one with the lowest foliage tx block
+    # hash. We prefer a block with foliage over one without (i.e. where foliage
+    # is None)
+    if num_duplicates == 0 and not include_none:
+        assert store.get_unfinished_block(unf.partial_hash) is None
+    else:
+        best_unf = store.get_unfinished_block(unf.partial_hash)
+        assert best_unf is not None
+        if num_duplicates == 0:
+            # if a block without foliage is our only option, that's what we get
+            assert best_unf.foliage.foliage_transaction_block_hash is None
+        else:
+            assert best_unf.foliage.foliage_transaction_block_hash == bytes32([4] * 32)
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    "blocks,expected",
+    [
+        ([None, 1, 2, 3], 1),
+        ([None], None),
+        ([], None),
+        ([4, 5, 3], 3),
+        ([4], 4),
+    ],
+)
+async def test_find_best_block(
+    seeded_random: random.Random,
+    blocks: List[Optional[int]],
+    expected: Optional[int],
+) -> None:
+    result: Dict[Optional[bytes32], UnfinishedBlockEntry] = {}
+    for b in blocks:
+        if b is None:
+            result[b] = UnfinishedBlockEntry(None, None, 123)  # type: ignore
+        else:
+            result[bytes32(b.to_bytes(1, "big") * 32)] = UnfinishedBlockEntry(None, None, 123)  # type: ignore
+
+    foliage_hash, block = find_best_block(result)
+    if expected is None:
+        assert foliage_hash is None
+    else:
+        assert foliage_hash == bytes32(expected.to_bytes(1, "big") * 32)
+
+
 @pytest.mark.limit_consensus_modes(reason="save time")
 @pytest.mark.anyio
 @pytest.mark.parametrize("normalized_to_identity", [False, True])
@@ -122,12 +207,99 @@ async def test_basic_store(
     # Add/get unfinished block
     for height, unf_block in enumerate(unfinished_blocks):
         assert store.get_unfinished_block(unf_block.partial_hash) is None
+        assert store.get_unfinished_block2(unf_block.partial_hash, None) == (None, 0, False)
         store.add_unfinished_block(
             uint32(height), unf_block, PreValidationResult(None, uint64(123532), None, False, uint32(0))
         )
         assert store.get_unfinished_block(unf_block.partial_hash) == unf_block
+        assert store.get_unfinished_block2(
+            unf_block.partial_hash, unf_block.foliage.foliage_transaction_block_hash
+        ) == (unf_block, 1, False)
+
+        foliage_hash = unf_block.foliage.foliage_transaction_block_hash
+        dummy_hash = bytes32.fromhex("abababababababababababababababababababababababababababababababab")
+        assert store.get_unfinished_block2(unf_block.partial_hash, dummy_hash) == (
+            None,
+            1,
+            foliage_hash is not None and dummy_hash > foliage_hash,
+        )
+
+        ublock = store.get_unfinished_block_result(unf_block.partial_hash)
+        assert ublock is not None and ublock.required_iters == uint64(123532)
+        ublock = store.get_unfinished_block_result2(
+            unf_block.partial_hash, unf_block.foliage.foliage_transaction_block_hash
+        )
+
+        assert ublock is not None and ublock.required_iters == uint64(123532)
+
         store.remove_unfinished_block(unf_block.partial_hash)
         assert store.get_unfinished_block(unf_block.partial_hash) is None
+        assert store.get_unfinished_block2(
+            unf_block.partial_hash, unf_block.foliage.foliage_transaction_block_hash
+        ) == (None, 0, False)
+
+    # Multiple unfinished blocks with colliding partial hashes
+    unf1 = unfinished_blocks[0]
+    unf2 = dataclasses.replace(unf1, foliage=unfinished_blocks[1].foliage)
+    unf3 = dataclasses.replace(unf1, foliage=unfinished_blocks[2].foliage)
+    unf4 = dataclasses.replace(unf1, foliage=unfinished_blocks[3].foliage)
+
+    # we have none of these blocks in the store
+    for unf_block in [unf1, unf2, unf3, unf4]:
+        assert store.get_unfinished_block(unf_block.partial_hash) is None
+        assert store.get_unfinished_block2(unf_block.partial_hash, None) == (None, 0, False)
+
+    height = uint32(1)
+    # all blocks without a foliage all collapse down into being the same
+    assert unf1.foliage.foliage_transaction_block_hash is not None
+    assert unf2.foliage.foliage_transaction_block_hash is None
+    assert unf3.foliage.foliage_transaction_block_hash is None
+    assert unf4.foliage.foliage_transaction_block_hash is None
+    for val, unf_block in enumerate([unf1, unf2, unf3, unf4]):
+        store.add_unfinished_block(
+            uint32(height), unf_block, PreValidationResult(None, uint64(val), None, False, uint32(0))
+        )
+
+    # when not specifying a foliage hash, you get the "best" one
+    # best is defined as the lowest foliage hash
+    assert store.get_unfinished_block(unf1.partial_hash) == unf1
+    assert store.get_unfinished_block2(unf1.partial_hash, unf1.foliage.foliage_transaction_block_hash) == (
+        unf1,
+        2,
+        False,
+    )
+    # unf4 overwrote unf2 and unf3 (that's why there are only 2 blocks stored).
+    # however, there's no way to explicitly request the block with None foliage
+    # since when specifying None, you always get the first one. unf1 in this
+    # case
+    assert store.get_unfinished_block2(unf2.partial_hash, unf2.foliage.foliage_transaction_block_hash) == (
+        unf1,
+        2,
+        False,
+    )
+    assert store.get_unfinished_block2(unf3.partial_hash, unf3.foliage.foliage_transaction_block_hash) == (
+        unf1,
+        2,
+        False,
+    )
+    assert store.get_unfinished_block2(unf4.partial_hash, unf4.foliage.foliage_transaction_block_hash) == (
+        unf1,
+        2,
+        False,
+    )
+    assert store.get_unfinished_block2(unf4.partial_hash, None) == (unf1, 2, False)
+
+    ublock = store.get_unfinished_block_result(unf1.partial_hash)
+    assert ublock is not None and ublock.required_iters == uint64(0)
+    ublock = store.get_unfinished_block_result2(unf1.partial_hash, unf1.foliage.foliage_transaction_block_hash)
+    assert ublock is not None and ublock.required_iters == uint64(0)
+    # still, when not specifying a foliage hash, you just get the first ublock
+    ublock = store.get_unfinished_block_result2(unf1.partial_hash, None)
+    assert ublock is not None and ublock.required_iters == uint64(0)
+
+    # negative test cases
+    assert store.get_unfinished_block_result(bytes32([1] * 32)) is None
+    assert store.get_unfinished_block_result2(bytes32([1] * 32), None) is None
 
     blocks = custom_block_tools.get_consecutive_blocks(
         1,
@@ -958,3 +1130,61 @@ async def test_long_chain_slots(
         store.new_peak(
             peak, peak_full_block, sp_sub_slot, ip_sub_slot, None, blockchain, next_sub_slot_iters, next_difficulty
         )
+
+
+@pytest.mark.anyio
+async def test_mark_requesting(
+    seeded_random: random.Random,
+) -> None:
+    store = FullNodeStore(DEFAULT_CONSTANTS)
+    a = bytes32.random(seeded_random)
+    b = bytes32.random(seeded_random)
+    c = bytes32.random(seeded_random)
+
+    assert not store.is_requesting_unfinished_block(a, a)
+    assert not store.is_requesting_unfinished_block(a, b)
+    assert not store.is_requesting_unfinished_block(a, c)
+    assert not store.is_requesting_unfinished_block(b, b)
+    assert not store.is_requesting_unfinished_block(c, c)
+
+    store.mark_requesting_unfinished_block(a, b)
+    assert store.is_requesting_unfinished_block(a, b)
+    assert not store.is_requesting_unfinished_block(a, c)
+    assert not store.is_requesting_unfinished_block(a, a)
+    assert not store.is_requesting_unfinished_block(b, c)
+    assert not store.is_requesting_unfinished_block(b, b)
+
+    store.mark_requesting_unfinished_block(a, c)
+    assert store.is_requesting_unfinished_block(a, b)
+    assert store.is_requesting_unfinished_block(a, c)
+    assert not store.is_requesting_unfinished_block(a, a)
+    assert not store.is_requesting_unfinished_block(b, c)
+    assert not store.is_requesting_unfinished_block(b, b)
+
+    # this is a no-op
+    store.remove_requesting_unfinished_block(a, a)
+    store.remove_requesting_unfinished_block(c, a)
+
+    assert store.is_requesting_unfinished_block(a, b)
+    assert store.is_requesting_unfinished_block(a, c)
+    assert not store.is_requesting_unfinished_block(a, a)
+    assert not store.is_requesting_unfinished_block(b, c)
+    assert not store.is_requesting_unfinished_block(b, b)
+
+    store.remove_requesting_unfinished_block(a, b)
+
+    assert not store.is_requesting_unfinished_block(a, b)
+    assert store.is_requesting_unfinished_block(a, c)
+    assert not store.is_requesting_unfinished_block(a, a)
+    assert not store.is_requesting_unfinished_block(b, c)
+    assert not store.is_requesting_unfinished_block(b, b)
+
+    store.remove_requesting_unfinished_block(a, c)
+
+    assert not store.is_requesting_unfinished_block(a, b)
+    assert not store.is_requesting_unfinished_block(a, c)
+    assert not store.is_requesting_unfinished_block(a, a)
+    assert not store.is_requesting_unfinished_block(b, c)
+    assert not store.is_requesting_unfinished_block(b, b)
+
+    assert len(store.requesting_unfinished_blocks) == 0
diff --git a/tests/core/full_node/test_full_node.py b/tests/core/full_node/test_full_node.py
index 21afc75c73d0..92d05531d16e 100644
--- a/tests/core/full_node/test_full_node.py
+++ b/tests/core/full_node/test_full_node.py
@@ -11,6 +11,7 @@
 import pytest
 from chia_rs import AugSchemeMPL, G2Element, PrivateKey
 from clvm.casts import int_to_bytes
+from packaging.version import Version
 
 from chia.consensus.block_body_validation import ForkInfo
 from chia.consensus.pot_iterations import is_overflow_block
@@ -29,7 +30,7 @@
 from chia.server.address_manager import AddressManager
 from chia.server.outbound_message import Message, NodeType
 from chia.server.server import ChiaServer
-from chia.simulator.block_tools import BlockTools, create_block_tools_async, get_signage_point
+from chia.simulator.block_tools import BlockTools, create_block_tools_async, get_signage_point, make_unfinished_block
 from chia.simulator.full_node_simulator import FullNodeSimulator
 from chia.simulator.keyring import TempKeyring
 from chia.simulator.setup_services import setup_full_node
@@ -678,22 +679,8 @@ async def test_respond_unfinished(self, wallet_nodes, self_hostname):
         # Create empty slots
         blocks = bt.get_consecutive_blocks(1, block_list_input=blocks, skip_slots=6)
         block = blocks[-1]
-        if is_overflow_block(bt.constants, block.reward_chain_block.signage_point_index):
-            finished_ss = block.finished_sub_slots[:-1]
-        else:
-            finished_ss = block.finished_sub_slots
+        unf = make_unfinished_block(block, bt.constants)
 
-        unf = UnfinishedBlock(
-            finished_ss,
-            block.reward_chain_block.get_unfinished(),
-            block.challenge_chain_sp_proof,
-            block.reward_chain_sp_proof,
-            block.foliage,
-            block.foliage_transaction_block,
-            block.transactions_info,
-            block.transactions_generator,
-            [],
-        )
         # Can't add because no sub slots
         assert full_node_1.full_node.full_node_store.get_unfinished_block(unf.partial_hash) is None
 
@@ -709,22 +696,7 @@ async def test_respond_unfinished(self, wallet_nodes, self_hostname):
         blocks = bt.get_consecutive_blocks(1, block_list_input=blocks, skip_slots=3)
 
         block = blocks[-1]
-
-        if is_overflow_block(bt.constants, block.reward_chain_block.signage_point_index):
-            finished_ss = block.finished_sub_slots[:-1]
-        else:
-            finished_ss = block.finished_sub_slots
-        unf = UnfinishedBlock(
-            finished_ss,
-            block.reward_chain_block.get_unfinished(),
-            block.challenge_chain_sp_proof,
-            block.reward_chain_sp_proof,
-            block.foliage,
-            block.foliage_transaction_block,
-            block.transactions_info,
-            block.transactions_generator,
-            [],
-        )
+        unf = make_unfinished_block(block, bt.constants)
         assert full_node_1.full_node.full_node_store.get_unfinished_block(unf.partial_hash) is None
 
         for slot in blocks[-1].finished_sub_slots:
@@ -738,18 +710,7 @@ async def test_respond_unfinished(self, wallet_nodes, self_hostname):
         blocks = bt.get_consecutive_blocks(1, block_list_input=blocks, skip_slots=3, force_overflow=True)
 
         block = blocks[-1]
-
-        unf = UnfinishedBlock(
-            block.finished_sub_slots[:-1],
-            block.reward_chain_block.get_unfinished(),
-            block.challenge_chain_sp_proof,
-            block.reward_chain_sp_proof,
-            block.foliage,
-            block.foliage_transaction_block,
-            block.transactions_info,
-            block.transactions_generator,
-            [],
-        )
+        unf = make_unfinished_block(block, bt.constants, force_overflow=True)
         assert full_node_1.full_node.full_node_store.get_unfinished_block(unf.partial_hash) is None
 
         for slot in blocks[-1].finished_sub_slots:
@@ -784,17 +745,7 @@ async def test_respond_unfinished(self, wallet_nodes, self_hostname):
             seed=b"random seed",
         )
         block = blocks[-1]
-        unf = UnfinishedBlock(
-            block.finished_sub_slots[:-1],  # Since it's overflow
-            block.reward_chain_block.get_unfinished(),
-            block.challenge_chain_sp_proof,
-            block.reward_chain_sp_proof,
-            block.foliage,
-            block.foliage_transaction_block,
-            block.transactions_info,
-            block.transactions_generator,
-            [],
-        )
+        unf = make_unfinished_block(block, bt.constants, force_overflow=True)
         assert full_node_1.full_node.full_node_store.get_unfinished_block(unf.partial_hash) is None
         await full_node_1.full_node.add_unfinished_block(unf, None)
         assert full_node_1.full_node.full_node_store.get_unfinished_block(unf.partial_hash) is not None
@@ -1253,36 +1204,162 @@ async def test_request_blocks(self, wallet_nodes):
         assert std_hash(fetched_blocks[-1]) == std_hash(blocks_t[-1])
 
     @pytest.mark.anyio
-    async def test_new_unfinished_block(self, wallet_nodes, self_hostname):
+    @pytest.mark.parametrize("peer_version", ["0.0.35", "0.0.36"])
+    @pytest.mark.parametrize("requesting", [0, 1, 2])
+    async def test_new_unfinished_block(self, wallet_nodes, peer_version: str, requesting: int, self_hostname: str):
         full_node_1, full_node_2, server_1, server_2, wallet_a, wallet_receiver, bt = wallet_nodes
         blocks = await full_node_1.get_all_full_blocks()
 
         peer = await connect_and_get_peer(server_1, server_2, self_hostname)
+        assert peer in server_1.all_connections.values()
 
-        blocks = bt.get_consecutive_blocks(1, block_list_input=blocks)
+        blocks = bt.get_consecutive_blocks(2, block_list_input=blocks)
         block: FullBlock = blocks[-1]
-        overflow = is_overflow_block(bt.constants, block.reward_chain_block.signage_point_index)
-        unf = UnfinishedBlock(
-            block.finished_sub_slots[:] if not overflow else block.finished_sub_slots[:-1],
-            block.reward_chain_block.get_unfinished(),
-            block.challenge_chain_sp_proof,
-            block.reward_chain_sp_proof,
-            block.foliage,
-            block.foliage_transaction_block,
-            block.transactions_info,
-            block.transactions_generator,
-            [],
-        )
+        unf = make_unfinished_block(block, bt.constants)
 
         # Don't have
+        if requesting == 1:
+            full_node_1.full_node.full_node_store.requesting_unfinished_blocks.add(unf.partial_hash)
+            res = await full_node_1.new_unfinished_block(fnp.NewUnfinishedBlock(unf.partial_hash))
+            assert res is None
+        elif requesting == 2:
+            full_node_1.full_node.full_node_store.requesting_unfinished_blocks2.setdefault(unf.partial_hash, set()).add(
+                unf.foliage.foliage_transaction_block_hash
+            )
+            res = await full_node_1.new_unfinished_block(fnp.NewUnfinishedBlock(unf.partial_hash))
+            assert res is None
+        else:
+            res = await full_node_1.new_unfinished_block(fnp.NewUnfinishedBlock(unf.partial_hash))
+            assert res is not None
+            assert res is not None and res.data == bytes(fnp.RequestUnfinishedBlock(unf.partial_hash))
+
+        # when we receive a new unfinished block, we advertize it to our peers.
+        # We send new_unfinished_blocks to old peers (0.0.35 and earlier) and we
+        # send new_unfinishe_blocks2 to new peers (0.0.6 and later). Test both
+        peer.protocol_version = Version(peer_version)
+
+        await full_node_1.full_node.add_block(blocks[-2])
+        await full_node_1.full_node.add_unfinished_block(unf, None)
+
+        msg = peer.outgoing_queue.get_nowait()
+        assert msg.type == ProtocolMessageTypes.new_peak.value
+        msg = peer.outgoing_queue.get_nowait()
+        if peer_version == "0.0.35":
+            assert msg.type == ProtocolMessageTypes.new_unfinished_block.value
+            assert msg.data == bytes(fnp.NewUnfinishedBlock(unf.partial_hash))
+        elif peer_version == "0.0.36":
+            assert msg.type == ProtocolMessageTypes.new_unfinished_block2.value
+            assert msg.data == bytes(
+                fnp.NewUnfinishedBlock2(unf.partial_hash, unf.foliage.foliage_transaction_block_hash)
+            )
+        else:  # pragma: no cover
+            # the test parameters must have been updated, update the test too!
+            assert False
+
+        # Have
         res = await full_node_1.new_unfinished_block(fnp.NewUnfinishedBlock(unf.partial_hash))
-        assert res is not None
+        assert res is None
+
+    @pytest.mark.anyio
+    @pytest.mark.parametrize("requesting", [0, 1, 2])
+    async def test_new_unfinished_block2(self, wallet_nodes, requesting: int, self_hostname: str):
+        full_node_1, full_node_2, server_1, server_2, wallet_a, wallet_receiver, bt = wallet_nodes
+        blocks = await full_node_1.get_all_full_blocks()
+
+        peer = await connect_and_get_peer(server_1, server_2, self_hostname)
+
+        blocks = bt.get_consecutive_blocks(1, block_list_input=blocks)
+        block: FullBlock = blocks[-1]
+        unf = make_unfinished_block(block, bt.constants)
+
+        # Don't have
+        if requesting == 1:
+            full_node_1.full_node.full_node_store.requesting_unfinished_blocks.add(unf.partial_hash)
+
+        if requesting == 2:
+            full_node_1.full_node.full_node_store.requesting_unfinished_blocks2.setdefault(unf.partial_hash, set()).add(
+                unf.foliage.foliage_transaction_block_hash
+            )
+            res = await full_node_1.new_unfinished_block2(
+                fnp.NewUnfinishedBlock2(unf.partial_hash, unf.foliage.foliage_transaction_block_hash)
+            )
+            assert res is None
+        else:
+            res = await full_node_1.new_unfinished_block2(
+                fnp.NewUnfinishedBlock2(unf.partial_hash, unf.foliage.foliage_transaction_block_hash)
+            )
+            assert res is not None and res.data == bytes(
+                fnp.RequestUnfinishedBlock2(unf.partial_hash, unf.foliage.foliage_transaction_block_hash)
+            )
+
         await full_node_1.full_node.add_unfinished_block(unf, peer)
 
         # Have
-        res = await full_node_1.new_unfinished_block(fnp.NewUnfinishedBlock(unf.partial_hash))
+        res = await full_node_1.new_unfinished_block2(
+            fnp.NewUnfinishedBlock2(unf.partial_hash, unf.foliage.foliage_transaction_block_hash)
+        )
         assert res is None
 
+    @pytest.mark.anyio
+    async def test_new_unfinished_block2_forward_limit(self, wallet_nodes, self_hostname: str):
+        full_node_1, full_node_2, server_1, server_2, wallet_a, wallet_receiver, bt = wallet_nodes
+        blocks = bt.get_consecutive_blocks(3, guarantee_transaction_block=True)
+        for block in blocks:
+            await full_node_1.full_node.add_block(block)
+        coin = blocks[-1].get_included_reward_coins()[0]
+        puzzle_hash = wallet_receiver.get_new_puzzlehash()
+
+        peer = await connect_and_get_peer(server_1, server_2, self_hostname)
+
+        # notify the node of unfinished blocks for this reward block hash
+        # we forward 3 different blocks with the same reward block hash, but no
+        # more (it's configurable)
+        # also, we don't forward unfinished blocks that are "worse" than the
+        # best block we've already seen, so we may need to send more than 3
+        # blocks to the node for it to forward 3
+
+        unf_blocks: List[UnfinishedBlock] = []
+
+        last_reward_hash: Optional[bytes32] = None
+        for idx in range(0, 6):
+            # we include a different transaction in each block. This makes the
+            # foliage different in each of them, but the reward block (plot) the same
+            tx: SpendBundle = wallet_a.generate_signed_transaction(100 * (idx + 1), puzzle_hash, coin)
+
+            # note that we use the same chain to build the new block on top of every time
+            block = bt.get_consecutive_blocks(
+                1, block_list_input=blocks, guarantee_transaction_block=True, transaction_data=tx
+            )[-1]
+            unf = make_unfinished_block(block, bt.constants)
+            unf_blocks.append(unf)
+
+            if last_reward_hash is None:
+                last_reward_hash = unf.partial_hash
+            else:
+                assert last_reward_hash == unf.partial_hash
+
+        # sort the blocks from worst -> best
+        def sort_key(b: UnfinishedBlock) -> bytes32:
+            assert b.foliage.foliage_transaction_block_hash is not None
+            return b.foliage.foliage_transaction_block_hash
+
+        unf_blocks.sort(reverse=True, key=sort_key)
+
+        for idx, unf in enumerate(unf_blocks):
+            res = await full_node_1.new_unfinished_block2(
+                fnp.NewUnfinishedBlock2(unf.partial_hash, unf.foliage.foliage_transaction_block_hash)
+            )
+            # 3 is the default number of different unfinished blocks we forward
+            if idx <= 3:
+                # Don't have
+                assert res is not None and res.data == bytes(
+                    fnp.RequestUnfinishedBlock2(unf.partial_hash, unf.foliage.foliage_transaction_block_hash)
+                )
+            else:
+                # too many UnfinishedBlocks with the same reward hash
+                assert res is None
+            await full_node_1.full_node.add_unfinished_block(unf, peer)
+
     @pytest.mark.anyio
     @pytest.mark.parametrize(
         "committment,expected",
@@ -1457,18 +1534,7 @@ async def test_double_blocks_same_pospace(self, wallet_nodes, self_hostname):
         )
 
         block: FullBlock = blocks[-1]
-        overflow = is_overflow_block(bt.constants, block.reward_chain_block.signage_point_index)
-        unf: UnfinishedBlock = UnfinishedBlock(
-            block.finished_sub_slots[:] if not overflow else block.finished_sub_slots[:-1],
-            block.reward_chain_block.get_unfinished(),
-            block.challenge_chain_sp_proof,
-            block.reward_chain_sp_proof,
-            block.foliage,
-            block.foliage_transaction_block,
-            block.transactions_info,
-            block.transactions_generator,
-            [],
-        )
+        unf = make_unfinished_block(block, bt.constants)
         await full_node_1.full_node.add_unfinished_block(unf, dummy_peer)
         assert full_node_1.full_node.full_node_store.get_unfinished_block(unf.partial_hash)
 
@@ -1494,18 +1560,7 @@ async def test_request_unfinished_block(self, wallet_nodes, self_hostname):
         for block in blocks[:-1]:
             await full_node_1.full_node.add_block(block)
         block: FullBlock = blocks[-1]
-        overflow = is_overflow_block(bt.constants, block.reward_chain_block.signage_point_index)
-        unf = UnfinishedBlock(
-            block.finished_sub_slots[:] if not overflow else block.finished_sub_slots[:-1],
-            block.reward_chain_block.get_unfinished(),
-            block.challenge_chain_sp_proof,
-            block.reward_chain_sp_proof,
-            block.foliage,
-            block.foliage_transaction_block,
-            block.transactions_info,
-            block.transactions_generator,
-            [],
-        )
+        unf = make_unfinished_block(block, bt.constants)
 
         # Don't have
         res = await full_node_1.request_unfinished_block(fnp.RequestUnfinishedBlock(unf.partial_hash))
@@ -1515,6 +1570,61 @@ async def test_request_unfinished_block(self, wallet_nodes, self_hostname):
         res = await full_node_1.request_unfinished_block(fnp.RequestUnfinishedBlock(unf.partial_hash))
         assert res is not None
 
+    @pytest.mark.anyio
+    async def test_request_unfinished_block2(self, wallet_nodes, self_hostname):
+        full_node_1, full_node_2, server_1, server_2, wallet_a, wallet_receiver, bt = wallet_nodes
+        blocks = await full_node_1.get_all_full_blocks()
+        blocks = bt.get_consecutive_blocks(3, guarantee_transaction_block=True)
+        for block in blocks:
+            await full_node_1.full_node.add_block(block)
+        coin = blocks[-1].get_included_reward_coins()[0]
+        puzzle_hash = wallet_receiver.get_new_puzzlehash()
+
+        peer = await connect_and_get_peer(server_1, server_2, self_hostname)
+
+        # the "best" unfinished block according to the metric we use to pick one
+        # deterministically
+        best_unf: Optional[UnfinishedBlock] = None
+
+        for idx in range(0, 6):
+            # we include a different transaction in each block. This makes the
+            # foliage different in each of them, but the reward block (plot) the same
+            tx: SpendBundle = wallet_a.generate_signed_transaction(100 * (idx + 1), puzzle_hash, coin)
+
+            # note that we use the same chain to build the new block on top of every time
+            block = bt.get_consecutive_blocks(
+                1, block_list_input=blocks, guarantee_transaction_block=True, transaction_data=tx
+            )[-1]
+            unf = make_unfinished_block(block, bt.constants)
+            assert unf.foliage.foliage_transaction_block_hash is not None
+
+            if best_unf is None:
+                best_unf = unf
+            elif (
+                unf.foliage.foliage_transaction_block_hash is not None
+                and unf.foliage.foliage_transaction_block_hash < best_unf.foliage.foliage_transaction_block_hash
+            ):
+                best_unf = unf
+
+            # Don't have
+            res = await full_node_1.request_unfinished_block2(
+                fnp.RequestUnfinishedBlock2(unf.partial_hash, unf.foliage.foliage_transaction_block_hash)
+            )
+            assert res is None
+
+            await full_node_1.full_node.add_unfinished_block(unf, peer)
+            # Have
+            res = await full_node_1.request_unfinished_block2(
+                fnp.RequestUnfinishedBlock2(unf.partial_hash, unf.foliage.foliage_transaction_block_hash)
+            )
+            assert res.data == bytes(fnp.RespondUnfinishedBlock(unf))
+
+            res = await full_node_1.request_unfinished_block(fnp.RequestUnfinishedBlock(unf.partial_hash))
+            assert res.data == bytes(fnp.RespondUnfinishedBlock(best_unf))
+
+            res = await full_node_1.request_unfinished_block2(fnp.RequestUnfinishedBlock2(unf.partial_hash, None))
+            assert res.data == bytes(fnp.RespondUnfinishedBlock(best_unf))
+
     @pytest.mark.anyio
     async def test_new_signage_point_or_end_of_sub_slot(self, wallet_nodes, self_hostname):
         full_node_1, full_node_2, server_1, server_2, wallet_a, wallet_receiver, bt = wallet_nodes
diff --git a/tests/util/build_network_protocol_files.py b/tests/util/build_network_protocol_files.py
index c520397b0b12..4fff6b5e6b15 100644
--- a/tests/util/build_network_protocol_files.py
+++ b/tests/util/build_network_protocol_files.py
@@ -59,6 +59,8 @@ def visit_full_node(visitor: Callable[[Any, str], None]) -> None:
     visitor(respond_compact_vdf, "respond_compact_vdf")
     visitor(request_peers, "request_peers")
     visitor(respond_peers, "respond_peers")
+    visitor(new_unfinished_block2, "new_unfinished_block2")
+    visitor(request_unfinished_block2, "request_unfinished_block2")
 
 
 def visit_wallet_protocol(visitor: Callable[[Any, str], None]) -> None:
diff --git a/tests/util/full_sync.py b/tests/util/full_sync.py
index 43cb7aeb3afd..48a038522905 100644
--- a/tests/util/full_sync.py
+++ b/tests/util/full_sync.py
@@ -7,7 +7,7 @@
 import time
 from contextlib import contextmanager
 from pathlib import Path
-from typing import Iterator, List, Optional, cast
+from typing import Callable, Iterator, List, Optional, cast
 
 import aiosqlite
 import zstd
@@ -59,6 +59,15 @@ async def send_to_all(
     ) -> None:
         pass
 
+    async def send_to_all_if(
+        self,
+        messages: List[Message],
+        node_type: NodeType,
+        predicate: Callable[[WSChiaConnection], bool],
+        exclude: Optional[bytes32] = None,
+    ) -> None:
+        pass
+
     def set_received_message_callback(self, callback: ConnectionCallback) -> None:
         pass
 
diff --git a/tests/util/network_protocol_data.py b/tests/util/network_protocol_data.py
index 49eef486597b..01f0938066f0 100644
--- a/tests/util/network_protocol_data.py
+++ b/tests/util/network_protocol_data.py
@@ -408,11 +408,21 @@
 )
 
 new_unfinished_block = full_node_protocol.NewUnfinishedBlock(
-    bytes32(bytes.fromhex("229646fb33551966039d9324c0d10166c554d20e9a11e3f30942ec0bb346377e")),
+    bytes32.fromhex("229646fb33551966039d9324c0d10166c554d20e9a11e3f30942ec0bb346377e"),
 )
 
 request_unfinished_block = full_node_protocol.RequestUnfinishedBlock(
-    bytes32(bytes.fromhex("8b5e5a59f33bb89e1bfd5aca79409352864e70aa7765c331d641875f83d59d1d")),
+    bytes32.fromhex("8b5e5a59f33bb89e1bfd5aca79409352864e70aa7765c331d641875f83d59d1d"),
+)
+
+new_unfinished_block2 = full_node_protocol.NewUnfinishedBlock2(
+    bytes32.fromhex("229646fb33551966039d9324c0d10166c554d20e9a11e3f30942ec0bb346377e"),
+    bytes32.fromhex("166c554d20e9a11e3f30942ec0bb346377e229646fb33551966039d9324c0d10"),
+)
+
+request_unfinished_block2 = full_node_protocol.RequestUnfinishedBlock2(
+    bytes32.fromhex("8b5e5a59f33bb89e1bfd5aca79409352864e70aa7765c331d641875f83d59d1d"),
+    bytes32.fromhex("a79409352864e70aa7765c331d641875f83d59d1d8b5e5a59f33bb89e1bfd5ac"),
 )
 
 unfinished_block = UnfinishedBlock(
diff --git a/tests/util/protocol_messages_bytes-v1.0 b/tests/util/protocol_messages_bytes-v1.0
index 3b356b7d1473..007b72bb538a 100644
Binary files a/tests/util/protocol_messages_bytes-v1.0 and b/tests/util/protocol_messages_bytes-v1.0 differ
diff --git a/tests/util/protocol_messages_json.py b/tests/util/protocol_messages_json.py
index d1c5d7c83633..28989b7931e9 100644
--- a/tests/util/protocol_messages_json.py
+++ b/tests/util/protocol_messages_json.py
@@ -1222,6 +1222,16 @@
 
 respond_peers_json: Dict[str, Any] = {"peer_list": [{"host": "127.0.0.1", "port": 8444, "timestamp": 10796}]}
 
+new_unfinished_block2_json: Dict[str, Any] = {
+    "unfinished_reward_hash": "0x229646fb33551966039d9324c0d10166c554d20e9a11e3f30942ec0bb346377e",
+    "foliage_hash": "0x166c554d20e9a11e3f30942ec0bb346377e229646fb33551966039d9324c0d10",
+}
+
+request_unfinished_block2_json: Dict[str, Any] = {
+    "unfinished_reward_hash": "0x8b5e5a59f33bb89e1bfd5aca79409352864e70aa7765c331d641875f83d59d1d",
+    "foliage_hash": "0xa79409352864e70aa7765c331d641875f83d59d1d8b5e5a59f33bb89e1bfd5ac",
+}
+
 request_puzzle_solution_json: Dict[str, Any] = {
     "coin_name": "0x6edddb46bd154f50566b49c95812e0f1131a0a7162630349fc8d1d696e463e47",
     "height": 3905474497,
diff --git a/tests/util/test_network_protocol_files.py b/tests/util/test_network_protocol_files.py
index 2b80661b5152..bea36cb7dd5c 100644
--- a/tests/util/test_network_protocol_files.py
+++ b/tests/util/test_network_protocol_files.py
@@ -176,328 +176,338 @@ def test_protocol_bytes() -> None:
     assert bytes(message_29) == bytes(respond_peers)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_30 = type(request_puzzle_solution).from_bytes(message_bytes)
-    assert message_30 == request_puzzle_solution
-    assert bytes(message_30) == bytes(request_puzzle_solution)
+    message_30 = type(new_unfinished_block2).from_bytes(message_bytes)
+    assert message_30 == new_unfinished_block2
+    assert bytes(message_30) == bytes(new_unfinished_block2)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_31 = type(puzzle_solution_response).from_bytes(message_bytes)
-    assert message_31 == puzzle_solution_response
-    assert bytes(message_31) == bytes(puzzle_solution_response)
+    message_31 = type(request_unfinished_block2).from_bytes(message_bytes)
+    assert message_31 == request_unfinished_block2
+    assert bytes(message_31) == bytes(request_unfinished_block2)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_32 = type(respond_puzzle_solution).from_bytes(message_bytes)
-    assert message_32 == respond_puzzle_solution
-    assert bytes(message_32) == bytes(respond_puzzle_solution)
+    message_32 = type(request_puzzle_solution).from_bytes(message_bytes)
+    assert message_32 == request_puzzle_solution
+    assert bytes(message_32) == bytes(request_puzzle_solution)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_33 = type(reject_puzzle_solution).from_bytes(message_bytes)
-    assert message_33 == reject_puzzle_solution
-    assert bytes(message_33) == bytes(reject_puzzle_solution)
+    message_33 = type(puzzle_solution_response).from_bytes(message_bytes)
+    assert message_33 == puzzle_solution_response
+    assert bytes(message_33) == bytes(puzzle_solution_response)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_34 = type(send_transaction).from_bytes(message_bytes)
-    assert message_34 == send_transaction
-    assert bytes(message_34) == bytes(send_transaction)
+    message_34 = type(respond_puzzle_solution).from_bytes(message_bytes)
+    assert message_34 == respond_puzzle_solution
+    assert bytes(message_34) == bytes(respond_puzzle_solution)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_35 = type(transaction_ack).from_bytes(message_bytes)
-    assert message_35 == transaction_ack
-    assert bytes(message_35) == bytes(transaction_ack)
+    message_35 = type(reject_puzzle_solution).from_bytes(message_bytes)
+    assert message_35 == reject_puzzle_solution
+    assert bytes(message_35) == bytes(reject_puzzle_solution)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_36 = type(new_peak_wallet).from_bytes(message_bytes)
-    assert message_36 == new_peak_wallet
-    assert bytes(message_36) == bytes(new_peak_wallet)
+    message_36 = type(send_transaction).from_bytes(message_bytes)
+    assert message_36 == send_transaction
+    assert bytes(message_36) == bytes(send_transaction)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_37 = type(request_block_header).from_bytes(message_bytes)
-    assert message_37 == request_block_header
-    assert bytes(message_37) == bytes(request_block_header)
+    message_37 = type(transaction_ack).from_bytes(message_bytes)
+    assert message_37 == transaction_ack
+    assert bytes(message_37) == bytes(transaction_ack)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_38 = type(request_block_headers).from_bytes(message_bytes)
-    assert message_38 == request_block_headers
-    assert bytes(message_38) == bytes(request_block_headers)
+    message_38 = type(new_peak_wallet).from_bytes(message_bytes)
+    assert message_38 == new_peak_wallet
+    assert bytes(message_38) == bytes(new_peak_wallet)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_39 = type(respond_header_block).from_bytes(message_bytes)
-    assert message_39 == respond_header_block
-    assert bytes(message_39) == bytes(respond_header_block)
+    message_39 = type(request_block_header).from_bytes(message_bytes)
+    assert message_39 == request_block_header
+    assert bytes(message_39) == bytes(request_block_header)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_40 = type(respond_block_headers).from_bytes(message_bytes)
-    assert message_40 == respond_block_headers
-    assert bytes(message_40) == bytes(respond_block_headers)
+    message_40 = type(request_block_headers).from_bytes(message_bytes)
+    assert message_40 == request_block_headers
+    assert bytes(message_40) == bytes(request_block_headers)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_41 = type(reject_header_request).from_bytes(message_bytes)
-    assert message_41 == reject_header_request
-    assert bytes(message_41) == bytes(reject_header_request)
+    message_41 = type(respond_header_block).from_bytes(message_bytes)
+    assert message_41 == respond_header_block
+    assert bytes(message_41) == bytes(respond_header_block)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_42 = type(request_removals).from_bytes(message_bytes)
-    assert message_42 == request_removals
-    assert bytes(message_42) == bytes(request_removals)
+    message_42 = type(respond_block_headers).from_bytes(message_bytes)
+    assert message_42 == respond_block_headers
+    assert bytes(message_42) == bytes(respond_block_headers)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_43 = type(respond_removals).from_bytes(message_bytes)
-    assert message_43 == respond_removals
-    assert bytes(message_43) == bytes(respond_removals)
+    message_43 = type(reject_header_request).from_bytes(message_bytes)
+    assert message_43 == reject_header_request
+    assert bytes(message_43) == bytes(reject_header_request)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_44 = type(reject_removals_request).from_bytes(message_bytes)
-    assert message_44 == reject_removals_request
-    assert bytes(message_44) == bytes(reject_removals_request)
+    message_44 = type(request_removals).from_bytes(message_bytes)
+    assert message_44 == request_removals
+    assert bytes(message_44) == bytes(request_removals)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_45 = type(request_additions).from_bytes(message_bytes)
-    assert message_45 == request_additions
-    assert bytes(message_45) == bytes(request_additions)
+    message_45 = type(respond_removals).from_bytes(message_bytes)
+    assert message_45 == respond_removals
+    assert bytes(message_45) == bytes(respond_removals)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_46 = type(respond_additions).from_bytes(message_bytes)
-    assert message_46 == respond_additions
-    assert bytes(message_46) == bytes(respond_additions)
+    message_46 = type(reject_removals_request).from_bytes(message_bytes)
+    assert message_46 == reject_removals_request
+    assert bytes(message_46) == bytes(reject_removals_request)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_47 = type(reject_additions).from_bytes(message_bytes)
-    assert message_47 == reject_additions
-    assert bytes(message_47) == bytes(reject_additions)
+    message_47 = type(request_additions).from_bytes(message_bytes)
+    assert message_47 == request_additions
+    assert bytes(message_47) == bytes(request_additions)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_48 = type(request_header_blocks).from_bytes(message_bytes)
-    assert message_48 == request_header_blocks
-    assert bytes(message_48) == bytes(request_header_blocks)
+    message_48 = type(respond_additions).from_bytes(message_bytes)
+    assert message_48 == respond_additions
+    assert bytes(message_48) == bytes(respond_additions)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_49 = type(reject_header_blocks).from_bytes(message_bytes)
-    assert message_49 == reject_header_blocks
-    assert bytes(message_49) == bytes(reject_header_blocks)
+    message_49 = type(reject_additions).from_bytes(message_bytes)
+    assert message_49 == reject_additions
+    assert bytes(message_49) == bytes(reject_additions)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_50 = type(respond_header_blocks).from_bytes(message_bytes)
-    assert message_50 == respond_header_blocks
-    assert bytes(message_50) == bytes(respond_header_blocks)
+    message_50 = type(request_header_blocks).from_bytes(message_bytes)
+    assert message_50 == request_header_blocks
+    assert bytes(message_50) == bytes(request_header_blocks)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_51 = type(coin_state).from_bytes(message_bytes)
-    assert message_51 == coin_state
-    assert bytes(message_51) == bytes(coin_state)
+    message_51 = type(reject_header_blocks).from_bytes(message_bytes)
+    assert message_51 == reject_header_blocks
+    assert bytes(message_51) == bytes(reject_header_blocks)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_52 = type(register_for_ph_updates).from_bytes(message_bytes)
-    assert message_52 == register_for_ph_updates
-    assert bytes(message_52) == bytes(register_for_ph_updates)
+    message_52 = type(respond_header_blocks).from_bytes(message_bytes)
+    assert message_52 == respond_header_blocks
+    assert bytes(message_52) == bytes(respond_header_blocks)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_53 = type(reject_block_headers).from_bytes(message_bytes)
-    assert message_53 == reject_block_headers
-    assert bytes(message_53) == bytes(reject_block_headers)
+    message_53 = type(coin_state).from_bytes(message_bytes)
+    assert message_53 == coin_state
+    assert bytes(message_53) == bytes(coin_state)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_54 = type(respond_to_ph_updates).from_bytes(message_bytes)
-    assert message_54 == respond_to_ph_updates
-    assert bytes(message_54) == bytes(respond_to_ph_updates)
+    message_54 = type(register_for_ph_updates).from_bytes(message_bytes)
+    assert message_54 == register_for_ph_updates
+    assert bytes(message_54) == bytes(register_for_ph_updates)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_55 = type(register_for_coin_updates).from_bytes(message_bytes)
-    assert message_55 == register_for_coin_updates
-    assert bytes(message_55) == bytes(register_for_coin_updates)
+    message_55 = type(reject_block_headers).from_bytes(message_bytes)
+    assert message_55 == reject_block_headers
+    assert bytes(message_55) == bytes(reject_block_headers)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_56 = type(respond_to_coin_updates).from_bytes(message_bytes)
-    assert message_56 == respond_to_coin_updates
-    assert bytes(message_56) == bytes(respond_to_coin_updates)
+    message_56 = type(respond_to_ph_updates).from_bytes(message_bytes)
+    assert message_56 == respond_to_ph_updates
+    assert bytes(message_56) == bytes(respond_to_ph_updates)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_57 = type(coin_state_update).from_bytes(message_bytes)
-    assert message_57 == coin_state_update
-    assert bytes(message_57) == bytes(coin_state_update)
+    message_57 = type(register_for_coin_updates).from_bytes(message_bytes)
+    assert message_57 == register_for_coin_updates
+    assert bytes(message_57) == bytes(register_for_coin_updates)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_58 = type(request_children).from_bytes(message_bytes)
-    assert message_58 == request_children
-    assert bytes(message_58) == bytes(request_children)
+    message_58 = type(respond_to_coin_updates).from_bytes(message_bytes)
+    assert message_58 == respond_to_coin_updates
+    assert bytes(message_58) == bytes(respond_to_coin_updates)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_59 = type(respond_children).from_bytes(message_bytes)
-    assert message_59 == respond_children
-    assert bytes(message_59) == bytes(respond_children)
+    message_59 = type(coin_state_update).from_bytes(message_bytes)
+    assert message_59 == coin_state_update
+    assert bytes(message_59) == bytes(coin_state_update)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_60 = type(request_ses_info).from_bytes(message_bytes)
-    assert message_60 == request_ses_info
-    assert bytes(message_60) == bytes(request_ses_info)
+    message_60 = type(request_children).from_bytes(message_bytes)
+    assert message_60 == request_children
+    assert bytes(message_60) == bytes(request_children)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_61 = type(respond_ses_info).from_bytes(message_bytes)
-    assert message_61 == respond_ses_info
-    assert bytes(message_61) == bytes(respond_ses_info)
+    message_61 = type(respond_children).from_bytes(message_bytes)
+    assert message_61 == respond_children
+    assert bytes(message_61) == bytes(respond_children)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_62 = type(pool_difficulty).from_bytes(message_bytes)
-    assert message_62 == pool_difficulty
-    assert bytes(message_62) == bytes(pool_difficulty)
+    message_62 = type(request_ses_info).from_bytes(message_bytes)
+    assert message_62 == request_ses_info
+    assert bytes(message_62) == bytes(request_ses_info)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_63 = type(harvester_handhsake).from_bytes(message_bytes)
-    assert message_63 == harvester_handhsake
-    assert bytes(message_63) == bytes(harvester_handhsake)
+    message_63 = type(respond_ses_info).from_bytes(message_bytes)
+    assert message_63 == respond_ses_info
+    assert bytes(message_63) == bytes(respond_ses_info)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_64 = type(new_signage_point_harvester).from_bytes(message_bytes)
-    assert message_64 == new_signage_point_harvester
-    assert bytes(message_64) == bytes(new_signage_point_harvester)
+    message_64 = type(pool_difficulty).from_bytes(message_bytes)
+    assert message_64 == pool_difficulty
+    assert bytes(message_64) == bytes(pool_difficulty)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_65 = type(new_proof_of_space).from_bytes(message_bytes)
-    assert message_65 == new_proof_of_space
-    assert bytes(message_65) == bytes(new_proof_of_space)
+    message_65 = type(harvester_handhsake).from_bytes(message_bytes)
+    assert message_65 == harvester_handhsake
+    assert bytes(message_65) == bytes(harvester_handhsake)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_66 = type(request_signatures).from_bytes(message_bytes)
-    assert message_66 == request_signatures
-    assert bytes(message_66) == bytes(request_signatures)
+    message_66 = type(new_signage_point_harvester).from_bytes(message_bytes)
+    assert message_66 == new_signage_point_harvester
+    assert bytes(message_66) == bytes(new_signage_point_harvester)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_67 = type(respond_signatures).from_bytes(message_bytes)
-    assert message_67 == respond_signatures
-    assert bytes(message_67) == bytes(respond_signatures)
+    message_67 = type(new_proof_of_space).from_bytes(message_bytes)
+    assert message_67 == new_proof_of_space
+    assert bytes(message_67) == bytes(new_proof_of_space)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_68 = type(plot).from_bytes(message_bytes)
-    assert message_68 == plot
-    assert bytes(message_68) == bytes(plot)
+    message_68 = type(request_signatures).from_bytes(message_bytes)
+    assert message_68 == request_signatures
+    assert bytes(message_68) == bytes(request_signatures)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_69 = type(request_plots).from_bytes(message_bytes)
-    assert message_69 == request_plots
-    assert bytes(message_69) == bytes(request_plots)
+    message_69 = type(respond_signatures).from_bytes(message_bytes)
+    assert message_69 == respond_signatures
+    assert bytes(message_69) == bytes(respond_signatures)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_70 = type(respond_plots).from_bytes(message_bytes)
-    assert message_70 == respond_plots
-    assert bytes(message_70) == bytes(respond_plots)
+    message_70 = type(plot).from_bytes(message_bytes)
+    assert message_70 == plot
+    assert bytes(message_70) == bytes(plot)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_71 = type(request_peers_introducer).from_bytes(message_bytes)
-    assert message_71 == request_peers_introducer
-    assert bytes(message_71) == bytes(request_peers_introducer)
+    message_71 = type(request_plots).from_bytes(message_bytes)
+    assert message_71 == request_plots
+    assert bytes(message_71) == bytes(request_plots)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_72 = type(respond_peers_introducer).from_bytes(message_bytes)
-    assert message_72 == respond_peers_introducer
-    assert bytes(message_72) == bytes(respond_peers_introducer)
+    message_72 = type(respond_plots).from_bytes(message_bytes)
+    assert message_72 == respond_plots
+    assert bytes(message_72) == bytes(respond_plots)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_73 = type(authentication_payload).from_bytes(message_bytes)
-    assert message_73 == authentication_payload
-    assert bytes(message_73) == bytes(authentication_payload)
+    message_73 = type(request_peers_introducer).from_bytes(message_bytes)
+    assert message_73 == request_peers_introducer
+    assert bytes(message_73) == bytes(request_peers_introducer)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_74 = type(get_pool_info_response).from_bytes(message_bytes)
-    assert message_74 == get_pool_info_response
-    assert bytes(message_74) == bytes(get_pool_info_response)
+    message_74 = type(respond_peers_introducer).from_bytes(message_bytes)
+    assert message_74 == respond_peers_introducer
+    assert bytes(message_74) == bytes(respond_peers_introducer)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_75 = type(post_partial_payload).from_bytes(message_bytes)
-    assert message_75 == post_partial_payload
-    assert bytes(message_75) == bytes(post_partial_payload)
+    message_75 = type(authentication_payload).from_bytes(message_bytes)
+    assert message_75 == authentication_payload
+    assert bytes(message_75) == bytes(authentication_payload)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_76 = type(post_partial_request).from_bytes(message_bytes)
-    assert message_76 == post_partial_request
-    assert bytes(message_76) == bytes(post_partial_request)
+    message_76 = type(get_pool_info_response).from_bytes(message_bytes)
+    assert message_76 == get_pool_info_response
+    assert bytes(message_76) == bytes(get_pool_info_response)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_77 = type(post_partial_response).from_bytes(message_bytes)
-    assert message_77 == post_partial_response
-    assert bytes(message_77) == bytes(post_partial_response)
+    message_77 = type(post_partial_payload).from_bytes(message_bytes)
+    assert message_77 == post_partial_payload
+    assert bytes(message_77) == bytes(post_partial_payload)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_78 = type(get_farmer_response).from_bytes(message_bytes)
-    assert message_78 == get_farmer_response
-    assert bytes(message_78) == bytes(get_farmer_response)
+    message_78 = type(post_partial_request).from_bytes(message_bytes)
+    assert message_78 == post_partial_request
+    assert bytes(message_78) == bytes(post_partial_request)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_79 = type(post_farmer_payload).from_bytes(message_bytes)
-    assert message_79 == post_farmer_payload
-    assert bytes(message_79) == bytes(post_farmer_payload)
+    message_79 = type(post_partial_response).from_bytes(message_bytes)
+    assert message_79 == post_partial_response
+    assert bytes(message_79) == bytes(post_partial_response)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_80 = type(post_farmer_request).from_bytes(message_bytes)
-    assert message_80 == post_farmer_request
-    assert bytes(message_80) == bytes(post_farmer_request)
+    message_80 = type(get_farmer_response).from_bytes(message_bytes)
+    assert message_80 == get_farmer_response
+    assert bytes(message_80) == bytes(get_farmer_response)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_81 = type(post_farmer_response).from_bytes(message_bytes)
-    assert message_81 == post_farmer_response
-    assert bytes(message_81) == bytes(post_farmer_response)
+    message_81 = type(post_farmer_payload).from_bytes(message_bytes)
+    assert message_81 == post_farmer_payload
+    assert bytes(message_81) == bytes(post_farmer_payload)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_82 = type(put_farmer_payload).from_bytes(message_bytes)
-    assert message_82 == put_farmer_payload
-    assert bytes(message_82) == bytes(put_farmer_payload)
+    message_82 = type(post_farmer_request).from_bytes(message_bytes)
+    assert message_82 == post_farmer_request
+    assert bytes(message_82) == bytes(post_farmer_request)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_83 = type(put_farmer_request).from_bytes(message_bytes)
-    assert message_83 == put_farmer_request
-    assert bytes(message_83) == bytes(put_farmer_request)
+    message_83 = type(post_farmer_response).from_bytes(message_bytes)
+    assert message_83 == post_farmer_response
+    assert bytes(message_83) == bytes(post_farmer_response)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_84 = type(put_farmer_response).from_bytes(message_bytes)
-    assert message_84 == put_farmer_response
-    assert bytes(message_84) == bytes(put_farmer_response)
+    message_84 = type(put_farmer_payload).from_bytes(message_bytes)
+    assert message_84 == put_farmer_payload
+    assert bytes(message_84) == bytes(put_farmer_payload)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_85 = type(error_response).from_bytes(message_bytes)
-    assert message_85 == error_response
-    assert bytes(message_85) == bytes(error_response)
+    message_85 = type(put_farmer_request).from_bytes(message_bytes)
+    assert message_85 == put_farmer_request
+    assert bytes(message_85) == bytes(put_farmer_request)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_86 = type(new_peak_timelord).from_bytes(message_bytes)
-    assert message_86 == new_peak_timelord
-    assert bytes(message_86) == bytes(new_peak_timelord)
+    message_86 = type(put_farmer_response).from_bytes(message_bytes)
+    assert message_86 == put_farmer_response
+    assert bytes(message_86) == bytes(put_farmer_response)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_87 = type(new_unfinished_block_timelord).from_bytes(message_bytes)
-    assert message_87 == new_unfinished_block_timelord
-    assert bytes(message_87) == bytes(new_unfinished_block_timelord)
+    message_87 = type(error_response).from_bytes(message_bytes)
+    assert message_87 == error_response
+    assert bytes(message_87) == bytes(error_response)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_88 = type(new_infusion_point_vdf).from_bytes(message_bytes)
-    assert message_88 == new_infusion_point_vdf
-    assert bytes(message_88) == bytes(new_infusion_point_vdf)
+    message_88 = type(new_peak_timelord).from_bytes(message_bytes)
+    assert message_88 == new_peak_timelord
+    assert bytes(message_88) == bytes(new_peak_timelord)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_89 = type(new_signage_point_vdf).from_bytes(message_bytes)
-    assert message_89 == new_signage_point_vdf
-    assert bytes(message_89) == bytes(new_signage_point_vdf)
+    message_89 = type(new_unfinished_block_timelord).from_bytes(message_bytes)
+    assert message_89 == new_unfinished_block_timelord
+    assert bytes(message_89) == bytes(new_unfinished_block_timelord)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_90 = type(new_end_of_sub_slot_bundle).from_bytes(message_bytes)
-    assert message_90 == new_end_of_sub_slot_bundle
-    assert bytes(message_90) == bytes(new_end_of_sub_slot_bundle)
+    message_90 = type(new_infusion_point_vdf).from_bytes(message_bytes)
+    assert message_90 == new_infusion_point_vdf
+    assert bytes(message_90) == bytes(new_infusion_point_vdf)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_91 = type(request_compact_proof_of_time).from_bytes(message_bytes)
-    assert message_91 == request_compact_proof_of_time
-    assert bytes(message_91) == bytes(request_compact_proof_of_time)
+    message_91 = type(new_signage_point_vdf).from_bytes(message_bytes)
+    assert message_91 == new_signage_point_vdf
+    assert bytes(message_91) == bytes(new_signage_point_vdf)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_92 = type(respond_compact_proof_of_time).from_bytes(message_bytes)
-    assert message_92 == respond_compact_proof_of_time
-    assert bytes(message_92) == bytes(respond_compact_proof_of_time)
+    message_92 = type(new_end_of_sub_slot_bundle).from_bytes(message_bytes)
+    assert message_92 == new_end_of_sub_slot_bundle
+    assert bytes(message_92) == bytes(new_end_of_sub_slot_bundle)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_93 = type(error_without_data).from_bytes(message_bytes)
-    assert message_93 == error_without_data
-    assert bytes(message_93) == bytes(error_without_data)
+    message_93 = type(request_compact_proof_of_time).from_bytes(message_bytes)
+    assert message_93 == request_compact_proof_of_time
+    assert bytes(message_93) == bytes(request_compact_proof_of_time)
 
     message_bytes, input_bytes = parse_blob(input_bytes)
-    message_94 = type(error_with_data).from_bytes(message_bytes)
-    assert message_94 == error_with_data
-    assert bytes(message_94) == bytes(error_with_data)
+    message_94 = type(respond_compact_proof_of_time).from_bytes(message_bytes)
+    assert message_94 == respond_compact_proof_of_time
+    assert bytes(message_94) == bytes(respond_compact_proof_of_time)
+
+    message_bytes, input_bytes = parse_blob(input_bytes)
+    message_95 = type(error_without_data).from_bytes(message_bytes)
+    assert message_95 == error_without_data
+    assert bytes(message_95) == bytes(error_without_data)
+
+    message_bytes, input_bytes = parse_blob(input_bytes)
+    message_96 = type(error_with_data).from_bytes(message_bytes)
+    assert message_96 == error_with_data
+    assert bytes(message_96) == bytes(error_with_data)
 
     assert input_bytes == b""
diff --git a/tests/util/test_network_protocol_json.py b/tests/util/test_network_protocol_json.py
index 3f403ad0d79e..4362223765f2 100644
--- a/tests/util/test_network_protocol_json.py
+++ b/tests/util/test_network_protocol_json.py
@@ -78,6 +78,10 @@ def test_protocol_json() -> None:
     assert type(request_peers).from_json_dict(request_peers_json) == request_peers
     assert str(respond_peers_json) == str(respond_peers.to_json_dict())
     assert type(respond_peers).from_json_dict(respond_peers_json) == respond_peers
+    assert str(new_unfinished_block2_json) == str(new_unfinished_block2.to_json_dict())
+    assert type(new_unfinished_block2).from_json_dict(new_unfinished_block2_json) == new_unfinished_block2
+    assert str(request_unfinished_block2_json) == str(request_unfinished_block2.to_json_dict())
+    assert type(request_unfinished_block2).from_json_dict(request_unfinished_block2_json) == request_unfinished_block2
     assert str(request_puzzle_solution_json) == str(request_puzzle_solution.to_json_dict())
     assert type(request_puzzle_solution).from_json_dict(request_puzzle_solution_json) == request_puzzle_solution
     assert str(puzzle_solution_response_json) == str(puzzle_solution_response.to_json_dict())
diff --git a/tests/util/test_network_protocol_test.py b/tests/util/test_network_protocol_test.py
index 33261598e56a..b22d8bf1a0b9 100644
--- a/tests/util/test_network_protocol_test.py
+++ b/tests/util/test_network_protocol_test.py
@@ -41,10 +41,10 @@ def test_missing_messages_state_machine() -> None:
     # to the visitor in build_network_protocol_files.py and rerun it. Then
     # update this test
     assert (
-        len(VALID_REPLY_MESSAGE_MAP) == 20
+        len(VALID_REPLY_MESSAGE_MAP) == 21
     ), "A message was added to the protocol state machine. Make sure to update the protocol message regression test to include the new message"
     assert (
-        len(NO_REPLY_EXPECTED) == 7
+        len(NO_REPLY_EXPECTED) == 8
     ), "A message was added to the protocol state machine. Make sure to update the protocol message regression test to include the new message"
 
 
@@ -124,6 +124,7 @@ def test_missing_messages() -> None:
         "NewSignagePointOrEndOfSubSlot",
         "NewTransaction",
         "NewUnfinishedBlock",
+        "NewUnfinishedBlock2",
         "RejectBlock",
         "RejectBlocks",
         "RequestBlock",
@@ -135,6 +136,7 @@ def test_missing_messages() -> None:
         "RequestSignagePointOrEndOfSubSlot",
         "RequestTransaction",
         "RequestUnfinishedBlock",
+        "RequestUnfinishedBlock2",
         "RespondBlock",
         "RespondBlocks",
         "RespondCompactVDF",