initial operator message types

Cargo.toml

@@ -1,6 +1,6 @@
 [workspace]
 default-members = ["crates/relay"]
-members = ["crates/common", "crates/data-api", "crates/database", "crates/relay", "crates/simulator", "crates/tcp-types", "crates/types", "crates/test-harness", "crates/website"]
+members = ["crates/common", "crates/data-api", "crates/database", "crates/relay", "crates/simulator", "crates/tcp-types", "crates/types", "crates/test-harness", "crates/website", "crates/operator"]
 resolver = "2"
 
 [workspace.package]
@@ -21,6 +21,7 @@ alloy-signer = { version = "1.0.41" }
 alloy-signer-local = { version = "1.0.41" }
 alloy-sol-types = { version = "1.4.1", default-features = false }
 askama = "0.12.0"
+async-channel = "2.5.0"
 async-trait = "0.1"
 axum = { version = "0.8", features = ["matched-path", "ws"] }
 backtrace = "0.3.69"
@@ -59,6 +60,7 @@ helix-database = { path = "crates/database" }
 helix-tcp-types = { path = "./crates/tcp-types" }
 helix-types = { path = "./crates/types" }
 helix-website = { path = "crates/website" }
+hex = "0.4.3"
 http = "1.0"
 http-body = "1.0"
 http-body-util = "0.1.3"
