feat: add support for Beacon API Event Stream (#1848)

Author: KolbyML

Cargo.lock

@@ -38,7 +38,7 @@ rust-version = "1.87.0"
 version = "0.3.2"
 
 [workspace.dependencies]
-alloy = { version = "1.0", default-features = false, features = ["std", "serde", "getrandom"] }
+alloy = { version = "1.0", default-features = false, features = ["std", "serde", "getrandom", "rpc-types-beacon"] }
 alloy-chains = "0.2"
 alloy-hardforks = "0.2.0"
 alloy-rlp = { version = "0.3.8", default-features = false, features = ["derive"] }
@@ -56,6 +56,7 @@ ethereum_hashing = "0.7.0"
 ethereum_serde_utils = "0.8"
 ethereum_ssz = "0.9"
 ethereum_ssz_derive = "0.9"
+eventsource-client = "0.15.0"
 futures = "0.3.23"
 futures-util = "0.3.23"
 hex = "0.4.3"

Cargo.toml

@@ -18,6 +18,8 @@ anyhow.workspace = true
 ethereum_ssz.workspace = true
 ethereum_ssz_derive.workspace = true
 ethportal-api.workspace = true
+eventsource-client.workspace = true
+futures.workspace = true
 reqwest.workspace = true
 reqwest-middleware = { version = "0.4", features = ["json"] }
 reqwest-retry = "0.7"

crates/ethereum-rpc-client/Cargo.toml

@@ -0,0 +1,45 @@
+use alloy::rpc::types::beacon::events::{
+    BeaconNodeEventTopic, ChainReorgEvent, FinalizedCheckpointEvent, HeadEvent,
+    LightClientOptimisticUpdateEvent,
+};
+use eventsource_client::Event;
+use serde::de::{DeserializeOwned, Error};
+
+pub enum BeaconEvent {
+    ChainReorg(ChainReorgEvent),
+    Head(HeadEvent),
+    LightClientOptimisticUpdate(LightClientOptimisticUpdateEvent),
+    FinalizedCheckpoint(FinalizedCheckpointEvent),
+}
+
+impl BeaconEvent {
+    fn from_json<T: DeserializeOwned>(
+        json: &str,
+        constructor: impl FnOnce(T) -> Self,
+    ) -> Result<Self, serde_json::Error> {
+        serde_json::from_str(json).map(constructor)
+    }
+}
+
+impl TryFrom<Event> for BeaconEvent {
+    type Error = serde_json::Error;
+
+    fn try_from(event: Event) -> Result<Self, Self::Error> {
+        if event.event_type == BeaconNodeEventTopic::ChainReorg.query_value() {
+            Self::from_json(&event.data, Self::ChainReorg)
+        } else if event.event_type == BeaconNodeEventTopic::Head.query_value() {
+            Self::from_json(&event.data, Self::Head)
+        } else if event.event_type
+            == BeaconNodeEventTopic::LightClientOptimisticUpdate.query_value()
+        {
+            Self::from_json(&event.data, Self::LightClientOptimisticUpdate)
+        } else if event.event_type == BeaconNodeEventTopic::FinalizedCheckpoint.query_value() {
+            Self::from_json(&event.data, Self::FinalizedCheckpoint)
+        } else {
+            Err(Self::Error::custom(format!(
+                "Can't create BeaconEvent: unexpected event type: {}",
+                event.event_type,
+            )))
+        }
+    }
+}

crates/ethereum-rpc-client/src/consensus/event.rs

@@ -1,9 +1,10 @@
 pub mod constants;
+pub mod event;
 pub mod rpc_types;
 
-use std::{fmt::Display, time::Duration};
+use std::{fmt::Display, pin::Pin, time::Duration};
 
-use alloy::primitives::B256;
+use alloy::{primitives::B256, rpc::types::beacon::events::BeaconNodeEventTopic};
 use anyhow::{anyhow, bail, ensure};
 use constants::DEFAULT_BEACON_STATE_REQUEST_TIMEOUT;
 use ethportal_api::{
@@ -17,19 +18,23 @@ use ethportal_api::{
         optimistic_update::LightClientOptimisticUpdateElectra, update::LightClientUpdateElectra,
     },
 };
+use event::BeaconEvent;
+use eventsource_client::{Client, ClientBuilder, SSE};
+use futures::{Stream, StreamExt};
 use reqwest::{
     header::{HeaderMap, HeaderValue, ACCEPT, CONTENT_TYPE},
     Response,
 };
 use rpc_types::{RootResponse, VersionResponse, VersionedDataResponse, VersionedDataResult};
 use serde::de::DeserializeOwned;
 use ssz::Decode;
-use tracing::{debug, warn};
+use tracing::{debug, error, info, warn};
 use url::Url;
 
 use super::http_client::{
     ClientWithBaseUrl, ContentType, JSON_ACCEPT_PRIORITY, JSON_CONTENT_TYPE, SSZ_CONTENT_TYPE,
 };
+use crate::http_client::get_authorization_headers;
 
 /// Implements endpoints from the Beacon API to access data from the consensus layer.
 #[derive(Clone, Debug)]
@@ -235,6 +240,55 @@ impl ConsensusApi {
             .message)
     }
 
