Change validation & collation protocol names to include genesis hash & fork id (#5876)

This commit is contained in:
Dmitry Markin
2022-08-30 19:50:22 +03:00
committed by GitHub
parent 6e04112b93
commit 13ea167bd7
19 changed files with 739 additions and 177 deletions
+73 -21
View File
@@ -27,9 +27,11 @@ use sp_consensus::SyncOracle;
use polkadot_node_network_protocol::{
self as net_protocol,
peer_set::{PeerSet, PerPeerSet},
v1 as protocol_v1, ObservedRole, OurView, PeerId, ProtocolVersion,
UnifiedReputationChange as Rep, View,
peer_set::{
CollationVersion, PeerSet, PeerSetProtocolNames, PerPeerSet, ProtocolVersion,
ValidationVersion,
},
v1 as protocol_v1, ObservedRole, OurView, PeerId, UnifiedReputationChange as Rep, View,
};
use polkadot_node_subsystem::{
@@ -80,6 +82,7 @@ pub struct NetworkBridgeRx<N, AD> {
sync_oracle: Box<dyn SyncOracle + Send>,
shared: Shared,
metrics: Metrics,
peerset_protocol_names: PeerSetProtocolNames,
}
impl<N, AD> NetworkBridgeRx<N, AD> {
@@ -92,9 +95,17 @@ impl<N, AD> NetworkBridgeRx<N, AD> {
authority_discovery_service: AD,
sync_oracle: Box<dyn SyncOracle + Send>,
metrics: Metrics,
peerset_protocol_names: PeerSetProtocolNames,
) -> Self {
let shared = Shared::default();
Self { network_service, authority_discovery_service, sync_oracle, shared, metrics }
Self {
network_service,
authority_discovery_service,
sync_oracle,
shared,
metrics,
peerset_protocol_names,
}
}
}
@@ -147,6 +158,7 @@ async fn handle_network_messages<AD>(
mut authority_discovery_service: AD,
metrics: Metrics,
shared: Shared,
peerset_protocol_names: PeerSetProtocolNames,
) -> Result<(), Error>
where
AD: validator_discovery::AuthorityDiscovery + Send,
@@ -166,13 +178,14 @@ where
}) => {
let role = ObservedRole::from(role);
let (peer_set, version) = {
let (peer_set, version) = match PeerSet::try_from_protocol_name(&protocol) {
None => continue,
Some(p) => p,
};
let (peer_set, version) =
match peerset_protocol_names.try_get_protocol(&protocol) {
None => continue,
Some(p) => p,
};
if let Some(fallback) = negotiated_fallback {
match PeerSet::try_from_protocol_name(&fallback) {
match peerset_protocol_names.try_get_protocol(&fallback) {
None => {
gum::debug!(
target: LOG_TARGET,
@@ -210,7 +223,7 @@ where
target: LOG_TARGET,
action = "PeerConnected",
peer_set = ?peer_set,
version,
version = %version,
peer = ?peer,
role = ?role
);
@@ -245,7 +258,7 @@ where
NetworkBridgeEvent::PeerConnected(
peer.clone(),
role,
1,
version,
maybe_authority,
),
NetworkBridgeEvent::PeerViewChange(peer.clone(), View::default()),
@@ -259,6 +272,7 @@ where
vec![peer],
PeerSet::Validation,
version,
&peerset_protocol_names,
WireMessage::<protocol_v1::ValidationProtocol>::ViewUpdate(local_view),
&metrics,
);
@@ -269,7 +283,7 @@ where
NetworkBridgeEvent::PeerConnected(
peer.clone(),
role,
1,
version,
maybe_authority,
),
NetworkBridgeEvent::PeerViewChange(peer.clone(), View::default()),
@@ -283,6 +297,7 @@ where
vec![peer],
PeerSet::Collation,
version,
&peerset_protocol_names,
WireMessage::<protocol_v1::CollationProtocol>::ViewUpdate(local_view),
&metrics,
);
@@ -290,7 +305,7 @@ where
}
},
Some(NetworkEvent::NotificationStreamClosed { remote: peer, protocol }) => {
let (peer_set, version) = match PeerSet::try_from_protocol_name(&protocol) {
let (peer_set, version) = match peerset_protocol_names.try_get_protocol(&protocol) {
None => continue,
Some(peer_set) => peer_set,
};
@@ -317,7 +332,7 @@ where
w
};
if was_connected && version == peer_set.get_default_version() {
if was_connected && version == peer_set.get_main_version() {
match peer_set {
PeerSet::Validation =>
dispatch_validation_event_to_all(
@@ -355,7 +370,8 @@ where
.filter_map(|(protocol, msg_bytes)| {
// version doesn't matter because we always receive on the 'correct'
// protocol name, not the negotiated fallback.
let (peer_set, _version) = PeerSet::try_from_protocol_name(protocol)?;
let (peer_set, _version) =
peerset_protocol_names.try_get_protocol(protocol)?;
if peer_set == PeerSet::Validation {
if expected_versions[PeerSet::Validation].is_none() {
return Some(Err(UNCONNECTED_PEERSET_COST))
@@ -384,7 +400,8 @@ where
.filter_map(|(protocol, msg_bytes)| {
// version doesn't matter because we always receive on the 'correct'
// protocol name, not the negotiated fallback.
let (peer_set, _version) = PeerSet::try_from_protocol_name(protocol)?;
let (peer_set, _version) =
peerset_protocol_names.try_get_protocol(protocol)?;
if peer_set == PeerSet::Collation {
if expected_versions[PeerSet::Collation].is_none() {
@@ -422,7 +439,9 @@ where
if !v_messages.is_empty() {
let (events, reports) =
if expected_versions[PeerSet::Validation] == Some(1) {
if expected_versions[PeerSet::Validation] ==
Some(ValidationVersion::V1.into())
{
handle_v1_peer_messages::<protocol_v1::ValidationProtocol, _>(
remote.clone(),
PeerSet::Validation,
@@ -453,7 +472,9 @@ where
if !c_messages.is_empty() {
let (events, reports) =
if expected_versions[PeerSet::Collation] == Some(1) {
if expected_versions[PeerSet::Collation] ==
Some(CollationVersion::V1.into())
{
handle_v1_peer_messages::<protocol_v1::CollationProtocol, _>(
remote.clone(),
PeerSet::Collation,
@@ -494,6 +515,7 @@ async fn run_incoming_orchestra_signals<Context, N, AD>(
shared: Shared,
sync_oracle: Box<dyn SyncOracle + Send>,
metrics: Metrics,
peerset_protocol_names: PeerSetProtocolNames,
) -> Result<(), Error>
where
N: Network,
@@ -574,6 +596,7 @@ where
&shared,
finalized_number,
&metrics,
&peerset_protocol_names,
);
}
}
@@ -619,6 +642,7 @@ where
metrics,
sync_oracle,
shared,
peerset_protocol_names,
} = bridge;
let (task, network_event_handler) = handle_network_messages(
@@ -628,6 +652,7 @@ where
authority_discovery_service.clone(),
metrics.clone(),
shared.clone(),
peerset_protocol_names.clone(),
)
.remote_handle();
@@ -641,6 +666,7 @@ where
shared,
sync_oracle,
metrics,
peerset_protocol_names,
);
futures::pin_mut!(orchestra_signal_handler);
@@ -667,6 +693,7 @@ fn update_our_view<Net, Context>(
shared: &Shared,
finalized_number: BlockNumber,
metrics: &Metrics,
peerset_protocol_names: &PeerSetProtocolNames,
) where
Net: Network,
{
@@ -700,11 +727,18 @@ fn update_our_view<Net, Context>(
send_validation_message_v1(
net,
validation_peers,
peerset_protocol_names,
WireMessage::ViewUpdate(new_view.clone()),
metrics,
);
send_collation_message_v1(net, collation_peers, WireMessage::ViewUpdate(new_view), metrics);
send_collation_message_v1(
net,
collation_peers,
peerset_protocol_names,
WireMessage::ViewUpdate(new_view),
metrics,
);
let our_view = OurView::new(
live_heads.iter().take(MAX_VIEW_HEADS).cloned().map(|a| (a.hash, a.span)),
@@ -778,19 +812,37 @@ fn handle_v1_peer_messages<RawMessage: Decode, OutMessage: From<RawMessage>>(
fn send_validation_message_v1(
net: &mut impl Network,
peers: Vec<PeerId>,
peerset_protocol_names: &PeerSetProtocolNames,
message: WireMessage<protocol_v1::ValidationProtocol>,
metrics: &Metrics,
) {
send_message(net, peers, PeerSet::Validation, 1, message, metrics);
send_message(
net,
peers,
PeerSet::Validation,
ValidationVersion::V1.into(),
peerset_protocol_names,
message,
metrics,
);
}
fn send_collation_message_v1(
net: &mut impl Network,
peers: Vec<PeerId>,
peerset_protocol_names: &PeerSetProtocolNames,
message: WireMessage<protocol_v1::CollationProtocol>,
metrics: &Metrics,
) {
send_message(net, peers, PeerSet::Collation, 1, message, metrics)
send_message(
net,
peers,
PeerSet::Collation,
CollationVersion::V1.into(),
peerset_protocol_names,
message,
metrics,
);
}
async fn dispatch_validation_event_to_all(
+74 -17
View File
@@ -31,6 +31,7 @@ use std::{
use sc_network::{Event as NetworkEvent, IfDisconnected};
use polkadot_node_network_protocol::{
peer_set::PeerSetProtocolNames,
request_response::{outgoing::Requests, ReqProtocolNames},
view, ObservedRole, Versioned,
};
@@ -46,7 +47,7 @@ use polkadot_node_subsystem_test_helpers::{
SingleItemSink, SingleItemStream, TestSubsystemContextHandle,
};
use polkadot_node_subsystem_util::metered;
use polkadot_primitives::v2::AuthorityDiscoveryId;
use polkadot_primitives::v2::{AuthorityDiscoveryId, Hash};
use sc_network::Multiaddr;
use sp_keyring::Sr25519Keyring;
@@ -68,6 +69,7 @@ pub enum NetworkAction {
struct TestNetwork {
net_events: Arc<Mutex<Option<SingleItemStream<NetworkEvent>>>>,
action_tx: Arc<Mutex<metered::UnboundedMeteredSender<NetworkAction>>>,
protocol_names: Arc<PeerSetProtocolNames>,
}
#[derive(Clone, Debug)]
@@ -78,9 +80,12 @@ struct TestAuthorityDiscovery;
struct TestNetworkHandle {
action_rx: metered::UnboundedMeteredReceiver<NetworkAction>,
net_tx: SingleItemSink<NetworkEvent>,
protocol_names: PeerSetProtocolNames,
}
fn new_test_network() -> (TestNetwork, TestNetworkHandle, TestAuthorityDiscovery) {
fn new_test_network(
protocol_names: PeerSetProtocolNames,
) -> (TestNetwork, TestNetworkHandle, TestAuthorityDiscovery) {
let (net_tx, net_rx) = polkadot_node_subsystem_test_helpers::single_item_sink();
let (action_tx, action_rx) = metered::unbounded();
@@ -88,8 +93,9 @@ fn new_test_network() -> (TestNetwork, TestNetworkHandle, TestAuthorityDiscovery
TestNetwork {
net_events: Arc::new(Mutex::new(Some(net_rx))),
action_tx: Arc::new(Mutex::new(action_tx)),
protocol_names: Arc::new(protocol_names.clone()),
},
TestNetworkHandle { action_rx, net_tx },
TestNetworkHandle { action_rx, net_tx, protocol_names },
TestAuthorityDiscovery,
)
}
@@ -130,14 +136,20 @@ impl Network for TestNetwork {
.unwrap();
}
fn disconnect_peer(&self, who: PeerId, peer_set: PeerSet) {
fn disconnect_peer(&self, who: PeerId, protocol: Cow<'static, str>) {
let (peer_set, version) = self.protocol_names.try_get_protocol(&protocol).unwrap();
assert_eq!(version, peer_set.get_main_version());
self.action_tx
.lock()
.unbounded_send(NetworkAction::DisconnectPeer(who, peer_set))
.unwrap();
}
fn write_notification(&self, who: PeerId, peer_set: PeerSet, message: Vec<u8>) {
fn write_notification(&self, who: PeerId, protocol: Cow<'static, str>, message: Vec<u8>) {
let (peer_set, version) = self.protocol_names.try_get_protocol(&protocol).unwrap();
assert_eq!(version, peer_set.get_main_version());
self.action_tx
.lock()
.unbounded_send(NetworkAction::WriteNotification(who, peer_set, message))
@@ -181,7 +193,7 @@ impl TestNetworkHandle {
async fn connect_peer(&mut self, peer: PeerId, peer_set: PeerSet, role: ObservedRole) {
self.send_network_event(NetworkEvent::NotificationStreamOpened {
remote: peer,
protocol: peer_set.into_default_protocol_name(),
protocol: self.protocol_names.get_main_name(peer_set),
negotiated_fallback: None,
role: role.into(),
})
@@ -191,7 +203,7 @@ impl TestNetworkHandle {
async fn disconnect_peer(&mut self, peer: PeerId, peer_set: PeerSet) {
self.send_network_event(NetworkEvent::NotificationStreamClosed {
remote: peer,
protocol: peer_set.into_default_protocol_name(),
protocol: self.protocol_names.get_main_name(peer_set),
})
.await;
}
@@ -199,7 +211,7 @@ impl TestNetworkHandle {
async fn peer_message(&mut self, peer: PeerId, peer_set: PeerSet, message: Vec<u8>) {
self.send_network_event(NetworkEvent::NotificationsReceived {
remote: peer,
messages: vec![(peer_set.into_default_protocol_name(), message.into())],
messages: vec![(self.protocol_names.get_main_name(peer_set), message.into())],
})
.await;
}
@@ -285,8 +297,12 @@ fn test_harness<T: Future<Output = VirtualOverseer>>(
sync_oracle: Box<dyn SyncOracle + Send>,
test: impl FnOnce(TestHarness) -> T,
) {
let genesis_hash = Hash::repeat_byte(0xff);
let fork_id = None;
let peerset_protocol_names = PeerSetProtocolNames::new(genesis_hash, fork_id);
let pool = sp_core::testing::TaskExecutor::new();
let (mut network, network_handle, discovery) = new_test_network();
let (mut network, network_handle, discovery) = new_test_network(peerset_protocol_names.clone());
let (context, virtual_overseer) =
polkadot_node_subsystem_test_helpers::make_subsystem_context(pool);
let network_stream = network.event_stream();
@@ -297,6 +313,7 @@ fn test_harness<T: Future<Output = VirtualOverseer>>(
metrics: Metrics(None),
sync_oracle,
shared: Shared::default(),
peerset_protocol_names,
};
let network_bridge = run_network_in(bridge, context, network_stream)
@@ -656,7 +673,12 @@ fn peer_view_updates_sent_via_overseer() {
// bridge will inform about all connected peers.
{
assert_sends_validation_event_to_all(
NetworkBridgeEvent::PeerConnected(peer.clone(), ObservedRole::Full, 1, None),
NetworkBridgeEvent::PeerConnected(
peer.clone(),
ObservedRole::Full,
ValidationVersion::V1.into(),
None,
),
&mut virtual_overseer,
)
.await;
@@ -699,7 +721,12 @@ fn peer_messages_sent_via_overseer() {
// bridge will inform about all connected peers.
{
assert_sends_validation_event_to_all(
NetworkBridgeEvent::PeerConnected(peer.clone(), ObservedRole::Full, 1, None),
NetworkBridgeEvent::PeerConnected(
peer.clone(),
ObservedRole::Full,
ValidationVersion::V1.into(),
None,
),
&mut virtual_overseer,
)
.await;
@@ -769,7 +796,12 @@ fn peer_disconnect_from_just_one_peerset() {
// bridge will inform about all connected peers.
{
assert_sends_validation_event_to_all(
NetworkBridgeEvent::PeerConnected(peer.clone(), ObservedRole::Full, 1, None),
NetworkBridgeEvent::PeerConnected(
peer.clone(),
ObservedRole::Full,
ValidationVersion::V1.into(),
None,
),
&mut virtual_overseer,
)
.await;
@@ -783,7 +815,12 @@ fn peer_disconnect_from_just_one_peerset() {
{
assert_sends_collation_event_to_all(
NetworkBridgeEvent::PeerConnected(peer.clone(), ObservedRole::Full, 1, None),
NetworkBridgeEvent::PeerConnected(
peer.clone(),
ObservedRole::Full,
ValidationVersion::V1.into(),
None,
),
&mut virtual_overseer,
)
.await;
@@ -852,7 +889,12 @@ fn relays_collation_protocol_messages() {
// bridge will inform about all connected peers.
{
assert_sends_validation_event_to_all(
NetworkBridgeEvent::PeerConnected(peer_a.clone(), ObservedRole::Full, 1, None),
NetworkBridgeEvent::PeerConnected(
peer_a.clone(),
ObservedRole::Full,
ValidationVersion::V1.into(),
None,
),
&mut virtual_overseer,
)
.await;
@@ -866,7 +908,12 @@ fn relays_collation_protocol_messages() {
{
assert_sends_collation_event_to_all(
NetworkBridgeEvent::PeerConnected(peer_b.clone(), ObservedRole::Full, 1, None),
NetworkBridgeEvent::PeerConnected(
peer_b.clone(),
ObservedRole::Full,
ValidationVersion::V1.into(),
None,
),
&mut virtual_overseer,
)
.await;
@@ -945,7 +992,12 @@ fn different_views_on_different_peer_sets() {
// bridge will inform about all connected peers.
{
assert_sends_validation_event_to_all(
NetworkBridgeEvent::PeerConnected(peer.clone(), ObservedRole::Full, 1, None),
NetworkBridgeEvent::PeerConnected(
peer.clone(),
ObservedRole::Full,
ValidationVersion::V1.into(),
None,
),
&mut virtual_overseer,
)
.await;
@@ -959,7 +1011,12 @@ fn different_views_on_different_peer_sets() {
{
assert_sends_collation_event_to_all(
NetworkBridgeEvent::PeerConnected(peer.clone(), ObservedRole::Full, 1, None),
NetworkBridgeEvent::PeerConnected(
peer.clone(),
ObservedRole::Full,
ValidationVersion::V1.into(),
None,
),
&mut virtual_overseer,
)
.await;