Skip to content

A0-1664 Stop exposing sync connected events from substrate network #788

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
timorl merged 8 commits into from
Dec 7, 2022
Merged

A0-1664 Stop exposing sync connected events from substrate network #788

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
698d03d
removing stuff
Dec 5, 2022
d0895b3
remove tests that do not work...
Dec 6, 2022
1ac10c8
Merge branch 'main' into A0-1664-stop-exposing!
lesniak43 Dec 6, 2022
09da9e7
polishing
Dec 6, 2022
a1de729
no ()
Dec 6, 2022
ad797ba
fmt
Dec 6, 2022
dac5bec
Merge branch 'main' into A0-1664-stop-exposing
lesniak43 Dec 6, 2022
c8607b0
Merge remote-tracking branch 'origin' into A0-1664-stop-exposing
maciejnems Dec 7, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
remove tests that do not work...
  • Loading branch information
Damian Leśniak committed Dec 6, 2022
commit d0895b3a19fb26bec2d7b0910232dce48d05af76
7 changes: 1 addition & 6 deletions finality-aleph/src/network/mock.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,4 @@
use std::{
collections::VecDeque,
fmt,
sync::Arc,
time::Duration,
};
use std::{collections::VecDeque, fmt, sync::Arc, time::Duration};

use aleph_primitives::KEY_TYPE;
use async_trait::async_trait;
Expand Down
195 changes: 4 additions & 191 deletions finality-aleph/src/network/service.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -326,7 +326,7 @@ where

#[cfg(test)]
mod tests {
use std::{collections::HashSet, iter, iter::FromIterator};
use std::iter;

use codec::Encode;
use futures::{channel::oneshot, StreamExt};
Expand All @@ -337,13 +337,12 @@ mod tests {
use crate::{
network::{
manager::{SessionHandler, VersionedAuthentication},
mock::{crypto_basics, MockData, MockEvent, MockIO, MockNetwork, MockSenderError},
mock::{crypto_basics, MockData, MockEvent, MockIO, MockNetwork},
testing::DiscoveryMessage,
NetworkIdentity, Protocol,
Protocol,
},
testing::mocks::validator_network::{
random_peer_id, MockMultiaddress,
MockNetwork as MockValidatorNetwork,
random_peer_id, MockMultiaddress, MockNetwork as MockValidatorNetwork,
},
SessionId,
};
Expand Down Expand Up @@ -425,98 +424,6 @@ mod tests {
))
}

#[tokio::test]
async fn test_notification_stream_opened() {
let mut test_data = TestData::prepare().await;

let peer_ids: Vec<_> = (0..3).map(|_| random_peer_id()).collect();

peer_ids.iter().for_each(|peer_id| {
test_data.network.emit_event(MockEvent::StreamOpened(
peer_id.clone(),
Protocol::Authentication,
));
});

let message = authentication(test_data.validator_network.identity().0).await;
test_data
.mock_io
.messages_for_network
.unbounded_send(message.clone())
.unwrap();

let broadcasted_messages = HashSet::<_>::from_iter(
test_data
.network
.send_message
.take(peer_ids.len())
.await
.into_iter(),
);

let expected_messages = HashSet::from_iter(
peer_ids
.into_iter()
.map(|peer_id| (message.clone().encode(), peer_id, Protocol::Authentication)),
);

assert_eq!(broadcasted_messages, expected_messages);

test_data.cleanup().await
}

#[tokio::test]
async fn test_notification_stream_closed() {
let mut test_data = TestData::prepare().await;

let peer_ids: Vec<_> = (0..3).map(|_| random_peer_id()).collect();
let opened_authorities_n = 2;

peer_ids.iter().for_each(|peer_id| {
test_data.network.emit_event(MockEvent::StreamOpened(
peer_id.clone(),
Protocol::Authentication,
));
});

peer_ids
.iter()
.skip(opened_authorities_n)
.for_each(|peer_id| {
test_data.network.emit_event(MockEvent::StreamClosed(
peer_id.clone(),
Protocol::Authentication,
));
});

let message = authentication(test_data.validator_network.identity().0).await;
test_data
.mock_io
.messages_for_network
.unbounded_send(message.clone())
.unwrap();

let broadcasted_messages = HashSet::<_>::from_iter(
test_data
.network
.send_message
.take(opened_authorities_n)
.await
.into_iter(),
);

let expected_messages = HashSet::from_iter(
peer_ids
.into_iter()
.take(opened_authorities_n)
.map(|peer_id| (message.clone().encode(), peer_id, Protocol::Authentication)),
);

assert_eq!(broadcasted_messages, expected_messages);

test_data.cleanup().await
}

