Add a test to cover new peak returning added transactions as a result of retrying potential transactions.

Author: AmineKhaldi

chia/_tests/core/mempool/test_mempool_manager.py

@@ -41,6 +41,7 @@
     QUOTE_BYTES,
     QUOTE_EXECUTION_COST,
     MempoolManager,
+    NewPeakItem,
     TimelockConditions,
     can_replace,
     check_removals,
@@ -3260,3 +3261,41 @@ async def test_different_ff_versions() -> None:
     latest_singleton_lineage2 = new_mi2.bundle_coin_spends[version2_id].latest_singleton_lineage
     assert latest_singleton_lineage2 is not None
     assert latest_singleton_lineage2.coin_id == new_latest_lineage_id
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("optimized_path", [True, False])
+async def test_new_peak_txs_added(optimized_path: bool) -> None:
+    """
+    Tests that deferred transactions because of time-lock are retried once the
+    time-lock allows them to be reconsidered.
+    """
+    coins = TestCoins([TEST_COIN], {})
+    mempool_manager = await setup_mempool(coins)
+    # Add an item that should go to the pending cache
+    assert mempool_manager.peak is not None
+    asserted_abs_height = mempool_manager.peak.height + 1
+    sb, sb_name, result = await generate_and_add_spendbundle(
+        mempool_manager, [[ConditionOpcode.ASSERT_HEIGHT_ABSOLUTE, asserted_abs_height]]
+    )
+    _, status, error = result
+    assert status == MempoolInclusionStatus.PENDING
+    assert error == Err.ASSERT_HEIGHT_ABSOLUTE_FAILED
+    # Advance the mempool beyond the asserted height to retry the test item
+    if optimized_path:
+        spent_coins: Optional[list[bytes32]] = []
+        new_peak_info = await mempool_manager.new_peak(
+            create_test_block_record(height=uint32(asserted_abs_height)), spent_coins
+        )
+        # We're not there yet (needs to be higher, not equal)
+        assert mempool_manager.get_mempool_item(sb_name, include_pending=False) is None
+        assert new_peak_info.items == []
+    else:
+        spent_coins = None
+    new_peak_info = await mempool_manager.new_peak(
+        create_test_block_record(height=uint32(asserted_abs_height + 1)), spent_coins
+    )
+    # The item gets retried successfully now
+    mi = mempool_manager.get_mempool_item(sb_name, include_pending=False)
+    assert mi is not None
+    assert new_peak_info.items == [NewPeakItem(sb_name, sb, mi.conds)]