Implement SyncEventStream

Remove `SyncConnected`/`SyncDisconnected` events from
`NetworkEvenStream` and provide those events through
`ChainSyncInterface` instead.

Modify BEEFY/GRANDPA/transactions protocol and `NetworkGossip` to take
`SyncEventStream` object which they listen to for incoming sync peer
events.

bin/node-template/node/src/service.rs

@@ -240,7 +240,7 @@ pub fn new_full(mut config: Configuration) -> Result<TaskManager, ServiceError>
 		backend,
 		system_rpc_tx,
 		tx_handler_controller,
-		sync_service,
+		sync_service: sync_service.clone(),
 		config,
 		telemetry: telemetry.as_mut(),
 	})?;
@@ -321,6 +321,8 @@ pub fn new_full(mut config: Configuration) -> Result<TaskManager, ServiceError>
 			config: grandpa_config,
 			link: grandpa_link,
 			network,
+			// TODo(aaro): fix arc
+			sync: Arc::new(sync_service),
 			voting_rule: sc_finality_grandpa::VotingRulesBuilder::default().build(),
 			prometheus_registry,
 			shared_voter_state: SharedVoterState::empty(),

bin/node/cli/src/service.rs

@@ -29,7 +29,7 @@ use node_primitives::Block;
 use sc_client_api::BlockBackend;
 use sc_consensus_babe::{self, SlotProportion};
 use sc_executor::NativeElseWasmExecutor;
-use sc_network::NetworkService;
+use sc_network::{ChainSyncInterface, NetworkService};
 use sc_network_common::{protocol::event::Event, service::NetworkEventStream};
 use sc_service::{config::Configuration, error::Error as ServiceError, RpcHandlers, TaskManager};
 use sc_telemetry::{Telemetry, TelemetryWorker};
@@ -393,7 +393,7 @@ pub fn new_full_base(
 		task_manager: &mut task_manager,
 		system_rpc_tx,
 		tx_handler_controller,
-		sync_service,
+		sync_service: sync_service.clone(),
 		telemetry: telemetry.as_mut(),
 	})?;
 
