mirror of
https://github.com/pezkuwichain/pezkuwi-subxt.git
synced 2026-05-29 21:41:03 +00:00
Kill the light client, CHTs and change tries. (#10080)
* Remove light client, change tries and CHTs * Update tests * fmt * Restore changes_root * Fixed benches * Cargo fmt * fmt * fmt
This commit is contained in:
Arkadiy Paronyan
@@ -20,14 +20,14 @@ use crate::{
|
||||
bitswap::Bitswap,
|
||||
config::ProtocolId,
|
||||
discovery::{DiscoveryBehaviour, DiscoveryConfig, DiscoveryOut},
|
||||
light_client_requests, peer_info,
|
||||
peer_info,
|
||||
protocol::{message::Roles, CustomMessageOutcome, NotificationsSink, Protocol},
|
||||
request_responses, DhtEvent, ObservedRole,
|
||||
};
|
||||
|
||||
use bytes::Bytes;
|
||||
use codec::Encode;
|
||||
use futures::{channel::oneshot, stream::StreamExt};
|
||||
use futures::channel::oneshot;
|
||||
use libp2p::{
|
||||
core::{Multiaddr, PeerId, PublicKey},
|
||||
identify::IdentifyInfo,
|
||||
@@ -76,10 +76,6 @@ pub struct Behaviour<B: BlockT> {
|
||||
#[behaviour(ignore)]
|
||||
events: VecDeque<BehaviourOut<B>>,
|
||||
|
||||
/// Light client request handling.
|
||||
#[behaviour(ignore)]
|
||||
light_client_request_sender: light_client_requests::sender::LightClientRequestSender<B>,
|
||||
|
||||
/// Protocol name used to send out block requests via
|
||||
/// [`request_responses::RequestResponsesBehaviour`].
|
||||
#[behaviour(ignore)]
|
||||
@@ -198,7 +194,6 @@ impl<B: BlockT> Behaviour<B> {
|
||||
substrate: Protocol<B>,
|
||||
user_agent: String,
|
||||
local_public_key: PublicKey,
|
||||
light_client_request_sender: light_client_requests::sender::LightClientRequestSender<B>,
|
||||
disco_config: DiscoveryConfig,
|
||||
block_request_protocol_config: request_responses::ProtocolConfig,
|
||||
state_request_protocol_config: request_responses::ProtocolConfig,
|
||||
@@ -233,7 +228,6 @@ impl<B: BlockT> Behaviour<B> {
|
||||
request_response_protocols.into_iter(),
|
||||
peerset,
|
||||
)?,
|
||||
light_client_request_sender,
|
||||
events: VecDeque::new(),
|
||||
block_request_protocol_name,
|
||||
state_request_protocol_name,
|
||||
@@ -316,14 +310,6 @@ impl<B: BlockT> Behaviour<B> {
|
||||
pub fn put_value(&mut self, key: record::Key, value: Vec<u8>) {
|
||||
self.discovery.put_value(key, value);
|
||||
}
|
||||
|
||||
/// Issue a light client request.
|
||||
pub fn light_client_request(
|
||||
&mut self,
|
||||
r: light_client_requests::sender::Request<B>,
|
||||
) -> Result<(), light_client_requests::sender::SendRequestError> {
|
||||
self.light_client_request_sender.request(r)
|
||||
}
|
||||
}
|
||||
|
||||
fn reported_roles_to_observed_role(roles: Roles) -> ObservedRole {
|
||||
@@ -436,17 +422,11 @@ impl<B: BlockT> NetworkBehaviourEventProcess<CustomMessageOutcome<B>> for Behavi
|
||||
CustomMessageOutcome::NotificationsReceived { remote, messages } => {
|
||||
self.events.push_back(BehaviourOut::NotificationsReceived { remote, messages });
|
||||
},
|
||||
CustomMessageOutcome::PeerNewBest(peer_id, number) => {
|
||||
self.light_client_request_sender.update_best_block(&peer_id, number);
|
||||
},
|
||||
CustomMessageOutcome::SyncConnected(peer_id) => {
|
||||
self.light_client_request_sender.inject_connected(peer_id);
|
||||
self.events.push_back(BehaviourOut::SyncConnected(peer_id))
|
||||
},
|
||||
CustomMessageOutcome::SyncDisconnected(peer_id) => {
|
||||
self.light_client_request_sender.inject_disconnected(peer_id);
|
||||
self.events.push_back(BehaviourOut::SyncDisconnected(peer_id))
|
||||
},
|
||||
CustomMessageOutcome::PeerNewBest(_peer_id, _number) => {},
|
||||
CustomMessageOutcome::SyncConnected(peer_id) =>
|
||||
self.events.push_back(BehaviourOut::SyncConnected(peer_id)),
|
||||
CustomMessageOutcome::SyncDisconnected(peer_id) =>
|
||||
self.events.push_back(BehaviourOut::SyncDisconnected(peer_id)),
|
||||
CustomMessageOutcome::None => {},
|
||||
}
|
||||
}
|
||||
@@ -534,23 +514,9 @@ impl<B: BlockT> NetworkBehaviourEventProcess<DiscoveryOut> for Behaviour<B> {
|
||||
impl<B: BlockT> Behaviour<B> {
|
||||
fn poll<TEv>(
|
||||
&mut self,
|
||||
cx: &mut Context,
|
||||
_cx: &mut Context,
|
||||
_: &mut impl PollParameters,
|
||||
) -> Poll<NetworkBehaviourAction<TEv, BehaviourOut<B>>> {
|
||||
use light_client_requests::sender::OutEvent;
|
||||
while let Poll::Ready(Some(event)) = self.light_client_request_sender.poll_next_unpin(cx) {
|
||||
match event {
|
||||
OutEvent::SendRequest { target, request, pending_response, protocol_name } =>
|
||||
self.request_responses.send_request(
|
||||
&target,
|
||||
&protocol_name,
|
||||
request,
|
||||
pending_response,
|
||||
IfDisconnected::ImmediateError,
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(event) = self.events.pop_front() {
|
||||
return Poll::Ready(NetworkBehaviourAction::GenerateEvent(event))
|
||||
}
|
||||
|
||||
@@ -23,7 +23,6 @@
|
||||
|
||||
pub use crate::{
|
||||
chain::Client,
|
||||
on_demand_layer::{AlwaysBadChecker, OnDemand},
|
||||
request_responses::{
|
||||
IncomingRequest, OutgoingResponse, ProtocolConfig as RequestResponseConfig,
|
||||
},
|
||||
@@ -83,11 +82,6 @@ pub struct Params<B: BlockT, H: ExHashT> {
|
||||
/// Client that contains the blockchain.
|
||||
pub chain: Arc<dyn Client<B>>,
|
||||
|
||||
/// The `OnDemand` object acts as a "receiver" for block data requests from the client.
|
||||
/// If `Some`, the network worker will process these requests and answer them.
|
||||
/// Normally used only for light clients.
|
||||
pub on_demand: Option<Arc<OnDemand<B>>>,
|
||||
|
||||
/// Pool of transactions.
|
||||
///
|
||||
/// The network worker will fetch transactions from this object in order to propagate them on
|
||||
|
||||
@@ -247,7 +247,6 @@
|
||||
mod behaviour;
|
||||
mod chain;
|
||||
mod discovery;
|
||||
mod on_demand_layer;
|
||||
mod peer_info;
|
||||
mod protocol;
|
||||
mod request_responses;
|
||||
|
||||
@@ -20,8 +20,6 @@
|
||||
|
||||
/// For incoming light client requests.
|
||||
pub mod handler;
|
||||
/// For outgoing light client requests.
|
||||
pub mod sender;
|
||||
|
||||
use crate::{config::ProtocolId, request_responses::ProtocolConfig};
|
||||
|
||||
@@ -47,269 +45,3 @@ pub fn generate_protocol_config(protocol_id: &ProtocolId) -> ProtocolConfig {
|
||||
inbound_queue: None,
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::{config::ProtocolId, request_responses::IncomingRequest};
|
||||
|
||||
use assert_matches::assert_matches;
|
||||
use futures::{
|
||||
channel::oneshot,
|
||||
executor::{block_on, LocalPool},
|
||||
prelude::*,
|
||||
task::Spawn,
|
||||
};
|
||||
use libp2p::PeerId;
|
||||
use sc_client_api::{
|
||||
light::{
|
||||
self, ChangesProof, RemoteBodyRequest, RemoteCallRequest, RemoteChangesRequest,
|
||||
RemoteHeaderRequest, RemoteReadRequest,
|
||||
},
|
||||
FetchChecker, RemoteReadChildRequest, StorageProof,
|
||||
};
|
||||
use sp_blockchain::Error as ClientError;
|
||||
use sp_core::storage::ChildInfo;
|
||||
use sp_runtime::{
|
||||
generic::Header,
|
||||
traits::{BlakeTwo256, Block as BlockT, NumberFor},
|
||||
};
|
||||
use std::{collections::HashMap, sync::Arc};
|
||||
|
||||
pub struct DummyFetchChecker<B> {
|
||||
pub ok: bool,
|
||||
pub _mark: std::marker::PhantomData<B>,
|
||||
}
|
||||
|
||||
impl<B: BlockT> FetchChecker<B> for DummyFetchChecker<B> {
|
||||
fn check_header_proof(
|
||||
&self,
|
||||
_request: &RemoteHeaderRequest<B::Header>,
|
||||
header: Option<B::Header>,
|
||||
_remote_proof: StorageProof,
|
||||
) -> Result<B::Header, ClientError> {
|
||||
match self.ok {
|
||||
true if header.is_some() => Ok(header.unwrap()),
|
||||
_ => Err(ClientError::Backend("Test error".into())),
|
||||
}
|
||||
}
|
||||
|
||||
fn check_read_proof(
|
||||
&self,
|
||||
request: &RemoteReadRequest<B::Header>,
|
||||
_: StorageProof,
|
||||
) -> Result<HashMap<Vec<u8>, Option<Vec<u8>>>, ClientError> {
|
||||
match self.ok {
|
||||
true => Ok(request.keys.iter().cloned().map(|k| (k, Some(vec![42]))).collect()),
|
||||
false => Err(ClientError::Backend("Test error".into())),
|
||||
}
|
||||
}
|
||||
|
||||
fn check_read_child_proof(
|
||||
&self,
|
||||
request: &RemoteReadChildRequest<B::Header>,
|
||||
_: StorageProof,
|
||||
) -> Result<HashMap<Vec<u8>, Option<Vec<u8>>>, ClientError> {
|
||||
match self.ok {
|
||||
true => Ok(request.keys.iter().cloned().map(|k| (k, Some(vec![42]))).collect()),
|
||||
false => Err(ClientError::Backend("Test error".into())),
|
||||
}
|
||||
}
|
||||
|
||||
fn check_execution_proof(
|
||||
&self,
|
||||
_: &RemoteCallRequest<B::Header>,
|
||||
_: StorageProof,
|
||||
) -> Result<Vec<u8>, ClientError> {
|
||||
match self.ok {
|
||||
true => Ok(vec![42]),
|
||||
false => Err(ClientError::Backend("Test error".into())),
|
||||
}
|
||||
}
|
||||
|
||||
fn check_changes_proof(
|
||||
&self,
|
||||
_: &RemoteChangesRequest<B::Header>,
|
||||
_: ChangesProof<B::Header>,
|
||||
) -> Result<Vec<(NumberFor<B>, u32)>, ClientError> {
|
||||
match self.ok {
|
||||
true => Ok(vec![(100u32.into(), 2)]),
|
||||
false => Err(ClientError::Backend("Test error".into())),
|
||||
}
|
||||
}
|
||||
|
||||
fn check_body_proof(
|
||||
&self,
|
||||
_: &RemoteBodyRequest<B::Header>,
|
||||
body: Vec<B::Extrinsic>,
|
||||
) -> Result<Vec<B::Extrinsic>, ClientError> {
|
||||
match self.ok {
|
||||
true => Ok(body),
|
||||
false => Err(ClientError::Backend("Test error".into())),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub fn protocol_id() -> ProtocolId {
|
||||
ProtocolId::from("test")
|
||||
}
|
||||
|
||||
pub fn peerset() -> (sc_peerset::Peerset, sc_peerset::PeersetHandle) {
|
||||
let cfg = sc_peerset::SetConfig {
|
||||
in_peers: 128,
|
||||
out_peers: 128,
|
||||
bootnodes: Default::default(),
|
||||
reserved_only: false,
|
||||
reserved_nodes: Default::default(),
|
||||
};
|
||||
sc_peerset::Peerset::from_config(sc_peerset::PeersetConfig { sets: vec![cfg] })
|
||||
}
|
||||
|
||||
pub fn dummy_header() -> sp_test_primitives::Header {
|
||||
sp_test_primitives::Header {
|
||||
parent_hash: Default::default(),
|
||||
number: 0,
|
||||
state_root: Default::default(),
|
||||
extrinsics_root: Default::default(),
|
||||
digest: Default::default(),
|
||||
}
|
||||
}
|
||||
|
||||
type Block =
|
||||
sp_runtime::generic::Block<Header<u64, BlakeTwo256>, substrate_test_runtime::Extrinsic>;
|
||||
|
||||
fn send_receive(request: sender::Request<Block>, pool: &LocalPool) {
|
||||
let client = Arc::new(substrate_test_runtime_client::new());
|
||||
let (handler, protocol_config) =
|
||||
handler::LightClientRequestHandler::new(&protocol_id(), client);
|
||||
pool.spawner().spawn_obj(handler.run().boxed().into()).unwrap();
|
||||
|
||||
let (_peer_set, peer_set_handle) = peerset();
|
||||
let mut sender = sender::LightClientRequestSender::<Block>::new(
|
||||
&protocol_id(),
|
||||
Arc::new(crate::light_client_requests::tests::DummyFetchChecker {
|
||||
ok: true,
|
||||
_mark: std::marker::PhantomData,
|
||||
}),
|
||||
peer_set_handle,
|
||||
);
|
||||
sender.inject_connected(PeerId::random());
|
||||
|
||||
sender.request(request).unwrap();
|
||||
let sender::OutEvent::SendRequest { pending_response, request, .. } =
|
||||
block_on(sender.next()).unwrap();
|
||||
let (tx, rx) = oneshot::channel();
|
||||
block_on(protocol_config.inbound_queue.unwrap().send(IncomingRequest {
|
||||
peer: PeerId::random(),
|
||||
payload: request,
|
||||
pending_response: tx,
|
||||
}))
|
||||
.unwrap();
|
||||
pool.spawner()
|
||||
.spawn_obj(
|
||||
async move {
|
||||
pending_response.send(Ok(rx.await.unwrap().result.unwrap())).unwrap();
|
||||
}
|
||||
.boxed()
|
||||
.into(),
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
pool.spawner()
|
||||
.spawn_obj(sender.for_each(|_| future::ready(())).boxed().into())
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn send_receive_call() {
|
||||
let chan = oneshot::channel();
|
||||
let request = light::RemoteCallRequest {
|
||||
block: Default::default(),
|
||||
header: dummy_header(),
|
||||
method: "test".into(),
|
||||
call_data: vec![],
|
||||
retry_count: None,
|
||||
};
|
||||
|
||||
let mut pool = LocalPool::new();
|
||||
send_receive(sender::Request::Call { request, sender: chan.0 }, &pool);
|
||||
assert_eq!(vec![42], pool.run_until(chan.1).unwrap().unwrap());
|
||||
// ^--- from `DummyFetchChecker::check_execution_proof`
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn send_receive_read() {
|
||||
let chan = oneshot::channel();
|
||||
let request = light::RemoteReadRequest {
|
||||
header: dummy_header(),
|
||||
block: Default::default(),
|
||||
keys: vec![b":key".to_vec()],
|
||||
retry_count: None,
|
||||
};
|
||||
let mut pool = LocalPool::new();
|
||||
send_receive(sender::Request::Read { request, sender: chan.0 }, &pool);
|
||||
assert_eq!(
|
||||
Some(vec![42]),
|
||||
pool.run_until(chan.1).unwrap().unwrap().remove(&b":key"[..]).unwrap()
|
||||
);
|
||||
// ^--- from `DummyFetchChecker::check_read_proof`
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn send_receive_read_child() {
|
||||
let chan = oneshot::channel();
|
||||
let child_info = ChildInfo::new_default(&b":child_storage:default:sub"[..]);
|
||||
let request = light::RemoteReadChildRequest {
|
||||
header: dummy_header(),
|
||||
block: Default::default(),
|
||||
storage_key: child_info.prefixed_storage_key(),
|
||||
keys: vec![b":key".to_vec()],
|
||||
retry_count: None,
|
||||
};
|
||||
let mut pool = LocalPool::new();
|
||||
send_receive(sender::Request::ReadChild { request, sender: chan.0 }, &pool);
|
||||
assert_eq!(
|
||||
Some(vec![42]),
|
||||
pool.run_until(chan.1).unwrap().unwrap().remove(&b":key"[..]).unwrap()
|
||||
);
|
||||
// ^--- from `DummyFetchChecker::check_read_child_proof`
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn send_receive_header() {
|
||||
sp_tracing::try_init_simple();
|
||||
let chan = oneshot::channel();
|
||||
let request = light::RemoteHeaderRequest {
|
||||
cht_root: Default::default(),
|
||||
block: 1,
|
||||
retry_count: None,
|
||||
};
|
||||
let mut pool = LocalPool::new();
|
||||
send_receive(sender::Request::Header { request, sender: chan.0 }, &pool);
|
||||
// The remote does not know block 1:
|
||||
assert_matches!(pool.run_until(chan.1).unwrap(), Err(ClientError::RemoteFetchFailed));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn send_receive_changes() {
|
||||
let chan = oneshot::channel();
|
||||
let request = light::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,
|
||||
};
|
||||
let mut pool = LocalPool::new();
|
||||
send_receive(sender::Request::Changes { request, sender: chan.0 }, &pool);
|
||||
assert_eq!(vec![(100, 2)], pool.run_until(chan.1).unwrap().unwrap());
|
||||
// ^--- from `DummyFetchChecker::check_changes_proof`
|
||||
}
|
||||
}
|
||||
|
||||
@@ -32,17 +32,14 @@ use codec::{self, Decode, Encode};
|
||||
use futures::{channel::mpsc, prelude::*};
|
||||
use log::{debug, trace};
|
||||
use prost::Message;
|
||||
use sc_client_api::{light, StorageProof};
|
||||
use sc_client_api::StorageProof;
|
||||
use sc_peerset::ReputationChange;
|
||||
use sp_core::{
|
||||
hexdisplay::HexDisplay,
|
||||
storage::{ChildInfo, ChildType, PrefixedStorageKey, StorageKey},
|
||||
storage::{ChildInfo, ChildType, PrefixedStorageKey},
|
||||
};
|
||||
use sp_runtime::{
|
||||
generic::BlockId,
|
||||
traits::{Block, Zero},
|
||||
};
|
||||
use std::{collections::BTreeMap, sync::Arc};
|
||||
use sp_runtime::{generic::BlockId, traits::Block};
|
||||
use std::sync::Arc;
|
||||
|
||||
const LOG_TARGET: &str = "light-client-request-handler";
|
||||
|
||||
@@ -137,12 +134,12 @@ impl<B: Block> LightClientRequestHandler<B> {
|
||||
self.on_remote_call_request(&peer, r)?,
|
||||
Some(schema::v1::light::request::Request::RemoteReadRequest(r)) =>
|
||||
self.on_remote_read_request(&peer, r)?,
|
||||
Some(schema::v1::light::request::Request::RemoteHeaderRequest(r)) =>
|
||||
self.on_remote_header_request(&peer, r)?,
|
||||
Some(schema::v1::light::request::Request::RemoteHeaderRequest(_r)) =>
|
||||
return Err(HandleRequestError::BadRequest("Not supported.")),
|
||||
Some(schema::v1::light::request::Request::RemoteReadChildRequest(r)) =>
|
||||
self.on_remote_read_child_request(&peer, r)?,
|
||||
Some(schema::v1::light::request::Request::RemoteChangesRequest(r)) =>
|
||||
self.on_remote_changes_request(&peer, r)?,
|
||||
Some(schema::v1::light::request::Request::RemoteChangesRequest(_r)) =>
|
||||
return Err(HandleRequestError::BadRequest("Not supported.")),
|
||||
None =>
|
||||
return Err(HandleRequestError::BadRequest("Remote request without request data.")),
|
||||
};
|
||||
@@ -285,106 +282,6 @@ impl<B: Block> LightClientRequestHandler<B> {
|
||||
|
||||
Ok(schema::v1::light::Response { response: Some(response) })
|
||||
}
|
||||
|
||||
fn on_remote_header_request(
|
||||
&mut self,
|
||||
peer: &PeerId,
|
||||
request: &schema::v1::light::RemoteHeaderRequest,
|
||||
) -> Result<schema::v1::light::Response, HandleRequestError> {
|
||||
trace!("Remote header proof request from {} ({:?}).", peer, request.block);
|
||||
|
||||
let block = Decode::decode(&mut request.block.as_ref())?;
|
||||
let (header, proof) = match self.client.header_proof(&BlockId::Number(block)) {
|
||||
Ok((header, proof)) => (header.encode(), proof),
|
||||
Err(error) => {
|
||||
trace!(
|
||||
"Remote header proof request from {} ({:?}) failed with: {}.",
|
||||
peer,
|
||||
request.block,
|
||||
error
|
||||
);
|
||||
(Default::default(), StorageProof::empty())
|
||||
},
|
||||
};
|
||||
|
||||
let response = {
|
||||
let r = schema::v1::light::RemoteHeaderResponse { header, proof: proof.encode() };
|
||||
schema::v1::light::response::Response::RemoteHeaderResponse(r)
|
||||
};
|
||||
|
||||
Ok(schema::v1::light::Response { response: Some(response) })
|
||||
}
|
||||
|
||||
fn on_remote_changes_request(
|
||||
&mut self,
|
||||
peer: &PeerId,
|
||||
request: &schema::v1::light::RemoteChangesRequest,
|
||||
) -> Result<schema::v1::light::Response, HandleRequestError> {
|
||||
trace!(
|
||||
"Remote changes proof request from {} for key {} ({:?}..{:?}).",
|
||||
peer,
|
||||
if !request.storage_key.is_empty() {
|
||||
format!(
|
||||
"{} : {}",
|
||||
HexDisplay::from(&request.storage_key),
|
||||
HexDisplay::from(&request.key)
|
||||
)
|
||||
} else {
|
||||
HexDisplay::from(&request.key).to_string()
|
||||
},
|
||||
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(PrefixedStorageKey::new_ref(&request.storage_key))
|
||||
};
|
||||
|
||||
let proof =
|
||||
match self.client.key_changes_proof(first, last, min, max, storage_key, &key) {
|
||||
Ok(proof) => proof,
|
||||
Err(error) => {
|
||||
trace!(
|
||||
"Remote changes proof request from {} for key {} ({:?}..{:?}) failed with: {}.",
|
||||
peer,
|
||||
format!("{} : {}", HexDisplay::from(&request.storage_key), HexDisplay::from(&key.0)),
|
||||
request.first,
|
||||
request.last,
|
||||
error,
|
||||
);
|
||||
|
||||
light::ChangesProof::<B::Header> {
|
||||
max_block: Zero::zero(),
|
||||
proof: Vec::new(),
|
||||
roots: BTreeMap::new(),
|
||||
roots_proof: StorageProof::empty(),
|
||||
}
|
||||
},
|
||||
};
|
||||
|
||||
let response = {
|
||||
let r = schema::v1::light::RemoteChangesResponse {
|
||||
max: proof.max_block.encode(),
|
||||
proof: proof.proof,
|
||||
roots: proof
|
||||
.roots
|
||||
.into_iter()
|
||||
.map(|(k, v)| schema::v1::light::Pair { fst: k.encode(), snd: v.encode() })
|
||||
.collect(),
|
||||
roots_proof: proof.roots_proof.encode(),
|
||||
};
|
||||
schema::v1::light::response::Response::RemoteChangesResponse(r)
|
||||
};
|
||||
|
||||
Ok(schema::v1::light::Response { response: Some(response) })
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(derive_more::Display, derive_more::From)]
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,241 +0,0 @@
|
||||
// This file is part of Substrate.
|
||||
|
||||
// Copyright (C) 2017-2021 Parity Technologies (UK) Ltd.
|
||||
// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
|
||||
|
||||
// This program is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
|
||||
// This program is distributed in the hope that it will be useful,
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU General Public License for more details.
|
||||
|
||||
// You should have received a copy of the GNU General Public License
|
||||
// along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
|
||||
//! On-demand requests service.
|
||||
|
||||
use crate::light_client_requests;
|
||||
|
||||
use futures::{channel::oneshot, prelude::*};
|
||||
use parking_lot::Mutex;
|
||||
use sc_client_api::{
|
||||
ChangesProof, FetchChecker, Fetcher, RemoteBodyRequest, RemoteCallRequest,
|
||||
RemoteChangesRequest, RemoteHeaderRequest, RemoteReadChildRequest, RemoteReadRequest,
|
||||
StorageProof,
|
||||
};
|
||||
use sc_utils::mpsc::{tracing_unbounded, TracingUnboundedReceiver, TracingUnboundedSender};
|
||||
use sp_blockchain::Error as ClientError;
|
||||
use sp_runtime::traits::{Block as BlockT, Header as HeaderT, NumberFor};
|
||||
use std::{
|
||||
collections::HashMap,
|
||||
pin::Pin,
|
||||
sync::Arc,
|
||||
task::{Context, Poll},
|
||||
};
|
||||
|
||||
/// Implements the `Fetcher` trait of the client. Makes it possible for the light client to perform
|
||||
/// network requests for some state.
|
||||
///
|
||||
/// This implementation stores all the requests in a queue. The network, in parallel, is then
|
||||
/// responsible for pulling elements out of that queue and fulfilling them.
|
||||
pub struct OnDemand<B: BlockT> {
|
||||
/// Objects that checks whether what has been retrieved is correct.
|
||||
checker: Arc<dyn FetchChecker<B>>,
|
||||
|
||||
/// Queue of requests. Set to `Some` at initialization, then extracted by the network.
|
||||
///
|
||||
/// Note that a better alternative would be to use a MPMC queue here, and add a `poll` method
|
||||
/// from the `OnDemand`. However there exists no popular implementation of MPMC channels in
|
||||
/// asynchronous Rust at the moment
|
||||
requests_queue:
|
||||
Mutex<Option<TracingUnboundedReceiver<light_client_requests::sender::Request<B>>>>,
|
||||
|
||||
/// Sending side of `requests_queue`.
|
||||
requests_send: TracingUnboundedSender<light_client_requests::sender::Request<B>>,
|
||||
}
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
#[error("AlwaysBadChecker")]
|
||||
struct ErrorAlwaysBadChecker;
|
||||
|
||||
impl Into<ClientError> for ErrorAlwaysBadChecker {
|
||||
fn into(self) -> ClientError {
|
||||
ClientError::Application(Box::new(self))
|
||||
}
|
||||
}
|
||||
|
||||
/// Dummy implementation of `FetchChecker` that always assumes that responses are bad.
|
||||
///
|
||||
/// Considering that it is the responsibility of the client to build the fetcher, it can use this
|
||||
/// implementation if it knows that it will never perform any request.
|
||||
#[derive(Default, Clone)]
|
||||
pub struct AlwaysBadChecker;
|
||||
|
||||
impl<Block: BlockT> FetchChecker<Block> for AlwaysBadChecker {
|
||||
fn check_header_proof(
|
||||
&self,
|
||||
_request: &RemoteHeaderRequest<Block::Header>,
|
||||
_remote_header: Option<Block::Header>,
|
||||
_remote_proof: StorageProof,
|
||||
) -> Result<Block::Header, ClientError> {
|
||||
Err(ErrorAlwaysBadChecker.into())
|
||||
}
|
||||
|
||||
fn check_read_proof(
|
||||
&self,
|
||||
_request: &RemoteReadRequest<Block::Header>,
|
||||
_remote_proof: StorageProof,
|
||||
) -> Result<HashMap<Vec<u8>, Option<Vec<u8>>>, ClientError> {
|
||||
Err(ErrorAlwaysBadChecker.into())
|
||||
}
|
||||
|
||||
fn check_read_child_proof(
|
||||
&self,
|
||||
_request: &RemoteReadChildRequest<Block::Header>,
|
||||
_remote_proof: StorageProof,
|
||||
) -> Result<HashMap<Vec<u8>, Option<Vec<u8>>>, ClientError> {
|
||||
Err(ErrorAlwaysBadChecker.into())
|
||||
}
|
||||
|
||||
fn check_execution_proof(
|
||||
&self,
|
||||
_request: &RemoteCallRequest<Block::Header>,
|
||||
_remote_proof: StorageProof,
|
||||
) -> Result<Vec<u8>, ClientError> {
|
||||
Err(ErrorAlwaysBadChecker.into())
|
||||
}
|
||||
|
||||
fn check_changes_proof(
|
||||
&self,
|
||||
_request: &RemoteChangesRequest<Block::Header>,
|
||||
_remote_proof: ChangesProof<Block::Header>,
|
||||
) -> Result<Vec<(NumberFor<Block>, u32)>, ClientError> {
|
||||
Err(ErrorAlwaysBadChecker.into())
|
||||
}
|
||||
|
||||
fn check_body_proof(
|
||||
&self,
|
||||
_request: &RemoteBodyRequest<Block::Header>,
|
||||
_body: Vec<Block::Extrinsic>,
|
||||
) -> Result<Vec<Block::Extrinsic>, ClientError> {
|
||||
Err(ErrorAlwaysBadChecker.into())
|
||||
}
|
||||
}
|
||||
|
||||
impl<B: BlockT> OnDemand<B>
|
||||
where
|
||||
B::Header: HeaderT,
|
||||
{
|
||||
/// Creates new on-demand service.
|
||||
pub fn new(checker: Arc<dyn FetchChecker<B>>) -> Self {
|
||||
let (requests_send, requests_queue) = tracing_unbounded("mpsc_ondemand");
|
||||
let requests_queue = Mutex::new(Some(requests_queue));
|
||||
|
||||
Self { checker, requests_queue, requests_send }
|
||||
}
|
||||
|
||||
/// Get checker reference.
|
||||
pub fn checker(&self) -> &Arc<dyn FetchChecker<B>> {
|
||||
&self.checker
|
||||
}
|
||||
|
||||
/// Extracts the queue of requests.
|
||||
///
|
||||
/// Whenever one of the methods of the `Fetcher` trait is called, an element is pushed on this
|
||||
/// channel.
|
||||
///
|
||||
/// If this function returns `None`, that means that the receiver has already been extracted in
|
||||
/// the past, and therefore that something already handles the requests.
|
||||
pub(crate) fn extract_receiver(
|
||||
&self,
|
||||
) -> Option<TracingUnboundedReceiver<light_client_requests::sender::Request<B>>> {
|
||||
self.requests_queue.lock().take()
|
||||
}
|
||||
}
|
||||
|
||||
impl<B> Fetcher<B> for OnDemand<B>
|
||||
where
|
||||
B: BlockT,
|
||||
B::Header: HeaderT,
|
||||
{
|
||||
type RemoteHeaderResult = RemoteResponse<B::Header>;
|
||||
type RemoteReadResult = RemoteResponse<HashMap<Vec<u8>, Option<Vec<u8>>>>;
|
||||
type RemoteCallResult = RemoteResponse<Vec<u8>>;
|
||||
type RemoteChangesResult = RemoteResponse<Vec<(NumberFor<B>, u32)>>;
|
||||
type RemoteBodyResult = RemoteResponse<Vec<B::Extrinsic>>;
|
||||
|
||||
fn remote_header(&self, request: RemoteHeaderRequest<B::Header>) -> Self::RemoteHeaderResult {
|
||||
let (sender, receiver) = oneshot::channel();
|
||||
let _ = self
|
||||
.requests_send
|
||||
.unbounded_send(light_client_requests::sender::Request::Header { request, sender });
|
||||
RemoteResponse { receiver }
|
||||
}
|
||||
|
||||
fn remote_read(&self, request: RemoteReadRequest<B::Header>) -> Self::RemoteReadResult {
|
||||
let (sender, receiver) = oneshot::channel();
|
||||
let _ = self
|
||||
.requests_send
|
||||
.unbounded_send(light_client_requests::sender::Request::Read { request, sender });
|
||||
RemoteResponse { receiver }
|
||||
}
|
||||
|
||||
fn remote_read_child(
|
||||
&self,
|
||||
request: RemoteReadChildRequest<B::Header>,
|
||||
) -> Self::RemoteReadResult {
|
||||
let (sender, receiver) = oneshot::channel();
|
||||
let _ = self
|
||||
.requests_send
|
||||
.unbounded_send(light_client_requests::sender::Request::ReadChild { request, sender });
|
||||
RemoteResponse { receiver }
|
||||
}
|
||||
|
||||
fn remote_call(&self, request: RemoteCallRequest<B::Header>) -> Self::RemoteCallResult {
|
||||
let (sender, receiver) = oneshot::channel();
|
||||
let _ = self
|
||||
.requests_send
|
||||
.unbounded_send(light_client_requests::sender::Request::Call { request, sender });
|
||||
RemoteResponse { receiver }
|
||||
}
|
||||
|
||||
fn remote_changes(
|
||||
&self,
|
||||
request: RemoteChangesRequest<B::Header>,
|
||||
) -> Self::RemoteChangesResult {
|
||||
let (sender, receiver) = oneshot::channel();
|
||||
let _ = self
|
||||
.requests_send
|
||||
.unbounded_send(light_client_requests::sender::Request::Changes { request, sender });
|
||||
RemoteResponse { receiver }
|
||||
}
|
||||
|
||||
fn remote_body(&self, request: RemoteBodyRequest<B::Header>) -> Self::RemoteBodyResult {
|
||||
let (sender, receiver) = oneshot::channel();
|
||||
let _ = self
|
||||
.requests_send
|
||||
.unbounded_send(light_client_requests::sender::Request::Body { request, sender });
|
||||
RemoteResponse { receiver }
|
||||
}
|
||||
}
|
||||
|
||||
/// Future for an on-demand remote call response.
|
||||
pub struct RemoteResponse<T> {
|
||||
receiver: oneshot::Receiver<Result<T, ClientError>>,
|
||||
}
|
||||
|
||||
impl<T> Future for RemoteResponse<T> {
|
||||
type Output = Result<T, ClientError>;
|
||||
|
||||
fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
|
||||
match self.receiver.poll_unpin(cx) {
|
||||
Poll::Ready(Ok(res)) => Poll::Ready(res),
|
||||
Poll::Ready(Err(_)) => Poll::Ready(Err(ClientError::RemoteFetchCancelled)),
|
||||
Poll::Pending => Poll::Pending,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -33,11 +33,9 @@ use crate::{
|
||||
config::{parse_str_addr, Params, TransportConfig},
|
||||
discovery::DiscoveryConfig,
|
||||
error::Error,
|
||||
light_client_requests,
|
||||
network_state::{
|
||||
NetworkState, NotConnectedPeer as NetworkStateNotConnectedPeer, Peer as NetworkStatePeer,
|
||||
},
|
||||
on_demand_layer::AlwaysBadChecker,
|
||||
protocol::{
|
||||
self,
|
||||
event::Event,
|
||||
@@ -238,12 +236,6 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkWorker<B, H> {
|
||||
}
|
||||
})?;
|
||||
|
||||
let checker = params
|
||||
.on_demand
|
||||
.as_ref()
|
||||
.map(|od| od.checker().clone())
|
||||
.unwrap_or_else(|| Arc::new(AlwaysBadChecker));
|
||||
|
||||
let num_connected = Arc::new(AtomicUsize::new(0));
|
||||
let is_major_syncing = Arc::new(AtomicBool::new(false));
|
||||
|
||||
@@ -255,14 +247,6 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkWorker<B, H> {
|
||||
params.network_config.client_version, params.network_config.node_name
|
||||
);
|
||||
|
||||
let light_client_request_sender = {
|
||||
light_client_requests::sender::LightClientRequestSender::new(
|
||||
¶ms.protocol_id,
|
||||
checker,
|
||||
peerset_handle.clone(),
|
||||
)
|
||||
};
|
||||
|
||||
let discovery_config = {
|
||||
let mut config = DiscoveryConfig::new(local_public.clone());
|
||||
config.with_permanent_addresses(known_addresses);
|
||||
@@ -347,7 +331,6 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkWorker<B, H> {
|
||||
protocol,
|
||||
user_agent,
|
||||
local_public,
|
||||
light_client_request_sender,
|
||||
discovery_config,
|
||||
params.block_request_protocol_config,
|
||||
params.state_request_protocol_config,
|
||||
@@ -447,7 +430,6 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkWorker<B, H> {
|
||||
service,
|
||||
import_queue: params.import_queue,
|
||||
from_service,
|
||||
light_client_rqs: params.on_demand.and_then(|od| od.extract_receiver()),
|
||||
event_streams: out_events::OutChannels::new(params.metrics_registry.as_ref())?,
|
||||
peers_notifications_sinks,
|
||||
tx_handler_controller,
|
||||
@@ -1464,8 +1446,6 @@ pub struct NetworkWorker<B: BlockT + 'static, H: ExHashT> {
|
||||
import_queue: Box<dyn ImportQueue<B>>,
|
||||
/// Messages from the [`NetworkService`] that must be processed.
|
||||
from_service: TracingUnboundedReceiver<ServiceToWorkerMsg<B, H>>,
|
||||
/// Receiver for queries from the light client that must be processed.
|
||||
light_client_rqs: Option<TracingUnboundedReceiver<light_client_requests::sender::Request<B>>>,
|
||||
/// Senders for events that happen on the network.
|
||||
event_streams: out_events::OutChannels,
|
||||
/// Prometheus network metrics.
|
||||
@@ -1489,23 +1469,6 @@ impl<B: BlockT + 'static, H: ExHashT> Future for NetworkWorker<B, H> {
|
||||
this.import_queue
|
||||
.poll_actions(cx, &mut NetworkLink { protocol: &mut this.network_service });
|
||||
|
||||
// Check for new incoming light client requests.
|
||||
if let Some(light_client_rqs) = this.light_client_rqs.as_mut() {
|
||||
while let Poll::Ready(Some(rq)) = light_client_rqs.poll_next_unpin(cx) {
|
||||
let result = this.network_service.behaviour_mut().light_client_request(rq);
|
||||
match result {
|
||||
Ok(()) => {},
|
||||
Err(light_client_requests::sender::SendRequestError::TooManyRequests) => {
|
||||
warn!("Couldn't start light client request: too many pending requests");
|
||||
},
|
||||
}
|
||||
|
||||
if let Some(metrics) = this.metrics.as_ref() {
|
||||
metrics.issued_light_requests.inc();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// At the time of writing of this comment, due to a high volume of messages, the network
|
||||
// worker sometimes takes a long time to process the loop below. When that happens, the
|
||||
// rest of the polling is frozen. In order to avoid negative side-effects caused by this
|
||||
|
||||
@@ -116,7 +116,6 @@ fn build_test_full_node(
|
||||
}),
|
||||
network_config: config,
|
||||
chain: client.clone(),
|
||||
on_demand: None,
|
||||
transaction_pool: Arc::new(crate::config::EmptyTransactionPool),
|
||||
protocol_id,
|
||||
import_queue,
|
||||
|
||||
Reference in New Issue
Block a user