@@ -73,6 +75,7 @@ lh-kzg = { package = "kzg", git = "https://github.com/sigp/lighthouse", tag = "v
 lh-slot-clock = { package = "slot_clock", git = "https://github.com/sigp/lighthouse", tag = "v8.0.1" }
 lh-test-random = { package = "test_random_derive", git = "https://github.com/sigp/lighthouse", tag = "v8.0.1" }
 lh-types = { package = "types", git = "https://github.com/sigp/lighthouse", tag = "v8.0.1" }
+libp2p = { version = "0.56", features = ["macros", "ping", "quic", "floodsub", "secp256k1", "tokio"] }
 metrics = "0.24.0"
 mockito = "1.1.1"
 moka = { version = "0.12.10", features = ["sync"] }

crates/operator/Cargo.toml

@@ -0,0 +1,21 @@
+[package]
+name = "helix-operator"
+edition.workspace = true
+license.workspace = true
+repository.workspace = true
+rust-version.workspace = true
+version.workspace = true
+
+[dependencies]
+async-channel.workspace = true
+ethereum_ssz.workspace = true
+helix-types.workspace = true
+hex.workspace = true
+libp2p.workspace = true
+serde.workspace = true
+thiserror.workspace = true
+tokio.workspace = true
+tracing.workspace = true
+
+[dev-dependencies]
+alloy-primitives.workspace = true

crates/operator/src/lib.rs

@@ -0,0 +1,146 @@
+use std::fmt::{Display, Formatter};
+
+use async_channel::{Receiver, RecvError, SendError, Sender, TryRecvError, TrySendError, bounded};
+use helix_types::OperatorMessage;
+use libp2p::{
+    BehaviourBuilderError, Multiaddr, TransportError,
+    identity::{Keypair, PublicKey},
+    multiaddr,
+};
+use serde::{Deserialize, Serialize};
+use thiserror::Error;
+use tokio::task::AbortHandle;
+
+mod pubsub;
+mod utils;
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub struct Operator {
+    pub name: String,
+    #[serde(
+        serialize_with = "utils::serialize_pubkey",
+        deserialize_with = "utils::deserialize_pubkey"
+    )]
+    pub pubkey: PublicKey,
+    pub multiaddr: Multiaddr,
+}
+
+#[derive(Debug, Error)]
+pub enum OperatorError {
+    MultiaddrParseError(#[from] multiaddr::Error),
+    SwarmNetworkError(#[from] TransportError<std::io::Error>),
+    SwarmBuildError(#[from] BehaviourBuilderError),
+    MessageSendError(#[from] SendError<OperatorMessage>),
+    MessageTrySendError(#[from] TrySendError<OperatorMessage>),
+    MessageRecvError(#[from] RecvError),
+    MessageTryRecvError(#[from] TryRecvError),
+}
+
+impl Display for OperatorError {
+    fn fmt(&self, f: &mut Formatter) -> Result<(), std::fmt::Error> {
+        f.write_fmt(format_args!("{self:?}"))
+    }
+}
+
+/// Handle to operator pubsub.
+/// When this is first created, current builder collateral and demotions messages should
+/// be sent. These are cached internally (and updated) and sent automatically when new
+/// peers connect. 
+pub struct OperatorPubSub {
+    outgoing_msgs: Sender<OperatorMessage>,
+    incoming_msgs: Receiver<OperatorMessage>,
+    task_handle: AbortHandle,
+}
+
+impl Drop for OperatorPubSub {
+    fn drop(&mut self) {
+        self.task_handle.abort();
+    }
+}
+
+impl OperatorPubSub {
+    pub fn new(quic_port: u16, local_keypair: Keypair, operators: Vec<Operator>) -> Self {
+        let (outgoing_msgs, out_recv) = bounded(128);
+        let (in_send, incoming_msgs) = bounded(128);
+
+        // p2p task.
+        let handle = tokio::spawn(pubsub::run_operator_connection(
+            quic_port,
+            local_keypair,
+            operators,
+            out_recv,
+            in_send,
+        ));
+
+        Self { outgoing_msgs, incoming_msgs, task_handle: handle.abort_handle() }
+    }
+
+    pub async fn send(&self, msg: OperatorMessage) -> Result<(), OperatorError> {
+        Ok(self.outgoing_msgs.send(msg).await?)
+    }
+
+    pub fn try_send(&self, msg: OperatorMessage) -> Result<(), OperatorError> {
+        Ok(self.outgoing_msgs.try_send(msg)?)
+    }
+
+    pub async fn recv(&self) -> Result<OperatorMessage, OperatorError> {
+        Ok(self.incoming_msgs.recv().await?)
+    }
+
+    pub fn try_recv(&self) -> Result<Option<OperatorMessage>, OperatorError> {
+        match self.incoming_msgs.try_recv() {
+            Ok(msg) => Ok(Some(msg)),
+            Err(TryRecvError::Empty) => Ok(None),
+            Err(e) => Err(e.into()),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::str::FromStr;
+
+    use alloy_primitives::B256;
+    use helix_types::{BlsPublicKeyBytes, Demotion, OperatorMessage, Promotion};
+    use libp2p::{Multiaddr, identity::Keypair};
+
+    use crate::{Operator, OperatorPubSub};
+
+    #[tokio::test]
+    async fn operator_p2p() {
+        let keypair_a = Keypair::generate_secp256k1();
+        let keypair_b = Keypair::generate_secp256k1();
+
+        let operator_a = Operator {
+            name: "operator A".into(),
+            pubkey: keypair_a.public(),
+            multiaddr: Multiaddr::from_str("/ip4/127.0.0.1/udp/23032/quic-v1").unwrap(),
+        };
+
+        let operator_b = Operator {
+            name: "operator B".into(),
+            pubkey: keypair_b.public(),
+            multiaddr: Multiaddr::from_str("/ip4/127.0.0.1/udp/32023/quic-v1").unwrap(),
+        };
+
+        let op_a = OperatorPubSub::new(23032, keypair_a, vec![operator_b]);
+        let op_b = OperatorPubSub::new(32023, keypair_b, vec![operator_a]);
+
+        let builder_pubkey = BlsPublicKeyBytes::random();
+        let demotion = Demotion {
+            ts_ms: 1,
+            slot: 1,
+            builder_pubkey,
+            block_hash: B256::random(),
+            reason_msg: "fail".as_bytes().to_vec(),
+        };
+        let promotion = Promotion { ts_ms: 2, slot: 2, builder_pubkey };
+        op_a.send(helix_types::OperatorMessage::Demotion(demotion)).await.unwrap();
+        let msg = op_b.recv().await.unwrap();
+        assert!(matches!(msg, OperatorMessage::Demotion(_)));
+
+        op_b.send(OperatorMessage::Promotion(promotion)).await.unwrap();
+        let msg = op_a.recv().await.unwrap();
+        assert!(matches!(msg, OperatorMessage::Promotion(_)));
+    }
+}

crates/operator/src/pubsub.rs

@@ -0,0 +1,152 @@
+use std::{collections::HashMap, time::Duration};
+
+use async_channel::{Receiver, Sender};
+use helix_types::OperatorMessage;
+use libp2p::{
+    PeerId, SwarmBuilder, allow_block_list::{self, AllowedPeers}, floodsub::{self, Event, FloodsubMessage}, futures::StreamExt, identity::Keypair, ping, swarm::{NetworkBehaviour, SwarmEvent},
+};
+use ssz::{Decode, Encode};
+
+use super::{Operator, OperatorError};
+
+#[derive(NetworkBehaviour)]
+struct NetBehaviour {
+    allow_list: allow_block_list::Behaviour<AllowedPeers>,
+    floodsub: floodsub::Behaviour,
+    ping: ping::Behaviour,
+}
+
+pub(super) async fn run_operator_connection(
+    quic_port: u16,
+    keypair: Keypair,
+    operators: Vec<Operator>,
+    outgoing: Receiver<OperatorMessage>,
+    incoming: Sender<OperatorMessage>,
+) -> Result<(), OperatorError> {
+    let floodsub_topic = floodsub::Topic::new("operator");
+    let local_peer_id = PeerId::from_public_key(&keypair.public());
+
+    let mut allow_list = allow_block_list::Behaviour::default();
+    for op in &operators {
+        allow_list.allow_peer(PeerId::from_public_key(&op.pubkey));
+    }
+
+    let mut swarm = SwarmBuilder::with_existing_identity(keypair)
+        .with_tokio()
+        .with_quic()
+        .with_behaviour(|_key| {
+            Ok(NetBehaviour {
+                allow_list,
+                floodsub: floodsub::Behaviour::new(local_peer_id),
+                ping: ping::Behaviour::default(),
+            })
+        })?
+        .with_swarm_config(|cfg| cfg.with_idle_connection_timeout(Duration::from_secs(u64::MAX)))
+        .build();
+
+    // Subscribe to operator topic.
+    swarm.behaviour_mut().floodsub.subscribe(floodsub_topic.clone());
+
+    // Listen for incoming connections.
+    swarm.listen_on(format!("/ip4/0.0.0.0/udp/{quic_port}/quic-v1").parse()?)?;
+
+    // Peers by id.
+    let peers = operators
+        .into_iter()
+        .map(|o| (PeerId::from_public_key(&o.pubkey), o))
+        .collect::<HashMap<_, _>>();
+
+    for (peer_id, operator) in &peers {
+        // Dial other operators.
+        swarm.behaviour_mut().floodsub.add_node_to_partial_view(*peer_id);
+        if let Err(e) = swarm.dial(operator.multiaddr.clone()) {
+            tracing::warn!(?operator, ?e, "failed to dial operator");
+        }
+    }
+
+    // Local demotions keyed by builder pubkey. Sent when a new operator subscribes.
+    let mut demotions = HashMap::new();
+    // Local collateral keyed by builder pubkey. Sent when a new operator subscribes.
+    let mut builder_collateral = HashMap::new();
+    // Number of connected peers
+    let mut connected_peers = 0u32;
+
+    loop {
+        tokio::select! {
+            to_send = outgoing.recv() => match to_send {
+                Ok(msg) => {
+                    match &msg {
+                        OperatorMessage::Demotion(demotion) => {
+                            demotions.insert(demotion.builder_pubkey, demotion.clone());
+                        }
+                        OperatorMessage::Promotion(promotion) => {
+                            demotions.remove(&promotion.builder_pubkey);
+                        }
+                        OperatorMessage::Collateral(collateral) => {
+                            builder_collateral.insert(collateral.builder_pubkey, collateral.clone());
+                        }
+                    }
+                    if connected_peers > 0 {
+                        swarm.behaviour_mut().floodsub.publish(floodsub_topic.clone(), msg.as_ssz_bytes());
+                    }
+                }
+                Err(_) => break, // channel closed
+            },
+            event = swarm.select_next_some() => match event {
+                SwarmEvent::Behaviour(b_event) => match b_event {
+                    NetBehaviourEvent::Floodsub(f_event) => match f_event {
+                        Event::Message(msg) => {
+                            let FloodsubMessage { source, data, .. } = msg;
+
+                            match peers.get(&source) {
+                                Some(operator) => {
+                                    let operator_msg = match OperatorMessage::from_ssz_bytes(&data) {
+                                        Ok(msg) => msg,
+                                        Err(e) => {
+                                            tracing::error!(?e, operator=operator.name, "failed to decode operator message");
+                                            continue;
+                                        }
+                                    };
+                                    tracing::info!(?operator_msg, operator=operator.name, "new operator message");
+                                    let _ = incoming.send(operator_msg).await;
+                                }
+                                None => {
+                                    tracing::warn!(?source, "received operator message from unknown peer");
+                                    let _ = swarm.disconnect_peer_id(source);
+                                }
+                            }
+                        }
+                        Event::Subscribed { peer_id, topic } => {
+                            if peers.contains_key(&peer_id) && topic == floodsub_topic {
+                                // Send current demotion and collateral state.
+                                for (_, demotion) in &demotions {
+                                    swarm.behaviour_mut().floodsub.publish(floodsub_topic.clone(), OperatorMessage::Demotion(demotion.clone()).as_ssz_bytes());
+                                }
+                                for (_, collateral) in &builder_collateral {
+                                    swarm.behaviour_mut().floodsub.publish(floodsub_topic.clone(), OperatorMessage::Collateral(collateral.clone()).as_ssz_bytes());
+                                }
+                            } else {
+                                let _ = swarm.disconnect_peer_id(peer_id);
+                            }
+                        }
+                        _ => {}
+                    }
+                    NetBehaviourEvent::Ping(_) => {}
+                }
+                SwarmEvent::ConnectionEstablished { peer_id, .. } => {
+                    if peers.contains_key(&peer_id) {
+                        connected_peers += 1;
+                    }
+                }
+                SwarmEvent::ConnectionClosed { peer_id, .. } => {
+                    if peers.contains_key(&peer_id) {
+                        connected_peers = connected_peers.saturating_sub(1);
+                    }
+                }
+                _ => {},
+            }
+        }
+    }
+
+    Ok(())
+}

crates/operator/src/utils.rs

@@ -0,0 +1,23 @@
+use libp2p::identity::{PublicKey, secp256k1};
+use serde::{Deserialize, Deserializer, Serializer, de::Error as _, ser::Error};
+
+/// Serialize a secp256k1 pubkey to hex.
+pub(super) fn serialize_pubkey<S>(value: &PublicKey, serializer: S) -> Result<S::Ok, S::Error>
+where
+    S: Serializer,
+{
+    let key = value.clone().try_into_secp256k1().map_err(|e| S::Error::custom(e.to_string()))?;
+    serializer.serialize_str(&hex::encode(&key.to_bytes()))
+}
+
+/// Deserialize a hex-encoded secp256k1 pubkey.
+pub(super) fn deserialize_pubkey<'de, D>(d: D) -> Result<PublicKey, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    let hex = String::deserialize(d)?;
+    let bytes = hex::decode(hex).map_err(|e| D::Error::custom(e.to_string()))?;
+    let secp_pubkey = secp256k1::PublicKey::try_from_bytes(&bytes)
+        .map_err(|e| D::Error::custom(e.to_string()))?;
+    Ok(PublicKey::from(secp_pubkey))
+}

crates/types/src/lib.rs

@@ -8,6 +8,7 @@ mod error;
 mod execution_payload;
 mod fields;
 mod hydration;
+mod operator;
 mod test_utils;
 mod utils;
 mod validator;
@@ -31,6 +32,7 @@ pub use lh_types::{
     Config as LhConfig, EthSpec, ForkVersionDecode, MainnetEthSpec, SignedRoot,
     fork_name::ForkName, payload::ExecPayload, test_utils::TestRandom,
 };
+pub use operator::*;
 use serde::{Deserialize, Serialize};
 use ssz_derive::{Decode, Encode};
 pub use test_utils::*;