API for LightClientBootstrap, LightClientFinalityUpdate, LightClientOptimisticUpdate and light client events

beacon_node/beacon_chain/src/events.rs

@@ -15,6 +15,8 @@ pub struct ServerSentEventHandler<T: EthSpec> {
     chain_reorg_tx: Sender<EventKind<T>>,
     contribution_tx: Sender<EventKind<T>>,
     late_head: Sender<EventKind<T>>,
+    light_client_finality_update_tx: Sender<EventKind<T>>,
+    light_client_optimistic_update_tx: Sender<EventKind<T>>,
     block_reward_tx: Sender<EventKind<T>>,
     log: Logger,
 }
@@ -33,6 +35,8 @@ impl<T: EthSpec> ServerSentEventHandler<T> {
         let (chain_reorg_tx, _) = broadcast::channel(capacity);
         let (contribution_tx, _) = broadcast::channel(capacity);
         let (late_head, _) = broadcast::channel(capacity);
+        let (light_client_finality_update_tx, _) = broadcast::channel(capacity);
+        let (light_client_optimistic_update_tx, _) = broadcast::channel(capacity);
         let (block_reward_tx, _) = broadcast::channel(capacity);
 
         Self {
@@ -44,6 +48,8 @@ impl<T: EthSpec> ServerSentEventHandler<T> {
             chain_reorg_tx,
             contribution_tx,
             late_head,
+            light_client_finality_update_tx,
+            light_client_optimistic_update_tx,
             block_reward_tx,
             log,
         }
@@ -70,6 +76,10 @@ impl<T: EthSpec> ServerSentEventHandler<T> {
                 .map(|count| trace!(self.log, "Registering server-sent contribution and proof event"; "receiver_count" => count)),
             EventKind::LateHead(late_head) => self.late_head.send(EventKind::LateHead(late_head))
                 .map(|count| trace!(self.log, "Registering server-sent late head event"; "receiver_count" => count)),
+            EventKind::LightClientFinalityUpdate(update) => self.light_client_finality_update_tx.send(EventKind::LightClientFinalityUpdate(update))
+                .map(|count| trace!(self.log, "Registering server-sent light client finality update event"; "receiver_count" => count)),
+            EventKind::LightClientOptimisticUpdate(update) => self.light_client_optimistic_update_tx.send(EventKind::LightClientOptimisticUpdate(update))
+                .map(|count| trace!(self.log, "Registering server-sent light client optimistic update event"; "receiver_count" => count)),
             EventKind::BlockReward(block_reward) => self.block_reward_tx.send(EventKind::BlockReward(block_reward))
                 .map(|count| trace!(self.log, "Registering server-sent contribution and proof event"; "receiver_count" => count)),
         };
@@ -110,6 +120,14 @@ impl<T: EthSpec> ServerSentEventHandler<T> {
         self.late_head.subscribe()
     }
 
+    pub fn subscribe_light_client_finality_update(&self) -> Receiver<EventKind<T>> {
+        self.light_client_finality_update_tx.subscribe()
+    }
+
+    pub fn subscribe_light_client_optimistic_update(&self) -> Receiver<EventKind<T>> {
+        self.light_client_optimistic_update_tx.subscribe()
+    }
+
     pub fn subscribe_block_reward(&self) -> Receiver<EventKind<T>> {
         self.block_reward_tx.subscribe()
     }

beacon_node/client/src/builder.rs

@@ -147,6 +147,7 @@ where
         } else {
             None
         };
+        self.http_api_config.enable_light_client_server = config.network.enable_light_client_server;
 
         let execution_layer = if let Some(config) = config.execution_layer {
             let context = runtime_context.service_context("exec".into());

beacon_node/http_api/src/lib.rs

@@ -54,11 +54,11 @@ use tokio::sync::mpsc::{Sender, UnboundedSender};
 use tokio_stream::{wrappers::BroadcastStream, StreamExt};
 use types::{
     Attestation, AttestationData, AttesterSlashing, BeaconStateError, BlindedPayload,
-    CommitteeCache, ConfigAndPreset, Epoch, EthSpec, ForkName, FullPayload,
-    ProposerPreparationData, ProposerSlashing, RelativeEpoch, SignedAggregateAndProof,
-    SignedBeaconBlock, SignedBlindedBeaconBlock, SignedContributionAndProof,
-    SignedValidatorRegistrationData, SignedVoluntaryExit, Slot, SyncCommitteeMessage,
-    SyncContributionData,
+    CommitteeCache, ConfigAndPreset, Epoch, EthSpec, ForkName, FullPayload, Hash256,
+    LightClientBootstrap, ProposerPreparationData, ProposerSlashing, RelativeEpoch,
+    SignedAggregateAndProof, SignedBeaconBlock, SignedBlindedBeaconBlock,
+    SignedContributionAndProof, SignedValidatorRegistrationData, SignedVoluntaryExit, Slot,
+    SyncCommitteeMessage, SyncContributionData,
 };
 use version::{
     add_consensus_version_header, execution_optimistic_fork_versioned_response,
@@ -118,6 +118,7 @@ pub struct Config {
     pub allow_sync_stalled: bool,
     pub spec_fork_name: Option<ForkName>,
     pub data_dir: PathBuf,
+    pub enable_light_client_server: bool,
 }
 
 impl Default for Config {
@@ -131,6 +132,7 @@ impl Default for Config {
             allow_sync_stalled: false,
             spec_fork_name: None,
             data_dir: PathBuf::from(DEFAULT_ROOT_DIR),
+            enable_light_client_server: false,
         }
     }
 }
@@ -251,6 +253,18 @@ pub fn prometheus_metrics() -> warp::filters::log::Log<impl Fn(warp::filters::lo
     })
 }
 