+    pub fn get_events_stream(
+        &self,
+        topics: &[BeaconNodeEventTopic],
+        stream_tag: &'static str,
+    ) -> anyhow::Result<Pin<Box<dyn Stream<Item = BeaconEvent> + Send>>> {
+        let endpoint = self.primary.base_url().join(&format!(
+            "/eth/v1/events?topics={}",
+            topics
+                .iter()
+                .map(|topic| topic.query_value())
+                .collect::<Vec<_>>()
+                .join(",")
+        ))?;
+
+        let mut client_builder = ClientBuilder::for_url(endpoint.as_str())?;
+        for (key, value) in get_authorization_headers(self.primary.base_url().clone())? {
+            client_builder = client_builder.header(key.as_ref(), value.to_str()?)?;
+        }
+
+        Ok(client_builder
+            .build()
+            .stream()
+            .filter_map(move |event| async move {
+                let event = match event {
+                    Ok(SSE::Event(event)) => event,
+                    Ok(SSE::Connected(connection_details)) => {
+                        info!("{stream_tag}: Connected to SSE stream: {connection_details:?}");
+                        return None;
+                    }
+                    Ok(SSE::Comment(comment)) => {
+                        info!("{stream_tag}: Received comment: {comment:?}");
+                        return None;
+                    }
+                    Err(err) => {
+                        error!("{stream_tag}: Error receiving event: {err:?}");
+                        return None;
+                    }
+                };
+                match BeaconEvent::try_from(event) {
+                    Ok(event) => Some(event),
+                    Err(err) => {
+                        error!("{stream_tag}: Failed to decode event: {err:?}");
+                        None
+                    }
+                }
+            })
+            .boxed())
+    }
+
     /// Make a request to the cl provider. If the primary provider fails, it will retry with the
     /// fallback.
     async fn request<T>(

crates/ethereum-rpc-client/src/consensus/mod.rs

@@ -1,7 +1,8 @@
 use std::{env, time::Duration};
 
+use anyhow::{anyhow, bail};
 use reqwest::{
-    header::{HeaderMap, HeaderValue, ACCEPT, CONTENT_TYPE},
+    header::{HeaderMap, HeaderName, HeaderValue, ACCEPT, CONTENT_TYPE},
     Client, IntoUrl, Request, Response,
 };
 use reqwest_middleware::{ClientBuilder, ClientWithMiddleware, RequestBuilder};
@@ -31,7 +32,7 @@ impl ClientWithBaseUrl {
         url: Url,
         request_timeout: u64,
         content_type: ContentType,
-    ) -> Result<ClientWithBaseUrl, String> {
+    ) -> anyhow::Result<ClientWithBaseUrl> {
         let mut headers = HeaderMap::new();
         match content_type {
             ContentType::Json => {
@@ -43,40 +44,17 @@ impl ClientWithBaseUrl {
             }
         }
 
-        if let Some(host) = url.host_str() {
-            if host.contains("pandaops.io") {
-                let client_id = env::var("PANDAOPS_CLIENT_ID").unwrap_or_else(|_| {
-                    error!("Pandaops provider detected without PANDAOPS_CLIENT_ID set");
-                    "null".to_string()
-                });
-
-                let client_secret = env::var("PANDAOPS_CLIENT_SECRET").unwrap_or_else(|_| {
-                    error!("Pandaops provider detected without PANDAOPS_CLIENT_SECRET set");
-                    "null".to_string()
-                });
-
-                headers.insert(
-                    "CF-Access-Client-Id",
-                    HeaderValue::from_str(&client_id)
-                        .map_err(|_| "Invalid client id header value")?,
-                );
-
-                headers.insert(
-                    "CF-Access-Client-Secret",
-                    HeaderValue::from_str(&client_secret)
-                        .map_err(|_| "Invalid client secret header value")?,
-                );
-            }
-        } else {
-            return Err("Failed to find host string".into());
+        let authentication_headers = get_authorization_headers(url.clone())?;
+        for (key, value) in authentication_headers {
+            headers.insert(key, value);
         }
 
         // Add retry middleware
         let reqwest_client = Client::builder()
             .default_headers(headers)
             .timeout(Duration::from_secs(request_timeout))
             .build()
-            .map_err(|_| "Failed to build HTTP client")?;
+            .map_err(|err| anyhow!("Failed to build HTTP client {err:?}"))?;
         let client = ClientBuilder::new(reqwest_client)
             .with(RetryTransientMiddleware::new_with_policy(
                 ExponentialBackoff::builder().build_with_max_retries(3),
@@ -111,3 +89,32 @@ impl ClientWithBaseUrl {
         self.client.execute(request).await
     }
 }
+
+pub fn get_authorization_headers(url: Url) -> anyhow::Result<Vec<(HeaderName, HeaderValue)>> {
+    let mut headers = vec![];
+    if let Some(host) = url.host_str() {
+        if host.contains("pandaops.io") {
+            let client_id = env::var("PANDAOPS_CLIENT_ID").unwrap_or_else(|_| {
+                error!("Pandaops provider detected without PANDAOPS_CLIENT_ID set");
+                "null".to_string()
+            });
+
+            let client_secret = env::var("PANDAOPS_CLIENT_SECRET").unwrap_or_else(|_| {
+                error!("Pandaops provider detected without PANDAOPS_CLIENT_SECRET set");
+                "null".to_string()
+            });
+
+            headers.push((
+                HeaderName::from_static("cf-access-client-id"),
+                HeaderValue::from_str(&client_id)?,
+            ));
+            headers.push((
+                HeaderName::from_static("cf-access-client-secret"),
+                HeaderValue::from_str(&client_secret)?,
+            ));
+        }
+    } else {
+        bail!("Failed to find host string");
+    }
+    Ok(headers)
+}