Bound oracle approval polling and handle registry root change gracefully

src/config/settings.py

@@ -354,6 +354,10 @@ def network_config(self) -> NetworkConfig:
     'ORACLES_CONSOLIDATION_TIMEOUT', default=10, cast=int
 )
 ORACLES_EXITS_TIMEOUT: int = decouple_config('ORACLES_EXITS_TIMEOUT', default=10, cast=int)
+# Max number of polling attempts to collect oracle approvals/signatures before giving up
+ORACLES_APPROVALS_MAX_ATTEMPTS: int = decouple_config(
+    'ORACLES_APPROVALS_MAX_ATTEMPTS', default=10, cast=int
+)
 # withdrawals
 WITHDRAWALS_INTERVAL: int = decouple_config(
     'WITHDRAWALS_INTERVAL', default=43200, cast=int  # every 12 hr

src/validators/oracles.py

@@ -24,13 +24,14 @@
     warning_verbose,
 )
 from src.config.settings import (
+    ORACLES_APPROVALS_MAX_ATTEMPTS,
     ORACLES_CONSOLIDATION_TIMEOUT,
     ORACLES_EXITS_TIMEOUT,
     ORACLES_VALIDATORS_TIMEOUT,
     settings,
 )
 from src.validators.event_processors import get_validators_start_index
-from src.validators.exceptions import RegistryRootChangedError
+from src.validators.exceptions import ConsolidationError, RegistryRootChangedError
 from src.validators.keystores.base import BaseKeystore
 from src.validators.signing.common import get_encrypted_exit_signature_shards
 from src.validators.typings import ApprovalRequest, ConsolidationRequest, Validator
