Newer
Older
dispute_req_receiver,
dispute_coordinator_config,
chain_selection_config,
})
};
let (network, system_rpc_tx, tx_handler_controller, network_starter, sync_service) =
service::build_network(service::BuildNetworkParams {
config: &config,
client: client.clone(),
transaction_pool: transaction_pool.clone(),
spawn_handle: task_manager.spawn_handle(),
import_queue,
block_announce_validator_builder: None,
warp_sync_params: Some(WarpSyncParams::WithProvider(warp_sync)),
})?;
if config.offchain_worker.enabled {
use futures::FutureExt;
task_manager.spawn_handle().spawn(
"offchain-workers-runner",
"offchain-work",
sc_offchain::OffchainWorkers::new(sc_offchain::OffchainWorkerOptions {
runtime_api_provider: client.clone(),
keystore: Some(keystore_container.keystore()),
offchain_db: backend.offchain_storage(),
transaction_pool: Some(OffchainTransactionPoolFactory::new(
transaction_pool.clone(),
)),
network_provider: Arc::new(network.clone()),
is_validator: role.is_authority(),
enable_http_requests: false,
custom_extensions: move |_| vec![],
})
.run(client.clone(), task_manager.spawn_handle())
.boxed(),
);
}
let rpc_handlers = service::spawn_tasks(service::SpawnTasksParams {
config,
backend: backend.clone(),
client: client.clone(),
keystore: keystore_container.keystore(),
network: network.clone(),
sync_service: sync_service.clone(),
rpc_builder: Box::new(rpc_extensions_builder),
transaction_pool: transaction_pool.clone(),
task_manager: &mut task_manager,
})?;
if let Some(hwbench) = hwbench {
sc_sysinfo::print_hwbench(&hwbench);
match SUBSTRATE_REFERENCE_HARDWARE.check_hardware(&hwbench) {
Err(err) if role.is_authority() => {
log::warn!(
"⚠️ The hardware does not meet the minimal requirements {} for role 'Authority' find out more at:\n\
https://wiki.polkadot.network/docs/maintain-guides-how-to-validate-polkadot#reference-hardware",
err
},
_ => {},
if let Some(ref mut telemetry) = telemetry {
let telemetry_handle = telemetry.handle();
task_manager.spawn_handle().spawn(
"telemetry_hwbench",
None,
sc_sysinfo::initialize_hwbench_telemetry(telemetry_handle, hwbench),
);
}
}
let (block_import, link_half, babe_link, beefy_links) = import_setup;
let overseer_client = client.clone();
let spawner = task_manager.spawn_handle();
Bastian Köcher
committed
let authority_discovery_service =
// We need the authority discovery if this node is either a validator or running alongside a parachain node.
// Parachains node require the authority discovery for finding relay chain validators for sending
// their PoVs or recovering PoVs.
if role.is_authority() || is_parachain_node.is_running_alongside_parachain_node() {
use futures::StreamExt;
use sc_network::{Event, NetworkEventStream};
Bastian Köcher
committed
let authority_discovery_role = if role.is_authority() {
sc_authority_discovery::Role::PublishAndDiscover(keystore_container.keystore())
} else {
// don't publish our addresses when we're not an authority (collator, cumulus, ..)
sc_authority_discovery::Role::Discover
};
let dht_event_stream =
network.event_stream("authority-discovery").filter_map(|e| async move {
match e {
Event::Dht(e) => Some(e),
_ => None,
}
});
let (worker, service) = sc_authority_discovery::new_worker_and_service_with_config(
sc_authority_discovery::WorkerConfig {
publish_non_global_ips: auth_disc_publish_non_global_ips,
public_addresses: auth_disc_public_addresses,
Bastian Köcher
committed
// Require that authority discovery records are signed.
strict_record_validation: true,
..Default::default()
},
client.clone(),
Bastian Köcher
committed
Box::pin(dht_event_stream),
authority_discovery_role,
prometheus_registry.clone(),
);
task_manager.spawn_handle().spawn(
"authority-discovery-worker",
Some("authority-discovery"),
Box::pin(worker.run()),
);
Some(service)
Bastian Köcher
committed
None
let runtime_client = Arc::new(DefaultSubsystemClient::new(
overseer_client.clone(),
OffchainTransactionPoolFactory::new(transaction_pool.clone()),
));
let overseer_handle = if let Some(authority_discovery_service) = authority_discovery_service {
let (overseer, overseer_handle) = overseer_gen
.generate::<service::SpawnTaskHandle, DefaultSubsystemClient<FullClient>>(
overseer_connector,
sync_service: sync_service.clone(),
registry: prometheus_registry.as_ref(),
spawner,
Bastian Köcher
committed
is_parachain_node,
overseer_message_channel_capacity_override,
Dmitry Markin
committed
req_protocol_names,
Dmitry Markin
committed
peerset_protocol_names,
ext_overseer_args,
)
.map_err(|e| {
gum::error!("Failed to init overseer: {}", e);
let handle = Handle::new(overseer_handle.clone());
{
let handle = handle.clone();
task_manager.spawn_essential_handle().spawn_blocking(
"overseer",
Box::pin(async move {
use futures::{pin_mut, select, FutureExt};
let forward = polkadot_overseer::forward_events(overseer_client, handle);
let forward = forward.fuse();
let overseer_fut = overseer.run().fuse();
pin_mut!(overseer_fut);
pin_mut!(forward);
select! {
() = forward => (),
() = overseer_fut => (),
complete => (),
}
}),
);
!auth_or_collator,
"Precondition congruence (false) is guaranteed by manual checking. qed"
);
if role.is_authority() {
let proposer = sc_basic_authorship::ProposerFactory::new(
transaction_pool.clone(),
prometheus_registry.as_ref(),
telemetry.as_ref().map(|x| x.handle()),
let client_clone = client.clone();
let overseer_handle =
overseer_handle.as_ref().ok_or(Error::AuthoritiesRequireRealOverseer)?.clone();
let slot_duration = babe_link.config().slot_duration();
let babe_config = babe::BabeParams {
keystore: keystore_container.keystore(),
client: client.clone(),
select_chain,
block_import,
env: proposer,
sync_oracle: sync_service.clone(),
justification_sync_link: sync_service.clone(),
create_inherent_data_providers: move |parent, ()| {
let client_clone = client_clone.clone();
let parachain =
polkadot_node_core_parachains_inherent::ParachainsInherentDataProvider::new(
client_clone,
overseer_handle,
parent,
);
let timestamp = sp_timestamp::InherentDataProvider::from_system_time();
let slot =
sp_consensus_babe::inherents::InherentDataProvider::from_timestamp_and_slot_duration(
*timestamp,
slot_duration,
);
Ok((slot, timestamp, parachain))
block_proposal_slot_portion: babe::SlotProportion::new(2f32 / 3f32),
max_block_proposal_slot_portion: None,
telemetry: telemetry.as_ref().map(|x| x.handle()),
let babe = babe::start_babe(babe_config)?;
task_manager.spawn_essential_handle().spawn_blocking("babe", None, babe);
// if the node isn't actively participating in consensus then it doesn't
// need a keystore, regardless of which protocol we use below.
let keystore_opt = if role.is_authority() { Some(keystore_container.keystore()) } else { None };
// beefy is enabled if its notification service exists
if let Some(notification_service) = beefy_notification_service {
Adrian Catangiu
committed
let justifications_protocol_name = beefy_on_demand_justifications_handler.protocol_name();
let network_params = beefy::BeefyNetworkParams {
network: Arc::new(network.clone()),
sync: sync_service.clone(),
Adrian Catangiu
committed
gossip_protocol_name: beefy_gossip_proto_name,
justifications_protocol_name,
Adrian Catangiu
committed
_phantom: core::marker::PhantomData::<Block>,
};
Adrian Catangiu
committed
let payload_provider = beefy_primitives::mmr::MmrRootProvider::new(client.clone());
let beefy_params = beefy::BeefyParams {
client: client.clone(),
backend: backend.clone(),
Adrian Catangiu
committed
payload_provider,
Adrian Catangiu
committed
runtime: client.clone(),
Adrian Catangiu
committed
network_params,
min_block_delta: if chain_spec.is_wococo() { 4 } else { 8 },
prometheus_registry: prometheus_registry.clone(),
links: beefy_links,
Adrian Catangiu
committed
on_demand_justifications_handler: beefy_on_demand_justifications_handler,
is_authority: role.is_authority(),
let gadget = beefy::start_beefy_gadget::<_, _, _, _, _, _, _>(beefy_params);
// BEEFY is part of consensus, if it fails we'll bring the node down with it to make sure it
// is noticed.
task_manager
.spawn_essential_handle()
.spawn_blocking("beefy-gadget", None, gadget);
}
// When offchain indexing is enabled, MMR gadget should also run.
if is_offchain_indexing_enabled {
task_manager.spawn_essential_handle().spawn_blocking(
"mmr-gadget",
None,
MmrGadget::start(
client.clone(),
backend.clone(),
sp_mmr_primitives::INDEXING_PREFIX.to_vec(),
),
);
let config = grandpa::Config {
// FIXME substrate#1578 make this available through chainspec
// Grandpa performance can be improved a bit by tuning this parameter, see:
// https://github.com/paritytech/polkadot/issues/5464
gossip_duration: Duration::from_millis(1000),
justification_generation_period: GRANDPA_JUSTIFICATION_PERIOD,
name: Some(name),
observer_enabled: false,
keystore: keystore_opt,
telemetry: telemetry.as_ref().map(|x| x.handle()),
protocol_name: grandpa_protocol_name,
let enable_grandpa = !disable_grandpa;
if enable_grandpa {
// start the full GRANDPA voter
// NOTE: unlike in substrate we are currently running the full
// GRANDPA voter protocol for all full nodes (regardless of whether
// they're validators or not). at this point the full voter should
// provide better guarantees of block and vote data availability than
// the observer.
let mut voting_rules_builder = grandpa::VotingRulesBuilder::default();
#[cfg(not(feature = "malus"))]
let _malus_finality_delay = None;
if let Some(delay) = _malus_finality_delay {
info!(?delay, "Enabling malus finality delay",);
voting_rules_builder = voting_rules_builder.add(grandpa::BeforeBestBlockBy(delay));
let grandpa_config = grandpa::GrandpaParams {
config,
link: link_half,
network: network.clone(),
sync: sync_service.clone(),
voting_rule: voting_rules_builder.build(),
prometheus_registry: prometheus_registry.clone(),
telemetry: telemetry.as_ref().map(|x| x.handle()),
notification_service: grandpa_notification_service,
offchain_tx_pool_factory: OffchainTransactionPoolFactory::new(transaction_pool.clone()),
task_manager.spawn_essential_handle().spawn_blocking(
"grandpa-voter",
None,
grandpa::run_grandpa_voter(grandpa_config)?,
);
network_starter.start_network();
Ok(NewFull {
task_manager,
client,
overseer_handle,
network,
sync_service,
rpc_handlers,
backend,
})
#[cfg(feature = "full-node")]
macro_rules! chain_ops {
($config:expr, $jaeger_agent:expr, $telemetry_worker_handle:expr) => {{
let telemetry_worker_handle = $telemetry_worker_handle;
let jaeger_agent = $jaeger_agent;
let mut config = $config;
let basics = new_partial_basics(config, jaeger_agent, telemetry_worker_handle)?;
use ::sc_consensus::LongestChain;
// use the longest chain selection, since there is no overseer available
let chain_selection = LongestChain::new(basics.backend.clone());
let service::PartialComponents { client, backend, import_queue, task_manager, .. } =
new_partial::<LongestChain<_, Block>>(&mut config, basics, chain_selection)?;
Ok((client, backend, import_queue, task_manager))
}};
}
/// Builds a new object suitable for chain operations.
#[cfg(feature = "full-node")]
jaeger_agent: Option<std::net::SocketAddr>,
) -> Result<(Arc<FullClient>, Arc<FullBackend>, sc_consensus::BasicQueue<Block>, TaskManager), Error>
{
config.keystore = service::config::KeystoreConfig::InMemory;
if config.chain_spec.is_rococo() ||
config.chain_spec.is_wococo() ||
config.chain_spec.is_versi()
{
chain_ops!(config, jaeger_agent, None)
} else if config.chain_spec.is_kusama() {
chain_ops!(config, jaeger_agent, None)
} else if config.chain_spec.is_westend() {
return chain_ops!(config, jaeger_agent, None)
} else {
chain_ops!(config, jaeger_agent, None)
/// Build a full node.
///
/// The actual "flavor", aka if it will use `Polkadot`, `Rococo` or `Kusama` is determined based on
/// [`IdentifyVariant`] using the chain spec.
pub fn build_full<OverseerGenerator: OverseerGen>(
mut params: NewFullParams<OverseerGenerator>,
) -> Result<NewFull, Error> {
let is_polkadot = config.chain_spec.is_polkadot();
params.overseer_message_channel_capacity_override =
params.overseer_message_channel_capacity_override.map(move |capacity| {
if is_polkadot {
gum::warn!("Channel capacity should _never_ be tampered with on polkadot!");
}
capacity
match config.network.network_backend {
sc_network::config::NetworkBackendType::Libp2p =>
new_full::<_, sc_network::NetworkWorker<Block, Hash>>(config, params),
sc_network::config::NetworkBackendType::Litep2p =>
new_full::<_, sc_network::Litep2pNetworkBackend>(config, params),
}
/// Reverts the node state down to at most the last finalized block.
///
/// In particular this reverts:
/// - `ApprovalVotingSubsystem` data in the parachains-db;
/// - `ChainSelectionSubsystem` data in the parachains-db;
/// - Low level Babe and Grandpa consensus data.
#[cfg(feature = "full-node")]
pub fn revert_backend(
client: Arc<FullClient>,
backend: Arc<FullBackend>,
blocks: BlockNumber,
config: Configuration,
) -> Result<(), Error> {
let best_number = client.info().best_number;
let finalized = client.info().finalized_number;
let revertible = blocks.min(best_number - finalized);
if revertible == 0 {
return Ok(())
}
let number = best_number - revertible;
let hash = client.block_hash_from_id(&BlockId::Number(number))?.ok_or(
sp_blockchain::Error::Backend(format!(
"Unexpected hash lookup failure for block number: {}",
number
)),
)?;
let parachains_db = open_database(&config.database)
.map_err(|err| sp_blockchain::Error::Backend(err.to_string()))?;
revert_approval_voting(parachains_db.clone(), hash)?;
revert_chain_selection(parachains_db, hash)?;
// Revert Substrate consensus related components
babe::revert(client.clone(), backend, blocks)?;
grandpa::revert(client, blocks)?;
Ok(())
}
fn revert_chain_selection(db: Arc<dyn Database>, hash: Hash) -> sp_blockchain::Result<()> {
let config = chain_selection_subsystem::Config {
col_data: parachains_db::REAL_COLUMNS.col_chain_selection_data,
stagnant_check_interval: chain_selection_subsystem::StagnantCheckInterval::never(),
stagnant_check_mode: chain_selection_subsystem::StagnantCheckMode::PruneOnly,
let chain_selection = chain_selection_subsystem::ChainSelectionSubsystem::new(config, db);
.revert_to(hash)
.map_err(|err| sp_blockchain::Error::Backend(err.to_string()))
}
fn revert_approval_voting(db: Arc<dyn Database>, hash: Hash) -> sp_blockchain::Result<()> {
let config = approval_voting_subsystem::Config {
col_approval_data: parachains_db::REAL_COLUMNS.col_approval_data,
slot_duration_millis: Default::default(),
};
let approval_voting = approval_voting_subsystem::ApprovalVotingSubsystem::with_config(
config,
db,
Arc::new(sc_keystore::LocalKeystore::in_memory()),
Box::new(consensus_common::NoNetwork),
approval_voting_subsystem::Metrics::default(),
);
approval_voting
.revert_to(hash)
.map_err(|err| sp_blockchain::Error::Backend(err.to_string()))
}