Unverified Commit d4022633 authored by Andronik Ordian's avatar Andronik Ordian Committed by GitHub
Browse files

Chain API subsystem (#1498)

* chain-api subsystem skeleton

* chain-api subsystem: draft impl

* chain-api subsystem: mock testclient

* chain-api subsystem: impl HeaderBacked for TestClient

* chain-api subsystem: impl basic tests

* chain-api subsystem: tiny guide

* chain-api subsystem: rename ChainApiRequestMessage to ChainApiMessage

* chain-api subsystem: add the page to the ToC

* chain-api subsystem: proper error type

* chain-api subsystem: impl ancestors request

* chain-api subsystem: tests for ancestors request

* guide: fix ancestor return type

* runtime-api subsystem: remove unused dep

* fix fmt

* fix outdated comment

* chain-api subsystem: s/format/to_string

* lower-case subsystem names

* chain-api subsystem: resolve Finalized todo

* chain-api subsystem: remove TODO

* extract request errors into a module

* remove caching TODO

* fix imports
parent 473f204f
Pipeline #102368 passed with stages
in 22 minutes and 18 seconds
......@@ -4653,6 +4653,18 @@ dependencies = [
"sp-keyring",
]
[[package]]
name = "polkadot-node-core-chain-api"
version = "0.1.0"
dependencies = [
"futures 0.3.5",
"maplit",
"polkadot-node-subsystem",
"polkadot-primitives",
"sp-blockchain",
"sp-core",
]
[[package]]
name = "polkadot-node-core-proposer"
version = "0.1.0"
......@@ -4683,7 +4695,6 @@ dependencies = [
name = "polkadot-node-core-runtime-api"
version = "0.1.0"
dependencies = [
"assert_matches",
"futures 0.3.5",
"polkadot-node-primitives",
"polkadot-node-subsystem",
......
......@@ -47,6 +47,7 @@ members = [
"node/core/backing",
"node/core/bitfield-signing",
"node/core/candidate-validation",
"node/core/chain-api",
"node/core/proposer",
"node/core/runtime-api",
"node/network/bridge",
......
......@@ -26,8 +26,9 @@ use keystore::KeyStorePtr;
use polkadot_node_subsystem::{
messages::{
self, AllMessages, AvailabilityStoreMessage, BitfieldDistributionMessage,
BitfieldSigningMessage, CandidateBackingMessage, RuntimeApiMessage, RuntimeApiError
BitfieldSigningMessage, CandidateBackingMessage, RuntimeApiMessage,
},
errors::RuntimeApiError,
util::{self, JobManager, JobTrait, ToJobTrait, Validator},
};
use polkadot_primitives::v1::{AvailabilityBitfield, CoreState, Hash, ValidatorIndex};
......
......@@ -26,8 +26,8 @@ use polkadot_subsystem::{
};
use polkadot_subsystem::messages::{
AllMessages, CandidateValidationMessage, RuntimeApiMessage, ValidationFailed, RuntimeApiRequest,
RuntimeApiError,
};
use polkadot_subsystem::errors::RuntimeApiError;
use polkadot_node_primitives::{ValidationResult, ValidationOutputs};
use polkadot_primitives::v1::{
ValidationCode, OmittedValidationData, PoV, CandidateDescriptor, LocalValidationData,
......
[package]
name = "polkadot-node-core-chain-api"
version = "0.1.0"
authors = ["Parity Technologies <admin@parity.io>"]
edition = "2018"
[dependencies]
futures = { version = "0.3.5" }
sp-blockchain = { git = "https://github.com/paritytech/substrate", branch = "master" }
polkadot-primitives = { path = "../../../primitives" }
polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsystem" }
[dev-dependencies]
futures = { version = "0.3.5", features = ["thread-pool"] }
maplit = "1.0.2"
polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsystem", features = ["test-helpers"] }
sp-core = { git = "https://github.com/paritytech/substrate", branch = "master" }
// Copyright 2020 Parity Technologies (UK) Ltd.
// This file is part of Polkadot.
// Polkadot 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.
// Polkadot 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 Polkadot. If not, see <http://www.gnu.org/licenses/>.
//! Implements the Chain API Subsystem
//!
//! Provides access to the chain data. Every request may return an error.
//! At the moment, the implementation requires `Client` to implement `HeaderBackend`,
//! we may add more bounds in the future if we will need e.g. block bodies.
//!
//! Supported requests:
//! * Block hash to number
//! * Finalized block number to hash
//! * Last finalized block number
//! * Ancestors
use polkadot_subsystem::{
FromOverseer, OverseerSignal,
SpawnedSubsystem, Subsystem, SubsystemResult, SubsystemContext,
messages::ChainApiMessage,
};
use polkadot_primitives::v1::{Block, BlockId};
use sp_blockchain::HeaderBackend;
use futures::prelude::*;
/// The Chain API Subsystem implementation.
pub struct ChainApiSubsystem<Client> {
client: Client,
}
impl<Client> ChainApiSubsystem<Client> {
/// Create a new Chain API subsystem with the given client.
pub fn new(client: Client) -> Self {
ChainApiSubsystem {
client
}
}
}
impl<Client, Context> Subsystem<Context> for ChainApiSubsystem<Client> where
Client: HeaderBackend<Block> + 'static,
Context: SubsystemContext<Message = ChainApiMessage>
{
fn start(self, ctx: Context) -> SpawnedSubsystem {
SpawnedSubsystem {
future: run(ctx, self.client).map(|_| ()).boxed(),
name: "chain-api-subsystem",
}
}
}
async fn run<Client>(
mut ctx: impl SubsystemContext<Message = ChainApiMessage>,
client: Client,
) -> SubsystemResult<()>
where
Client: HeaderBackend<Block>,
{
loop {
match ctx.recv().await? {
FromOverseer::Signal(OverseerSignal::Conclude) => return Ok(()),
FromOverseer::Signal(OverseerSignal::ActiveLeaves(_)) => {},
FromOverseer::Signal(OverseerSignal::BlockFinalized(_)) => {},
FromOverseer::Communication { msg } => match msg {
ChainApiMessage::BlockNumber(hash, response_channel) => {
let result = client.number(hash).map_err(|e| e.to_string().into());
let _ = response_channel.send(result);
},
ChainApiMessage::FinalizedBlockHash(number, response_channel) => {
// Note: we don't verify it's finalized
let result = client.hash(number).map_err(|e| e.to_string().into());
let _ = response_channel.send(result);
},
ChainApiMessage::FinalizedBlockNumber(response_channel) => {
let result = client.info().finalized_number;
let _ = response_channel.send(Ok(result));
},
ChainApiMessage::Ancestors { hash, k, response_channel } => {
let mut hash = hash;
let next_parent = core::iter::from_fn(|| {
let maybe_header = client.header(BlockId::Hash(hash));
match maybe_header {
// propagate the error
Err(e) => Some(Err(e.to_string().into())),
// fewer than `k` ancestors are available
Ok(None) => None,
Ok(Some(header)) => {
hash = header.parent_hash;
Some(Ok(hash))
}
}
});
let result = next_parent.take(k).collect::<Result<Vec<_>, _>>();
let _ = response_channel.send(result);
},
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::collections::BTreeMap;
use futures::{future::BoxFuture, channel::oneshot};
use polkadot_primitives::v1::{Hash, BlockNumber, BlockId, Header};
use polkadot_subsystem::test_helpers::{make_subsystem_context, TestSubsystemContextHandle};
use sp_blockchain::Info as BlockInfo;
use sp_core::testing::TaskExecutor;
#[derive(Clone)]
struct TestClient {
blocks: BTreeMap<Hash, BlockNumber>,
finalized_blocks: BTreeMap<BlockNumber, Hash>,
headers: BTreeMap<Hash, Header>,
}
const ONE: Hash = Hash::repeat_byte(0x01);
const TWO: Hash = Hash::repeat_byte(0x02);
const THREE: Hash = Hash::repeat_byte(0x03);
const FOUR: Hash = Hash::repeat_byte(0x04);
const ERROR_PATH: Hash = Hash::repeat_byte(0xFF);
fn default_header() -> Header {
Header {
parent_hash: Hash::zero(),
number: 100500,
state_root: Hash::zero(),
extrinsics_root: Hash::zero(),
digest: Default::default(),
}
}
impl Default for TestClient {
fn default() -> Self {
Self {
blocks: maplit::btreemap! {
ONE => 1,
TWO => 2,
THREE => 3,
FOUR => 4,
},
finalized_blocks: maplit::btreemap! {
1 => ONE,
3 => THREE,
},
headers: maplit::btreemap! {
TWO => Header {
parent_hash: ONE,
number: 2,
..default_header()
},
THREE => Header {
parent_hash: TWO,
number: 3,
..default_header()
},
FOUR => Header {
parent_hash: THREE,
number: 4,
..default_header()
},
ERROR_PATH => Header {
..default_header()
}
}
}
}
}
fn last_key_value<K: Clone, V: Clone>(map: &BTreeMap<K, V>) -> (K, V) {
assert!(!map.is_empty());
map.iter()
.last()
.map(|(k, v)| (k.clone(), v.clone()))
.unwrap()
}
impl HeaderBackend<Block> for TestClient {
fn info(&self) -> BlockInfo<Block> {
let genesis_hash = self.blocks.iter().next().map(|(h, _)| *h).unwrap();
let (best_hash, best_number) = last_key_value(&self.blocks);
let (finalized_number, finalized_hash) = last_key_value(&self.finalized_blocks);
BlockInfo {
best_hash,
best_number,
genesis_hash,
finalized_hash,
finalized_number,
number_leaves: 0,
}
}
fn number(&self, hash: Hash) -> sp_blockchain::Result<Option<BlockNumber>> {
Ok(self.blocks.get(&hash).copied())
}
fn hash(&self, number: BlockNumber) -> sp_blockchain::Result<Option<Hash>> {
Ok(self.finalized_blocks.get(&number).copied())
}
fn header(&self, id: BlockId) -> sp_blockchain::Result<Option<Header>> {
match id {
// for error path testing
BlockId::Hash(hash) if hash.is_zero() => {
Err(sp_blockchain::Error::Backend("Zero hashes are illegal!".into()))
}
BlockId::Hash(hash) => {
Ok(self.headers.get(&hash).cloned())
}
_ => unreachable!(),
}
}
fn status(&self, _id: BlockId) -> sp_blockchain::Result<sp_blockchain::BlockStatus> {
unimplemented!()
}
}
fn test_harness(
test: impl FnOnce(TestClient, TestSubsystemContextHandle<ChainApiMessage>)
-> BoxFuture<'static, ()>,
) {
let (ctx, ctx_handle) = make_subsystem_context(TaskExecutor::new());
let client = TestClient::default();
let chain_api_task = run(ctx, client.clone()).map(|x| x.unwrap());
let test_task = test(client, ctx_handle);
futures::executor::block_on(future::join(chain_api_task, test_task));
}
#[test]
fn request_block_number() {
test_harness(|client, mut sender| {
async move {
let zero = Hash::zero();
let test_cases = [
(TWO, client.number(TWO).unwrap()),
(zero, client.number(zero).unwrap()), // not here
];
for (hash, expected) in &test_cases {
let (tx, rx) = oneshot::channel();
sender.send(FromOverseer::Communication {
msg: ChainApiMessage::BlockNumber(*hash, tx),
}).await;
assert_eq!(rx.await.unwrap().unwrap(), *expected);
}
sender.send(FromOverseer::Signal(OverseerSignal::Conclude)).await;
}.boxed()
})
}
#[test]
fn request_finalized_hash() {
test_harness(|client, mut sender| {
async move {
let test_cases = [
(1, client.hash(1).unwrap()), // not here
(2, client.hash(2).unwrap()),
];
for (number, expected) in &test_cases {
let (tx, rx) = oneshot::channel();
sender.send(FromOverseer::Communication {
msg: ChainApiMessage::FinalizedBlockHash(*number, tx),
}).await;
assert_eq!(rx.await.unwrap().unwrap(), *expected);
}
sender.send(FromOverseer::Signal(OverseerSignal::Conclude)).await;
}.boxed()
})
}
#[test]
fn request_last_finalized_number() {
test_harness(|client, mut sender| {
async move {
let (tx, rx) = oneshot::channel();
let expected = client.info().finalized_number;
sender.send(FromOverseer::Communication {
msg: ChainApiMessage::FinalizedBlockNumber(tx),
}).await;
assert_eq!(rx.await.unwrap().unwrap(), expected);
sender.send(FromOverseer::Signal(OverseerSignal::Conclude)).await;
}.boxed()
})
}
#[test]
fn request_ancestors() {
test_harness(|_client, mut sender| {
async move {
let (tx, rx) = oneshot::channel();
sender.send(FromOverseer::Communication {
msg: ChainApiMessage::Ancestors { hash: THREE, k: 4, response_channel: tx },
}).await;
assert_eq!(rx.await.unwrap().unwrap(), vec![TWO, ONE]);
let (tx, rx) = oneshot::channel();
sender.send(FromOverseer::Communication {
msg: ChainApiMessage::Ancestors { hash: TWO, k: 1, response_channel: tx },
}).await;
assert_eq!(rx.await.unwrap().unwrap(), vec![ONE]);
let (tx, rx) = oneshot::channel();
sender.send(FromOverseer::Communication {
msg: ChainApiMessage::Ancestors { hash: ERROR_PATH, k: 2, response_channel: tx },
}).await;
assert!(rx.await.unwrap().is_err());
sender.send(FromOverseer::Signal(OverseerSignal::Conclude)).await;
}.boxed()
})
}
}
......@@ -17,4 +17,3 @@ polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsys
sp-core = { git = "https://github.com/paritytech/substrate", branch = "master" }
futures = { version = "0.3.5", features = ["thread-pool"] }
polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsystem", features = ["test-helpers"] }
assert_matches = "1.3.0"
......@@ -24,8 +24,9 @@ use polkadot_subsystem::{
FromOverseer, OverseerSignal,
};
use polkadot_subsystem::messages::{
RuntimeApiMessage, RuntimeApiRequest as Request, RuntimeApiError,
RuntimeApiMessage, RuntimeApiRequest as Request,
};
use polkadot_subsystem::errors::RuntimeApiError;
use polkadot_primitives::v1::{Block, BlockId, Hash, ParachainHost};
use sp_api::{ProvideRuntimeApi};
......@@ -50,7 +51,7 @@ impl<Client, Context> Subsystem<Context> for RuntimeApiSubsystem<Client> where
fn start(self, ctx: Context) -> SpawnedSubsystem {
SpawnedSubsystem {
future: run(ctx, self.0).map(|_| ()).boxed(),
name: "RuntimeApiSubsystem",
name: "runtime-api-subsystem",
}
}
}
......
// Copyright 2020 Parity Technologies (UK) Ltd.
// This file is part of Polkadot.
// Polkadot 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.
// Polkadot 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 Polkadot. If not, see <http://www.gnu.org/licenses/>.
//! Error types for the subsystem requests.
/// A description of an error causing the runtime API request to be unservable.
#[derive(Debug, Clone)]
pub struct RuntimeApiError(String);
impl From<String> for RuntimeApiError {
fn from(s: String) -> Self {
RuntimeApiError(s)
}
}
impl core::fmt::Display for RuntimeApiError {
fn fmt(&self, f: &mut core::fmt::Formatter) -> Result<(), core::fmt::Error> {
write!(f, "{}", self.0)
}
}
/// A description of an error causing the chain API request to be unservable.
#[derive(Debug, Clone)]
pub struct ChainApiError {
msg: String,
}
impl From<&str> for ChainApiError {
fn from(s: &str) -> Self {
s.to_owned().into()
}
}
impl From<String> for ChainApiError {
fn from(msg: String) -> Self {
Self { msg }
}
}
impl core::fmt::Display for ChainApiError {
fn fmt(&self, f: &mut core::fmt::Formatter) -> Result<(), core::fmt::Error> {
write!(f, "{}", self.msg)
}
}
......@@ -34,6 +34,7 @@ use smallvec::SmallVec;
use crate::messages::AllMessages;
pub mod errors;
pub mod messages;
pub mod util;
#[cfg(any(test, feature = "test-helpers"))]
......
......@@ -31,7 +31,7 @@ use polkadot_primitives::v1::{
CoreAssignment, CoreOccupied, CandidateDescriptor,
ValidatorSignature, OmittedValidationData, AvailableData, GroupRotationInfo,
CoreState, LocalValidationData, GlobalValidationData, OccupiedCoreAssumption,
CandidateEvent, SessionIndex,
CandidateEvent, SessionIndex, BlockNumber,
};
use polkadot_node_primitives::{
MisbehaviorReport, SignedFullStatement, View, ProtocolId, ValidationResult,
......@@ -285,6 +285,43 @@ impl AvailabilityStoreMessage {
}
}
/// A response channel for the result of a chain API request.
pub type ChainApiResponseChannel<T> = oneshot::Sender<Result<T, crate::errors::ChainApiError>>;
/// Chain API request subsystem message.
#[derive(Debug)]
pub enum ChainApiMessage {
/// Request the block number by hash.
/// Returns `None` if a block with the given hash is not present in the db.
BlockNumber(Hash, ChainApiResponseChannel<Option<BlockNumber>>),
/// Request the finalized block hash by number.
/// Returns `None` if a block with the given number is not present in the db.
/// Note: the caller must ensure the block is finalized.
FinalizedBlockHash(BlockNumber, ChainApiResponseChannel<Option<Hash>>),
/// Request the last finalized block number.
/// This request always succeeds.
FinalizedBlockNumber(ChainApiResponseChannel<BlockNumber>),
/// Request the `k` ancestors block hashes of a block with the given hash.
/// The response channel may return a `Vec` of size up to `k`
/// filled with ancestors hashes with the following order:
/// `parent`, `grandparent`, ...
Ancestors {
/// The hash of the block in question.
hash: Hash,
/// The number of ancestors to request.
k: usize,
/// The response channel.
response_channel: ChainApiResponseChannel<Vec<Hash>>,
},
}
impl ChainApiMessage {
/// If the current variant contains the relay parent hash, return it.
pub fn relay_parent(&self) -> Option<Hash> {
None
}
}
/// The information on scheduler assignments that some somesystems may be querying.
#[derive(Debug, Clone)]
pub struct SchedulerRoster {
......@@ -298,18 +335,8 @@ pub struct SchedulerRoster {
pub availability_cores: Vec<Option<CoreOccupied>>,
}
/// A description of an error causing the runtime API request to be unservable.
#[derive(Debug, Clone)]
pub struct RuntimeApiError(String);
impl From<String> for RuntimeApiError {
fn from(s: String) -> Self {
RuntimeApiError(s)
}
}
/// A sender for the result of a runtime API request.
pub type RuntimeApiSender<T> = oneshot::Sender<Result<T, RuntimeApiError>>;
pub type RuntimeApiSender<T> = oneshot::Sender<Result<T, crate::errors::RuntimeApiError>>;
/// A request to the Runtime API subsystem.
#[derive(Debug)]
......
......@@ -22,8 +22,9 @@
use crate::{
messages::{
AllMessages, RuntimeApiError, RuntimeApiMessage, RuntimeApiRequest, RuntimeApiSender,
AllMessages, RuntimeApiMessage, RuntimeApiRequest, RuntimeApiSender,
},
errors::{ChainApiError, RuntimeApiError},
FromOverseer, SpawnedSubsystem, Subsystem, SubsystemContext, SubsystemError, SubsystemResult,
};
use futures::{
......@@ -72,7 +73,10 @@ pub enum Error {
/// A subsystem error
#[from]
Subsystem(SubsystemError),
/// An error in the runtime API.
/// An error in the Chain API.
#[from]
ChainApi(ChainApiError),
/// An error in the Runtime API.
#[from]
RuntimeApi(RuntimeApiError),
/// The type system wants this even though it doesn't make sense
......