@@ -54,7 +55,7 @@ async def poll_validation_approval(
     approvals_min_interval = 1
     rate_limiter = RateLimiter(approvals_min_interval)
 
-    while True:
+    for attempt in range(1, ORACLES_APPROVALS_MAX_ATTEMPTS + 1):
         # Keep min interval between requests
         await rate_limiter.ensure_interval()
 
@@ -85,11 +86,18 @@ async def poll_validation_approval(
             oracles_approval = await send_approval_requests(protocol_config, oracles_request)
             return oracles_request, oracles_approval
         except NotEnoughOracleApprovalsError as e:
-            logger.error(
-                'Not enough oracle approvals for validator registration: %d. Threshold is %d.',
+            logger.warning(
+                'Not enough oracle approvals for validator registration: %d. Threshold is %d. '
+                'Attempt %d out of %d.',
                 e.num_votes,
                 e.threshold,
+                attempt,
+                ORACLES_APPROVALS_MAX_ATTEMPTS,
             )
+            if attempt >= ORACLES_APPROVALS_MAX_ATTEMPTS:
+                raise
+
+    raise RuntimeError('ORACLES_APPROVALS_MAX_ATTEMPTS must be >= 1')
 
 
 async def poll_consolidation_signature(
@@ -107,7 +115,7 @@ async def poll_consolidation_signature(
         public_keys=target_public_keys,
         vault_address=vault,
     )
-    while True:
+    for attempt in range(1, ORACLES_APPROVALS_MAX_ATTEMPTS + 1):
         # Keep min interval between requests
         await rate_limiter.ensure_interval()
 
@@ -117,10 +125,13 @@ async def poll_consolidation_signature(
         )
 
         if len(consolidation_signatures) < votes_threshold:
-            logger.error(
-                'Not enough oracle approvals for validator consolidation: %d. Threshold is %d.',
+            logger.warning(
+                'Not enough oracle approvals for validator consolidation: %d. Threshold is %d. '
+                'Attempt %d out of %d.',
                 len(consolidation_signatures),
                 votes_threshold,
+                attempt,
+                ORACLES_APPROVALS_MAX_ATTEMPTS,
             )
             continue
         signatures = b''
@@ -130,6 +141,11 @@ async def poll_consolidation_signature(
             signatures += signature
         return signatures
 
+    raise ConsolidationError(
+        'Not enough oracle approvals for validator consolidation after '
+        f'{ORACLES_APPROVALS_MAX_ATTEMPTS} attempts'
+    )
+
 
 async def send_approval_requests(
     protocol_config: ProtocolConfig, request: ApprovalRequest

src/validators/tasks.py

@@ -10,6 +10,7 @@
 
 from src.common.clients import execution_client
 from src.common.contracts import VaultContract, validators_registry_contract
+from src.common.exceptions import NotEnoughOracleApprovalsError
 from src.common.execution import check_gas_price
 from src.common.harvest import get_harvest_params
 from src.common.metrics import metrics
@@ -23,7 +24,11 @@
 from src.validators.consensus import fetch_funding_validators_balances
 from src.validators.database import NetworkValidatorCrud
 from src.validators.event_processors import get_validators_start_index
-from src.validators.exceptions import EmptyRelayerResponseException, FundingException
+from src.validators.exceptions import (
+    EmptyRelayerResponseException,
+    FundingException,
+    RegistryRootChangedError,
+)
 from src.validators.execution import (
     get_withdrawable_assets,
     tx_fund_validators,
@@ -183,7 +188,7 @@ async def fund_validators_chunk(
     return tx_hash
 
 
-# pylint: disable-next=too-many-locals
+# pylint: disable-next=too-many-locals,too-many-return-statements
 async def register_new_validators(
     vault_assets: Gwei,
     harvest_params: HarvestParams | None,
@@ -244,12 +249,22 @@ async def register_new_validators(
         'Started registration of %d %s validator(s)', len(validators), settings.validator_type.value
     )
 
-    oracles_request, oracles_approval = await poll_validation_approval(
-        keystore=keystore,
-        validators=validators,
-        validators_registry_root=validators_registry_root,
-        validators_manager_signature=validators_manager_signature,
-    )
+    try:
+        oracles_request, oracles_approval = await poll_validation_approval(
+            keystore=keystore,
+            validators=validators,
+            validators_registry_root=validators_registry_root,
+            validators_manager_signature=validators_manager_signature,
+        )
+    except RegistryRootChangedError:
+        logger.info('Validators registry root changed, retrying in the next cycle')
+        return None
+    except NotEnoughOracleApprovalsError:
+        logger.warning(
+            'Could not collect enough oracle approvals for validator registration this cycle'
+        )
+        return None
+
     tx_hash = await validate_index_and_register_validators(
         approval=oracles_approval,
         validators=validators,

src/validators/tests/test_oracles.py

@@ -0,0 +1,181 @@
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from eth_typing import HexStr
+from sw_utils.tests import faker
+from sw_utils.tests.factories import get_mocked_protocol_config
+from web3 import Web3
+
+from src.common.exceptions import NotEnoughOracleApprovalsError
+from src.common.typings import OraclesApproval
+from src.validators.exceptions import ConsolidationError, RegistryRootChangedError
+from src.validators.oracles import (
+    poll_consolidation_signature,
+    poll_validation_approval,
+)
+from src.validators.typings import ApprovalRequest
+
+
+@pytest.fixture(autouse=True)
+def _no_rate_limit_delay():
+    """RateLimiter sleeps between polling iterations; keep the loops instant in tests."""
+    with patch('src.common.utils.asyncio.sleep', new=AsyncMock()):
+        yield
+
+
+class TestPollValidationApproval:
+    async def test_returns_on_first_success(self):
+        """Returns the request/approval pair as soon as send_approval_requests succeeds."""
+        registry_root = faker.eth_proof()
+        approval_request = _fake_approval_request()
+        oracles_approval = OraclesApproval(signatures=b'\x01', ipfs_hash='ipfs', deadline=1)
+
+        with (
+            patch(
+                'src.validators.oracles.get_protocol_config',
+                new=AsyncMock(return_value=get_mocked_protocol_config()),
+            ),
+            patch(
+                'src.validators.oracles.validators_registry_contract.get_registry_root',
+                new=AsyncMock(return_value=registry_root),
+            ),
+            patch(
+                'src.validators.oracles.create_approval_request',
+                new=AsyncMock(return_value=approval_request),
+            ),
+            patch(
+                'src.validators.oracles.send_approval_requests',
+                new=AsyncMock(return_value=oracles_approval),
+            ),
+        ):
+            request, approval = await poll_validation_approval(
+                keystore=None,
+                validators=[],
+                validators_registry_root=registry_root,
+                validators_manager_signature=HexStr('0x'),
+            )
+
+        assert request is approval_request
+        assert approval is oracles_approval
+
+    async def test_raises_after_cap_attempts(self):
+        """Re-raises the last NotEnoughOracleApprovalsError once the attempt cap is exhausted."""
+        registry_root = faker.eth_proof()
+        approval_request = _fake_approval_request()
+        error = NotEnoughOracleApprovalsError(num_votes=1, threshold=2)
+        send_approval_requests_mock = AsyncMock(side_effect=error)
+
+        with (
+            patch(
+                'src.validators.oracles.get_protocol_config',
+                new=AsyncMock(return_value=get_mocked_protocol_config()),
+            ),
+            patch(
+                'src.validators.oracles.validators_registry_contract.get_registry_root',
+                new=AsyncMock(return_value=registry_root),
+            ),
+            patch(
+                'src.validators.oracles.create_approval_request',
+                new=AsyncMock(return_value=approval_request),
+            ),
+            patch('src.validators.oracles.send_approval_requests', new=send_approval_requests_mock),
+            patch('src.validators.oracles.ORACLES_APPROVALS_MAX_ATTEMPTS', 3),
+        ):
+            with pytest.raises(NotEnoughOracleApprovalsError):
+                await poll_validation_approval(
+                    keystore=None,
+                    validators=[],
+                    validators_registry_root=registry_root,
+                    validators_manager_signature=HexStr('0x'),
+                )
+
+        assert send_approval_requests_mock.call_count == 3
+
+    async def test_raises_registry_root_changed_immediately(self):
+        """Raises RegistryRootChangedError without ever sending approval requests."""
+        passed_registry_root = faker.eth_proof()
+        current_registry_root = faker.eth_proof()
+        send_approval_requests_mock = AsyncMock()
+
+        with (
+            patch(
+                'src.validators.oracles.get_protocol_config',
+                new=AsyncMock(return_value=get_mocked_protocol_config()),
+            ),
+            patch(
+                'src.validators.oracles.validators_registry_contract.get_registry_root',
+                new=AsyncMock(return_value=current_registry_root),
+            ),
+            patch('src.validators.oracles.send_approval_requests', new=send_approval_requests_mock),
+        ):
+            with pytest.raises(RegistryRootChangedError):
+                await poll_validation_approval(
+                    keystore=None,
+                    validators=[],
+                    validators_registry_root=passed_registry_root,
+                    validators_manager_signature=HexStr('0x'),
+                )
+
+        send_approval_requests_mock.assert_not_called()
+
+
+class TestPollConsolidationSignature:
+    async def test_raises_after_cap_attempts(self):
+        """Raises ConsolidationError once the attempt cap is exhausted below threshold."""
+        protocol_config = get_mocked_protocol_config(validators_threshold=2)
+        send_requests_mock = AsyncMock(return_value=[])
+
+        with (
+            patch('src.validators.oracles._send_consolidation_requests', new=send_requests_mock),
+            patch('src.validators.oracles.ORACLES_APPROVALS_MAX_ATTEMPTS', 3),
+        ):
+            with pytest.raises(ConsolidationError):
+                await poll_consolidation_signature(
+                    protocol_config=protocol_config,
+                    target_public_keys=[faker.validator_public_key()],
+                    vault=faker.eth_address(),
+                )
+
+        assert send_requests_mock.call_count == 3
+
+    async def test_returns_sorted_truncated_signatures_when_threshold_met(self):
+        """Concatenates signatures sorted by oracle address, truncated to the threshold."""
+        protocol_config = get_mocked_protocol_config(validators_threshold=2)
+        low_address = Web3.to_checksum_address('0x' + '01' * 20)
+        mid_address = Web3.to_checksum_address('0x' + '02' * 20)
+        high_address = Web3.to_checksum_address('0x' + '03' * 20)
+        low_signature = b'\x11'
+        mid_signature = b'\x22'
+        high_signature = b'\x33'
+        # Returned out of order and above threshold, to verify sorting and truncation both apply
+        consolidation_signatures = [
+            (high_address, high_signature),
+            (low_address, low_signature),
+            (mid_address, mid_signature),
+        ]
+
+        with patch(
+            'src.validators.oracles._send_consolidation_requests',
+            new=AsyncMock(return_value=consolidation_signatures),
+        ):
+            signatures = await poll_consolidation_signature(
+                protocol_config=protocol_config,
+                target_public_keys=[faker.validator_public_key()],
+                vault=faker.eth_address(),
+            )
+
+        assert signatures == low_signature + mid_signature
+
+
+def _fake_approval_request() -> ApprovalRequest:
+    return ApprovalRequest(
+        validator_index=0,
+        vault_address=faker.eth_address(),
+        validators_root=faker.eth_proof(),
+        public_keys=[],
+        deposit_signatures=[],
+        public_key_shards=[],
+        exit_signature_shards=[],
+        deadline=0,
+        validators_manager_signature=HexStr('0x'),
+    )