Make attestation consensus threshold configurable

docs/running_vero.md

@@ -54,6 +54,24 @@ exclusively use for block proposals. When performing a block proposal duty,
 only these beacon nodes will be used to produce and publish a block.
 ___
 
+#### `--attestation-consensus-threshold`
+
+Specify the required number of beacon nodes that need to agree
+on the attestation data before the validators proceed to attest.
+
+There are a few situations where you may want to change the default:
+- when running against 2 beacon nodes, where you only want to use
+  the second node as a fallback
+- when running against a large client-diverse set of beacon nodes
+  where a lower threshold (like 2 or 3 out of 6 beacon nodes agreeing)
+  may be sufficient to avoid single-client bugs.
+
+See https://github.com/serenita-org/vero/issues/38 for more
+information.
+
+Defaults to a majority of beacon nodes (>50%) agreeing.
+___
+
 #### `--fee-recipient`
 
 **[required]** The fee recipient address to use during block proposals.

docs/using_multiple_beacon_nodes.md

@@ -16,6 +16,11 @@ beacon nodes to be temporarily offline, whether that's
 because of an unexpected technical issue or for planned
 maintenance.
 
+Note: it is possible to override the default mode of
+reaching consensus on attestation data among a majority
+of the beacon nodes using the
+`--attestation-consensus-threshold` CLI parameter.
+
 ## Attestations
 
 When the time comes to attest to the head of the chain,

src/args.py

@@ -9,16 +9,17 @@
 class CLIArgs(msgspec.Struct, kw_only=True):
     remote_signer_url: str
     beacon_node_urls: list[str]
-    beacon_node_urls_proposal: list[str] = []
+    beacon_node_urls_proposal: list[str]
+    attestation_consensus_threshold: int
     fee_recipient: str
     data_dir: str
     graffiti: bytes
     gas_limit: int
-    use_external_builder: bool = False
+    use_external_builder: bool
     builder_boost_factor: int
     metrics_address: str
     metrics_port: int
-    metrics_multiprocess_mode: bool = False
+    metrics_multiprocess_mode: bool
     log_level: str
 
 
@@ -40,6 +41,25 @@ def _validate_comma_separated_strings(
     return items
 
 
+def _process_attestation_consensus_threshold(
+    value: int | None, beacon_node_urls: list[str]
+) -> int:
+    if value is None:
+        # If no value provided, default to a majority of beacon nodes
+        return len(beacon_node_urls) // 2 + 1
+
+    if value <= 0:
+        raise ValueError(f"Invalid value for attestation_consensus_threshold: {value}")
+
+    if len(beacon_node_urls) < value:
+        raise ValueError(
+            f"Invalid value for attestation_consensus_threshold ({value})"
+            f" with {len(beacon_node_urls)} beacon node(s)"
+        )
+
+    return value
+
+
 def _process_fee_recipient(input_string: str) -> str:
     _fee_recipient_byte_length = 20
 
@@ -87,6 +107,13 @@ def parse_cli_args(args: Sequence[str]) -> CLIArgs:
         default="",
         help="A comma-separated list of beacon node URLs to exclusively use for block proposals.",
     )
+    parser.add_argument(
+        "--attestation-consensus-threshold",
+        type=int,
+        required=False,
+        default=None,
+        help="Specify the required number of beacon nodes that need to agree on the attestation data before the validators proceed to attest. Defaults to a majority of beacon nodes (>50%) agreeing.",
+    )
     parser.add_argument(
         "--fee-recipient",
         type=str,
@@ -157,16 +184,17 @@ def parse_cli_args(args: Sequence[str]) -> CLIArgs:
 
     try:
         # Process and validate parsed args
+        beacon_node_urls = [
+            _validate_url(url)
+            for url in _validate_comma_separated_strings(
+                input_string=parsed_args.beacon_node_urls,
+                entity_name="beacon node url",
+                min_values_required=1,
+            )
+        ]
         return CLIArgs(
             remote_signer_url=_validate_url(parsed_args.remote_signer_url),
-            beacon_node_urls=[
-                _validate_url(url)
-                for url in _validate_comma_separated_strings(
-                    input_string=parsed_args.beacon_node_urls,
-                    entity_name="beacon node url",
-                    min_values_required=1,
-                )
-            ],
+            beacon_node_urls=beacon_node_urls,
             beacon_node_urls_proposal=[
                 _validate_url(url)
                 for url in _validate_comma_separated_strings(
@@ -175,6 +203,10 @@ def parse_cli_args(args: Sequence[str]) -> CLIArgs:
                     min_values_required=0,
                 )
             ],
+            attestation_consensus_threshold=_process_attestation_consensus_threshold(
+                value=parsed_args.attestation_consensus_threshold,
+                beacon_node_urls=beacon_node_urls,
+            ),
             fee_recipient=_process_fee_recipient(parsed_args.fee_recipient),
             data_dir=parsed_args.data_dir,
             graffiti=_process_graffiti(parsed_args.graffiti),

src/initialize.py

@@ -107,6 +107,7 @@ async def run_services(cli_args: CLIArgs) -> None:
             beacon_node_urls=cli_args.beacon_node_urls,
             beacon_node_urls_proposal=cli_args.beacon_node_urls_proposal,
             scheduler=scheduler,
+            cli_args=cli_args,
         ) as multi_beacon_node,
     ):
         beacon_chain = BeaconChain(multi_beacon_node=multi_beacon_node)

