Implement bitswap in the network behaviour using libp2p_bitswap.

client/network-gossip/src/bridge.rs

@@ -210,7 +210,8 @@ impl<B: BlockT> Future for GossipEngine<B> {
 
 								this.forwarding_state = ForwardingState::Busy(to_forward.into());
 							},
-							Event::Dht(_) => {}
+							Event::Dht(_) => {},
+							Event::Bitswap(_) => {},
 						}
 						// The network event stream closed. Do the same for [`GossipValidator`].
 						Poll::Ready(None) => return Poll::Ready(()),

client/network/Cargo.toml

@@ -61,6 +61,8 @@ unsigned-varint = { version = "0.4.0", features = ["futures", "futures-codec"] }
 void = "1.0.2"
 wasm-timer = "0.2"
 zeroize = "1.0.0"
+libp2p-bitswap = "0.7.0"
+tiny-cid = "0.2.5"
 
 [dependencies.libp2p]
 version = "0.28.1"

client/network/src/behaviour.rs

@@ -54,6 +54,8 @@ pub struct Behaviour<B: BlockT, H: ExHashT> {
 	peer_info: peer_info::PeerInfoBehaviour,
 	/// Discovers nodes of the network.
 	discovery: DiscoveryBehaviour,
+	/// Exchanges blocks of data with other nodes.
+	pub(crate) bitswap: libp2p_bitswap::Bitswap,
 	/// Generic request-reponse protocols.
 	request_responses: request_responses::RequestResponsesBehaviour,
 	/// Block request handling.
@@ -172,6 +174,18 @@ pub enum BehaviourOut<B: BlockT> {
 	/// Events generated by a DHT as a response to get_value or put_value requests as well as the
 	/// request duration.
 	Dht(DhtEvent, Duration),
+
+	/// Event generated by bitswap.
+	Bitswap(BitswapEvent)
+}
+
+/// An event generated by bitswap.
+#[derive(Clone, Debug)]
+pub enum BitswapEvent {
+	/// A block was received.
+	ReceivedBlock(PeerId, tiny_cid::Cid, Box<[u8]>),
+	/// A WANT request was received.
+	ReceivedWant(PeerId, tiny_cid::Cid, i32),
 }
 
 impl<B: BlockT, H: ExHashT> Behaviour<B, H> {
@@ -197,6 +211,7 @@ impl<B: BlockT, H: ExHashT> Behaviour<B, H> {
 			finality_proof_requests,
 			light_client_handler,
 			events: VecDeque::new(),
+			bitswap: libp2p_bitswap::Bitswap::new(),
 			role,
 		})
 	}
@@ -537,6 +552,25 @@ impl<B: BlockT, H: ExHashT> NetworkBehaviourEventProcess<DiscoveryOut>
 	}
 }
 
