pezkuwichain/pezkuwi-subxt
1
0
Fork 0
mirror of https://github.com/pezkuwichain/pezkuwi-subxt.git synced 2026-04-26 16:57:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

overseer: send_msg should not return an error (#1995)

Browse Source 
* send_message should not return an error

* Apply suggestions from code review

Co-authored-by: Peter Goodspeed-Niklaus <coriolinus@users.noreply.github.com>

* s/send_logging_error/send_and_log_error

Co-authored-by: Peter Goodspeed-Niklaus <coriolinus@users.noreply.github.com>
This commit is contained in:
Andronik Ordian
2020-11-23 12:42:14 +01:00
committed by GitHub
parent 8cdb063f72
commit 69b103b1d5
21 changed files with 330 additions and 462 deletions
Download Patch File Download Diff File Expand all files Collapse all files
polkadot/node/network/statement-distribution/src/lib.rs
+38 -51
View File
@@ -523,7 +523,7 @@ async fn circulate_statement_and_dependents(
relay_parent: Hash,
statement: SignedFullStatement,
metrics: &Metrics,
) -> SubsystemResult<()> {
) {
if let Some(active_head)= active_heads.get_mut(&relay_parent) {
// First circulate the statement directly to all peers needing it.
@@ -532,7 +532,7 @@ async fn circulate_statement_and_dependents(
match active_head.note_statement(statement) {
NotedStatement::Fresh(stored) => Some((
*stored.compact().candidate_hash(),
circulate_statement(peers, ctx, relay_parent, stored).await?,
circulate_statement(peers, ctx, relay_parent, stored).await,
)),
_ => None,
}
@@ -552,13 +552,11 @@ async fn circulate_statement_and_dependents(
candidate_hash,
&*active_head,
metrics,
).await?;
).await;
}
}
}
}
Ok(())
}
fn statement_message(relay_parent: Hash, statement: SignedFullStatement)
@@ -577,7 +575,7 @@ async fn circulate_statement(
ctx: &mut impl SubsystemContext<Message = StatementDistributionMessage>,
relay_parent: Hash,
stored: &StoredStatement,
) -> SubsystemResult<Vec<PeerId>> {
) -> Vec<PeerId> {
let fingerprint = stored.fingerprint();
let mut peers_to_send = HashMap::new();
@@ -594,14 +592,14 @@ async fn circulate_statement(
ctx.send_message(AllMessages::NetworkBridge(NetworkBridgeMessage::SendValidationMessage(
peers_to_send.keys().cloned().collect(),
payload,
))).await?;
))).await;
}
Ok(peers_to_send.into_iter().filter_map(|(peer, needs_dependent)| if needs_dependent {
peers_to_send.into_iter().filter_map(|(peer, needs_dependent)| if needs_dependent {
Some(peer)
} else {
None
}).collect())
}).collect()
}
/// Send all statements about a given candidate hash to a peer.
@@ -614,7 +612,7 @@ async fn send_statements_about(
candidate_hash: CandidateHash,
active_head: &ActiveHeadData,
metrics: &Metrics,
) -> SubsystemResult<()> {
) {
for statement in active_head.statements_about(candidate_hash) {
if peer_data.send(&relay_parent, &statement.fingerprint()).is_some() {
let payload = statement_message(
@@ -624,13 +622,11 @@ async fn send_statements_about(
ctx.send_message(AllMessages::NetworkBridge(
NetworkBridgeMessage::SendValidationMessage(vec![peer.clone()], payload)
)).await?;
)).await;
metrics.on_statement_distributed();
}
}
Ok(())
}
/// Send all statements at a given relay-parent to a peer.
@@ -642,7 +638,7 @@ async fn send_statements(
relay_parent: Hash,
active_head: &ActiveHeadData,
metrics: &Metrics,
) -> SubsystemResult<()> {
) {
for statement in active_head.statements() {
if peer_data.send(&relay_parent, &statement.fingerprint()).is_some() {
let payload = statement_message(
@@ -652,20 +648,18 @@ async fn send_statements(
ctx.send_message(AllMessages::NetworkBridge(
NetworkBridgeMessage::SendValidationMessage(vec![peer.clone()], payload)
)).await?;
)).await;
metrics.on_statement_distributed();
}
}
Ok(())
}
async fn report_peer(
ctx: &mut impl SubsystemContext,
peer: PeerId,
rep: Rep,
) -> SubsystemResult<()> {
) {
ctx.send_message(AllMessages::NetworkBridge(
NetworkBridgeMessage::ReportPeer(peer, rep)
)).await
@@ -685,13 +679,14 @@ async fn handle_incoming_message<'a>(
ctx: &mut impl SubsystemContext<Message = StatementDistributionMessage>,
message: protocol_v1::StatementDistributionMessage,
metrics: &Metrics,
) -> SubsystemResult<Option<(Hash, &'a StoredStatement)>> {
) -> Option<(Hash, &'a StoredStatement)> {
let (relay_parent, statement) = match message {
protocol_v1::StatementDistributionMessage::Statement(r, s) => (r, s),
};
if !our_view.contains(&relay_parent) {
return report_peer(ctx, peer, COST_UNEXPECTED_STATEMENT).await.map(|_| None);
report_peer(ctx, peer, COST_UNEXPECTED_STATEMENT).await;
return None;
}
let active_head = match active_heads.get_mut(&relay_parent) {
@@ -703,13 +698,14 @@ async fn handle_incoming_message<'a>(
requested_relay_parent = %relay_parent,
"our view out-of-sync with active heads; head not found",
);
return Ok(None);
return None;
}
};
// check the signature on the statement.
if let Err(()) = check_statement_signature(&active_head, relay_parent, &statement) {
return report_peer(ctx, peer, COST_INVALID_SIGNATURE).await.map(|_| None);
report_peer(ctx, peer, COST_INVALID_SIGNATURE).await;
return None;
}
// Ensure the statement is stored in the peer data.
@@ -720,8 +716,8 @@ async fn handle_incoming_message<'a>(
let max_message_count = active_head.validators.len() * 2;
match peer_data.receive(&relay_parent, &fingerprint, max_message_count) {
Err(rep) => {
report_peer(ctx, peer, rep).await?;
return Ok(None)
report_peer(ctx, peer, rep).await;
return None;
}
Ok(true) => {
// Send the peer all statements concerning the candidate that we have,
@@ -734,7 +730,7 @@ async fn handle_incoming_message<'a>(
fingerprint.0.candidate_hash().clone(),
&*active_head,
metrics,
).await?
).await;
}
Ok(false) => {}
}
@@ -742,14 +738,14 @@ async fn handle_incoming_message<'a>(
// Note: `peer_data.receive` already ensures that the statement is not an unbounded equivocation
// or unpinned to a seconded candidate. So it is safe to place it into the storage.
match active_head.note_statement(statement) {
NotedStatement::NotUseful => Ok(None),
NotedStatement::NotUseful => None,
NotedStatement::UsefulButKnown => {
report_peer(ctx, peer, BENEFIT_VALID_STATEMENT).await?;
Ok(None)
report_peer(ctx, peer, BENEFIT_VALID_STATEMENT).await;
None
}
NotedStatement::Fresh(statement) => {
report_peer(ctx, peer, BENEFIT_VALID_STATEMENT_FIRST).await?;
Ok(Some((relay_parent, statement)))
report_peer(ctx, peer, BENEFIT_VALID_STATEMENT_FIRST).await;
Some((relay_parent, statement))
}
}
}
@@ -763,7 +759,7 @@ async fn update_peer_view_and_send_unlocked(
active_heads: &HashMap<Hash, ActiveHeadData>,
new_view: View,
metrics: &Metrics,
) -> SubsystemResult<()> {
) {
let old_view = std::mem::replace(&mut peer_data.view, new_view);
// Remove entries for all relay-parents in the old view but not the new.
@@ -785,11 +781,9 @@ async fn update_peer_view_and_send_unlocked(
new,
active_head,
metrics,
).await?;
).await;
}
}
Ok(())
}
#[tracing::instrument(level = "trace", skip(peers, active_heads, ctx, metrics), fields(subsystem = LOG_TARGET))]
@@ -800,19 +794,16 @@ async fn handle_network_update(
our_view: &mut View,
update: NetworkBridgeEvent<protocol_v1::StatementDistributionMessage>,
metrics: &Metrics,
) -> SubsystemResult<()> {
) {
match update {
NetworkBridgeEvent::PeerConnected(peer, _role) => {
peers.insert(peer, PeerData {
view: Default::default(),
view_knowledge: Default::default(),
});
Ok(())
}
NetworkBridgeEvent::PeerDisconnected(peer) => {
peers.remove(&peer);
Ok(())
}
NetworkBridgeEvent::PeerMessage(peer, message) => {
match peers.get_mut(&peer) {
@@ -825,7 +816,7 @@ async fn handle_network_update(
ctx,
message,
metrics,
).await?;
).await;
if let Some((relay_parent, new)) = new_stored {
// When we receive a new message from a peer, we forward it to the
@@ -833,12 +824,10 @@ async fn handle_network_update(
let message = AllMessages::CandidateBacking(
CandidateBackingMessage::Statement(relay_parent, new.statement.clone())
);
ctx.send_message(message).await?;
ctx.send_message(message).await;
}
Ok(())
}
None => Ok(()),
None => (),
}
}
@@ -854,7 +843,7 @@ async fn handle_network_update(
metrics,
).await
}
None => Ok(()),
None => (),
}
}
NetworkBridgeEvent::OurViewChange(view) => {
@@ -872,8 +861,6 @@ async fn handle_network_update(
);
}
}
Ok(())
}
}
@@ -917,7 +904,7 @@ impl StatementDistribution {
ctx.send_messages(
std::iter::once(val_message).chain(std::iter::once(session_message))
).await?;
).await;
match (val_rx.await?, session_rx.await?) {
(Ok(v), Ok(s)) => (v, s),
@@ -959,7 +946,7 @@ impl StatementDistribution {
relay_parent,
statement,
&metrics,
).await?;
).await;
}
StatementDistributionMessage::NetworkBridgeUpdateV1(event) => {
let _timer = metrics.time_network_bridge_update_v1();
@@ -971,7 +958,7 @@ impl StatementDistribution {
&mut our_view,
event,
&metrics,
).await?
).await;
}
StatementDistributionMessage::RegisterStatementListener(tx) => {
statement_listeners.push(tx);
@@ -1428,7 +1415,7 @@ mod tests {
&active_heads,
new_view.clone(),
&Default::default(),
).await.unwrap();
).await;
assert_eq!(peer_data.view, new_view);
assert!(!peer_data.view_knowledge.contains_key(&hash_a));
@@ -1544,7 +1531,7 @@ mod tests {
&mut ctx,
hash_b,
&statement,
).await.unwrap();
).await;
{
assert_eq!(needs_dependents.len(), 2);