#[tokio::test]
async fn test_send_validator_data() {
let mut test_data = TestData::prepare().await;
Expand Down Expand Up @@ -567,100 +474,6 @@ mod tests {
test_data.cleanup().await
}

#[tokio::test]
async fn test_create_sender_error() {
let mut test_data = TestData::prepare().await;

test_data
.network
.create_sender_errors
.lock()
.push_back(MockSenderError::SomeError);

let peer_id = random_peer_id();

let message_1 = authentication(vec![(random_peer_id(), String::from("other_1"))]).await;
let message_2 = authentication(vec![(random_peer_id(), String::from("other_2"))]).await;

test_data.network.emit_event(MockEvent::StreamOpened(
peer_id.clone(),
Protocol::Authentication,
));

test_data
.mock_io
.messages_for_network
.unbounded_send(message_1)
.unwrap();

test_data
.mock_io
.messages_for_network
.unbounded_send(message_2.clone())
.unwrap();

let expected = (message_2.encode(), peer_id, Protocol::Authentication);

assert_eq!(
test_data
.network
.send_message
.next()
.await
.expect("Should receive message"),
expected,
);

test_data.cleanup().await
}

#[tokio::test]
async fn test_send_error() {
let mut test_data = TestData::prepare().await;

test_data
.network
.send_errors
.lock()
.push_back(MockSenderError::SomeError);

let peer_id = random_peer_id();

let message_1 = authentication(vec![(random_peer_id(), String::from("other_1"))]).await;
let message_2 = authentication(vec![(random_peer_id(), String::from("other_2"))]).await;

test_data.network.emit_event(MockEvent::StreamOpened(
peer_id.clone(),
Protocol::Authentication,
));

test_data
.mock_io
.messages_for_network
.unbounded_send(message_1)
.unwrap();

test_data
.mock_io
.messages_for_network
.unbounded_send(message_2.clone())
.unwrap();

let expected = (message_2.encode(), peer_id, Protocol::Authentication);

assert_eq!(
test_data
.network
.send_message
.next()
.await
.expect("Should receive message"),
expected,
);

test_data.cleanup().await
}

#[tokio::test]
async fn test_notification_received() {
let mut test_data = TestData::prepare().await;
Expand Down
19 changes: 12 additions & 7 deletions finality-aleph/src/substrate_network.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,21 +136,26 @@ impl<B: Block, H: ExHashT> EventStream<PeerId> for NetworkEventStream<B, H> {
match self.stream.next().await {
Some(event) => match event {
SyncConnected { remote } => {
let multiaddress = iter::once(MultiaddressProtocol::P2p(remote.into())).collect();
let multiaddress =
iter::once(MultiaddressProtocol::P2p(remote.into())).collect();
trace!(target: "aleph-network", "Connected event from address {:?}", multiaddress);
if let Err(e) = self.network
.add_peers_to_reserved_set(protocol_name(&Protocol::Authentication), iter::once(multiaddress).collect())
{
if let Err(e) = self.network.add_peers_to_reserved_set(
protocol_name(&Protocol::Authentication),
iter::once(multiaddress).collect(),
) {
error!(target: "aleph-network", "add_reserved failed: {}", e);
}
continue;
},
}
SyncDisconnected { remote } => {
trace!(target: "aleph-network", "Disconnected event for peer {:?}", remote);
let addresses = iter::once(remote).collect();
self.network.remove_peers_from_reserved_set(protocol_name(&Protocol::Authentication), addresses);
self.network.remove_peers_from_reserved_set(
protocol_name(&Protocol::Authentication),
addresses,
);
continue;
},
}
NotificationStreamOpened {
remote, protocol, ..
} => match to_protocol(protocol.as_ref()) {
Expand Down