+impl<B: BlockT, H: ExHashT> NetworkBehaviourEventProcess<libp2p_bitswap::BitswapEvent>
+	for Behaviour<B, H> {
+	fn inject_event(&mut self, event: libp2p_bitswap::BitswapEvent) {
+		match event {
+			libp2p_bitswap::BitswapEvent::ReceivedBlock(peer_id, cid, data) => {
+				self.events.push_back(
+					BehaviourOut::Bitswap(BitswapEvent::ReceivedBlock(peer_id, cid, data))
+				);
+			},
+			libp2p_bitswap::BitswapEvent::ReceivedWant(peer_id, cid, priority) => {
+				self.events.push_back(
+					BehaviourOut::Bitswap(BitswapEvent::ReceivedWant(peer_id, cid, priority))
+				);
+			}
+			libp2p_bitswap::BitswapEvent::ReceivedCancel(..) => {},
+		}
+	}
+}
+
 impl<B: BlockT, H: ExHashT> Behaviour<B, H> {
 	fn poll<TEv>(&mut self, _: &mut Context, _: &mut impl PollParameters) -> Poll<NetworkBehaviourAction<TEv, BehaviourOut<B>>> {
 		if let Some(event) = self.events.pop_front() {

client/network/src/gossip.rs

@@ -81,7 +81,7 @@ impl<M> QueuedSender<M> {
 		protocol: ConsensusEngineId,
 		queue_size_limit: usize,
 		messages_encode: F
-	) -> (Self, impl Future<Output = ()> + Send + 'static)
+	) -> (Self, impl Future<Output = ()> + 'static)
 	where
 		M: Send + 'static,
 		B: BlockT + 'static,

client/network/src/lib.rs

@@ -266,6 +266,7 @@ pub mod network_state;
 
 #[doc(inline)]
 pub use libp2p::{multiaddr, Multiaddr, PeerId};
+pub use behaviour::BitswapEvent;
 pub use protocol::{event::{DhtEvent, Event, ObservedRole}, sync::SyncState, PeerInfo};
 pub use service::{
 	NetworkService, NetworkWorker, RequestFailure, OutboundFailure, NotificationSender,

client/network/src/protocol/event.rs

@@ -21,6 +21,7 @@ use bytes::Bytes;
 use libp2p::core::PeerId;
 use libp2p::kad::record::Key;
 use sp_runtime::ConsensusEngineId;
+use crate::behaviour::BitswapEvent;
 
 /// Events generated by DHT as a response to get_value and put_value requests.
 #[derive(Debug, Clone)]
@@ -45,6 +46,8 @@ pub enum DhtEvent {
 pub enum Event {
 	/// Event generated by a DHT.
 	Dht(DhtEvent),
+	/// Event generated by Bitswap.
+	Bitswap(BitswapEvent),
 
 	/// Opened a substream with the given node with the given notifications protocol.
 	///

client/network/src/service.rs

@@ -597,6 +597,21 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkWorker<B, H> {
 	pub fn add_reserved_peer(&self, peer: String) -> Result<(), String> {
 		self.service.add_reserved_peer(peer)
 	}
+
+	/// Get the number of bitswap peers we are connected to.
+	pub fn bitswap_num_peers(&self) -> usize {
+		self.network_service.bitswap.peers().count()
+	}
+
+	/// Get the number of bitswap peers who want a block.
+	pub fn bitswap_num_peers_want(&self, cid: &tiny_cid::Cid) -> usize {
+		self.network_service.bitswap.peers_want(cid).count()
+	}
+
+	/// Get if a specific bitswap peer wants a block.
+	pub fn bitswap_peer_wants_cid(&self, peer_id: &PeerId, cid: &tiny_cid::Cid) -> bool {
+		self.network_service.bitswap.peers_want(cid).find(|id| **id == *peer_id).is_some()
+	}
 }
 
 impl<B: BlockT + 'static, H: ExHashT> NetworkService<B, H> {
@@ -1027,6 +1042,34 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkService<B, H> {
 			.to_worker
 			.unbounded_send(ServiceToWorkerMsg::OwnBlockImported(hash, number));
 	}
+
+	/// Send a bitswap block to a peer.
+	pub fn bitswap_send_block(&self, peer_id: PeerId, cid: tiny_cid::Cid, data: Box<[u8]>) {
+		let _ = self
+			.to_worker
+			.unbounded_send(ServiceToWorkerMsg::BitswapSendBlock(peer_id, cid, data));
+	}
+
+	/// Send a bitswap block to all peers that have the block in their wantlist.
+	pub fn bitswap_send_block_all(&self, cid: tiny_cid::Cid, data: Box<[u8]>) {
+		let _ = self
+			.to_worker
+			.unbounded_send(ServiceToWorkerMsg::BitswapSendBlockAll(cid, data));
+	}
+
+	/// Send a bitswap WANT request to all peers for a block.
+	pub fn bitswap_want_block(&self, cid: tiny_cid::Cid, priority: i32) {
+		let _ = self
+			.to_worker
+			.unbounded_send(ServiceToWorkerMsg::BitswapWantBlock(cid, priority));
+	}
+
+	/// Cancel a bitswap WANT request.
+	pub fn bitswap_cancel_block(&self, cid: tiny_cid::Cid) {
+		let _ = self
+			.to_worker
+			.unbounded_send(ServiceToWorkerMsg::BitswapCancelBlock(cid));
+	}
 }
 
 impl<B: BlockT + 'static, H: ExHashT> sp_consensus::SyncOracle
@@ -1159,6 +1202,10 @@ enum ServiceToWorkerMsg<B: BlockT, H: ExHashT> {
 	DisconnectPeer(PeerId),
 	UpdateChain,
 	OwnBlockImported(B::Hash, NumberFor<B>),
+	BitswapSendBlock(PeerId, tiny_cid::Cid, Box<[u8]>),
+	BitswapSendBlockAll(tiny_cid::Cid, Box<[u8]>),
+	BitswapWantBlock(tiny_cid::Cid, i32),
+	BitswapCancelBlock(tiny_cid::Cid),
 }
 
 /// Main network worker. Must be polled in order for the network to advance.
@@ -1301,6 +1348,14 @@ impl<B: BlockT + 'static, H: ExHashT> Future for NetworkWorker<B, H> {
 					this.network_service.user_protocol_mut().update_chain(),
 				ServiceToWorkerMsg::OwnBlockImported(hash, number) =>
 					this.network_service.user_protocol_mut().own_block_imported(hash, number),
+				ServiceToWorkerMsg::BitswapSendBlock(peer_id, cid, block) =>
+					this.network_service.bitswap.send_block(&peer_id, cid, block),
+				ServiceToWorkerMsg::BitswapSendBlockAll(cid, block) =>
+					this.network_service.bitswap.send_block_all(&cid, &block),
+				ServiceToWorkerMsg::BitswapWantBlock(cid, priority) =>
+					this.network_service.bitswap.want_block(cid, priority),
+				ServiceToWorkerMsg::BitswapCancelBlock(cid) =>
+					this.network_service.bitswap.cancel_block(&cid),
 			}
 		}
 
@@ -1506,6 +1561,9 @@ impl<B: BlockT + 'static, H: ExHashT> Future for NetworkWorker<B, H> {
 
 					this.event_streams.send(Event::Dht(event));
 				},
+				Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::Bitswap(ev))) => {
+					this.event_streams.send(Event::Bitswap(ev));
+				},
 				Poll::Ready(SwarmEvent::ConnectionEstablished { peer_id, endpoint, num_established }) => {
 					trace!(target: "sub-libp2p", "Libp2p => Connected({:?})", peer_id);
 
@@ -1531,14 +1589,14 @@ impl<B: BlockT + 'static, H: ExHashT> Future for NetworkWorker<B, H> {
 						let reason = match cause {
 							Some(ConnectionError::IO(_)) => "transport-error",
 							Some(ConnectionError::Handler(NodeHandlerWrapperError::Handler(EitherError::A(EitherError::A(
-								EitherError::A(EitherError::A(EitherError::A(EitherError::B(
-								EitherError::A(PingFailure::Timeout)))))))))) => "ping-timeout",
+								EitherError::A(EitherError::A(EitherError::A(EitherError::A(EitherError::B(
+								EitherError::A(PingFailure::Timeout))))))))))) => "ping-timeout",
 							Some(ConnectionError::Handler(NodeHandlerWrapperError::Handler(EitherError::A(EitherError::A(
-								EitherError::A(EitherError::A(EitherError::A(EitherError::A(
-								NotifsHandlerError::Legacy(LegacyConnectionKillError)))))))))) => "force-closed",
+								EitherError::A(EitherError::A(EitherError::A(EitherError::A(EitherError::A(
+								NotifsHandlerError::Legacy(LegacyConnectionKillError))))))))))) => "force-closed",
 							Some(ConnectionError::Handler(NodeHandlerWrapperError::Handler(EitherError::A(EitherError::A(
-								EitherError::A(EitherError::A(EitherError::A(EitherError::A(
-								NotifsHandlerError::SyncNotificationsClogged))))))))) => "sync-notifications-clogged",
+								EitherError::A(EitherError::A(EitherError::A(EitherError::A(EitherError::A(
+								NotifsHandlerError::SyncNotificationsClogged)))))))))) => "sync-notifications-clogged",
 							Some(ConnectionError::Handler(NodeHandlerWrapperError::Handler(_))) => "protocol-error",
 							Some(ConnectionError::Handler(NodeHandlerWrapperError::KeepAliveTimeout)) => "keep-alive-timeout",
 							None => "actively-closed",