@@ -533,6 +533,7 @@ pub fn new_full_base(
 			config,
 			link: grandpa_link,
 			network: network.clone(),
+			sync: Arc::new(sync_service),
 			telemetry: telemetry.as_ref().map(|x| x.handle()),
 			voting_rule: grandpa::VotingRulesBuilder::default().build(),
 			prometheus_registry,

client/beefy/src/lib.rs

@@ -42,7 +42,7 @@ use prometheus::Registry;
 use sc_client_api::{Backend, BlockBackend, BlockchainEvents, FinalityNotifications, Finalizer};
 use sc_consensus::BlockImport;
 use sc_network::ProtocolName;
-use sc_network_common::service::NetworkRequest;
+use sc_network_common::{service::NetworkRequest, sync::SyncEventStream};
 use sc_network_gossip::{GossipEngine, Network as GossipNetwork};
 use sp_api::{HeaderT, NumberFor, ProvideRuntimeApi};
 use sp_blockchain::{
@@ -168,6 +168,8 @@ where
 pub struct BeefyNetworkParams<B: Block, N> {
 	/// Network implementing gossip, requests and sync-oracle.
 	pub network: Arc<N>,
+	/// Syncing service implementing event stream for peers.
+	pub sync: Arc<dyn SyncEventStream>,
 	/// Chain specific BEEFY gossip protocol name. See
 	/// [`communication::beefy_protocol_name::gossip_protocol_name`].
 	pub gossip_protocol_name: ProtocolName,
@@ -228,14 +230,20 @@ where
 		on_demand_justifications_handler,
 	} = beefy_params;
 
-	let BeefyNetworkParams { network, gossip_protocol_name, justifications_protocol_name, .. } =
-		network_params;
+	let BeefyNetworkParams {
+		network,
+		sync,
+		gossip_protocol_name,
+		justifications_protocol_name,
+		..
+	} = network_params;
 
 	let known_peers = Arc::new(Mutex::new(KnownPeers::new()));
 	let gossip_validator =
 		Arc::new(communication::gossip::GossipValidator::new(known_peers.clone()));
 	let mut gossip_engine = sc_network_gossip::GossipEngine::new(
 		network.clone(),
+		sync.clone(),
 		gossip_protocol_name,
 		gossip_validator.clone(),
 		None,
@@ -285,6 +293,7 @@ where
 		backend,
 		payload_provider,
 		network,
+		sync,
 		key_store: key_store.into(),
 		known_peers,
 		gossip_engine,

client/beefy/src/tests.rs

@@ -139,6 +139,10 @@ impl BeefyTestNet {
 		});
 	}
 
+	pub(crate) fn block_until_sync_connected(&mut self) {
+		todo!();
+	}
+
 	pub(crate) fn generate_blocks_and_sync(
 		&mut self,
 		count: usize,
@@ -342,6 +346,7 @@ where
 
 		let network_params = crate::BeefyNetworkParams {
 			network: peer.network_service().clone(),
+			sync: peer.sync_service().clone(),
 			gossip_protocol_name: beefy_gossip_proto_name(),
 			justifications_protocol_name: on_demand_justif_handler.protocol_name(),
 			_phantom: PhantomData,
@@ -954,6 +959,7 @@ fn test_voter_init_setup(
 		Arc::new(crate::communication::gossip::GossipValidator::new(known_peers));
 	let mut gossip_engine = sc_network_gossip::GossipEngine::new(
 		net.peer(0).network_service().clone(),
+		net.peer(0).sync_service().clone(),
 		"/beefy/whatever",
 		gossip_validator,
 		None,

client/beefy/src/worker.rs

@@ -32,6 +32,7 @@ use sc_client_api::{Backend, FinalityNotification, FinalityNotifications, Header
 use sc_network_common::{
 	protocol::event::Event as NetEvent,
 	service::{NetworkEventStream, NetworkRequest},
+	sync::{SyncEvent, SyncEventStream},
 };
 use sc_network_gossip::GossipEngine;
 
@@ -252,6 +253,7 @@ pub(crate) struct WorkerParams<B: Block, BE, P, R, N> {
 	pub backend: Arc<BE>,
 	pub payload_provider: P,
 	pub network: N,
+	pub sync: Arc<dyn SyncEventStream>,
 	pub key_store: BeefyKeystore,
 	pub known_peers: Arc<Mutex<KnownPeers<B>>>,
 	pub gossip_engine: GossipEngine<B>,
@@ -302,6 +304,7 @@ pub(crate) struct BeefyWorker<B: Block, BE, P, R, N> {
 	backend: Arc<BE>,
 	payload_provider: P,
 	network: N,
+	sync: Arc<dyn SyncEventStream>,
 	key_store: BeefyKeystore,
 
 	// communication
@@ -346,6 +349,7 @@ where
 			payload_provider,
 			key_store,
 			network,
+			sync,
 			gossip_engine,
 			gossip_validator,
 			on_demand_justifications,
@@ -359,6 +363,7 @@ where
 			backend,
 			payload_provider,
 			network,
+			sync,
 			known_peers,
 			key_store,
 			gossip_engine,
@@ -794,7 +799,7 @@ where
 	) {
 		info!(target: "beefy", "🥩 run BEEFY worker, best grandpa: #{:?}.", self.best_grandpa_block());
 
-		let mut network_events = self.network.event_stream("network-gossip").fuse();
+		let mut sync_events = self.sync.event_stream("network-gossipzzz").fuse();
 		let mut votes = Box::pin(
 			self.gossip_engine
 				.messages_for(topic::<B>())
@@ -838,11 +843,11 @@ where
 					return;
 				},
 				// Keep track of connected peers.
-				net_event = network_events.next() => {
-					if let Some(net_event) = net_event {
-						self.handle_network_event(net_event);
+				sync_event = sync_events.next() => {
+					if let Some(sync_event) = sync_event {
+						self.handle_sync_event(sync_event);
 					} else {
-						error!(target: "beefy", "🥩 Network events stream terminated, closing worker.");
+						error!(target: "beefy", "🥩 Syncing events stream terminated, closing worker.");
 						return;
 					}
 				},
@@ -897,16 +902,14 @@ where
 	}
 
 	/// Update known peers based on network events.
-	fn handle_network_event(&mut self, event: NetEvent) {
+	fn handle_sync_event(&mut self, event: SyncEvent) {
 		match event {
-			NetEvent::SyncConnected { remote } => {
+			SyncEvent::PeerConnected(remote) => {
 				self.known_peers.lock().add_new(remote);
 			},
-			NetEvent::SyncDisconnected { remote } => {
+			SyncEvent::PeerDisconnected(remote) => {
 				self.known_peers.lock().remove(&remote);
 			},
-			// We don't care about other events.
-			_ => (),
 		}
 	}
 }
@@ -982,7 +985,8 @@ pub(crate) mod tests {
 	use beefy_primitives::{known_payloads, mmr::MmrRootProvider};
 	use futures::{executor::block_on, future::poll_fn, task::Poll};
 	use sc_client_api::{Backend as BackendT, HeaderBackend};
-	use sc_network::NetworkService;
+	use sc_network::{ChainSyncInterface, NetworkService};
+	use sc_network_common::sync::SyncEventStream;
 	use sc_network_test::TestNetFactory;
 	use sp_api::HeaderT;
 	use sp_blockchain::Backend as BlockchainBackendT;
@@ -1050,10 +1054,16 @@ pub(crate) mod tests {
 		let backend = peer.client().as_backend();
 		let api = Arc::new(TestApi {});
 		let network = peer.network_service().clone();
+		let sync = peer.sync_service().clone();
 		let known_peers = Arc::new(Mutex::new(KnownPeers::new()));
 		let gossip_validator = Arc::new(GossipValidator::new(known_peers.clone()));
-		let gossip_engine =
-			GossipEngine::new(network.clone(), "/beefy/1", gossip_validator.clone(), None);
+		let gossip_engine = GossipEngine::new(
+			network.clone(),
+			sync.clone(),
+			"/beefy/1",
+			gossip_validator.clone(),
+			None,
+		);
 		let on_demand_justifications = OnDemandJustificationsEngine::new(
 			network.clone(),
 			api.clone(),
@@ -1080,6 +1090,7 @@ pub(crate) mod tests {
 			gossip_validator,
 			metrics: None,
 			network,
+			sync,
 			on_demand_justifications,
 			persisted_state,
 		};

client/finality-grandpa/src/communication/mod.rs

@@ -59,7 +59,10 @@ use crate::{
 use gossip::{
 	FullCatchUpMessage, FullCommitMessage, GossipMessage, GossipValidator, PeerReport, VoteMessage,
 };
-use sc_network_common::service::{NetworkBlock, NetworkSyncForkRequest};
+use sc_network_common::{
+	service::{NetworkBlock, NetworkSyncForkRequest},
+	sync::SyncEventStream,
+};
 use sc_utils::mpsc::TracingUnboundedReceiver;
 use sp_finality_grandpa::{AuthorityId, AuthoritySignature, RoundNumber, SetId as SetIdNumber};
 
@@ -234,6 +237,7 @@ impl<B: BlockT, N: Network<B>> NetworkBridge<B, N> {
 	/// service taken from the VoterSetState.
 	pub(crate) fn new(
 		service: N,
+		sync: Arc<dyn SyncEventStream>,
 		config: crate::Config,
 		set_state: crate::environment::SharedVoterSetState<B>,
 		prometheus_registry: Option<&Registry>,
@@ -246,6 +250,7 @@ impl<B: BlockT, N: Network<B>> NetworkBridge<B, N> {
 		let validator = Arc::new(validator);
 		let gossip_engine = Arc::new(Mutex::new(GossipEngine::new(
 			service.clone(),
+			sync.clone(),
 			protocol,
 			validator.clone(),
 			prometheus_registry,

client/finality-grandpa/src/communication/tests.rs

@@ -33,6 +33,7 @@ use sc_network_common::{
 		NetworkBlock, NetworkEventStream, NetworkNotification, NetworkPeers,
 		NetworkSyncForkRequest, NotificationSender, NotificationSenderError,
 	},
+	sync::{SyncEvent as SyncStreamEvent, SyncEventStream},
 };
 use sc_network_gossip::Validator;
 use sc_network_test::{Block, Hash};
@@ -190,6 +191,24 @@ impl sc_network_gossip::ValidatorContext<Block> for TestNetwork {
 	fn send_topic(&mut self, _: &PeerId, _: Hash, _: bool) {}
 }
 
+pub(crate) enum SyncEvent {
+	EventStream(TracingUnboundedSender<SyncStreamEvent>),
+}
+
+#[derive(Clone)]
+pub(crate) struct TestSync {
+	sender: TracingUnboundedSender<SyncEvent>,
+}
+
+impl SyncEventStream for TestSync {
+	fn event_stream(
+		&self,
+		_name: &'static str,
+	) -> Pin<Box<dyn Stream<Item = SyncStreamEvent> + Send>> {
+		Box::pin(futures::stream::pending())
+	}
+}
+
 pub(crate) struct Tester {
 	pub(crate) net_handle: super::NetworkBridge<Block, TestNetwork>,
 	gossip_validator: Arc<GossipValidator<Block>>,
@@ -259,6 +278,8 @@ fn voter_set_state() -> SharedVoterSetState<Block> {
 pub(crate) fn make_test_network() -> (impl Future<Output = Tester>, TestNetwork) {
 	let (tx, rx) = tracing_unbounded("test");
 	let net = TestNetwork { sender: tx };
+	let (stx, srx) = tracing_unbounded("sync");
+	let sync = Arc::new(TestSync { sender: stx });
 
 	#[derive(Clone)]
 	struct Exit;
@@ -271,7 +292,8 @@ pub(crate) fn make_test_network() -> (impl Future<Output = Tester>, TestNetwork)
 		}
 	}
 
-	let bridge = super::NetworkBridge::new(net.clone(), config(), voter_set_state(), None, None);
+	let bridge =
+		super::NetworkBridge::new(net.clone(), sync, config(), voter_set_state(), None, None);
 
 	(
 		futures::future::ready(Tester {

client/finality-grandpa/src/lib.rs

@@ -68,7 +68,7 @@ use sc_client_api::{
 	StorageProvider, TransactionFor,
 };
 use sc_consensus::BlockImport;
-use sc_network_common::protocol::ProtocolName;
+use sc_network_common::{protocol::ProtocolName, sync::SyncEventStream};
 use sc_telemetry::{telemetry, TelemetryHandle, CONSENSUS_DEBUG, CONSENSUS_INFO};
 use sc_utils::mpsc::{tracing_unbounded, TracingUnboundedReceiver};
 use sp_api::ProvideRuntimeApi;
@@ -673,6 +673,8 @@ pub struct GrandpaParams<Block: BlockT, C, N, SC, VR> {
 	/// `sc_network` crate, it is assumed that the Grandpa notifications protocol has been passed
 	/// to the configuration of the networking. See [`grandpa_peers_set_config`].
 	pub network: N,
+	/// Event stream for syncing-related events.
+	pub sync: Arc<dyn SyncEventStream>,
 	/// A voting rule used to potentially restrict target votes.
 	pub voting_rule: VR,
 	/// The prometheus metrics registry.
@@ -724,6 +726,7 @@ where
 		mut config,
 		link,
 		network,
+		sync,
 		voting_rule,
 		prometheus_registry,
 		shared_voter_state,
@@ -748,6 +751,7 @@ where
 
 	let network = NetworkBridge::new(
 		network,
+		sync,
 		config.clone(),
 		persistent_data.set_state.clone(),
 		prometheus_registry.as_ref(),

client/finality-grandpa/src/observer.rs

@@ -28,6 +28,7 @@ use futures::prelude::*;
 use log::{debug, info, warn};
 
 use sc_client_api::backend::Backend;
+use sc_network_common::sync::SyncEventStream;
 use sc_telemetry::TelemetryHandle;
 use sc_utils::mpsc::TracingUnboundedReceiver;
 use sp_blockchain::HeaderMetadata;
@@ -167,6 +168,7 @@ pub fn run_grandpa_observer<BE, Block: BlockT, Client, N, SC>(
 	config: Config,
 	link: LinkHalf<Block, Client, SC>,
 	network: N,
+	sync: Arc<dyn SyncEventStream>,
 ) -> sp_blockchain::Result<impl Future<Output = ()> + Send>
 where
 	BE: Backend<Block> + Unpin + 'static,
@@ -186,6 +188,7 @@ where
 
 	let network = NetworkBridge::new(
 		network,
+		sync,
 		config.clone(),
 		persistent_data.set_state.clone(),
 		None,