Persist light client updates (#5545)

* persist light client updates

* update beacon chain to serve light client updates

* resolve todos

* cache best update

* extend cache parts

* is better light client update

* resolve merge conflict

* initial api changes

* add lc update db column

* fmt

* added tests

* add sim

* Merge branch 'unstable' of https://github.com/sigp/lighthouse into persist-light-client-updates

* fix some weird issues with the simulator

* tests

* Merge branch 'unstable' of https://github.com/sigp/lighthouse into persist-light-client-updates

* test changes

* merge conflict

* testing

* started work on ef tests and some code clean up

* update tests

* linting

* noop pre altair, were still failing on electra though

* allow for zeroed light client header

* Merge branch 'unstable' of https://github.com/sigp/lighthouse into persist-light-client-updates

* merge unstable

* remove unwraps

* remove unwraps

* Update light_client_update.rs

* merge unstable

* move functionality to helper methods

* refactor is best update fn

* refactor is best update fn

* improve organization of light client server cache logic

* fork diget calc, and only spawn as many blcoks as we need for the lc update test

* fetch lc update from the cache if it exists

* fmt

* Fix beacon_chain tests

* Add debug code to update ranking_order ef test

* Fix compare code

* merge conflicts

* fix test

* Merge branch 'persist-light-client-updates' of https://github.com/eserilev/lighthouse into persist-light-client-updates

* Use blinded blocks for light client proofs

* fix ef test

* merge conflicts

* fix lc update check

* Lint

* resolve merge conflict

* Merge branch 'persist-light-client-updates' of https://github.com/eserilev/lighthouse into persist-light-client-updates

* revert basic sim

* small fix

* revert sim

* Review PR

* resolve merge conflicts

* Merge branch 'unstable' into persist-light-client-updates

beacon_node/http_api/src/lib.rs

@@ -13,6 +13,7 @@ mod block_rewards;
 mod build_block_contents;
 mod builder_states;
 mod database;
+mod light_client;
 mod metrics;
 mod produce_block;
 mod proposer_duties;
@@ -30,6 +31,7 @@ mod validator_inclusion;
 mod validators;
 mod version;
 
+use crate::light_client::get_light_client_updates;
 use crate::produce_block::{produce_blinded_block_v2, produce_block_v2, produce_block_v3};
 use crate::version::fork_versioned_response;
 use beacon_chain::{
@@ -44,8 +46,8 @@ use bytes::Bytes;
 use directory::DEFAULT_ROOT_DIR;
 use eth2::types::{
     self as api_types, BroadcastValidation, EndpointVersion, ForkChoice, ForkChoiceNode,
-    PublishBlockRequest, ValidatorBalancesRequestBody, ValidatorId, ValidatorStatus,
-    ValidatorsRequestBody,
+    LightClientUpdatesQuery, PublishBlockRequest, ValidatorBalancesRequestBody, ValidatorId,
+    ValidatorStatus, ValidatorsRequestBody,
 };
 use eth2::{CONSENSUS_VERSION_HEADER, CONTENT_TYPE_HEADER, SSZ_CONTENT_TYPE_HEADER};
 use lighthouse_network::{types::SyncState, EnrExt, NetworkGlobals, PeerId, PubsubMessage};
@@ -2484,6 +2486,25 @@ pub fn serve<T: BeaconChainTypes>(
             },
         );
 
+    // GET beacon/light_client/updates
+    let get_beacon_light_client_updates = beacon_light_client_path
+        .clone()
+        .and(task_spawner_filter.clone())
+        .and(warp::path("updates"))
+        .and(warp::path::end())
+        .and(warp::query::<api_types::LightClientUpdatesQuery>())
+        .and(warp::header::optional::<api_types::Accept>("accept"))
+        .then(
+            |chain: Arc<BeaconChain<T>>,
+             task_spawner: TaskSpawner<T::EthSpec>,
+             query: LightClientUpdatesQuery,
+             accept_header: Option<api_types::Accept>| {
+                task_spawner.blocking_response_task(Priority::P1, move || {
+                    get_light_client_updates::<T>(chain, query, accept_header)
+                })
+            },
+        );
+
     /*
      * beacon/rewards
      */
@@ -4640,6 +4661,10 @@ pub fn serve<T: BeaconChainTypes>(
                     enable(ctx.config.enable_light_client_server)
                         .and(get_beacon_light_client_bootstrap),
                 )
+                .uor(
+                    enable(ctx.config.enable_light_client_server)
+                        .and(get_beacon_light_client_updates),
+                )
                 .uor(get_lighthouse_block_packing_efficiency)
                 .uor(get_lighthouse_merge_readiness)
                 .uor(get_events)

beacon_node/http_api/src/light_client.rs

@@ -0,0 +1,143 @@
+use beacon_chain::{BeaconChain, BeaconChainTypes};
+use eth2::types::{
+    self as api_types, ChainSpec, ForkVersionedResponse, LightClientUpdate,
+    LightClientUpdateResponseChunk, LightClientUpdateSszResponse, LightClientUpdatesQuery,
+};
+use ssz::Encode;
+use std::sync::Arc;
+use warp::{
+    hyper::{Body, Response},
+    reply::Reply,
+    Rejection,
+};
+
+use crate::version::{add_ssz_content_type_header, fork_versioned_response, V1};
+
+const MAX_REQUEST_LIGHT_CLIENT_UPDATES: u64 = 128;
+
+pub fn get_light_client_updates<T: BeaconChainTypes>(
+    chain: Arc<BeaconChain<T>>,
+    query: LightClientUpdatesQuery,
+    accept_header: Option<api_types::Accept>,
+) -> Result<Response<Body>, Rejection> {
+    validate_light_client_updates_request(&chain, &query)?;
+
+    let light_client_updates = chain
+        .get_light_client_updates(query.start_period, query.count)
+        .map_err(|_| {
+            warp_utils::reject::custom_not_found("No LightClientUpdates found".to_string())
+        })?;
+
+    match accept_header {
+        Some(api_types::Accept::Ssz) => {
+            let response_chunks = light_client_updates
+                .iter()
+                .map(|update| map_light_client_update_to_ssz_chunk::<T>(&chain, update))
+                .collect::<Vec<LightClientUpdateResponseChunk>>();
+
+            let ssz_response = LightClientUpdateSszResponse {
+                response_chunk_len: (light_client_updates.len() as u64).to_le_bytes().to_vec(),
+                response_chunk: response_chunks.as_ssz_bytes(),
+            }
+            .as_ssz_bytes();
+
+            Response::builder()
+                .status(200)
+                .body(ssz_response)
+                .map(|res: Response<Vec<u8>>| add_ssz_content_type_header(res))
+                .map_err(|e| {
+                    warp_utils::reject::custom_server_error(format!(
+                        "failed to create response: {}",
+                        e
+                    ))
+                })
+        }
+        _ => {
+            let fork_versioned_response = light_client_updates
+                .iter()
+                .map(|update| map_light_client_update_to_json_response::<T>(&chain, update.clone()))
+                .collect::<Result<Vec<ForkVersionedResponse<LightClientUpdate<T::EthSpec>>>, Rejection>>()?;
+            Ok(warp::reply::json(&fork_versioned_response).into_response())
+        }
+    }
+}
+
+pub fn validate_light_client_updates_request<T: BeaconChainTypes>(
+    chain: &BeaconChain<T>,
+    query: &LightClientUpdatesQuery,
+) -> Result<(), Rejection> {
+    if query.count > MAX_REQUEST_LIGHT_CLIENT_UPDATES {
+        return Err(warp_utils::reject::custom_bad_request(
+            "Invalid count requested".to_string(),
+        ));
+    }
+
+    let current_sync_period = chain
+        .epoch()
+        .map_err(|_| {
+            warp_utils::reject::custom_server_error("failed to get current epoch".to_string())
+        })?
+        .sync_committee_period(&chain.spec)
+        .map_err(|_| {
+            warp_utils::reject::custom_server_error(
+                "failed to get current sync committee period".to_string(),
+            )
+        })?;
+
+    if query.start_period > current_sync_period {
+        return Err(warp_utils::reject::custom_bad_request(
+            "Invalid sync committee period requested".to_string(),
+        ));
+    }
+
+    let earliest_altair_sync_committee = chain
+        .spec
+        .altair_fork_epoch
+        .ok_or(warp_utils::reject::custom_server_error(
+            "failed to get altair fork epoch".to_string(),
+        ))?
+        .sync_committee_period(&chain.spec)
+        .map_err(|_| {
+            warp_utils::reject::custom_server_error(
+                "failed to get earliest altair sync committee".to_string(),
+            )
+        })?;
+
+    if query.start_period < earliest_altair_sync_committee {
+        return Err(warp_utils::reject::custom_bad_request(
+            "Invalid sync committee period requested".to_string(),
+        ));
+    }
+
+    Ok(())
+}
+
+fn map_light_client_update_to_ssz_chunk<T: BeaconChainTypes>(
+    chain: &BeaconChain<T>,
+    light_client_update: &LightClientUpdate<T::EthSpec>,
+) -> LightClientUpdateResponseChunk {
+    let fork_name = chain
+        .spec
+        .fork_name_at_slot::<T::EthSpec>(*light_client_update.signature_slot());
+
+    let fork_digest = ChainSpec::compute_fork_digest(
+        chain.spec.fork_version_for_name(fork_name),
+        chain.genesis_validators_root,
+    );
+
+    LightClientUpdateResponseChunk {
+        context: fork_digest,
+        payload: light_client_update.as_ssz_bytes(),
+    }
+}
+
+fn map_light_client_update_to_json_response<T: BeaconChainTypes>(
+    chain: &BeaconChain<T>,
+    light_client_update: LightClientUpdate<T::EthSpec>,
+) -> Result<ForkVersionedResponse<LightClientUpdate<T::EthSpec>>, Rejection> {
+    let fork_name = chain
+        .spec
+        .fork_name_at_slot::<T::EthSpec>(*light_client_update.signature_slot());
+
+    fork_versioned_response(V1, fork_name, light_client_update)
+}

beacon_node/http_api/tests/tests.rs

@@ -1813,6 +1813,36 @@ impl ApiTester {
         self
     }
 
+    pub async fn test_get_beacon_light_client_updates(self) -> Self {
+        let current_epoch = self.chain.epoch().unwrap();
+        let current_sync_committee_period = current_epoch
+            .sync_committee_period(&self.chain.spec)
+            .unwrap();
+
+        let result = match self
+            .client
+            .get_beacon_light_client_updates::<E>(current_sync_committee_period as u64, 1)
+            .await
+        {
+            Ok(result) => result,
+            Err(e) => panic!("query failed incorrectly: {e:?}"),
+        };
+
+        let expected = self
+            .chain
+            .light_client_server_cache
+            .get_light_client_updates(
+                &self.chain.store,
+                current_sync_committee_period as u64,
+                1,
+                &self.chain.spec,
+            )
+            .unwrap();
+
+        assert_eq!(result.clone().unwrap().len(), expected.len());
+        self
+    }
+
     pub async fn test_get_beacon_light_client_bootstrap(self) -> Self {
         let block_id = BlockId(CoreBlockId::Finalized);
         let (block_root, _, _) = block_id.root(&self.chain).unwrap();
@@ -6171,6 +6201,18 @@ async fn node_get() {
         .await;
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn get_light_client_updates() {
+    let config = ApiTesterConfig {
+        spec: ForkName::Altair.make_genesis_spec(E::default_spec()),
+        ..<_>::default()
+    };
+    ApiTester::new_from_config(config)
+        .await
+        .test_get_beacon_light_client_updates()
+        .await;
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn get_light_client_bootstrap() {
     let config = ApiTesterConfig {