client/network/src/service/out_events.rs

@@ -227,6 +227,11 @@ impl Metrics {
 				self.events_total
 					.with_label_values(&["dht", "sent", name])
 					.inc_by(num);
+			},
+			Event::Bitswap(_) => {
+				self.events_total
+					.with_label_values(&["bitswap", "sent", name])
+					.inc_by(num);
 			}
 			Event::NotificationStreamOpened { engine_id, .. } => {
 				self.events_total
@@ -258,6 +263,11 @@ impl Metrics {
 					.with_label_values(&["dht", "received", name])
 					.inc();
 			}
+			Event::Bitswap(_) => {
+				self.events_total
+					.with_label_values(&["bitswap", "received", name])
+					.inc();
+			}
 			Event::NotificationStreamOpened { engine_id, .. } => {
 				self.events_total
 					.with_label_values(&[&format!("notif-open-{:?}", engine_id), "received", name])

client/network/src/service/tests.rs

@@ -271,6 +271,8 @@ fn notifications_state_consistent() {
 				// Add new events here.
 				future::Either::Left(Event::Dht(_)) => {}
 				future::Either::Right(Event::Dht(_)) => {}
+				future::Either::Left(Event::Bitswap(_)) => {}
+				future::Either::Right(Event::Bitswap(_)) => {}
 			};
 		}
 	});

