Skip to content

Commit 33fd007

Browse files
Vid201Vid201
committed
feat: p2p status message
1 parent 0395f24 commit 33fd007

File tree

14 files changed

+125
-33
lines changed

14 files changed

+125
-33
lines changed

‎crates/grpc/src/uopool.rs

Lines changed: 17 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -449,9 +449,24 @@ where
449449
info!("Starting p2p mode without bootnodes");
450450
}
451451

452-
let mut p2p_network = Network::new(config.clone(), mempool_channels)
452+
// fetch latest block information for p2p
453+
let latest_block_number = eth_client
454+
.get_block_number()
453455
.await
454-
.expect("p2p network init failed");
456+
.expect("get block number failed (needed for p2p)");
457+
let latest_block_hash = eth_client
458+
.get_block(latest_block_number)
459+
.await
460+
.expect("get block hash failed (needed for p2p)")
461+
.expect("get block hash failed (needed for p2p)");
462+
463+
let mut p2p_network = Network::new(
464+
config.clone(),
465+
(latest_block_hash.hash.unwrap_or_default(), latest_block_number.as_u64()),
466+
mempool_channels,
467+
)
468+
.await
469+
.expect("p2p network init failed");
455470

456471
tokio::spawn(async move {
457472
loop {

‎crates/mempool/src/builder.rs

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -107,13 +107,31 @@ where
107107

108108
pub fn register_block_updates(&self, mut block_stream: BlockStream) {
109109
let mut uopool = self.uopool();
110+
let network = self.network.clone();
110111
tokio::spawn(async move {
111112
while let Some(hash) = block_stream.next().await {
112113
if let Ok(hash) = hash {
113114
let h: H256 = hash;
114115
let _ = Self::handle_block_update(h, &mut uopool)
115116
.await
116117
.map_err(|e| warn!("Failed to handle block update: {:?}", e));
118+
119+
// update p2p latest block info
120+
if let Some(ref network) = network {
121+
if let Ok(block_number) =
122+
uopool.entry_point.eth_client().get_block_number().await.map_err(|e| {
123+
warn!("Failed to get block number: {:?}", e);
124+
e
125+
})
126+
{
127+
let _ = network
128+
.unbounded_send(NetworkMessage::NewBlock {
129+
block_hash: hash,
130+
block_number: block_number.as_u64(),
131+
})
132+
.map_err(|e| warn!("Failed to send new block message: {:?}", e));
133+
}
134+
}
117135
}
118136
}
119137
});

‎crates/mempool/src/mempool.rs

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -312,6 +312,7 @@ pub trait UserOperationAct:
312312
AddRemoveUserOp + UserOperationOp + ClearOp + Send + Sync + DynClone
313313
{
314314
}
315+
315316
dyn_clone::clone_trait_object!(UserOperationAct);
316317
impl<T> UserOperationAct for T where
317318
T: AddRemoveUserOp + UserOperationOp + ClearOp + Send + Sync + Clone
@@ -328,6 +329,7 @@ pub trait UserOperationAddrAct:
328329
AddRemoveUserOpHash + UserOperationAddrOp + ClearOp + Send + Sync + DynClone
329330
{
330331
}
332+
331333
dyn_clone::clone_trait_object!(UserOperationAddrAct);
332334
impl<T> UserOperationAddrAct for T where
333335
T: AddRemoveUserOpHash + UserOperationAddrOp + ClearOp + Send + Sync + Clone
@@ -338,6 +340,7 @@ pub trait UserOperationCodeHashAct:
338340
UserOperationCodeHashOp + ClearOp + Send + Sync + DynClone
339341
{
340342
}
343+
341344
dyn_clone::clone_trait_object!(UserOperationCodeHashAct);
342345
impl<T> UserOperationCodeHashAct for T where
343346
T: UserOperationCodeHashOp + ClearOp + Send + Sync + Clone
@@ -366,6 +369,7 @@ impl Mempool {
366369
user_operations_code_hashes,
367370
}
368371
}
372+
369373
pub fn add(&mut self, uo: UserOperation) -> Result<UserOperationHash, MempoolErrorKind> {
370374
let (sender, factory, paymaster) = uo.get_entities();
371375
let uo_hash = uo.hash;
@@ -379,12 +383,14 @@ impl Mempool {
379383
}
380384
Ok(uo_hash)
381385
}
386+
382387
pub fn get(
383388
&self,
384389
uo_hash: &UserOperationHash,
385390
) -> Result<Option<UserOperation>, MempoolErrorKind> {
386391
self.user_operations.get_by_uo_hash(uo_hash)
387392
}
393+
388394
pub fn get_all_by_sender(&self, addr: &Address) -> Vec<UserOperation> {
389395
let uos_by_sender = self.user_operations_by_sender.get_all_by_address(addr);
390396
uos_by_sender
@@ -393,12 +399,15 @@ impl Mempool {
393399
.flatten()
394400
.collect()
395401
}
402+
396403
pub fn get_number_by_sender(&self, addr: &Address) -> usize {
397404
self.user_operations_by_sender.get_number_by_address(addr)
398405
}
406+
399407
pub fn get_number_by_entity(&self, addr: &Address) -> usize {
400408
self.user_operations_by_entity.get_number_by_address(addr)
401409
}
410+
402411
pub fn get_prev_by_sender(&self, uo: &UserOperation) -> Option<UserOperation> {
403412
self.user_operations_by_sender
404413
.get_all_by_address(&uo.sender)
@@ -408,22 +417,26 @@ impl Mempool {
408417
.filter(|uo_prev| uo_prev.nonce == uo.nonce)
409418
.max_by_key(|uo_prev| uo_prev.max_priority_fee_per_gas)
410419
}
420+
411421
pub fn has_code_hashes(&self, uo_hash: &UserOperationHash) -> Result<bool, MempoolErrorKind> {
412422
self.user_operations_code_hashes.has_code_hashes(uo_hash)
413423
}
424+
414425
pub fn set_code_hashes(
415426
&mut self,
416427
uo_hash: &UserOperationHash,
417428
hashes: Vec<CodeHash>,
418429
) -> Result<(), MempoolErrorKind> {
419430
self.user_operations_code_hashes.set_code_hashes(uo_hash, hashes)
420431
}
432+
421433
pub fn get_code_hashes(
422434
&self,
423435
uo_hash: &UserOperationHash,
424436
) -> Result<Vec<CodeHash>, MempoolErrorKind> {
425437
self.user_operations_code_hashes.get_code_hashes(uo_hash)
426438
}
439+
427440
pub fn remove(&mut self, uo_hash: &UserOperationHash) -> Result<bool, MempoolErrorKind> {
428441
let uo = if let Some(user_op) = self.user_operations.get_by_uo_hash(uo_hash)? {
429442
user_op
@@ -449,6 +462,7 @@ impl Mempool {
449462

450463
Ok(true)
451464
}
465+
452466
pub fn remove_by_entity(&mut self, entity: &Address) -> Result<(), MempoolErrorKind> {
453467
let uos = self.user_operations_by_entity.get_all_by_address(entity);
454468

@@ -458,13 +472,16 @@ impl Mempool {
458472

459473
Ok(())
460474
}
475+
461476
// Get UserOperations sorted by max_priority_fee_per_gas without dup sender
462477
pub fn get_sorted(&self) -> Result<Vec<UserOperation>, MempoolErrorKind> {
463478
self.user_operations.get_sorted()
464479
}
480+
465481
pub fn get_all(&self) -> Result<Vec<UserOperation>, MempoolErrorKind> {
466482
self.user_operations.get_all()
467483
}
484+
468485
pub fn clear(&mut self) {
469486
self.user_operations.clear();
470487
self.user_operations_by_sender.clear();

‎crates/mempool/src/validate/sanity/sender.rs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -42,7 +42,7 @@ impl<M: Middleware> SanityCheck<M> for Sender {
4242
(!code.is_empty() && !uo.init_code.is_empty())
4343
{
4444
return Err(SanityError::Sender {
45-
inner: format!("sender {0} is an existing contract, or the initCode {1} is not empty (but not both)", uo.sender, uo.init_code),
45+
inner: format!("sender {0:?} is an existing contract, or the initCode {1} is not empty (but not both)", uo.sender, uo.init_code),
4646
});
4747
}
4848

‎crates/p2p/src/rpc/codec/ssz_snappy.rs

Lines changed: 3 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ use crate::rpc::{
33
methods::{
44
GoodbyeReason, MetaData, MetaDataRequest, Ping, PooledUserOpHashesRequest,
55
PooledUserOpHashesResponse, PooledUserOpsByHashRequest, PooledUserOpsByHashResponse,
6-
RPCResponse, StatusMessage,
6+
RPCResponse, Status,
77
},
88
outbound::OutboundRequest,
99
protocol::{InboundRequest, Protocol, ProtocolId},
@@ -80,10 +80,8 @@ impl Decoder for SSZSnappyInboundCodec {
8080
let mut buffer = vec![];
8181
snap::read::FrameDecoder::<&[u8]>::new(src).read_to_end(&mut buffer)?;
8282

83-
trace!("Inbound request buffer {:?}", buffer);
84-
8583
let request = match self.protocol.protocol {
86-
Protocol::Status => InboundRequest::Status(StatusMessage::deserialize(&buffer)?),
84+
Protocol::Status => InboundRequest::Status(Status::deserialize(&buffer)?),
8785
Protocol::Goodbye => InboundRequest::Goodbye(GoodbyeReason::deserialize(&buffer)?),
8886
Protocol::Ping => InboundRequest::Ping(Ping::deserialize(&buffer)?),
8987
Protocol::MetaData => InboundRequest::MetaData(MetaDataRequest::deserialize(&buffer)?),
@@ -156,8 +154,6 @@ impl Decoder for SSZSnappyOutboundCodec {
156154
type Error = Error;
157155

158156
fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
159-
trace!("Outbound response buffer {:?}", src);
160-
161157
// response_chunk ::= <result> | <encoding-dependent-header> | <encoded-payload>
162158

163159
// TODO: response chunks
@@ -173,9 +169,7 @@ impl Decoder for SSZSnappyOutboundCodec {
173169
snap::read::FrameDecoder::<&[u8]>::new(src).read_to_end(&mut decompressed_data)?;
174170

175171
let response = match self.protocol.protocol {
176-
Protocol::Status => {
177-
RPCResponse::Status(StatusMessage::deserialize(&decompressed_data)?)
178-
}
172+
Protocol::Status => RPCResponse::Status(Status::deserialize(&decompressed_data)?),
179173
Protocol::Goodbye => {
180174
RPCResponse::Goodbye(GoodbyeReason::deserialize(&decompressed_data)?)
181175
}

‎crates/p2p/src/rpc/handler.rs

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -193,8 +193,6 @@ impl RPCHandler {
193193
bytes.clear();
194194
codec.encode(response, &mut bytes)?;
195195

196-
trace!("Sending {:?} bytes", bytes.len());
197-
198196
socket.write_all(&bytes).await?;
199197
socket.close().await?;
200198

@@ -228,8 +226,6 @@ impl RPCHandler {
228226
let mut codec = SSZSnappyOutboundCodec::new(protocol_id);
229227
codec.encode(request, &mut bytes)?;
230228

231-
trace!("Sending {:?} bytes", bytes.len());
232-
233229
socket.write_all(&bytes).await?;
234230
socket.close().await?;
235231

‎crates/p2p/src/rpc/methods.rs

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -14,10 +14,10 @@ pub struct MetaData {
1414
}
1515

1616
#[derive(ssz_rs_derive::Serializable, Clone, Debug, PartialEq, Default)]
17-
pub struct StatusMessage {
18-
chain_id: u64,
19-
block_hash: [u8; 32],
20-
block_number: u64,
17+
pub struct Status {
18+
pub chain_id: u64,
19+
pub block_hash: [u8; 32],
20+
pub block_number: u64,
2121
}
2222

2323
#[derive(Clone, Debug, PartialEq, Default)]
@@ -127,7 +127,7 @@ pub struct RequestId(pub(crate) u64);
127127

128128
#[derive(Debug, Clone, PartialEq)]
129129
pub enum RPCResponse {
130-
Status(StatusMessage),
130+
Status(Status),
131131
Goodbye(GoodbyeReason),
132132
Pong(Ping),
133133
MetaData(MetaData),

‎crates/p2p/src/rpc/outbound.rs

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
use super::{
22
methods::{
33
GoodbyeReason, MetaDataRequest, Ping, PooledUserOpHashesRequest,
4-
PooledUserOpsByHashRequest, StatusMessage,
4+
PooledUserOpsByHashRequest, Status,
55
},
66
protocol::{InboundRequest, Protocol, ProtocolId},
77
};
@@ -10,7 +10,7 @@ use libp2p::{core::UpgradeInfo, OutboundUpgrade, Stream};
1010

1111
#[derive(Debug, Clone, PartialEq)]
1212
pub enum OutboundRequest {
13-
Status(StatusMessage),
13+
Status(Status),
1414
Goodbye(GoodbyeReason),
1515
Ping(Ping),
1616
MetaData(MetaDataRequest),

‎crates/p2p/src/rpc/protocol.rs

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
use super::{
22
methods::{
33
GoodbyeReason, MetaDataRequest, Ping, PooledUserOpHashesRequest,
4-
PooledUserOpsByHashRequest, StatusMessage,
4+
PooledUserOpsByHashRequest, Status,
55
},
66
outbound::OutboundRequest,
77
};
@@ -99,7 +99,7 @@ impl Display for Encoding {
9999

100100
#[derive(Debug, Clone, PartialEq)]
101101
pub enum InboundRequest {
102-
Status(StatusMessage),
102+
Status(Status),
103103
Goodbye(GoodbyeReason),
104104
Ping(Ping),
105105
MetaData(MetaDataRequest),

‎crates/p2p/src/service/mod.rs

Lines changed: 29 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@ use crate::{
1515
},
1616
peer_manager::{PeerManager, PeerManagerEvent},
1717
rpc::{
18-
methods::{MetaData, MetaDataRequest, Ping, RPCResponse, RequestId},
18+
methods::{MetaData, MetaDataRequest, Ping, RPCResponse, RequestId, Status},
1919
outbound::OutboundRequest,
2020
protocol::InboundRequest,
2121
RPCEvent, RPC,
@@ -32,7 +32,7 @@ use crate::{
3232
};
3333
use alloy_chains::Chain;
3434
use discv5::Enr;
35-
use ethers::types::Address;
35+
use ethers::types::{Address, H256};
3636
use futures::channel::{
3737
mpsc::{UnboundedReceiver, UnboundedSender},
3838
oneshot::Sender,
@@ -129,7 +129,11 @@ impl From<Network> for Swarm<Behaviour> {
129129
}
130130

131131
impl Network {
132-
pub async fn new(config: Config, mempool_channels: Vec<MempoolChannel>) -> eyre::Result<Self> {
132+
pub async fn new(
133+
config: Config,
134+
latest_block: (H256, u64),
135+
mempool_channels: Vec<MempoolChannel>,
136+
) -> eyre::Result<Self> {
133137
// Handle private key
134138
let key = if let Some(key) = load_private_key_from_file(&config.node_key_file) {
135139
key
@@ -205,6 +209,8 @@ impl Network {
205209
metadata,
206210
trusted_peers,
207211
config.chain_spec.clone(),
212+
latest_block.0,
213+
latest_block.1,
208214
))
209215
};
210216

@@ -274,6 +280,14 @@ impl Network {
274280
self.network_globals.local_metadata()
275281
}
276282

283+
pub fn status(&self) -> Status {
284+
Status {
285+
chain_id: self.network_globals.chain_spec().chain.id(),
286+
block_hash: *self.network_globals.latest_block_hash().as_fixed_bytes(),
287+
block_number: self.network_globals.latest_block_number(),
288+
}
289+
}
290+
277291
/// handle gossipsub event
278292
fn handle_gossipsub_event(&self, event: Box<gossipsub::Event>) -> Option<NetworkEvent> {
279293
match *event {
@@ -357,6 +371,11 @@ impl Network {
357371
.expect("channel should exist");
358372
None
359373
}
374+
InboundRequest::Status(_status) => {
375+
// TODO: verify status message
376+
sender.send(RPCResponse::Status(self.status())).expect("channel should exist");
377+
None
378+
}
360379
InboundRequest::Goodbye(_) => None,
361380
_ => Some(NetworkEvent::RequestMessage { peer_id, request, sender }),
362381
},
@@ -475,6 +494,13 @@ impl Network {
475494
}
476495
}
477496
}
497+
NetworkMessage::NewBlock { block_hash, block_number } => {
498+
let mut latest_block_hash = self.network_globals.latest_block_hash.write();
499+
*latest_block_hash = block_hash;
500+
let mut latest_block_number =
501+
self.network_globals.latest_block_number.write();
502+
*latest_block_number = block_number;
503+
}
478504
_ => {}
479505
}
480506
}

0 commit comments

Comments
 (0)