chore(consume): mark sync tests flaky (#2252)

* chore(consume): Mark all consume sync tests as flaky

* refactor: simplify consume sync logic

* chore: CHANGELOG entry

Author: fselmo

docs/CHANGELOG.md

@@ -22,12 +22,13 @@ Test fixtures for use by clients are available for each release on the [Github r
 #### `consume`
 
 - ✨ Add retry logic to RPC requests to fix flaky connection issues in Hive ([#2205](https://github.com/ethereum/execution-spec-tests/pull/2205)).
+- 🛠️ Mark `consume sync` tests as `flaky` with 3 retires due to client sync inconsistencies ([#2252](https://github.com/ethereum/execution-spec-tests/pull/2252)).
 
 ### 📋 Misc
 
 - ✨ Add tighter validation for EIP-7928 model coming from t8n when filling ([#2138](https://github.com/ethereum/execution-spec-tests/pull/2138)).
 - ✨ Add flexible API for absence checks for EIP-7928 (BAL) tests ([#2124](https://github.com/ethereum/execution-spec-tests/pull/2124)).
-- 🐞 Use ``engine_newPayloadV5`` for `>=Amsterdam` forks in `consume engine` ([#2170](https://github.com/ethereum/execution-spec-tests/pull/2170)).
+- 🐞 Use `engine_newPayloadV5` for `>=Amsterdam` forks in `consume engine` ([#2170](https://github.com/ethereum/execution-spec-tests/pull/2170)).
 - 🔀 Refactor EIP-7928 (BAL) absence checks into a friendlier class-based DevEx ([#2175](https://github.com/ethereum/execution-spec-tests/pull/2175)).
 
 ### 🧪 Test Cases

pyproject.toml

@@ -46,6 +46,7 @@ dependencies = [
     "questionary>=2.1.0,<3",
     "ethereum-rlp>=0.1.3,<0.2",
     "pytest-regex>=0.2.0,<0.3",
+    "pytest-rerunfailures>=16.0,<17",
     "eth-abi>=5.2.0",
     "joblib>=1.4.2",
     "ckzg>=2.1.3,<3",

src/pytest_plugins/consume/simulators/simulator_logic/test_via_sync.py

@@ -16,7 +16,6 @@
 
 import pytest
 
-from ethereum_test_base_types import Hash
 from ethereum_test_exceptions import UndefinedException
 from ethereum_test_fixtures import BlockchainEngineSyncFixture
 from ethereum_test_rpc import AdminRPC, EngineRPC, EthRPC, NetRPC
@@ -42,65 +41,6 @@ def __init__(self, *args: object) -> None:
         logger.fail(str(self))
 
 
-def wait_for_sync(
-    sync_eth_rpc: EthRPC,
-    expected_block_hash: str | Hash,
-    timeout: int = 10,
-    poll_interval: float = 1.0,
-) -> bool:
-    """Wait for the sync client to reach the expected block hash."""
-    start_time = time.time()
-    last_block_number = 0
-    no_progress_count = 0
-
-    while time.time() - start_time < timeout:
-        try:
-            # First check if we have the expected block
-            block = sync_eth_rpc.get_block_by_hash(Hash(expected_block_hash))
-            if block is not None:
-                logger.info(f"Sync complete! Client has block {expected_block_hash}")
-                return True
-
-            # Check current sync progress
-            current_block = sync_eth_rpc.get_block_by_number("latest")
-            if current_block:
-                current_number = int(current_block.get("number", "0x0"), 16)
-                current_hash = current_block.get("hash", "unknown")
-                if current_number > last_block_number:
-                    logger.info(f"Sync progress: block {current_number} (hash: {current_hash})")
-                    last_block_number = current_number
-                    no_progress_count = 0
-                else:
-                    no_progress_count += 1
-                    if no_progress_count == 1:
-                        logger.info(
-                            f"Sync client is at block {current_number} (hash: {current_hash})"
-                        )
-                    elif no_progress_count % 10 == 0:
-                        logger.debug(
-                            f"No sync progress for {no_progress_count} polls, "
-                            f"still at block {current_number}"
-                        )
-
-        except Exception as e:
-            logger.debug(f"Error checking sync status: {e}")
-
-        time.sleep(poll_interval)
-
-    # Log final state
-    try:
-        final_block = sync_eth_rpc.get_block_by_number("latest")
-        if final_block:
-            logger.warning(
-                f"Sync timeout! Final block: {final_block.get('number', 'unknown')} "
-                f"(hash: {final_block.get('hash', 'unknown')})"
-            )
-    except Exception:
-        pass
-
-    return False
-
-
 def test_blockchain_via_sync(
     timing_data: TimingData,
     eth_rpc: EthRPC,
@@ -445,7 +385,7 @@ def test_blockchain_via_sync(
         last_forkchoice_time = time.time()
         forkchoice_interval = 2.0  # Send forkchoice updates every 2 seconds
 
-        while time.time() - sync_start_time < 60:  # 60 second timeout
+        while time.time() - sync_start_time < 15:  # 15 second timeout
             # Send periodic forkchoice updates to keep sync alive
             if time.time() - last_forkchoice_time >= forkchoice_interval:
                 try:
@@ -473,35 +413,35 @@ def test_blockchain_via_sync(
                 f"within timeout"
             )
 
-        # Final verification
+        logger.info("Sync verification successful!")
+
+        # Verify the final state but give a few tries
+        assert eth_rpc is not None, "eth_rpc is required"
         assert sync_eth_rpc is not None, "sync_eth_rpc is required"
-        assert sync_engine_rpc is not None, "sync_engine_rpc is required"
-        if wait_for_sync(sync_eth_rpc, last_valid_block_hash, timeout=5):
-            logger.info("Sync verification successful!")
-
-            # Verify the final state
-            sync_block = sync_eth_rpc.get_block_by_hash(last_valid_block_hash)
-            client_block = eth_rpc.get_block_by_hash(last_valid_block_hash)
-
-            if sync_block["stateRoot"] != client_block["stateRoot"]:
-                raise LoggedError(
-                    f"State root mismatch after sync. "
-                    f"Sync client: {sync_block['stateRoot']}, "
-                    f"Client under test: {client_block['stateRoot']}"
-                )
 
-            # Verify post state if available
-            if fixture.post_state_hash:
-                if sync_block["stateRoot"] != str(fixture.post_state_hash):
+        for attempt in range(3):
+            try:
+                sync_block = sync_eth_rpc.get_block_by_hash(last_valid_block_hash)
+                client_block = eth_rpc.get_block_by_hash(last_valid_block_hash)
+
+                if sync_block["stateRoot"] != client_block["stateRoot"]:
                     raise LoggedError(
-                        f"Final state root mismatch. "
-                        f"Expected: {fixture.post_state_hash}, "
-                        f"Got: {sync_block['stateRoot']}"
+                        f"State root mismatch after sync. "
+                        f"Sync client: {sync_block['stateRoot']}, "
+                        f"Client under test: {client_block['stateRoot']}"
                     )
-        else:
-            raise LoggedError(
-                f"Sync client failed to synchronize to block {last_valid_block_hash} "
-                f"within timeout"
-            )
+
+                if fixture.post_state_hash:
+                    if sync_block["stateRoot"] != str(fixture.post_state_hash):
+                        raise LoggedError(
+                            f"Final state root mismatch. "
+                            f"Expected: {fixture.post_state_hash}, "
+                            f"Got: {sync_block['stateRoot']}"
+                        )
+            except Exception as e:
+                if attempt < 2:
+                    time.sleep(1)
+                    continue
+                raise e
 
     logger.info("Sync test completed successfully!")

src/pytest_plugins/consume/simulators/sync/conftest.py

@@ -47,6 +47,10 @@ def pytest_collection_modifyitems(session, config, items):
     del session, config
 
     for item in items:
+        # Auto-mark all verify_sync tests as flaky with 3 reruns
+        if item.get_closest_marker("blockchain_test_sync"):
+            item.add_marker(pytest.mark.flaky(reruns=3))
+
         # Check if this test has both client_type and sync_client_type
         if (
             hasattr(item, "callspec")