client/rpc-api/src/system/error.rs

@@ -48,7 +48,7 @@ impl From<Error> for rpc::Error {
 				data: serde_json::to_value(h).ok(),
 			},
 			Error::MalformattedPeerArg(ref e) => rpc::Error {
-				code :rpc::ErrorCode::ServerError(BASE_ERROR + 2),
+				code: rpc::ErrorCode::ServerError(BASE_ERROR + 2),
 				message: e.clone(),
 				data: None,
 			}

client/service/src/lib.rs

@@ -200,7 +200,7 @@ pub struct PartialComponents<Client, Backend, SelectChain, ImportQueue, Transact
 async fn build_network_future<
 	B: BlockT,
 	C: BlockchainEvents<B>,
-	H: sc_network::ExHashT
+	H: sc_network::ExHashT,
 > (
 	role: Role,
 	mut network: sc_network::NetworkWorker<B, H>,

primitives/core/src/offchain/storage.rs

@@ -24,7 +24,7 @@ use std::iter::Iterator;
 /// In-memory storage for offchain workers.
 #[derive(Debug, Clone, Default)]
 pub struct InMemOffchainStorage {
-	storage: HashMap<Vec<u8>, Vec<u8>>,
+	pub(crate) storage: HashMap<Vec<u8>, Vec<u8>>,
 }
 
 impl InMemOffchainStorage {

primitives/core/src/offchain/testing.rs

@@ -87,6 +87,11 @@ impl TestPersistentOffchainDB {
 			}
 		}
 	}
+
+	/// Get whether the DB is empty.
+	pub fn is_empty(&self) -> bool {
+		self.persistent.read().storage.is_empty()
+	}
 }
 
 impl OffchainStorage for TestPersistentOffchainDB {