Tree sync WIP

beacon_node/beacon_chain/src/block_verification_types.rs

@@ -188,7 +188,7 @@ impl<E: EthSpec> RpcBlock<E> {
         block: Arc<SignedBeaconBlock<E>>,
         custody_columns: Vec<CustodyDataColumn<E>>,
         spec: &ChainSpec,
-    ) -> Result<Self, AvailabilityCheckError> {
+    ) -> Result<Self, String> {
         let block_root = block_root.unwrap_or_else(|| get_block_root(&block));
 
         let inner = RpcBlockInner::BlockAndCustodyColumns {
@@ -197,11 +197,7 @@ impl<E: EthSpec> RpcBlock<E> {
                 custody_columns,
                 spec.number_of_columns as usize,
             )
-            .map_err(|e| {
-                AvailabilityCheckError::Unexpected(format!(
-                    "custody_columns len exceeds number_of_columns: {e:?}"
-                ))
-            })?,
+            .map_err(|e| format!("custody_columns len exceeds number_of_columns: {e:?}"))?,
         };
         Ok(Self {
             block_root,

beacon_node/beacon_chain/src/data_availability_checker/overflow_lru_cache.rs

@@ -1183,8 +1183,13 @@ mod pending_components_tests {
     pub fn pre_setup() -> Setup<E> {
         let mut rng = StdRng::seed_from_u64(0xDEADBEEF0BAD5EEDu64);
         let spec = test_spec::<E>();
-        let (block, blobs_vec) =
-            generate_rand_block_and_blobs::<E>(ForkName::Deneb, NumBlobs::Random, &mut rng, &spec);
+        let (block, blobs_vec) = generate_rand_block_and_blobs::<E>(
+            ForkName::Deneb,
+            NumBlobs::Random,
+            None,
+            &mut rng,
+            &spec,
+        );
         let max_len = spec.max_blobs_per_block(block.epoch()) as usize;
         let mut blobs: RuntimeFixedVector<Option<Arc<BlobSidecar<E>>>> =
             RuntimeFixedVector::default(max_len);

beacon_node/beacon_chain/src/test_utils.rs

@@ -2392,7 +2392,8 @@ where
                     .take(sampling_column_count)
                     .map(CustodyDataColumn::from_asserted_custody)
                     .collect::<Vec<_>>();
-                RpcBlock::new_with_custody_columns(Some(block_root), block, columns, &self.spec)?
+                RpcBlock::new_with_custody_columns(Some(block_root), block, columns, &self.spec)
+                    .map_err(BlockError::InternalError)?
             } else {
                 RpcBlock::new_without_blobs(Some(block_root), block)
             }
@@ -3144,10 +3145,14 @@ pub enum NumBlobs {
 pub fn generate_rand_block_and_blobs<E: EthSpec>(
     fork_name: ForkName,
     num_blobs: NumBlobs,
+    parent_root: Option<Hash256>,
     rng: &mut impl Rng,
     spec: &ChainSpec,
 ) -> (SignedBeaconBlock<E, FullPayload<E>>, Vec<BlobSidecar<E>>) {
-    let inner = map_fork_name!(fork_name, BeaconBlock, <_>::random_for_test(rng));
+    let mut inner = map_fork_name!(fork_name, BeaconBlock, <_>::random_for_test(rng));
+    if let Some(parent_root) = parent_root {
+        *inner.parent_root_mut() = parent_root;
+    }
 
     let mut block = SignedBeaconBlock::from_block(inner, types::Signature::random_for_test(rng));
     let max_blobs = spec.max_blobs_per_block(block.epoch()) as usize;
@@ -3246,13 +3251,15 @@ pub fn generate_rand_block_and_blobs<E: EthSpec>(
 pub fn generate_rand_block_and_data_columns<E: EthSpec>(
     fork_name: ForkName,
     num_blobs: NumBlobs,
+    parent_root: Option<Hash256>,
     rng: &mut impl Rng,
     spec: &ChainSpec,
 ) -> (
     SignedBeaconBlock<E, FullPayload<E>>,
     DataColumnSidecarList<E>,
 ) {
-    let (block, _blobs) = generate_rand_block_and_blobs(fork_name, num_blobs, rng, spec);
+    let (block, _blobs) =
+        generate_rand_block_and_blobs(fork_name, num_blobs, parent_root, rng, spec);
     let data_columns = generate_data_column_sidecars_from_block(&block, spec);
     (block, data_columns)
 }

beacon_node/beacon_chain/tests/store_tests.rs

@@ -2603,7 +2603,7 @@ async fn weak_subjectivity_sync_test(slots: Vec<Slot>, checkpoint_slot: Slot) {
                 .deconstruct();
             if wss_fork.fulu_enabled() {
                 info!(block_slot = %block.slot(), ?block_root, "Corrupting data column KZG proof");
-                let (mut data_columns, expected_column_indices) = cols.unwrap();
+                let mut data_columns = cols.unwrap();
                 assert!(
                     !data_columns.is_empty(),
                     "data column sidecars shouldn't be empty"
@@ -2618,7 +2618,6 @@ async fn weak_subjectivity_sync_test(slots: Vec<Slot>, checkpoint_slot: Slot) {
                     Some(block_root),
                     block,
                     data_columns.to_vec(),
-                    expected_column_indices,
                     &harness.spec,
                 )
                 .unwrap()
@@ -3819,7 +3818,6 @@ fn available_to_rpc_block<E: EthSpec>(block: AvailableBlock<E>, spec: &ChainSpec
                 .into_iter()
                 .map(|d| CustodyDataColumn::from_asserted_custody(d))
                 .collect(),
-            vec![],
             spec,
         )
         .unwrap(),

beacon_node/client/src/notifier.rs

@@ -146,9 +146,7 @@ pub fn spawn_notifier<T: BeaconChainTypes>(
                             Instant::now(),
                         );
                 }
-                SyncState::SyncingFinalized { .. }
-                | SyncState::SyncingHead { .. }
-                | SyncState::SyncTransition => {
+                SyncState::Syncing { .. } | SyncState::SyncTransition => {
                     speedo.observe(head_slot, Instant::now());
                 }
                 SyncState::Stalled | SyncState::Synced => {}

beacon_node/http_api/src/lib.rs

@@ -472,7 +472,8 @@ pub fn serve<T: BeaconChainTypes>(
                 move |network_globals: Arc<NetworkGlobals<T::EthSpec>>,
                       chain: Arc<BeaconChain<T>>| async move {
                     match *network_globals.sync_state.read() {
-                        SyncState::SyncingFinalized { .. } => {
+                        // TODO(tree-sync): review, we don't have a notion of finalized sync now
+                        SyncState::Syncing { .. } => {
                             let head_slot = chain.canonical_head.cached_head().head_slot();
 
                             let current_slot =
@@ -494,9 +495,7 @@ pub fn serve<T: BeaconChainTypes>(
                                 )))
                             }
                         }
-                        SyncState::SyncingHead { .. }
-                        | SyncState::SyncTransition
-                        | SyncState::BackFillSyncing { .. } => Ok(()),
+                        SyncState::SyncTransition | SyncState::BackFillSyncing { .. } => Ok(()),
                         SyncState::Synced => Ok(()),
                         SyncState::Stalled => Ok(()),
                     }

beacon_node/lighthouse_network/src/peer_manager/network_behaviour.rs

@@ -79,10 +79,7 @@ impl<E: EthSpec> NetworkBehaviour for PeerManager<E> {
             }
         }
 
-        if !matches!(
-            self.network_globals.sync_state(),
-            SyncState::SyncingFinalized { .. } | SyncState::SyncingHead { .. }
-        ) {
+        if !matches!(self.network_globals.sync_state(), SyncState::Syncing { .. }) {
             loop {
                 match self.status_peers.poll_next_unpin(cx) {
                     Poll::Ready(Some(Ok(peer_id))) => {

beacon_node/lighthouse_network/src/rpc/codec.rs

@@ -1088,7 +1088,12 @@ mod tests {
     }
 
     fn bbroot_request_v2(fork_name: ForkName) -> BlocksByRootRequest {
-        BlocksByRootRequest::new(vec![Hash256::zero()], &fork_context(fork_name))
+        let fork_context = fork_context(fork_name);
+        BlocksByRootRequest::new(
+            vec![Hash256::zero()],
+            &fork_context.spec,
+            fork_context.current_fork(),
+        )
     }
 
     fn blbroot_request(fork_name: ForkName) -> BlobsByRootRequest {

beacon_node/lighthouse_network/src/rpc/methods.rs

@@ -16,9 +16,9 @@ use types::blob_sidecar::BlobIdentifier;
 use types::light_client_update::MAX_REQUEST_LIGHT_CLIENT_UPDATES;
 use types::{
     blob_sidecar::BlobSidecar, ChainSpec, ColumnIndex, DataColumnSidecar,
-    DataColumnsByRootIdentifier, Epoch, EthSpec, ForkContext, Hash256, LightClientBootstrap,
-    LightClientFinalityUpdate, LightClientOptimisticUpdate, LightClientUpdate, RuntimeVariableList,
-    SignedBeaconBlock, Slot,
+    DataColumnsByRootIdentifier, Epoch, EthSpec, ForkContext, ForkName, Hash256,
+    LightClientBootstrap, LightClientFinalityUpdate, LightClientOptimisticUpdate,
+    LightClientUpdate, RuntimeVariableList, SignedBeaconBlock, Slot,
 };
 
 /// Maximum length of error message.
@@ -440,10 +440,8 @@ pub struct BlocksByRootRequest {
 }
 
 impl BlocksByRootRequest {
-    pub fn new(block_roots: Vec<Hash256>, fork_context: &ForkContext) -> Self {
-        let max_request_blocks = fork_context
-            .spec
-            .max_request_blocks(fork_context.current_fork());
+    pub fn new(block_roots: Vec<Hash256>, spec: &ChainSpec, current_fork: ForkName) -> Self {
+        let max_request_blocks = spec.max_request_blocks(current_fork);
         let block_roots = RuntimeVariableList::from_vec(block_roots, max_request_blocks);
         Self::V2(BlocksByRootRequestV2 { block_roots })
     }

beacon_node/lighthouse_network/src/rpc/self_limiter.rs

@@ -313,13 +313,22 @@ mod tests {
     use crate::rpc::rate_limiter::Quota;
     use crate::rpc::self_limiter::SelfRateLimiter;
     use crate::rpc::{Ping, Protocol, RPCSend, RequestType};
-    use crate::service::api_types::{AppRequestId, SingleLookupReqId, SyncRequestId};
+    use crate::service::api_types::{
+        AppRequestId, BlocksByRootRequestId, BlocksByRootRequester, HeaderLookupId, SyncRequestId,
+    };
     use libp2p::PeerId;
     use logging::create_test_tracing_subscriber;
     use std::num::NonZeroU64;
     use std::time::Duration;
     use types::{EthSpec, ForkContext, Hash256, MainnetEthSpec, Slot};
 
+    fn get_parent_request_id() -> BlocksByRootRequester {
+        BlocksByRootRequester::Header(HeaderLookupId {
+            id: 0,
+            block_root: Hash256::ZERO,
+        })
+    }
+
     /// Test that `next_peer_request_ready` correctly maintains the queue.
     #[tokio::test]
     async fn test_next_peer_request_ready() {
@@ -336,17 +345,15 @@ mod tests {
         let mut limiter: SelfRateLimiter<AppRequestId, MainnetEthSpec> =
             SelfRateLimiter::new(Some(config), fork_context).unwrap();
         let peer_id = PeerId::random();
-        let lookup_id = 0;
+        let parent_request_id = get_parent_request_id();
 
         for i in 1..=5u32 {
             let _ = limiter.allows(
                 peer_id,
-                AppRequestId::Sync(SyncRequestId::SingleBlock {
-                    id: SingleLookupReqId {
-                        lookup_id,
-                        req_id: i,
-                    },
-                }),
+                AppRequestId::Sync(SyncRequestId::BlocksByRoot(BlocksByRootRequestId {
+                    id: i,
+                    parent_request_id,
+                })),
                 RequestType::Ping(Ping { data: i as u64 }),
             );
         }
@@ -363,9 +370,7 @@ mod tests {
             for i in 2..=5u32 {
                 assert!(matches!(
                     iter.next().unwrap().request_id,
-                    AppRequestId::Sync(SyncRequestId::SingleBlock {
-                        id: SingleLookupReqId { req_id, .. },
-                    }) if req_id == i,
+                    AppRequestId::Sync(SyncRequestId::BlocksByRoot(BlocksByRootRequestId{id,..})) if id == i,
                 ));
             }
 
@@ -388,9 +393,7 @@ mod tests {
             for i in 3..=5 {
                 assert!(matches!(
                     iter.next().unwrap().request_id,
-                    AppRequestId::Sync(SyncRequestId::SingleBlock {
-                        id: SingleLookupReqId { req_id, .. },
-                    }) if req_id == i,
+                    AppRequestId::Sync(SyncRequestId::BlocksByRoot(BlocksByRootRequestId{id,..})) if id == i,
                 ));
             }
 
@@ -409,16 +412,15 @@ mod tests {
         let mut limiter: SelfRateLimiter<AppRequestId, MainnetEthSpec> =
             SelfRateLimiter::new(None, fork_context).unwrap();
         let peer_id = PeerId::random();
+        let parent_request_id = get_parent_request_id();
 
         for i in 1..=5u32 {
             let result = limiter.allows(
                 peer_id,
-                AppRequestId::Sync(SyncRequestId::SingleBlock {
-                    id: SingleLookupReqId {
-                        lookup_id: i,
-                        req_id: i,
-                    },
-                }),
+                AppRequestId::Sync(SyncRequestId::BlocksByRoot(BlocksByRootRequestId {
+                    id: i,
+                    parent_request_id,
+                })),
                 RequestType::Ping(Ping { data: i as u64 }),
             );
 
@@ -469,9 +471,7 @@ mod tests {
 
             assert!(matches!(
                 request_id,
-                AppRequestId::Sync(SyncRequestId::SingleBlock {
-                    id: SingleLookupReqId { req_id, .. },
-                }) if *req_id == i
+                AppRequestId::Sync(SyncRequestId::BlocksByRoot(BlocksByRootRequestId {id,..})) if *id == i
             ));
         }
     }
@@ -487,17 +487,16 @@ mod tests {
             SelfRateLimiter::new(None, fork_context).unwrap();
         let peer1 = PeerId::random();
         let peer2 = PeerId::random();
+        let parent_request_id = get_parent_request_id();
 
         for peer in [peer1, peer2] {
             for i in 1..=5u32 {
                 let result = limiter.allows(
                     peer,
-                    AppRequestId::Sync(SyncRequestId::SingleBlock {
-                        id: SingleLookupReqId {
-                            lookup_id: i,
-                            req_id: i,
-                        },
-                    }),
+                    AppRequestId::Sync(SyncRequestId::BlocksByRoot(BlocksByRootRequestId {
+                        id: i,
+                        parent_request_id,
+                    })),
                     RequestType::Ping(Ping { data: i as u64 }),
                 );
 
@@ -525,9 +524,7 @@ mod tests {
             let (request_id, _) = failed_requests.remove(0);
             assert!(matches!(
                 request_id,
-                AppRequestId::Sync(SyncRequestId::SingleBlock {
-                        id: SingleLookupReqId { req_id, .. },
-                }) if req_id == i
+                AppRequestId::Sync(SyncRequestId::BlocksByRoot(BlocksByRootRequestId{id,..})) if id == i
             ));
         }