src/providers/multi_beacon_node.py

@@ -1,8 +1,8 @@
 """Uses multiple beacon nodes to provide the validator client with data.
 
 The biggest advantage of using multiple beacon nodes is that we can
-request attestation data from all of them, and only attest if a majority
-of them agrees on the state of the chain, providing resilience against
+request attestation data from all of them, and only attest if enough
+of them agree on the state of the chain, providing resilience against
 single-client bugs.
 
 This provider has 2 important internal methods:
@@ -46,6 +46,7 @@
 from opentelemetry import trace
 from remerkleable.complex import Container
 
+from args import CLIArgs
 from observability import ErrorType, get_shared_metrics
 from providers.beacon_node import BeaconNode
 from schemas import SchemaBeaconAPI, SchemaValidator
@@ -66,6 +67,7 @@ def __init__(
         beacon_node_urls: list[str],
         beacon_node_urls_proposal: list[str],
         scheduler: AsyncIOScheduler,
+        cli_args: CLIArgs,
     ):
         self.logger = logging.getLogger(self.__class__.__name__)
         self.logger.setLevel(logging.getLogger().level)
@@ -80,21 +82,15 @@ def __init__(
             BeaconNode(base_url=base_url, scheduler=scheduler)
             for base_url in beacon_node_urls_proposal
         ]
-        # TODO Consider renaming to consensus_threshold
-        #  allowing for overrides of this value through a CLI argument.
-        #  Usecase would be larger node operators, running say 5 nodes with
-        #  diverse client combinations.
-        #  They could use this override to attest as soon as 2 of them agree
-        #  and don't need to wait for the 3rd confirmation. This would allow
-        #  them to attest quickly yet still avoid single-client bugs.
-        self._majority_threshold = len(self.beacon_nodes) // 2 + 1
+
+        self._attestation_consensus_threshold = cli_args.attestation_consensus_threshold
 
     async def initialize(self) -> None:
         # Attempt to fully initialize the connected beacon nodes
         await asyncio.gather(*(bn.initialize_full() for bn in self.beacon_nodes))
 
         successfully_initialized = len([b for b in self.beacon_nodes if b.initialized])
-        if successfully_initialized < self._majority_threshold:
+        if successfully_initialized < self._attestation_consensus_threshold:
             raise RuntimeError(
                 f"Failed to fully initialize a sufficient amount of beacon nodes -"
                 f" {successfully_initialized}/{len(self.beacon_nodes)} initialized",
@@ -454,7 +450,7 @@ async def _produce_attestation_data_from_head_event(
             try:
                 att_data = await coro
                 head_match_count += 1
-                if head_match_count >= self._majority_threshold:
+                if head_match_count >= self._attestation_consensus_threshold:
                     # Cancel pending tasks
                     for task in tasks:
                         task.cancel()
@@ -509,7 +505,10 @@ async def _produce_attestation_data_without_head_event(
 
                 block_root = att_data.beacon_block_root.to_obj()
                 head_block_root_counter[block_root] += 1
-                if head_block_root_counter[block_root] >= self._majority_threshold:
+                if (
+                    head_block_root_counter[block_root]
+                    >= self._attestation_consensus_threshold
+                ):
                     # Cancel pending tasks
                     for task in tasks:
                         task.cancel()
@@ -533,10 +532,10 @@ async def _produce_attestation_data(
         # Slightly different algorithms depending on whether
         # a head event has been emitted.
         # A) A head event was emitted
-        #    We wait for a majority of beacon nodes to report the same
+        #    We wait for enough beacon nodes to report the same
         #    head block root as is present in the head event.
         # B) No head event was emitted
-        #    We wait for a majority of beacon nodes to report the same
+        #    We wait for enough beacon nodes to report the same
         #    head block root.
         if head_event:
             return await self._produce_attestation_data_from_head_event(
@@ -560,17 +559,17 @@ async def produce_attestation_data(
     ) -> AttestationData:
         """Returns attestation data from the connected beacon nodes.
 
-        If a head event is provided, the function will wait until a majority of beacon nodes
+        If a head event is provided, the function will wait until enough beacon nodes
         has processed the same head block.
 
         Some example situations that can occur and how they are handled:
         - 2s into the slot, we receive a head event from one beacon node,
           but the rest of connected beacon nodes hasn't processed that block yet
