::from(
+ "Unable to produce collation",
+ ))
+ }
+ }
+
+ /// Get the underlying collator service.
+ pub fn collator_service(&self) -> &CS {
+ &self.collator_service
+ }
+}
+
+/// A claim on an Aura slot.
+pub struct SlotClaim {
+ author_id: NimbusId,
+ timestamp: Timestamp,
+}
+
+impl SlotClaim {
+ /// Create a slot-claim from the given author public key, slot, and timestamp.
+ ///
+ /// This does not check whether the author actually owns the slot or the timestamp
+ /// falls within the slot.
+ pub fn unchecked(author_id: NimbusId, timestamp: Timestamp) -> Self
+ where
+ P: Pair,
+ P::Public: Codec,
+ P::Signature: Codec,
+ {
+ SlotClaim {
+ author_id,
+ timestamp,
+ }
+ }
+
+ /// Get the timestamp corresponding to the relay-chain slot this claim was
+ /// generated against.
+ pub fn timestamp(&self) -> Timestamp {
+ self.timestamp
+ }
+
+ /// Get the timestamp corresponding to the relay-chain slot this claim was
+ /// generated against.
+ pub fn author_id(&self) -> NimbusId {
+ self.author_id.clone()
+ }
+}
+
+/// Seal a block with a signature in the header.
+pub fn seal(
+ pre_sealed: B,
+ storage_changes: StorageChanges>,
+ author_pub: &P::Public,
+ keystore: &KeystorePtr,
+) -> Result, Box>
+where
+ P: Pair,
+ P::Signature: Codec + TryFrom>,
+ P::Public: AppPublic,
+{
+ let (pre_header, body) = pre_sealed.deconstruct();
+ let pre_hash = pre_header.hash();
+ let block_number = *pre_header.number();
+
+ // seal the block.
+ let block_import_params = {
+ //let seal_digest = aura_internal::seal::<_, P>(&pre_hash, &author_pub, keystore).map_err(Box::new)?;
+ let seal_digest = crate::collators::seal_header::(
+ &pre_header,
+ keystore,
+ &author_pub.to_raw_vec(),
+ &sp_core::sr25519::CRYPTO_ID,
+ );
+ let mut block_import_params = BlockImportParams::new(BlockOrigin::Own, pre_header);
+ block_import_params.post_digests.push(seal_digest);
+ block_import_params.body = Some(body);
+ block_import_params.state_action =
+ StateAction::ApplyChanges(sc_consensus::StorageChanges::Changes(storage_changes));
+ block_import_params.fork_choice = Some(ForkChoiceStrategy::LongestChain);
+ block_import_params
+ };
+ let post_hash = block_import_params.post_hash();
+
+ tracing::info!(
+ target: crate::LOG_TARGET,
+ "🔖 Pre-sealed block for proposal at {}. Hash now {:?}, previously {:?}.",
+ block_number,
+ post_hash,
+ pre_hash,
+ );
+
+ Ok(block_import_params)
+}
diff --git a/client/consensus/nimbus-consensus/src/collators.rs b/client/consensus/nimbus-consensus/src/collators.rs
index 5d3c252..c336693 100644
--- a/client/consensus/nimbus-consensus/src/collators.rs
+++ b/client/consensus/nimbus-consensus/src/collators.rs
@@ -17,157 +17,48 @@
//! Stock, pure Nimbus collators.
//!
//! This includes the [`basic`] collator, which only builds on top of the most recently
-//! included parachain block, as well as the [`lookahead`] collator, which prospectively
-//! builds on parachain blocks which have not yet been included in the relay chain.
+//! included parachain block, the [`lookahead`] collator, which prospectively builds on
+//! parachain blocks which have not yet been included in the relay chain, and the
+//! [`slot_based`] collator, which uses a slot-based block production approach with a
+//! separate block builder task that is timed by the parachain's slot duration.
pub mod basic;
pub mod lookahead;
+pub mod slot_based;
-use crate::*;
-use cumulus_client_collator::service::ServiceInterface as CollatorServiceInterface;
-use cumulus_client_consensus_common::{ParachainBlockImportMarker, ParachainCandidate};
-use cumulus_client_consensus_proposer::ProposerInterface;
+use crate::{collator::SlotClaim, *};
+use async_backing_primitives::Slot;
+use async_backing_primitives::UnincludedSegmentApi;
+use cumulus_client_consensus_common::{ParentSearchParams, PotentialParent};
use cumulus_primitives_core::{
- relay_chain::{OccupiedCoreAssumption, ValidationCodeHash},
- ParachainBlockData,
+ relay_chain::{Header as RelayHeader, OccupiedCoreAssumption, ValidationCodeHash},
+ ParaId,
};
-use cumulus_primitives_parachain_inherent::ParachainInherentData;
-use futures::prelude::*;
-use log::{debug, info};
-use nimbus_primitives::{CompatibleDigestItem, DigestsProvider, NimbusId, NIMBUS_KEY_ID};
-use polkadot_node_primitives::{Collation, MaybeCompressedPoV};
+use cumulus_relay_chain_interface::{OverseerHandle, RelayChainInterface};
+use log::debug;
+use nimbus_primitives::{CompatibleDigestItem, NimbusId, NIMBUS_KEY_ID};
+use parity_scale_codec::Codec;
+use polkadot_node_subsystem::messages::{CollatorProtocolMessage, RuntimeApiRequest};
+use polkadot_node_subsystem_util::runtime::ClaimQueueSnapshot;
use polkadot_primitives::Hash as RelayHash;
-use sc_consensus::{BlockImport, BlockImportParams};
-use sp_application_crypto::ByteArray;
-use sp_consensus::{BlockOrigin, Proposal};
-use sp_core::{crypto::CryptoTypeId, sr25519};
-use sp_inherents::InherentData;
-use sp_keystore::Keystore;
+use sp_api::{ApiExt, RuntimeApiInfo};
+use sp_core::crypto::CryptoTypeId;
+use sp_keystore::{Keystore, KeystorePtr};
use sp_runtime::{
traits::{Block as BlockT, Header as HeaderT},
DigestItem,
};
+use sp_timestamp::Timestamp;
use std::convert::TryInto;
-use std::error::Error;
-use std::time::Duration;
-/// Propose, seal, and import a block, packaging it into a collation.
-///
-/// Provide the slot to build at as well as any other necessary pre-digest logs,
-/// the inherent data, and the proposal duration and PoV size limits.
-///
-/// The Aura pre-digest should not be explicitly provided and is set internally.
-///
-/// This does not announce the collation to the parachain network or the relay chain.
-pub(crate) async fn collate(
- additional_digests_provider: &ADP,
- author_id: NimbusId,
- block_import: &mut BI,
- collator_service: &CS,
- keystore: &dyn Keystore,
- parent_header: &Block::Header,
- proposer: &mut Proposer,
- inherent_data: (ParachainInherentData, InherentData),
- proposal_duration: Duration,
- max_pov_size: usize,
-) -> Result)>, Box>
-where
- ADP: DigestsProvider::Hash> + 'static,
- Block: BlockT,
- BI: BlockImport + ParachainBlockImportMarker + Send + Sync + 'static,
- CS: CollatorServiceInterface,
- Proposer: ProposerInterface + Send + Sync + 'static,
-{
- let mut logs = vec![CompatibleDigestItem::nimbus_pre_digest(author_id.clone())];
- logs.extend(
- additional_digests_provider.provide_digests(author_id.clone(), parent_header.hash()),
- );
-
- let maybe_proposal = proposer
- .propose(
- parent_header,
- &inherent_data.0,
- inherent_data.1,
- sp_runtime::generic::Digest { logs },
- proposal_duration,
- Some(max_pov_size),
- )
- .await
- .map_err(|e| Box::new(e) as Box)?;
-
- let Proposal {
- block,
- storage_changes,
- proof,
- } = match maybe_proposal {
- None => return Ok(None),
- Some(p) => p,
- };
-
- let (header, extrinsics) = block.clone().deconstruct();
-
- let sig_digest = seal_header::(
- &header,
- keystore,
- &author_id.to_raw_vec(),
- &sr25519::CRYPTO_ID,
- );
-
- let mut block_import_params = BlockImportParams::new(BlockOrigin::Own, header.clone());
- block_import_params.post_digests.push(sig_digest.clone());
- block_import_params.body = Some(extrinsics.clone());
- block_import_params.state_action = sc_consensus::StateAction::ApplyChanges(
- sc_consensus::StorageChanges::Changes(storage_changes),
- );
- // The collator should follow the longest chain
- block_import_params.fork_choice = Some(sc_consensus::ForkChoiceStrategy::LongestChain);
-
- let post_hash = block_import_params.post_hash();
-
- // Print the same log line as slots (aura and babe)
- info!(
- "🔖 Sealed block for proposal at {}. Hash now {:?}, previously {:?}.",
- *header.number(),
- &post_hash,
- header.hash(),
- );
-
- block_import
- .import_block(block_import_params)
- .map_err(|e| Box::new(e) as Box)
- .await?;
-
- // Compute info about the block after the digest is added
- let mut post_header = header.clone();
- post_header.digest_mut().logs.push(sig_digest.clone());
- let post_block = Block::new(post_header, extrinsics);
-
- if let Some((collation, block_data)) = collator_service.build_collation(
- parent_header,
- post_hash,
- ParachainCandidate {
- block: post_block,
- proof,
- },
- ) {
- block_data.log_size_info();
-
- if let MaybeCompressedPoV::Compressed(ref pov) = collation.proof_of_validity {
- tracing::info!(
- target: crate::LOG_TARGET,
- "Compressed PoV size: {}kb",
- pov.block_data.0.len() as f64 / 1024f64,
- );
- }
-
- Ok(Some((collation, block_data)))
- } else {
- Err(
- Box::::from("Unable to produce collation")
- as Box,
- )
- }
-}
+// This is an arbitrary value which is likely guaranteed to exceed any reasonable
+// limit, as it would correspond to 30 non-included blocks.
+//
+// Since we only search for parent blocks which have already been imported,
+// we can guarantee that all imported blocks respect the unincluded segment
+// rules specified by the parachain's runtime and thus will never be too deep. This is just an extra
+// sanity check.
+const PARENT_SEARCH_DEPTH: usize = 30;
pub(crate) fn seal_header(
header: &Block::Header,
@@ -250,3 +141,311 @@ async fn check_validation_code_or_log(
}
}
}
+
+/// Holds a relay parent and its descendants.
+pub struct RelayParentData {
+ /// The relay parent block header
+ relay_parent: RelayHeader,
+ /// Ordered collection of descendant block headers, from oldest to newest
+ descendants: Vec,
+}
+
+impl RelayParentData {
+ /// Creates a new instance with the given relay parent and no descendants.
+ pub fn new(relay_parent: RelayHeader) -> Self {
+ Self {
+ relay_parent,
+ descendants: Default::default(),
+ }
+ }
+
+ /// Creates a new instance with the given relay parent and descendants.
+ pub fn new_with_descendants(relay_parent: RelayHeader, descendants: Vec) -> Self {
+ Self {
+ relay_parent,
+ descendants,
+ }
+ }
+
+ /// Returns a reference to the relay parent header.
+ pub fn relay_parent(&self) -> &RelayHeader {
+ &self.relay_parent
+ }
+
+ /// Returns the number of descendants.
+ #[cfg(test)]
+ pub fn descendants_len(&self) -> usize {
+ self.descendants.len()
+ }
+
+ /// Consumes the structure and returns a vector containing the relay parent followed by its
+ /// descendants in chronological order. The resulting list should be provided to the parachain
+ /// inherent data.
+ pub fn into_inherent_descendant_list(self) -> Vec {
+ let Self {
+ relay_parent,
+ mut descendants,
+ } = self;
+
+ if descendants.is_empty() {
+ return Default::default();
+ }
+
+ let mut result = vec![relay_parent];
+ result.append(&mut descendants);
+ result
+ }
+}
+
+/// Fetch scheduling lookahead at given relay parent.
+async fn scheduling_lookahead(
+ relay_parent: RelayHash,
+ relay_client: &impl RelayChainInterface,
+) -> Option {
+ let runtime_api_version = relay_client
+ .version(relay_parent)
+ .await
+ .map_err(|e| {
+ tracing::error!(
+ target: super::LOG_TARGET,
+ error = ?e,
+ "Failed to fetch relay chain runtime version.",
+ )
+ })
+ .ok()?;
+
+ let parachain_host_runtime_api_version = runtime_api_version
+ .api_version(
+ &>::ID,
+ )
+ .unwrap_or_default();
+
+ if parachain_host_runtime_api_version
+ < RuntimeApiRequest::SCHEDULING_LOOKAHEAD_RUNTIME_REQUIREMENT
+ {
+ return None;
+ }
+
+ match relay_client.scheduling_lookahead(relay_parent).await {
+ Ok(scheduling_lookahead) => Some(scheduling_lookahead),
+ Err(err) => {
+ tracing::error!(
+ target: crate::LOG_TARGET,
+ ?err,
+ ?relay_parent,
+ "Failed to fetch scheduling lookahead from relay chain",
+ );
+ None
+ }
+ }
+}
+
+/// Use [`cumulus_client_consensus_common::find_potential_parents`] to find parachain blocks that
+/// we can build on. Once a list of potential parents is retrieved, return the last one of the
+/// longest chain.
+async fn find_parent(
+ relay_parent: RelayHash,
+ para_id: ParaId,
+ para_backend: &impl sc_client_api::Backend,
+ relay_client: &impl RelayChainInterface,
+) -> Option<(::Header, PotentialParent)>
+where
+ Block: BlockT,
+{
+ let parent_search_params = ParentSearchParams {
+ relay_parent,
+ para_id,
+ ancestry_lookback: scheduling_lookahead(relay_parent, relay_client)
+ .await
+ .unwrap_or(polkadot_primitives::DEFAULT_SCHEDULING_LOOKAHEAD)
+ .saturating_sub(1) as usize,
+ max_depth: PARENT_SEARCH_DEPTH,
+ ignore_alternative_branches: true,
+ };
+
+ let potential_parents = cumulus_client_consensus_common::find_potential_parents::(
+ parent_search_params,
+ para_backend,
+ relay_client,
+ )
+ .await;
+
+ let potential_parents = match potential_parents {
+ Err(e) => {
+ tracing::error!(
+ target: crate::LOG_TARGET,
+ ?relay_parent,
+ err = ?e,
+ "Could not fetch potential parents to build upon"
+ );
+
+ return None;
+ }
+ Ok(x) => x,
+ };
+
+ let included_block = potential_parents
+ .iter()
+ .find(|x| x.depth == 0)?
+ .header
+ .clone();
+ potential_parents
+ .into_iter()
+ .max_by_key(|a| a.depth)
+ .map(|parent| (included_block, parent))
+}
+
+/// Helper for managing pre-connections to backing groups.
+///
+/// This ensures collators establish connections to backing groups proactively,
+/// even in single-collator scenarios where the collator always owns the slot.
+///
+/// For Nimbus consensus, since there's no round-robin authority rotation like Aura,
+/// we simply check if we have any Nimbus keys and pre-connect if so.
+pub(crate) struct BackingGroupConnectionHelper {
+ keystore: KeystorePtr,
+ overseer_handle: OverseerHandle,
+ /// Track the last slot we connected for to avoid sending duplicate messages.
+ last_slot: Option,
+ /// Whether we are currently connected.
+ connected: bool,
+}
+
+impl BackingGroupConnectionHelper {
+ /// Create a new [`BackingGroupConnectionHelper`].
+ pub fn new(keystore: KeystorePtr, overseer_handle: OverseerHandle) -> Self {
+ Self {
+ keystore,
+ overseer_handle,
+ last_slot: None,
+ connected: false,
+ }
+ }
+
+ /// Update the connection helper with the current slot.
+ ///
+ /// For Nimbus, we check if we have any Nimbus keys and pre-connect if so.
+ /// This is simpler than Aura since Nimbus doesn't have round-robin slot assignment.
+ pub async fn update(&mut self, current_slot: Slot) {
+ // If we already processed this slot, skip
+ if self.last_slot == Some(current_slot) {
+ return;
+ }
+ self.last_slot = Some(current_slot);
+
+ // Check if we have any Nimbus keys
+ let has_keys = self
+ .keystore
+ .keys(NIMBUS_KEY_ID)
+ .map(|keys| !keys.is_empty())
+ .unwrap_or(false);
+
+ if has_keys && !self.connected {
+ tracing::debug!(
+ target: crate::LOG_TARGET,
+ ?current_slot,
+ "Pre-connecting to backing groups",
+ );
+ self.overseer_handle
+ .send_msg(
+ CollatorProtocolMessage::ConnectToBackingGroups,
+ "BackingGroupConnectionHelper",
+ )
+ .await;
+ self.connected = true;
+ } else if !has_keys && self.connected {
+ tracing::debug!(
+ target: crate::LOG_TARGET,
+ ?current_slot,
+ "Disconnecting from backing groups - no keys",
+ );
+ self.overseer_handle
+ .send_msg(
+ CollatorProtocolMessage::DisconnectFromBackingGroups,
+ "BackingGroupConnectionHelper",
+ )
+ .await;
+ self.connected = false;
+ }
+ }
+}
+
+// Returns the claim queue at the given relay parent.
+async fn claim_queue_at(
+ relay_parent: RelayHash,
+ relay_client: &impl RelayChainInterface,
+) -> ClaimQueueSnapshot {
+ // Get `ClaimQueue` from runtime
+ match relay_client.claim_queue(relay_parent).await {
+ Ok(claim_queue) => claim_queue.into(),
+ Err(error) => {
+ tracing::error!(
+ target: crate::LOG_TARGET,
+ ?error,
+ ?relay_parent,
+ "Failed to query claim queue runtime API",
+ );
+ Default::default()
+ }
+ }
+}
+
+// Checks if we own the slot at the given block and whether there
+// is space in the unincluded segment.
+async fn can_build_upon(
+ para_slot: Slot,
+ relay_slot: Slot,
+ timestamp: Timestamp,
+ relay_parent: PHeader,
+ parent_header: Block::Header,
+ included_block: Block::Hash,
+ client: &Client,
+ keystore: &KeystorePtr,
+ force_authoring: bool,
+) -> Option
+where
+ Client: ProvideRuntimeApi,
+ Client::Api: NimbusApi + UnincludedSegmentApi + ApiExt,
+ P: sp_core::Pair,
+ P::Public: Codec,
+ P::Signature: Codec,
+{
+ let runtime_api = client.runtime_api();
+ let author_pub = crate::claim_slot::(
+ keystore,
+ client,
+ &parent_header,
+ &relay_parent,
+ force_authoring,
+ )
+ .await
+ .ok()
+ .flatten()?;
+
+ let parent_hash = parent_header.hash();
+
+ // This function is typically called when we want to build block N. At that point, the
+ // unincluded segment in the runtime is unaware of the hash of block N-1. If the unincluded
+ // segment in the runtime is full, but block N-1 is the included block, the unincluded segment
+ // should have length 0 and we can build. Since the hash is not available to the runtime
+ // however, we need this extra check here.
+ if parent_hash == included_block {
+ return Some(SlotClaim::unchecked::(author_pub, timestamp));
+ }
+
+ let api_version = runtime_api
+ .api_version::>(parent_hash)
+ .ok()
+ .flatten()?;
+
+ let slot = if api_version > 1 {
+ relay_slot
+ } else {
+ para_slot
+ };
+
+ runtime_api
+ .can_build_upon(parent_hash, included_block, slot)
+ .ok()?
+ .then(|| SlotClaim::unchecked::(author_pub, timestamp))
+}
diff --git a/client/consensus/nimbus-consensus/src/collators/basic.rs b/client/consensus/nimbus-consensus/src/collators/basic.rs
index 92c5904..5d624c8 100644
--- a/client/consensus/nimbus-consensus/src/collators/basic.rs
+++ b/client/consensus/nimbus-consensus/src/collators/basic.rs
@@ -69,8 +69,9 @@ pub struct Params {
pub additional_relay_state_keys: Vec>,
/// Force production of the block even if the collator is not eligible
pub force_authoring: bool,
- /// Maximum percentage of POV size to use (0-85)
- pub max_pov_percentage: u8,
+ /// The maximum percentage of the maximum PoV size that the collator can use.
+ /// It will be removed once https://github.com/paritytech/polkadot-sdk/issues/6020 is fixed.
+ pub max_pov_percentage: Option,
/// A builder for inherent data builders.
pub create_inherent_data_providers: CIDP,
/// The collator service used for bundling proposals into collations and announcing
@@ -112,13 +113,13 @@ pub async fn run(
let Params {
additional_digests_provider,
- mut block_import,
+ block_import,
collator_service,
create_inherent_data_providers,
keystore,
collator_peer_id,
para_id,
- mut proposer,
+ proposer,
para_client,
relay_client,
force_authoring,
@@ -126,6 +127,28 @@ pub async fn run(
..
} = params;
+ // Create a Collator instance for inherent data creation and block building
+ type CIDPContext = (PHash, PersistedValidationData, NimbusId);
+ let collator_params = crate::collator::Params {
+ create_inherent_data_providers,
+ block_import,
+ relay_client: relay_client.clone(),
+ keystore: keystore.clone(),
+ para_id,
+ proposer,
+ collator_service,
+ };
+ let mut collator = crate::collator::Collator::<
+ Block,
+ nimbus_primitives::NimbusPair,
+ _,
+ _,
+ _,
+ _,
+ _,
+ CIDPContext,
+ >::new(collator_params);
+
let mut last_processed_slot = 0;
let mut last_relay_chain_block = Default::default();
@@ -155,7 +178,10 @@ pub async fn run(
let parent_hash = parent_header.hash();
- if !collator_service.check_block_status(parent_hash, &parent_header) {
+ if !collator
+ .collator_service()
+ .check_block_status(parent_hash, &parent_header)
+ {
continue;
}
@@ -249,52 +275,67 @@ pub async fn run(
continue;
}
+ let cidp_context = (
+ *request.relay_parent(),
+ validation_data.clone(),
+ nimbus_id.clone(),
+ );
let inherent_data = try_request!(
- create_inherent_data(
- &create_inherent_data_providers,
- para_id,
- parent_header.hash(),
- validation_data,
- &relay_client,
- *request.relay_parent(),
- nimbus_id.clone(),
- Some(timestamp),
- collator_peer_id,
- params.additional_relay_state_keys.clone(),
- )
- .await
+ collator
+ .create_inherent_data(
+ *request.relay_parent(),
+ validation_data,
+ parent_header.hash(),
+ Some(timestamp),
+ params.additional_relay_state_keys.clone(),
+ collator_peer_id,
+ cidp_context,
+ )
+ .await
);
- let allowed_pov_size = {
- // Cap the percentage at 85% (see https://github.com/paritytech/polkadot-sdk/issues/6020)
- let capped_percentage = max_pov_percentage.min(85);
- // Calculate the allowed POV size based on the percentage
- (validation_data.max_pov_size as u128)
- .saturating_mul(capped_percentage as u128)
- .saturating_div(100) as usize
- };
+ let allowed_pov_size = if let Some(max_pov_percentage) = max_pov_percentage {
+ validation_data.max_pov_size * max_pov_percentage / 100
+ } else {
+ // Set the block limit to 85% of the maximum PoV size.
+ //
+ // Once https://github.com/paritytech/polkadot-sdk/issues/6020 issue is
+ // fixed, this should be removed.
+ validation_data.max_pov_size * 85 / 100
+ } as usize;
+
+ // Create slot claim and get additional digests
+ let slot_claim = crate::collator::SlotClaim::unchecked::(
+ nimbus_id.clone(),
+ timestamp,
+ );
+ let additional_digests: Vec<_> = additional_digests_provider
+ .provide_digests(nimbus_id, parent_hash)
+ .into_iter()
+ .collect();
let maybe_collation = try_request!(
- super::collate::(
- &additional_digests_provider,
- nimbus_id,
- &mut block_import,
- &collator_service,
- &*keystore,
- &parent_header,
- &mut proposer,
- inherent_data,
- params.authoring_duration,
- allowed_pov_size,
- )
- .await
+ collator
+ .collate(
+ &parent_header,
+ &slot_claim,
+ additional_digests,
+ inherent_data,
+ params.authoring_duration,
+ allowed_pov_size,
+ )
+ .await
);
if let Some((collation, block_data)) = maybe_collation {
let Some(block_hash) = block_data.blocks().first().map(|b| b.hash()) else {
continue;
};
- let result_sender = Some(collator_service.announce_with_barrier(block_hash));
+ let result_sender = Some(
+ collator
+ .collator_service()
+ .announce_with_barrier(block_hash),
+ );
request.complete(Some(CollationResult {
collation,
result_sender,
diff --git a/client/consensus/nimbus-consensus/src/collators/lookahead.rs b/client/consensus/nimbus-consensus/src/collators/lookahead.rs
index 31869c8..3e8a012 100644
--- a/client/consensus/nimbus-consensus/src/collators/lookahead.rs
+++ b/client/consensus/nimbus-consensus/src/collators/lookahead.rs
@@ -69,8 +69,9 @@ pub struct Params,
/// The underlying keystore, which should contain Aura consensus keys.
pub keystore: KeystorePtr,
/// A handle to the relay-chain client's "Overseer" or task orchestrator.
@@ -152,6 +153,34 @@ where
}
};
+ // Helper for pre-connecting to backing groups (fix for single collator scenarios)
+ let mut connection_helper = super::BackingGroupConnectionHelper::new(
+ params.keystore.clone(),
+ params.overseer_handle.clone(),
+ );
+
+ // Create a Collator instance for inherent data creation and block building
+ type CIDPContext = (PHash, PersistedValidationData, NimbusId);
+ let collator_params = crate::collator::Params {
+ create_inherent_data_providers: params.create_inherent_data_providers,
+ block_import: params.block_import,
+ relay_client: params.relay_client.clone(),
+ keystore: params.keystore.clone(),
+ para_id: params.para_id,
+ proposer: params.proposer,
+ collator_service: params.collator_service,
+ };
+ let mut collator = crate::collator::Collator::<
+ Block,
+ nimbus_primitives::NimbusPair,
+ _,
+ _,
+ _,
+ _,
+ _,
+ CIDPContext,
+ >::new(collator_params);
+
// React to each new relay block
while let Some(relay_parent_header) = import_notifications.next().await {
let relay_parent = relay_parent_header.hash();
@@ -299,6 +328,12 @@ where
let mut parent_hash = initial_parent.hash;
let mut parent_header = initial_parent.header;
let overseer_handle = &mut params.overseer_handle;
+
+ // Proactively connect to backing groups. This is especially important for single
+ // collator setups where we always own the slot and thus would never trigger
+ // pre-connection through the normal code path.
+ connection_helper.update(slot_now).await;
+
for n_built in 0..2 {
// Ask to the runtime if we are authorized to create a new parablock on top of this parent.
// (This will claim the slot internally)
@@ -333,32 +368,31 @@ where
//
// Create inherents data for the next parablock
- let (parachain_inherent_data, other_inherent_data) =
- match crate::create_inherent_data(
- ¶ms.create_inherent_data_providers,
- params.para_id,
- parent_hash,
- &PersistedValidationData {
- parent_head: parent_header.encode().into(),
- relay_parent_number: *relay_parent_header.number(),
- relay_parent_storage_root: *relay_parent_header.state_root(),
- max_pov_size,
- },
- ¶ms.relay_client,
+ let validation_data = PersistedValidationData {
+ parent_head: parent_header.encode().into(),
+ relay_parent_number: *relay_parent_header.number(),
+ relay_parent_storage_root: *relay_parent_header.state_root(),
+ max_pov_size,
+ };
+ let cidp_context = (relay_parent, validation_data.clone(), author_id.clone());
+ let (parachain_inherent_data, other_inherent_data) = match collator
+ .create_inherent_data(
relay_parent,
- author_id.clone(),
+ &validation_data,
+ parent_hash,
Some(timestamp),
- params.collator_peer_id,
params.additional_relay_state_keys.clone(),
+ params.collator_peer_id,
+ cidp_context,
)
.await
- {
- Err(err) => {
- tracing::error!(target: crate::LOG_TARGET, ?err);
- break;
- }
- Ok(x) => x,
- };
+ {
+ Err(err) => {
+ tracing::error!(target: crate::LOG_TARGET, ?err);
+ break;
+ }
+ Ok(x) => x,
+ };
// Compute the hash of the parachain runtime bytecode that we using to build the block.
// The hash will be send to relay validators alongside the candidate.
@@ -377,28 +411,36 @@ where
)
.await;
- let allowed_pov_size = {
- // Cap the percentage at 85% (see https://github.com/paritytech/polkadot-sdk/issues/6020)
- let capped_percentage = params.max_pov_percentage.min(85);
- // Calculate the allowed POV size based on the percentage
- (max_pov_size as u128)
- .saturating_mul(capped_percentage as u128)
- .saturating_div(100) as usize
- };
-
- match super::collate(
- ¶ms.additional_digests_provider,
- author_id,
- &mut params.block_import,
- ¶ms.collator_service,
- keystore,
- &parent_header,
- &mut params.proposer,
- (parachain_inherent_data, other_inherent_data),
- params.authoring_duration,
- allowed_pov_size,
- )
- .await
+ let allowed_pov_size = if let Some(max_pov_percentage) = params.max_pov_percentage {
+ max_pov_size * max_pov_percentage / 100
+ } else {
+ // Set the block limit to 85% of the maximum PoV size.
+ //
+ // Once https://github.com/paritytech/polkadot-sdk/issues/6020 issue is
+ // fixed, this should be removed.
+ max_pov_size * 85 / 100
+ } as usize;
+
+ // Create slot claim and get additional digests
+ let slot_claim = crate::collator::SlotClaim::unchecked::<
+ nimbus_primitives::NimbusPair,
+ >(author_id.clone(), timestamp);
+ let additional_digests: Vec<_> = params
+ .additional_digests_provider
+ .provide_digests(author_id, parent_hash)
+ .into_iter()
+ .collect();
+
+ match collator
+ .collate(
+ &parent_header,
+ &slot_claim,
+ additional_digests,
+ (parachain_inherent_data, other_inherent_data),
+ params.authoring_duration,
+ allowed_pov_size,
+ )
+ .await
{
Ok(Some((collation, block_data))) => {
let Some(new_block_header) =
@@ -412,7 +454,9 @@ where
// Here we are assuming that the import logic protects against equivocations
// and provides sybil-resistance, as it should.
- params.collator_service.announce_block(new_block_hash, None);
+ collator
+ .collator_service()
+ .announce_block(new_block_hash, None);
// TODO: Add PoV export functionality
diff --git a/client/consensus/nimbus-consensus/src/collators/slot_based/block_builder_task.rs b/client/consensus/nimbus-consensus/src/collators/slot_based/block_builder_task.rs
new file mode 100644
index 0000000..b972fce
--- /dev/null
+++ b/client/consensus/nimbus-consensus/src/collators/slot_based/block_builder_task.rs
@@ -0,0 +1,686 @@
+// Copyright Moonsong Labs
+// This file is part of Moonkit.
+
+// Moonkit 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.
+
+// Moonkit 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 Moonkit. If not, see .
+
+use nimbus_primitives::NimbusId;
+use parity_scale_codec::{Codec, Encode};
+
+use cumulus_client_collator::service::ServiceInterface as CollatorServiceInterface;
+use cumulus_client_consensus_common::{self as consensus_common, ParachainBlockImportMarker};
+use cumulus_client_consensus_proposer::ProposerInterface;
+use cumulus_relay_chain_interface::RelayChainInterface;
+use sp_consensus_slots::SlotDuration;
+use sp_runtime::SaturatedConversion;
+
+use super::CollatorMessage;
+use crate::{
+ collator as collator_util,
+ collators::{
+ check_validation_code_or_log,
+ slot_based::{
+ relay_chain_data_cache::{RelayChainData, RelayChainDataCache},
+ slot_timer::{SlotInfo, SlotTimer},
+ },
+ BackingGroupConnectionHelper, RelayParentData,
+ },
+ NimbusApi, LOG_TARGET,
+};
+use async_backing_primitives::{Slot, UnincludedSegmentApi};
+use cumulus_primitives_core::{
+ extract_relay_parent, rpsr_digest, ClaimQueueOffset, CoreSelector, CumulusDigestItem,
+ PersistedValidationData, RelayParentOffsetApi,
+};
+use futures::prelude::*;
+use polkadot_primitives::{
+ Block as RelayBlock, CoreIndex, Hash as RelayHash, Header as RelayHeader, Id as ParaId,
+};
+use sc_client_api::{backend::AuxStore, BlockBackend, BlockOf, UsageProvider};
+use sc_consensus::BlockImport;
+use sc_network_types::PeerId;
+use sp_api::ProvideRuntimeApi;
+use sp_application_crypto::AppPublic;
+use sp_blockchain::HeaderBackend;
+use sp_core::crypto::Pair;
+use sp_inherents::CreateInherentDataProviders;
+use sp_keystore::KeystorePtr;
+use sp_runtime::traits::{Block as BlockT, Header as HeaderT, Member, Zero};
+use std::{collections::VecDeque, sync::Arc, time::Duration};
+
+/// Parameters for [`run_block_builder`].
+pub struct BuilderTaskParams<
+ Block: BlockT,
+ BI,
+ CIDP,
+ Client,
+ Backend,
+ RelayClient,
+ CHP,
+ Proposer,
+ CS,
+ DP,
+> {
+ /// Additional digest provider
+ pub additional_digests_provider: DP,
+ /// Inherent data providers. Only non-consensus inherent data should be provided, i.e.
+ /// the timestamp, slot, and paras inherents should be omitted, as they are set by this
+ /// collator.
+ pub create_inherent_data_providers: CIDP,
+ /// Used to actually import blocks.
+ pub block_import: BI,
+ /// The underlying para client.
+ pub para_client: Arc,
+ /// The para client's backend, used to access the database.
+ pub para_backend: Arc,
+ /// A handle to the relay-chain client.
+ pub relay_client: RelayClient,
+ /// A validation code hash provider, used to get the current validation code hash.
+ pub code_hash_provider: CHP,
+ /// The underlying keystore, which should contain Aura consensus keys.
+ pub keystore: KeystorePtr,
+ /// The collator network peer id.
+ pub collator_peer_id: PeerId,
+ /// The para's ID.
+ pub para_id: ParaId,
+ /// The underlying block proposer this should call into.
+ pub proposer: Proposer,
+ /// The generic collator service used to plug into this consensus engine.
+ pub collator_service: CS,
+ /// The amount of time to spend authoring each block.
+ pub authoring_duration: Duration,
+ /// Channel to send built blocks to the collation task.
+ pub collator_sender: sc_utils::mpsc::TracingUnboundedSender>,
+ /// Slot duration of the relay chain.
+ pub relay_chain_slot_duration: Duration,
+ /// The length of slots in this parachain.
+ /// If the parachain doesn't have slot and rely only on relay slots, set it to None.
+ pub para_slot_duration: Option,
+ /// Offset all time operations by this duration.
+ ///
+ /// This is a time quantity that is subtracted from the actual timestamp when computing
+ /// the time left to enter a new slot. In practice, this *left-shifts* the clock time with the
+ /// intent to keep our "clock" slightly behind the relay chain one and thus reducing the
+ /// likelihood of encountering unfavorable notification arrival timings (i.e. we don't want to
+ /// wait for relay chain notifications because we woke up too early).
+ pub slot_offset: Duration,
+ /// The maximum percentage of the maximum PoV size that the collator can use.
+ /// It will be removed once https://github.com/paritytech/polkadot-sdk/issues/6020 is fixed.
+ pub max_pov_percentage: Option,
+ /// Force production of the block even if the collator is not eligible
+ pub force_authoring: bool,
+ pub additional_relay_state_keys: Vec>,
+}
+
+/// Run block-builder.
+pub fn run_block_builder(
+ params: BuilderTaskParams,
+) -> impl Future