From b0774e0274b54bc7b88ac5b731d7edef3dbf9b41 Mon Sep 17 00:00:00 2001 From: dapplion <35266934+dapplion@users.noreply.github.com> Date: Sun, 26 Jan 2025 14:17:04 +0400 Subject: [PATCH 1/4] Remove unnecessary range checks --- .../network/src/sync/range_sync/batch.rs | 30 ------------------- 1 file changed, 30 deletions(-) diff --git a/beacon_node/network/src/sync/range_sync/batch.rs b/beacon_node/network/src/sync/range_sync/batch.rs index 53fb55b14da..fd7a8094a93 100644 --- a/beacon_node/network/src/sync/range_sync/batch.rs +++ b/beacon_node/network/src/sync/range_sync/batch.rs @@ -277,36 +277,6 @@ impl BatchInfo { > { match self.state.poison() { BatchState::Downloading(peer, _request_id) => { - // verify that blocks are in range - if let Some(last_slot) = blocks.last().map(|b| b.slot()) { - // the batch is non-empty - let first_slot = blocks[0].slot(); - - let failed_range = if first_slot < self.start_slot { - Some((self.start_slot, first_slot)) - } else if self.end_slot < last_slot { - Some((self.end_slot, last_slot)) - } else { - None - }; - - if let Some((expected, received)) = failed_range { - // this is a failed download, register the attempt and check if the batch - // can be tried again - self.failed_download_attempts.push(peer); - self.state = if self.failed_download_attempts.len() - >= B::max_batch_download_attempts() as usize - { - BatchState::Failed - } else { - // drop the blocks - BatchState::AwaitingDownload - }; - - return Err(Ok((expected, received, self.outcome()))); - } - } - let received = blocks.len(); self.state = BatchState::AwaitingProcessing(peer, blocks, Instant::now()); Ok(received) From f60ce3fc8de13c8e4f208cd35b88a356c618b51f Mon Sep 17 00:00:00 2001 From: dapplion <35266934+dapplion@users.noreply.github.com> Date: Sun, 26 Jan 2025 12:00:03 +0400 Subject: [PATCH 2/4] Consistent logging in sync rpc requests --- .../network/src/sync/network_context.rs | 161 ++++++++++-------- 1 file changed, 92 insertions(+), 69 deletions(-) diff --git a/beacon_node/network/src/sync/network_context.rs b/beacon_node/network/src/sync/network_context.rs index 3047d5d3721..c4447f4ed98 100644 --- a/beacon_node/network/src/sync/network_context.rs +++ b/beacon_node/network/src/sync/network_context.rs @@ -37,6 +37,7 @@ use requests::{ use slog::{debug, error, warn}; use std::collections::hash_map::Entry; use std::collections::{HashMap, HashSet}; +use std::fmt::Debug; use std::sync::Arc; use std::time::Duration; use tokio::sync::mpsc; @@ -533,15 +534,6 @@ impl SyncNetworkContext { let req_id = self.next_id(); let id = SingleLookupReqId { lookup_id, req_id }; - debug!( - self.log, - "Sending BlocksByRoot Request"; - "method" => "BlocksByRoot", - "block_root" => ?block_root, - "peer" => %peer_id, - "id" => ?id - ); - let request = BlocksByRootSingleRequest(block_root); // Lookup sync event safety: If network_send.send() returns Ok(_) we are guaranteed that @@ -558,6 +550,15 @@ impl SyncNetworkContext { }) .map_err(|_| RpcRequestSendError::NetworkSendError)?; + debug!( + self.log, + "Sync RPC request sent"; + "method" => "BlocksByRoot", + "block_root" => ?block_root, + "peer" => %peer_id, + "id" => ?id + ); + self.blocks_by_root_requests.insert( id, peer_id, @@ -616,19 +617,9 @@ impl SyncNetworkContext { let req_id = self.next_id(); let id = SingleLookupReqId { lookup_id, req_id }; - debug!( - self.log, - "Sending BlobsByRoot Request"; - "method" => "BlobsByRoot", - "block_root" => ?block_root, - "blob_indices" => ?indices, - "peer" => %peer_id, - "id" => ?id - ); - let request = BlobsByRootSingleBlockRequest { block_root, - indices, + indices: indices.clone(), }; // Lookup sync event safety: Refer to `Self::block_lookup_request` `network_send.send` call @@ -640,6 +631,16 @@ impl SyncNetworkContext { }) .map_err(|_| RpcRequestSendError::NetworkSendError)?; + debug!( + self.log, + "Sync RPC request sent"; + "method" => "BlobsByRoot", + "block_root" => ?block_root, + "blob_indices" => ?indices, + "peer" => %peer_id, + "id" => ?id + ); + self.blobs_by_root_requests.insert( id, peer_id, @@ -665,9 +666,16 @@ impl SyncNetworkContext { id: self.next_id(), requester, }; + + self.send_network_msg(NetworkMessage::SendRequest { + peer_id, + request: RequestType::DataColumnsByRoot(request.clone().into_request(&self.chain.spec)), + request_id: AppRequestId::Sync(SyncRequestId::DataColumnsByRoot(req_id)), + })?; + debug!( self.log, - "Sending DataColumnsByRoot Request"; + "Sync RPC request sent"; "method" => "DataColumnsByRoot", "block_root" => ?request.block_root, "indices" => ?request.indices, @@ -676,12 +684,6 @@ impl SyncNetworkContext { "req_id" => %req_id, ); - self.send_network_msg(NetworkMessage::SendRequest { - peer_id, - request: RequestType::DataColumnsByRoot(request.clone().into_request(&self.chain.spec)), - request_id: AppRequestId::Sync(SyncRequestId::DataColumnsByRoot(req_id)), - })?; - self.data_columns_by_root_requests.insert( req_id, peer_id, @@ -765,22 +767,23 @@ impl SyncNetworkContext { id: self.next_id(), parent_request_id, }; + self.network_send + .send(NetworkMessage::SendRequest { + peer_id, + request: RequestType::BlocksByRange(request.clone().into()), + request_id: AppRequestId::Sync(SyncRequestId::BlocksByRange(id)), + }) + .map_err(|_| RpcRequestSendError::NetworkSendError)?; + debug!( self.log, - "Sending BlocksByRange request"; + "Sync RPC request sent"; "method" => "BlocksByRange", "count" => request.count(), "epoch" => Slot::new(*request.start_slot()).epoch(T::EthSpec::slots_per_epoch()), "peer" => %peer_id, "id" => ?id, ); - self.network_send - .send(NetworkMessage::SendRequest { - peer_id, - request: RequestType::BlocksByRange(request.clone().into()), - request_id: AppRequestId::Sync(SyncRequestId::BlocksByRange(id)), - }) - .map_err(|_| RpcRequestSendError::NetworkSendError)?; self.blocks_by_range_requests.insert( id, @@ -804,15 +807,6 @@ impl SyncNetworkContext { parent_request_id, }; let request_epoch = Slot::new(request.start_slot).epoch(T::EthSpec::slots_per_epoch()); - debug!( - self.log, - "Sending BlobsByRange requests"; - "method" => "BlobsByRange", - "count" => request.count, - "epoch" => request_epoch, - "peer" => %peer_id, - "id" => ?id, - ); // Create the blob request based on the blocks request. self.network_send @@ -823,6 +817,16 @@ impl SyncNetworkContext { }) .map_err(|_| RpcRequestSendError::NetworkSendError)?; + debug!( + self.log, + "Sync RPC request sent"; + "method" => "BlobsByRange", + "count" => request.count, + "epoch" => request_epoch, + "peer" => %peer_id, + "id" => ?id, + ); + let max_blobs_per_block = self.chain.spec.max_blobs_per_block(request_epoch); self.blobs_by_range_requests.insert( id, @@ -845,9 +849,17 @@ impl SyncNetworkContext { id: self.next_id(), parent_request_id, }; + + self.send_network_msg(NetworkMessage::SendRequest { + peer_id, + request: RequestType::DataColumnsByRange(request.clone()), + request_id: AppRequestId::Sync(SyncRequestId::DataColumnsByRange(id)), + }) + .map_err(|_| RpcRequestSendError::NetworkSendError)?; + debug!( self.log, - "Sending DataColumnsByRange requests"; + "Sync RPC request sent"; "method" => "DataColumnsByRange", "count" => request.count, "epoch" => Slot::new(request.start_slot).epoch(T::EthSpec::slots_per_epoch()), @@ -856,13 +868,6 @@ impl SyncNetworkContext { "id" => ?id, ); - self.send_network_msg(NetworkMessage::SendRequest { - peer_id, - request: RequestType::DataColumnsByRange(request.clone()), - request_id: AppRequestId::Sync(SyncRequestId::DataColumnsByRange(id)), - }) - .map_err(|_| RpcRequestSendError::NetworkSendError)?; - self.data_columns_by_range_requests.insert( id, peer_id, @@ -1006,8 +1011,8 @@ impl SyncNetworkContext { peer_id: PeerId, rpc_event: RpcEvent>>, ) -> Option>>> { - let response = self.blocks_by_root_requests.on_response(id, rpc_event); - let response = response.map(|res| { + let resp = self.blocks_by_root_requests.on_response(id, rpc_event); + let resp = resp.map(|res| { res.and_then(|(mut blocks, seen_timestamp)| { // Enforce that exactly one chunk = one block is returned. ReqResp behavior limits the // response count to at most 1. @@ -1019,10 +1024,7 @@ impl SyncNetworkContext { } }) }); - if let Some(Err(RpcResponseError::VerifyError(e))) = &response { - self.report_peer(peer_id, PeerAction::LowToleranceError, e.into()); - } - response + self.on_rpc_response_result(id, "BlocksByRoot", resp, peer_id, |_| 1) } pub(crate) fn on_single_blob_response( @@ -1031,8 +1033,8 @@ impl SyncNetworkContext { peer_id: PeerId, rpc_event: RpcEvent>>, ) -> Option>> { - let response = self.blobs_by_root_requests.on_response(id, rpc_event); - let response = response.map(|res| { + let resp = self.blobs_by_root_requests.on_response(id, rpc_event); + let resp = resp.map(|res| { res.and_then(|(blobs, seen_timestamp)| { if let Some(max_len) = blobs .first() @@ -1051,10 +1053,7 @@ impl SyncNetworkContext { } }) }); - if let Some(Err(RpcResponseError::VerifyError(e))) = &response { - self.report_peer(peer_id, PeerAction::LowToleranceError, e.into()); - } - response + self.on_rpc_response_result(id, "BlobsByRoot", resp, peer_id, |_| 1) } #[allow(clippy::type_complexity)] @@ -1067,7 +1066,7 @@ impl SyncNetworkContext { let resp = self .data_columns_by_root_requests .on_response(id, rpc_event); - self.report_rpc_response_errors(resp, peer_id) + self.on_rpc_response_result(id, "DataColumnsByRoot", resp, peer_id, |_| 1) } #[allow(clippy::type_complexity)] @@ -1078,7 +1077,7 @@ impl SyncNetworkContext { rpc_event: RpcEvent>>, ) -> Option>>>> { let resp = self.blocks_by_range_requests.on_response(id, rpc_event); - self.report_rpc_response_errors(resp, peer_id) + self.on_rpc_response_result(id, "BlocksByRange", resp, peer_id, |b| b.len()) } #[allow(clippy::type_complexity)] @@ -1089,7 +1088,7 @@ impl SyncNetworkContext { rpc_event: RpcEvent>>, ) -> Option>>>> { let resp = self.blobs_by_range_requests.on_response(id, rpc_event); - self.report_rpc_response_errors(resp, peer_id) + self.on_rpc_response_result(id, "BlobsByRangeRequest", resp, peer_id, |b| b.len()) } #[allow(clippy::type_complexity)] @@ -1102,14 +1101,38 @@ impl SyncNetworkContext { let resp = self .data_columns_by_range_requests .on_response(id, rpc_event); - self.report_rpc_response_errors(resp, peer_id) + self.on_rpc_response_result(id, "DataColumnsByRange", resp, peer_id, |d| d.len()) } - fn report_rpc_response_errors( + fn on_rpc_response_result usize>( &mut self, + id: I, + method: &'static str, resp: Option>, peer_id: PeerId, + get_count: F, ) -> Option> { + match &resp { + None => {} + Some(Ok((v, _))) => { + debug!( + self.log, + "Sync RPC request completed"; + "id" => ?id, + "method" => method, + "count" => get_count(v) + ); + } + Some(Err(e)) => { + debug!( + self.log, + "Sync RPC request error"; + "id" => ?id, + "method" => method, + "error" => ?e + ); + } + } if let Some(Err(RpcResponseError::VerifyError(e))) = &resp { self.report_peer(peer_id, PeerAction::LowToleranceError, e.into()); } From cff9910c4a5b4dd6de56309cec3dfea187a08621 Mon Sep 17 00:00:00 2001 From: dapplion <35266934+dapplion@users.noreply.github.com> Date: Sun, 26 Jan 2025 13:01:02 +0400 Subject: [PATCH 3/4] Concise request Ids for sync --- .../src/service/api_types.rs | 148 ++++++++++++++++-- .../network/src/sync/network_context.rs | 59 +++---- 2 files changed, 170 insertions(+), 37 deletions(-) diff --git a/beacon_node/lighthouse_network/src/service/api_types.rs b/beacon_node/lighthouse_network/src/service/api_types.rs index c4bdc37d982..3b86e3e3a93 100644 --- a/beacon_node/lighthouse_network/src/service/api_types.rs +++ b/beacon_node/lighthouse_network/src/service/api_types.rs @@ -1,16 +1,15 @@ -use std::sync::Arc; - +use crate::rpc::{ + methods::{ResponseTermination, RpcResponse, RpcSuccessResponse, StatusMessage}, + SubstreamId, +}; use libp2p::swarm::ConnectionId; +use std::fmt::{Display, Formatter}; +use std::sync::Arc; use types::{ BlobSidecar, DataColumnSidecar, Epoch, EthSpec, Hash256, LightClientBootstrap, LightClientFinalityUpdate, LightClientOptimisticUpdate, LightClientUpdate, SignedBeaconBlock, }; -use crate::rpc::{ - methods::{ResponseTermination, RpcResponse, RpcSuccessResponse, StatusMessage}, - SubstreamId, -}; - /// Identifier of requests sent by a peer. pub type PeerRequestId = (ConnectionId, SubstreamId); @@ -235,9 +234,138 @@ impl slog::Value for RequestId { } } +// Since each request Id is deeply nested with various types, if rendered with Debug on logs they +// take too much visual space. This custom Display implementations make the overall Id short while +// not losing information +impl Display for BlocksByRangeRequestId { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "{}/{}", self.id, self.parent_request_id) + } +} + +impl Display for BlobsByRangeRequestId { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "{}/{}", self.id, self.parent_request_id) + } +} + +impl Display for DataColumnsByRangeRequestId { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "{}/{}", self.id, self.parent_request_id) + } +} + +impl Display for ComponentsByRangeRequestId { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "{}/{}", self.id, self.requester) + } +} + +impl Display for SingleLookupReqId { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "{}/Lookup/{}", self.req_id, self.lookup_id) + } +} + // This custom impl reduces log boilerplate not printing `DataColumnsByRootRequestId` on each id log -impl std::fmt::Display for DataColumnsByRootRequestId { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "{} {:?}", self.id, self.requester) +impl Display for DataColumnsByRootRequestId { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "{}/{}", self.id, self.requester) + } +} + +impl Display for DataColumnsByRootRequester { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + match self { + Self::Custody(id) => write!(f, "Custody/{id}"), + Self::Sampling(id) => write!(f, "Sampling/{id}"), + } + } +} + +impl Display for CustodyId { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.requester) + } +} + +impl Display for CustodyRequester { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.0) + } +} + +impl Display for RangeRequestId { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + match self { + Self::RangeSync { chain_id, batch_id } => write!(f, "RangeSync/{batch_id}/{chain_id}"), + Self::BackfillSync { batch_id } => write!(f, "BackfillSync/{batch_id}"), + } + } +} + +impl Display for SamplingId { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "{}/{}", self.sampling_request_id, self.id) + } +} + +impl Display for SamplingRequestId { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.0) + } +} + +impl Display for SamplingRequester { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + match self { + Self::ImportedBlock(block) => write!(f, "ImportedBlock/{block}"), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn display_id_data_columns_by_root_custody() { + let id = DataColumnsByRootRequestId { + id: 123, + requester: DataColumnsByRootRequester::Custody(CustodyId { + requester: CustodyRequester(SingleLookupReqId { + req_id: 121, + lookup_id: 101, + }), + }), + }; + assert_eq!(format!("{id}"), "123/Custody/121/Lookup/101"); + } + + #[test] + fn display_id_data_columns_by_root_sampling() { + let id = DataColumnsByRootRequestId { + id: 123, + requester: DataColumnsByRootRequester::Sampling(SamplingId { + id: SamplingRequester::ImportedBlock(Hash256::ZERO), + sampling_request_id: SamplingRequestId(101), + }), + }; + assert_eq!(format!("{id}"), "123/Sampling/101/ImportedBlock/0x0000000000000000000000000000000000000000000000000000000000000000"); + } + + #[test] + fn display_id_data_columns_by_range() { + let id = DataColumnsByRangeRequestId { + id: 123, + parent_request_id: ComponentsByRangeRequestId { + id: 122, + requester: RangeRequestId::RangeSync { + chain_id: 5492900659401505034, + batch_id: Epoch::new(0), + }, + }, + }; + assert_eq!(format!("{id}"), "123/122/RangeSync/0/5492900659401505034"); } } diff --git a/beacon_node/network/src/sync/network_context.rs b/beacon_node/network/src/sync/network_context.rs index c4447f4ed98..94d9c3dfbac 100644 --- a/beacon_node/network/src/sync/network_context.rs +++ b/beacon_node/network/src/sync/network_context.rs @@ -531,8 +531,10 @@ impl SyncNetworkContext { } } - let req_id = self.next_id(); - let id = SingleLookupReqId { lookup_id, req_id }; + let id = SingleLookupReqId { + lookup_id, + req_id: self.next_id(), + }; let request = BlocksByRootSingleRequest(block_root); @@ -556,7 +558,7 @@ impl SyncNetworkContext { "method" => "BlocksByRoot", "block_root" => ?block_root, "peer" => %peer_id, - "id" => ?id + "id" => %id ); self.blocks_by_root_requests.insert( @@ -568,7 +570,7 @@ impl SyncNetworkContext { BlocksByRootRequestItems::new(request), ); - Ok(LookupRequestResult::RequestSent(req_id)) + Ok(LookupRequestResult::RequestSent(id.req_id)) } /// Request necessary blobs for `block_root`. Requests only the necessary blobs by checking: @@ -614,8 +616,10 @@ impl SyncNetworkContext { return Ok(LookupRequestResult::NoRequestNeeded("no indices to fetch")); } - let req_id = self.next_id(); - let id = SingleLookupReqId { lookup_id, req_id }; + let id = SingleLookupReqId { + lookup_id, + req_id: self.next_id(), + }; let request = BlobsByRootSingleBlockRequest { block_root, @@ -638,7 +642,7 @@ impl SyncNetworkContext { "block_root" => ?block_root, "blob_indices" => ?indices, "peer" => %peer_id, - "id" => ?id + "id" => %id ); self.blobs_by_root_requests.insert( @@ -651,7 +655,7 @@ impl SyncNetworkContext { BlobsByRootRequestItems::new(request), ); - Ok(LookupRequestResult::RequestSent(req_id)) + Ok(LookupRequestResult::RequestSent(id.req_id)) } /// Request to send a single `data_columns_by_root` request to the network. @@ -662,7 +666,7 @@ impl SyncNetworkContext { request: DataColumnsByRootSingleBlockRequest, expect_max_responses: bool, ) -> Result, &'static str> { - let req_id = DataColumnsByRootRequestId { + let id = DataColumnsByRootRequestId { id: self.next_id(), requester, }; @@ -670,7 +674,7 @@ impl SyncNetworkContext { self.send_network_msg(NetworkMessage::SendRequest { peer_id, request: RequestType::DataColumnsByRoot(request.clone().into_request(&self.chain.spec)), - request_id: AppRequestId::Sync(SyncRequestId::DataColumnsByRoot(req_id)), + request_id: AppRequestId::Sync(SyncRequestId::DataColumnsByRoot(id)), })?; debug!( @@ -680,18 +684,17 @@ impl SyncNetworkContext { "block_root" => ?request.block_root, "indices" => ?request.indices, "peer" => %peer_id, - "requester" => ?requester, - "req_id" => %req_id, + "id" => %id, ); self.data_columns_by_root_requests.insert( - req_id, + id, peer_id, expect_max_responses, DataColumnsByRootRequestItems::new(request), ); - Ok(LookupRequestResult::RequestSent(req_id)) + Ok(LookupRequestResult::RequestSent(id)) } /// Request to fetch all needed custody columns of a specific block. This function may not send @@ -724,15 +727,17 @@ impl SyncNetworkContext { return Ok(LookupRequestResult::NoRequestNeeded("no indices to fetch")); } - let req_id = self.next_id(); - let id = SingleLookupReqId { lookup_id, req_id }; + let id = SingleLookupReqId { + lookup_id, + req_id: self.next_id(), + }; debug!( self.log, "Starting custody columns request"; "block_root" => ?block_root, "indices" => ?custody_indexes_to_fetch, - "id" => ?id + "id" => %id ); let requester = CustodyRequester(id); @@ -751,7 +756,7 @@ impl SyncNetworkContext { // created cannot return data immediately, it must send some request to the network // first. And there must exist some request, `custody_indexes_to_fetch` is not empty. self.custody_by_root_requests.insert(requester, request); - Ok(LookupRequestResult::RequestSent(req_id)) + Ok(LookupRequestResult::RequestSent(id.req_id)) } Err(e) => Err(RpcRequestSendError::CustodyRequestError(e)), } @@ -779,10 +784,10 @@ impl SyncNetworkContext { self.log, "Sync RPC request sent"; "method" => "BlocksByRange", - "count" => request.count(), + "slots" => request.count(), "epoch" => Slot::new(*request.start_slot()).epoch(T::EthSpec::slots_per_epoch()), "peer" => %peer_id, - "id" => ?id, + "id" => %id, ); self.blocks_by_range_requests.insert( @@ -821,10 +826,10 @@ impl SyncNetworkContext { self.log, "Sync RPC request sent"; "method" => "BlobsByRange", - "count" => request.count, + "slots" => request.count, "epoch" => request_epoch, "peer" => %peer_id, - "id" => ?id, + "id" => %id, ); let max_blobs_per_block = self.chain.spec.max_blobs_per_block(request_epoch); @@ -861,11 +866,11 @@ impl SyncNetworkContext { self.log, "Sync RPC request sent"; "method" => "DataColumnsByRange", - "count" => request.count, + "slots" => request.count, "epoch" => Slot::new(request.start_slot).epoch(T::EthSpec::slots_per_epoch()), "columns" => ?request.columns, "peer" => %peer_id, - "id" => ?id, + "id" => %id, ); self.data_columns_by_range_requests.insert( @@ -1104,7 +1109,7 @@ impl SyncNetworkContext { self.on_rpc_response_result(id, "DataColumnsByRange", resp, peer_id, |d| d.len()) } - fn on_rpc_response_result usize>( + fn on_rpc_response_result usize>( &mut self, id: I, method: &'static str, @@ -1118,7 +1123,7 @@ impl SyncNetworkContext { debug!( self.log, "Sync RPC request completed"; - "id" => ?id, + "id" => %id, "method" => method, "count" => get_count(v) ); @@ -1127,7 +1132,7 @@ impl SyncNetworkContext { debug!( self.log, "Sync RPC request error"; - "id" => ?id, + "id" => %id, "method" => method, "error" => ?e ); From 3919f2a7524fa389bc822c555379e1ec6ef9fbe7 Mon Sep 17 00:00:00 2001 From: dapplion <35266934+dapplion@users.noreply.github.com> Date: Sun, 26 Jan 2025 14:16:55 +0400 Subject: [PATCH 4/4] Add log when sending batch to processor --- .../network/src/network_beacon_processor/mod.rs | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/beacon_node/network/src/network_beacon_processor/mod.rs b/beacon_node/network/src/network_beacon_processor/mod.rs index 4a3fb28e102..21de2c3400e 100644 --- a/beacon_node/network/src/network_beacon_processor/mod.rs +++ b/beacon_node/network/src/network_beacon_processor/mod.rs @@ -612,8 +612,13 @@ impl NetworkBeaconProcessor { process_id: ChainSegmentProcessId, blocks: Vec>, ) -> Result<(), Error> { - let is_backfill = matches!(&process_id, ChainSegmentProcessId::BackSyncBatchId { .. }); + debug!(self.log, "Batch sending for process"; + "blocks" => blocks.len(), + "id" => ?process_id, + ); + let processor = self.clone(); + let id = process_id.clone(); let process_fn = async move { let notify_execution_layer = if processor .network_globals @@ -626,17 +631,16 @@ impl NetworkBeaconProcessor { NotifyExecutionLayer::Yes }; processor - .process_chain_segment(process_id, blocks, notify_execution_layer) + .process_chain_segment(id, blocks, notify_execution_layer) .await; }; let process_fn = Box::pin(process_fn); // Back-sync batches are dispatched with a different `Work` variant so // they can be rate-limited. - let work = if is_backfill { - Work::ChainSegmentBackfill(process_fn) - } else { - Work::ChainSegment(process_fn) + let work = match process_id { + ChainSegmentProcessId::RangeBatchId { .. } => Work::ChainSegment(process_fn), + ChainSegmentProcessId::BackSyncBatchId { .. } => Work::ChainSegmentBackfill(process_fn), }; self.try_send(BeaconWorkEvent {