Wire local attestation admission through optional cluster consensus

crates/bin/src/main.rs

@@ -25,11 +25,11 @@ use silver_common::{
     tracing::initialise_tracing_log,
 };
 use silver_config::Config;
-use silver_control::{Controller, sync_engine::SyncEngine};
+use silver_control::{Controller, cluster::AttestationClusterConfig, sync_engine::SyncEngine};
 use silver_discovery::{DiscV5, Discovery};
 use silver_gossip::GossipHandler;
 use silver_httpcore::Bind;
-use silver_network::{Context, NetworkTile, P2p};
+use silver_network::{ClusterNodes, Context, NetworkTile, P2p};
 use silver_peer::PeerManager;
 use silver_storage::{latest_local_checkpoint, tile::StorageTile};
 
@@ -123,6 +123,9 @@ fn main() -> Result<(), Box<dyn Error>> {
         incoming_engine_resp_producer.cache_ref().random_access("ds_engine_incoming_resp", true)?;
 
     let cluster_inbound_producer = TCache::producer("cluster_inbound", CLUSTER_MESSAGE_TCACHE_SIZE);
+    let cluster_inbound_consumer = cluster_inbound_producer
+        .cache_ref()
+        .strict_random_access("control_cluster_inbound", true)?;
     let cluster_outbound_producer =
         TCache::producer("cluster_outbound", CLUSTER_MESSAGE_TCACHE_SIZE);
     let cluster_outbound_consumer = cluster_outbound_producer
@@ -157,11 +160,27 @@ fn main() -> Result<(), Box<dyn Error>> {
     // Long-lived attnets: advertised from boot (peer retention exempts us
     // from excess-peer pruning); the gossip subscriptions themselves
     // activate once Following — see `Controller::pending_subnet_topics`.
-    let boot_epoch = ticker.current_slot() / SLOTS_PER_EPOCH;
+    let boot_wall_slot = ticker.current_slot();
+    let boot_epoch = boot_wall_slot / SLOTS_PER_EPOCH;
     let attnet_count = config.attestation_subnet_count();
     let subnets = local_enr.node_id().attestation_subnets(boot_epoch, attnet_count);
     local_enr.set_attnets(subnets, keypair.secret_key())?;
 
+    // Cluster configuration
+    let (cluster_config, cluster_nodes) = config
+        .cluster_config()
+        .map(|c| {
+            let voters = c.nodes.clone();
+            let node_id = match voters.iter().find(|(_, enr)| enr.node_id() == local_enr.node_id())
+            {
+                Some((id, _)) => *id,
+                None => return Err("no local node configured in cluster config"),
+            };
+            Ok((AttestationClusterConfig::new(node_id, voters.keys().copied().collect()), voters))
+        })
+        .transpose()?
+        .unzip();
+
     let discv5_addr = config.discovery_bind_addr().expect("no discovery port");
     let p2p_addr = config.p2p_bind_addr().expect("no p2p port");
     let mut discv5 = DiscV5::new(
@@ -170,6 +189,20 @@ fn main() -> Result<(), Box<dyn Error>> {
         local_enr,
         config.fork_digest(),
     );
+
+    // Cluster peers are added as trusted peers
+    let cluster_peers = cluster_nodes
+        .as_ref()
+        .map(|m| m.values())
+        .unwrap_or_default()
+        .filter(|enr| enr.node_id() != local_enr.node_id());
+    let trusted_peers =
+        config.trusted_peers().iter().chain(cluster_peers).cloned().collect::<Vec<_>>();
+    let trusted_ips = trusted_peers
+        .iter()
+        .filter_map(|enr| enr.ip4().map(IpAddr::from).or(enr.ip6().map(IpAddr::from)))
+        .collect();
+
     let server_config = silver_network::create_server_config(&keypair)?;
     let p2p_endpoint = P2p::new(
         keypair,
@@ -180,11 +213,7 @@ fn main() -> Result<(), Box<dyn Error>> {
             None,
         ),
         config.max_connections(),
-        config
-            .trusted_peers()
-            .iter()
-            .filter_map(|enr| enr.ip4().map(IpAddr::from).or(enr.ip6().map(IpAddr::from)))
-            .collect(),
+        trusted_ips,
     );
     let identify = config.identify()?;
     let p2p_context = Context {
@@ -195,7 +224,7 @@ fn main() -> Result<(), Box<dyn Error>> {
         rpc_producer: incoming_rpc_producer,
         rpc_consumer: outgoing_rpc_producer.cache_ref().random_access("p2p_outgoing_rpc", true)?,
         identify: Some(ProtoIdentify::from((&identify, &keypair))),
-        cluster_nodes: None,
+        cluster_nodes: cluster_nodes.map(ClusterNodes::new),
         cluster_inbound_producer,
         cluster_outbound_consumer,
     };
@@ -238,7 +267,7 @@ fn main() -> Result<(), Box<dyn Error>> {
     let mut control_tile = Controller::new(
         PeerManager::new(
             keypair.peer_id(),
-            config.trusted_peers().to_vec(),
+            trusted_peers,
             gossip_topics,
             config.peer_score_params(),
             config.syncing_config(),
@@ -249,13 +278,16 @@ fn main() -> Result<(), Box<dyn Error>> {
         gossip_handler,
         outgoing_rpc_producer.clone(),
         incoming_rpc_consumer_ctl,
+        cluster_outbound_producer,
+        cluster_inbound_consumer,
+        cluster_config,
         SyncEngine::new(
             config.syncing_config(),
             booting_from_local_checkpoint,
             das_custody_groups,
             spec.clone(),
         ),
-    );
+    )?;
     control_tile.set_pending_subnet_topics(
         silver_common::attnet_subnets(subnets)
             .map(silver_common::GossipTopic::BeaconAttestation)

crates/control/Cargo.toml

@@ -22,6 +22,7 @@ buffa-build = "0.2.0"
 
 [dev-dependencies]
 silver_common = { workspace = true, features = ["test-util"] }
+tempfile = "3"
 
 [lints]
 workspace = true

crates/control/src/cluster/mod.rs

@@ -9,9 +9,12 @@ mod node;
 mod wire;
 
 pub use admission::AdmissionError;
+pub(crate) use admission::AttestationAdmission;
 pub use command::{AttestationKey, AttestationLockCommand, CommandDecodeError};
+pub(crate) use lock_store::AttestationLockStore;
 pub use lock_store::LockResult;
 pub use node::{
     AttestationCluster, AttestationClusterConfig, AttestationDecision, ClusterError, ClusterEvent,
     ProposalId, ProposeError,
 };
+pub(crate) use wire::{decode_message, encode_message};

crates/control/src/tile.rs

@@ -2,15 +2,22 @@ use std::time::{Duration, Instant};
 
 use flux::{spine::SpineAdapter, tile::Tile};
 use silver_common::{
-    BeaconStateEvent, GossipTopic, Nanos, P2pSend, PeerControl, PeerEvent, PeerStats, RpcInbound,
-    RpcOutbound, RpcRequest, RpcRequestOutbound, RpcResponse, RpcResponseInbound, SilverSpine,
-    SilverSpineProducers, SyncNeed, SyncUpdate, TMultiProducer, TRandomAccess,
+    BeaconApiRequest, BeaconStateEvent, GossipTopic, LOCAL_GOSSIP_STREAM_ID, Nanos, P2pSend,
+    PeerControl, PeerEvent, PeerStats, RpcInbound, RpcOutbound, RpcRequest, RpcRequestOutbound,
+    RpcResponse, RpcResponseInbound, SilverSpine, SilverSpineProducers, SyncNeed, SyncUpdate,
+    TMultiProducer, TProducer, TRandomAccess,
     ssz_view::{METADATA_SIZE, STATUS_V2_SIZE, StatusView},
 };
 use silver_gossip::{GossipHandler, GossipHandlerEvent};
 use silver_peer::PeerManager;
 
-use crate::sync_engine::{SyncAction, SyncEngine};
+use self::attestation_cluster::AttestationClusterHandler;
+use crate::{
+    cluster::{AttestationClusterConfig, ClusterError},
+    sync_engine::{SyncAction, SyncEngine},
+};
+
+mod attestation_cluster;
 
 const PEER_PERSIST_INTERVAL: Duration = Duration::from_secs(300);
 
@@ -28,6 +35,7 @@ pub struct Controller {
     /// Reads `incoming_rpc` sidecar payloads referenced by
     /// `PeerEvent::PublishDataColumn`.
     rpc_ssz_consumer: TRandomAccess,
+    attestation_cluster: AttestationClusterHandler,
     last_tick: Instant,
     last_ping: Instant,
     last_status: Instant,
@@ -50,26 +58,39 @@ impl Controller {
     /// Build a Controller. `status` and `metadata` start empty — callers
     /// update them via `set_status` / `set_metadata` once chain state is
     /// available.
+    #[allow(clippy::too_many_arguments)]
     pub fn new(
         peer_manager: PeerManager,
         gossip_handler: GossipHandler,
         rpc_producer: TMultiProducer,
         rpc_ssz_consumer: TRandomAccess,
+        cluster_outbound_producer: TProducer,
+        cluster_inbound_consumer: TRandomAccess,
+        cluster_config: Option<AttestationClusterConfig>,
         sync_engine: SyncEngine,
-    ) -> Self {
-        Self {
+    ) -> Result<Self, ClusterError> {
+        let now = Instant::now();
+        let attestation_cluster = AttestationClusterHandler::new(
+            cluster_outbound_producer,
+            cluster_inbound_consumer,
+            cluster_config,
+            now,
+        )?;
+
+        Ok(Self {
             peer_manager,
             gossip_handler,
             sync_engine,
             rpc_producer,
             rpc_ssz_consumer,
-            last_tick: Instant::now(),
-            last_ping: Instant::now(),
-            last_status: Instant::now(),
-            last_peer_persist: Instant::now(),
+            attestation_cluster,
+            last_tick: now,
+            last_ping: now,
+            last_status: now,
+            last_peer_persist: now,
             auto_ping: true,
             pending_subnet_topics: Vec::new(),
-        }
+        })
     }
 
     pub fn set_pending_subnet_topics(&mut self, topics: Vec<GossipTopic>) {
@@ -136,6 +157,7 @@ impl Tile<SilverSpine> for Controller {
     fn loop_body(&mut self, adapter: &mut SpineAdapter<SilverSpine>) {
         let now = Instant::now();
         self.rpc_ssz_consumer.free();
+        self.attestation_cluster.free();
 
         // Local status must land before the sync drive below: issuance is
         // capped against the imported head, and a one-loop-stale watermark
@@ -146,6 +168,7 @@ impl Tile<SilverSpine> for Controller {
 
             match beacon_event {
                 BeaconStateEvent::Status { ssz, latest_block_slot, wall_slot, .. } => {
+                    self.attestation_cluster.on_status(StatusView::head_slot(&ssz), wall_slot);
                     latest_status_event = Some((ssz, latest_block_slot, wall_slot));
                 }
                 // PM keeps the reject for peer eviction (Status backing a
@@ -157,6 +180,17 @@ impl Tile<SilverSpine> for Controller {
             }
         });
 
+        adapter.consume(|request: BeaconApiRequest, producers| {
+            self.attestation_cluster.on_beacon_api_request(
+                request,
+                now,
+                &mut self.gossip_handler,
+                producers,
+            );
+        });
+
+        self.attestation_cluster.spin(now, adapter, &mut self.gossip_handler);
+
         adapter.consume(|need: SyncNeed, _producers| self.sync_engine.on_sync_need(need, now));
 
         let fork_digest_changed = self.handle_latest_status(latest_status_event);
@@ -206,6 +240,18 @@ impl Tile<SilverSpine> for Controller {
                 return;
             }
 
+            // Beacon State uses the synthetic local stream for a terminal
+            // local validation failure. Complete the API request without
+            // counting that failure against a (non-existent) network peer.
+            if matches!(
+                event,
+                PeerEvent::P2pGossipInvalidMsg { p2p_peer, .. }
+                    if p2p_peer == LOCAL_GOSSIP_STREAM_ID.peer()
+            ) {
+                self.attestation_cluster.on_peer_event(&event, producers);
+                return;
+            }
+
             self.sync_engine.on_peer_event(event, self.peer_manager.our_fork_digest());
 
             if let PeerEvent::SendGossip {
@@ -228,8 +274,16 @@ impl Tile<SilverSpine> for Controller {
                     producers,
                 )
             });
+
+            // A local Beacon API request completes only after Beacon State's
+            // validation result has gone through the normal gossip path.
+            self.attestation_cluster.on_peer_event(&event, producers);
         });
 
+        // Consume every validation outcome already queued before expiring
+        // requests, so an event arriving at the deadline wins the race.
+        self.attestation_cluster.expire_pending_validation(now, &mut adapter.producers);
+
         adapter.consume(|rpc: RpcInbound, producers| {
             self.sync_engine.rpc_event(&rpc, self.peer_manager.our_fork_digest());
             if let Some(request_id) = Self::msg_served_for(&rpc) {