From c5b5c0d6d2e2d96885894b752d76acf77b406077 Mon Sep 17 00:00:00 2001 From: Toralf Wittner Date: Mon, 10 Feb 2020 16:20:11 +0100 Subject: [PATCH 1/3] network: Use "one shot" protocol handler. Add two new `NetworkBehaviour`s, one handling remote block requests and another one to handle light client requests (both local and from remote). The change is motivated by the desire to use multiple substreams of a single connection for different protocols. To achieve this, libp2p's `OneShotHandler` is used as a protocol handler in each behaviour. It will open a fresh substream for the duration of the request and close it afterwards. For block requests, we currently only handle incoming requests from remote and tests are missing. For light client handling we support incoming requests from remote and also ported a substantial amount of functionality over from `light_dispatch.rs` (including several tests). However the result lacks in at least two aspects: (1) We require external updates w.r.t. the best block per peer and currently nothing updates this information. (2) We carry a lot of peer-related state around. Both aspects could be simplified by externalising peer selection and just requiring a specific peer ID where the request should be sent to. We still have to maintain some peer related state due to the way libp2p's swarm and network behaviour work (e.g. we must make sure to always issue `NetworkBehaviourAction::SendEvent`s to peers we are connected to, otherwise the actions die a silent death. Another change implemented here is the use of protocol buffers as the encoding for network messages. Certain individual fields of messages are still SCALE encoded. There has been some discussion about this in another PR (https://github.com/paritytech/substrate/pull/3452), so far without resolution. --- Cargo.lock | 7 + client/network/Cargo.toml | 10 + client/network/build.rs | 8 + client/network/src/behaviour.rs | 18 +- client/network/src/protocol.rs | 15 + client/network/src/protocol/block_requests.rs | 338 ++++ .../src/protocol/light_client_handler.rs | 1777 +++++++++++++++++ client/network/src/protocol/light_dispatch.rs | 47 +- .../network/src/protocol/schema/api.v1.proto | 59 + .../src/protocol/schema/light.v1.proto | 128 ++ client/network/src/service.rs | 23 +- 11 files changed, 2403 insertions(+), 27 deletions(-) create mode 100644 client/network/build.rs create mode 100644 client/network/src/protocol/block_requests.rs create mode 100644 client/network/src/protocol/light_client_handler.rs create mode 100644 client/network/src/protocol/schema/api.v1.proto create mode 100644 client/network/src/protocol/schema/light.v1.proto diff --git a/Cargo.lock b/Cargo.lock index b93b32fd931f5..3ae1cb9211e32 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6034,6 +6034,8 @@ dependencies = [ name = "sc-network" version = "0.8.0" dependencies = [ + "assert_matches", + "async-std", "bitflags", "bytes 0.5.4", "derive_more", @@ -6050,8 +6052,11 @@ dependencies = [ "linked_hash_set", "log 0.4.8", "lru 0.4.3", + "nohash-hasher", "parity-scale-codec", "parking_lot 0.10.0", + "prost", + "prost-build", "quickcheck", "rand 0.7.3", "rustc-hex", @@ -6073,8 +6078,10 @@ dependencies = [ "sp-runtime", "sp-test-primitives", "substrate-test-client", + "substrate-test-runtime", "substrate-test-runtime-client", "tempfile", + "thiserror", "unsigned-varint", "void", "wasm-timer", diff --git a/client/network/Cargo.toml b/client/network/Cargo.toml index 10d5f1ce9b6a8..ca9882f606ea2 100644 --- a/client/network/Cargo.toml +++ b/client/network/Cargo.toml @@ -6,6 +6,9 @@ license = "GPL-3.0" authors = ["Parity Technologies "] edition = "2018" +[build-dependencies] +prost-build = "0.6.1" + [dependencies] bitflags = "1.2.0" bytes = "0.5.0" @@ -24,7 +27,9 @@ linked-hash-map = "0.5.2" linked_hash_set = "0.1.3" log = "0.4.8" lru = "0.4.0" +nohash-hasher = "0.1.3" parking_lot = "0.10.0" +prost = "0.6.1" rand = "0.7.2" rustc-hex = "2.0.1" sc-block-builder = { version = "0.8", path = "../block-builder" } @@ -45,16 +50,21 @@ sp-keyring = { version = "2.0.0", optional = true, path = "../../primitives/keyr sp-runtime = { version = "2.0.0", path = "../../primitives/runtime" } substrate-test-client = { version = "2.0.0", optional = true, path = "../../test-utils/client" } substrate-test-runtime-client = { version = "2.0.0", optional = true, path = "../../test-utils/runtime/client" } +thiserror = "1" unsigned-varint = { version = "0.3.0", features = ["codec"] } void = "1.0.2" zeroize = "1.0.0" [dev-dependencies] +async-std = "1.5" +assert_matches = "1.3" env_logger = "0.7.0" quickcheck = "0.9.0" rand = "0.7.2" sp-keyring = { version = "2.0.0", path = "../../primitives/keyring" } sp-test-primitives = { version = "2.0.0", path = "../../primitives/test-primitives" } +substrate-test-runtime = { version = "2.0.0", path = "../../test-utils/runtime" } +substrate-test-runtime-client = { version = "2.0.0", path = "../../test-utils/runtime/client" } tempfile = "3.1.0" [features] diff --git a/client/network/build.rs b/client/network/build.rs new file mode 100644 index 0000000000000..0fd1f128660e9 --- /dev/null +++ b/client/network/build.rs @@ -0,0 +1,8 @@ +const PROTOS: &[&str] = &[ + "src/protocol/schema/api.v1.proto", + "src/protocol/schema/light.v1.proto" +]; + +fn main() { + prost_build::compile_protos(PROTOS, &["src/protocol"]).unwrap(); +} diff --git a/client/network/src/behaviour.rs b/client/network/src/behaviour.rs index 8b903cec351f5..03634ce103d36 100644 --- a/client/network/src/behaviour.rs +++ b/client/network/src/behaviour.rs @@ -19,7 +19,7 @@ use crate::{ Event, protocol::event::DhtEvent }; use crate::{ExHashT, specialization::NetworkSpecialization}; -use crate::protocol::{CustomMessageOutcome, Protocol}; +use crate::protocol::{self, CustomMessageOutcome, Protocol}; use libp2p::NetworkBehaviour; use libp2p::core::{Multiaddr, PeerId, PublicKey}; use libp2p::kad::record; @@ -42,7 +42,10 @@ pub struct Behaviour, H: ExHashT> { debug_info: debug_info::DebugInfoBehaviour>, /// Discovers nodes of the network. discovery: DiscoveryBehaviour>, - + /// Block request handling. + block_requests: protocol::BlockRequests, B>, + /// Light client request handling. + light_client_handler: protocol::LightClientHandler, B>, /// Queue of events to produce for the outside. #[behaviour(ignore)] events: Vec>, @@ -65,6 +68,8 @@ impl, H: ExHashT> Behaviour { known_addresses: Vec<(PeerId, Multiaddr)>, enable_mdns: bool, allow_private_ipv4: bool, + block_requests: protocol::BlockRequests, B>, + light_client_handler: protocol::LightClientHandler, B> ) -> Self { Behaviour { substrate, @@ -75,7 +80,9 @@ impl, H: ExHashT> Behaviour { enable_mdns, allow_private_ipv4 ).await, - events: Vec::new(), + block_requests, + light_client_handler, + events: Vec::new() } } @@ -117,6 +124,11 @@ impl, H: ExHashT> Behaviour { pub fn put_value(&mut self, key: record::Key, value: Vec) { self.discovery.put_value(key, value); } + +// /// Issue a light client request. +// pub fn light_client_request(&mut self, r: light_client_handler::Request) -> Result<(), light_client_handler::Error> { +// self.light_client_handler.request(r) +// } } impl, H: ExHashT> NetworkBehaviourEventProcess for diff --git a/client/network/src/protocol.rs b/client/network/src/protocol.rs index 99b0117882c28..8e5f9d6e7bc8f 100644 --- a/client/network/src/protocol.rs +++ b/client/network/src/protocol.rs @@ -54,15 +54,30 @@ use crate::error; use util::LruHashSet; use wasm_timer::Instant; +// Include sources generated from protobuf definitions. +pub mod api { + pub mod v1 { + include!(concat!(env!("OUT_DIR"), "/api.v1.rs")); + pub mod light { + include!(concat!(env!("OUT_DIR"), "/api.v1.light.rs")); + } + } +} + mod legacy_proto; mod util; +pub mod block_requests; pub mod message; pub mod event; +pub mod light_client_handler; pub mod light_dispatch; pub mod specialization; pub mod sync; +pub use block_requests::BlockRequests; +pub use light_client_handler::LightClientHandler; + const REQUEST_TIMEOUT_SEC: u64 = 40; /// Interval at which we perform time based maintenance const TICK_TIMEOUT: time::Duration = time::Duration::from_millis(1100); diff --git a/client/network/src/protocol/block_requests.rs b/client/network/src/protocol/block_requests.rs new file mode 100644 index 0000000000000..c67d096c78ec1 --- /dev/null +++ b/client/network/src/protocol/block_requests.rs @@ -0,0 +1,338 @@ +//! `NetworkBehaviour` implementation which handles incoming block requests. +//! +//! Every request is coming in on a separate connection substream which gets +//! closed after we have sent the response back. Incoming requests are encoded +//! as protocol buffers (cf. `api.v1.proto`). + +#![allow(unused)] + +use bytes::Bytes; +use codec::{Encode, Decode}; +use crate::{ + chain::Client, + config::ProtocolId, + protocol::{api, message::BlockAttributes} +}; +use futures::{future::BoxFuture, prelude::*, stream::FuturesUnordered}; +use libp2p::{ + core::{ + ConnectedPoint, + Multiaddr, + PeerId, + upgrade::{InboundUpgrade, ReadOneError, UpgradeInfo, Negotiated}, + upgrade::{DeniedUpgrade, read_one, write_one} + }, + swarm::{NetworkBehaviour, NetworkBehaviourAction, OneShotHandler, PollParameters, SubstreamProtocol} +}; +use prost::Message; +use sp_runtime::{generic::BlockId, traits::{Block, Header, One, Zero}}; +use std::{ + cmp::min, + io, + iter, + sync::Arc, + time::Duration, + task::{Context, Poll} +}; +use void::{Void, unreachable}; + +// Type alias for convenience. +pub type Error = Box; + +/// Configuration options for `BlockRequests`. +#[derive(Debug, Clone)] +pub struct Config { + max_block_data_response: u32, + max_request_len: usize, + inactivity_timeout: Duration, + protocol: Bytes, +} + +impl Config { + /// Create a fresh configuration with the following options: + /// + /// - max. block data in response = 128 + /// - max. request size = 1 MiB + /// - inactivity timeout = 15s + pub fn new(id: &ProtocolId) -> Self { + let mut c = Config { + max_block_data_response: 128, + max_request_len: 1024 * 1024, + inactivity_timeout: Duration::from_secs(15), + protocol: Bytes::new(), + }; + c.set_protocol(id); + c + } + + /// Limit the max. number of block data in a response. + pub fn set_max_block_data_response(&mut self, v: u32) -> &mut Self { + self.max_block_data_response = v; + self + } + + /// Limit the max. length of incoming block request bytes. + pub fn set_max_request_len(&mut self, v: usize) -> &mut Self { + self.max_request_len = v; + self + } + + /// Limit the max. duration the substream may remain inactive before closing it. + pub fn set_inactivity_timeout(&mut self, v: Duration) -> &mut Self { + self.inactivity_timeout = v; + self + } + + /// Set protocol to use for upgrade negotiation. + pub fn set_protocol(&mut self, id: &ProtocolId) -> &mut Self { + let mut v = Vec::new(); + v.extend_from_slice(b"/"); + v.extend_from_slice(id.as_bytes()); + v.extend_from_slice(b"/sync/1"); + self.protocol = v.into(); + self + } +} + +/// The block request handling behaviour. +pub struct BlockRequests { + /// This behaviour's configuration. + config: Config, + /// Blockchain client. + chain: Arc>, + /// Futures sending back the block request response. + outgoing: FuturesUnordered>, + /// Type witness term. + _marker: std::marker::PhantomData +} + +impl BlockRequests +where + T: AsyncRead + AsyncWrite + Unpin + Send + 'static, + B: Block, +{ + pub fn new(cfg: Config, chain: Arc>) -> Self { + BlockRequests { + config: cfg, + chain, + outgoing: FuturesUnordered::new(), + _marker: std::marker::PhantomData + } + } + + /// Callback, invoked when a new block request has been received from remote. + fn on_block_request + ( &mut self + , peer: &PeerId + , request: &api::v1::BlockRequest + ) -> Result + { + log::trace!("block request {} from peer {}: from block {:?} to block {:?}, max blocks {:?}", + request.id, + peer, + request.from_block, + request.to_block, + request.max_blocks); + + let from_block_id = + match request.from_block { + Some(api::v1::block_request::FromBlock::Hash(ref h)) => { + let h = Decode::decode(&mut h.as_ref())?; + BlockId::::Hash(h) + } + Some(api::v1::block_request::FromBlock::Number(ref n)) => { + let n = Decode::decode(&mut n.as_ref())?; + BlockId::::Number(n) + } + None => { + let msg = "missing `BlockRequest::from_block` field"; + return Err(io::Error::new(io::ErrorKind::Other, msg).into()) + } + }; + + let max_blocks = + if request.max_blocks == 0 { + self.config.max_block_data_response + } else { + min(request.max_blocks, self.config.max_block_data_response) + }; + + let direction = + if request.direction == api::v1::Direction::Ascending as i32 { + api::v1::Direction::Ascending + } else if request.direction == api::v1::Direction::Descending as i32 { + api::v1::Direction::Descending + } else { + let msg = format!("invalid `BlockRequest::direction` value: {}", request.direction); + return Err(io::Error::new(io::ErrorKind::Other, msg).into()) + }; + + let attributes = BlockAttributes::decode(&mut request.fields.to_be_bytes().as_ref())?; + let get_header = attributes.contains(BlockAttributes::HEADER); + let get_body = attributes.contains(BlockAttributes::BODY); + let get_justification = attributes.contains(BlockAttributes::JUSTIFICATION); + + let mut blocks = Vec::new(); + let mut block_id = from_block_id; + while let Some(header) = self.chain.header(&block_id).unwrap_or(None) { + if blocks.len() >= max_blocks as usize { + break + } + + let number = header.number().clone(); + let hash = header.hash(); + let parent_hash = header.parent_hash().clone(); + + let block_data = api::v1::BlockData { + hash: hash.encode(), + header: if get_header { + header.encode() + } else { + Vec::new() + }, + body: if get_body { + self.chain.body(&BlockId::Hash(hash))? + .unwrap_or(Vec::new()) + .iter_mut() + .map(|extrinsic| extrinsic.encode()) + .collect() + } else { + Vec::new() + }, + receipt: Vec::new(), + message_queue: Vec::new(), + justification: if get_justification { + self.chain.justification(&BlockId::Hash(hash))?.unwrap_or(Vec::new()) + } else { + Vec::new() + } + }; + + blocks.push(block_data); + + match direction { + api::v1::Direction::Ascending => { + block_id = BlockId::Number(number + One::one()) + } + api::v1::Direction::Descending => { + if number.is_zero() { + break + } + block_id = BlockId::Hash(parent_hash) + } + } + } + + Ok(api::v1::BlockResponse { id: request.id, blocks }) + } +} + +impl NetworkBehaviour for BlockRequests +where + T: AsyncRead + AsyncWrite + Unpin + Send + 'static, + B: Block +{ + type ProtocolsHandler = OneShotHandler>>; + type OutEvent = Void; + + fn new_handler(&mut self) -> Self::ProtocolsHandler { + let p = Protocol { + max_request_len: self.config.max_request_len, + protocol: self.config.protocol.clone(), + }; + OneShotHandler::new(SubstreamProtocol::new(p), self.config.inactivity_timeout) + } + + fn addresses_of_peer(&mut self, _: &PeerId) -> Vec { + Vec::new() + } + + fn inject_connected(&mut self, _peer: PeerId, _info: ConnectedPoint) { + } + + fn inject_disconnected(&mut self, _peer: &PeerId, _info: ConnectedPoint) { + } + + fn inject_node_event(&mut self, peer: PeerId, Request(request, mut stream): Request>) { + match self.on_block_request(&peer, &request) { + Ok(res) => { + log::trace!("enqueueing block response {} for peer {} with {} blocks", res.id, peer, res.blocks.len()); + let mut data = Vec::with_capacity(res.encoded_len()); + if let Err(e) = res.encode(&mut data) { + log::debug!("error encoding block response {} for peer {}: {}", res.id, peer, e) + } else { + let future = async move { + if let Err(e) = write_one(&mut stream, data).await { + log::debug!("error writing block response: {}", e) + } + }; + self.outgoing.push(future.boxed()) + } + } + Err(e) => log::debug!("error handling block request {} from peer {}: {}", request.id, peer, e) + } + } + + fn poll(&mut self, cx: &mut Context, _: &mut impl PollParameters) -> Poll> { + while let Poll::Ready(Some(_)) = self.outgoing.poll_next_unpin(cx) {} + Poll::Pending + } +} + +/// The incoming block request. +/// +/// Holds the protobuf value and the connection substream which made the +/// request and over which to send the response. +#[derive(Debug)] +pub struct Request(api::v1::BlockRequest, T); + +impl From for Request { + fn from(v: Void) -> Self { + unreachable(v) + } +} + +/// Substream upgrade protocol. +/// +/// We attempt to parse an incoming protobuf encoded request (cf. `Request`) +/// which will be handled by the `BlockRequests` behaviour, i.e. the request +/// will become visible via `inject_node_event` which then dispatches to the +/// relevant callback to process the message and prepare a response. +#[derive(Debug, Clone)] +pub struct Protocol { + /// The max. request length in bytes. + max_request_len: usize, + /// The protocol to use during upgrade negotiation. + protocol: Bytes, +} + +impl UpgradeInfo for Protocol { + type Info = Bytes; + type InfoIter = iter::Once; + + fn protocol_info(&self) -> Self::InfoIter { + iter::once(self.protocol.clone()) + } +} + +impl InboundUpgrade for Protocol +where + T: AsyncRead + AsyncWrite + Unpin + Send + 'static +{ + type Output = Request; + type Error = ReadOneError; + type Future = BoxFuture<'static, Result>; + + fn upgrade_inbound(self, mut s: T, _: Self::Info) -> Self::Future { + let future = async move { + let len = self.max_request_len; + let vec = read_one(&mut s, len).await?; + match api::v1::BlockRequest::decode(&vec[..]) { + Ok(r) => Ok(Request(r, s)), + Err(e) => Err(ReadOneError::Io(io::Error::new(io::ErrorKind::Other, e))) + } + }; + future.boxed() + } +} + diff --git a/client/network/src/protocol/light_client_handler.rs b/client/network/src/protocol/light_client_handler.rs new file mode 100644 index 0000000000000..117c55cd3c9bb --- /dev/null +++ b/client/network/src/protocol/light_client_handler.rs @@ -0,0 +1,1777 @@ +//! [`NetworkBehaviour`] implementation which handles light client requests. +//! +//! Every request is coming in on a separate connection substream which gets +//! closed after we have sent the response back. Requests and responses are +//! encoded as protocol buffers (cf. `api.v1.proto`). +//! +//! For every outgoing request we likewise open a separate substream. + +#![allow(unused)] + +use bytes::Bytes; +use codec::{self, Encode, Decode}; +use crate::{ + chain::Client, + config::ProtocolId, + protocol::{api, light_dispatch::TIMEOUT_REPUTATION_CHANGE} +}; +use futures::{channel::oneshot, future::BoxFuture, prelude::*, stream::FuturesUnordered}; +use libp2p::{ + core::{ + ConnectedPoint, + Multiaddr, + PeerId, + upgrade::{InboundUpgrade, ReadOneError, UpgradeInfo, Negotiated}, + upgrade::{OutboundUpgrade, read_one, write_one} + }, + swarm::{NetworkBehaviour, NetworkBehaviourAction, OneShotHandler, PollParameters, SubstreamProtocol} +}; +use nohash_hasher::IntMap; +use prost::Message; +use rustc_hex::ToHex; +use sc_client::light::fetcher; +use sc_client_api::StorageProof; +use sc_peerset::ReputationChange; +use sp_core::storage::{ChildInfo, StorageKey}; +use sp_blockchain::{Error as ClientError}; +use sp_runtime::traits::{Block, Header, NumberFor, Zero}; +use std::{ + collections::{BTreeMap, VecDeque, HashMap}, + iter, + io, + sync::Arc, + time::{Duration, Instant}, + task::{Context, Poll} +}; +use void::Void; + +/// Configuration options for `LightClientHandler` behaviour. +#[derive(Debug, Clone)] +pub struct Config { + max_data_size: usize, + max_pending_requests: usize, + inactivity_timeout: Duration, + request_timeout: Duration, + protocol: Bytes, +} + +impl Config { + /// Create a fresh configuration with the following options: + /// + /// - max. data size = 1 MiB + /// - max. pending requests = 128 + /// - inactivity timeout = 15s + /// - request timeout = 15s + pub fn new(id: &ProtocolId) -> Self { + let mut c = Config { + max_data_size: 1024 * 1024, + max_pending_requests: 128, + inactivity_timeout: Duration::from_secs(15), + request_timeout: Duration::from_secs(15), + protocol: Bytes::new(), + }; + c.set_protocol(id); + c + } + + /// Limit the max. length of incoming request bytes. + pub fn set_max_data_size(&mut self, v: usize) -> &mut Self { + self.max_data_size = v; + self + } + + /// Limit the max. number of pending requests. + pub fn set_max_pending_requests(&mut self, v: usize) -> &mut Self { + self.max_pending_requests = v; + self + } + + /// Limit the max. duration the connection may remain inactive before closing it. + pub fn set_inactivity_timeout(&mut self, v: Duration) -> &mut Self { + self.inactivity_timeout = v; + self + } + + /// Limit the max. request duration. + pub fn set_request_timeout(&mut self, v: Duration) -> &mut Self { + self.request_timeout = v; + self + } + + /// Set protocol to use for upgrade negotiation. + pub fn set_protocol(&mut self, id: &ProtocolId) -> &mut Self { + let mut v = Vec::new(); + v.extend_from_slice(b"/"); + v.extend_from_slice(id.as_bytes()); + v.extend_from_slice(b"/light/1"); + self.protocol = v.into(); + self + } +} + +/// Possible errors while handling light clients. +#[derive(Debug, thiserror::Error)] +pub enum Error { + /// There are currently too many pending request. + #[error("too many pending requests")] + TooManyRequests, + /// The response type does not correspond to the issued request. + #[error("unexpected response")] + UnexpectedResponse, + /// A bad request has been received. + #[error("bad request: {0}")] + BadRequest(&'static str), + /// The chain client errored. + #[error("client error: {0}")] + Client(#[from] ClientError), + /// Encoding or decoding of some data failed. + #[error("codec error: {0}")] + Codec(#[from] codec::Error), +} + +/// The possible light client requests we support. +/// +/// The associated `oneshot::Sender` will be used to convey the result of +/// their request back to them (cf. `Reply`). +// +// This is modeled after light_dispatch.rs's `RequestData` which is not +// used because we currently only support a subset of those. +#[derive(Debug)] +pub enum Request { + Header { + request: fetcher::RemoteHeaderRequest, + sender: oneshot::Sender> + }, + Read { + request: fetcher::RemoteReadRequest, + sender: oneshot::Sender, Option>>, ClientError>> + }, + ReadChild { + request: fetcher::RemoteReadChildRequest, + sender: oneshot::Sender, Option>>, ClientError>> + }, + Call { + request: fetcher::RemoteCallRequest, + sender: oneshot::Sender, ClientError>> + }, + Changes { + request: fetcher::RemoteChangesRequest, + sender: oneshot::Sender, u32)>, ClientError>> + } +} + +/// The data to send back to the light client over the oneshot channel. +// +// It is unified here in order to be able to return it as a function +// result instead of delivering it to the client as a side effect of +// response processing. +#[derive(Debug)] +enum Reply { + VecU8(Vec), + VecNumberU32(Vec<(::Number, u32)>), + MapVecU8OptVecU8(HashMap, Option>>), + Header(B::Header) +} + +/// Augments a light client request with metadata. +#[derive(Debug)] +struct RequestWrapper { + /// Time when this value was created. + timestamp: Instant, + /// Remaining retries. + retries: usize, + /// The actual request. + request: Request, + /// Peer information, e.g. `PeerId`. + peer: P +} + +/// Information we have about some peer. +#[derive(Debug)] +struct PeerInfo { + address: Multiaddr, + best_block: Option>, + status: PeerStatus, +} + +/// A peer is either idle or busy processing a request from us. +#[derive(Debug, Clone, PartialEq, Eq)] +enum PeerStatus { + /// The peer is available. + Idle, + /// We wait for the peer to return us a response for the given request ID. + BusyWith(u64), +} + +/// The light client handler behaviour. +pub struct LightClientHandler { + /// This behaviour's configuration. + config: Config, + /// Blockchain client. + chain: Arc>, + /// Verifies that received responses are correct. + checker: Arc>, + /// Peer information (addresses, their best block, etc.) + peers: HashMap>, + /// Futures sending back response to remote clients. + responses: FuturesUnordered>, + /// Pending (local) requests. + pending_requests: VecDeque>, + /// Requests on their way to remote peers. + outstanding: IntMap>, + /// (Local) Request ID counter + next_request_id: u64, + /// Handle to use for reporting misbehaviour of peers. + peerset: sc_peerset::PeersetHandle, + /// Type witness term. + _marker: std::marker::PhantomData +} + +impl LightClientHandler +where + T: AsyncRead + AsyncWrite + Unpin + Send + 'static, + B: Block, +{ + /// Construct a new light client handler. + pub fn new + ( cfg: Config + , chain: Arc> + , checker: Arc> + , peerset: sc_peerset::PeersetHandle + ) -> Self + { + LightClientHandler { + config: cfg, + chain, + checker, + peers: HashMap::new(), + responses: FuturesUnordered::new(), + pending_requests: VecDeque::new(), + outstanding: IntMap::default(), + next_request_id: 1, + peerset, + _marker: std::marker::PhantomData + } + } + + /// We rely on external information about peers best blocks as we lack the + /// means to determine it ourselves. + pub fn update_best_block(&mut self, peer: &PeerId, num: NumberFor) { + if let Some(info) = self.peers.get_mut(peer) { + info.best_block = Some(num) + } + } + + /// Issue a new light client request. + pub fn request(&mut self, req: Request) -> Result<(), Error> { + if self.pending_requests.len() >= self.config.max_pending_requests { + return Err(Error::TooManyRequests) + } + let rw = RequestWrapper { + timestamp: Instant::now(), + retries: retries(&req), + request: req, + peer: (), // we do not know the peer yet + }; + self.pending_requests.push_back(rw); + Ok(()) + } + + fn next_request_id(&mut self) -> u64 { + let id = self.next_request_id; + self.next_request_id += 1; + id + } + + // Iterate over peers known to possess a certain block. + fn idle_peers_with_block(&mut self, num: NumberFor) -> impl Iterator + '_ { + self.peers.iter() + .filter(move |(_, info)| { + info.status == PeerStatus::Idle && info.best_block >= Some(num) + }) + .map(|(peer, _)| peer.clone()) + } + + // Iterate over peers without a known block. + fn idle_peers_with_unknown_block(&mut self) -> impl Iterator + '_ { + self.peers.iter() + .filter(|(_, info)| { + info.status == PeerStatus::Idle && info.best_block.is_none() + }) + .map(|(peer, _)| peer.clone()) + } + + /// Remove the given peer. + /// + /// If we have a request to this peer in flight, we move it back to + /// the pending requests queue. + fn remove_peer(&mut self, peer: &PeerId) { + if let Some(id) = self.outstanding.iter().find(|(_, rw)| &rw.peer == peer).map(|(k, _)| *k) { + let rw = self.outstanding.remove(&id).expect("key belongs to entry in this map"); + let rw = RequestWrapper { + timestamp: rw.timestamp, + retries: rw.retries, + request: rw.request, + peer: (), // need to find another peer + }; + self.pending_requests.push_back(rw); + } + self.peers.remove(peer); + } + + /// Process a local request's response from remote. + /// + /// If successful, this will give us the actual, checked data we should be + /// sending back to the client, otherwise an error. + fn on_response + ( &mut self + , peer: &PeerId + , request: &Request + , response: api::v1::light::Response + ) -> Result, Error> + { + log::trace!("response {} from {}", response.id, peer); + use api::v1::light::response::Response; + match response.response { + Some(Response::RemoteCallResponse(response)) => + if let Request::Call { request , .. } = request { + let proof = Decode::decode(&mut response.proof.as_ref())?; + let reply = self.checker.check_execution_proof(request, proof)?; + Ok(Reply::VecU8(reply)) + } else { + Err(Error::UnexpectedResponse) + } + Some(Response::RemoteReadResponse(response)) => + match request { + Request::Read { request, .. } => { + let proof = Decode::decode(&mut response.proof.as_ref())?; + let reply = self.checker.check_read_proof(&request, proof)?; + Ok(Reply::MapVecU8OptVecU8(reply)) + } + Request::ReadChild { request, .. } => { + let proof = Decode::decode(&mut response.proof.as_ref())?; + let reply = self.checker.check_read_child_proof(&request, proof)?; + Ok(Reply::MapVecU8OptVecU8(reply)) + } + _ => Err(Error::UnexpectedResponse) + } + Some(Response::RemoteChangesResponse(response)) => + if let Request::Changes { request, .. } = request { + let max_block = Decode::decode(&mut response.max.as_ref())?; + let roots_proof = Decode::decode(&mut response.roots_proof.as_ref())?; + let roots = { + let mut r = BTreeMap::new(); + for pair in response.roots { + let k = Decode::decode(&mut pair.fst.as_ref())?; + let v = Decode::decode(&mut pair.snd.as_ref())?; + r.insert(k, v); + } + r + }; + let reply = self.checker.check_changes_proof(&request, fetcher::ChangesProof { + max_block, + proof: response.proof, + roots, + roots_proof, + })?; + Ok(Reply::VecNumberU32(reply)) + } else { + Err(Error::UnexpectedResponse) + } + Some(Response::RemoteHeaderResponse(response)) => + if let Request::Header { request, .. } = request { + let header = + if response.header.is_empty() { + None + } else { + Some(Decode::decode(&mut response.header.as_ref())?) + }; + let proof = Decode::decode(&mut response.proof.as_ref())?; + let reply = self.checker.check_header_proof(&request, header, proof)?; + Ok(Reply::Header(reply)) + } else { + Err(Error::UnexpectedResponse) + } + None => Err(Error::UnexpectedResponse) + } + } + + fn on_remote_call_request + ( &mut self + , peer: &PeerId + , request_id: u64 + , request: &api::v1::light::RemoteCallRequest + ) -> Result + { + log::trace!("remote call request {} from {} ({} at {:?})", + request_id, + peer, + request.method, + request.block); + + let block = Decode::decode(&mut request.block.as_ref())?; + + let proof = match self.chain.execution_proof(&block, &request.method, &request.data) { + Ok((_, proof)) => proof, + Err(e) => { + log::trace!("remote call request {} from {} ({} at {:?}) failed with: {}", + request_id, + peer, + request.method, + request.block, + e); + StorageProof::empty() + } + }; + + let response = { + let r = api::v1::light::RemoteCallResponse { proof: proof.encode() }; + api::v1::light::response::Response::RemoteCallResponse(r) + }; + + Ok(api::v1::light::Response { id: request_id, response: Some(response) }) + } + + fn on_remote_read_request + ( &mut self + , peer: &PeerId + , request_id: u64 + , request: &api::v1::light::RemoteReadRequest + ) -> Result + { + if request.keys.is_empty() { + log::debug!("invalid remote read request sent by {}", peer); + return Err(Error::BadRequest("remote read request without keys")) + } + + log::trace!("remote read request {} from {} ({} at {:?})", + request_id, + peer, + fmt_keys(request.keys.first(), request.keys.last()), + request.block); + + let block = Decode::decode(&mut request.block.as_ref())?; + + let proof = match self.chain.read_proof(&block, &request.keys) { + Ok(proof) => proof, + Err(error) => { + log::trace!("remote read request {} from {} ({} at {:?}) failed with: {}", + request_id, + peer, + fmt_keys(request.keys.first(), request.keys.last()), + request.block, + error); + StorageProof::empty() + } + }; + + let response = { + let r = api::v1::light::RemoteReadResponse { proof: proof.encode() }; + api::v1::light::response::Response::RemoteReadResponse(r) + }; + + Ok(api::v1::light::Response { id: request_id, response: Some(response) }) + } + + fn on_remote_read_child_request + ( &mut self + , peer: &PeerId + , request_id: u64 + , request: &api::v1::light::RemoteReadChildRequest + ) -> Result + { + if request.keys.is_empty() { + log::debug!("invalid remote child read request sent by {}", peer); + return Err(Error::BadRequest("remove read child request without keys")) + } + + log::trace!("remote read child request {} from {} ({} {} at {:?})", + request_id, + peer, + request.storage_key.to_hex::(), + fmt_keys(request.keys.first(), request.keys.last()), + request.block); + + let block = Decode::decode(&mut request.block.as_ref())?; + + let proof = + if let Some(info) = ChildInfo::resolve_child_info(request.child_type, &request.child_info[..]) { + match self.chain.read_child_proof(&block, &request.storage_key, info, &request.keys) { + Ok(proof) => proof, + Err(error) => { + log::trace!("remote read child request {} from {} ({} {} at {:?}) failed with: {}", + request_id, + peer, + request.storage_key.to_hex::(), + fmt_keys(request.keys.first(), request.keys.last()), + request.block, + error); + StorageProof::empty() + } + } + } else { + log::trace!("remote read child request {} from {} ({} {} at {:?}) failed with: {}", + request_id, + peer, + request.storage_key.to_hex::(), + fmt_keys(request.keys.first(), request.keys.last()), + request.block, + "invalid child info and type" + ); + StorageProof::empty() + }; + + let response = { + let r = api::v1::light::RemoteReadResponse { proof: proof.encode() }; + api::v1::light::response::Response::RemoteReadResponse(r) + }; + + Ok(api::v1::light::Response { id: request_id, response: Some(response) }) + } + + fn on_remote_header_request + ( &mut self + , peer: &PeerId + , request_id: u64 + , request: &api::v1::light::RemoteHeaderRequest + ) -> Result + { + log::trace!("remote header proof request {} from {} ({:?})", request_id, peer, request.block); + + let block = Decode::decode(&mut request.block.as_ref())?; + + let (header, proof) = match self.chain.header_proof(block) { + Ok((header, proof)) => (header.encode(), proof), + Err(error) => { + log::trace!("remote header proof request {} from {} ({:?}) failed with: {}", + request_id, + peer, + request.block, + error); + (Default::default(), StorageProof::empty()) + } + }; + + let response = { + let r = api::v1::light::RemoteHeaderResponse { header, proof: proof.encode() }; + api::v1::light::response::Response::RemoteHeaderResponse(r) + }; + + Ok(api::v1::light::Response { id: request_id, response: Some(response) }) + } + + fn on_remote_changes_request + ( &mut self + , peer: &PeerId + , request_id: u64 + , request: &api::v1::light::RemoteChangesRequest + ) -> Result + { + log::trace!("remote changes proof request {} from {} for key {} ({:?}..{:?})", + request_id, + peer, + if !request.storage_key.is_empty() { + format!("{} : {}", request.storage_key.to_hex::(), request.key.to_hex::()) + } else { + request.key.to_hex::() + }, + request.first, + request.last); + + let first = Decode::decode(&mut request.first.as_ref())?; + let last = Decode::decode(&mut request.last.as_ref())?; + let min = Decode::decode(&mut request.min.as_ref())?; + let max = Decode::decode(&mut request.max.as_ref())?; + let key = StorageKey(request.key.clone()); + let storage_key = + if request.storage_key.is_empty() { + None + } else { + Some(StorageKey(request.storage_key.clone())) + }; + + let proof = match self.chain.key_changes_proof(first, last, min, max, storage_key.as_ref(), &key) { + Ok(proof) => proof, + Err(error) => { + log::trace!("remote changes proof request {} from {} for key {} ({:?}..{:?}) failed with: {}", + request_id, + peer, + if let Some(sk) = storage_key { + format!("{} : {}", sk.0.to_hex::(), key.0.to_hex::()) + } else { + key.0.to_hex::() + }, + request.first, + request.last, + error); + + fetcher::ChangesProof:: { + max_block: Zero::zero(), + proof: Vec::new(), + roots: BTreeMap::new(), + roots_proof: StorageProof::empty(), + } + } + }; + + let response = { + let r = api::v1::light::RemoteChangesResponse { + max: proof.max_block.encode(), + proof: proof.proof, + roots: proof.roots.into_iter() + .map(|(k, v)| api::v1::light::Pair { fst: k.encode(), snd: v.encode() }) + .collect(), + roots_proof: proof.roots_proof.encode(), + }; + api::v1::light::response::Response::RemoteChangesResponse(r) + }; + + Ok(api::v1::light::Response { id: request_id, response: Some(response) }) + } +} + +impl NetworkBehaviour for LightClientHandler +where + T: AsyncRead + AsyncWrite + Unpin + Send + 'static, + B: Block +{ + type ProtocolsHandler = OneShotHandler>>; + type OutEvent = Void; + + fn new_handler(&mut self) -> Self::ProtocolsHandler { + let p = InboundProtocol { + max_data_size: self.config.max_data_size, + protocol: self.config.protocol.clone(), + }; + OneShotHandler::new(SubstreamProtocol::new(p), self.config.inactivity_timeout) + } + + fn addresses_of_peer(&mut self, peer: &PeerId) -> Vec { + self.peers.get(peer) + .map(|info| vec![info.address.clone()]) + .unwrap_or_default() + } + + fn inject_connected(&mut self, peer: PeerId, info: ConnectedPoint) { + let peer_address = match info { + ConnectedPoint::Listener { send_back_addr, .. } => send_back_addr, + ConnectedPoint::Dialer { address } => address + }; + + log::trace!("peer {} connected with address {}", peer, peer_address); + + let info = PeerInfo { + address: peer_address, + best_block: None, + status: PeerStatus::Idle, + }; + + self.peers.insert(peer, info); + } + + fn inject_disconnected(&mut self, peer: &PeerId, _: ConnectedPoint) { + log::trace!("peer {} disconnected", peer); + self.remove_peer(peer) + } + + fn inject_node_event(&mut self, peer: PeerId, event: Event>) { + match event { + // An incoming request from remote has been received. + Event::Request(request, mut stream) => { + log::trace!("incoming request {} from {}", peer, request.id); + let result = match &request.request { + Some(api::v1::light::request::Request::RemoteCallRequest(r)) => + self.on_remote_call_request(&peer, request.id, r), + Some(api::v1::light::request::Request::RemoteReadRequest(r)) => + self.on_remote_read_request(&peer, request.id, r), + Some(api::v1::light::request::Request::RemoteHeaderRequest(r)) => + self.on_remote_header_request(&peer, request.id, r), + Some(api::v1::light::request::Request::RemoteReadChildRequest(r)) => + self.on_remote_read_child_request(&peer, request.id, r), + Some(api::v1::light::request::Request::RemoteChangesRequest(r)) => + self.on_remote_changes_request(&peer, request.id, r), + None => { + log::debug!("ignoring request {} without request data from peer {}", request.id, peer); + return + } + }; + match result { + Ok(response) => { + log::trace!("enqueueing response {} for peer {}", response.id, peer); + let mut data = Vec::new(); + if let Err(e) = response.encode(&mut data) { + log::debug!("error encoding response {} for peer {}: {}", response.id, peer, e) + } else { + let future = async move { + if let Err(e) = write_one(&mut stream, data).await { + log::debug!("error writing response: {}", e) + } + }; + self.responses.push(future.boxed()) + } + } + Err(Error::BadRequest(_)) => { + self.remove_peer(&peer); + self.peerset.report_peer(peer, ReputationChange::new(-(1 << 12), "bad request")) + } + Err(e) => log::debug!("error handling request {} from peer {}: {}", request.id, peer, e) + } + } + // A response to one of our own requests has been received. + Event::Response(response) => { + let id = response.id; + if let Some(request) = self.outstanding.remove(&id) { + // We first just check if the response originates from the expected peer. + if request.peer != peer { + log::debug!("was expecting response {} from {} instead of {}", id, request.peer, peer); + self.outstanding.insert(id, request); + self.remove_peer(&peer); + self.peerset.report_peer(peer, ReputationChange::new_fatal("response from unexpected peer")); + return + } + + if let Some(info) = self.peers.get_mut(&peer) { + if info.status != PeerStatus::BusyWith(id) { + // If we get here, something is wrong with our internal handling of peer + // status information. At any time, a single peer processes at most one + // request from us and its status should contain the request ID we are + // expecting a response for. If a peer would send us a response with a + // random ID, we should not have an entry for it with this peer ID in + // our `outstanding` map, so a malicious peer should not be able to get + // us here. It is our own fault and must be fixed! + panic!("unexpected peer status {:?} for {}", info.status, peer); + } + + info.status = PeerStatus::Idle; // Make peer available again. + + match self.on_response(&peer, &request.request, response) { + Ok(reply) => send_reply(Ok(reply), request.request), + Err(Error::UnexpectedResponse) => { + log::debug!("unexpected response {} from peer {}", id, peer); + self.remove_peer(&peer); + self.peerset.report_peer(peer, ReputationChange::new_fatal("unexpected response from peer")); + let rw = RequestWrapper { + timestamp: request.timestamp, + retries: request.retries, + request: request.request, + peer: (), + }; + self.pending_requests.push_back(rw); + } + Err(other) => { + log::debug!("error handling response {} from peer {}: {}", id, peer, other); + self.remove_peer(&peer); + self.peerset.report_peer(peer, ReputationChange::new_fatal("invalid response from peer")); + if request.retries > 0 { + let rw = RequestWrapper { + timestamp: request.timestamp, + retries: request.retries - 1, + request: request.request, + peer: (), + }; + self.pending_requests.push_back(rw) + } else { + send_reply(Err(ClientError::RemoteFetchFailed), request.request) + } + } + } + } else { + // If we get here, something is wrong with our internal handling of peers. + // We apparently have an entry in our `outstanding` map and the peer is the one we + // expected. So, if we can not find an entry for it in our peer information table, + // then these two collections are out of sync which must not happen and is a clear + // programmer error that must be fixed! + panic!("missing peer information for {}; response {}", peer, id); + } + } else { + log::debug!("unexpected response {} from peer {}", id, peer); + self.remove_peer(&peer); + self.peerset.report_peer(peer, ReputationChange::new_fatal("response from unexpected peer")); + } + } + } + } + + fn poll(&mut self, cx: &mut Context, _: &mut impl PollParameters) -> Poll> { + // Process response sending futures. + while let Poll::Ready(Some(_)) = self.responses.poll_next_unpin(cx) {} + + // If we have a pending request to send, try to find an available peer and send it. + let now = Instant::now(); + while let Some(mut request) = self.pending_requests.pop_front() { + if now > request.timestamp + self.config.request_timeout { + if request.retries == 0 { + send_reply(Err(ClientError::RemoteFetchFailed), request.request); + continue + } + request.timestamp = Instant::now(); + request.retries -= 1 + } + let number = required_block(&request.request); + let available_peer = { + let p = self.idle_peers_with_block(number).next(); + if p.is_none() { + self.idle_peers_with_unknown_block().next() + } else { + p + } + }; + if let Some(peer) = available_peer { + let id = self.next_request_id(); + let rq = serialise_request(id, &request.request); + let mut buf = Vec::with_capacity(rq.encoded_len()); + if let Err(e) = rq.encode(&mut buf) { + log::debug!("failed to serialise request {}: {}", id, e); + send_reply(Err(ClientError::RemoteFetchFailed), request.request) + } else { + log::trace!("sending request {} to peer {}", id, peer); + let protocol = OutboundProtocol { + request: buf, + max_data_size: self.config.max_data_size, + protocol: self.config.protocol.clone(), + }; + self.peers.get_mut(&peer).map(|info| info.status = PeerStatus::BusyWith(id)); + let rw = RequestWrapper { + timestamp: request.timestamp, + retries: request.retries, + request: request.request, + peer: peer.clone(), + }; + self.outstanding.insert(id, rw); + return Poll::Ready(NetworkBehaviourAction::SendEvent { peer_id: peer, event: protocol }) + } + } else { + self.pending_requests.push_front(request); + log::debug!("no peer available to send request to"); + break + } + } + + // Look for ongoing requests that have timed out. + let mut expired = Vec::new(); + for (id, rw) in &self.outstanding { + if now > rw.timestamp + self.config.request_timeout { + log::debug!("request {} timed out", id); + expired.push(*id) + } + } + for id in expired { + if let Some(rw) = self.outstanding.remove(&id) { + self.remove_peer(&rw.peer); + self.peerset.report_peer(rw.peer.clone(), + ReputationChange::new(TIMEOUT_REPUTATION_CHANGE, "light request timeout")); + if rw.retries == 0 { + send_reply(Err(ClientError::RemoteFetchFailed), rw.request); + continue + } + let rw = RequestWrapper { + timestamp: Instant::now(), + retries: rw.retries - 1, + request: rw.request, + peer: (), + }; + self.pending_requests.push_back(rw) + } + } + + Poll::Pending + } +} + +fn required_block(request: &Request) -> NumberFor { + match request { + Request::Header { request, .. } => request.block, + Request::Read { request, .. } => *request.header.number(), + Request::ReadChild { request, .. } => *request.header.number(), + Request::Call { request, .. } => *request.header.number(), + Request::Changes { request, .. } => request.max_block.0, + } +} + +fn retries(request: &Request) -> usize { + let rc = match request { + Request::Header { request, .. } => request.retry_count, + Request::Read { request, .. } => request.retry_count, + Request::ReadChild { request, .. } => request.retry_count, + Request::Call { request, .. } => request.retry_count, + Request::Changes { request, .. } => request.retry_count, + }; + rc.unwrap_or(0) +} + +fn serialise_request(id: u64, request: &Request) -> api::v1::light::Request { + let request = match request { + Request::Header { request, .. } => { + let r = api::v1::light::RemoteHeaderRequest { block: request.block.encode() }; + api::v1::light::request::Request::RemoteHeaderRequest(r) + } + Request::Read { request, .. } => { + let r = api::v1::light::RemoteReadRequest { + block: request.block.encode(), + keys: request.keys.clone(), + }; + api::v1::light::request::Request::RemoteReadRequest(r) + } + Request::ReadChild { request, .. } => { + let r = api::v1::light::RemoteReadChildRequest { + block: request.block.encode(), + storage_key: request.storage_key.clone(), + child_type: request.child_type.clone(), + child_info: request.child_info.clone(), + keys: request.keys.clone(), + }; + api::v1::light::request::Request::RemoteReadChildRequest(r) + } + Request::Call { request, .. } => { + let r = api::v1::light::RemoteCallRequest { + block: request.block.encode(), + method: request.method.clone(), + data: request.call_data.clone(), + }; + api::v1::light::request::Request::RemoteCallRequest(r) + } + Request::Changes { request, .. } => { + let r = api::v1::light::RemoteChangesRequest { + first: request.first_block.1.encode(), + last: request.last_block.1.encode(), + min: request.tries_roots.1.encode(), + max: request.max_block.1.encode(), + storage_key: request.storage_key.clone().unwrap_or_default(), + key: request.key.clone(), + }; + api::v1::light::request::Request::RemoteChangesRequest(r) + } + }; + + api::v1::light::Request { id, request: Some(request) } +} + +fn send_reply(result: Result, ClientError>, request: Request) { + fn send(item: T, sender: oneshot::Sender) { + let _ = sender.send(item); // It is okay if the other end already hung up. + } + match request { + Request::Header { request, sender } => match result { + Err(e) => send(Err(e), sender), + Ok(Reply::Header(x)) => send(Ok(x), sender), + reply => log::error!("invalid reply for header request: {:?}, {:?}", reply, request), + } + Request::Read { request, sender } => match result { + Err(e) => send(Err(e), sender), + Ok(Reply::MapVecU8OptVecU8(x)) => send(Ok(x), sender), + reply => log::error!("invalid reply for read request: {:?}, {:?}", reply, request), + } + Request::ReadChild { request, sender } => match result { + Err(e) => send(Err(e), sender), + Ok(Reply::MapVecU8OptVecU8(x)) => send(Ok(x), sender), + reply => log::error!("invalid reply for read child request: {:?}, {:?}", reply, request), + } + Request::Call { request, sender } => match result { + Err(e) => send(Err(e), sender), + Ok(Reply::VecU8(x)) => send(Ok(x), sender), + reply => log::error!("invalid reply for call request: {:?}, {:?}", reply, request), + } + Request::Changes { request, sender } => match result { + Err(e) => send(Err(e), sender), + Ok(Reply::VecNumberU32(x)) => send(Ok(x), sender), + reply => log::error!("invalid reply for changes request: {:?}, {:?}", reply, request), + } + } +} + +/// Output type of inbound and outbound substream upgrades. +#[derive(Debug)] +pub enum Event { + /// Incoming request from remote and substream to use for the response. + Request(api::v1::light::Request, T), + /// Incoming response from remote. + Response(api::v1::light::Response), +} + +/// Substream upgrade protocol. +/// +/// Reads incoming requests from remote. +#[derive(Debug, Clone)] +pub struct InboundProtocol { + /// The max. request length in bytes. + max_data_size: usize, + /// The protocol to use for upgrade negotiation. + protocol: Bytes, +} + +impl UpgradeInfo for InboundProtocol { + type Info = Bytes; + type InfoIter = iter::Once; + + fn protocol_info(&self) -> Self::InfoIter { + iter::once(self.protocol.clone()) + } +} + +impl InboundUpgrade for InboundProtocol +where + T: AsyncRead + AsyncWrite + Unpin + Send + 'static +{ + type Output = Event; + type Error = ReadOneError; + type Future = BoxFuture<'static, Result>; + + fn upgrade_inbound(self, mut s: T, _: Self::Info) -> Self::Future { + let future = async move { + let vec = read_one(&mut s, self.max_data_size).await?; + match api::v1::light::Request::decode(&vec[..]) { + Ok(r) => Ok(Event::Request(r, s)), + Err(e) => Err(ReadOneError::Io(io::Error::new(io::ErrorKind::Other, e))) + } + }; + future.boxed() + } +} + +/// Substream upgrade protocol. +/// +/// Sends a request to remote and awaits the response. +#[derive(Debug, Clone)] +pub struct OutboundProtocol { + /// The serialised protobuf request. + request: Vec, + /// The max. request length in bytes. + max_data_size: usize, + /// The protocol to use for upgrade negotiation. + protocol: Bytes, +} + +impl UpgradeInfo for OutboundProtocol { + type Info = Bytes; + type InfoIter = iter::Once; + + fn protocol_info(&self) -> Self::InfoIter { + iter::once(self.protocol.clone()) + } +} + +impl OutboundUpgrade for OutboundProtocol +where + T: AsyncRead + AsyncWrite + Unpin + Send + 'static +{ + type Output = Event; + type Error = ReadOneError; + type Future = BoxFuture<'static, Result>; + + fn upgrade_outbound(self, mut s: T, _: Self::Info) -> Self::Future { + let future = async move { + write_one(&mut s, &self.request).await?; + let vec = read_one(&mut s, self.max_data_size).await?; + api::v1::light::Response::decode(&vec[..]) + .map(Event::Response) + .map_err(|e| { + ReadOneError::Io(io::Error::new(io::ErrorKind::Other, e)) + }) + }; + future.boxed() + } +} + +fn fmt_keys(first: Option<&Vec>, last: Option<&Vec>) -> String { + if let (Some(first), Some(last)) = (first, last) { + if first == last { + first.to_hex::() + } else { + format!("{}..{}", first.to_hex::(), last.to_hex::()) + } + } else { + String::from("n/a") + } +} + +#[cfg(test)] +mod tests { + use async_std::task; + use assert_matches::assert_matches; + use codec::Encode; + use crate::{ + chain::Client, + config::ProtocolId, + protocol::{api, light_dispatch::tests::{DummyFetchChecker, dummy_header}} + }; + use futures::{channel::oneshot, prelude::*}; + use libp2p::{ + PeerId, + Multiaddr, + core::{ + ConnectedPoint, + identity, + muxing::{StreamMuxerBox, SubstreamRef}, + transport::{Transport, boxed::Boxed, memory::MemoryTransport}, + upgrade + }, + noise::{self, Keypair, X25519, NoiseConfig}, + swarm::{NetworkBehaviour, NetworkBehaviourAction, PollParameters}, + yamux + }; + use sc_client_api::StorageProof; + use sc_client::light::fetcher; + use sp_blockchain::{Error as ClientError}; + use sp_core::storage::ChildInfo; + use std::{ + collections::HashSet, + io, + iter::{self, FromIterator}, + pin::Pin, + sync::Arc, + task::{Context, Poll} + }; + use sp_runtime::{generic::Header, traits::BlakeTwo256}; + use super::{Event, LightClientHandler, Request, OutboundProtocol, PeerStatus}; + use void::Void; + + const CHILD_INFO: ChildInfo<'static> = ChildInfo::new_default(b"foobarbaz"); + + type Block = sp_runtime::generic::Block, substrate_test_runtime::Extrinsic>; + type Handler = LightClientHandler>, Block>; + type Swarm = libp2p::swarm::Swarm, Handler>; + + fn empty_proof() -> Vec { + StorageProof::empty().encode() + } + + fn make_swarm(ok: bool, ps: sc_peerset::PeersetHandle, cf: super::Config) -> Swarm { + let client = Arc::new(substrate_test_runtime_client::new()); + let checker = Arc::new(DummyFetchChecker::new(ok)); + let id_key = identity::Keypair::generate_ed25519(); + let dh_key = Keypair::::new().into_authentic(&id_key).unwrap(); + let local_peer = id_key.public().into_peer_id(); + let transport = MemoryTransport::default() + .upgrade(upgrade::Version::V1) + .authenticate(NoiseConfig::xx(dh_key).into_authenticated()) + .multiplex(yamux::Config::default()) + .map(|(peer, muxer), _| (peer, StreamMuxerBox::new(muxer))) + .map_err(|e| io::Error::new(io::ErrorKind::Other, e)) + .boxed(); + Swarm::new(transport, LightClientHandler::new(cf, client, checker, ps), local_peer) + } + + fn make_config() -> super::Config { + super::Config::new(&ProtocolId::from(&b"foo"[..])) + } + + struct EmptyPollParams(PeerId); + + impl PollParameters for EmptyPollParams { + type SupportedProtocolsIter = iter::Empty>; + type ListenedAddressesIter = iter::Empty; + type ExternalAddressesIter = iter::Empty; + + fn supported_protocols(&self) -> Self::SupportedProtocolsIter { + iter::empty() + } + + fn listened_addresses(&self) -> Self::ListenedAddressesIter { + iter::empty() + } + + fn external_addresses(&self) -> Self::ExternalAddressesIter { + iter::empty() + } + + fn local_peer_id(&self) -> &PeerId { + &self.0 + } + } + + fn peerset() -> (sc_peerset::Peerset, sc_peerset::PeersetHandle) { + let cfg = sc_peerset::PeersetConfig { + in_peers: 128, + out_peers: 128, + bootnodes: Vec::new(), + reserved_only: false, + reserved_nodes: Vec::new(), + }; + sc_peerset::Peerset::from_config(cfg) + } + + fn make_behaviour + ( ok: bool + , ps: sc_peerset::PeersetHandle + , cf: super::Config + ) -> LightClientHandler>, Block> + { + let client = Arc::new(substrate_test_runtime_client::new()); + let checker = Arc::new(DummyFetchChecker::new(ok)); + LightClientHandler::new(cf, client, checker, ps) + } + + fn empty_dialer() -> ConnectedPoint { + ConnectedPoint::Dialer { address: Multiaddr::empty() } + } + + fn poll(mut b: &mut LightClientHandler) -> Poll> + where + T: AsyncRead + AsyncWrite + Unpin + Send + 'static + { + let mut p = EmptyPollParams(PeerId::random()); + match future::poll_fn(|cx| Pin::new(&mut b).poll(cx, &mut p)).now_or_never() { + Some(a) => Poll::Ready(a), + None => Poll::Pending + } + } + + #[test] + fn disconnects_from_peer_if_told() { + let peer = PeerId::random(); + let pset = peerset(); + let mut behaviour = make_behaviour(true, pset.1, make_config()); + + behaviour.inject_connected(peer.clone(), empty_dialer()); + assert_eq!(1, behaviour.peers.len()); + + behaviour.inject_disconnected(&peer, empty_dialer()); + assert_eq!(0, behaviour.peers.len()) + } + + #[test] + fn disconnects_from_peer_if_request_times_out() { + let peer0 = PeerId::random(); + let peer1 = PeerId::random(); + let pset = peerset(); + let mut behaviour = make_behaviour(true, pset.1, make_config()); + + behaviour.inject_connected(peer0.clone(), empty_dialer()); + behaviour.inject_connected(peer1.clone(), empty_dialer()); + + // We now know about two peers. + assert_eq!(HashSet::from_iter(&[peer0.clone(), peer1.clone()]), behaviour.peers.keys().collect::>()); + + // No requests have been made yet. + assert!(behaviour.pending_requests.is_empty()); + assert!(behaviour.outstanding.is_empty()); + + // Issue our first request! + let chan = oneshot::channel(); + let request = fetcher::RemoteCallRequest { + block: Default::default(), + header: dummy_header(), + method: "test".into(), + call_data: vec![], + retry_count: Some(1), + }; + behaviour.request(Request::Call { request, sender: chan.0 }).unwrap(); + assert_eq!(1, behaviour.pending_requests.len()); + + // The behaviour should now attempt to send the request. + assert_matches!(poll(&mut behaviour), Poll::Ready(NetworkBehaviourAction::SendEvent { peer_id, .. }) => { + assert!(peer_id == peer0 || peer_id == peer1) + }); + + // And we should have one busy peer. + assert!({ + let (idle, busy): (Vec<_>, Vec<_>) = + behaviour.peers.iter().partition(|(_, info)| info.status == PeerStatus::Idle); + + idle.len() == 1 && busy.len() == 1 + && (idle[0].0 == &peer0 || busy[0].0 == &peer0) + && (idle[0].0 == &peer1 || busy[0].0 == &peer1) + }); + + // No more pending requests, but one should be outstanding. + assert_eq!(0, behaviour.pending_requests.len()); + assert_eq!(1, behaviour.outstanding.len()); + + // We now set back the timestamp of the outstanding request to make it expire. + let request = behaviour.outstanding.values_mut().next().unwrap(); + request.timestamp -= make_config().request_timeout; + + // Make progress, but do not expect some action. + assert_matches!(poll(&mut behaviour), Poll::Pending); + + // The request should have timed out by now and the corresponding peer be removed. + assert_eq!(1, behaviour.peers.len()); + // Since we asked for one retry, the request should be back in the pending queue. + assert_eq!(1, behaviour.pending_requests.len()); + // No other request should be ongoing. + assert_eq!(0, behaviour.outstanding.len()); + } + + #[test] + fn disconnects_from_peer_on_response_with_wrong_id() { + let peer = PeerId::random(); + let pset = peerset(); + let mut behaviour = make_behaviour(true, pset.1, make_config()); + + behaviour.inject_connected(peer.clone(), empty_dialer()); + assert_eq!(1, behaviour.peers.len()); + + let chan = oneshot::channel(); + let request = fetcher::RemoteCallRequest { + block: Default::default(), + header: dummy_header(), + method: "test".into(), + call_data: vec![], + retry_count: Some(1), + }; + behaviour.request(Request::Call { request, sender: chan.0 }).unwrap(); + + assert_eq!(1, behaviour.pending_requests.len()); + assert_eq!(0, behaviour.outstanding.len()); + poll(&mut behaviour); // Make progress + assert_eq!(0, behaviour.pending_requests.len()); + assert_eq!(1, behaviour.outstanding.len()); + + // Construct response with bogus ID + let response = { + let r = api::v1::light::RemoteCallResponse { proof: empty_proof() }; + api::v1::light::Response { + id: 2365789, + response: Some(api::v1::light::response::Response::RemoteCallResponse(r)), + } + }; + + // Make sure our bogus ID is really not used. + assert!(!behaviour.outstanding.keys().any(|id| id == &response.id)); + + behaviour.inject_node_event(peer.clone(), Event::Response(response)); + assert!(behaviour.peers.is_empty()); + + poll(&mut behaviour); // More progress + + // The request should be back in the pending queue + assert_eq!(1, behaviour.pending_requests.len()); + assert_eq!(0, behaviour.outstanding.len()); + } + + #[test] + fn disconnects_from_peer_on_incorrect_response() { + let peer = PeerId::random(); + let pset = peerset(); + let mut behaviour = make_behaviour(false, pset.1, make_config()); + // ^--- Making sure the response data check fails. + + behaviour.inject_connected(peer.clone(), empty_dialer()); + assert_eq!(1, behaviour.peers.len()); + + let chan = oneshot::channel(); + let request = fetcher::RemoteCallRequest { + block: Default::default(), + header: dummy_header(), + method: "test".into(), + call_data: vec![], + retry_count: Some(1), + }; + behaviour.request(Request::Call { request, sender: chan.0 }).unwrap(); + + assert_eq!(1, behaviour.pending_requests.len()); + assert_eq!(0, behaviour.outstanding.len()); + poll(&mut behaviour); // Make progress + assert_eq!(0, behaviour.pending_requests.len()); + assert_eq!(1, behaviour.outstanding.len()); + + let request_id = *behaviour.outstanding.keys().next().unwrap(); + + let response = { + let r = api::v1::light::RemoteCallResponse { proof: empty_proof() }; + api::v1::light::Response { + id: request_id, + response: Some(api::v1::light::response::Response::RemoteCallResponse(r)), + } + }; + + behaviour.inject_node_event(peer.clone(), Event::Response(response)); + assert!(behaviour.peers.is_empty()); + + poll(&mut behaviour); // More progress + + // The request should be back in the pending queue + assert_eq!(1, behaviour.pending_requests.len()); + assert_eq!(0, behaviour.outstanding.len()); + } + + #[test] + fn disconnects_from_peer_on_unexpected_response() { + let peer = PeerId::random(); + let pset = peerset(); + let mut behaviour = make_behaviour(true, pset.1, make_config()); + + behaviour.inject_connected(peer.clone(), empty_dialer()); + assert_eq!(1, behaviour.peers.len()); + assert_eq!(0, behaviour.pending_requests.len()); + assert_eq!(0, behaviour.outstanding.len()); + + // Some unsolicited response + let response = { + let r = api::v1::light::RemoteCallResponse { proof: empty_proof() }; + api::v1::light::Response { + id: 2347895932, + response: Some(api::v1::light::response::Response::RemoteCallResponse(r)), + } + }; + + behaviour.inject_node_event(peer.clone(), Event::Response(response)); + + assert!(behaviour.peers.is_empty()); + poll(&mut behaviour); + assert_eq!(0, behaviour.pending_requests.len()); + assert_eq!(0, behaviour.outstanding.len()); + } + + #[test] + fn disconnects_from_peer_on_wrong_response_type() { + let peer = PeerId::random(); + let pset = peerset(); + let mut behaviour = make_behaviour(true, pset.1, make_config()); + + behaviour.inject_connected(peer.clone(), empty_dialer()); + assert_eq!(1, behaviour.peers.len()); + + let chan = oneshot::channel(); + let request = fetcher::RemoteCallRequest { + block: Default::default(), + header: dummy_header(), + method: "test".into(), + call_data: vec![], + retry_count: Some(1), + }; + behaviour.request(Request::Call { request, sender: chan.0 }).unwrap(); + + assert_eq!(1, behaviour.pending_requests.len()); + assert_eq!(0, behaviour.outstanding.len()); + poll(&mut behaviour); // Make progress + assert_eq!(0, behaviour.pending_requests.len()); + assert_eq!(1, behaviour.outstanding.len()); + + let request_id = *behaviour.outstanding.keys().next().unwrap(); + + let response = { + let r = api::v1::light::RemoteReadResponse { proof: empty_proof() }; // Not a RemoteCallResponse! + api::v1::light::Response { + id: request_id, + response: Some(api::v1::light::response::Response::RemoteReadResponse(r)), + } + }; + + behaviour.inject_node_event(peer.clone(), Event::Response(response)); + assert!(behaviour.peers.is_empty()); + + poll(&mut behaviour); // More progress + + // The request should be back in the pending queue + assert_eq!(1, behaviour.pending_requests.len()); + assert_eq!(0, behaviour.outstanding.len()); + } + + #[test] + fn receives_remote_failure_after_retry_count_failures() { + let peer1 = PeerId::random(); + let peer2 = PeerId::random(); + let peer3 = PeerId::random(); + let peer4 = PeerId::random(); + let pset = peerset(); + let mut behaviour = make_behaviour(false, pset.1, make_config()); + // ^--- Making sure the response data check fails. + + behaviour.inject_connected(peer1.clone(), empty_dialer()); + behaviour.inject_connected(peer2.clone(), empty_dialer()); + behaviour.inject_connected(peer3.clone(), empty_dialer()); + behaviour.inject_connected(peer4.clone(), empty_dialer()); + assert_eq!(4, behaviour.peers.len()); + + let mut chan = oneshot::channel(); + let request = fetcher::RemoteCallRequest { + block: Default::default(), + header: dummy_header(), + method: "test".into(), + call_data: vec![], + retry_count: Some(3), // Attempt up to three retries. + }; + behaviour.request(Request::Call { request, sender: chan.0 }).unwrap(); + + assert_eq!(1, behaviour.pending_requests.len()); + assert_eq!(0, behaviour.outstanding.len()); + assert_matches!(poll(&mut behaviour), Poll::Ready(NetworkBehaviourAction::SendEvent { .. })); + assert_eq!(0, behaviour.pending_requests.len()); + assert_eq!(1, behaviour.outstanding.len()); + + for _ in 0 .. 3 { + // Construct an invalid response + let request_id = *behaviour.outstanding.keys().next().unwrap(); + let responding_peer = behaviour.outstanding.values().next().unwrap().peer.clone(); + let response = { + let r = api::v1::light::RemoteCallResponse { proof: empty_proof() }; + api::v1::light::Response { + id: request_id, + response: Some(api::v1::light::response::Response::RemoteCallResponse(r)) + } + }; + behaviour.inject_node_event(responding_peer, Event::Response(response.clone())); + assert_matches!(poll(&mut behaviour), Poll::Ready(NetworkBehaviourAction::SendEvent { .. })); + assert_matches!(chan.1.try_recv(), Ok(None)) + } + // Final invalid response + let request_id = *behaviour.outstanding.keys().next().unwrap(); + let responding_peer = behaviour.outstanding.values().next().unwrap().peer.clone(); + let response = { + let r = api::v1::light::RemoteCallResponse { proof: empty_proof() }; + api::v1::light::Response { + id: request_id, + response: Some(api::v1::light::response::Response::RemoteCallResponse(r)), + } + }; + behaviour.inject_node_event(responding_peer, Event::Response(response)); + assert_matches!(poll(&mut behaviour), Poll::Pending); + assert_matches!(chan.1.try_recv(), Ok(Some(Err(ClientError::RemoteFetchFailed)))) + } + + fn issue_request(request: Request) { + let peer = PeerId::random(); + let pset = peerset(); + let mut behaviour = make_behaviour(true, pset.1, make_config()); + + behaviour.inject_connected(peer.clone(), empty_dialer()); + assert_eq!(1, behaviour.peers.len()); + + let response = match request { + Request::Header{..} => { + let r = api::v1::light::RemoteHeaderResponse { + header: dummy_header().encode(), + proof: empty_proof() + }; + api::v1::light::Response { + id: 1, + response: Some(api::v1::light::response::Response::RemoteHeaderResponse(r)), + } + } + Request::Read{..} => { + let r = api::v1::light::RemoteReadResponse { proof: empty_proof() }; + api::v1::light::Response { + id: 1, + response: Some(api::v1::light::response::Response::RemoteReadResponse(r)), + } + } + Request::ReadChild{..} => { + let r = api::v1::light::RemoteReadResponse { proof: empty_proof() }; + api::v1::light::Response { + id: 1, + response: Some(api::v1::light::response::Response::RemoteReadResponse(r)), + } + } + Request::Call{..} => { + let r = api::v1::light::RemoteCallResponse { proof: empty_proof() }; + api::v1::light::Response { + id: 1, + response: Some(api::v1::light::response::Response::RemoteCallResponse(r)), + } + } + Request::Changes{..} => { + let r = api::v1::light::RemoteChangesResponse { + max: iter::repeat(1).take(32).collect(), + proof: Vec::new(), + roots: Vec::new(), + roots_proof: empty_proof() + }; + api::v1::light::Response { + id: 1, + response: Some(api::v1::light::response::Response::RemoteChangesResponse(r)), + } + } + }; + + behaviour.request(request).unwrap(); + + assert_eq!(1, behaviour.pending_requests.len()); + assert_eq!(0, behaviour.outstanding.len()); + assert_matches!(poll(&mut behaviour), Poll::Ready(NetworkBehaviourAction::SendEvent { .. })); + assert_eq!(0, behaviour.pending_requests.len()); + assert_eq!(1, behaviour.outstanding.len()); + assert_eq!(1, *behaviour.outstanding.keys().next().unwrap()); + + behaviour.inject_node_event(peer.clone(), Event::Response(response)); + + poll(&mut behaviour); + + assert_eq!(0, behaviour.pending_requests.len()); + assert_eq!(0, behaviour.outstanding.len()) + } + + #[test] + fn receives_remote_call_response() { + let mut chan = oneshot::channel(); + let request = fetcher::RemoteCallRequest { + block: Default::default(), + header: dummy_header(), + method: "test".into(), + call_data: vec![], + retry_count: None, + }; + issue_request(Request::Call { request, sender: chan.0 }); + assert_matches!(chan.1.try_recv(), Ok(Some(Ok(_)))) + } + + #[test] + fn receives_remote_read_response() { + let mut chan = oneshot::channel(); + let request = fetcher::RemoteReadRequest { + header: dummy_header(), + block: Default::default(), + keys: vec![b":key".to_vec()], + retry_count: None, + }; + issue_request(Request::Read { request, sender: chan.0 }); + assert_matches!(chan.1.try_recv(), Ok(Some(Ok(_)))) + } + + #[test] + fn receives_remote_read_child_response() { + let info = CHILD_INFO.info(); + let mut chan = oneshot::channel(); + let request = fetcher::RemoteReadChildRequest { + header: dummy_header(), + block: Default::default(), + storage_key: b":child_storage:sub".to_vec(), + keys: vec![b":key".to_vec()], + child_info: info.0.to_vec(), + child_type: info.1, + retry_count: None, + }; + issue_request(Request::ReadChild { request, sender: chan.0 }); + assert_matches!(chan.1.try_recv(), Ok(Some(Ok(_)))) + } + + #[test] + fn receives_remote_header_response() { + let mut chan = oneshot::channel(); + let request = fetcher::RemoteHeaderRequest { + cht_root: Default::default(), + block: 1, + retry_count: None, + }; + issue_request(Request::Header { request, sender: chan.0 }); + assert_matches!(chan.1.try_recv(), Ok(Some(Ok(_)))) + } + + #[test] + fn receives_remote_changes_response() { + let mut chan = oneshot::channel(); + let request = fetcher::RemoteChangesRequest { + changes_trie_configs: vec![sp_core::ChangesTrieConfigurationRange { + zero: (0, Default::default()), + end: None, + config: Some(sp_core::ChangesTrieConfiguration::new(4, 2)), + }], + first_block: (1, Default::default()), + last_block: (100, Default::default()), + max_block: (100, Default::default()), + tries_roots: (1, Default::default(), Vec::new()), + key: Vec::new(), + storage_key: None, + retry_count: None, + }; + issue_request(Request::Changes { request, sender: chan.0 }); + assert_matches!(chan.1.try_recv(), Ok(Some(Ok(_)))) + } + + fn send_receive(request: Request) { + // We start a swarm on the listening side which awaits incoming requests and answers them: + let local_pset = peerset(); + let local_listen_addr: libp2p::Multiaddr = libp2p::multiaddr::Protocol::Memory(rand::random()).into(); + let mut local_swarm = make_swarm(true, local_pset.1, make_config()); + Swarm::listen_on(&mut local_swarm, local_listen_addr.clone()).unwrap(); + + // We also start a swarm that makes requests and awaits responses: + let remote_pset = peerset(); + let mut remote_swarm = make_swarm(true, remote_pset.1, make_config()); + + // We now schedule a request, dial the remote and let the two swarm work it out: + remote_swarm.request(request).unwrap(); + Swarm::dial_addr(&mut remote_swarm, local_listen_addr).unwrap(); + + let future = { + let a = local_swarm.for_each(|_| future::ready(())); + let b = remote_swarm.for_each(|_| future::ready(())); + future::join(a, b).map(|_| ()) + }; + + task::spawn(future); + } + + #[test] + fn send_receive_call() { + let chan = oneshot::channel(); + let request = fetcher::RemoteCallRequest { + block: Default::default(), + header: dummy_header(), + method: "test".into(), + call_data: vec![], + retry_count: None, + }; + send_receive(Request::Call { request, sender: chan.0 }); + assert_eq!(vec![42], task::block_on(chan.1).unwrap().unwrap()); + // ^--- from `DummyFetchChecker::check_execution_proof` + } + + #[test] + fn send_receive_read() { + let chan = oneshot::channel(); + let request = fetcher::RemoteReadRequest { + header: dummy_header(), + block: Default::default(), + keys: vec![b":key".to_vec()], + retry_count: None + }; + send_receive(Request::Read { request, sender: chan.0 }); + assert_eq!(Some(vec![42]), task::block_on(chan.1).unwrap().unwrap().remove(&b":key"[..]).unwrap()); + // ^--- from `DummyFetchChecker::check_read_proof` + } + + #[test] + fn send_receive_read_child() { + let info = CHILD_INFO.info(); + let chan = oneshot::channel(); + let request = fetcher::RemoteReadChildRequest { + header: dummy_header(), + block: Default::default(), + storage_key: b":child_storage:sub".to_vec(), + keys: vec![b":key".to_vec()], + child_info: info.0.to_vec(), + child_type: info.1, + retry_count: None, + }; + send_receive(Request::ReadChild { request, sender: chan.0 }); + assert_eq!(Some(vec![42]), task::block_on(chan.1).unwrap().unwrap().remove(&b":key"[..]).unwrap()); + // ^--- from `DummyFetchChecker::check_read_child_proof` + } + + #[test] + fn send_receive_header() { + let _ = env_logger::try_init(); + let chan = oneshot::channel(); + let request = fetcher::RemoteHeaderRequest { + cht_root: Default::default(), + block: 1, + retry_count: None, + }; + send_receive(Request::Header { request, sender: chan.0 }); + // The remote does not know block 1: + assert_matches!(task::block_on(chan.1).unwrap(), Err(ClientError::RemoteFetchFailed)); + } + + #[test] + fn send_receive_changes() { + let chan = oneshot::channel(); + let request = fetcher::RemoteChangesRequest { + changes_trie_configs: vec![sp_core::ChangesTrieConfigurationRange { + zero: (0, Default::default()), + end: None, + config: Some(sp_core::ChangesTrieConfiguration::new(4, 2)), + }], + first_block: (1, Default::default()), + last_block: (100, Default::default()), + max_block: (100, Default::default()), + tries_roots: (1, Default::default(), Vec::new()), + key: Vec::new(), + storage_key: None, + retry_count: None, + }; + send_receive(Request::Changes { request, sender: chan.0 }); + assert_eq!(vec![(100, 2)], task::block_on(chan.1).unwrap().unwrap()); + // ^--- from `DummyFetchChecker::check_changes_proof` + } +} diff --git a/client/network/src/protocol/light_dispatch.rs b/client/network/src/protocol/light_dispatch.rs index 5b587d2045d9f..2eab0a5a3ffe0 100644 --- a/client/network/src/protocol/light_dispatch.rs +++ b/client/network/src/protocol/light_dispatch.rs @@ -41,7 +41,7 @@ const REQUEST_TIMEOUT: Duration = Duration::from_secs(15); /// Default request retry count. const RETRY_COUNT: usize = 1; /// Reputation change for a peer when a request timed out. -const TIMEOUT_REPUTATION_CHANGE: i32 = -(1 << 8); +pub(crate) const TIMEOUT_REPUTATION_CHANGE: i32 = -(1 << 8); /// Trait used by the `LightDispatch` service to communicate messages back to the network. pub trait LightDispatchNetwork { @@ -692,17 +692,26 @@ pub mod tests { use crate::message::{self, BlockAttributes, Direction, FromBlock, RequestId}; use libp2p::PeerId; use super::{REQUEST_TIMEOUT, LightDispatch, LightDispatchNetwork, RequestData, StorageProof}; - use sp_test_primitives::{Block, Extrinsic, Header}; + use sp_test_primitives::{Block, Header}; - struct DummyFetchChecker { ok: bool } + pub(crate) struct DummyFetchChecker { + pub(crate) ok: bool, + _mark: std::marker::PhantomData + } + + impl DummyFetchChecker { + pub(crate) fn new(ok: bool) -> Self { + DummyFetchChecker { ok, _mark: std::marker::PhantomData } + } + } - impl FetchChecker for DummyFetchChecker { + impl FetchChecker for DummyFetchChecker { fn check_header_proof( &self, - _request: &RemoteHeaderRequest
, - header: Option
, + _request: &RemoteHeaderRequest, + header: Option, _remote_proof: StorageProof, - ) -> ClientResult
{ + ) -> ClientResult { match self.ok { true if header.is_some() => Ok(header.unwrap()), _ => Err(ClientError::Backend("Test error".into())), @@ -711,7 +720,7 @@ pub mod tests { fn check_read_proof( &self, - request: &RemoteReadRequest
, + request: &RemoteReadRequest, _: StorageProof, ) -> ClientResult, Option>>> { match self.ok { @@ -727,7 +736,7 @@ pub mod tests { fn check_read_child_proof( &self, - request: &RemoteReadChildRequest
, + request: &RemoteReadChildRequest, _: StorageProof, ) -> ClientResult, Option>>> { match self.ok { @@ -741,7 +750,7 @@ pub mod tests { } } - fn check_execution_proof(&self, _: &RemoteCallRequest
, _: StorageProof) -> ClientResult> { + fn check_execution_proof(&self, _: &RemoteCallRequest, _: StorageProof) -> ClientResult> { match self.ok { true => Ok(vec![42]), false => Err(ClientError::Backend("Test error".into())), @@ -750,20 +759,20 @@ pub mod tests { fn check_changes_proof( &self, - _: &RemoteChangesRequest
, - _: ChangesProof
- ) -> ClientResult, u32)>> { + _: &RemoteChangesRequest, + _: ChangesProof + ) -> ClientResult, u32)>> { match self.ok { - true => Ok(vec![(100, 2)]), + true => Ok(vec![(100.into(), 2)]), false => Err(ClientError::Backend("Test error".into())), } } fn check_body_proof( &self, - _: &RemoteBodyRequest
, - body: Vec - ) -> ClientResult> { + _: &RemoteBodyRequest, + body: Vec + ) -> ClientResult> { match self.ok { true => Ok(body), false => Err(ClientError::Backend("Test error".into())), @@ -772,7 +781,7 @@ pub mod tests { } fn dummy(ok: bool) -> LightDispatch { - LightDispatch::new(Arc::new(DummyFetchChecker { ok })) + LightDispatch::new(Arc::new(DummyFetchChecker::new(ok))) } fn total_peers(light_dispatch: &LightDispatch) -> usize { @@ -791,7 +800,7 @@ pub mod tests { }); } - fn dummy_header() -> Header { + pub(crate) fn dummy_header() -> Header { Header { parent_hash: Default::default(), number: 0, diff --git a/client/network/src/protocol/schema/api.v1.proto b/client/network/src/protocol/schema/api.v1.proto new file mode 100644 index 0000000000000..e4c32ec585d9c --- /dev/null +++ b/client/network/src/protocol/schema/api.v1.proto @@ -0,0 +1,59 @@ +// Schema definition for block request/response messages. + +syntax = "proto3"; + +package api.v1; + +// Block enumeration direction. +enum Direction { + // Enumerate in ascending order (from child to parent). + Ascending = 0; + // Enumerate in descendfing order (from parent to canonical child). + Descending = 1; +} + +// Request block data from a peer. +message BlockRequest { + // Unique request id. + uint64 id = 1; + // Bits of block data to request. + uint32 fields = 2; + // Start from this block. + oneof from_block { + // Start with given hash. + bytes hash = 3; + // Start with given block number. + bytes number = 4; + } + // End at this block. An implementation defined maximum is used when unspecified. + bytes to_block = 5; // optional + // Sequence direction. + Direction direction = 6; + // Maximum number of blocks to return. An implementation defined maximum is used when unspecified. + uint32 max_blocks = 7; // optional +} + +// Response to `BlockRequest` +message BlockResponse { + // Id of a request this response was made for. + uint64 id = 1; + // Block data for the requested sequence. + repeated BlockData blocks = 2; +} + +// Block data sent in the response. +message BlockData { + // Block header hash. + bytes hash = 1; + // Block header if requested. + bytes header = 2; // optional + // Block body if requested. + repeated bytes body = 3; // optional + // Block receipt if requested. + bytes receipt = 4; // optional + // Block message queue if requested. + bytes message_queue = 5; // optional + // Justification if requested. + bytes justification = 6; // optional +} + diff --git a/client/network/src/protocol/schema/light.v1.proto b/client/network/src/protocol/schema/light.v1.proto new file mode 100644 index 0000000000000..b9aee67b5ee24 --- /dev/null +++ b/client/network/src/protocol/schema/light.v1.proto @@ -0,0 +1,128 @@ +// Schema definition for light client messages. + +syntax = "proto3"; + +package api.v1.light; + +// A pair of arbitrary bytes. +message Pair { + // The first element of the pair. + bytes fst = 1; + // The second element of the pair. + bytes snd = 2; +} + +// Enumerate all possible light client request messages. +message Request { + // Unique request id. + uint64 id = 1; + oneof request { + RemoteCallRequest remote_call_request = 2; + RemoteReadRequest remote_read_request = 3; + RemoteHeaderRequest remote_header_request = 4; + RemoteReadChildRequest remote_read_child_request = 5; + RemoteChangesRequest remote_changes_request = 6; + } +} + +// Enumerate all possible light client response messages. +message Response { + /// Id of a request this response was made for. + uint64 id = 1; + oneof response { + RemoteCallResponse remote_call_response = 2; + RemoteReadResponse remote_read_response = 3; + RemoteHeaderResponse remote_header_response = 4; + RemoteChangesResponse remote_changes_response = 6; + } +} + +// Remote call request. +message RemoteCallRequest { + // Block at which to perform call. + bytes block = 2; + // Method name. + string method = 3; + // Call data. + bytes data = 4; +} + +// Remote call response. +message RemoteCallResponse { + // Execution proof. + bytes proof = 2; +} + +// Remote storage read request. +message RemoteReadRequest { + // Block at which to perform call. + bytes block = 2; + // Storage keys. + repeated bytes keys = 3; +} + +// Remote read response. +message RemoteReadResponse { + // Read proof. + bytes proof = 2; +} + +// Remote storage read child request. +message RemoteReadChildRequest { + // Block at which to perform call. + bytes block = 2; + // Child Storage key. + bytes storage_key = 3; + // Child trie source information. + bytes child_info = 4; + /// Child type, its required to resolve `child_info` + /// content and choose child implementation. + uint32 child_type = 5; + // Storage keys. + repeated bytes keys = 6; +} + +// Remote header request. +message RemoteHeaderRequest { + // Block number to request header for. + bytes block = 2; +} + +// Remote header response. +message RemoteHeaderResponse { + // Header. None if proof generation has failed (e.g. header is unknown). + bytes header = 2; // optional + // Header proof. + bytes proof = 3; +} + +/// Remote changes request. +message RemoteChangesRequest { + // Hash of the first block of the range (including first) where changes are requested. + bytes first = 2; + // Hash of the last block of the range (including last) where changes are requested. + bytes last = 3; + // Hash of the first block for which the requester has the changes trie root. All other + // affected roots must be proved. + bytes min = 4; + // Hash of the last block that we can use when querying changes. + bytes max = 5; + // Storage child node key which changes are requested. + bytes storage_key = 6; // optional + // Storage key which changes are requested. + bytes key = 7; +} + +// Remote changes response. +message RemoteChangesResponse { + // Proof has been generated using block with this number as a max block. Should be + // less than or equal to the RemoteChangesRequest::max block number. + bytes max = 2; + // Changes proof. + repeated bytes proof = 3; + // Changes tries roots missing on the requester' node. + repeated Pair roots = 4; + // Missing changes tries roots proof. + bytes roots_proof = 5; +} + diff --git a/client/network/src/service.rs b/client/network/src/service.rs index e3eaf6c31d566..b37d9afcbee64 100644 --- a/client/network/src/service.rs +++ b/client/network/src/service.rs @@ -193,6 +193,10 @@ impl, H: ExHashT> NetworkWorker let local_peer_id = local_public.clone().into_peer_id(); info!(target: "sub-libp2p", "Local node identity is: {}", local_peer_id.to_base58()); + let checker = params.on_demand.as_ref() + .map(|od| od.checker().clone()) + .unwrap_or(Arc::new(AlwaysBadChecker)); + let num_connected = Arc::new(AtomicUsize::new(0)); let is_major_syncing = Arc::new(AtomicBool::new(false)); let (protocol, peerset_handle) = Protocol::new( @@ -200,14 +204,13 @@ impl, H: ExHashT> NetworkWorker roles: params.roles, max_parallel_downloads: params.network_config.max_parallel_downloads, }, - params.chain, - params.on_demand.as_ref().map(|od| od.checker().clone()) - .unwrap_or(Arc::new(AlwaysBadChecker)), + params.chain.clone(), + checker.clone(), params.specialization, params.transaction_pool, - params.finality_proof_provider, + params.finality_proof_provider.clone(), params.finality_proof_request_builder, - params.protocol_id, + params.protocol_id.clone(), peerset_config, params.block_announce_validator )?; @@ -219,6 +222,14 @@ impl, H: ExHashT> NetworkWorker params.network_config.client_version, params.network_config.node_name ); + let block_requests = { + let config = protocol::block_requests::Config::new(¶ms.protocol_id); + protocol::BlockRequests::new(config, params.chain.clone()) + }; + let light_client_handler = { + let config = protocol::light_client_handler::Config::new(¶ms.protocol_id); + protocol::LightClientHandler::new(config, params.chain, checker, peerset_handle.clone()) + }; let behaviour = futures::executor::block_on(Behaviour::new( protocol, user_agent, @@ -232,6 +243,8 @@ impl, H: ExHashT> NetworkWorker TransportConfig::MemoryOnly => false, TransportConfig::Normal { allow_private_ipv4, .. } => allow_private_ipv4, }, + block_requests, + light_client_handler )); let (transport, bandwidth) = { let (config_mem, config_wasm) = match params.network_config.transport { From cb038dd49c451d624525df546db6b8abec820fd1 Mon Sep 17 00:00:00 2001 From: Toralf Wittner Date: Mon, 10 Feb 2020 16:57:58 +0100 Subject: [PATCH 2/3] Uncomment `Behaviour::light_client_request`. --- client/network/src/behaviour.rs | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/client/network/src/behaviour.rs b/client/network/src/behaviour.rs index 03634ce103d36..facbe1417b473 100644 --- a/client/network/src/behaviour.rs +++ b/client/network/src/behaviour.rs @@ -19,7 +19,7 @@ use crate::{ Event, protocol::event::DhtEvent }; use crate::{ExHashT, specialization::NetworkSpecialization}; -use crate::protocol::{self, CustomMessageOutcome, Protocol}; +use crate::protocol::{self, light_client_handler, CustomMessageOutcome, Protocol}; use libp2p::NetworkBehaviour; use libp2p::core::{Multiaddr, PeerId, PublicKey}; use libp2p::kad::record; @@ -125,10 +125,11 @@ impl, H: ExHashT> Behaviour { self.discovery.put_value(key, value); } -// /// Issue a light client request. -// pub fn light_client_request(&mut self, r: light_client_handler::Request) -> Result<(), light_client_handler::Error> { -// self.light_client_handler.request(r) -// } + /// Issue a light client request. + #[allow(unused)] + pub fn light_client_request(&mut self, r: light_client_handler::Request) -> Result<(), light_client_handler::Error> { + self.light_client_handler.request(r) + } } impl, H: ExHashT> NetworkBehaviourEventProcess for From 79a65a04aa0dc536c1e0b8cb05c63d161f62927b Mon Sep 17 00:00:00 2001 From: Toralf Wittner Date: Tue, 11 Feb 2020 17:14:51 +0100 Subject: [PATCH 3/3] Add license headers. --- client/network/src/protocol/block_requests.rs | 16 ++++++++++++++++ .../network/src/protocol/light_client_handler.rs | 16 ++++++++++++++++ 2 files changed, 32 insertions(+) diff --git a/client/network/src/protocol/block_requests.rs b/client/network/src/protocol/block_requests.rs index c67d096c78ec1..f8a905c288b9d 100644 --- a/client/network/src/protocol/block_requests.rs +++ b/client/network/src/protocol/block_requests.rs @@ -1,3 +1,19 @@ +// Copyright 2020 Parity Technologies (UK) Ltd. +// This file is part of Substrate. +// +// Substrate 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. +// +// Substrate 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 Substrate. If not, see . + //! `NetworkBehaviour` implementation which handles incoming block requests. //! //! Every request is coming in on a separate connection substream which gets diff --git a/client/network/src/protocol/light_client_handler.rs b/client/network/src/protocol/light_client_handler.rs index 117c55cd3c9bb..f5be23c0d4d49 100644 --- a/client/network/src/protocol/light_client_handler.rs +++ b/client/network/src/protocol/light_client_handler.rs @@ -1,3 +1,19 @@ +// Copyright 2020 Parity Technologies (UK) Ltd. +// This file is part of Substrate. +// +// Substrate 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. +// +// Substrate 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 Substrate. If not, see . + //! [`NetworkBehaviour`] implementation which handles light client requests. //! //! Every request is coming in on a separate connection substream which gets