Add cumulus-service (#187)

* Add cumulus-service

Crate that abstracts the service of a parachain.

* Make finalization infallible
This commit is contained in:
Bastian Köcher
2020-08-08 08:07:33 +02:00
committed by GitHub
parent 1fcd05503a
commit 9603b2fe5f
11 changed files with 422 additions and 145 deletions
Generated
+22 -1
View File
@@ -920,7 +920,6 @@ dependencies = [
"polkadot-validation",
"sc-cli",
"sc-client-api",
"sc-service",
"sp-api",
"sp-blockchain",
"sp-consensus",
@@ -1093,6 +1092,27 @@ dependencies = [
"trie-db",
]
[[package]]
name = "cumulus-service"
version = "0.1.0"
dependencies = [
"cumulus-collator",
"cumulus-consensus",
"cumulus-network",
"cumulus-primitives",
"polkadot-collator",
"polkadot-primitives",
"polkadot-service",
"sc-client-api",
"sc-service",
"sp-api",
"sp-blockchain",
"sp-consensus",
"sp-core",
"sp-inherents",
"sp-runtime",
]
[[package]]
name = "cumulus-test-client"
version = "0.1.0"
@@ -5788,6 +5808,7 @@ dependencies = [
"cumulus-contracts-parachain-runtime",
"cumulus-network",
"cumulus-primitives",
"cumulus-service",
"cumulus-test-parachain-runtime",
"derive_more 0.15.0",
"exit-future 0.1.4",
+1
View File
@@ -11,6 +11,7 @@ members = [
"rococo-parachains/pallets/token-dealer",
"rococo-parachains/runtime",
"runtime",
"service",
"test/runtime",
"test/client",
"upward-message",
-1
View File
@@ -13,7 +13,6 @@ sp-consensus = { git = "https://github.com/paritytech/substrate", branch = "roco
sp-inherents = { git = "https://github.com/paritytech/substrate", branch = "rococo-branch" }
sp-api = { git = "https://github.com/paritytech/substrate", branch = "rococo-branch" }
sc-client-api = { git = "https://github.com/paritytech/substrate", branch = "rococo-branch" }
sc-service = { git = "https://github.com/paritytech/substrate", branch = "rococo-branch" }
sc-cli = { git = "https://github.com/paritytech/substrate", branch = "rococo-branch" }
# Polkadot dependencies
-11
View File
@@ -30,7 +30,6 @@ use cumulus_primitives::{
use cumulus_runtime::ParachainBlockData;
use sc_client_api::{BlockBackend, Finalizer, StateBackend, UsageProvider};
use sc_service::Configuration;
use sp_api::ApiExt;
use sp_blockchain::HeaderBackend;
use sp_consensus::{
@@ -547,16 +546,6 @@ where
}
}
/// Prepare the collator's node condifugration
///
/// This function will disable the default announcement of Substrate for the parachain in favor
/// of the one of Cumulus.
pub fn prepare_collator_config(mut parachain_config: Configuration) -> Configuration {
parachain_config.announce_block = false;
parachain_config
}
#[cfg(test)]
mod tests {
use super::*;
+31 -20
View File
@@ -26,12 +26,10 @@ use sp_runtime::{
traits::{Block as BlockT, Header as HeaderT},
};
use polkadot_primitives::v0::{
Id as ParaId, ParachainHost, Block as PBlock, Hash as PHash,
};
use polkadot_primitives::v0::{Block as PBlock, Hash as PHash, Id as ParaId, ParachainHost};
use codec::Decode;
use futures::{future, Future, FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt};
use futures::{future, Future, FutureExt, Stream, StreamExt};
use log::{error, trace, warn};
use std::{marker::PhantomData, sync::Arc};
@@ -118,26 +116,39 @@ where
polkadot
.finalized_heads(para_id)?
.map(|head_data| {
<<Block as BlockT>::Header>::decode(&mut &head_data[..])
.map_err(|_| Error::InvalidHeadData)
.filter_map(|head_data| {
let res = match <<Block as BlockT>::Header>::decode(&mut &head_data[..]) {
Ok(header) => Some(header),
Err(err) => {
warn!(
target: "cumulus-consensus",
"Could not decode Parachain header for finalizing: {:?}",
err,
);
None
}
};
future::ready(res)
})
.try_for_each(move |p_head| {
future::ready(
finalize_block(&*local, p_head.hash())
.map_err(Error::Client)
.map(|_| ()),
)
.for_each(move |p_head| {
if let Err(e) = finalize_block(&*local, p_head.hash()) {
warn!(
target: "cumulus-consensus",
"Failed to finalize block: {:?}",
e,
);
}
future::ready(())
})
.map_err(|e| {
warn!(
target: "cumulus-consensus",
"Failed to finalize block: {:?}", e)
})
.map(|_| ())
};
Ok(future::select(follow_finalized, follow_new_best(para_id, local, polkadot, announce_block)?).map(|_| ()))
Ok(future::select(
follow_finalized,
follow_new_best(para_id, local, polkadot, announce_block)?,
)
.map(|_| ()))
}
/// Follow the relay chain new best head, to update the Parachain new best head.
+1
View File
@@ -57,6 +57,7 @@ cumulus-consensus = { path = "../consensus" }
cumulus-collator = { path = "../collator" }
cumulus-network = { path = "../network" }
cumulus-primitives = { path = "../primitives" }
cumulus-service = { path = "../service" }
# Polkadot dependencies
polkadot-primitives = { git = "https://github.com/paritytech/polkadot", branch = "rococo-branch" }
+1 -1
View File
@@ -256,7 +256,7 @@ pub fn run() -> Result<()> {
if cli.run.base.validator { "yes" } else { "no" }
);
crate::service::run_collator(
crate::service::run_node(
config,
key,
polkadot_config,
+1 -1
View File
@@ -105,7 +105,7 @@ async fn integration_test() {
let parachain_config =
parachain_config(task_executor.clone(), Charlie, vec![], para_id).unwrap();
let (_service, charlie_client) =
crate::service::run_collator(parachain_config, key, polkadot_config, para_id, true)
crate::service::run_node(parachain_config, key, polkadot_config, para_id, true)
.unwrap();
sleep(Duration::from_secs(3)).await;
charlie_client.wait_for_blocks(4).await;
+92 -110
View File
@@ -15,19 +15,18 @@
// along with Cumulus. If not, see <http://www.gnu.org/licenses/>.
use ansi_term::Color;
use cumulus_collator::{prepare_collator_config, CollatorBuilder};
use cumulus_network::{DelayedBlockAnnounceValidator, JustifiedBlockAnnounceValidator};
use polkadot_primitives::v0::{CollatorPair, Block as PBlock, Id as ParaId};
use cumulus_network::DelayedBlockAnnounceValidator;
use cumulus_service::{
prepare_node_config, start_collator, start_full_node, StartCollatorParams, StartFullNodeParams,
};
use polkadot_primitives::v0::CollatorPair;
use sc_executor::native_executor_instance;
pub use sc_executor::NativeExecutor;
use sc_informant::OutputFormat;
use sc_service::{Configuration, PartialComponents, TaskManager, TFullBackend, TFullClient, Role};
use std::sync::Arc;
use sp_core::crypto::Pair;
use sc_service::{Configuration, PartialComponents, Role, TFullBackend, TFullClient, TaskManager};
use sp_runtime::traits::BlakeTwo256;
use sp_trie::PrefixedMemoryDB;
use sp_runtime::traits::{BlakeTwo256, Block as BlockT};
use polkadot_service::{AbstractClient, RuntimeApiCollection};
use sp_consensus::SyncOracle;
use std::sync::Arc;
// Our native executor instance.
native_executor_instance!(
@@ -40,26 +39,40 @@ native_executor_instance!(
///
/// Use this macro if you don't actually need the full service, but just the builder in order to
/// be able to perform chain operations.
pub fn new_partial(config: &mut Configuration) -> Result<
pub fn new_partial(
config: &mut Configuration,
) -> Result<
PartialComponents<
TFullClient<parachain_runtime::opaque::Block, parachain_runtime::RuntimeApi, crate::service::Executor>,
TFullBackend<parachain_runtime::opaque::Block>,
(),
sp_consensus::import_queue::BasicQueue<parachain_runtime::opaque::Block, PrefixedMemoryDB<BlakeTwo256>>,
sc_transaction_pool::FullPool<parachain_runtime::opaque::Block, TFullClient<parachain_runtime::opaque::Block, parachain_runtime::RuntimeApi, crate::service::Executor>>,
(),
>,
sc_service::Error,
>
{
let inherent_data_providers = sp_inherents::InherentDataProviders::new();
let (client, backend, keystore, task_manager) =
sc_service::new_full_parts::<
TFullClient<
parachain_runtime::opaque::Block,
parachain_runtime::RuntimeApi,
crate::service::Executor,
>(&config)?;
>,
TFullBackend<parachain_runtime::opaque::Block>,
(),
sp_consensus::import_queue::BasicQueue<
parachain_runtime::opaque::Block,
PrefixedMemoryDB<BlakeTwo256>,
>,
sc_transaction_pool::FullPool<
parachain_runtime::opaque::Block,
TFullClient<
parachain_runtime::opaque::Block,
parachain_runtime::RuntimeApi,
crate::service::Executor,
>,
>,
(),
>,
sc_service::Error,
> {
let inherent_data_providers = sp_inherents::InherentDataProviders::new();
let (client, backend, keystore, task_manager) = sc_service::new_full_parts::<
parachain_runtime::opaque::Block,
parachain_runtime::RuntimeApi,
crate::service::Executor,
>(&config)?;
let client = Arc::new(client);
//let select_chain = sc_consensus::LongestChain::new(backend.clone());
@@ -96,24 +109,30 @@ pub fn new_partial(config: &mut Configuration) -> Result<
Ok(params)
}
/// Run a collator node with the given parachain `Configuration` and relaychain `Configuration`
/// Run a node with the given parachain `Configuration` and relay chain `Configuration`
///
/// This function blocks until done.
pub fn run_collator(
pub fn run_node(
parachain_config: Configuration,
key: Arc<CollatorPair>,
collator_key: Arc<CollatorPair>,
mut polkadot_config: polkadot_collator::Configuration,
id: polkadot_primitives::v0::Id,
validator: bool,
) -> sc_service::error::Result<(
TaskManager,
Arc<TFullClient<parachain_runtime::opaque::Block, parachain_runtime::RuntimeApi, crate::service::Executor>>,
Arc<
TFullClient<
parachain_runtime::opaque::Block,
parachain_runtime::RuntimeApi,
crate::service::Executor,
>,
>,
)> {
if matches!(parachain_config.role, Role::Light) {
return Err("Light client not supported!".into());
}
let mut parachain_config = prepare_collator_config(parachain_config);
let mut parachain_config = prepare_node_config(parachain_config);
parachain_config.informant_output_format = OutputFormat {
enable_color: true,
@@ -125,7 +144,8 @@ pub fn run_collator(
};
let params = new_partial(&mut parachain_config)?;
params.inherent_data_providers
params
.inherent_data_providers
.register_provider(sp_timestamp::InherentDataProvider)
.unwrap();
@@ -171,22 +191,24 @@ pub fn run_collator(
})
};
let _rpc_handlers = sc_service::spawn_tasks(sc_service::SpawnTasksParams {
on_demand: None,
remote_blockchain: None,
rpc_extensions_builder,
client: client.clone(),
transaction_pool: transaction_pool.clone(),
task_manager: &mut task_manager,
telemetry_connection_sinks: Default::default(),
config: parachain_config,
keystore: params.keystore,
backend,
network: network.clone(),
network_status_sinks,
system_rpc_tx,
sc_service::spawn_tasks(sc_service::SpawnTasksParams {
on_demand: None,
remote_blockchain: None,
rpc_extensions_builder,
client: client.clone(),
transaction_pool: transaction_pool.clone(),
task_manager: &mut task_manager,
telemetry_connection_sinks: Default::default(),
config: parachain_config,
keystore: params.keystore,
backend,
network: network.clone(),
network_status_sinks,
system_rpc_tx,
})?;
let announce_block = Arc::new(move |hash, data| network.announce_block(hash, data));
if validator {
let proposer_factory = sc_basic_authorship::ProposerFactory::new(
client.clone(),
@@ -194,76 +216,36 @@ pub fn run_collator(
prometheus_registry.as_ref(),
);
let block_import = client.clone();
let announce_block = Arc::new(move |hash, data| network.announce_block(hash, data));
let builder = CollatorBuilder::new(
proposer_factory,
params.inherent_data_providers,
block_import,
client.clone(),
id,
client.clone(),
announce_block,
block_announce_validator,
);
let (polkadot_future, polkadot_task_manager) =
polkadot_collator::start_collator(builder, id, key, polkadot_config)?;
task_manager
.spawn_essential_handle()
.spawn("polkadot", polkadot_future);
task_manager.add_child(polkadot_task_manager);
} else {
let is_light = matches!(polkadot_config.role, Role::Light);
let (polkadot_task_manager, client, handles) = if is_light {
Err("Light client not supported.".into())
} else {
polkadot_service::build_full(
polkadot_config,
Some((key.public(), id)),
None,
false,
6000,
None,
)
}?;
let polkadot_network = handles.polkadot_network.expect("polkadot service is started; qed");
client.execute_with(SetDelayedBlockAnnounceValidator {
block_announce_validator,
let params = StartCollatorParams {
para_id: id,
polkadot_sync_oracle: Box::new(polkadot_network),
});
block_import: client.clone(),
proposer_factory,
inherent_data_providers: params.inherent_data_providers,
block_status: client.clone(),
announce_block,
client: client.clone(),
block_announce_validator,
task_manager: &mut task_manager,
polkadot_config,
collator_key,
};
task_manager.add_child(polkadot_task_manager);
start_collator(params)?;
} else {
let params = StartFullNodeParams {
client: client.clone(),
announce_block,
polkadot_config,
collator_key,
block_announce_validator,
task_manager: &mut task_manager,
para_id: id,
};
start_full_node(params)?;
}
start_network.start_network();
Ok((task_manager, client))
}
struct SetDelayedBlockAnnounceValidator<B: BlockT> {
block_announce_validator: DelayedBlockAnnounceValidator<B>,
para_id: ParaId,
polkadot_sync_oracle: Box<dyn SyncOracle + Send>,
}
impl<B: BlockT> polkadot_service::ExecuteWithClient for SetDelayedBlockAnnounceValidator<B> {
type Output = ();
fn execute_with_client<Client, Api, Backend>(self, client: Arc<Client>) -> Self::Output
where<Api as sp_api::ApiExt<PBlock>>::StateBackend: sp_api::StateBackend<BlakeTwo256>,
Backend: sc_client_api::Backend<PBlock>,
Backend::State: sp_api::StateBackend<BlakeTwo256>,
Api: RuntimeApiCollection<StateBackend = Backend::State>,
Client: AbstractClient<PBlock, Backend, Api = Api> + 'static
{
self.block_announce_validator.set(Box::new(JustifiedBlockAnnounceValidator::new(
client,
self.para_id,
self.polkadot_sync_oracle,
)));
}
}
+27
View File
@@ -0,0 +1,27 @@
[package]
name = "cumulus-service"
version = "0.1.0"
authors = ["Parity Technologies <admin@parity.io>"]
edition = "2018"
[dependencies]
# Cumulus dependencies
cumulus-consensus = { path = "../consensus" }
cumulus-collator = { path = "../collator" }
cumulus-primitives = { path = "../primitives" }
cumulus-network = { path = "../network" }
# Substrate dependencies
sc-service = { git = "https://github.com/paritytech/substrate", branch = "rococo-branch" }
sc-client-api = { git = "https://github.com/paritytech/substrate", branch = "rococo-branch" }
sp-consensus = { git = "https://github.com/paritytech/substrate", branch = "rococo-branch" }
sp-runtime = { git = "https://github.com/paritytech/substrate", branch = "rococo-branch" }
sp-api = { git = "https://github.com/paritytech/substrate", branch = "rococo-branch" }
sp-core = { git = "https://github.com/paritytech/substrate", branch = "rococo-branch" }
sp-inherents = { git = "https://github.com/paritytech/substrate", branch = "rococo-branch" }
sp-blockchain = { git = "https://github.com/paritytech/substrate", branch = "rococo-branch" }
# Polkadot dependencies
polkadot-collator = { git = "https://github.com/paritytech/polkadot", branch = "rococo-branch" }
polkadot-primitives = { git = "https://github.com/paritytech/polkadot", branch = "rococo-branch" }
polkadot-service = { git = "https://github.com/paritytech/polkadot", branch = "rococo-branch" }
+246
View File
@@ -0,0 +1,246 @@
// Copyright 2020 Parity Technologies (UK) Ltd.
// This file is part of Cumulus.
// 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 Cumulus. If not, see <http://www.gnu.org/licenses/>.
//! Cumulus service
//!
//! Provides functions for starting a collator node or a normal full node.
use cumulus_collator::CollatorBuilder;
use cumulus_network::{DelayedBlockAnnounceValidator, JustifiedBlockAnnounceValidator};
use cumulus_primitives::ParaId;
use polkadot_primitives::v0::{Block as PBlock, CollatorPair};
use polkadot_service::{AbstractClient, RuntimeApiCollection};
use sc_client_api::{Backend as BackendT, BlockBackend, Finalizer, UsageProvider};
use sc_service::{Configuration, Role, TaskManager};
use sp_blockchain::{HeaderBackend, Result as ClientResult};
use sp_consensus::{BlockImport, Environment, Error as ConsensusError, Proposer, SyncOracle};
use sp_core::crypto::Pair;
use sp_inherents::InherentDataProviders;
use sp_runtime::traits::{BlakeTwo256, Block as BlockT};
use std::{marker::PhantomData, sync::Arc};
/// Parameters given to [`start_collator`].
pub struct StartCollatorParams<'a, Block: BlockT, PF, BI, BS, Client> {
pub para_id: ParaId,
pub proposer_factory: PF,
pub inherent_data_providers: InherentDataProviders,
pub block_import: BI,
pub block_status: Arc<BS>,
pub announce_block: Arc<dyn Fn(Block::Hash, Vec<u8>) + Send + Sync>,
pub client: Arc<Client>,
pub block_announce_validator: DelayedBlockAnnounceValidator<Block>,
pub task_manager: &'a mut TaskManager,
pub polkadot_config: Configuration,
pub collator_key: Arc<CollatorPair>,
}
/// Start a collator node for a parachain.
///
/// A collator is similar to a validator in a normal blockchain.
/// It is reponsible for producing blocks and sending the blocks to a
/// parachain validator for validation and inclusion into the relay chain.
pub fn start_collator<'a, Block, PF, BI, BS, Client, Backend>(
StartCollatorParams {
para_id,
proposer_factory,
inherent_data_providers,
block_import,
block_status,
announce_block,
client,
block_announce_validator,
task_manager,
polkadot_config,
collator_key,
}: StartCollatorParams<'a, Block, PF, BI, BS, Client>,
) -> sc_service::error::Result<()>
where
Block: BlockT,
PF: Environment<Block> + Send + 'static,
BI: BlockImport<
Block,
Error = ConsensusError,
Transaction = <PF::Proposer as Proposer<Block>>::Transaction,
> + Send
+ Sync
+ 'static,
BS: BlockBackend<Block> + Send + Sync + 'static,
Client: Finalizer<Block, Backend>
+ UsageProvider<Block>
+ HeaderBackend<Block>
+ Send
+ Sync
+ BlockBackend<Block>
+ 'static,
for<'b> &'b Client: BlockImport<Block>,
Backend: BackendT<Block> + 'static,
{
let builder = CollatorBuilder::new(
proposer_factory,
inherent_data_providers,
block_import,
block_status,
para_id,
client,
announce_block,
block_announce_validator,
);
let (polkadot_future, polkadot_task_manager) =
polkadot_collator::start_collator(builder, para_id, collator_key, polkadot_config)?;
task_manager
.spawn_essential_handle()
.spawn("polkadot", polkadot_future);
task_manager.add_child(polkadot_task_manager);
Ok(())
}
/// Parameters given to [`start_full_node`].
pub struct StartFullNodeParams<'a, Block: BlockT, Client> {
pub polkadot_config: Configuration,
pub collator_key: Arc<CollatorPair>,
pub para_id: ParaId,
pub block_announce_validator: DelayedBlockAnnounceValidator<Block>,
pub client: Arc<Client>,
pub announce_block: Arc<dyn Fn(Block::Hash, Vec<u8>) + Send + Sync>,
pub task_manager: &'a mut TaskManager,
}
/// Start a full node for a parachain.
///
/// A full node will only sync the given parachain and will follow the
/// tip of the chain.
pub fn start_full_node<Block, Client, Backend>(
StartFullNodeParams {
polkadot_config,
collator_key,
para_id,
block_announce_validator,
client,
announce_block,
task_manager,
}: StartFullNodeParams<Block, Client>,
) -> sc_service::error::Result<()>
where
Block: BlockT,
Client: Finalizer<Block, Backend>
+ UsageProvider<Block>
+ Send
+ Sync
+ BlockBackend<Block>
+ 'static,
for<'a> &'a Client: BlockImport<Block>,
Backend: BackendT<Block> + 'static,
{
let is_light = matches!(polkadot_config.role, Role::Light);
let (polkadot_task_manager, pclient, handles) = if is_light {
Err("Light client not supported.".into())
} else {
polkadot_service::build_full(
polkadot_config,
Some((collator_key.public(), para_id)),
None,
false,
6000,
None,
)
}?;
let polkadot_network = handles
.polkadot_network
.expect("Polkadot service is started; qed");
pclient.execute_with(InitParachainFullNode {
block_announce_validator,
para_id,
polkadot_sync_oracle: Box::new(polkadot_network),
announce_block,
client,
task_manager,
_phantom: PhantomData,
})?;
task_manager.add_child(polkadot_task_manager);
Ok(())
}
/// Prepare the parachain's node condifugration
///
/// This function will disable the default announcement of Substrate for the parachain in favor
/// of the one of Cumulus.
pub fn prepare_node_config(mut parachain_config: Configuration) -> Configuration {
parachain_config.announce_block = false;
parachain_config
}
struct InitParachainFullNode<'a, Block: BlockT, Client, Backend> {
block_announce_validator: DelayedBlockAnnounceValidator<Block>,
para_id: ParaId,
polkadot_sync_oracle: Box<dyn SyncOracle + Send>,
announce_block: Arc<dyn Fn(Block::Hash, Vec<u8>) + Send + Sync>,
client: Arc<Client>,
task_manager: &'a mut TaskManager,
_phantom: PhantomData<Backend>,
}
impl<'a, Block, Client, Backend> polkadot_service::ExecuteWithClient
for InitParachainFullNode<'a, Block, Client, Backend>
where
Block: BlockT,
Client: Finalizer<Block, Backend>
+ UsageProvider<Block>
+ Send
+ Sync
+ BlockBackend<Block>
+ 'static,
for<'b> &'b Client: BlockImport<Block>,
Backend: BackendT<Block> + 'static,
{
type Output = ClientResult<()>;
fn execute_with_client<PClient, Api, PBackend>(self, client: Arc<PClient>) -> Self::Output
where
<Api as sp_api::ApiExt<PBlock>>::StateBackend: sp_api::StateBackend<BlakeTwo256>,
PBackend: sc_client_api::Backend<PBlock>,
PBackend::State: sp_api::StateBackend<BlakeTwo256>,
Api: RuntimeApiCollection<StateBackend = PBackend::State>,
PClient: AbstractClient<PBlock, PBackend, Api = Api> + 'static,
{
self.block_announce_validator
.set(Box::new(JustifiedBlockAnnounceValidator::new(
client.clone(),
self.para_id,
self.polkadot_sync_oracle,
)));
let future = cumulus_consensus::follow_polkadot(
self.para_id,
self.client,
client,
self.announce_block,
)?;
self.task_manager
.spawn_essential_handle()
.spawn("cumulus-consensus", future);
Ok(())
}
}