-          --> we wait for a majority of beacon nodes to report the same head block
+          --> we wait for enough beacon nodes to report the same head block
               (even if that means submitting the attestation later than 4s into the slot)
         - 4s into the slot, we haven't received a head event.
-          --> We request all beacon nodes to produce attestation data and wait until a majority
-              of beacon nodes agrees on a head block. Then we attest to that.
+          --> We request all beacon nodes to produce attestation data and wait until enough
+              beacon nodes agrees on a head block. Then we attest to that.
         """
         with self.tracer.start_as_current_span(
             name=f"{self.__class__.__name__}.produce_attestation_data",

tests/conftest.py

@@ -5,7 +5,7 @@
 import pytest
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 
-from args import CLIArgs
+from args import CLIArgs, _process_attestation_consensus_threshold
 from observability import init_observability
 from providers import BeaconChain, MultiBeaconNode, RemoteSigner
 from schemas import SchemaBeaconAPI
@@ -39,6 +39,9 @@ def cli_args(
         remote_signer_url=remote_signer_url,
         beacon_node_urls=[beacon_node_url],
         beacon_node_urls_proposal=beacon_node_urls_proposal,
+        attestation_consensus_threshold=_process_attestation_consensus_threshold(
+            None, [beacon_node_url]
+        ),
         fee_recipient="0x0000000000000000000000000000000000000000",
         data_dir="/tmp/vero_tests",
         use_external_builder=False,
@@ -47,6 +50,7 @@ def cli_args(
         gas_limit=30_000_000,
         metrics_address="localhost",
         metrics_port=8000,
+        metrics_multiprocess_mode=False,
         log_level="INFO",
     )
 
@@ -145,6 +149,7 @@ async def multi_beacon_node(
         beacon_node_urls=cli_args.beacon_node_urls,
         beacon_node_urls_proposal=cli_args.beacon_node_urls_proposal,
         scheduler=scheduler,
+        cli_args=cli_args,
     ) as mbn:
         yield mbn
 

tests/providers/conftest.py

@@ -5,6 +5,7 @@
 from aioresponses import CallbackResult, aioresponses
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 
+from args import CLIArgs
 from providers import MultiBeaconNode
 from spec.base import SpecDeneb
 
@@ -15,6 +16,7 @@ async def multi_beacon_node_three_inited_nodes(
     mocked_genesis_response: dict,  # type: ignore[type-arg]
     spec_deneb: SpecDeneb,
     scheduler: AsyncIOScheduler,
+    cli_args: CLIArgs,
 ) -> AsyncGenerator[MultiBeaconNode, None]:
     mbn = MultiBeaconNode(
         beacon_node_urls=[
@@ -24,6 +26,7 @@ async def multi_beacon_node_three_inited_nodes(
         ],
         beacon_node_urls_proposal=[],
         scheduler=scheduler,
+        cli_args=cli_args,
     )
     with aioresponses() as m:
         m.get(

tests/providers/test_multi_beacon_node.py

@@ -15,6 +15,7 @@
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 from remerkleable.bitfields import Bitlist, Bitvector
 
+from args import CLIArgs, _process_attestation_consensus_threshold
 from providers import MultiBeaconNode
 from spec.attestation import Attestation, AttestationData
 from spec.base import SpecDeneb
@@ -23,7 +24,7 @@
 
 @pytest.mark.parametrize(
     argnames=[
-        "beacon_node_base_urls",
+        "beacon_node_urls",
         "beacon_node_availabilities",
         "expected_initialization_success",
     ],
@@ -61,28 +62,33 @@
     ],
 )
 async def test_initialize(
-    beacon_node_base_urls: list[str],
+    beacon_node_urls: list[str],
     beacon_node_availabilities: list[bool],
     expected_initialization_success: bool,
     mocked_fork_response: dict,  # type: ignore[type-arg]
     mocked_genesis_response: dict,  # type: ignore[type-arg]
     spec_deneb: SpecDeneb,
     scheduler: AsyncIOScheduler,
+    cli_args: CLIArgs,
 ) -> None:
-    """Tests that the multi-beacon node is able to initialize if a majority
-    of its supplied beacon nodes is available.
+    """Tests that the multi-beacon node is able to initialize if enough
+    of its supplied beacon nodes are available.
     """
-    assert len(beacon_node_base_urls) == len(beacon_node_availabilities)
+    assert len(beacon_node_urls) == len(beacon_node_availabilities)
+    cli_args.attestation_consensus_threshold = _process_attestation_consensus_threshold(
+        None, beacon_node_urls
+    )
 
     mbn = MultiBeaconNode(
-        beacon_node_urls=beacon_node_base_urls,
+        beacon_node_urls=beacon_node_urls,
         beacon_node_urls_proposal=[],
         scheduler=scheduler,
+        cli_args=cli_args,
     )
 
     with aioresponses() as m:
         for _url, beacon_node_available in zip(
-            beacon_node_base_urls,
+            beacon_node_urls,
             beacon_node_availabilities,
             strict=True,
         ):