from __future__ import annotations import pytest from chia_rs import BlockRecord, FullBlock, SubEpochSummary, UnfinishedBlock, is_overflow_block from chia_rs.sized_bytes import bytes32 from chia_rs.sized_ints import uint64, uint128 from chia._tests.blockchain.blockchain_test_utils import _validate_and_add_block from chia._tests.util.blockchain import create_blockchain from chia._tests.util.time_out_assert import time_out_assert from chia.consensus.blockchain import Blockchain from chia.consensus.difficulty_adjustment import get_next_sub_slot_iters_and_difficulty from chia.consensus.make_sub_epoch_summary import next_sub_epoch_summary from chia.full_node.full_node_service import FullNodeService from chia.protocols import timelord_protocol from chia.server.server import ChiaServer from chia.simulator.block_tools import BlockTools from chia.simulator.full_node_simulator import FullNodeSimulator from chia.simulator.wallet_tools import WalletTool from chia.timelord.timelord_api import TimelordAPI def last_unfinished(tl: TimelordAPI, *, overflow: bool) -> timelord_protocol.NewUnfinishedBlockTimelord: if overflow: return tl.timelord.overflow_blocks[-1] else: return tl.timelord.unfinished_blocks[-1] class TestNewPeak: @pytest.mark.anyio async def test_timelord_new_peak_basic( self, timelord: tuple[TimelordAPI, ChiaServer], default_1000_blocks: list[FullBlock], one_node: tuple[list[FullNodeService], list[FullNodeSimulator], BlockTools], ) -> None: [full_node_service], _, bt = one_node full_node = full_node_service._node async with create_blockchain(bt.constants, 2) as (b1, _): async with create_blockchain(bt.constants, 2) as (b2, _): timelord_api, _ = timelord for block in default_1000_blocks: await _validate_and_add_block(b1, block) await _validate_and_add_block(b2, block) await full_node.add_block(block) peak = timelord_peak_from_block(b1, default_1000_blocks[-1]) assert peak is not None assert timelord_api.timelord.new_peak is None await timelord_api.new_peak_timelord(peak) await time_out_assert(60, tl_new_peak_is_none, True, timelord_api) assert timelord_api.timelord.last_state.peak is not None assert ( timelord_api.timelord.last_state.peak.reward_chain_block.get_hash() == peak.reward_chain_block.get_hash() ) blocks = bt.get_consecutive_blocks(1, default_1000_blocks) await _validate_and_add_block(b1, blocks[-1]) await _validate_and_add_block(b2, blocks[-1]) await timelord_api.new_peak_timelord(timelord_peak_from_block(b1, blocks[-1])) await full_node.add_block(blocks[-1]) await time_out_assert(60, tl_new_peak_is_none, True, timelord_api) assert timelord_api.timelord.last_state.peak is not None assert ( timelord_api.timelord.last_state.peak.reward_chain_block.get_hash() == blocks[-1].reward_chain_block.get_hash() ) fn_peak = full_node.blockchain.get_peak() assert fn_peak is not None and fn_peak.header_hash == blocks[-1].header_hash blocks_1 = bt.get_consecutive_blocks(2, blocks) await _validate_and_add_block(b1, blocks_1[-2]) await _validate_and_add_block(b1, blocks_1[-1]) await timelord_api.new_peak_timelord(timelord_peak_from_block(b1, blocks_1[-2])) await full_node.add_block(blocks_1[-2]) await timelord_api.new_peak_timelord(timelord_peak_from_block(b1, blocks_1[-1])) await full_node.add_block(blocks_1[-1]) await time_out_assert(60, tl_new_peak_is_none, True, timelord_api) assert timelord_api.timelord.last_state.peak is not None assert ( timelord_api.timelord.last_state.peak.reward_chain_block.get_hash() == blocks_1[-1].reward_chain_block.get_hash() ) fn_peak = full_node.blockchain.get_peak() assert fn_peak is not None and fn_peak.header_hash == blocks_1[-1].header_hash # new unknown peak, weight less then curr peak blocks_2 = bt.get_consecutive_blocks(1, blocks) await _validate_and_add_block(b2, blocks_2[-1]) await timelord_api.new_peak_timelord(timelord_peak_from_block(b2, blocks_2[-1])) await full_node.add_block(blocks_2[-1]) await time_out_assert(60, tl_new_peak_is_none, True, timelord_api) assert timelord_api.timelord.last_state.last_weight == blocks_1[-1].weight assert timelord_api.timelord.last_state.total_iters == blocks_1[-1].reward_chain_block.total_iters fn_peak = full_node.blockchain.get_peak() assert fn_peak is not None and fn_peak.header_hash == blocks_1[-1].header_hash @pytest.mark.anyio async def test_timelord_new_peak_unfinished_not_orphaned( self, bt: BlockTools, timelord: tuple[TimelordAPI, ChiaServer], default_1000_blocks: list[FullBlock] ) -> None: async with create_blockchain(bt.constants, 2) as (b1, _): timelord_api, _ = timelord for block in default_1000_blocks: await _validate_and_add_block(b1, block) peak = timelord_peak_from_block(b1, default_1000_blocks[-1]) assert peak is not None assert timelord_api.timelord.new_peak is None await timelord_api.new_peak_timelord(peak) assert timelord_api.timelord.new_peak is not None assert timelord_api.timelord.new_peak.reward_chain_block.get_hash() == peak.reward_chain_block.get_hash() # make two new blocks on tip blocks_1 = bt.get_consecutive_blocks(2, default_1000_blocks) block_1 = blocks_1[-2] block_2 = blocks_1[-1] await _validate_and_add_block(b1, block_1) await _validate_and_add_block(b1, block_2) block_record = b1.block_record(block_2.header_hash) timelord_unf_block = timelord_protocol.NewUnfinishedBlockTimelord( block_1.reward_chain_block.get_unfinished(), uint64(block_record.weight - default_1000_blocks[-1].weight), block_record.sub_slot_iters, block_1.foliage, next_sub_epoch_summary(bt.constants, b1, block_record.required_iters, block_1, True), await get_rc_prev(b1, block_1), ) await timelord_api.new_unfinished_block_timelord(timelord_unf_block) overflow = is_overflow_block(bt.constants, timelord_unf_block.reward_chain_block.signage_point_index) assert last_unfinished(timelord_api, overflow=overflow).get_hash() == timelord_unf_block.get_hash() new_peak = timelord_peak_from_block(b1, block_2) assert timelord_unf_block.reward_chain_block.total_iters <= new_peak.reward_chain_block.total_iters await timelord_api.new_peak_timelord(new_peak) await time_out_assert(60, tl_new_peak_is_none, True, timelord_api) assert ( timelord_api.timelord.last_state.peak.reward_chain_block.get_hash() == new_peak.reward_chain_block.get_hash() ) @pytest.mark.anyio async def test_timelord_new_peak_unfinished_orphaned( self, one_node: tuple[list[FullNodeService], list[FullNodeSimulator], BlockTools], timelord: tuple[TimelordAPI, ChiaServer], default_1000_blocks: list[FullBlock], ) -> None: [full_node_service], _, bt = one_node full_node = full_node_service._node async with create_blockchain(bt.constants, 2) as (b1, _): async with create_blockchain(bt.constants, 2) as (b2, _): timelord_api, _ = timelord for block in default_1000_blocks: await _validate_and_add_block(b1, block) await _validate_and_add_block(b2, block) await full_node.add_block(block) peak = timelord_peak_from_block(b1, default_1000_blocks[-1]) assert peak is not None assert timelord_api.timelord.new_peak is None await timelord_api.new_peak_timelord(peak) assert timelord_api.timelord.new_peak is not None await time_out_assert(60, tl_new_peak_is_none, True, timelord_api) assert timelord_api.timelord.last_state.peak is not None assert ( timelord_api.timelord.last_state.peak.reward_chain_block.get_hash() == peak.reward_chain_block.get_hash() ) # make two new blocks on tip, block_2 has higher total iterations block_1 = bt.get_consecutive_blocks(1, default_1000_blocks)[-1] block_2 = bt.get_consecutive_blocks( 1, default_1000_blocks, min_signage_point=block_1.reward_chain_block.signage_point_index )[-1] # make sure block_2 has higher iterations then block_1 assert block_2.total_iters > block_1.total_iters # make sure block_1 and block_2 have higher iterations then peak assert block_1.total_iters > default_1000_blocks[-1].total_iters await _validate_and_add_block(b1, block_1) await _validate_and_add_block(b2, block_2) block_record_1 = b1.block_record(block_1.header_hash) timelord_unf_block = timelord_protocol.NewUnfinishedBlockTimelord( block_1.reward_chain_block.get_unfinished(), uint64(block_record_1.weight - default_1000_blocks[-1].weight), block_record_1.sub_slot_iters, block_1.foliage, next_sub_epoch_summary(bt.constants, b1, block_record_1.required_iters, block_1, True), await get_rc_prev(b1, block_1), ) await timelord_api.new_unfinished_block_timelord(timelord_unf_block) overflow = is_overflow_block(bt.constants, timelord_unf_block.reward_chain_block.signage_point_index) assert last_unfinished(timelord_api, overflow=overflow).get_hash() == timelord_unf_block.get_hash() new_peak = timelord_peak_from_block(b2, block_2) # timelord knows unfinished block_1 that has lower iterations, # add block_2 peak and make sure we skip it and prefer to finish block_1 assert timelord_unf_block.reward_chain_block.total_iters <= new_peak.reward_chain_block.total_iters await timelord_api.new_peak_timelord(new_peak) await time_out_assert(60, tl_new_peak_is_none, True, timelord_api) # check that peak did not change assert ( timelord_api.timelord.last_state.peak.reward_chain_block.get_hash() == peak.reward_chain_block.get_hash() ) # check unfinished block_1 is still in cache overflow = is_overflow_block(bt.constants, timelord_unf_block.reward_chain_block.signage_point_index) assert last_unfinished(timelord_api, overflow=overflow).get_hash() == timelord_unf_block.get_hash() # full node gets block_1 unfinished block_1_unf = UnfinishedBlock( block_1.finished_sub_slots, block_1.reward_chain_block.get_unfinished(), block_1.challenge_chain_sp_proof, block_1.reward_chain_sp_proof, block_1.foliage, block_1.foliage_transaction_block, block_1.transactions_info, block_1.transactions_generator, [], ) await full_node.add_unfinished_block(block_1_unf, None) unf: UnfinishedBlock = full_node.full_node_store.get_unfinished_block(block_1_unf.partial_hash) assert unf.get_hash() == block_1_unf.get_hash() # full node peak is block_2 await full_node.add_block(block_2) curr = await full_node.blockchain.get_full_peak() assert block_2.header_hash == curr.header_hash # full_node gets finished block_1 response = timelord_protocol.NewInfusionPointVDF( block_1_unf.partial_hash, block_1.reward_chain_block.challenge_chain_ip_vdf, block_1.challenge_chain_ip_proof, block_1.reward_chain_block.reward_chain_ip_vdf, block_1.reward_chain_ip_proof, block_1.reward_chain_block.infused_challenge_chain_ip_vdf, block_1.infused_challenge_chain_ip_proof, ) await full_node.new_infusion_point_vdf(response) peak_after_unf_infusion = await full_node.blockchain.get_full_peak() # assert full node switched peak to block_1 since it has the same height as block_2 but lower iterations assert peak_after_unf_infusion.header_hash == block_1.header_hash @pytest.mark.anyio async def test_timelord_new_peak_unfinished_orphaned_overflow( self, bt: BlockTools, timelord: tuple[TimelordAPI, ChiaServer], default_1000_blocks: list[FullBlock] ) -> None: async with create_blockchain(bt.constants, 2) as (b1, _): async with create_blockchain(bt.constants, 2) as (b2, _): timelord_api, _ = timelord for block in default_1000_blocks: await _validate_and_add_block(b1, block) await _validate_and_add_block(b2, block) peak = timelord_peak_from_block(b1, default_1000_blocks[-1]) assert peak is not None assert timelord_api.timelord.new_peak is None await timelord_api.new_peak_timelord(peak) await time_out_assert(60, tl_new_peak_is_none, True, timelord_api) assert timelord_api.timelord.last_state.peak is not None assert ( timelord_api.timelord.last_state.peak.reward_chain_block.get_hash() == peak.reward_chain_block.get_hash() ) # make two new blocks on tip block_1 = bt.get_consecutive_blocks(1, default_1000_blocks, time_per_block=9, force_overflow=True)[-1] block_2 = bt.get_consecutive_blocks( 1, default_1000_blocks, seed=b"data", time_per_block=50, skip_slots=1, min_signage_point=block_1.reward_chain_block.signage_point_index, )[-1] # make sure block_2 has higher iterations assert block_2.total_iters >= block_1.total_iters await _validate_and_add_block(b1, block_1) await _validate_and_add_block(b2, block_2) block_record = b1.block_record(block_1.header_hash) sub_slot_iters, difficulty = get_next_sub_slot_iters_and_difficulty( bt.constants, len(block_1.finished_sub_slots) > 0, b1.block_record(block_1.prev_header_hash), b1, ) timelord_unf_block = timelord_protocol.NewUnfinishedBlockTimelord( block_1.reward_chain_block.get_unfinished(), difficulty, sub_slot_iters, block_1.foliage, next_sub_epoch_summary(bt.constants, b1, block_record.required_iters, block_1, True), await get_rc_prev(b1, block_1), ) await timelord_api.new_unfinished_block_timelord(timelord_unf_block) assert timelord_api.timelord.overflow_blocks[-1].get_hash() == timelord_unf_block.get_hash() new_peak = timelord_peak_from_block(b2, block_2) assert timelord_unf_block.reward_chain_block.total_iters <= new_peak.reward_chain_block.total_iters assert block_1.reward_chain_block.get_hash() != new_peak.reward_chain_block.get_hash() await timelord_api.new_peak_timelord(new_peak) await time_out_assert(60, tl_new_peak_is_none, True, timelord_api) assert ( timelord_api.timelord.last_state.peak.reward_chain_block.get_hash() == peak.reward_chain_block.get_hash() ) @pytest.mark.anyio async def test_timelord_new_peak_unfinished_eos( self, one_node: tuple[list[FullNodeService], list[FullNodeSimulator], BlockTools], timelord: tuple[TimelordAPI, ChiaServer], default_1000_blocks: list[FullBlock], ) -> None: [full_node_service], _, bt = one_node full_node = full_node_service._node async with create_blockchain(bt.constants, 2) as (b1, _): async with create_blockchain(bt.constants, 2) as (b2, _): timelord_api, _ = timelord for block in default_1000_blocks: await _validate_and_add_block(b1, block) await _validate_and_add_block(b2, block) await full_node.add_block(block) peak = timelord_peak_from_block(b1, default_1000_blocks[-1]) assert peak is not None assert timelord_api.timelord.new_peak is None await timelord_api.new_peak_timelord(peak) await time_out_assert(60, tl_new_peak_is_none, True, timelord_api) assert timelord_api.timelord.last_state.peak is not None assert ( timelord_api.timelord.last_state.peak.reward_chain_block.get_hash() == peak.reward_chain_block.get_hash() ) # make two new blocks on tip, block_2 is in a new slot block_1 = bt.get_consecutive_blocks(1, default_1000_blocks)[-1] block_2 = bt.get_consecutive_blocks( 1, default_1000_blocks, skip_slots=1, skip_overflow=True, seed=b"data" )[-1] # make sure block_2 has higher iterations assert block_2.total_iters >= block_1.total_iters await _validate_and_add_block(b1, block_1) await _validate_and_add_block(b2, block_2) await full_node.add_block(block_2) fn_peak = full_node.blockchain.get_peak() assert fn_peak is not None and fn_peak.header_hash == block_2.header_hash block_record = b2.block_record(block_2.header_hash) timelord_unf_block = timelord_protocol.NewUnfinishedBlockTimelord( block_2.reward_chain_block.get_unfinished(), uint64(block_record.weight - default_1000_blocks[-1].weight), block_record.sub_slot_iters, block_2.foliage, next_sub_epoch_summary(bt.constants, b1, block_record.required_iters, block_2, True), await get_rc_prev(b2, block_2), ) timelord_api.timelord.last_state.set_state(block_2.finished_sub_slots[-1]) # add unfinished and make sure we cache it await timelord_api.new_unfinished_block_timelord(timelord_unf_block) overflow = is_overflow_block(bt.constants, timelord_unf_block.reward_chain_block.signage_point_index) assert last_unfinished(timelord_api, overflow=overflow).get_hash() == timelord_unf_block.get_hash() new_peak = timelord_peak_from_block(b1, block_1) assert timelord_unf_block.reward_chain_block.total_iters >= new_peak.reward_chain_block.total_iters await timelord_api.new_peak_timelord(new_peak) await full_node.add_block(block_1) await time_out_assert(60, tl_new_peak_is_none, True, timelord_api) # make sure we switch to lower iteration peak assert ( timelord_api.timelord.last_state.peak.reward_chain_block.get_hash() == new_peak.reward_chain_block.get_hash() ) fn_peak = full_node.blockchain.get_peak() assert fn_peak is not None and fn_peak.header_hash == block_1.header_hash @pytest.mark.anyio async def test_timelord_new_peak_node_sync( self, one_node: tuple[list[FullNodeService], list[FullNodeSimulator], BlockTools], timelord: tuple[TimelordAPI, ChiaServer], default_1000_blocks: list[FullBlock], ) -> None: [full_node_service], _, bt = one_node full_node = full_node_service._node async with create_blockchain(bt.constants, 2) as (b1, _): async with create_blockchain(bt.constants, 2) as (b2, _): timelord_api, _ = timelord for block in default_1000_blocks: await _validate_and_add_block(b1, block) await _validate_and_add_block(b2, block) await full_node.add_block(block) peak = timelord_peak_from_block(b1, default_1000_blocks[-1]) assert peak is not None assert timelord_api.timelord.new_peak is None await timelord_api.new_peak_timelord(peak) assert timelord_api.timelord.new_peak is not None assert ( timelord_api.timelord.new_peak.reward_chain_block.get_hash() == peak.reward_chain_block.get_hash() ) await time_out_assert(60, tl_new_peak_is_none, True, timelord_api) # make two new blocks on tip, block_2 has higher total iterations block_1 = bt.get_consecutive_blocks(1, default_1000_blocks)[-1] block_2 = bt.get_consecutive_blocks( 1, default_1000_blocks, min_signage_point=block_1.reward_chain_block.signage_point_index )[-1] assert block_2.weight == block_1.weight # make sure block_2 has higher iterations then block_1 assert block_2.total_iters > block_1.total_iters # make sure block_1 and block_2 have higher iterations then peak assert block_1.total_iters > default_1000_blocks[-1].total_iters await full_node.add_block(block_2) await _validate_and_add_block(b1, block_2) peak_tl = timelord_peak_from_block(b1, block_2) peak = await full_node.blockchain.get_full_peak() assert peak is not None assert timelord_api.timelord.new_peak is None await timelord_api.new_peak_timelord(peak_tl) assert timelord_api.timelord.new_peak is not None assert peak.header_hash == block_2.header_hash assert peak_tl.reward_chain_block.get_hash() == peak.reward_chain_block.get_hash() await time_out_assert(60, tl_new_peak_is_none, True, timelord_api) await full_node.add_block(block_1) await _validate_and_add_block(b1, block_1) peak = timelord_peak_from_block(b1, block_1) assert peak is not None await timelord_api.new_peak_timelord(peak) peak = await full_node.blockchain.get_full_peak() assert peak == block_1 peak_tl = timelord_api.timelord.new_peak assert peak_tl.reward_chain_block.get_hash() == peak.reward_chain_block.get_hash() @pytest.mark.anyio @pytest.mark.parametrize("different_foliage", [False, True]) async def test_timelord_new_peak_is_in_unfinished_cache( self, one_node: tuple[list[FullNodeService], list[FullNodeSimulator], BlockTools], timelord: tuple[TimelordAPI, ChiaServer], default_1000_blocks: list[FullBlock], different_foliage: bool, ) -> None: _, _, bt = one_node wallet = WalletTool(bt.constants) coinbase_puzzlehash = wallet.get_new_puzzlehash() timelord_api, _ = timelord blocks = bt.get_consecutive_blocks( num_blocks=10, block_list_input=default_1000_blocks, skip_overflow=True, force_overflow=False, farmer_reward_puzzle_hash=coinbase_puzzlehash, guarantee_transaction_block=True, ) async with create_blockchain(bt.constants, 2) as (b1, _): for block in blocks: await _validate_and_add_block(b1, block) peak = timelord_peak_from_block(b1, blocks[-1]) assert peak is not None and timelord_api.timelord.new_peak is None await timelord_api.new_peak_timelord(peak) assert timelord_api.timelord.new_peak is not None await time_out_assert(60, tl_new_peak_is_none, True, timelord_api) assert ( timelord_api.timelord.last_state.peak is not None and timelord_api.timelord.last_state.peak.reward_chain_block.get_hash() == peak.reward_chain_block.get_hash() ) block_1 = bt.get_consecutive_blocks( block_list_input=blocks, num_blocks=1, farmer_reward_puzzle_hash=coinbase_puzzlehash, guarantee_transaction_block=True, skip_overflow=True, )[-1] await _validate_and_add_block(b1, block_1) block_record_1 = b1.block_record(block_1.header_hash) if len(block_1.finished_sub_slots) > 0: timelord_api.timelord.last_state.set_state(block_1.finished_sub_slots[-1]) block_1_diffrent_foliage = None if not different_foliage: timelord_unf_block = timelord_protocol.NewUnfinishedBlockTimelord( block_1.reward_chain_block.get_unfinished(), uint64(block_record_1.weight - blocks[-1].weight), block_record_1.sub_slot_iters, block_1.foliage, next_sub_epoch_summary(bt.constants, b1, block_record_1.required_iters, block_1, True), await get_rc_prev(b1, block_1), ) else: spend_coin = None for coin in blocks[-8].get_included_reward_coins(): if coin.puzzle_hash == coinbase_puzzlehash: spend_coin = coin assert spend_coin is not None sb = wallet.generate_signed_transaction(uint64(1000), bytes32(b"0" * 32), spend_coin) block_1_diffrent_foliage = bt.get_consecutive_blocks( block_list_input=blocks, num_blocks=1, farmer_reward_puzzle_hash=coinbase_puzzlehash, guarantee_transaction_block=True, transaction_data=sb, skip_overflow=True, )[-1] assert block_1_diffrent_foliage.header_hash != block_1.header_hash timelord_unf_block = timelord_protocol.NewUnfinishedBlockTimelord( block_1_diffrent_foliage.reward_chain_block.get_unfinished(), uint64(block_1_diffrent_foliage.weight - blocks[-1].weight), block_record_1.sub_slot_iters, block_1_diffrent_foliage.foliage, next_sub_epoch_summary(bt.constants, b1, block_record_1.required_iters, block_1, True), await get_rc_prev(b1, block_1), ) await timelord_api.new_unfinished_block_timelord(timelord_unf_block) overflow = is_overflow_block(bt.constants, timelord_unf_block.reward_chain_block.signage_point_index) assert last_unfinished(timelord_api, overflow=overflow).get_hash() == timelord_unf_block.get_hash() assert ( last_unfinished(timelord_api, overflow=overflow).reward_chain_block.get_hash() == timelord_unf_block.reward_chain_block.get_hash() ) new_peak = timelord_peak_from_block(b1, block_1) assert timelord_unf_block.reward_chain_block.total_iters == new_peak.reward_chain_block.total_iters await timelord_api.new_peak_timelord(new_peak) await time_out_assert(60, tl_new_peak_is_none, True, timelord_api) # check that peak was not skipped assert ( timelord_api.timelord.last_state.peak.reward_chain_block.get_hash() == new_peak.reward_chain_block.get_hash() ) # check unfinished block_1 is not in cache assert len(timelord_api.timelord.unfinished_blocks) == 0 async def get_rc_prev(blockchain: Blockchain, block: FullBlock) -> bytes32: if block.reward_chain_block.signage_point_index == 0: # find first in slot and find slot challenge blk = blockchain.block_record(block.header_hash) while blk.first_in_sub_slot is False: blk = blockchain.block_record(block.prev_header_hash) full_blk = await blockchain.get_full_block(blk.header_hash) assert full_blk is not None sub_slot = None for s in full_blk.finished_sub_slots: if s is not None and s.challenge_chain.get_hash() == block.reward_chain_block.pos_ss_cc_challenge_hash: sub_slot = s if sub_slot is None: assert block.reward_chain_block.pos_ss_cc_challenge_hash == blockchain.constants.GENESIS_CHALLENGE rc_prev = blockchain.constants.GENESIS_CHALLENGE else: rc_prev = sub_slot.reward_chain.get_hash() else: assert block.reward_chain_block.reward_chain_sp_vdf is not None rc_prev = block.reward_chain_block.reward_chain_sp_vdf.challenge return rc_prev def get_recent_reward_challenges(blockchain: Blockchain) -> list[tuple[bytes32, uint128]]: peak = blockchain.get_peak() if peak is None: return [] recent_rc: list[tuple[bytes32, uint128]] = [] curr: BlockRecord | None = peak while curr is not None and len(recent_rc) < 2 * blockchain.constants.MAX_SUB_SLOT_BLOCKS: if curr != peak: recent_rc.append((curr.reward_infusion_new_challenge, curr.total_iters)) if curr.first_in_sub_slot: assert curr.finished_reward_slot_hashes is not None sub_slot_total_iters = curr.ip_sub_slot_total_iters(blockchain.constants) # Start from the most recent for rc in reversed(curr.finished_reward_slot_hashes): if sub_slot_total_iters < curr.sub_slot_iters: break recent_rc.append((rc, sub_slot_total_iters)) sub_slot_total_iters = uint128(sub_slot_total_iters - curr.sub_slot_iters) curr = blockchain.try_block_record(curr.prev_hash) return list(reversed(recent_rc)) def timelord_peak_from_block( blockchain: Blockchain, block: FullBlock, ) -> timelord_protocol.NewPeakTimelord: peak = blockchain.block_record(block.header_hash) _, difficulty = get_next_sub_slot_iters_and_difficulty(blockchain.constants, False, peak, blockchain) ses: SubEpochSummary | None = next_sub_epoch_summary( blockchain.constants, blockchain, peak.required_iters, block, True ) curr = peak while not curr.is_challenge_block(blockchain.constants) and not curr.first_in_sub_slot: curr = blockchain.block_record(curr.prev_hash) if curr.is_challenge_block(blockchain.constants): last_csb_or_eos = curr.total_iters else: last_csb_or_eos = curr.ip_sub_slot_total_iters(blockchain.constants) curr = peak passed_ses_height_but_not_yet_included = True while (curr.height % blockchain.constants.SUB_EPOCH_BLOCKS) != 0: if curr.sub_epoch_summary_included: passed_ses_height_but_not_yet_included = False curr = blockchain.block_record(curr.prev_hash) if curr.sub_epoch_summary_included or curr.height == 0: passed_ses_height_but_not_yet_included = False return timelord_protocol.NewPeakTimelord( block.reward_chain_block, difficulty, peak.deficit, peak.sub_slot_iters, ses, get_recent_reward_challenges(blockchain), last_csb_or_eos, passed_ses_height_but_not_yet_included, ) def tl_new_peak_is_none(timelord: TimelordAPI) -> bool: return timelord.timelord.new_peak is None