+fn enable(is_enabled: bool) -> impl Filter<Extract = (), Error = warp::Rejection> + Clone {
+    warp::any()
+        .and_then(move || async move {
+            if is_enabled {
+                Ok(())
+            } else {
+                Err(warp::reject::not_found())
+            }
+        })
+        .untuple_one()
+}
+
 /// Creates a server that will serve requests using information from `ctx`.
 ///
 /// The server will shut down gracefully when the `shutdown` future resolves.
@@ -1726,6 +1740,169 @@ pub fn serve<T: BeaconChainTypes>(
      * beacon/rewards
      */
 
+    let beacon_light_client_path = eth_v1
+        .and(warp::path("beacon"))
+        .and(warp::path("light_client"))
+        .and(chain_filter.clone());
+
+    // GET beacon/light_client/bootrap/{block_root}
+    let get_beacon_light_client_bootstrap = beacon_light_client_path
+        .clone()
+        .and(warp::path("bootstrap"))
+        .and(warp::path::param::<Hash256>().or_else(|_| async {
+            Err(warp_utils::reject::custom_bad_request(
+                "Invalid block root value".to_string(),
+            ))
+        }))
+        .and(warp::path::end())
+        .and(warp::header::optional::<api_types::Accept>("accept"))
+        .and_then(
+            |chain: Arc<BeaconChain<T>>,
+             block_root: Hash256,
+             accept_header: Option<api_types::Accept>| {
+                blocking_task(move || {
+                    let state_root = chain
+                        .get_blinded_block(&block_root)
+                        .map_err(|_| {
+                            warp_utils::reject::custom_server_error(
+                                "Error retrieving block".to_string(),
+                            )
+                        })?
+                        .map(|signed_block| signed_block.state_root())
+                        .ok_or_else(|| {
+                            warp_utils::reject::custom_not_found(
+                                "Light client bootstrap unavailable".to_string(),
+                            )
+                        })?;
+
+                    let mut state = chain
+                        .get_state(&state_root, None)
+                        .map_err(|_| {
+                            warp_utils::reject::custom_server_error(
+                                "Error retrieving state".to_string(),
+                            )
+                        })?
+                        .ok_or_else(|| {
+                            warp_utils::reject::custom_not_found(
+                                "Light client bootstrap unavailable".to_string(),
+                            )
+                        })?;
+                    let fork_name = state
+                        .fork_name(&chain.spec)
+                        .map_err(inconsistent_fork_rejection)?;
+                    let bootstrap =
+                        LightClientBootstrap::from_beacon_state(&mut state).map_err(|_| {
+                            warp_utils::reject::custom_server_error(
+                                "Failed to create light client bootstrap".to_string(),
+                            )
+                        })?;
+                    match accept_header {
+                        Some(api_types::Accept::Ssz) => Response::builder()
+                            .status(200)
+                            .header("Content-Type", "application/octet-stream")
+                            .body(bootstrap.as_ssz_bytes().into())
+                            .map_err(|e| {
+                                warp_utils::reject::custom_server_error(format!(
+                                    "failed to create response: {}",
+                                    e
+                                ))
+                            }),
+                        _ => Ok(
+                            warp::reply::json(&api_types::GenericResponse::from(bootstrap))
+                                .into_response(),
+                        ),
+                    }
+                    .map(|resp| add_consensus_version_header(resp, fork_name))
+                })
+            },
+        );
+
+    // GET beacon/light_client/optimistic_update
+    let get_beacon_light_client_optimistic_update = beacon_light_client_path
+        .clone()
+        .and(warp::path("optimistic_update"))
+        .and(warp::path::end())
+        .and(warp::header::optional::<api_types::Accept>("accept"))
+        .and_then(
+            |chain: Arc<BeaconChain<T>>, accept_header: Option<api_types::Accept>| {
+                blocking_task(move || {
+                    let update = chain
+                        .latest_seen_optimistic_update
+                        .lock()
+                        .clone()
+                        .ok_or_else(|| {
+                            warp_utils::reject::custom_not_found(
+                                "No LightClientOptimisticUpdate is available".to_string(),
+                            )
+                        })?;
+
+                    let fork_name = chain
+                        .spec
+                        .fork_name_at_slot::<T::EthSpec>(update.signature_slot);
+                    match accept_header {
+                        Some(api_types::Accept::Ssz) => Response::builder()
+                            .status(200)
+                            .header("Content-Type", "application/octet-stream")
+                            .body(update.as_ssz_bytes().into())
+                            .map_err(|e| {
+                                warp_utils::reject::custom_server_error(format!(
+                                    "failed to create response: {}",
+                                    e
+                                ))
+                            }),
+                        _ => Ok(warp::reply::json(&api_types::GenericResponse::from(update))
+                            .into_response()),
+                    }
+                    .map(|resp| add_consensus_version_header(resp, fork_name))
+                })
+            },
+        );
+
+    // GET beacon/light_client/finality_update
+    let get_beacon_light_client_finality_update = beacon_light_client_path
+        .clone()
+        .and(warp::path("finality_update"))
+        .and(warp::path::end())
+        .and(warp::header::optional::<api_types::Accept>("accept"))
+        .and_then(
+            |chain: Arc<BeaconChain<T>>, accept_header: Option<api_types::Accept>| {
+                blocking_task(move || {
+                    let update = chain
+                        .latest_seen_finality_update
+                        .lock()
+                        .clone()
+                        .ok_or_else(|| {
+                            warp_utils::reject::custom_not_found(
+                                "No LightClientFinalityUpdate is available".to_string(),
+                            )
+                        })?;
+
+                    let fork_name = chain
+                        .spec
+                        .fork_name_at_slot::<T::EthSpec>(update.signature_slot);
+                    match accept_header {
+                        Some(api_types::Accept::Ssz) => Response::builder()
+                            .status(200)
+                            .header("Content-Type", "application/octet-stream")
+                            .body(update.as_ssz_bytes().into())
+                            .map_err(|e| {
+                                warp_utils::reject::custom_server_error(format!(
+                                    "failed to create response: {}",
+                                    e
+                                ))
+                            }),
+                        _ => Ok(warp::reply::json(&api_types::GenericResponse::from(update))
+                            .into_response()),
+                    }
+                    .map(|resp| add_consensus_version_header(resp, fork_name))
+                })
+            },
+        );
+
+    /*
+     * beacon/rewards
+     */
+
     let beacon_rewards_path = eth_v1
         .and(warp::path("beacon"))
         .and(warp::path("rewards"))
@@ -3395,6 +3572,12 @@ pub fn serve<T: BeaconChainTypes>(
                                 api_types::EventTopic::LateHead => {
                                     event_handler.subscribe_late_head()
                                 }
+                                api_types::EventTopic::LightClientFinalityUpdate => {
+                                    event_handler.subscribe_light_client_finality_update()
+                                }
+                                api_types::EventTopic::LightClientOptimisticUpdate => {
+                                    event_handler.subscribe_light_client_optimistic_update()
+                                }
                                 api_types::EventTopic::BlockReward => {
                                     event_handler.subscribe_block_reward()
                                 }
@@ -3492,6 +3675,12 @@ pub fn serve<T: BeaconChainTypes>(
                 .or(get_lighthouse_database_info.boxed())
                 .or(get_lighthouse_block_rewards.boxed())
                 .or(get_lighthouse_attestation_performance.boxed())
+                .or(enable(ctx.config.enable_light_client_server)
+                    .and(get_beacon_light_client_optimistic_update.boxed()))
+                .or(enable(ctx.config.enable_light_client_server)
+                    .and(get_beacon_light_client_finality_update.boxed()))
+                .or(enable(ctx.config.enable_light_client_server)
+                    .and(get_beacon_light_client_bootstrap.boxed()))
                 .or(get_lighthouse_block_packing_efficiency.boxed())
                 .or(get_lighthouse_merge_readiness.boxed())
                 .or(get_events.boxed())

beacon_node/http_api/tests/common.rs

@@ -176,6 +176,7 @@ pub async fn create_api_server_on_port<T: BeaconChainTypes>(
             allow_sync_stalled: false,
             data_dir: std::path::PathBuf::from(DEFAULT_ROOT_DIR),
             spec_fork_name: None,
+            enable_light_client_server: true,
         },
         chain: Some(chain.clone()),
         network_senders: Some(network_senders),

beacon_node/http_api/tests/main.rs

@@ -1,5 +1,5 @@
 #![cfg(not(debug_assertions))] // Tests are too slow in debug.
-#![recursion_limit = "256"]
+#![recursion_limit = "512"]
 
 pub mod common;
 pub mod fork_tests;

beacon_node/http_api/tests/tests.rs

@@ -1202,6 +1202,39 @@ impl ApiTester {
         self
     }
 
+    pub async fn test_get_beacon_light_client_optimistic_update(self) -> Self {
+        // get_beacon_light_client_optimistic_update returns Ok(None) on 404 NOT FOUND
+        let result = match self
+            .client
+            .get_beacon_light_client_optimistic_update::<E>()
+            .await
+        {
+            Ok(result) => result,
+            Err(_) => panic!("query did not fail correctly"),
+        };
+
+        let expected = self.chain.latest_seen_optimistic_update.lock().clone();
+        assert_eq!(result, expected);
+
+        self
+    }
+
+    pub async fn test_get_beacon_light_client_finality_update(self) -> Self {
+        let result = match self
+            .client
+            .get_beacon_light_client_finality_update::<E>()
+            .await
+        {
+            Ok(result) => result,
+            Err(_) => panic!("query did not fail correctly"),
+        };
+
+        let expected = self.chain.latest_seen_finality_update.lock().clone();
+        assert_eq!(result, expected);
+
+        self
+    }
+
     pub async fn test_get_beacon_pool_attestations(self) -> Self {
         let result = self
             .client
@@ -3947,6 +3980,22 @@ async fn node_get() {
         .await;
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn get_light_client_optimistic_update() {
+    ApiTester::new()
+        .await
+        .test_get_beacon_light_client_optimistic_update()
+        .await;
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn get_light_client_finality_update() {
+    ApiTester::new()
+        .await
+        .test_get_beacon_light_client_finality_update()
+        .await;
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn get_validator_duties_early() {
     ApiTester::new()

beacon_node/lighthouse_network/src/rpc/codec/ssz_snappy.rs

@@ -16,8 +16,8 @@ use std::marker::PhantomData;
 use std::sync::Arc;
 use tokio_util::codec::{Decoder, Encoder};
 use types::{
-    light_client_bootstrap::LightClientBootstrap, EthSpec, ForkContext, ForkName, Hash256,
-    SignedBeaconBlock, SignedBeaconBlockAltair, SignedBeaconBlockBase, SignedBeaconBlockMerge,
+    EthSpec, ForkContext, ForkName, Hash256, LightClientBootstrap, SignedBeaconBlock,
+    SignedBeaconBlockAltair, SignedBeaconBlockBase, SignedBeaconBlockMerge,
 };
 use unsigned_varint::codec::Uvi;