Network sync refactoring (part 6) (#11940)

* Extract `NetworkKVProvider` trait in `sc-authority-discovery` and remove unnecessary dependency

* Extract `NetworkSyncForkRequest` trait in `sc-finality-grandpa`

* Relax requirements on `SyncOracle` trait, remove extra native methods from `NetworkService` that are already provided by trait impls

* Move `NetworkSigner` trait from `sc-authority-discovery` into `sc-network-common` and de-duplicate methods on `NetworkService`

* Move `NetworkKVProvider` trait from `sc-authority-discovery` into `sc-network-common` and de-duplicate methods on `NetworkService`

* Minimize `sc-authority-discovery` dependency on `sc-network`

* Move `NetworkSyncForkRequest` trait from `sc-finality-grandpa` to `sc-network-common` and de-duplicate methods in `NetworkService`

* Extract `NetworkStatusProvider` trait and de-duplicate methods on `NetworkService`

* Extract `NetworkPeers` trait and de-duplicate methods on `NetworkService`

* Extract `NetworkEventStream` trait and de-duplicate methods on `NetworkService`

* Move more methods from `NetworkService` into `NetworkPeers` trait

* Move `NetworkStateInfo` trait into `sc-network-common`

* Extract `NetworkNotification` trait and de-duplicate methods on `NetworkService`

* Extract `NetworkRequest` trait and de-duplicate methods on `NetworkService`

* Remove `NetworkService::local_peer_id()`, it is already provided by `NetworkStateInfo` impl

* Extract `NetworkTransaction` trait and de-duplicate methods on `NetworkService`

* Extract `NetworkBlock` trait and de-duplicate methods on `NetworkService`

* Remove dependencies on `NetworkService` from most of the methods of `sc-service`

* Address simple review comments

substrate/client/network/src/behaviour.rs

@@ -21,7 +21,7 @@ use crate::{
 	discovery::{DiscoveryBehaviour, DiscoveryConfig, DiscoveryOut},
 	peer_info,
 	protocol::{message::Roles, CustomMessageOutcome, NotificationsSink, Protocol},
-	request_responses, DhtEvent, ObservedRole,
+	request_responses,
 };
 
 use bytes::Bytes;
@@ -41,7 +41,11 @@ use log::debug;
 
 use sc_client_api::{BlockBackend, ProofProvider};
 use sc_consensus::import_queue::{IncomingBlock, Origin};
-use sc_network_common::{config::ProtocolId, request_responses::ProtocolConfig};
+use sc_network_common::{
+	config::ProtocolId,
+	protocol::event::{DhtEvent, ObservedRole},
+	request_responses::{IfDisconnected, ProtocolConfig, RequestFailure},
+};
 use sc_peerset::PeersetHandle;
 use sp_blockchain::{HeaderBackend, HeaderMetadata};
 use sp_consensus::BlockOrigin;
@@ -57,9 +61,7 @@ use std::{
 	time::Duration,
 };
 
-pub use crate::request_responses::{
-	IfDisconnected, InboundFailure, OutboundFailure, RequestFailure, RequestId, ResponseFailure,
-};
+pub use crate::request_responses::{InboundFailure, OutboundFailure, RequestId, ResponseFailure};
 
 /// General behaviour of the network. Combines all protocols together.
 #[derive(NetworkBehaviour)]

substrate/client/network/src/lib.rs

@@ -262,22 +262,26 @@ pub mod transactions;
 
 #[doc(inline)]
 pub use libp2p::{multiaddr, Multiaddr, PeerId};
-pub use protocol::{
-	event::{DhtEvent, Event, ObservedRole},
-	PeerInfo,
-};
-pub use sc_network_common::sync::{
-	warp::{WarpSyncPhase, WarpSyncProgress},
-	StateDownloadProgress, SyncState,
+pub use protocol::PeerInfo;
+pub use sc_network_common::{
+	protocol::event::{DhtEvent, Event, ObservedRole},
+	request_responses::{IfDisconnected, RequestFailure},
+	service::{
+		KademliaKey, NetworkBlock, NetworkDHTProvider, NetworkRequest, NetworkSigner,
+		NetworkStateInfo, NetworkStatus, NetworkStatusProvider, NetworkSyncForkRequest,
+		NetworkTransaction, Signature, SigningError,
+	},
+	sync::{
+		warp::{WarpSyncPhase, WarpSyncProgress},
+		StateDownloadProgress, SyncState,
+	},
 };
 pub use service::{
-	DecodingError, IfDisconnected, KademliaKey, Keypair, NetworkService, NetworkWorker,
-	NotificationSender, NotificationSenderReady, OutboundFailure, PublicKey, RequestFailure,
-	Signature, SigningError,
+	DecodingError, Keypair, NetworkService, NetworkWorker, NotificationSender,
+	NotificationSenderReady, OutboundFailure, PublicKey,
 };
 
 pub use sc_peerset::ReputationChange;
-use sp_runtime::traits::{Block as BlockT, NumberFor};
 
 /// The maximum allowed number of established connections per peer.
 ///
@@ -296,35 +300,3 @@ pub trait ExHashT: std::hash::Hash + Eq + std::fmt::Debug + Clone + Send + Sync
 
 impl<T> ExHashT for T where T: std::hash::Hash + Eq + std::fmt::Debug + Clone + Send + Sync + 'static
 {}
-
-/// Trait for providing information about the local network state
-pub trait NetworkStateInfo {
-	/// Returns the local external addresses.
-	fn external_addresses(&self) -> Vec<Multiaddr>;
-
-	/// Returns the local Peer ID.
-	fn local_peer_id(&self) -> PeerId;
-}
-
-/// Overview status of the network.
-#[derive(Clone)]
-pub struct NetworkStatus<B: BlockT> {
-	/// Current global sync state.
-	pub sync_state: SyncState,
-	/// Target sync block number.
-	pub best_seen_block: Option<NumberFor<B>>,
-	/// Number of peers participating in syncing.
-	pub num_sync_peers: u32,
-	/// Total number of connected peers
-	pub num_connected_peers: usize,
-	/// Total number of active peers.
-	pub num_active_peers: usize,
-	/// The total number of bytes received.
-	pub total_bytes_inbound: u64,
-	/// The total number of bytes sent.
-	pub total_bytes_outbound: u64,
-	/// State sync in progress.
-	pub state_sync: Option<StateDownloadProgress>,
-	/// Warp sync in progress.
-	pub warp_sync: Option<WarpSyncProgress<B>>,
-}

substrate/client/network/src/protocol.rs

@@ -18,7 +18,6 @@
 
 use crate::{
 	config, error,
-	request_responses::RequestFailure,
 	utils::{interval, LruHashSet},
 };
 
@@ -45,6 +44,7 @@ use sc_client_api::{BlockBackend, HeaderBackend, ProofProvider};
 use sc_consensus::import_queue::{BlockImportError, BlockImportStatus, IncomingBlock, Origin};
 use sc_network_common::{
 	config::ProtocolId,
+	request_responses::RequestFailure,
 	sync::{
 		message::{
 			BlockAnnounce, BlockAttributes, BlockData, BlockRequest, BlockResponse, BlockState,
@@ -76,7 +76,6 @@ use std::{
 
 mod notifications;
 
-pub mod event;
 pub mod message;
 
 pub use notifications::{NotificationsSink, NotifsHandlerError, Ready};

substrate/client/network/src/protocol/event.rs

@@ -1,122 +0,0 @@
-// This file is part of Substrate.
-
-// Copyright (C) 2019-2022 Parity Technologies (UK) Ltd.
-// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
-
-// This program is free software: you can redistribute it and/or modify
-// it under the terms of the GNU General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-
-// This program is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-// GNU General Public License for more details.
-
-// You should have received a copy of the GNU General Public License
-// along with this program. If not, see <https://www.gnu.org/licenses/>.
-
-//! Network event types. These are are not the part of the protocol, but rather
-//! events that happen on the network like DHT get/put results received.
-
-use bytes::Bytes;
-use libp2p::{core::PeerId, kad::record::Key};
-use std::borrow::Cow;
-
-/// Events generated by DHT as a response to get_value and put_value requests.
-#[derive(Debug, Clone)]
-#[must_use]
-pub enum DhtEvent {
-	/// The value was found.
-	ValueFound(Vec<(Key, Vec<u8>)>),
-
-	/// The requested record has not been found in the DHT.
-	ValueNotFound(Key),
-
-	/// The record has been successfully inserted into the DHT.
-	ValuePut(Key),
-
-	/// An error has occurred while putting a record into the DHT.
-	ValuePutFailed(Key),
-}
-
-/// Type for events generated by networking layer.
-#[derive(Debug, Clone)]
-#[must_use]
-pub enum Event {
-	/// Event generated by a DHT.
-	Dht(DhtEvent),
-
-	/// Now connected to a new peer for syncing purposes.
-	SyncConnected {
-		/// Node we are now syncing from.
-		remote: PeerId,
-	},
-
-	/// Now disconnected from a peer for syncing purposes.
-	SyncDisconnected {
-		/// Node we are no longer syncing from.
-		remote: PeerId,
-	},
-
-	/// Opened a substream with the given node with the given notifications protocol.
-	///
-	/// The protocol is always one of the notification protocols that have been registered.
-	NotificationStreamOpened {
-		/// Node we opened the substream with.
-		remote: PeerId,
-		/// The concerned protocol. Each protocol uses a different substream.
-		/// This is always equal to the value of
-		/// [`crate::config::NonDefaultSetConfig::notifications_protocol`] of one of the
-		/// configured sets.
-		protocol: Cow<'static, str>,
-		/// If the negotiation didn't use the main name of the protocol (the one in
-		/// `notifications_protocol`), then this field contains which name has actually been
-		/// used.
-		/// Always contains a value equal to the value in
-		/// [`crate::config::NonDefaultSetConfig::fallback_names`].
-		negotiated_fallback: Option<Cow<'static, str>>,
-		/// Role of the remote.
-		role: ObservedRole,
-	},
-
-	/// Closed a substream with the given node. Always matches a corresponding previous
-	/// `NotificationStreamOpened` message.
-	NotificationStreamClosed {
-		/// Node we closed the substream with.
-		remote: PeerId,
-		/// The concerned protocol. Each protocol uses a different substream.
-		protocol: Cow<'static, str>,
-	},
-
-	/// Received one or more messages from the given node using the given protocol.
-	NotificationsReceived {
-		/// Node we received the message from.
-		remote: PeerId,
-		/// Concerned protocol and associated message.
-		messages: Vec<(Cow<'static, str>, Bytes)>,
-	},
-}
-
-/// Role that the peer sent to us during the handshake, with the addition of what our local node
-/// knows about that peer.
-///
-/// > **Note**: This enum is different from the `Role` enum. The `Role` enum indicates what a
-/// >			node says about itself, while `ObservedRole` is a `Role` merged with the
-/// >			information known locally about that node.
-#[derive(Debug, Clone)]
-pub enum ObservedRole {
-	/// Full node.
-	Full,
-	/// Light node.
-	Light,
-	/// Third-party authority.
-	Authority,
-}
-
-impl ObservedRole {
-	/// Returns `true` for `ObservedRole::Light`.
-	pub fn is_light(&self) -> bool {
-		matches!(self, Self::Light)
-	}
-}

substrate/client/network/src/request_responses.rs

@@ -50,7 +50,9 @@ use libp2p::{
 		NetworkBehaviourAction, PollParameters,
 	},
 };
-use sc_network_common::request_responses::{IncomingRequest, OutgoingResponse, ProtocolConfig};
+use sc_network_common::request_responses::{
+	IfDisconnected, IncomingRequest, OutgoingResponse, ProtocolConfig, RequestFailure,
+};
 use std::{
 	borrow::Cow,
 	collections::{hash_map::Entry, HashMap},
@@ -118,26 +120,6 @@ impl From<(Cow<'static, str>, RequestId)> for ProtocolRequestId {
 	}
 }
 
-/// When sending a request, what to do on a disconnected recipient.
-#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)]
-pub enum IfDisconnected {
-	/// Try to connect to the peer.
-	TryConnect,
-	/// Just fail if the destination is not yet connected.
-	ImmediateError,
-}
-
-/// Convenience functions for `IfDisconnected`.
-impl IfDisconnected {
-	/// Shall we connect to a disconnected peer?
-	pub fn should_connect(self) -> bool {
-		match self {
-			Self::TryConnect => true,
-			Self::ImmediateError => false,
-		}
-	}
-}
-
 /// Implementation of `NetworkBehaviour` that provides support for request-response protocols.
 pub struct RequestResponsesBehaviour {
 	/// The multiple sub-protocols, by name.
@@ -787,23 +769,6 @@ pub enum RegisterError {
 	DuplicateProtocol(Cow<'static, str>),
 }
 
-/// Error in a request.
-#[derive(Debug, thiserror::Error)]
-#[allow(missing_docs)]
-pub enum RequestFailure {
-	#[error("We are not currently connected to the requested peer.")]
-	NotConnected,
-	#[error("Given protocol hasn't been registered.")]
-	UnknownProtocol,
-	#[error("Remote has closed the substream before answering, thereby signaling that it considers the request as valid, but refused to answer it.")]
-	Refused,
-	#[error("The remote replied, but the local node is no longer interested in the response.")]
-	Obsolete,
-	/// Problem on the network.
-	#[error("Problem on the network: {0}")]
-	Network(OutboundFailure),
-}
-
 /// Error when processing a request sent by a remote.
 #[derive(Debug, thiserror::Error)]
 pub enum ResponseFailure {

substrate/client/network/src/service/out_events.rs

@@ -31,11 +31,10 @@
 //! - Send events by calling [`OutChannels::send`]. Events are cloned for each sender in the
 //! collection.
 
-use crate::Event;
-
 use futures::{channel::mpsc, prelude::*, ready, stream::FusedStream};
 use parking_lot::Mutex;
 use prometheus_endpoint::{register, CounterVec, GaugeVec, Opts, PrometheusError, Registry, U64};
+use sc_network_common::protocol::event::Event;
 use std::{
 	cell::RefCell,
 	fmt,

substrate/client/network/src/service/signature.rs

@@ -1,50 +0,0 @@
-// This file is part of Substrate.
-//
-// Copyright (C) 2017-2022 Parity Technologies (UK) Ltd.
-// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
-//
-// This program is free software: you can redistribute it and/or modify
-// it under the terms of the GNU General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// This program is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-// GNU General Public License for more details.
-//
-// You should have received a copy of the GNU General Public License
-// along with this program. If not, see <https://www.gnu.org/licenses/>.
-//
-// If you read this, you are very thorough, congratulations.
-
-use super::*;
-
-/// A result of signing a message with a network identity. Since `PeerId` is potentially a hash of a
-/// `PublicKey`, you need to reveal the `PublicKey` next to the signature, so the verifier can check
-/// if the signature was made by the entity that controls a given `PeerId`.
-pub struct Signature {
-	/// The public key derived from the network identity that signed the message.
-	pub public_key: PublicKey,
-	/// The actual signature made for the message signed.
-	pub bytes: Vec<u8>,
-}
-
-impl Signature {
-	/// Create a signature for a message with a given network identity.
-	pub fn sign_message(
-		message: impl AsRef<[u8]>,
-		keypair: &Keypair,
-	) -> Result<Self, SigningError> {
-		let public_key = keypair.public();
-		let bytes = keypair.sign(message.as_ref())?;
-		Ok(Self { public_key, bytes })
-	}
-
-	/// Verify whether the signature was made for the given message by the entity that controls the
-	/// given `PeerId`.
-	pub fn verify(&self, message: impl AsRef<[u8]>, peer_id: &PeerId) -> bool {
-		*peer_id == self.public_key.to_peer_id() &&
-			self.public_key.verify(message.as_ref(), &self.bytes)
-	}
-}

substrate/client/network/src/service/tests.rs

@@ -16,11 +16,15 @@
 // You should have received a copy of the GNU General Public License
 // along with this program. If not, see <https://www.gnu.org/licenses/>.
 
-use crate::{config, Event, NetworkService, NetworkWorker};
+use crate::{config, NetworkService, NetworkWorker};
 
 use futures::prelude::*;
 use libp2p::PeerId;
-use sc_network_common::config::ProtocolId;
+use sc_network_common::{
+	config::ProtocolId,
+	protocol::event::Event,
+	service::{NetworkEventStream, NetworkNotification, NetworkPeers, NetworkStateInfo},
+};
 use sc_network_light::light_client_requests::handler::LightClientRequestHandler;
 use sc_network_sync::{
 	block_request_handler::BlockRequestHandler, state_request_handler::StateRequestHandler,
@@ -192,7 +196,7 @@ fn build_nodes_one_proto() -> (
 			set_config: config::SetConfig {
 				reserved_nodes: vec![config::MultiaddrWithPeerId {
 					multiaddr: listen_addr,
-					peer_id: node1.local_peer_id().clone(),
+					peer_id: node1.local_peer_id(),
 				}],
 				..Default::default()
 			},
@@ -214,18 +218,10 @@ fn notifications_state_consistent() {
 
 	// Write some initial notifications that shouldn't get through.
 	for _ in 0..(rand::random::<u8>() % 5) {
-		node1.write_notification(
-			node2.local_peer_id().clone(),
-			PROTOCOL_NAME,
-			b"hello world".to_vec(),
-		);
+		node1.write_notification(node2.local_peer_id(), PROTOCOL_NAME, b"hello world".to_vec());
 	}
 	for _ in 0..(rand::random::<u8>() % 5) {
-		node2.write_notification(
-			node1.local_peer_id().clone(),
-			PROTOCOL_NAME,
-			b"hello world".to_vec(),
-		);
+		node2.write_notification(node1.local_peer_id(), PROTOCOL_NAME, b"hello world".to_vec());
 	}
 
 	async_std::task::block_on(async move {
@@ -249,14 +245,14 @@ fn notifications_state_consistent() {
 			// test consists in ensuring that notifications get ignored if the stream isn't open.
 			if rand::random::<u8>() % 5 >= 3 {
 				node1.write_notification(
-					node2.local_peer_id().clone(),
+					node2.local_peer_id(),
 					PROTOCOL_NAME,
 					b"hello world".to_vec(),
 				);
 			}
 			if rand::random::<u8>() % 5 >= 3 {
 				node2.write_notification(
-					node1.local_peer_id().clone(),
+					node1.local_peer_id(),
 					PROTOCOL_NAME,
 					b"hello world".to_vec(),
 				);
@@ -264,10 +260,10 @@ fn notifications_state_consistent() {
 
 			// Also randomly disconnect the two nodes from time to time.
 			if rand::random::<u8>() % 20 == 0 {
-				node1.disconnect_peer(node2.local_peer_id().clone(), PROTOCOL_NAME);
+				node1.disconnect_peer(node2.local_peer_id(), PROTOCOL_NAME);
 			}
 			if rand::random::<u8>() % 20 == 0 {
-				node2.disconnect_peer(node1.local_peer_id().clone(), PROTOCOL_NAME);
+				node2.disconnect_peer(node1.local_peer_id(), PROTOCOL_NAME);
 			}
 
 			// Grab next event from either `events_stream1` or `events_stream2`.
@@ -295,7 +291,7 @@ fn notifications_state_consistent() {
 						something_happened = true;
 						assert!(!node1_to_node2_open);
 						node1_to_node2_open = true;
-						assert_eq!(remote, *node2.local_peer_id());
+						assert_eq!(remote, node2.local_peer_id());
 					},
 				future::Either::Right(Event::NotificationStreamOpened {
 					remote, protocol, ..
@@ -304,7 +300,7 @@ fn notifications_state_consistent() {
 						something_happened = true;
 						assert!(!node2_to_node1_open);
 						node2_to_node1_open = true;
-						assert_eq!(remote, *node1.local_peer_id());
+						assert_eq!(remote, node1.local_peer_id());
 					},
 				future::Either::Left(Event::NotificationStreamClosed {
 					remote, protocol, ..
@@ -312,7 +308,7 @@ fn notifications_state_consistent() {
 					if protocol == PROTOCOL_NAME {
 						assert!(node1_to_node2_open);
 						node1_to_node2_open = false;
-						assert_eq!(remote, *node2.local_peer_id());
+						assert_eq!(remote, node2.local_peer_id());
 					},
 				future::Either::Right(Event::NotificationStreamClosed {
 					remote, protocol, ..
@@ -320,14 +316,14 @@ fn notifications_state_consistent() {
 					if protocol == PROTOCOL_NAME {
 						assert!(node2_to_node1_open);
 						node2_to_node1_open = false;
-						assert_eq!(remote, *node1.local_peer_id());
+						assert_eq!(remote, node1.local_peer_id());
 					},
 				future::Either::Left(Event::NotificationsReceived { remote, .. }) => {
 					assert!(node1_to_node2_open);
-					assert_eq!(remote, *node2.local_peer_id());
+					assert_eq!(remote, node2.local_peer_id());
 					if rand::random::<u8>() % 5 >= 4 {
 						node1.write_notification(
-							node2.local_peer_id().clone(),
+							node2.local_peer_id(),
 							PROTOCOL_NAME,
 							b"hello world".to_vec(),
 						);
@@ -335,10 +331,10 @@ fn notifications_state_consistent() {
 				},
 				future::Either::Right(Event::NotificationsReceived { remote, .. }) => {
 					assert!(node2_to_node1_open);
-					assert_eq!(remote, *node1.local_peer_id());
+					assert_eq!(remote, node1.local_peer_id());
 					if rand::random::<u8>() % 5 >= 4 {
 						node2.write_notification(
-							node1.local_peer_id().clone(),
+							node1.local_peer_id(),
 							PROTOCOL_NAME,
 							b"hello world".to_vec(),
 						);
@@ -373,7 +369,7 @@ fn lots_of_incoming_peers_works() {
 		..config::NetworkConfiguration::new_local()
 	});
 
-	let main_node_peer_id = *main_node.local_peer_id();
+	let main_node_peer_id = main_node.local_peer_id();
 
 	// We spawn background tasks and push them in this `Vec`. They will all be waited upon before
 	// this test ends.
@@ -476,8 +472,13 @@ fn notifications_back_pressure() {
 
 		// Sending!
 		for num in 0..TOTAL_NOTIFS {
-			let notif = node1.notification_sender(node2_id.clone(), PROTOCOL_NAME).unwrap();
-			notif.ready().await.unwrap().send(format!("hello #{}", num)).unwrap();
+			let notif = node1.notification_sender(node2_id, PROTOCOL_NAME).unwrap();
+			notif
+				.ready()
+				.await
+				.unwrap()
+				.send(format!("hello #{}", num).into_bytes())
+				.unwrap();
 		}
 
 		receiver.await;
@@ -514,7 +515,7 @@ fn fallback_name_working() {
 			set_config: config::SetConfig {
 				reserved_nodes: vec![config::MultiaddrWithPeerId {
 					multiaddr: listen_addr,
-					peer_id: node1.local_peer_id().clone(),
+					peer_id: node1.local_peer_id(),
 				}],
 				..Default::default()
 			},

substrate/client/network/src/transactions.rs

@@ -32,7 +32,7 @@ use crate::{
 	protocol::message,
 	service::NetworkService,
 	utils::{interval, LruHashSet},
-	Event, ExHashT, ObservedRole,
+	ExHashT,
 };
 
 use codec::{Decode, Encode};
@@ -40,7 +40,11 @@ use futures::{channel::mpsc, prelude::*, stream::FuturesUnordered};
 use libp2p::{multiaddr, PeerId};
 use log::{debug, trace, warn};
 use prometheus_endpoint::{register, Counter, PrometheusError, Registry, U64};
-use sc_network_common::config::ProtocolId;
+use sc_network_common::{
+	config::ProtocolId,
+	protocol::event::{Event, ObservedRole},
+	service::{NetworkEventStream, NetworkNotification, NetworkPeers},
+};
 use sp_runtime::traits::Block as BlockT;
 use std::{
 	borrow::Cow,
@@ -176,7 +180,7 @@ impl TransactionsHandlerPrototype {
 		transaction_pool: Arc<dyn TransactionPool<H, B>>,
 		metrics_registry: Option<&Registry>,
 	) -> error::Result<(TransactionsHandler<B, H>, TransactionsHandlerController<H>)> {
-		let event_stream = service.event_stream("transactions-handler").boxed();
+		let event_stream = service.event_stream("transactions-handler");
 		let (to_handler, from_controller) = mpsc::unbounded();
 		let gossip_enabled = Arc::new(AtomicBool::new(false));