gnosis/lighthouse
1
0
Fork 0
mirror of https://github.com/sigp/lighthouse.git synced 2026-04-17 12:58:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'sigp/unstable' into validator-custody-cgc-mutex

Browse Source
This commit is contained in:
dapplion
2025-04-11 18:08:57 -03:00
parent b18bb4e8d0 70f8ab9a6f
commit 5e8fb9ca55
155 changed files with 4073 additions and 3192 deletions
Download Patch File Download Diff File Expand all files Collapse all files

135
beacon_node/lighthouse_network/tests/rpc_tests.rs
View File

@@ -5,7 +5,7 @@ mod common;
use common::{build_tracing_subscriber, Protocol};
use lighthouse_network::rpc::{methods::*, RequestType};
use lighthouse_network::service::api_types::AppRequestId;
use lighthouse_network::{rpc::max_rpc_size, NetworkEvent, ReportSource, Response};
use lighthouse_network::{NetworkEvent, ReportSource, Response};
use ssz::Encode;
use ssz_types::VariableList;
use std::sync::Arc;
@@ -15,37 +15,37 @@ use tokio::time::sleep;
use tracing::{debug, warn};
use types::{
BeaconBlock, BeaconBlockAltair, BeaconBlockBase, BeaconBlockBellatrix, BlobSidecar, ChainSpec,
EmptyBlock, Epoch, EthSpec, FixedBytesExtended, ForkContext, ForkName, Hash256, MinimalEthSpec,
EmptyBlock, Epoch, EthSpec, FixedBytesExtended, ForkName, Hash256, MinimalEthSpec,
RuntimeVariableList, Signature, SignedBeaconBlock, Slot,
};
type E = MinimalEthSpec;
/// Bellatrix block with length < max_rpc_size.
fn bellatrix_block_small(fork_context: &ForkContext, spec: &ChainSpec) -> BeaconBlock<E> {
fn bellatrix_block_small(spec: &ChainSpec) -> BeaconBlock<E> {
let mut block = BeaconBlockBellatrix::<E>::empty(spec);
let tx = VariableList::from(vec![0; 1024]);
let txs = VariableList::from(std::iter::repeat(tx).take(5000).collect::<Vec<_>>());
let txs = VariableList::from(std::iter::repeat_n(tx, 5000).collect::<Vec<_>>());
block.body.execution_payload.execution_payload.transactions = txs;
let block = BeaconBlock::Bellatrix(block);
assert!(block.ssz_bytes_len() <= max_rpc_size(fork_context, spec.max_chunk_size as usize));
assert!(block.ssz_bytes_len() <= spec.max_payload_size as usize);
block
}
/// Bellatrix block with length > MAX_RPC_SIZE.
/// The max limit for a bellatrix block is in the order of ~16GiB which wouldn't fit in memory.
/// Hence, we generate a bellatrix block just greater than `MAX_RPC_SIZE` to test rejection on the rpc layer.
fn bellatrix_block_large(fork_context: &ForkContext, spec: &ChainSpec) -> BeaconBlock<E> {
fn bellatrix_block_large(spec: &ChainSpec) -> BeaconBlock<E> {
let mut block = BeaconBlockBellatrix::<E>::empty(spec);
let tx = VariableList::from(vec![0; 1024]);
let txs = VariableList::from(std::iter::repeat(tx).take(100000).collect::<Vec<_>>());
let txs = VariableList::from(std::iter::repeat_n(tx, 100000).collect::<Vec<_>>());
block.body.execution_payload.execution_payload.transactions = txs;
let block = BeaconBlock::Bellatrix(block);
assert!(block.ssz_bytes_len() > max_rpc_size(fork_context, spec.max_chunk_size as usize));
assert!(block.ssz_bytes_len() > spec.max_payload_size as usize);
block
}
@@ -98,7 +98,7 @@ fn test_tcp_status_rpc() {
}
NetworkEvent::ResponseReceived {
peer_id: _,
id: AppRequestId::Router,
app_request_id: AppRequestId::Router,
response,
} => {
// Should receive the RPC response
@@ -118,13 +118,17 @@ fn test_tcp_status_rpc() {
match receiver.next_event().await {
NetworkEvent::RequestReceived {
peer_id,
id,
request,
inbound_request_id,
request_type,
} => {
if request.r#type == rpc_request {
if request_type == rpc_request {
// send the response
debug!("Receiver Received");
receiver.send_response(peer_id, id, request.id, rpc_response.clone());
receiver.send_response(
peer_id,
inbound_request_id,
rpc_response.clone(),
);
}
}
_ => {} // Ignore other events
@@ -184,7 +188,7 @@ fn test_tcp_blocks_by_range_chunked_rpc() {
let signed_full_block = SignedBeaconBlock::from_block(full_block, Signature::empty());
let rpc_response_altair = Response::BlocksByRange(Some(Arc::new(signed_full_block)));
let full_block = bellatrix_block_small(&common::fork_context(ForkName::Bellatrix), &spec);
let full_block = bellatrix_block_small(&spec);
let signed_full_block = SignedBeaconBlock::from_block(full_block, Signature::empty());
let rpc_response_bellatrix_small =
Response::BlocksByRange(Some(Arc::new(signed_full_block)));
@@ -204,7 +208,7 @@ fn test_tcp_blocks_by_range_chunked_rpc() {
}
NetworkEvent::ResponseReceived {
peer_id: _,
id: _,
app_request_id: _,
response,
} => {
warn!("Sender received a response");
@@ -240,10 +244,10 @@ fn test_tcp_blocks_by_range_chunked_rpc() {
match receiver.next_event().await {
NetworkEvent::RequestReceived {
peer_id,
id,
request,
inbound_request_id,
request_type,
} => {
if request.r#type == rpc_request {
if request_type == rpc_request {
// send the response
warn!("Receiver got request");
for i in 0..messages_to_send {
@@ -258,16 +262,14 @@ fn test_tcp_blocks_by_range_chunked_rpc() {
};
receiver.send_response(
peer_id,
id,
request.id,
inbound_request_id,
rpc_response.clone(),
);
}
// send the stream termination
receiver.send_response(
peer_id,
id,
request.id,
inbound_request_id,
Response::BlocksByRange(None),
);
}
@@ -338,7 +340,7 @@ fn test_blobs_by_range_chunked_rpc() {
}
NetworkEvent::ResponseReceived {
peer_id: _,
id: _,
app_request_id: _,
response,
} => {
warn!("Sender received a response");
@@ -368,10 +370,10 @@ fn test_blobs_by_range_chunked_rpc() {
match receiver.next_event().await {
NetworkEvent::RequestReceived {
peer_id,
id,
request,
inbound_request_id,
request_type,
} => {
if request.r#type == rpc_request {
if request_type == rpc_request {
// send the response
warn!("Receiver got request");
for _ in 0..messages_to_send {
@@ -379,16 +381,14 @@ fn test_blobs_by_range_chunked_rpc() {
// second as altair and third as bellatrix.
receiver.send_response(
peer_id,
id,
request.id,
inbound_request_id,
rpc_response.clone(),
);
}
// send the stream termination
receiver.send_response(
peer_id,
id,
request.id,
inbound_request_id,
Response::BlobsByRange(None),
);
}
@@ -442,7 +442,7 @@ fn test_tcp_blocks_by_range_over_limit() {
}));
// BlocksByRange Response
let full_block = bellatrix_block_large(&common::fork_context(ForkName::Bellatrix), &spec);
let full_block = bellatrix_block_large(&spec);
let signed_full_block = SignedBeaconBlock::from_block(full_block, Signature::empty());
let rpc_response_bellatrix_large =
Response::BlocksByRange(Some(Arc::new(signed_full_block)));
@@ -459,8 +459,8 @@ fn test_tcp_blocks_by_range_over_limit() {
.unwrap();
}
// The request will fail because the sender will refuse to send anything > MAX_RPC_SIZE
NetworkEvent::RPCFailed { id, .. } => {
assert!(matches!(id, AppRequestId::Router));
NetworkEvent::RPCFailed { app_request_id, .. } => {
assert!(matches!(app_request_id, AppRequestId::Router));
return;
}
_ => {} // Ignore other behaviour events
@@ -474,26 +474,24 @@ fn test_tcp_blocks_by_range_over_limit() {
match receiver.next_event().await {
NetworkEvent::RequestReceived {
peer_id,
id,
request,
inbound_request_id,
request_type,
} => {
if request.r#type == rpc_request {
if request_type == rpc_request {
// send the response
warn!("Receiver got request");
for _ in 0..messages_to_send {
let rpc_response = rpc_response_bellatrix_large.clone();
receiver.send_response(
peer_id,
id,
request.id,
inbound_request_id,
rpc_response.clone(),
);
}
// send the stream termination
receiver.send_response(
peer_id,
id,
request.id,
inbound_request_id,
Response::BlocksByRange(None),
);
}
@@ -566,7 +564,7 @@ fn test_tcp_blocks_by_range_chunked_rpc_terminates_correctly() {
}
NetworkEvent::ResponseReceived {
peer_id: _,
id: _,
app_request_id: _,
response,
} =>
// Should receive the RPC response
@@ -608,15 +606,15 @@ fn test_tcp_blocks_by_range_chunked_rpc_terminates_correctly() {
futures::future::Either::Left((
NetworkEvent::RequestReceived {
peer_id,
id,
request,
inbound_request_id,
request_type,
},
_,
)) => {
if request.r#type == rpc_request {
if request_type == rpc_request {
// send the response
warn!("Receiver got request");
message_info = Some((peer_id, id, request.id));
message_info = Some((peer_id, inbound_request_id));
}
}
futures::future::Either::Right((_, _)) => {} // The timeout hit, send messages if required
@@ -626,8 +624,8 @@ fn test_tcp_blocks_by_range_chunked_rpc_terminates_correctly() {
// if we need to send messages send them here. This will happen after a delay
if message_info.is_some() {
messages_sent += 1;
let (peer_id, stream_id, request_id) = message_info.as_ref().unwrap();
receiver.send_response(*peer_id, *stream_id, *request_id, rpc_response.clone());
let (peer_id, inbound_request_id) = message_info.as_ref().unwrap();
receiver.send_response(*peer_id, *inbound_request_id, rpc_response.clone());
debug!("Sending message {}", messages_sent);
if messages_sent == messages_to_send + extra_messages_to_send {
// stop sending messages
@@ -700,7 +698,7 @@ fn test_tcp_blocks_by_range_single_empty_rpc() {
}
NetworkEvent::ResponseReceived {
peer_id: _,
id: AppRequestId::Router,
app_request_id: AppRequestId::Router,
response,
} => match response {
Response::BlocksByRange(Some(_)) => {
@@ -727,26 +725,24 @@ fn test_tcp_blocks_by_range_single_empty_rpc() {
match receiver.next_event().await {
NetworkEvent::RequestReceived {
peer_id,
id,
request,
inbound_request_id,
request_type,
} => {
if request.r#type == rpc_request {
if request_type == rpc_request {
// send the response
warn!("Receiver got request");
for _ in 1..=messages_to_send {
receiver.send_response(
peer_id,
id,
request.id,
inbound_request_id,
rpc_response.clone(),
);
}
// send the stream termination
receiver.send_response(
peer_id,
id,
request.id,
inbound_request_id,
Response::BlocksByRange(None),
);
}
@@ -817,7 +813,7 @@ fn test_tcp_blocks_by_root_chunked_rpc() {
let signed_full_block = SignedBeaconBlock::from_block(full_block, Signature::empty());
let rpc_response_altair = Response::BlocksByRoot(Some(Arc::new(signed_full_block)));
let full_block = bellatrix_block_small(&common::fork_context(ForkName::Bellatrix), &spec);
let full_block = bellatrix_block_small(&spec);
let signed_full_block = SignedBeaconBlock::from_block(full_block, Signature::empty());
let rpc_response_bellatrix_small =
Response::BlocksByRoot(Some(Arc::new(signed_full_block)));
@@ -837,7 +833,7 @@ fn test_tcp_blocks_by_root_chunked_rpc() {
}
NetworkEvent::ResponseReceived {
peer_id: _,
id: AppRequestId::Router,
app_request_id: AppRequestId::Router,
response,
} => match response {
Response::BlocksByRoot(Some(_)) => {
@@ -870,10 +866,10 @@ fn test_tcp_blocks_by_root_chunked_rpc() {
match receiver.next_event().await {
NetworkEvent::RequestReceived {
peer_id,
id,
request,
inbound_request_id,
request_type,
} => {
if request.r#type == rpc_request {
if request_type == rpc_request {
// send the response
debug!("Receiver got request");
@@ -886,14 +882,13 @@ fn test_tcp_blocks_by_root_chunked_rpc() {
} else {
rpc_response_bellatrix_small.clone()
};
receiver.send_response(peer_id, id, request.id, rpc_response);
receiver.send_response(peer_id, inbound_request_id, rpc_response);
debug!("Sending message");
}
// send the stream termination
receiver.send_response(
peer_id,
id,
request.id,
inbound_request_id,
Response::BlocksByRange(None),
);
debug!("Send stream term");
@@ -977,7 +972,7 @@ fn test_tcp_blocks_by_root_chunked_rpc_terminates_correctly() {
}
NetworkEvent::ResponseReceived {
peer_id: _,
id: AppRequestId::Router,
app_request_id: AppRequestId::Router,
response,
} => {
debug!("Sender received a response");
@@ -1019,15 +1014,15 @@ fn test_tcp_blocks_by_root_chunked_rpc_terminates_correctly() {
futures::future::Either::Left((
NetworkEvent::RequestReceived {
peer_id,
id,
request,
inbound_request_id,
request_type,
},
_,
)) => {
if request.r#type == rpc_request {
if request_type == rpc_request {
// send the response
warn!("Receiver got request");
message_info = Some((peer_id, id, request.id));
message_info = Some((peer_id, inbound_request_id));
}
}
futures::future::Either::Right((_, _)) => {} // The timeout hit, send messages if required
@@ -1037,8 +1032,8 @@ fn test_tcp_blocks_by_root_chunked_rpc_terminates_correctly() {
// if we need to send messages send them here. This will happen after a delay
if message_info.is_some() {
messages_sent += 1;
let (peer_id, stream_id, request_id) = message_info.as_ref().unwrap();
receiver.send_response(*peer_id, *stream_id, *request_id, rpc_response.clone());
let (peer_id, inbound_request_id) = message_info.as_ref().unwrap();
receiver.send_response(*peer_id, *inbound_request_id, rpc_response.clone());
debug!("Sending message {}", messages_sent);
if messages_sent == messages_to_send + extra_messages_to_send {
// stop sending messages