fix(consume): harden enginex hive reporting against connection errors

packages/testing/src/execution_testing/cli/pytest_commands/plugins/consume/simulators/enginex/conftest.py

@@ -10,16 +10,21 @@
 import io
 import json
 import logging
+from pathlib import Path
 from typing import TYPE_CHECKING, Generator, cast
 
 import pytest
 from hive.client import Client, ClientType
-from hive.testing import HiveTest
+from hive.testing import HiveTest, HiveTestSuite
 
 from execution_testing.fixtures import BlockchainEngineXFixture
 from execution_testing.fixtures.blockchain import FixtureHeader
 from execution_testing.fixtures.pre_alloc_groups import PreAllocGroup
 
+from ....pytest_hive.reporting import (
+    retry_on_connection_error,
+    write_reported_test_count,
+)
 from ..helpers.test_tracker import (
     PreAllocGroupTestTracker,
     enginex_group_counts_key,
@@ -47,6 +52,10 @@
 def pytest_configure(config: pytest.Config) -> None:
     """Set the supported fixture formats for the enginex simulator."""
     config.supported_fixture_formats = [BlockchainEngineXFixture]  # type: ignore[attr-defined]
+    # Detect silent test loss: after the run, assert that every collected
+    # test reported its result to hive (see `_verify_reported_test_count`
+    # in the pytest_hive plugin).
+    config.assert_reported_test_count = True  # type: ignore[attr-defined]
 
 
 @pytest.hookimpl(trylast=True)
@@ -139,7 +148,29 @@ def _per_test_reporting(
     the hive node still exists, before `client` teardown calls
     `mark_test_completed` / `client.stop()`.
     """
-    hive_test.register_multi_test_client(client)
+    retry_on_connection_error(
+        "register multi-test client",
+        lambda: hive_test.register_multi_test_client(client),
+    )
+
+
+@pytest.fixture(scope="module", autouse=True)
+def _reported_test_count_flush(
+    test_suite: HiveTestSuite,
+    session_temp_folder: Path,
+    request: pytest.FixtureRequest,
+) -> Generator[None, None, None]:
+    """
+    Persist this worker's reported-test counts before suite teardown.
+
+    Depending on `test_suite` guarantees this fixture's teardown runs
+    before the suite teardown, in which the last-finishing xdist worker
+    aggregates all workers' counts and asserts that no test results
+    were silently lost (never started or never reported to hive).
+    """
+    del test_suite
+    yield
+    write_reported_test_count(request.config, session_temp_folder)
 
 
 @pytest.fixture(scope="function")

packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/__init__.py

@@ -0,0 +1 @@
+"""Hive pytest plugin providing common functionality for simulators."""

packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/pytest_hive.py

@@ -50,6 +50,13 @@
 from execution_testing.logging import get_logger
 
 from .hive_info import ClientFile, HiveInfo
+from .reporting import (
+    HiveReportedTestCountError,
+    count_skipped_test,
+    end_test_and_count,
+    retry_on_connection_error,
+    verify_reported_test_count,
+)
 
 logger = get_logger(__name__)
 
@@ -162,6 +169,8 @@ def pytest_runtest_makereport(
     outcome = yield
     report = outcome.get_result()
     setattr(item, f"result_{report.when}", report)
+    if report.when == "setup" and report.skipped:
+        count_skipped_test(item.config)
 
 
 @pytest.fixture(scope="session")
@@ -199,6 +208,7 @@ def get_test_suite_scope(fixture_name: str, config: pytest.Config) -> str:
 
 @pytest.fixture(scope=get_test_suite_scope)  # type: ignore[arg-type]
 def test_suite(
+    request: pytest.FixtureRequest,
     simulator: Simulation,
     session_temp_folder: Path,
     test_suite_name: str,
@@ -235,16 +245,23 @@ def test_suite(
 
     yield suite
 
+    count_check_error: str | None = None
     with FileLock(users_lock_file):
         with open(users_file, "r") as f:
             users = json.load(f)
         users -= 1
         with open(users_file, "w") as f:
             json.dump(users, f)
         if users == 0:
+            if getattr(request.config, "assert_reported_test_count", False):
+                count_check_error = verify_reported_test_count(
+                    request, suite, session_temp_folder
+                )
             suite.end()
             suite_file.unlink()
             users_file.unlink()
+    if count_check_error is not None:
+        raise HiveReportedTestCountError(count_check_error)
 
 
 @pytest.fixture(scope="module")
@@ -325,9 +342,12 @@ def hive_test(
         )
 
     test_parameter_string = request.node.name
-    test: HiveTest = test_suite.start_test(
-        name=test_parameter_string,
-        description=test_case_description,
+    test: HiveTest = retry_on_connection_error(
+        f"start test {test_parameter_string}",
+        lambda: test_suite.start_test(
+            name=test_parameter_string,
+            description=test_case_description,
+        ),
     )
     yield test
 
@@ -403,10 +423,10 @@ def hive_test(
                 "unknown).\n\n" + captured_output
             )
 
-        test.end(
-            result=HiveTestResult(
-                test_pass=test_passed, details=test_result_details
-            )
+        end_test_and_count(
+            request.config,
+            test,
+            HiveTestResult(test_pass=test_passed, details=test_result_details),
         )
         logger.verbose(
             f"Finished processing logs for test: {request.node.nodeid}"
@@ -420,8 +440,8 @@ def hive_test(
         test_result_details = (
             f"Exception whilst processing test result: {str(e)}"
         )
-        test.end(
-            result=HiveTestResult(
-                test_pass=test_passed, details=test_result_details
-            )
+        end_test_and_count(
+            request.config,
+            test,
+            HiveTestResult(test_pass=test_passed, details=test_result_details),
         )

packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/reporting.py

@@ -0,0 +1,197 @@
+"""
+Helpers guarding hive test-result reporting against connection errors
+and silent test loss.
+
+At very high test throughput (e.g. `consume-enginex`), the simulator
+can transiently exhaust its ephemeral port range toward the hive API
+endpoint: `connect()` then fails with `EADDRNOTAVAIL` for a few seconds
+on all xdist workers at once. Tests in flight during such a burst
+either error at setup (`start_test` failed: the test is then silently
+missing from the hive results), or run and pass but never report their
+verdict (`end_test` failed: the dangling hive test case is later
+force-closed as "Test was terminated by host").
+
+The root fix is connection pooling in the `ethereum-hive` package;
+these helpers are the complementary hardening on the simulator side:
+
+- `retry_on_connection_error` retries hive API calls with backoff so a
+  seconds-long burst does not error tests or lose results.
+- The reported-test counting (`end_test_and_count`,
+  `write_reported_test_count`, `verify_reported_test_count`) detects
+  any remaining silent loss by comparing the number of test results
+  successfully reported to hive against the number of collected tests
+  once the run has finished.
+"""
+
+import json
+import os
+import time
+from pathlib import Path
+from typing import Callable, Dict, TypeVar
+
+import pytest
+import requests
+from hive.testing import HiveTest, HiveTestResult, HiveTestSuite
+
+from execution_testing.logging import get_logger
+
+logger = get_logger(__name__)
+
+reported_test_count_key = pytest.StashKey[int]()
+"""Tests whose results this process successfully reported to hive."""
+
+skipped_test_count_key = pytest.StashKey[int]()
+"""Tests this process skipped during setup (never reported to hive)."""
+
+T = TypeVar("T")
+
+
+class HiveReportedTestCountError(Exception):
+    """Fewer test results than collected tests were reported to hive."""
+
+
+def retry_on_connection_error(
+    description: str,
+    call: Callable[[], T],
+    *,
+    attempts: int = 5,
+    initial_backoff: float = 0.5,
+) -> T:
+    """
+    Call `call`, retrying connection errors with exponential backoff.
+
+    A safety net for hive API calls during transient client-side
+    ephemeral-port exhaustion (`EADDRNOTAVAIL`) bursts; see the module
+    docstring. Connection errors are raised before the request has been
+    sent, so retrying cannot duplicate a non-idempotent call.
+    """
+    backoff = initial_backoff
+    for attempt in range(1, attempts + 1):
+        try:
+            return call()
+        except requests.exceptions.ConnectionError as e:
+            if attempt == attempts:
+                raise
+            logger.warning(
+                f"Hive API call '{description}' raised a connection error "
+                f"(attempt {attempt}/{attempts}), retrying in "
+                f"{backoff:.1f}s: {e}"
+            )
+            time.sleep(backoff)
+            backoff *= 2
+    raise AssertionError("unreachable")
+
+
+def end_test_and_count(
+    config: pytest.Config, test: HiveTest, result: HiveTestResult
+) -> None:
+    """End a hive test with retry and count the reported result."""
+    retry_on_connection_error(
+        f"end test {test.id}", lambda: test.end(result=result)
+    )
+    config.stash[reported_test_count_key] = (
+        config.stash.get(reported_test_count_key, 0) + 1
+    )
+
+
+def count_skipped_test(config: pytest.Config) -> None:
+    """
+    Count a test skipped during setup.
+
+    Such tests never start a hive test; counting them lets the
+    collected-vs-reported check account for them.
+    """
+    config.stash[skipped_test_count_key] = (
+        config.stash.get(skipped_test_count_key, 0) + 1
+    )
+
+
+def write_reported_test_count(
+    config: pytest.Config, session_temp_folder: Path
+) -> None:
+    """
+    Persist this process's reported/skipped test counts to the shared
+    session folder so that the last-finishing xdist worker can verify
+    the total against the number of collected tests.
+    """
+    worker_id = os.environ.get("PYTEST_XDIST_WORKER", "master")
+    counts = {
+        "reported": config.stash.get(reported_test_count_key, 0),
+        "skipped": config.stash.get(skipped_test_count_key, 0),
+    }
+    count_file = (
+        session_temp_folder / f"hive_reported_test_count_{worker_id}.json"
+    )
+    with open(count_file, "w") as f:
+        json.dump(counts, f)
+
+
+def verify_reported_test_count(
+    request: pytest.FixtureRequest,
+    suite: HiveTestSuite,
+    session_temp_folder: Path,
+) -> str | None:
+    """
+    Verify that every collected test reported a result to hive.
+
+    Return an error message if results were lost, `None` otherwise. On
+    loss, additionally report a failing meta-test case to hive so that
+    the loss is visible in the hive results, where the affected tests
+    are otherwise silently missing.
+    """
+    session = request.session
+    if session.shouldstop or session.shouldfail:
+        logger.warning(
+            "Test run was interrupted; skipping the collected-vs-reported "
+            "hive test count check."
+        )
+        return None
+    collected = session.testscollected
+    reported = 0
+    skipped = 0
+    worker_counts: Dict[str, Dict[str, int]] = {}
+    for count_file in sorted(
+        session_temp_folder.glob("hive_reported_test_count_*.json")
+    ):
+        with open(count_file, "r") as f:
+            counts = json.load(f)
+        worker_id = count_file.stem.removeprefix("hive_reported_test_count_")
+        worker_counts[worker_id] = counts
+        reported += counts["reported"]
+        skipped += counts["skipped"]
+        count_file.unlink()
+    if reported + skipped >= collected:
+        if reported + skipped > collected:
+            logger.warning(
+                f"More test results than collected tests: "
+                f"{reported} reported + {skipped} skipped > {collected} "
+                f"collected ({worker_counts})."
+            )
+        else:
+            logger.info(
+                f"All {collected} collected tests are accounted for "
+                f"({reported} reported to hive, {skipped} skipped)."
+            )
+        return None
+    missing = collected - reported - skipped
+    message = (
+        f"{missing} test result(s) were not reported to hive: collected "
+        f"{collected} tests, but only {reported} result(s) were reported "
+        f"and {skipped} test(s) were skipped (per-worker counts: "
+        f"{worker_counts}). Results are typically lost when hive API "
+        f"calls fail, e.g. due to client startup failures or ephemeral "
+        f"port exhaustion (EADDRNOTAVAIL) at high test throughput."
+    )
+    logger.error(message)
+    try:
+        check_test = suite.start_test(
+            name="reported-test-count-check",
+            description=(
+                "Meta-test verifying that every collected test reported "
+                "its result to hive; fails if test results were lost."
+            ),
+        )
+        check_test.end(result=HiveTestResult(test_pass=False, details=message))
+    except Exception as e:
+        logger.error(f"Failed to report the test count mismatch: {e}")
+    return message

packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/tests/__init__.py

@@ -0,0 +1 @@
+"""Tests for the pytest_hive plugin."""