,
P: Pair,
P::Public: Codec,
P::Signature: Codec,
@@ -235,36 +234,58 @@ where
runtime_api.set_call_context(sp_core::traits::CallContext::Onchain { import: false });
let authorities = runtime_api.authorities(parent_hash).ok()?;
let author_pub = aura_internal::claim_slot::(para_slot, &authorities, keystore).await?;
+ Some(SlotClaim::unchecked::
(author_pub, para_slot, timestamp))
+}
+// Checks if there is space in the unincluded segment.
+async fn can_build_upon(
+ parent_hash: Block::Hash,
+ included_block: Block::Hash,
+ relay_slot: Slot,
+ para_slot: Slot,
+ client: &Client,
+) -> bool
+where
+ Client: ProvideRuntimeApi,
+ Client::Api: AuraUnincludedSegmentApi + ApiExt,
+{
// 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, para_slot, timestamp));
+ return true;
}
- let api_version = runtime_api
+ let runtime_api = client.runtime_api();
+ let Some(api_version) = runtime_api
.api_version::>(parent_hash)
.ok()
- .flatten()?;
+ .flatten()
+ else {
+ return false;
+ };
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, para_slot, timestamp))
+ .ok()
+ .unwrap_or(false)
}
/// Use [`cumulus_client_consensus_common::find_parent_for_building`] to find the best parachain
/// block to build on.
+///
+/// If the best parent does not pass `filter_parent`, walks backwards through ancestors
+/// until finding one that does, or reaching the included block.
async fn find_parent(
relay_parent: RelayHash,
para_id: ParaId,
para_backend: &impl sc_client_api::Backend,
relay_client: &impl RelayChainInterface,
+ filter_parent: impl Fn(&Block::Header) -> bool,
) -> Option>
where
Block: BlockT,
@@ -278,21 +299,21 @@ where
.saturating_sub(1) as usize,
};
- match cumulus_client_consensus_common::find_parent_for_building::(
+ let mut result = match cumulus_client_consensus_common::find_parent_for_building::(
parent_search_params,
para_backend,
relay_client,
)
.await
{
- Ok(Some(result)) => Some(result),
+ Ok(Some(result)) => result,
Ok(None) => {
tracing::warn!(
target: crate::LOG_TARGET,
?relay_parent,
"Could not find parent to build upon.",
);
- None
+ return None;
},
Err(e) => {
tracing::error!(
@@ -301,22 +322,44 @@ where
err = ?e,
"Could not find parent to build upon"
);
- None
+ return None;
},
+ };
+
+ // If the best parent doesn't pass the filter (e.g. it's a middle block in a bundle),
+ // walk backwards towards the included block until we find one that does.
+ // This avoids falling all the way back to the included block when there are valid
+ // last-in-core ancestors closer to the chain tip.
+ while !filter_parent(&result.best_parent_header) {
+ let parent_hash = *result.best_parent_header.parent_hash();
+ match para_backend.blockchain().header(parent_hash) {
+ Ok(Some(header)) => {
+ result.best_parent_header = header;
+ if parent_hash == result.included_header.hash() {
+ break;
+ }
+ },
+ _ => {
+ result.best_parent_header = result.included_header.clone();
+ break;
+ },
+ }
}
+
+ Some(result)
}
#[cfg(test)]
mod tests {
use super::*;
- use crate::collators::{can_build_upon, BackingGroupConnectionHelper};
+ use crate::collators::BackingGroupConnectionHelper;
use codec::Encode;
use cumulus_primitives_aura::Slot;
use cumulus_primitives_core::BlockT;
use cumulus_relay_chain_interface::PHash;
use cumulus_test_client::{
runtime::{Block, Hash},
- Client, DefaultTestClientBuilderExt, InitBlockBuilder, TestClientBuilder,
+ BuildBlockBuilder, Client, DefaultTestClientBuilderExt, TestClientBuilder,
TestClientBuilderExt,
};
use cumulus_test_relay_sproof_builder::RelayStateSproofBuilder;
@@ -326,7 +369,6 @@ mod tests {
use sc_consensus::{BlockImport, BlockImportParams, ForkChoiceStrategy};
use sp_consensus::BlockOrigin;
use sp_keystore::{Keystore, KeystorePtr};
- use sp_timestamp::Timestamp;
use std::sync::{Arc, Mutex};
async fn import_block>(
@@ -355,7 +397,11 @@ mod tests {
async fn build_and_import_block(client: &Client, included: Hash) -> Block {
let sproof = sproof_with_parent_by_hash(client, included);
- let block_builder = client.init_block_builder(None, sproof).block_builder;
+ let block_builder = client
+ .init_block_builder_builder()
+ .with_relay_sproof_builder(sproof)
+ .build()
+ .block_builder;
let block = block_builder.build().unwrap().block;
@@ -384,23 +430,22 @@ mod tests {
/// we are ensuring on the node side that we are are always able to build on the included block.
#[tokio::test]
async fn test_can_build_upon() {
- let (client, keystore) = set_up_components(6);
+ sp_tracing::try_init_simple();
+
+ let (client, _keystore) = set_up_components(6);
let genesis_hash = client.chain_info().genesis_hash;
let mut last_hash = genesis_hash;
// Fill up the unincluded segment tracker in the runtime.
- while can_build_upon::<_, _, sp_consensus_aura::sr25519::AuthorityPair>(
- Slot::from(u64::MAX),
- Slot::from(u64::MAX),
- Timestamp::default(),
+ while can_build_upon::<_, _>(
last_hash,
genesis_hash,
+ Slot::from(u64::MAX),
+ Slot::from(u64::MAX),
&*client,
- &keystore,
)
.await
- .is_some()
{
let block = build_and_import_block(&client, genesis_hash).await;
last_hash = block.header().hash();
@@ -408,17 +453,15 @@ mod tests {
// Blocks were built with the genesis hash set as included block.
// We call `can_build_upon` with the last built block as the included block.
- let result = can_build_upon::<_, _, sp_consensus_aura::sr25519::AuthorityPair>(
- Slot::from(u64::MAX),
- Slot::from(u64::MAX),
- Timestamp::default(),
+ let result = can_build_upon::<_, _>(
last_hash,
last_hash,
+ Slot::from(u64::MAX),
+ Slot::from(u64::MAX),
&*client,
- &keystore,
)
.await;
- assert!(result.is_some());
+ assert!(result);
}
/// Helper to create a mock overseer handle and message recorder
@@ -656,7 +699,7 @@ mod tests {
/// (both top-level and child trie keys) should be included in the relay chain state proof.
///
/// Falls back to an empty request if the runtime API call fails or is not implemented.
-fn get_relay_proof_request(
+pub(crate) fn get_relay_proof_request(
client: &Client,
parent_hash: Block::Hash,
) -> RelayProofRequest
@@ -676,6 +719,7 @@ where
}
/// Holds a relay parent and its descendants.
+#[derive(Clone)]
pub struct RelayParentData {
/// The relay parent block header
relay_parent: RelayHeader,
diff --git a/cumulus/client/consensus/aura/src/collators/slot_based/block_builder_task.rs b/cumulus/client/consensus/aura/src/collators/slot_based/block_builder_task.rs
index 248167e6d995c..7f6b4b4142ccc 100644
--- a/cumulus/client/consensus/aura/src/collators/slot_based/block_builder_task.rs
+++ b/cumulus/client/consensus/aura/src/collators/slot_based/block_builder_task.rs
@@ -15,27 +15,27 @@
// You should have received a copy of the GNU General Public License
// along with Cumulus. If not, see .
-use codec::{Codec, Encode};
-
use super::CollatorMessage;
use crate::{
- collator::{self as collator_util, BuildBlockAndImportParams},
+ collator::{self as collator_util, BuildBlockAndImportParams, Collator, SlotClaim},
collators::{
check_validation_code_or_log,
slot_based::{
- relay_chain_data_cache::{RelayChainData, RelayChainDataCache},
+ relay_chain_data_cache::RelayChainDataCache,
slot_timer::{SlotInfo, SlotTimer},
},
- BackingGroupConnectionHelper, RelayParentData,
+ BackingGroupConnectionHelper, RelayHash, RelayParentData,
},
LOG_TARGET,
};
+use codec::{Codec, Encode};
use cumulus_client_collator::service::ServiceInterface as CollatorServiceInterface;
use cumulus_client_consensus_common::{self as consensus_common, ParachainBlockImportMarker};
+use cumulus_client_proof_size_recording::prepare_proof_size_recording_aux_data;
use cumulus_primitives_aura::{AuraUnincludedSegmentApi, Slot};
use cumulus_primitives_core::{
- extract_relay_parent, rpsr_digest, ClaimQueueOffset, CoreInfo, CoreSelector, CumulusDigestItem,
- KeyToIncludeInRelayProof, PersistedValidationData, RelayParentOffsetApi,
+ BlockBundleInfo, ClaimQueueOffset, CoreInfo, CoreSelector, CumulusDigestItem,
+ PersistedValidationData, RelayParentOffsetApi, TargetBlockRate,
};
use cumulus_relay_chain_interface::RelayChainInterface;
use futures::prelude::*;
@@ -44,19 +44,29 @@ use sc_client_api::{backend::AuxStore, BlockBackend, BlockOf, UsageProvider};
use sc_consensus::BlockImport;
use sc_consensus_aura::SlotDuration;
use sc_network_types::PeerId;
-use sp_api::{ApiExt, ProvideRuntimeApi};
+use sp_api::{ApiExt, ProofRecorder, ProvideRuntimeApi, StorageProof};
use sp_application_crypto::AppPublic;
+use sp_block_builder::BlockBuilder;
use sp_blockchain::HeaderBackend;
use sp_consensus::Environment;
use sp_consensus_aura::AuraApi;
use sp_core::crypto::Pair;
+use sp_externalities::Extensions;
use sp_inherents::CreateInherentDataProviders;
use sp_keystore::KeystorePtr;
use sp_runtime::{
- traits::{Block as BlockT, Header as HeaderT, Member, Zero},
+ traits::{Block as BlockT, HashingFor, Header as HeaderT, Member},
Saturating,
};
-use std::{collections::VecDeque, sync::Arc, time::Duration};
+use sp_trie::{
+ proof_size_extension::{ProofSizeExt, RecordingProofSizeProvider},
+ recorder::IgnoredNodes,
+};
+use std::{
+ collections::VecDeque,
+ sync::Arc,
+ time::{Duration, Instant},
+};
/// Parameters for [`run_block_builder`].
pub struct BuilderTaskParams<
@@ -94,8 +104,6 @@ pub struct BuilderTaskParams<
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.
@@ -131,7 +139,9 @@ where
Client::Api: AuraApi
+ RelayParentOffsetApi
+ AuraUnincludedSegmentApi
- + KeyToIncludeInRelayProof,
+ + TargetBlockRate
+ + BlockBuilder
+ + cumulus_primitives_core::KeyToIncludeInRelayProof,
Backend: sc_client_api::Backend + 'static,
RelayClient: RelayChainInterface + Clone + 'static,
CIDP: CreateInherentDataProviders + 'static,
@@ -139,7 +149,7 @@ where
BI: BlockImport + ParachainBlockImportMarker + Send + Sync + 'static,
Proposer: Environment + Send + Sync + 'static,
CS: CollatorServiceInterface + Send + Sync + 'static,
- CHP: consensus_common::ValidationCodeHashProvider + Send + 'static,
+ CHP: consensus_common::ValidationCodeHashProvider + Send + Sync + 'static,
P: Pair + Send + Sync + 'static,
P::Public: AppPublic + Member + Codec,
P::Signature: TryFrom> + Member + Codec,
@@ -158,18 +168,13 @@ where
collator_service,
collator_sender,
code_hash_provider,
- authoring_duration,
relay_chain_slot_duration,
para_backend,
slot_offset,
max_pov_percentage,
} = params;
- let mut slot_timer = SlotTimer::<_, _, P>::new_with_offset(
- para_client.clone(),
- slot_offset,
- relay_chain_slot_duration,
- );
+ let mut slot_timer = SlotTimer::new_with_offset(slot_offset, relay_chain_slot_duration);
let mut collator = {
let params = collator_util::Params {
@@ -210,7 +215,7 @@ where
loop {
// We wait here until the next slot arrives.
- if slot_timer.wait_until_next_slot().await.is_err() {
+ let Ok(slot_time) = slot_timer.wait_until_next_slot().await else {
tracing::error!(target: LOG_TARGET, "Unable to wait for next slot.");
return;
};
@@ -251,6 +256,7 @@ where
continue;
};
+ // Use the slot calculated from relay parent
let Some(para_slot) = adjust_para_to_relay_parent_slot(
rp_data.relay_parent(),
relay_chain_slot_duration,
@@ -262,72 +268,53 @@ where
let relay_parent = rp_data.relay_parent().hash();
let relay_parent_header = rp_data.relay_parent().clone();
- let Some(parent_search_result) =
- crate::collators::find_parent(relay_parent, para_id, &*para_backend, &relay_client)
- .await
+ let Some(parent_search_result) = crate::collators::find_parent(
+ relay_parent,
+ para_id,
+ &*para_backend,
+ &relay_client,
+ |parent| {
+ // We never want to build on any "middle block" that isn't the last block in a
+ // core.
+ // When the digest item doesn't exist, we are running in compatibility
+ // mode and all parents are valid.
+ CumulusDigestItem::is_last_block_in_core(parent.digest()).unwrap_or(true)
+ },
+ )
+ .await
else {
continue;
};
- let parent_hash = parent_search_result.best_parent_header.hash();
let included_header = parent_search_result.included_header;
- let parent_header = &parent_search_result.best_parent_header;
- // Distance from included block to best parent (unincluded segment length).
+ let initial_parent_hash = parent_search_result.best_parent_header.hash();
+ let initial_parent_header = parent_search_result.best_parent_header;
let unincluded_segment_len =
- parent_header.number().saturating_sub(*included_header.number());
-
- // Retrieve the core.
- let core = match determine_core(
- &mut relay_chain_data_cache,
- &relay_parent_header,
- para_id,
- parent_header,
- relay_parent_offset,
- )
- .await
- {
- Err(()) => {
- tracing::debug!(
- target: LOG_TARGET,
- ?relay_parent,
- "Failed to determine core"
- );
+ initial_parent_header.number().saturating_sub(*included_header.number());
- continue;
- },
- Ok(Some(cores)) => {
- tracing::debug!(
- target: LOG_TARGET,
- ?relay_parent,
- core_selector = ?cores.selector,
- claim_queue_offset = ?cores.claim_queue_offset,
- "Going to claim core",
- );
-
- cores
- },
- Ok(None) => {
- tracing::debug!(
- target: LOG_TARGET,
- ?relay_parent,
- "No core scheduled"
- );
-
- continue;
- },
- };
-
- let Ok(RelayChainData { max_pov_size, last_claimed_core_selector, .. }) =
- relay_chain_data_cache.get_mut_relay_chain_data(relay_parent).await
+ let Ok(max_pov_size) = relay_chain_data_cache
+ .get_mut_relay_chain_data(relay_parent)
+ .await
+ .map(|d| d.max_pov_size)
else {
continue;
};
- slot_timer.update_scheduling(core.total_cores().into());
+ let allowed_pov_size = if let Some(max_pov_percentage) = 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;
// We mainly call this to inform users at genesis if there is a mismatch with the
// on-chain data.
- collator.collator_service().check_block_status(parent_hash, parent_header);
+ collator
+ .collator_service()
+ .check_block_status(initial_parent_hash, &initial_parent_header);
let Ok(relay_slot) =
sc_consensus_babe::find_pre_digest::(&relay_parent_header)
@@ -343,167 +330,518 @@ where
let mut runtime_api = para_client.runtime_api();
runtime_api
.set_call_context(sp_core::traits::CallContext::Onchain { import: false });
- if let Ok(authorities) = runtime_api.authorities(parent_hash) {
+ if let Ok(authorities) = runtime_api.authorities(initial_parent_hash) {
connection_helper.update::(para_slot.slot, &authorities).await;
}
}
- let slot_claim = match crate::collators::can_build_upon::<_, _, P>(
+ let Some(slot_claim) = crate::collators::claim_slot::<_, _, P>(
para_slot.slot,
- relay_slot,
para_slot.timestamp,
- parent_hash,
- included_header_hash,
+ initial_parent_hash,
&*para_client,
&keystore,
)
.await
- {
- Some(slot) => slot,
- None => {
- tracing::debug!(
- target: crate::LOG_TARGET,
- ?unincluded_segment_len,
- relay_parent = ?relay_parent,
- relay_parent_num = %relay_parent_header.number(),
- included_hash = ?included_header_hash,
- included_num = %included_header.number(),
- parent = ?parent_hash,
- slot = ?para_slot.slot,
- "Not building block."
- );
- continue;
- },
+ else {
+ tracing::debug!(
+ target: crate::LOG_TARGET,
+ ?unincluded_segment_len,
+ relay_parent = ?relay_parent,
+ relay_parent_num = %relay_parent_header.number(),
+ included_hash = ?included_header_hash,
+ included_num = %included_header.number(),
+ initial_parent = ?initial_parent_hash,
+ slot = ?para_slot.slot,
+ "Not eligible to claim slot."
+ );
+ continue;
};
tracing::debug!(
target: crate::LOG_TARGET,
?unincluded_segment_len,
- relay_parent = %relay_parent,
+ relay_parent = ?relay_parent,
relay_parent_num = %relay_parent_header.number(),
relay_parent_offset,
- included_hash = %included_header_hash,
+ included_hash = ?included_header_hash,
included_num = %included_header.number(),
- parent = %parent_hash,
+ initial_parent = ?initial_parent_hash,
slot = ?para_slot.slot,
- "Building block."
+ "Claiming slot."
);
- 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: *max_pov_size,
- };
-
- let relay_proof_request =
- super::super::get_relay_proof_request(&*para_client, parent_hash);
-
- let (parachain_inherent_data, other_inherent_data) = match collator
- .create_inherent_data_with_rp_offset(
- relay_parent,
- &validation_data,
- parent_hash,
- slot_claim.timestamp(),
- Some(rp_data),
- relay_proof_request,
- collator_peer_id,
- )
- .await
+ let mut cores = match determine_cores(
+ &mut relay_chain_data_cache,
+ &relay_parent_header,
+ para_id,
+ relay_parent_offset,
+ )
+ .await
{
- Err(err) => {
- tracing::error!(target: crate::LOG_TARGET, ?err);
- break;
+ Ok(Some(core)) => core,
+ Ok(None) => {
+ tracing::debug!(
+ target: crate::LOG_TARGET,
+ relay_parent = ?relay_parent,
+ "No cores scheduled."
+ );
+ continue;
},
- Ok(x) => x,
- };
+ Err(()) => {
+ tracing::error!(
+ target: crate::LOG_TARGET,
+ relay_parent = ?relay_parent,
+ "Failed to determine cores."
+ );
- let validation_code_hash = match code_hash_provider.code_hash_at(parent_hash) {
- None => {
- tracing::error!(target: crate::LOG_TARGET, ?parent_hash, "Could not fetch validation code hash");
break;
},
- Some(v) => v,
};
- check_validation_code_or_log(
- &validation_code_hash,
- para_id,
- &relay_client,
- relay_parent,
+ let number_of_blocks =
+ match para_client.runtime_api().target_block_rate(initial_parent_hash) {
+ Ok(interval) => interval,
+ Err(error) => {
+ tracing::debug!(
+ target: crate::LOG_TARGET,
+ block = ?initial_parent_hash,
+ ?error,
+ "Failed to fetch `slot_schedule`, assuming one block per core"
+ );
+
+ // Backwards compatible we use the number of cores as number of blocks.
+ cores.total_cores()
+ },
+ };
+
+ // In total we want to have at max `number_of_blocks` cores to use.
+ cores.truncate_cores(number_of_blocks);
+ let raw_blocks_per_core = (number_of_blocks / cores.total_cores()).max(1);
+ let left_over_blocks = number_of_blocks % cores.total_cores();
+ let blocks_per_cores = (0..cores.total_cores())
+ .map(|i| {
+ // We distribute the left over blocks across the cores.
+ raw_blocks_per_core + u32::from(i < left_over_blocks)
+ })
+ .collect::>();
+
+ tracing::debug!(
+ target: crate::LOG_TARGET,
+ ?blocks_per_cores,
+ core_indices = ?cores.core_indices(),
+ "Core configuration",
+ );
+
+ let mut pov_parent_header = initial_parent_header;
+ let mut pov_parent_hash = initial_parent_hash;
+ let block_time = relay_chain_slot_duration / number_of_blocks;
+
+ for blocks_per_core in blocks_per_cores {
+ let time_for_core = slot_time.time_left() / cores.cores_left();
+
+ match build_collation_for_core(BuildCollationParams {
+ pov_parent_header,
+ pov_parent_hash,
+ relay_parent_header: &relay_parent_header,
+ relay_parent_hash: relay_parent,
+ max_pov_size,
+ para_id,
+ relay_client: &relay_client,
+ code_hash_provider: &code_hash_provider,
+ slot_claim: &slot_claim,
+ collator_sender: &collator_sender,
+ collator: &mut collator,
+ allowed_pov_size,
+ core_info: cores.core_info(),
+ core_index: cores.core_index(),
+ block_time,
+ blocks_per_core,
+ time_for_core,
+ is_last_core_in_parachain_slot: cores.is_last_core() &&
+ slot_time.is_parachain_slot_ending(para_slot_duration.as_duration()),
+ collator_peer_id,
+ relay_parent_data: rp_data.clone(),
+ total_number_of_blocks: number_of_blocks,
+ included_header_hash,
+ relay_slot,
+ para_slot: para_slot.slot,
+ para_client: &*para_client,
+ })
+ .await
+ {
+ Ok(Some(header)) => {
+ pov_parent_header = header;
+ pov_parent_hash = pov_parent_header.hash();
+ },
+ // Let's wait for the next slot
+ Ok(None) => break,
+ Err(()) => return,
+ }
+
+ if !cores.advance() {
+ break;
+ }
+ }
+ }
+ }
+}
+
+/// Parameters for [`build_collation_for_core`].
+struct BuildCollationParams<
+ 'a,
+ Block: BlockT,
+ P: Pair,
+ RelayClient,
+ BI,
+ CIDP,
+ Proposer,
+ CS,
+ CHP,
+ Client,
+> {
+ pov_parent_header: Block::Header,
+ pov_parent_hash: Block::Hash,
+ relay_parent_header: &'a RelayHeader,
+ relay_parent_hash: RelayHash,
+ max_pov_size: u32,
+ para_id: ParaId,
+ relay_client: &'a RelayClient,
+ code_hash_provider: &'a CHP,
+ slot_claim: &'a SlotClaim,
+ collator_sender: &'a sc_utils::mpsc::TracingUnboundedSender>,
+ collator: &'a mut Collator,
+ allowed_pov_size: usize,
+ core_info: CoreInfo,
+ core_index: CoreIndex,
+ block_time: Duration,
+ blocks_per_core: u32,
+ /// Time allocated for the core.
+ time_for_core: Duration,
+ is_last_core_in_parachain_slot: bool,
+ collator_peer_id: PeerId,
+ relay_parent_data: RelayParentData,
+ total_number_of_blocks: u32,
+ included_header_hash: Block::Hash,
+ relay_slot: cumulus_primitives_aura::Slot,
+ para_slot: cumulus_primitives_aura::Slot,
+ para_client: &'a Client,
+}
+
+/// Build a collation for one core.
+///
+/// One collation can be composed of multiple blocks.
+async fn build_collation_for_core<
+ Block: BlockT,
+ P,
+ RelayClient,
+ BI,
+ CIDP,
+ Proposer,
+ CS,
+ CHP,
+ Client,
+>(
+ BuildCollationParams {
+ pov_parent_header,
+ pov_parent_hash,
+ relay_parent_header,
+ relay_parent_hash,
+ max_pov_size,
+ para_id,
+ relay_client,
+ code_hash_provider,
+ slot_claim,
+ collator_sender,
+ collator,
+ allowed_pov_size,
+ core_info,
+ core_index,
+ block_time,
+ blocks_per_core,
+ time_for_core: slot_time_for_core,
+ is_last_core_in_parachain_slot,
+ collator_peer_id,
+ relay_parent_data,
+ total_number_of_blocks,
+ included_header_hash,
+ relay_slot,
+ para_slot,
+ para_client,
+ }: BuildCollationParams<'_, Block, P, RelayClient, BI, CIDP, Proposer, CS, CHP, Client>,
+) -> Result, ()>
+where
+ RelayClient: RelayChainInterface + 'static,
+ P: Pair,
+ P::Public: AppPublic + Member + Codec,
+ P::Signature: TryFrom> + Member + Codec,
+ CIDP: CreateInherentDataProviders + 'static,
+ CIDP::InherentDataProviders: Send,
+ BI: BlockImport + ParachainBlockImportMarker + Send + Sync + 'static,
+ Proposer: Environment + Send + Sync + 'static,
+ CS: CollatorServiceInterface + Send + Sync + 'static,
+ CHP: consensus_common::ValidationCodeHashProvider + Send + Sync + 'static,
+ Client: ProvideRuntimeApi,
+ Client::Api: AuraUnincludedSegmentApi
+ + ApiExt
+ + cumulus_primitives_core::KeyToIncludeInRelayProof,
+{
+ let core_start = Instant::now();
+
+ let validation_data = PersistedValidationData {
+ parent_head: pov_parent_header.encode().into(),
+ relay_parent_number: *relay_parent_header.number(),
+ relay_parent_storage_root: *relay_parent_header.state_root(),
+ max_pov_size,
+ };
+
+ let Some(validation_code_hash) = code_hash_provider.code_hash_at(pov_parent_hash) else {
+ tracing::error!(
+ target: crate::LOG_TARGET,
+ ?pov_parent_hash,
+ "Could not fetch validation code hash",
+ );
+
+ return Err(());
+ };
+
+ check_validation_code_or_log(&validation_code_hash, para_id, relay_client, relay_parent_hash)
+ .await;
+
+ let mut blocks = Vec::new();
+ let mut proofs = Vec::new();
+ let mut ignored_nodes = IgnoredNodes::default();
+
+ let mut parent_hash = pov_parent_hash;
+ let mut parent_header = pov_parent_header.clone();
+
+ for block_index in 0..blocks_per_core {
+ // Check if we can build the next block
+ if !crate::collators::can_build_upon::(
+ parent_hash,
+ included_header_hash,
+ relay_slot,
+ para_slot,
+ para_client,
+ )
+ .await
+ {
+ tracing::debug!(
+ target: LOG_TARGET,
+ ?parent_hash,
+ ?included_header_hash,
+ "Cannot build next block due to unincluded segment constraints, skipping entire bundle. Will continue at the next slot."
+ );
+
+ return Ok(None);
+ }
+
+ // Create schedule for this block to determine timing decisions
+ let schedule = BlockProductionSchedule::new(
+ block_index,
+ blocks_per_core,
+ total_number_of_blocks,
+ is_last_core_in_parachain_slot,
+ );
+
+ if schedule.should_skip_production() {
+ tracing::debug!(
+ target: LOG_TARGET,
+ "Skipping block production so that the next node is able to import all blocks before its slot."
+ );
+ break;
+ }
+
+ tracing::trace!(
+ target: LOG_TARGET,
+ %block_index,
+ core_index = %core_index.0,
+ "Preparing to build block"
+ );
+
+ let relay_proof_request =
+ crate::collators::get_relay_proof_request::(para_client, parent_hash);
+
+ let (parachain_inherent_data, other_inherent_data) = match collator
+ .create_inherent_data_with_rp_offset(
+ relay_parent_hash,
+ &validation_data,
+ parent_hash,
+ slot_claim.timestamp(),
+ Some(relay_parent_data.clone()),
+ relay_proof_request,
+ collator_peer_id,
)
- .await;
+ .await
+ {
+ Err(err) => {
+ tracing::error!(target: crate::LOG_TARGET, ?err, "Failed to create inherent data.");
+ return Ok(None);
+ },
+ Ok(x) => x,
+ };
- 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;
+ let storage_proof_recorder =
+ ProofRecorder::::with_ignored_nodes(ignored_nodes.clone());
+
+ let proof_size_recorder = RecordingProofSizeProvider::new(storage_proof_recorder.clone());
+
+ let mut extra_extensions = Extensions::default();
+ extra_extensions.register(ProofSizeExt::new(proof_size_recorder.clone()));
+
+ let block_production_start = Instant::now();
+ // The time we have left to spent for the block.
+ let time_left_for_block = slot_time_for_core.saturating_sub(core_start.elapsed()) /
+ (blocks_per_core - block_index) as u32;
+
+ // The first block on a multi-block core gets the full remaining core time so that the
+ // runtime's `FullCore` weight mode can actually be utilized. Subsequent blocks are
+ // capped at `block_time` because they only carry fractional weight.
+ //
+ // Single-block cores (blocks_per_core == 1) go through schedule.authoring_duration()
+ // so that slot handover adjustments (e.g., Shorten) are applied on the last core.
+ let authoring_duration = if block_index == 0 && blocks_per_core > 1 {
+ slot_time_for_core.saturating_sub(core_start.elapsed())
+ } else {
+ schedule.authoring_duration(time_left_for_block, block_time)
+ };
- let adjusted_authoring_duration =
- slot_timer.adjust_authoring_duration(authoring_duration);
- tracing::debug!(target: crate::LOG_TARGET, duration = ?adjusted_authoring_duration, "Adjusted proposal duration.");
+ tracing::trace!(
+ target: LOG_TARGET,
+ ?authoring_duration,
+ "Building block"
+ );
- let Some(adjusted_authoring_duration) = adjusted_authoring_duration else {
- tracing::debug!(
- target: crate::LOG_TARGET,
- ?unincluded_segment_len,
- relay_parent = ?relay_parent,
- relay_parent_num = %relay_parent_header.number(),
- included_hash = ?included_header_hash,
- included_num = %included_header.number(),
- parent = ?parent_hash,
- slot = ?para_slot.slot,
- "Not building block due to insufficient authoring duration."
- );
+ let Ok(Some((built_block, mut import_block))) = collator
+ .build_block(BuildBlockAndImportParams {
+ parent_header: &parent_header,
+ slot_claim,
+ additional_pre_digest: vec![
+ CumulusDigestItem::CoreInfo(core_info.clone()).to_digest_item(),
+ CumulusDigestItem::BlockBundleInfo(BlockBundleInfo {
+ index: block_index as u8,
+ is_last: schedule.block_ends_bundle(),
+ })
+ .to_digest_item(),
+ ],
+ parachain_inherent_data,
+ extra_inherent_data: other_inherent_data,
+ proposal_duration: authoring_duration,
+ max_pov_size: allowed_pov_size,
+ storage_proof_recorder: storage_proof_recorder.into(),
+ extra_extensions,
+ })
+ .await
+ else {
+ tracing::error!(target: crate::LOG_TARGET, "Unable to build block at slot.");
+ return Ok(None);
+ };
- continue;
- };
+ parent_hash = built_block.block.header().hash();
+ parent_header = built_block.block.header().clone();
- let Ok(Some(candidate)) = collator
- .build_block_and_import(BuildBlockAndImportParams {
- parent_header: &parent_header,
- slot_claim: &slot_claim,
- additional_pre_digest: vec![
- CumulusDigestItem::CoreInfo(core.core_info()).to_digest_item()
- ],
- parachain_inherent_data,
- extra_inherent_data: other_inherent_data,
- proposal_duration: adjusted_authoring_duration,
- max_pov_size: allowed_pov_size,
- storage_proof_recorder: None,
- extra_extensions: Default::default(),
- })
- .await
- else {
- tracing::error!(target: crate::LOG_TARGET, "Unable to build block at slot.");
- continue;
- };
+ // Extract and add proof size recordings to the import block
+ let recorded_sizes = proof_size_recorder
+ .recorded_estimations()
+ .into_iter()
+ .map(|size| size as u32)
+ .collect::>();
- let new_block_hash = candidate.block.header().hash();
+ if !recorded_sizes.is_empty() {
+ prepare_proof_size_recording_aux_data(parent_hash, recorded_sizes).for_each(
+ |(k, v)| {
+ import_block.auxiliary.push((k, Some(v)));
+ },
+ );
+ }
- // Announce the newly built block to our peers.
- collator.collator_service().announce_block(new_block_hash, None);
+ if let Err(error) = collator.import_block(import_block).await {
+ tracing::error!(target: crate::LOG_TARGET, ?error, "Failed to import built block.");
+ return Ok(None);
+ }
- *last_claimed_core_selector = Some(core.core_selector());
+ // Announce the newly built block to our peers.
+ collator.collator_service().announce_block(parent_hash, None);
- if let Err(err) = collator_sender.unbounded_send(CollatorMessage {
- relay_parent,
- parent_header: parent_header.clone(),
- parachain_candidate: candidate.into(),
- validation_code_hash,
- core_index: core.core_index(),
- validation_data,
- }) {
- tracing::error!(target: crate::LOG_TARGET, ?err, "Unable to send block to collation task.");
- return;
- }
+ blocks.push(built_block.block);
+ proofs.push(built_block.proof);
+
+ let full_core_digest = CumulusDigestItem::contains_use_full_core(parent_header.digest());
+ let runtime_upgrade_digest = parent_header
+ .digest()
+ .logs
+ .iter()
+ .any(|it| matches!(it, sp_runtime::DigestItem::RuntimeEnvironmentUpdated));
+
+ if full_core_digest || runtime_upgrade_digest {
+ tracing::trace!(
+ target: crate::LOG_TARGET,
+ block_hash = ?parent_hash,
+ time_used_by_block_in_secs = %block_production_start.elapsed().as_secs_f32(),
+ %full_core_digest,
+ %runtime_upgrade_digest,
+ "Stopping block production for core",
+ );
+ break;
+ }
+
+ ignored_nodes.extend(IgnoredNodes::from_storage_proof::>(
+ proofs.last().expect("We just pushed the proof into the vector; qed"),
+ ));
+ ignored_nodes.extend(IgnoredNodes::from_memory_db(built_block.backend_transaction));
+
+ // If there is still time left for the block in the slot, we sleep the rest of the time.
+ // This ensures that we have some steady block rate.
+ if let Some(sleep) = time_left_for_block
+ .checked_sub(block_production_start.elapsed())
+ // Let's not sleep for the last block here, to send out the collation as early as
+ // possible.
+ .filter(|_| !schedule.is_effective_last_block())
+ {
+ tokio::time::sleep(sleep).await;
}
}
+
+ if blocks.is_empty() {
+ tracing::debug!(
+ target: LOG_TARGET,
+ ?core_index,
+ relay_parent = ?relay_parent_hash,
+ "Did not build any blocks, returning"
+ );
+
+ return Ok(None);
+ }
+
+ let proof = StorageProof::merge(proofs);
+
+ tracing::trace!(
+ target: LOG_TARGET,
+ ?core_index,
+ relay_parent = ?relay_parent_hash,
+ blocks = ?blocks.iter().map(|b| b.hash()).collect::>(),
+ "Sending out PoV"
+ );
+
+ if let Err(err) = collator_sender.unbounded_send(CollatorMessage {
+ relay_parent: relay_parent_hash,
+ parent_header: pov_parent_header.clone(),
+ blocks,
+ proof,
+ validation_code_hash,
+ core_index,
+ validation_data,
+ }) {
+ tracing::error!(target: crate::LOG_TARGET, ?err, "Unable to send block to collation task.");
+ Err(())
+ } else {
+ // Now let's sleep for the rest of the core.
+ if let Some(sleep) = slot_time_for_core.checked_sub(core_start.elapsed()) {
+ tokio::time::sleep(sleep).await;
+ }
+
+ Ok(Some(parent_header))
+ }
}
/// Translate the slot of the relay parent to the slot of the parachain.
@@ -617,7 +955,7 @@ where
///
/// The function traverses backwards from the best block until it finds the block at the specified
/// offset, collecting all blocks in between to maintain the chain of ancestry.
-pub(crate) async fn offset_relay_parent_find_descendants(
+pub async fn offset_relay_parent_find_descendants(
relay_chain_data_cache: &mut RelayChainDataCache,
mut relay_header: RelayHeader,
relay_parent_offset: u32,
@@ -631,7 +969,12 @@ where
}
if sc_consensus_babe::contains_epoch_change::(&relay_header) {
- tracing::debug!(target: LOG_TARGET, ?relay_best_block, relay_best_block_number = relay_header.number(), "Relay parent is in previous session.");
+ tracing::debug!(
+ target: LOG_TARGET,
+ ?relay_best_block,
+ relay_best_block_number = relay_header.number(),
+ "Relay parent is in previous session.",
+ );
return Ok(None);
}
@@ -644,7 +987,13 @@ where
.relay_parent_header
.clone();
if sc_consensus_babe::contains_epoch_change::(&next_header) {
- tracing::debug!(target: LOG_TARGET, ?relay_best_block, ancestor = %next_header.hash(), ancestor_block_number = next_header.number(), "Ancestor of best block is in previous session.");
+ tracing::debug!(
+ target: LOG_TARGET,
+ ?relay_best_block, ancestor = %next_header.hash(),
+ ancestor_block_number = next_header.number(),
+ "Ancestor of best block is in previous session.",
+ );
+
return Ok(None);
}
required_ancestors.push_front(next_header.clone());
@@ -668,98 +1017,449 @@ where
Ok(Some(RelayParentData::new_with_descendants(relay_parent, required_ancestors.into())))
}
-/// Return value of [`determine_core`].
-pub(crate) struct Core {
+/// Return value of [`determine_cores`].
+pub struct Cores {
selector: CoreSelector,
claim_queue_offset: ClaimQueueOffset,
- core_index: CoreIndex,
- number_of_cores: u16,
+ core_indices: Vec,
}
-impl Core {
+impl Cores {
/// Returns the current [`CoreInfo`].
- fn core_info(&self) -> CoreInfo {
+ pub fn core_info(&self) -> CoreInfo {
CoreInfo {
selector: self.selector,
claim_queue_offset: self.claim_queue_offset,
- number_of_cores: self.number_of_cores.into(),
+ number_of_cores: (self.core_indices.len() as u16).into(),
}
}
- /// Returns the current [`CoreSelector`].
- pub(crate) fn core_selector(&self) -> CoreSelector {
- self.selector
+ /// Returns the core indices.
+ fn core_indices(&self) -> &[CoreIndex] {
+ &self.core_indices
}
/// Returns the current [`CoreIndex`].
- pub(crate) fn core_index(&self) -> CoreIndex {
- self.core_index
+ pub fn core_index(&self) -> CoreIndex {
+ self.core_indices[self.selector.0 as usize]
+ }
+
+ /// Advance to the next available core.
+ ///
+ /// Returns `false` if there is no core left.
+ fn advance(&mut self) -> bool {
+ if self.selector.0 as usize + 1 < self.core_indices.len() {
+ self.selector.0 += 1;
+ true
+ } else {
+ false
+ }
}
/// Returns the total number of cores.
- pub(crate) fn total_cores(&self) -> u16 {
- self.number_of_cores
+ pub fn total_cores(&self) -> u32 {
+ self.core_indices.len() as u32
+ }
+
+ /// Truncate `cores` to `max_cores`.
+ pub fn truncate_cores(&mut self, max_cores: u32) {
+ self.core_indices.truncate(max_cores as usize);
+ }
+
+ /// Returns the number of cores left.
+ fn cores_left(&self) -> u32 {
+ self.total_cores() - self.selector.0 as u32
+ }
+
+ /// Returns if the current core is the last core.
+ fn is_last_core(&self) -> bool {
+ self.cores_left() == 1
}
}
-/// Determine the core for the given `para_id`.
-pub(crate) async fn determine_core(
+/// Slot handover adjustment strategy based on total block rate.
+///
+/// These adjustments exist because without transaction streaming, the next author
+/// must sequentially import all blocks before building their own. Each variant
+/// uses a different strategy to provide import buffer time.
+// TODO: Once transaction streaming is implemented, this can be removed.
+#[derive(Debug, Clone, Copy)]
+enum SlotHandoverAdjustment {
+ /// 0-1 blocks per slot - no adjustment needed.
+ /// The next author has plenty of time to import.
+ None,
+
+ /// 2-3 blocks per slot (~2-3s block time) - shorten authoring time.
+ Shorten {
+ /// Time adjustment factor of last block authoring time.
+ time_factor: f32,
+ },
+
+ /// >3 blocks per slot (<2s block time) - skip last block.
+ ///
+ /// Block time is too fast for time reduction alone, so we skip
+ /// producing the last block in each parachain slot entirely.
+ Skip,
+}
+
+impl SlotHandoverAdjustment {
+ /// Determine the appropriate adjustment based on total blocks per relay slot and blocks per
+ /// core.
+ fn from_total_blocks(total_blocks: u32, blocks_per_core: u32) -> Self {
+ match total_blocks {
+ 0..=1 => Self::None,
+ 2..=3 if blocks_per_core == 1 || blocks_per_core == total_blocks => {
+ Self::Shorten { time_factor: 0.5 }
+ },
+ _ => Self::Skip,
+ }
+ }
+
+ /// Whether this adjustment skips the last block (vs adjusting time).
+ fn skips_last_block(&self) -> bool {
+ matches!(self, Self::Skip)
+ }
+}
+
+/// Policy object that determines block production timing decisions.
+///
+/// Encapsulates the complex timing logic for block production, making decisions
+/// about when to skip blocks, how long to spend authoring, and when to sleep.
+#[derive(Debug, Clone, Copy)]
+struct BlockProductionSchedule {
+ mode: SlotHandoverAdjustment,
+ block_index: u32,
+ blocks_per_core: u32,
+ is_last_core_in_parachain_slot: bool,
+}
+
+impl BlockProductionSchedule {
+ fn new(
+ block_index: u32,
+ blocks_per_core: u32,
+ total_blocks: u32,
+ is_last_core_in_parachain_slot: bool,
+ ) -> Self {
+ Self {
+ mode: SlotHandoverAdjustment::from_total_blocks(total_blocks, blocks_per_core),
+ block_index,
+ blocks_per_core,
+ is_last_core_in_parachain_slot,
+ }
+ }
+
+ /// Whether this is the actual last block index in the core.
+ fn is_last_block_in_core(&self) -> bool {
+ self.block_index + 1 == self.blocks_per_core
+ }
+
+ /// Whether this is the second-to-last block index.
+ fn is_second_to_last(&self) -> bool {
+ self.block_index + 2 == self.blocks_per_core
+ }
+
+ /// Whether to skip producing this block entirely.
+ ///
+ /// In Bundling mode, we skip the last block in the parachain slot
+ /// to give the next author time to import all previous blocks.
+ fn should_skip_production(&self) -> bool {
+ self.mode.skips_last_block() &&
+ self.is_last_block_in_core() &&
+ self.is_last_core_in_parachain_slot
+ }
+
+ /// Whether this is effectively the last block we'll produce for this core.
+ ///
+ /// Used for `BundleInfo { is_last }` - validators need to know which
+ /// block might be final. Also used for sleep decisions - we don't sleep
+ /// after the last or second-to-last block to speed up the final stretch.
+ ///
+ /// The second-to-last block is always included because:
+ /// 1. In Bundling mode on the last core, we skip the actual last block
+ /// 2. Even when not skipping, avoiding sleep on the last two blocks speeds things up
+ fn is_effective_last_block(&self) -> bool {
+ self.is_last_block_in_core() || self.is_second_to_last()
+ }
+
+ /// Whether the node stops block production after this block for this bundle.
+ ///
+ /// Returns `true` when:
+ /// - This is the last block in the core, OR
+ /// - This is the second-to-last and the actual last will be skipped (Skip mode on the last core
+ /// of the parachain slot).
+ fn block_ends_bundle(&self) -> bool {
+ self.is_last_block_in_core() ||
+ (self.is_second_to_last() &&
+ self.mode.skips_last_block() &&
+ self.is_last_core_in_parachain_slot)
+ }
+
+ /// Compute the authoring duration given available time.
+ fn authoring_duration(&self, time_left: Duration, block_time: Duration) -> Duration {
+ let adjusted = match &self.mode {
+ SlotHandoverAdjustment::Shorten { time_factor }
+ if self.is_last_core_in_parachain_slot =>
+ {
+ time_left.mul_f32(*time_factor)
+ },
+ _ => time_left,
+ };
+
+ block_time.min(adjusted)
+ }
+}
+
+/// Determine the cores for the given `para_id`.
+///
+/// Takes into account the `parent` core to find the next available cores.
+pub async fn determine_cores(
relay_chain_data_cache: &mut RelayChainDataCache,
relay_parent: &RelayHeader,
para_id: ParaId,
- para_parent: &H,
relay_parent_offset: u32,
-) -> Result, ()> {
- let cores_at_offset = &relay_chain_data_cache
+) -> Result, ()> {
+ let claim_queue = &relay_chain_data_cache
.get_mut_relay_chain_data(relay_parent.hash())
.await?
- .claim_queue
- .iter_claims_at_depth_for_para(relay_parent_offset as usize, para_id)
+ .claim_queue;
+
+ let core_indices = claim_queue
+ .iter_claims_at_depth_for_para(relay_parent_offset as _, para_id)
.collect::>();
- let is_new_relay_parent = if para_parent.number().is_zero() {
- true
+ Ok(if core_indices.is_empty() {
+ None
} else {
- match extract_relay_parent(para_parent.digest()) {
- Some(last_relay_parent) => last_relay_parent != relay_parent.hash(),
- None => {
- rpsr_digest::extract_relay_parent_storage_root(para_parent.digest())
- .ok_or(())?
- .0 != *relay_parent.state_root()
- },
+ Some(Cores {
+ selector: CoreSelector(0),
+ claim_queue_offset: ClaimQueueOffset(relay_parent_offset as u8),
+ core_indices,
+ })
+ })
+}
+
+#[cfg(test)]
+mod block_production_schedule_tests {
+ use super::*;
+
+ mod mode_tests {
+ use super::*;
+
+ #[test]
+ fn mode_selection_from_total_blocks() {
+ // 0-1 blocks = None
+ assert!(matches!(
+ SlotHandoverAdjustment::from_total_blocks(0, 1),
+ SlotHandoverAdjustment::None
+ ));
+ assert!(matches!(
+ SlotHandoverAdjustment::from_total_blocks(1, 1),
+ SlotHandoverAdjustment::None
+ ));
+
+ // 2-3 blocks = Shorten with half time
+ assert!(matches!(
+ SlotHandoverAdjustment::from_total_blocks(2, 1),
+ SlotHandoverAdjustment::Shorten { time_factor: 0.5 }
+ ));
+ assert!(matches!(
+ SlotHandoverAdjustment::from_total_blocks(3, 1),
+ SlotHandoverAdjustment::Shorten { time_factor: 0.5 }
+ ));
+
+ assert!(matches!(
+ SlotHandoverAdjustment::from_total_blocks(3, 2),
+ SlotHandoverAdjustment::Skip
+ ));
+
+ // >3 blocks = Skip
+ assert!(matches!(
+ SlotHandoverAdjustment::from_total_blocks(4, 2),
+ SlotHandoverAdjustment::Skip
+ ));
+ assert!(matches!(
+ SlotHandoverAdjustment::from_total_blocks(12, 4),
+ SlotHandoverAdjustment::Skip
+ ));
}
- };
- let core_info = CumulusDigestItem::find_core_info(para_parent.digest());
+ #[test]
+ fn mode_behavior_flags() {
+ assert!(!SlotHandoverAdjustment::None.skips_last_block());
- // If we are using a new relay parent, we can start over from the start.
- let (selector, core_index) = if is_new_relay_parent {
- let Some(core_index) = cores_at_offset.get(0) else { return Ok(None) };
+ let shorten = SlotHandoverAdjustment::Shorten { time_factor: 0.5 };
+ assert!(!shorten.skips_last_block());
- (0, *core_index)
- } else if let Some(core_info) = core_info {
- let selector = core_info.selector.0 as usize + 1;
- let Some(core_index) = cores_at_offset.get(selector) else { return Ok(None) };
+ assert!(SlotHandoverAdjustment::Skip.skips_last_block());
+ }
+ }
- (selector, *core_index)
- } else {
- let last_claimed_core_selector = relay_chain_data_cache
- .get_mut_relay_chain_data(relay_parent.hash())
- .await?
- .last_claimed_core_selector;
+ mod schedule_tests {
+ use super::*;
- let selector = last_claimed_core_selector.map_or(0, |cs| cs.0 as usize) + 1;
- let Some(core_index) = cores_at_offset.get(selector) else { return Ok(None) };
+ #[test]
+ fn skip_production_only_in_fast_mode_last_core_last_block() {
+ // Should skip: Fast mode, last core, last block
+ assert!(BlockProductionSchedule::new(0, 1, 4, true).should_skip_production());
- (selector, *core_index)
- };
+ // Should NOT skip: not last core in parachain slot
+ assert!(!BlockProductionSchedule::new(0, 1, 4, false).should_skip_production());
- Ok(Some(Core {
- selector: CoreSelector(selector as u8),
- core_index,
- claim_queue_offset: ClaimQueueOffset(relay_parent_offset as u8),
- number_of_cores: cores_at_offset.len() as u16,
- }))
+ // Should NOT skip: Medium mode (uses time adjustment instead)
+ assert!(!BlockProductionSchedule::new(0, 1, 3, true).should_skip_production());
+
+ // Should NOT skip: not last block in core
+ assert!(!BlockProductionSchedule::new(0, 2, 4, true).should_skip_production());
+
+ // Should skip: Fast mode, last core, last block
+ assert!(BlockProductionSchedule::new(3, 4, 12, true).should_skip_production());
+ // Should skip: Fast mode, last core, second to last block
+ assert!(!BlockProductionSchedule::new(2, 4, 12, true).should_skip_production());
+
+ // Should NOT skip: Fast mode, not last core, last block
+ assert!(!BlockProductionSchedule::new(3, 4, 12, false).should_skip_production());
+ assert!(!BlockProductionSchedule::new(2, 4, 12, false).should_skip_production());
+ }
+
+ #[test]
+ fn effective_last_block_includes_second_to_last() {
+ // block_index 2 is second-to-last (2+2 == 4), always effective last
+ let schedule = BlockProductionSchedule::new(2, 4, 12, true);
+ assert!(schedule.is_effective_last_block());
+ assert!(!schedule.is_last_block_in_core());
+ assert!(schedule.is_second_to_last());
+
+ // Same config but not last core - second-to-last is STILL effective last
+ // (original logic doesn't gate on is_last_core_in_parachain_slot)
+ let schedule = BlockProductionSchedule::new(2, 4, 12, false);
+ assert!(schedule.is_effective_last_block());
+
+ let schedule = BlockProductionSchedule::new(3, 4, 12, false);
+ assert!(schedule.is_effective_last_block());
+
+ // First block is not effective last
+ let schedule = BlockProductionSchedule::new(0, 4, 12, true);
+ assert!(!schedule.is_effective_last_block());
+
+ // With only 1 block per core, there's no second-to-last
+ let schedule = BlockProductionSchedule::new(0, 1, 3, true);
+ assert!(schedule.is_effective_last_block()); // actual last
+ assert!(!schedule.is_second_to_last());
+ }
+
+ #[test]
+ fn authoring_duration_halved_in_medium_mode() {
+ let time_left = Duration::from_millis(2000);
+ let block_time = Duration::from_millis(3000);
+
+ // Medium mode, last block, 1 block per core -> halved
+ let schedule = BlockProductionSchedule::new(0, 1, 2, true);
+ assert_eq!(
+ schedule.authoring_duration(time_left, block_time),
+ Duration::from_millis(1000) // halved, capped by time_left/2
+ );
+
+ // Medium mode but NOT last block -> full time
+ let schedule = BlockProductionSchedule::new(0, 2, 2, true);
+ assert_eq!(
+ schedule.authoring_duration(time_left, block_time),
+ Duration::from_millis(1000) // halved
+ );
+
+ // Fast mode -> no time adjustment (uses skip instead)
+ let schedule = BlockProductionSchedule::new(0, 1, 4, true);
+ assert_eq!(
+ schedule.authoring_duration(time_left, block_time),
+ Duration::from_millis(2000)
+ );
+ }
+
+ #[test]
+ fn block_ends_bundle_only_on_true_last_block() {
+ // 6 blocks per core, Skip mode, last core:
+ // only the actual last (index 5) and second-to-last (index 4, because last
+ // will be skipped) should return true.
+ assert!(!BlockProductionSchedule::new(0, 6, 12, true).block_ends_bundle());
+ assert!(!BlockProductionSchedule::new(3, 6, 12, true).block_ends_bundle());
+ assert!(BlockProductionSchedule::new(4, 6, 12, true).block_ends_bundle());
+ assert!(BlockProductionSchedule::new(5, 6, 12, true).block_ends_bundle());
+
+ // Same config but NOT last core: second-to-last must NOT end the bundle
+ // (skip only applies on last core).
+ assert!(!BlockProductionSchedule::new(4, 6, 12, false).block_ends_bundle());
+ assert!(BlockProductionSchedule::new(5, 6, 12, false).block_ends_bundle());
+
+ // Shorten mode (2 blocks, 1 per core, last core): no skipping, so only the
+ // actual last block ends the bundle.
+ assert!(BlockProductionSchedule::new(0, 1, 2, true).block_ends_bundle());
+
+ // None mode (1 block total): trivially the last.
+ assert!(BlockProductionSchedule::new(0, 1, 1, true).block_ends_bundle());
+ assert!(BlockProductionSchedule::new(0, 1, 1, false).block_ends_bundle());
+
+ // 2 blocks on 1 core (Shorten mode): only index 1 ends the bundle.
+ assert!(!BlockProductionSchedule::new(0, 2, 2, true).block_ends_bundle());
+ assert!(BlockProductionSchedule::new(1, 2, 2, true).block_ends_bundle());
+ }
+
+ /// This test verifies that the new schedule logic matches the original inline logic
+ /// for various block/core configurations.
+ #[test]
+ fn schedule_matches_original_logic() {
+ // Test various configurations to ensure schedule matches original behavior
+ let test_cases = [
+ // (block_index, blocks_per_core, total_blocks, is_last_core)
+ (0, 1, 1, false), // Normal: 1 block, not last core
+ (0, 1, 1, true), // Normal: 1 block, last core
+ (0, 1, 2, true), // Medium: 2 blocks, last core
+ (0, 1, 3, true), // Medium: 3 blocks, last core
+ (0, 1, 4, true), // Fast: 4 blocks, last core (should skip)
+ (0, 1, 4, false), // Fast: 4 blocks, not last core
+ (0, 2, 6, true), // Fast: 6 blocks, 2 per core, block 0
+ (1, 2, 6, true), // Fast: 6 blocks, 2 per core, block 1 (last)
+ (0, 4, 12, true), // Fast: 12 blocks, 4 per core, block 0
+ (2, 4, 12, true), // Fast: 12 blocks, 4 per core, block 2 (second-to-last)
+ (3, 4, 12, true), // Fast: 12 blocks, 4 per core, block 3 (last, should skip)
+ ];
+
+ for (block_index, blocks_per_core, total_blocks, is_last_core) in test_cases {
+ let schedule = BlockProductionSchedule::new(
+ block_index,
+ blocks_per_core,
+ total_blocks,
+ is_last_core,
+ );
+
+ // Original is_last_block_in_core logic
+ let original_is_last = block_index + 1 == blocks_per_core ||
+ (block_index + 2 == blocks_per_core && blocks_per_core > 1);
+
+ // Original skip logic
+ let original_skip =
+ block_index + 1 == blocks_per_core && total_blocks > 3 && is_last_core;
+
+ assert_eq!(
+ schedule.is_effective_last_block(),
+ original_is_last,
+ "is_effective_last_block mismatch for ({}, {}, {}, {})",
+ block_index,
+ blocks_per_core,
+ total_blocks,
+ is_last_core
+ );
+
+ assert_eq!(
+ schedule.should_skip_production(),
+ original_skip,
+ "should_skip_production mismatch for ({}, {}, {}, {})",
+ block_index,
+ blocks_per_core,
+ total_blocks,
+ is_last_core
+ );
+ }
+ }
+ }
}
#[cfg(test)]
diff --git a/cumulus/client/consensus/aura/src/collators/slot_based/block_import.rs b/cumulus/client/consensus/aura/src/collators/slot_based/block_import.rs
index 29cf59a6fadd7..399beae88f2d8 100644
--- a/cumulus/client/consensus/aura/src/collators/slot_based/block_import.rs
+++ b/cumulus/client/consensus/aura/src/collators/slot_based/block_import.rs
@@ -15,14 +15,59 @@
// You should have received a copy of the GNU General Public License
// along with Cumulus. If not, see .
+use crate::LOG_TARGET;
+use codec::{Decode, Encode};
+use cumulus_client_proof_size_recording::prepare_proof_size_recording_aux_data;
+use cumulus_primitives_core::{BlockBundleInfo, CoreInfo, CumulusDigestItem, RelayBlockIdentifier};
use futures::{stream::FusedStream, StreamExt};
+use sc_client_api::{
+ backend::AuxStore,
+ client::{AuxDataOperations, FinalityNotification, PreCommitActions},
+ HeaderBackend,
+};
use sc_consensus::{BlockImport, StateAction};
use sc_utils::mpsc::{tracing_unbounded, TracingUnboundedReceiver, TracingUnboundedSender};
-use sp_api::{ApiExt, CallApiAt, CallContext, Core, ProvideRuntimeApi, StorageProof};
-use sp_runtime::traits::{Block as BlockT, Header as _};
-use sp_trie::proof_size_extension::ProofSizeExt;
+use sp_api::{
+ ApiExt, CallApiAt, CallContext, Core, ProofRecorder, ProofRecorderIgnoredNodes,
+ ProvideRuntimeApi, StorageProof,
+};
+use sp_blockchain::{Error as ClientError, Result as ClientResult};
+use sp_consensus::BlockOrigin;
+use sp_runtime::traits::{Block as BlockT, HashingFor, Header as _};
+use sp_trie::proof_size_extension::{ProofSizeExt, RecordingProofSizeProvider};
use std::sync::Arc;
+/// The aux storage key used to store the ignored nodes for the given block hash.
+fn ignored_nodes_key(block_hash: H) -> Vec {
+ (b"cumulus_slot_based_nodes_to_ignore", block_hash).encode()
+}
+
+/// Prepare a transaction to write the ignored nodes to the aux storage.
+///
+/// Returns the key-value pairs that need to be written to the aux storage.
+fn prepare_ignored_nodes_transaction(
+ block_hash: Block::Hash,
+ ignored_nodes: ProofRecorderIgnoredNodes,
+) -> impl Iterator- , Vec)> {
+ let key = ignored_nodes_key(block_hash);
+ let encoded_nodes = as Encode>::encode(&ignored_nodes);
+
+ [(key, encoded_nodes)].into_iter()
+}
+
+/// Load the ignored nodes associated with a block.
+fn load_ignored_nodes(
+ backend: &B,
+ block_hash: Block::Hash,
+) -> ClientResult>> {
+ match backend.get_aux(&ignored_nodes_key(block_hash))? {
+ None => Ok(None),
+ Some(t) => ProofRecorderIgnoredNodes::::decode(&mut &t[..])
+ .map(Some)
+ .map_err(|e| ClientError::Backend(format!("Failed to decode ignored nodes: {}", e))),
+ }
+}
+
/// Handle for receiving the block and the storage proof from the [`SlotBasedBlockImport`].
///
/// This handle should be passed to [`Params`](super::Params) or can also be dropped if the node is
@@ -46,27 +91,238 @@ impl SlotBasedBlockImportHandle {
}
}
+/// Register the clean up method for cleaning ignored nodes from blocks on which no further blocks
+/// will be imported.
+fn register_ignored_nodes_cleanup(client: Arc)
+where
+ C: PreCommitActions + HeaderBackend + 'static,
+ Block: BlockT,
+{
+ let client_for_closure = client.clone();
+ let on_finality = move |notification: &FinalityNotification| -> AuxDataOperations {
+ // The old finalized block is the parent of the first block in the tree route,
+ // or the parent of the finalized block if the tree route is empty.
+ let old_finalized_hash = notification
+ .tree_route
+ .first()
+ .and_then(|hash| client_for_closure.header(*hash).ok().flatten())
+ .map(|h| *h.parent_hash())
+ .unwrap_or_else(|| *notification.header.parent_hash());
+
+ notification
+ .stale_blocks
+ .iter()
+ // Delete the ignored nodes for all stale blocks.
+ .map(|b| (ignored_nodes_key(b.hash), None))
+ // We can not delete the ignored nodes for the finalized block, because blocks can still
+ // be imported on top of this block. However, once multiple blocks are finalized at
+ // once, blocks on the route to the finalized parent can no longer become parents
+ // either.
+ .chain(
+ notification
+ .tree_route
+ .iter()
+ .copied()
+ .map(|hash| (ignored_nodes_key(hash), None)),
+ )
+ .chain(std::iter::once((ignored_nodes_key(old_finalized_hash), None)))
+ .collect()
+ };
+
+ client.register_finality_action(Box::new(on_finality));
+}
+
/// Special block import for the slot based collator.
-pub struct SlotBasedBlockImport {
+pub struct SlotBasedBlockImport {
inner: BI,
client: Arc,
sender: TracingUnboundedSender<(Block, StorageProof)>,
}
-impl SlotBasedBlockImport {
+impl SlotBasedBlockImport {
/// Create a new instance.
///
/// The returned [`SlotBasedBlockImportHandle`] needs to be passed to the
/// [`Params`](super::Params), so that this block import instance can communicate with the
/// collation task. If the node is not running as a collator, just dropping the handle is fine.
- pub fn new(inner: BI, client: Arc) -> (Self, SlotBasedBlockImportHandle) {
+ pub fn new(inner: BI, client: Arc) -> (Self, SlotBasedBlockImportHandle)
+ where
+ Client: PreCommitActions + HeaderBackend + 'static,
+ {
let (sender, receiver) = tracing_unbounded("SlotBasedBlockImportChannel", 1000);
+ register_ignored_nodes_cleanup(client.clone());
+
(Self { sender, client, inner }, SlotBasedBlockImportHandle { receiver })
}
+
+ /// Get the [`ProofRecorderIgnoredNodes`] for `parent`.
+ ///
+ /// If `parent` was not part of the same block bundle, the [`ProofRecorderIgnoredNodes`] are not
+ /// required and `None` will be returned.
+ fn get_ignored_nodes(
+ &self,
+ parent: Block::Hash,
+ core_info: &CoreInfo,
+ bundle_info: &BlockBundleInfo,
+ relay_block_identifier: &RelayBlockIdentifier,
+ ) -> Option>
+ where
+ Client: AuxStore + HeaderBackend + Send + Sync,
+ {
+ let parent_header = self.client.header(parent).ok().flatten()?;
+ let parent_core_info = CumulusDigestItem::find_core_info(parent_header.digest())?;
+ let parent_bundle_info = CumulusDigestItem::find_block_bundle_info(parent_header.digest())?;
+ let parent_relay_block_identifier =
+ CumulusDigestItem::find_relay_block_identifier(parent_header.digest())?;
+
+ if parent_relay_block_identifier != *relay_block_identifier {
+ tracing::trace!(target: LOG_TARGET, ?parent_relay_block_identifier, ?relay_block_identifier, "Relay block identifier doesn't match");
+ return None;
+ }
+
+ if parent_core_info != *core_info {
+ tracing::trace!(target: LOG_TARGET, ?parent_core_info, ?core_info, "Core info doesn't match");
+ return None;
+ }
+
+ if parent_bundle_info.index.saturating_add(1) != bundle_info.index {
+ tracing::trace!(target: LOG_TARGET, ?parent_bundle_info, ?bundle_info, "Block is not a child, based on the index");
+ return None;
+ }
+
+ match load_ignored_nodes::(&*self.client, parent) {
+ Ok(nodes) => nodes,
+ Err(error) => {
+ tracing::trace!(target: LOG_TARGET, ?parent, ?error, "Failed to load `IgnoredNodes` from aux store");
+ None
+ },
+ }
+ }
+
+ /// Execute the given block and collect the storage proof.
+ ///
+ /// We need to execute the block on this level here, because we are collecting the storage
+ /// proofs and combining them for blocks on the same core. So, blocks on the same core do not
+ /// need to include the same trie nodes multiple times and thus, not wasting storage proof size.
+ ///
+ /// The proof must be recorded in exactly the same manner as during block building, because the
+ /// proof size is tracked via `ProofSizeExt` and affects runtime state. Without identical proof
+ /// recording, the computed state root would differ and block import would fail.
+ fn execute_block_and_collect_storage_proof(
+ &self,
+ params: &mut sc_consensus::BlockImportParams,
+ ) -> Result<(), sp_consensus::Error>
+ where
+ Client: ProvideRuntimeApi
+ + CallApiAt
+ + AuxStore
+ + HeaderBackend
+ + Send
+ + Sync,
+ Client::StateBackend: Send,
+ Client::Api: Core,
+ {
+ let core_info = CumulusDigestItem::find_core_info(params.header.digest());
+ let bundle_info = CumulusDigestItem::find_block_bundle_info(params.header.digest());
+ let relay_block_identifier =
+ CumulusDigestItem::find_relay_block_identifier(params.header.digest());
+
+ let (Some(core_info), Some(bundle_info), Some(relay_block_identifier)) =
+ (core_info, bundle_info, relay_block_identifier)
+ else {
+ tracing::debug!(
+ target: LOG_TARGET,
+ number = ?params.header.number(),
+ "no bundle digests, skipping execute_block_and_collect_storage_proof",
+ );
+ return Ok(());
+ };
+
+ let parent_hash = *params.header.parent_hash();
+
+ let mut nodes_to_ignore = self
+ .get_ignored_nodes(parent_hash, &core_info, &bundle_info, &relay_block_identifier)
+ .unwrap_or_default();
+
+ let recorder = ProofRecorder::::with_ignored_nodes(nodes_to_ignore.clone());
+ let proof_size_recorder = RecordingProofSizeProvider::new(recorder.clone());
+
+ let mut runtime_api = self.client.runtime_api();
+
+ // `record_proof_with_recorder` captures trie accesses, while `ProofSizeExt` replays the
+ // proof-size estimations in the same order they were observed during block building.
+ runtime_api.set_call_context(CallContext::Onchain { import: true });
+ runtime_api.record_proof_with_recorder(recorder.clone());
+ runtime_api.register_extension(ProofSizeExt::new(proof_size_recorder.clone()));
+
+ let block = Block::new(params.header.clone(), params.body.clone().unwrap_or_default());
+
+ tracing::debug!(
+ target: LOG_TARGET,
+ ?parent_hash,
+ number = ?params.header.number(),
+ ?core_info,
+ ?bundle_info,
+ "execute_block_and_collect_storage_proof: calling runtime_api.execute_block",
+ );
+
+ runtime_api
+ .execute_block(parent_hash, block.into())
+ .map_err(|e| Box::new(e) as Box<_>)?;
+
+ let storage_proof =
+ runtime_api.extract_proof().expect("Proof recording was enabled above; qed");
+
+ let state = self.client.state_at(parent_hash).map_err(|e| Box::new(e) as Box<_>)?;
+ let gen_storage_changes = runtime_api
+ .into_storage_changes(&state, parent_hash)
+ .map_err(sp_consensus::Error::ChainLookup)?;
+
+ if params.header.state_root() != &gen_storage_changes.transaction_storage_root {
+ return Err(sp_consensus::Error::Other(Box::new(
+ sp_blockchain::Error::InvalidStateRoot,
+ )));
+ }
+
+ // Extend the ignored nodes with the nodes from the storage proof and the generated
+ // storage changes. This ensures that subsequent blocks in the same bundle don't
+ // redundantly include the same trie nodes in their proof.
+ nodes_to_ignore.extend(ProofRecorderIgnoredNodes::::from_storage_proof::<
+ HashingFor,
+ >(&storage_proof));
+ nodes_to_ignore.extend(ProofRecorderIgnoredNodes::::from_memory_db(
+ gen_storage_changes.transaction.clone(),
+ ));
+
+ let block_hash = params.post_hash();
+ prepare_ignored_nodes_transaction::(block_hash, nodes_to_ignore).for_each(
+ |(k, v)| {
+ params.auxiliary.push((k, Some(v)));
+ },
+ );
+
+ // Extract and store proof size recordings
+ let recorded_sizes = proof_size_recorder
+ .recorded_estimations()
+ .into_iter()
+ .map(|size| size as u32)
+ .collect::>();
+
+ if !recorded_sizes.is_empty() {
+ prepare_proof_size_recording_aux_data(block_hash, recorded_sizes).for_each(|(k, v)| {
+ params.auxiliary.push((k, Some(v)));
+ });
+ }
+
+ params.state_action =
+ StateAction::ApplyChanges(sc_consensus::StorageChanges::Changes(gen_storage_changes));
+
+ Ok(())
+ }
}
-impl Clone for SlotBasedBlockImport {
+impl Clone for SlotBasedBlockImport {
fn clone(&self) -> Self {
Self { inner: self.inner.clone(), client: self.client.clone(), sender: self.sender.clone() }
}
@@ -78,7 +334,8 @@ where
Block: BlockT,
BI: BlockImport + Send + Sync,
BI::Error: Into,
- Client: ProvideRuntimeApi + CallApiAt + Send + Sync,
+ Client:
+ ProvideRuntimeApi + CallApiAt + AuxStore + HeaderBackend + Send + Sync,
Client::StateBackend: Send,
Client::Api: Core,
{
@@ -95,191 +352,13 @@ where
&self,
mut params: sc_consensus::BlockImportParams,
) -> Result {
- // If the channel exists and it is required to execute the block, we will execute the block
- // here. This is done to collect the storage proof and to prevent re-execution, we push
- // downwards the state changes.
- //
- // The following states are ignored:
- // - `StateAction::ApplyChanges`: means that the node produced the block itself or the
- // block was imported via state sync.
- // - `StateAction::Skip`: means that the block should be skipped. This is evident in the
- // context of gap-sync with collators running in non-archive modes. The state of the
- // parent block has already been discarded and therefore any import would fail.
- if !self.sender.is_closed() &&
- !matches!(params.state_action, StateAction::ApplyChanges(_) | StateAction::Skip)
+ if !(params.origin == BlockOrigin::Own ||
+ params.with_state() ||
+ params.state_action.skip_execution_checks())
{
- let mut runtime_api = self.client.runtime_api();
-
- runtime_api.set_call_context(CallContext::Onchain { import: true });
-
- runtime_api.record_proof();
- let recorder = runtime_api
- .proof_recorder()
- .expect("Proof recording is enabled in the line above; qed.");
- runtime_api.register_extension(ProofSizeExt::new(recorder));
-
- let parent_hash = *params.header.parent_hash();
-
- let block = Block::new(params.header.clone(), params.body.clone().unwrap_or_default());
-
- runtime_api
- .execute_block(parent_hash, block.clone().into())
- .map_err(|e| Box::new(e) as Box<_>)?;
-
- let storage_proof =
- runtime_api.extract_proof().expect("Proof recording was enabled above; qed");
-
- let state = self.client.state_at(parent_hash).map_err(|e| Box::new(e) as Box<_>)?;
- let gen_storage_changes = runtime_api
- .into_storage_changes(&state, parent_hash)
- .map_err(sp_consensus::Error::ChainLookup)?;
-
- if params.header.state_root() != &gen_storage_changes.transaction_storage_root {
- return Err(sp_consensus::Error::Other(Box::new(
- sp_blockchain::Error::InvalidStateRoot,
- )));
- }
-
- params.state_action = StateAction::ApplyChanges(sc_consensus::StorageChanges::Changes(
- gen_storage_changes,
- ));
-
- let _ = self.sender.unbounded_send((block, storage_proof));
+ self.execute_block_and_collect_storage_proof(&mut params)?;
}
self.inner.import_block(params).await.map_err(Into::into)
}
}
-
-#[cfg(test)]
-mod tests {
- use super::*;
- use codec::Encode;
- use cumulus_test_client::{
- runtime::Block, DefaultTestClientBuilderExt, InitBlockBuilder, TestClientBuilder,
- TestClientBuilderExt,
- };
- use cumulus_test_relay_sproof_builder::RelayStateSproofBuilder;
- use polkadot_primitives::HeadData;
- use sc_consensus::{BlockImportParams, ImportResult, StateAction};
- use sp_blockchain::HeaderBackend;
- use sp_consensus::BlockOrigin;
-
- fn sproof_with_best_parent(client: &cumulus_test_client::Client) -> RelayStateSproofBuilder {
- let best_hash = client.info().best_hash;
- let header = client.header(best_hash).ok().flatten().expect("No header for best block");
- let mut builder = RelayStateSproofBuilder::default();
- builder.para_id = cumulus_test_client::runtime::PARACHAIN_ID.into();
- builder.included_para_head = Some(HeadData(header.encode()));
- builder
- }
-
- /// Mock inner block import that always succeeds.
- #[derive(Clone)]
- struct MockBlockImport;
-
- #[async_trait::async_trait]
- impl BlockImport for MockBlockImport {
- type Error = sp_consensus::Error;
-
- async fn check_block(
- &self,
- _block: sc_consensus::BlockCheckParams,
- ) -> Result {
- Ok(ImportResult::imported(false))
- }
-
- async fn import_block(
- &self,
- _block: BlockImportParams,
- ) -> Result {
- Ok(ImportResult::imported(true))
- }
- }
-
- /// Regression test for the gap-sync infinite loop issue.
- ///
- /// When a non-archive collator has a block gap of size 1, gap-sync downloads
- /// the block and marks it with `skip_execution: true` (which translates to
- /// `StateAction::Skip`). Before the fix, `SlotBasedBlockImport` would attempt
- /// to execute such blocks, fail with a consensus error ("State already
- /// discarded for parent"), and trigger a chain-sync restart that re-creates
- /// the same gap — leading to an infinite retry loop.
- ///
- /// This test verifies that `StateAction::Skip` blocks are forwarded to the
- /// inner block import without attempting runtime execution.
- #[tokio::test]
- async fn gap_sync_block_with_skip_execution_does_not_attempt_runtime_call() {
- sp_tracing::try_init_simple();
-
- let client = Arc::new(TestClientBuilder::new().build());
-
- // Build a valid block so we have realistic headers/bodies.
- let sproof = sproof_with_best_parent(&client);
- let block_builder_data = client.init_block_builder(None, sproof);
- let block = block_builder_data.block_builder.build().unwrap().block;
-
- let (slot_based_import, mut handle) =
- SlotBasedBlockImport::new(MockBlockImport, client.clone());
-
- // Simulate the gap-sync scenario: a block arrives with StateAction::Skip
- // because the parent state has been pruned.
- let mut params = BlockImportParams::new(BlockOrigin::NetworkInitialSync, block.header);
- params.body = Some(block.extrinsics);
- params.state_action = StateAction::Skip;
- params.import_existing = true;
-
- // Before the fix, this would fail with a consensus error because
- // SlotBasedBlockImport would try to call `runtime_api.execute_block()`
- // on the parent hash whose state is no longer available.
- //
- // After the fix, StateAction::Skip is recognized and the block is
- // forwarded directly to the inner import without execution.
- let result = slot_based_import.import_block(params).await;
- assert!(result.is_ok(), "Gap-sync block with StateAction::Skip must not fail: {result:?}");
-
- // The channel must be empty — execution should have been skipped entirely,
- // so no (block, proof) was sent. This is the key assertion: without the
- // StateAction::Skip guard, execute_block() would run and send a message.
- //
- // Drop the sender side so the channel closes, then verify no message was queued.
- drop(slot_based_import);
- assert!(
- handle.receiver.next().await.is_none(),
- "No block+proof should be sent through the channel for StateAction::Skip"
- );
- }
-
- /// Verify that `StateAction::Execute` still triggers runtime execution.
- ///
- /// This complements the gap-sync regression test by ensuring we did not
- /// accidentally disable execution for normal blocks.
- #[tokio::test]
- async fn normal_block_with_execute_action_triggers_runtime_execution() {
- sp_tracing::try_init_simple();
-
- let client = Arc::new(TestClientBuilder::new().build());
-
- let sproof = sproof_with_best_parent(&client);
- let block_builder_data = client.init_block_builder(None, sproof);
- let block = block_builder_data.block_builder.build().unwrap().block;
-
- let (slot_based_import, mut handle) =
- SlotBasedBlockImport::new(MockBlockImport, client.clone());
-
- // Normal import with StateAction::Execute should trigger execution
- // and send the block + proof through the channel.
- let mut params =
- BlockImportParams::new(BlockOrigin::NetworkInitialSync, block.header.clone());
- params.body = Some(block.extrinsics.clone());
- params.state_action = StateAction::Execute;
-
- let result = slot_based_import.import_block(params).await;
- assert!(result.is_ok(), "Normal block import should succeed: {result:?}");
-
- // The block and proof should have been sent through the channel,
- // confirming that execution actually happened.
- let (received_block, _proof) = handle.next().await;
- assert_eq!(*received_block.header(), block.header);
- }
-}
diff --git a/cumulus/client/consensus/aura/src/collators/slot_based/collation_task.rs b/cumulus/client/consensus/aura/src/collators/slot_based/collation_task.rs
index c0fa0846de001..a77fb2b89531a 100644
--- a/cumulus/client/consensus/aura/src/collators/slot_based/collation_task.rs
+++ b/cumulus/client/consensus/aura/src/collators/slot_based/collation_task.rs
@@ -121,20 +121,19 @@ async fn handle_collation_message collation,
None => {
- tracing::warn!(target: LOG_TARGET, %hash, ?number, ?core_index, "Unable to build collation.");
+ tracing::warn!(target: LOG_TARGET, ?core_index, "Unable to build collation.");
return;
},
};
@@ -165,6 +164,7 @@ async fn handle_collation_message>(),
"Compressed PoV size: {}kb",
pov.block_data.0.len() as f64 / 1024f64,
);
@@ -183,7 +183,12 @@ async fn handle_collation_message>(),
+ "Submitting collation for core.",
+ );
overseer_handle
.send_msg(
diff --git a/cumulus/client/consensus/aura/src/collators/slot_based/mod.rs b/cumulus/client/consensus/aura/src/collators/slot_based/mod.rs
index c9ce5404db03f..38eb099e7532d 100644
--- a/cumulus/client/consensus/aura/src/collators/slot_based/mod.rs
+++ b/cumulus/client/consensus/aura/src/collators/slot_based/mod.rs
@@ -43,7 +43,9 @@
//!
//! - Parachain slot duration
//! - Number of assigned parachain cores
-//! - Parachain runtime configuration
+//! - The `target_block_rate` runtime API, which determines how many blocks to produce per relay
+//! chain slot. When this API is unavailable, the block builder falls back to one block per core.
+//! When the target exceeds the number of cores, multiple blocks are bundled per core.
//!
//! ## Timing Examples
//!
@@ -69,23 +71,26 @@
use self::{block_builder_task::run_block_builder, collation_task::run_collation_task};
pub use block_import::{SlotBasedBlockImport, SlotBasedBlockImportHandle};
use codec::Codec;
-use consensus_common::ParachainCandidate;
use cumulus_client_collator::service::ServiceInterface as CollatorServiceInterface;
use cumulus_client_consensus_common::{self as consensus_common, ParachainBlockImportMarker};
+use cumulus_client_proof_size_recording::register_proof_size_recording_cleanup;
use cumulus_primitives_aura::AuraUnincludedSegmentApi;
-use cumulus_primitives_core::{KeyToIncludeInRelayProof, RelayParentOffsetApi};
+use cumulus_primitives_core::{KeyToIncludeInRelayProof, RelayParentOffsetApi, TargetBlockRate};
use cumulus_relay_chain_interface::RelayChainInterface;
use futures::FutureExt;
use polkadot_primitives::{
CollatorPair, CoreIndex, Hash as RelayHash, Id as ParaId, PersistedValidationData,
ValidationCodeHash,
};
-use sc_client_api::{backend::AuxStore, BlockBackend, BlockOf, UsageProvider};
+use sc_client_api::{
+ backend::AuxStore, client::PreCommitActions, BlockBackend, BlockOf, UsageProvider,
+};
use sc_consensus::BlockImport;
use sc_network_types::PeerId;
use sc_utils::mpsc::tracing_unbounded;
-use sp_api::ProvideRuntimeApi;
+use sp_api::{ProvideRuntimeApi, StorageProof};
use sp_application_crypto::AppPublic;
+use sp_block_builder::BlockBuilder;
use sp_blockchain::HeaderBackend;
use sp_consensus::Environment;
use sp_consensus_aura::AuraApi;
@@ -132,8 +137,6 @@ pub struct Params
+ BlockBackend
+ UsageProvider
+ + PreCommitActions
+ Send
+ Sync
+ 'static,
Client::Api: AuraApi
+ AuraUnincludedSegmentApi
+ RelayParentOffsetApi
+ + TargetBlockRate
+ + BlockBuilder
+ KeyToIncludeInRelayProof,
Backend: sc_client_api::Backend + 'static,
RClient: RelayChainInterface + Clone + 'static,
@@ -177,7 +183,7 @@ pub fn run + ParachainBlockImportMarker + Send + Sync + 'static,
Proposer: Environment + Send + Sync + 'static,
CS: CollatorServiceInterface + Send + Sync + Clone + 'static,
- CHP: consensus_common::ValidationCodeHashProvider + Send + 'static,
+ CHP: consensus_common::ValidationCodeHashProvider + Send + Sync + 'static,
P: Pair + Send + Sync + 'static,
P::Public: AppPublic + Member + Codec,
P::Signature: TryFrom> + Member + Codec,
@@ -196,7 +202,6 @@ pub fn run {
pub relay_parent: RelayHash,
/// The header of the parent block.
pub parent_header: Block::Header,
- /// The parachain block candidate.
- pub parachain_candidate: ParachainCandidate,
+ /// The built blocks.
+ pub blocks: Vec,
+ /// The storage proof that was collected while building all the blocks.
+ pub proof: StorageProof,
/// The validation code hash at the parent block.
pub validation_code_hash: ValidationCodeHash,
/// Core index that this block should be submitted on
diff --git a/cumulus/client/consensus/aura/src/collators/slot_based/relay_chain_data_cache.rs b/cumulus/client/consensus/aura/src/collators/slot_based/relay_chain_data_cache.rs
index fb22251e4f5c4..17e49479a2648 100644
--- a/cumulus/client/consensus/aura/src/collators/slot_based/relay_chain_data_cache.rs
+++ b/cumulus/client/consensus/aura/src/collators/slot_based/relay_chain_data_cache.rs
@@ -18,7 +18,6 @@
//! Utility for caching [`RelayChainData`] for different relay blocks.
use crate::collators::claim_queue_at;
-use cumulus_primitives_core::CoreSelector;
use cumulus_relay_chain_interface::RelayChainInterface;
use polkadot_node_subsystem_util::runtime::ClaimQueueSnapshot;
use polkadot_primitives::{
@@ -35,8 +34,6 @@ pub struct RelayChainData {
pub claim_queue: ClaimQueueSnapshot,
/// Maximum configured PoV size on the relay chain.
pub max_pov_size: u32,
- /// The last [`CoreSelector`] we used.
- pub last_claimed_core_selector: Option,
}
/// Simple helper to fetch relay chain data and cache it based on the current relay chain best block
@@ -107,16 +104,11 @@ where
},
};
- Ok(RelayChainData {
- relay_parent_header,
- claim_queue,
- max_pov_size,
- last_claimed_core_selector: None,
- })
+ Ok(RelayChainData { relay_parent_header, claim_queue, max_pov_size })
}
#[cfg(test)]
- pub(crate) fn insert_test_data(&mut self, relay_parent: RelayHash, data: RelayChainData) {
- self.cached_data.insert(relay_parent, data);
+ pub fn insert_test_data(&mut self, relay_parent_hash: RelayHash, data: RelayChainData) {
+ self.cached_data.insert(relay_parent_hash, data);
}
}
diff --git a/cumulus/client/consensus/aura/src/collators/slot_based/slot_timer.rs b/cumulus/client/consensus/aura/src/collators/slot_based/slot_timer.rs
index a5357dc8d4ef5..810ab247f5e06 100644
--- a/cumulus/client/consensus/aura/src/collators/slot_based/slot_timer.rs
+++ b/cumulus/client/consensus/aura/src/collators/slot_based/slot_timer.rs
@@ -16,44 +16,11 @@
// along with Cumulus. If not, see .
use crate::LOG_TARGET;
-use codec::Codec;
use cumulus_primitives_aura::Slot;
-use cumulus_primitives_core::BlockT;
-use sc_client_api::UsageProvider;
use sc_consensus_aura::SlotDuration;
-use sp_api::{ApiExt, ProvideRuntimeApi};
-use sp_application_crypto::AppPublic;
-use sp_consensus_aura::AuraApi;
-use sp_core::Pair;
-use sp_runtime::traits::Member;
+
use sp_timestamp::Timestamp;
-use std::{
- cmp::{max, min},
- sync::Arc,
- time::Duration,
-};
-
-/// Lower limits of allowed block production interval.
-/// Defensive mechanism, corresponds to 12 cores at 6 second block time.
-const BLOCK_PRODUCTION_MINIMUM_INTERVAL_MS: Duration = Duration::from_millis(500);
-
-/// Theoretically, the block production is capped at `BLOCK_PRODUCTION_MINIMUM_INTERVAL_MS`.
-/// In practice, there might be slight deviations due to timing inaccuracies and delays.
-///
-/// This constant is taken into account while adjusting the authoring duration to fit into the slot.
-/// Therefore, it will only reduce the authoring duration if we are within the
-/// `BLOCK_PRODUCTION_ADJUSTMENT_MS` threshold of the next slot.
-///
-/// ### 12 cores 500ms blocks
-///
-/// For example, for 12 cores 500ms blocks: the next slot is scheduled in 490ms due to delays.
-/// In that case, we still want to attempt producing the block, as missing the slot would be worse
-/// than producing slightly too fast.
-const BLOCK_PRODUCTION_THRESHOLD_MS: Duration = Duration::from_millis(100);
-
-/// The amount of time the authoring duration of the last block production attempt
-/// should be reduced by to fit into the slot timing.
-const BLOCK_PRODUCTION_ADJUSTMENT_MS: Duration = Duration::from_millis(1000);
+use std::time::Duration;
#[derive(Debug)]
pub(crate) struct SlotInfo {
@@ -61,82 +28,70 @@ pub(crate) struct SlotInfo {
pub slot: Slot,
}
-/// Manages block-production timings based on chain parameters and assigned cores.
-#[derive(Debug)]
-pub(crate) struct SlotTimer {
- /// Parachain client that is used for runtime calls
- client: Arc,
- /// Offset the current time by this duration.
- time_offset: Duration,
- /// Last reported core count.
- last_reported_core_num: Option,
- /// Slot duration of the relay chain. This is used to compute how man block-production
- /// attempts we should trigger per relay chain block.
+/// Information about a slot timing, including the relay chain slot duration and exact start
+/// timestamp.
+#[derive(Debug, Clone)]
+pub(crate) struct SlotTime {
+ /// The relay chain slot duration used for this timing
relay_slot_duration: Duration,
- /// Stores the latest slot that was reported by [`Self::wait_until_next_slot`].
- last_reported_slot: Option,
- _marker: std::marker::PhantomData<(Block, Box)>,
+ /// The exact timestamp when this relay chain slot started
+ slot_start_timestamp: Timestamp,
+ /// Time offset to apply when calculating time remaining
+ time_offset: Duration,
}
-/// Compute when to try block-authoring next.
-/// The exact time point is determined by the slot duration of relay- and parachain as
-/// well as the last observed core count. If more cores are available, we attempt to author blocks
-/// for them.
-///
-/// Returns a tuple with:
-/// - `Duration`: How long to wait until the next slot.
-/// - `Slot`: The AURA slot used for authoring
-fn compute_next_wake_up_time(
- para_slot_duration: SlotDuration,
- relay_slot_duration: Duration,
- core_count: Option,
- time_now: Duration,
- time_offset: Duration,
-) -> (Duration, Slot) {
- let para_slots_per_relay_block =
- (relay_slot_duration.as_millis() / para_slot_duration.as_millis() as u128) as u32;
- let assigned_core_num = core_count.unwrap_or(1);
-
- // Trigger at least once per relay block, if we have for example 12 second slot duration,
- // we should still produce two blocks if we are scheduled on every relay block.
- let mut block_production_interval = min(para_slot_duration.as_duration(), relay_slot_duration);
-
- if assigned_core_num > para_slots_per_relay_block &&
- para_slot_duration.as_duration() >= relay_slot_duration
- {
- block_production_interval =
- max(relay_slot_duration / assigned_core_num, BLOCK_PRODUCTION_MINIMUM_INTERVAL_MS);
- tracing::debug!(
- target: LOG_TARGET,
- ?block_production_interval,
- "Expected to produce for {assigned_core_num} cores but only have {para_slots_per_relay_block} slots. Attempting to produce multiple blocks per slot."
- );
+impl SlotTime {
+ /// Create a new SlotTime
+ pub fn new(
+ relay_slot_duration: Duration,
+ slot_start_timestamp: Timestamp,
+ time_offset: Duration,
+ ) -> Self {
+ Self { relay_slot_duration, slot_start_timestamp, time_offset }
}
- let (duration, timestamp) =
- time_until_next_attempt(time_now, block_production_interval, time_offset);
- let aura_slot = Slot::from_timestamp(timestamp, para_slot_duration);
- (duration, aura_slot)
-}
+ /// Get the time remaining in this slot
+ pub fn time_left(&self) -> Duration {
+ self.time_left_internal(duration_now())
+ }
-/// Compute the time until the next slot changes.
-///
-/// Returns None if the next slot cannot be computed.
-fn compute_time_until_next_slot_change(
- para_slot_duration: SlotDuration,
- time_now: Duration,
- time_offset: Duration,
- last_reported_slot: Slot,
-) -> Option<(Duration, Slot)> {
- let now = time_now.saturating_sub(time_offset);
- let next_slot = last_reported_slot + Slot::from(1);
+ /// Internal implementation of [`Self::time_left`] that takes `now` as parameter.
+ fn time_left_internal(&self, now: Duration) -> Duration {
+ let now = now.saturating_sub(self.time_offset);
+ let slot_end_time_millis =
+ self.slot_start_timestamp.as_millis() + self.relay_slot_duration.as_millis() as u64;
+ let slot_end_time = Duration::from_millis(slot_end_time_millis);
+
+ slot_end_time.saturating_sub(now)
+ }
+
+ /// Check if the next relay chain slot would be in a different parachain slot.
+ pub fn is_parachain_slot_ending(&self, parachain_slot_duration: Duration) -> bool {
+ let now = duration_now().saturating_sub(self.time_offset);
+ let next_relay_slot_start_time =
+ self.slot_start_timestamp.as_duration() + self.relay_slot_duration;
- let Some(next_slot_timestamp) = next_slot.timestamp(para_slot_duration) else {
- return None;
- };
- let remaining_time = next_slot_timestamp.as_duration().saturating_sub(now);
+ // Calculate current parachain slot
+ let current_parachain_slot = now.as_millis() / parachain_slot_duration.as_millis();
- Some((remaining_time, next_slot))
+ // Calculate parachain slot for next relay slot
+ let next_parachain_slot =
+ next_relay_slot_start_time.as_millis() / parachain_slot_duration.as_millis() as u128;
+
+ current_parachain_slot != next_parachain_slot
+ }
+}
+
+/// Manages block-production slots based on the relay chain slot duration.
+#[derive(Debug)]
+pub(crate) struct SlotTimer {
+ /// Offset the current time by this duration.
+ time_offset: Duration,
+ /// Slot duration of the relay chain. This is used to compute when to wake up for
+ /// block production attempts.
+ relay_slot_duration: Duration,
+ /// Stores the latest slot that was reported by [`Self::wait_until_next_slot`].
+ last_reported_slot: Option,
}
/// Returns current duration since Unix epoch.
@@ -148,90 +103,13 @@ pub(super) fn duration_now() -> Duration {
})
}
-/// Adjust the authoring duration.
-fn adjust_authoring_duration(
- mut authoring_duration: Duration,
- next_block: (Duration, Slot),
- next_slot_change: (Duration, Slot),
- different_authors: bool,
-) -> Option {
- let (duration, next_block_slot) = next_block;
- let (duration_until_next_slot, next_slot) = next_slot_change;
-
- // The authoring of blocks must stop 1 second before the slot ends.
- let duration_until_deadline =
- duration_until_next_slot.saturating_sub(BLOCK_PRODUCTION_ADJUSTMENT_MS);
- tracing::debug!(
- target: LOG_TARGET,
- ?authoring_duration,
- ?duration,
- ?next_block_slot,
- ?duration_until_next_slot,
- ?next_slot,
- ?duration_until_deadline,
- ?different_authors,
- "Adjusting authoring duration for slot.",
- );
-
- // Ensure no blocks are produced in the last second of the slot,
- // regardless of authoring duration.
- if duration_until_deadline == Duration::ZERO {
- if different_authors {
- tracing::debug!(
- target: LOG_TARGET,
- ?duration_until_next_slot,
- ?next_slot,
- "Not enough time left in the slot to adjust authoring duration. Skipping block production for the slot."
- );
-
- return None;
- }
-
- // If authors are the same, we can still attempt producing the block
- // considering the next block duration.
- return Some(authoring_duration.min(duration));
- }
-
- // Clamp the authoring duration to fit into the slot deadline only if authors are different.
- // For most cases, the deadline is farther in the future than the authoring duration.
- if different_authors && authoring_duration >= duration_until_deadline {
- authoring_duration = duration_until_deadline;
-
- // Ensure we are not going below the minimum interval within a reasonable threshold.
- // For 12 cores, we might have a scenario where the last 3 blocks are skipped:
- // - Block 10: next slot change in 1.493s:
- // - After adjusting the deadline: 1.493s - 1s = 0.493s the block could be produced
- // without issues.
- // - Block 11: next slot change in 0.993s - skipped by the deadline
- // - Block 12: next slot change in 0.493s - skipped by the deadline
- if authoring_duration <
- BLOCK_PRODUCTION_MINIMUM_INTERVAL_MS.saturating_sub(BLOCK_PRODUCTION_THRESHOLD_MS)
- {
- tracing::debug!(
- target: LOG_TARGET,
- ?authoring_duration,
- ?next_slot,
- "Authoring duration is below minimum. Skipping block production for the slot."
- );
- return None;
- }
- }
-
- // The `duration` intends to slightly adjust when then block production
- // attempt happens. This goes slightly below the `BLOCK_PRODUCTION_MINIMUM_INTERVAL_MS`
- // threshold.
- Some(authoring_duration.min(duration))
-}
-
-/// Returns the duration until the next block production should be attempted.
-/// Returns:
-/// - Duration: The duration until the next attempt.
-fn time_until_next_attempt(
+/// Returns the duration until the next block production slot and the timestamp at this slot.
+fn time_until_next_slot(
now: Duration,
block_production_interval: Duration,
offset: Duration,
) -> (Duration, Timestamp) {
- let now = now.as_millis().saturating_sub(offset.as_millis());
+ let now = now.saturating_sub(offset).as_millis();
let next_slot_time = ((now + block_production_interval.as_millis()) /
block_production_interval.as_millis()) *
@@ -240,158 +118,68 @@ fn time_until_next_attempt(
(Duration::from_millis(remaining_millis as u64), Timestamp::from(next_slot_time as u64))
}
-impl SlotTimer
-where
- Block: BlockT,
- Client: ProvideRuntimeApi + UsageProvider + Send + Sync + 'static,
- Client::Api: AuraApi,
- P: Pair,
- P::Public: AppPublic + Member + Codec,
- P::Signature: TryFrom> + Member + Codec,
-{
+impl SlotTimer {
/// Create a new slot timer.
- pub fn new_with_offset(
- client: Arc,
- time_offset: Duration,
- relay_slot_duration: Duration,
- ) -> Self {
- Self {
- client,
- time_offset,
- last_reported_core_num: None,
- relay_slot_duration,
- last_reported_slot: Default::default(),
- _marker: Default::default(),
- }
- }
-
- /// Inform the slot timer about the last seen number of cores.
- pub fn update_scheduling(&mut self, num_cores_next_block: u32) {
- self.last_reported_core_num = Some(num_cores_next_block);
- }
-
- /// Returns the slot and how much time left until the next block production attempt.
- pub fn time_until_next_block(&mut self, slot_duration: SlotDuration) -> (Duration, Slot) {
- compute_next_wake_up_time(
- slot_duration,
- self.relay_slot_duration,
- self.last_reported_core_num,
- duration_now(),
- self.time_offset,
- )
- }
-
- /// Compute the time until the next slot changes.
- fn time_until_next_slot_change(
- &mut self,
- slot_duration: SlotDuration,
- ) -> Option<(Duration, Slot)> {
- compute_time_until_next_slot_change(
- slot_duration,
- duration_now(),
- self.time_offset,
- self.last_reported_slot.unwrap_or_default(),
- )
- }
-
- /// Check if two slots have different authors based on AURA round-robin algorithm.
- ///
- /// Returns true if the authors for the two slots are different.
- fn check_different_slot_authors(&self, slot: Slot, next_slot: Slot) -> bool {
- let best_hash = self.client.usage_info().chain.best_hash;
-
- let mut runtime_api = self.client.runtime_api();
- runtime_api.set_call_context(sp_core::traits::CallContext::Onchain { import: false });
- let Ok(authorities) = runtime_api.authorities(best_hash) else {
- // Presume they are different, this will adjust the slot authoring duration more
- // conservatively.
- return true;
- };
-
- let authorities_len = authorities.len() as u64;
- if authorities_len <= 1 {
- return false;
- }
-
- let author1_idx = *slot % authorities_len;
- let author2_idx = *next_slot % authorities_len;
-
- author1_idx != author2_idx
- }
-
- /// Adjust the authoring duration to fit into the slot timing.
- ///
- /// Returns the adjusted authoring duration and the slot that it corresponds to.
- pub fn adjust_authoring_duration(&mut self, authoring_duration: Duration) -> Option {
- let Ok(slot_duration) = crate::slot_duration(&*self.client) else {
- tracing::error!(target: LOG_TARGET, "Failed to fetch slot duration from runtime.");
- return None;
- };
-
- let next_block = self.time_until_next_block(slot_duration);
- let Some(next_slot_change) = self.time_until_next_slot_change(slot_duration) else {
- tracing::error!(
- target: LOG_TARGET,
- "Failed to compute time until next slot change. Using unadjusted authoring duration."
- );
- return Some(authoring_duration);
- };
-
- // Check if authors at current and next slots are different
- let current_slot = self.last_reported_slot.unwrap_or(next_block.1);
- let different_authors = self.check_different_slot_authors(current_slot, next_slot_change.1);
-
- adjust_authoring_duration(
- authoring_duration,
- next_block,
- next_slot_change,
- different_authors,
- )
+ pub fn new_with_offset(time_offset: Duration, relay_slot_duration: Duration) -> Self {
+ Self { time_offset, relay_slot_duration, last_reported_slot: None }
}
/// Returns a future that resolves when the next block production should be attempted.
- pub async fn wait_until_next_slot(&mut self) -> Result<(), ()> {
- let slot_duration = match crate::slot_duration(&*self.client) {
- Ok(d) => d,
- Err(error) => {
- tracing::error!(target: LOG_TARGET, %error, "Failed to fetch slot duration from runtime.");
- return Err(());
- },
- };
+ pub async fn wait_until_next_slot(&mut self) -> Result {
+ let (time_until_next_attempt, timestamp) =
+ time_until_next_slot(duration_now(), self.relay_slot_duration, self.time_offset);
- let (time_until_next_attempt, mut next_aura_slot) =
- self.time_until_next_block(slot_duration);
+ // Calculate the current slot using the relay chain slot duration
+ let relay_slot_duration_for_slot = SlotDuration::from(self.relay_slot_duration);
+ let mut next_slot = Slot::from_timestamp(timestamp, relay_slot_duration_for_slot);
- tracing::trace!(
- target: LOG_TARGET,
- ?time_until_next_attempt,
- aura_slot = ?next_aura_slot,
- last_reported = ?self.last_reported_slot,
- "Determined next block production opportunity."
- );
+ // Calculate the actual slot start timestamp (may be different if we're catching up)
+ let mut slot_start_timestamp = timestamp;
match self.last_reported_slot {
// If we already reported a slot, we don't want to skip a slot. But we also don't want
// to go through all the slots if a node was halted for some reason.
- Some(ls) if ls + 1 < next_aura_slot && next_aura_slot <= ls + 3 => {
- next_aura_slot = ls + 1u64;
+ Some(ls) if ls + 1 < next_slot && next_slot <= ls + 3 => {
+ next_slot = ls + 1u64;
+ // Calculate the timestamp for the adjusted slot
+ slot_start_timestamp =
+ next_slot.timestamp(relay_slot_duration_for_slot).ok_or(())?;
+ // Don't sleep since we're catching up
+ tracing::debug!(
+ target: LOG_TARGET,
+ last_slot = ?ls,
+ next_slot = ?next_slot,
+ "Catching up on skipped slot."
+ );
},
None | Some(_) => {
- tracing::trace!(target: LOG_TARGET, ?time_until_next_attempt, "Sleeping until the next slot.");
- tokio::time::sleep(time_until_next_attempt).await;
+ tracing::trace!(
+ target: LOG_TARGET,
+ time_to_sleep = ?time_until_next_attempt,
+ "Feeling sleepy 😴"
+ );
+
+ // Wake up slightly before the next slot to avoid noisy "catching up" logs caused by
+ // scheduler jitter right at the slot boundary.
+ tokio::time::sleep(
+ time_until_next_attempt.saturating_sub(Duration::from_millis(2)),
+ )
+ .await;
},
}
tracing::debug!(
target: LOG_TARGET,
- ?slot_duration,
- aura_slot = ?next_aura_slot,
- "New block production opportunity."
+ relay_slot_duration = ?self.relay_slot_duration,
+ ?next_slot,
+ ?slot_start_timestamp,
+ "New block production slot."
);
- self.last_reported_slot = Some(next_aura_slot);
+ // Update internal slot tracking
+ self.last_reported_slot = Some(next_slot);
- Ok(())
+ Ok(SlotTime::new(self.relay_slot_duration, slot_start_timestamp, self.time_offset))
}
}
@@ -399,250 +187,69 @@ where
mod tests {
use super::*;
use rstest::rstest;
- use sc_consensus_aura::SlotDuration;
const RELAY_CHAIN_SLOT_DURATION: u64 = 6000;
#[rstest]
// Test that different now timestamps have correct impact
- // ||||
- #[case(6000, Some(1), 1000, 0, 5000)]
- #[case(6000, Some(1), 0, 0, 6000)]
- #[case(6000, Some(1), 6000, 0, 6000)]
- #[case(6000, Some(0), 6000, 0, 6000)]
- // Test that `None` core defaults to 1
- // ||||
- #[case(6000, None, 1000, 0, 5000)]
- #[case(6000, None, 0, 0, 6000)]
- #[case(6000, None, 6000, 0, 6000)]
+ #[case(1000, 0, 5000)]
+ #[case(0, 0, 6000)]
+ #[case(6000, 0, 6000)]
// Test that offset affects the current time correctly
- // ||||
- #[case(6000, Some(1), 1000, 1000, 6000)]
- #[case(6000, Some(1), 12000, 2000, 2000)]
- #[case(6000, Some(1), 12000, 6000, 6000)]
- #[case(6000, Some(1), 12000, 7000, 1000)]
- // Test that number of cores affects the block production interval
- // |||||||
- #[case(6000, Some(3), 12000, 0, 2000)]
- #[case(6000, Some(2), 12000, 0, 3000)]
- #[case(6000, Some(3), 11999, 0, 1)]
- // High core count
- // ||||||||
- #[case(6000, Some(12), 0, 0, 500)]
- /// Test that the minimum block interval is respected
- /// at high core counts.
- /// |||||||||
- #[case(6000, Some(100), 0, 0, 500)]
- // Test that slot_duration works correctly
- // ||||
- #[case(2000, Some(1), 1000, 0, 1000)]
- #[case(2000, Some(1), 3000, 0, 1000)]
- #[case(2000, Some(1), 10000, 0, 2000)]
- #[case(2000, Some(2), 1000, 0, 1000)]
- // Cores are ignored if relay_slot_duration != para_slot_duration
- // |||||||
- #[case(2000, Some(3), 3000, 0, 1000)]
- // For long slot durations, we should still check
- // every relay chain block for the slot.
- // |||||
- #[case(12000, None, 0, 0, 6000)]
- #[case(12000, None, 6100, 0, 5900)]
- #[case(12000, None, 6000, 2000, 2000)]
- #[case(12000, Some(2), 6000, 0, 3000)]
- #[case(12000, Some(3), 6000, 0, 2000)]
- #[case(12000, Some(3), 8100, 0, 1900)]
+ #[case(1000, 1000, 6000)]
+ #[case(12000, 2000, 2000)]
+ #[case(12000, 6000, 6000)]
+ #[case(12000, 7000, 1000)]
+ // Test basic timing with relay slot duration
+ #[case(11999, 0, 1)]
fn test_get_next_slot(
- #[case] para_slot_millis: u64,
- #[case] core_count: Option,
#[case] time_now: u64,
#[case] offset_millis: u64,
#[case] expected_wait_duration: u128,
) {
- let para_slot_duration = SlotDuration::from_millis(para_slot_millis); // 6 second slots
let relay_slot_duration = Duration::from_millis(RELAY_CHAIN_SLOT_DURATION);
- let time_now = Duration::from_millis(time_now); // 1 second passed
+ let time_now = Duration::from_millis(time_now);
let offset = Duration::from_millis(offset_millis);
- let (wait_duration, _) = compute_next_wake_up_time(
- para_slot_duration,
- relay_slot_duration,
- core_count,
- time_now,
- offset,
- );
+ let (wait_duration, _) = time_until_next_slot(time_now, relay_slot_duration, offset);
assert_eq!(wait_duration.as_millis(), expected_wait_duration, "Wait time mismatch.");
- // Should wait 5 seconds
}
#[rstest]
// Basic slot change scenarios
- #[case(6000, 0, 0, Slot::from(0), 6000, Slot::from(1))]
- #[case(6000, 1000, 0, Slot::from(0), 5000, Slot::from(1))]
- #[case(6000, 6000, 0, Slot::from(1), 6000, Slot::from(2))]
- #[case(6000, 12000, 0, Slot::from(2), 6000, Slot::from(3))]
+ #[case(6000, 0, 0, Slot::from(0), 6000)]
+ #[case(6000, 1000, 0, Slot::from(0), 5000)]
+ #[case(6000, 6000, 0, Slot::from(1), 6000)]
+ #[case(6000, 12000, 0, Slot::from(2), 6000)]
// Test with offset
- #[case(6000, 1000, 1000, Slot::from(0), 6000, Slot::from(1))]
- #[case(6000, 2000, 1000, Slot::from(0), 5000, Slot::from(1))]
- #[case(6000, 6000, 3000, Slot::from(0), 3000, Slot::from(1))]
+ #[case(6000, 1000, 1000, Slot::from(0), 6000)]
+ #[case(6000, 2000, 1000, Slot::from(0), 5000)]
+ #[case(6000, 6000, 3000, Slot::from(0), 3000)]
// Different slot durations
- #[case(3000, 1000, 0, Slot::from(0), 2000, Slot::from(1))]
- #[case(3000, 3000, 0, Slot::from(1), 3000, Slot::from(2))]
- #[case(12000, 6000, 0, Slot::from(0), 6000, Slot::from(1))]
- #[case(12000, 12000, 0, Slot::from(1), 12000, Slot::from(2))]
+ #[case(3000, 1000, 0, Slot::from(0), 2000)]
+ #[case(3000, 3000, 0, Slot::from(1), 3000)]
+ #[case(12000, 6000, 0, Slot::from(0), 6000)]
+ #[case(12000, 12000, 0, Slot::from(1), 12000)]
// Edge cases - at slot boundary
- #[case(6000, 5999, 0, Slot::from(0), 1, Slot::from(1))]
- #[case(6000, 11999, 0, Slot::from(1), 1, Slot::from(2))]
+ #[case(6000, 5999, 0, Slot::from(0), 1)]
+ #[case(6000, 11999, 0, Slot::from(1), 1)]
fn test_compute_time_until_next_slot_change(
#[case] para_slot_millis: u64,
#[case] time_now: u64,
#[case] offset_millis: u64,
#[case] last_reported_slot: Slot,
#[case] expected_duration: u128,
- #[case] expected_next_slot: Slot,
) {
- let para_slot_duration = SlotDuration::from_millis(para_slot_millis);
- let time_now = Duration::from_millis(time_now);
- let offset = Duration::from_millis(offset_millis);
-
- let result = compute_time_until_next_slot_change(
- para_slot_duration,
- time_now,
- offset,
- last_reported_slot,
- );
-
- assert!(result.is_some(), "Expected result to be Some");
- let (duration, next_slot) = result.unwrap();
- assert_eq!(duration.as_millis(), expected_duration, "Duration mismatch");
- assert_eq!(next_slot, expected_next_slot, "Next slot mismatch");
- }
+ let slot_time = SlotTime {
+ relay_slot_duration: Duration::from_millis(para_slot_millis),
+ time_offset: Duration::from_millis(offset_millis),
+ slot_start_timestamp: Timestamp::new(
+ Duration::from_millis(para_slot_millis).as_millis() as u64 * *last_reported_slot,
+ ),
+ };
- #[rstest]
- // Various scenarios for 2s block production adjustment.
- #[case::blocks_2s_fits_next_block(
- Duration::from_millis(2000), // Authoring duration
- (Duration::from_millis(2000), Slot::from(1)), // Next block
- (Duration::from_millis(4000), Slot::from(2)), // Next slot change
- true, // Different authors
- Some(Duration::from_millis(2000)), // Expected
- )]
- #[case::blocks_2s_closer_next_slot(
- Duration::from_millis(2000), // Authoring duration
- (Duration::from_millis(1950), Slot::from(1)), // Next block
- (Duration::from_millis(4000), Slot::from(2)), // Next slot change
- true, // Different authors
- Some(Duration::from_millis(1950)), // Expected
- )]
- #[case::blocks_2s_closer_next_slot_bigger(
- Duration::from_millis(2000), // Authoring duration
- (Duration::from_millis(1500), Slot::from(1)), // Next block
- (Duration::from_millis(4000), Slot::from(2)), // Next slot change
- true, // Different authors
- Some(Duration::from_millis(1500)), // Expected
- )]
- #[case::blocks_2s_reduce_by_1s(
- Duration::from_millis(2000), // Authoring duration
- (Duration::from_millis(2000), Slot::from(1)), // Next block
- (Duration::from_millis(2000), Slot::from(2)), // Next slot change
- true, // Different authors
- Some(Duration::from_millis(1000)), // Expected
- )]
- #[case::blocks_2s_reduce_by_1s_plus_offset(
- Duration::from_millis(2000), // Authoring duration
- (Duration::from_millis(1950), Slot::from(1)), // Next block
- (Duration::from_millis(1950), Slot::from(2)), // Next slot change
- true, // Different authors
- Some(Duration::from_millis(950)), // Expected
- )]
- #[case::blocks_2s_reduce_to_minimum(
- Duration::from_millis(2000), // Authoring duration
- (Duration::from_millis(1400), Slot::from(1)), // Next block
- (Duration::from_millis(1400), Slot::from(2)), // Next slot change
- true, // Different authors
- Some(Duration::from_millis(400)), // Expected
- )]
- #[case::blocks_2s_reduce_below_minimum(
- Duration::from_millis(2000), // Authoring duration
- (Duration::from_millis(1300), Slot::from(1)), // Next block
- (Duration::from_millis(1300), Slot::from(2)), // Next slot change
- true, // Different authors
- None, // Expected to reduce below minimum
- )]
- #[case::blocks_2s_same_author(
- Duration::from_millis(2000), // Authoring duration
- (Duration::from_millis(1400), Slot::from(1)), // Next block
- (Duration::from_millis(1400), Slot::from(2)), // Next slot change
- false, // Different authors
- Some(Duration::from_millis(1400)), // Expected no adjustment for last second.
- )]
- // Various scenarios for 500ms block production adjustment.
- #[case::blocks_500ms_fits_next_block(
- Duration::from_millis(500), // Authoring duration
- (Duration::from_millis(500), Slot::from(1)), // Next block
- (Duration::from_millis(2000), Slot::from(2)), // Next slot change
- true, // Different authors
- Some(Duration::from_millis(500)), // Expected
- )]
- #[case::blocks_500ms_closer_next_slot(
- Duration::from_millis(500), // Authoring duration
- (Duration::from_millis(450), Slot::from(1)), // Next block
- (Duration::from_millis(2000), Slot::from(2)), // Next slot change
- true, // Different authors
- Some(Duration::from_millis(450)), // Expected
- )]
- #[case::blocks_500ms_closer_next_slot_bigger(
- Duration::from_millis(500), // Authoring duration
- (Duration::from_millis(400), Slot::from(1)), // Next block
- (Duration::from_millis(1500), Slot::from(2)), // Next slot change
- true, // Different authors
- Some(Duration::from_millis(400)), // Expected
- )]
- #[case::blocks_500ms_reduce_by_1s(
- Duration::from_millis(500), // Authoring duration
- (Duration::from_millis(500), Slot::from(1)), // Next block
- (Duration::from_millis(1000), Slot::from(2)), // Next slot change
- true, // Different authors
- None, // Expected
- )]
- #[case::blocks_500ms_reduce_by_1s_closer(
- Duration::from_millis(500), // Authoring duration
- (Duration::from_millis(500), Slot::from(1)), // Next block
- (Duration::from_millis(500), Slot::from(2)), // Next slot change
- true, // Different authors
- None, // Expected
- )]
- // If we are producing with 1 collator for 500ms authoring duration,
- // we must produce the last two slots and ignore the 1s adjustment.
- #[case::blocks_500ms_same_author(
- Duration::from_millis(500), // Authoring duration
- (Duration::from_millis(410), Slot::from(1)), // Next block
- (Duration::from_millis(1000), Slot::from(2)), // Next slot change
- false, // Different authors
- Some(Duration::from_millis(410)), // Expected no adjustment for last second.
- )]
- #[case::blocks_500ms_same_author_closer(
- Duration::from_millis(500), // Authoring duration
- (Duration::from_millis(400), Slot::from(1)), // Next block
- (Duration::from_millis(400), Slot::from(2)), // Next slot change
- false, // Different authors
- Some(Duration::from_millis(400)), // Expected no adjustment for last second.
- )]
- fn test_adjust_authoring_duration(
- #[case] authoring_duration: Duration,
- #[case] next_block: (Duration, Slot),
- #[case] next_slot_change: (Duration, Slot),
- #[case] different_authors: bool,
- #[case] expected: Option,
- ) {
- sp_tracing::init_for_tests();
+ let time_left = slot_time.time_left_internal(Duration::from_millis(time_now));
- let result = adjust_authoring_duration(
- authoring_duration,
- next_block,
- next_slot_change,
- different_authors,
- );
- tracing::debug!("Adjusted authoring duration: {:?}", result);
- assert_eq!(result, expected);
+ assert_eq!(time_left.as_millis(), expected_duration, "Duration mismatch");
}
}
diff --git a/cumulus/client/consensus/aura/src/collators/slot_based/tests.rs b/cumulus/client/consensus/aura/src/collators/slot_based/tests.rs
index 51b48905c58e7..e9e2f8bd5b163 100644
--- a/cumulus/client/consensus/aura/src/collators/slot_based/tests.rs
+++ b/cumulus/client/consensus/aura/src/collators/slot_based/tests.rs
@@ -17,13 +17,13 @@
use super::{
block_builder_task::{
- determine_core, offset_relay_parent_find_descendants, wait_for_current_relay_block,
+ determine_cores, offset_relay_parent_find_descendants, wait_for_current_relay_block,
},
relay_chain_data_cache::{RelayChainData, RelayChainDataCache},
};
use async_trait::async_trait;
use codec::Encode;
-use cumulus_primitives_core::{ClaimQueueOffset, CoreInfo, CoreSelector, CumulusDigestItem};
+use cumulus_primitives_core::CoreSelector;
use cumulus_relay_chain_interface::*;
use futures::Stream;
use polkadot_node_subsystem_util::runtime::ClaimQueueSnapshot;
@@ -36,7 +36,7 @@ use sc_consensus_babe::{
AuthorityId, ConsensusLog as BabeConsensusLog, NextEpochDescriptor, BABE_ENGINE_ID,
};
use sp_core::sr25519;
-use sp_runtime::{generic::BlockId, testing::Header as TestHeader, traits::Header};
+use sp_runtime::{generic::BlockId, traits::Header};
use sp_version::RuntimeVersion;
use std::{
collections::{BTreeMap, HashMap, VecDeque},
@@ -156,75 +156,18 @@ async fn determine_core_new_relay_parent() {
digest: Default::default(),
};
- // Create a test para parent header at block 0 (genesis)
- let para_parent = TestHeader::new_from_number(0);
-
// Setup claim queue data for the cache
cache.set_test_data(relay_parent.clone(), vec![CoreIndex(0), CoreIndex(1)]);
- let result = determine_core(&mut cache, &relay_parent, 1.into(), ¶_parent, 0).await;
+ let result = determine_cores(&mut cache, &relay_parent, 1.into(), 0).await;
let core = result.unwrap();
let core = core.unwrap();
- assert_eq!(core.core_selector(), CoreSelector(0));
+ assert_eq!(core.core_info().selector, CoreSelector(0));
assert_eq!(core.core_index(), CoreIndex(0));
assert_eq!(core.total_cores(), 2);
}
-#[tokio::test]
-async fn determine_core_with_core_info() {
- let (headers, best_header) = create_header_chain();
- let best_hash = best_header.hash();
- let client = TestRelayClient::new(headers);
- let mut cache = RelayChainDataCache::new(client, 1.into());
-
- // Create a test relay parent header
- let relay_parent = RelayHeader {
- parent_hash: best_hash,
- number: 101,
- state_root: Default::default(),
- extrinsics_root: Default::default(),
- digest: Default::default(),
- };
-
- // Create a para parent header with core info in digest
- let core_info = CoreInfo {
- selector: CoreSelector(0),
- claim_queue_offset: ClaimQueueOffset(0),
- number_of_cores: 3.into(),
- };
- let mut digest = sp_runtime::generic::Digest::default();
- digest.push(CumulusDigestItem::CoreInfo(core_info).to_digest_item());
- // Add relay parent storage root to make it a non-new relay parent
- digest.push(cumulus_primitives_core::rpsr_digest::relay_parent_storage_root_item(
- *relay_parent.state_root(),
- *relay_parent.number(),
- ));
-
- let para_parent = TestHeader {
- parent_hash: best_hash.into(),
- number: 1,
- state_root: Default::default(),
- extrinsics_root: Default::default(),
- digest,
- };
-
- // Setup claim queue data for the cache
- cache.set_test_data(relay_parent.clone(), vec![CoreIndex(0), CoreIndex(1), CoreIndex(2)]);
-
- let result = determine_core(&mut cache, &relay_parent, 1.into(), ¶_parent, 0).await;
-
- match result {
- Ok(Some(core)) => {
- assert_eq!(core.core_selector(), CoreSelector(1)); // Should be next selector (0 + 1)
- assert_eq!(core.core_index(), CoreIndex(1));
- assert_eq!(core.total_cores(), 3);
- },
- Ok(None) => panic!("Expected Some core, got None"),
- Err(()) => panic!("determine_core returned error"),
- }
-}
-
#[tokio::test]
async fn determine_core_no_cores_available() {
let (headers, _best_hash) = create_header_chain();
@@ -240,221 +183,15 @@ async fn determine_core_no_cores_available() {
digest: Default::default(),
};
- // Create a test para parent header at block 0 (genesis)
- let para_parent = TestHeader::new_from_number(0);
-
// Setup empty claim queue
cache.set_test_data(relay_parent.clone(), vec![]);
- let result = determine_core(&mut cache, &relay_parent, 1.into(), ¶_parent, 0).await;
+ let result = determine_cores(&mut cache, &relay_parent, 1.into(), 0).await;
let core = result.unwrap();
assert!(core.is_none());
}
-#[tokio::test]
-async fn determine_core_selector_overflow() {
- let (headers, best_header) = create_header_chain();
- let best_hash = best_header.hash();
- let client = TestRelayClient::new(headers);
- let mut cache = RelayChainDataCache::new(client, 1.into());
-
- // Create a test relay parent header
- let relay_parent = RelayHeader {
- parent_hash: best_hash,
- number: 101,
- state_root: Default::default(),
- extrinsics_root: Default::default(),
- digest: Default::default(),
- };
-
- let core_info = CoreInfo {
- selector: CoreSelector(1),
- claim_queue_offset: ClaimQueueOffset(0),
- number_of_cores: 2.into(),
- };
- let mut digest = sp_runtime::generic::Digest::default();
- digest.push(CumulusDigestItem::CoreInfo(core_info).to_digest_item());
- // Add relay parent storage root to make it a non-new relay parent
- digest.push(cumulus_primitives_core::rpsr_digest::relay_parent_storage_root_item(
- *relay_parent.state_root(),
- *relay_parent.number(),
- ));
-
- let para_parent = TestHeader {
- parent_hash: best_hash.into(),
- number: 1,
- state_root: Default::default(),
- extrinsics_root: Default::default(),
- digest,
- };
-
- // Setup claim queue with only 2 cores
- cache.set_test_data(relay_parent.clone(), vec![CoreIndex(0), CoreIndex(1)]);
-
- let result = determine_core(&mut cache, &relay_parent, 1.into(), ¶_parent, 0).await;
-
- let core = result.unwrap();
- assert!(core.is_none()); // Should return None when selector overflows
-}
-
-#[tokio::test]
-async fn determine_core_uses_last_claimed_core_selector() {
- let (headers, best_header) = create_header_chain();
- let best_hash = best_header.hash();
- let client = TestRelayClient::new(headers);
- let mut cache = RelayChainDataCache::new(client, 1.into());
-
- // Create a test relay parent header
- let relay_parent = RelayHeader {
- parent_hash: best_hash,
- number: 101,
- state_root: Default::default(),
- extrinsics_root: Default::default(),
- digest: Default::default(),
- };
-
- // Create a para parent header without core info in digest (non-genesis)
- // Need to add relay parent storage root to digest to make it a non-new relay parent
- let mut digest = sp_runtime::generic::Digest::default();
- digest.push(cumulus_primitives_core::rpsr_digest::relay_parent_storage_root_item(
- *relay_parent.state_root(),
- *relay_parent.number(),
- ));
-
- let para_parent = TestHeader {
- parent_hash: best_hash.into(),
- number: 1,
- state_root: Default::default(),
- extrinsics_root: Default::default(),
- digest,
- };
-
- // Setup claim queue data with last_claimed_core_selector set to 1
- cache.set_test_data_with_last_selector(
- relay_parent.clone(),
- vec![CoreIndex(0), CoreIndex(1), CoreIndex(2)],
- Some(CoreSelector(1)),
- );
-
- let result = determine_core(&mut cache, &relay_parent, 1.into(), ¶_parent, 0).await;
-
- match result {
- Ok(Some(core)) => {
- // Should use last_claimed_core_selector (1) + 1 = 2
- assert_eq!(core.core_selector(), CoreSelector(2));
- assert_eq!(core.core_index(), CoreIndex(2));
- assert_eq!(core.total_cores(), 3);
- },
- Ok(None) => panic!("Expected Some core, got None"),
- Err(()) => panic!("determine_core returned error"),
- }
-}
-
-#[tokio::test]
-async fn determine_core_uses_last_claimed_core_selector_wraps_around() {
- let (headers, best_header) = create_header_chain();
- let best_hash = best_header.hash();
- let client = TestRelayClient::new(headers);
- let mut cache = RelayChainDataCache::new(client, 1.into());
-
- // Create a test relay parent header
- let relay_parent = RelayHeader {
- parent_hash: best_hash,
- number: 101,
- state_root: Default::default(),
- extrinsics_root: Default::default(),
- digest: Default::default(),
- };
-
- // Create a para parent header without core info in digest (non-genesis)
- // Need to add relay parent storage root to digest to make it a non-new relay parent
- let mut digest = sp_runtime::generic::Digest::default();
- digest.push(cumulus_primitives_core::rpsr_digest::relay_parent_storage_root_item(
- *relay_parent.state_root(),
- *relay_parent.number(),
- ));
-
- let para_parent = TestHeader {
- parent_hash: best_hash.into(),
- number: 1,
- state_root: Default::default(),
- extrinsics_root: Default::default(),
- digest,
- };
-
- // Setup claim queue data with last_claimed_core_selector set to 2 (last index)
- // Next selector should wrap around to out of bounds and return None
- cache.set_test_data_with_last_selector(
- relay_parent.clone(),
- vec![CoreIndex(0), CoreIndex(1), CoreIndex(2)],
- Some(CoreSelector(2)),
- );
-
- let result = determine_core(&mut cache, &relay_parent, 1.into(), ¶_parent, 0).await;
-
- match result {
- Ok(Some(_)) => panic!("Expected None due to selector overflow"),
- Ok(None) => {
- // This is expected - selector 2 + 1 = 3, but only cores 0,1,2 available
- },
- Err(()) => panic!("determine_core returned error"),
- }
-}
-
-#[tokio::test]
-async fn determine_core_no_last_claimed_core_selector() {
- let (headers, best_header) = create_header_chain();
- let best_hash = best_header.hash();
- let client = TestRelayClient::new(headers);
- let mut cache = RelayChainDataCache::new(client, 1.into());
-
- // Create a test relay parent header
- let relay_parent = RelayHeader {
- parent_hash: best_hash,
- number: 101,
- state_root: Default::default(),
- extrinsics_root: Default::default(),
- digest: Default::default(),
- };
-
- // Create a para parent header without core info in digest (non-genesis)
- // Need to add relay parent storage root to digest to make it a non-new relay parent
- let mut digest = sp_runtime::generic::Digest::default();
- digest.push(cumulus_primitives_core::rpsr_digest::relay_parent_storage_root_item(
- *relay_parent.state_root(),
- *relay_parent.number(),
- ));
-
- let para_parent = TestHeader {
- parent_hash: best_hash.into(),
- number: 1,
- state_root: Default::default(),
- extrinsics_root: Default::default(),
- digest,
- };
-
- // Setup claim queue data with no last_claimed_core_selector (None)
- cache.set_test_data_with_last_selector(
- relay_parent.clone(),
- vec![CoreIndex(0), CoreIndex(1), CoreIndex(2)],
- None,
- );
-
- let result = determine_core(&mut cache, &relay_parent, 1.into(), ¶_parent, 0).await;
-
- match result {
- Ok(Some(core)) => {
- // Should start from selector 0 + 1 = 1 when no last selector
- assert_eq!(core.core_selector(), CoreSelector(1));
- assert_eq!(core.core_index(), CoreIndex(1));
- assert_eq!(core.total_cores(), 3);
- },
- Ok(None) => panic!("Expected Some core, got None"),
- Err(()) => panic!("determine_core returned error"),
- }
-}
-
#[derive(Clone)]
struct TestRelayClient {
headers: HashMap,
@@ -741,14 +478,13 @@ fn create_header_chain() -> (HashMap, RelayHeader) {
// Test extension for RelayChainDataCache
impl RelayChainDataCache {
fn set_test_data(&mut self, relay_parent_header: RelayHeader, cores: Vec) {
- self.set_test_data_with_last_selector(relay_parent_header, cores, None);
+ self.set_test_data_with_last_selector(relay_parent_header, cores);
}
fn set_test_data_with_last_selector(
&mut self,
relay_parent_header: RelayHeader,
cores: Vec,
- last_claimed_core_selector: Option,
) {
let relay_parent_hash = relay_parent_header.hash();
@@ -763,7 +499,6 @@ impl RelayChainDataCache {
relay_parent_header,
claim_queue: claim_queue_snapshot,
max_pov_size: 1024 * 1024,
- last_claimed_core_selector,
};
self.insert_test_data(relay_parent_hash, data);
diff --git a/cumulus/client/consensus/aura/src/equivocation_import_queue.rs b/cumulus/client/consensus/aura/src/equivocation_import_queue.rs
index 4c228f0cf6d2d..9398f697583d3 100644
--- a/cumulus/client/consensus/aura/src/equivocation_import_queue.rs
+++ b/cumulus/client/consensus/aura/src/equivocation_import_queue.rs
@@ -306,7 +306,7 @@ mod test {
use super::*;
use codec::Encode;
use cumulus_test_client::{
- runtime::Block, seal_block, Client, InitBlockBuilder, TestClientBuilder,
+ runtime::Block, seal_block, BuildBlockBuilder, Client, TestClientBuilder,
TestClientBuilderExt,
};
use cumulus_test_relay_sproof_builder::RelayStateSproofBuilder;
@@ -344,7 +344,11 @@ mod test {
..Default::default()
};
- let block_builder = client.init_block_builder(Some(validation_data), sproof);
+ let block_builder = client
+ .init_block_builder_builder()
+ .with_validation_data(validation_data)
+ .with_relay_sproof_builder(sproof)
+ .build();
let block = block_builder.block_builder.build().unwrap();
let mut blocks = Vec::new();
diff --git a/cumulus/client/consensus/aura/src/lib.rs b/cumulus/client/consensus/aura/src/lib.rs
index bbfa6fe089eb4..78a07407db0b4 100644
--- a/cumulus/client/consensus/aura/src/lib.rs
+++ b/cumulus/client/consensus/aura/src/lib.rs
@@ -23,9 +23,8 @@
//! For more information about AuRa, the Substrate crate should be checked.
use codec::Encode;
-use cumulus_primitives_core::PersistedValidationData;
-
-use cumulus_primitives_core::relay_chain::HeadData;
+use cumulus_primitives_core::{relay_chain::HeadData, PersistedValidationData};
+use polkadot_node_primitives::PoV;
use polkadot_primitives::{BlockNumber as RBlockNumber, Hash as RHash};
use sp_runtime::traits::{Block as BlockT, NumberFor};
use std::{fs, fs::File, path::PathBuf};
@@ -33,7 +32,6 @@ use std::{fs, fs::File, path::PathBuf};
mod import_queue;
pub use import_queue::{build_verifier, import_queue, BuildVerifierParams, ImportQueueParams};
-use polkadot_node_primitives::PoV;
pub use sc_consensus_aura::{
slot_duration, standalone::slot_duration_at, AuraVerifier, BuildAuraWorkerParams,
SlotProportion,
diff --git a/cumulus/client/consensus/common/src/parent_search.rs b/cumulus/client/consensus/common/src/parent_search.rs
index fca93f1f87997..2914498eaed9f 100644
--- a/cumulus/client/consensus/common/src/parent_search.rs
+++ b/cumulus/client/consensus/common/src/parent_search.rs
@@ -44,7 +44,8 @@ pub struct ParentSearchParams {
pub ancestry_lookback: usize,
}
-/// Result of the parent search, containing the included block and the best parent to build on.
+/// A potential parent block returned from [`find_parent_for_building`]
+#[derive(PartialEq, Clone)]
pub struct ParentSearchResult {
/// The header of the included block (confirmed on relay chain).
pub included_header: B::Header,
diff --git a/cumulus/client/consensus/common/src/tests.rs b/cumulus/client/consensus/common/src/tests.rs
index d82ccd7654503..03292588b1a9f 100644
--- a/cumulus/client/consensus/common/src/tests.rs
+++ b/cumulus/client/consensus/common/src/tests.rs
@@ -32,7 +32,7 @@ use cumulus_relay_chain_interface::{
};
use cumulus_test_client::{
runtime::{Block, Hash, Header},
- Backend, Client, InitBlockBuilder, TestClientBuilder, TestClientBuilderExt,
+ Backend, BuildBlockBuilder, Client, TestClientBuilder, TestClientBuilderExt,
};
use cumulus_test_relay_sproof_builder::RelayStateSproofBuilder;
use futures::{channel::mpsc, executor::block_on, select, FutureExt, Stream, StreamExt};
@@ -324,19 +324,22 @@ fn sproof_with_parent(parent: HeadData) -> RelayStateSproofBuilder {
x
}
-fn build_block(
+fn build_block(
builder: &B,
sproof: RelayStateSproofBuilder,
at: Option,
timestamp: Option,
relay_parent: Option,
) -> Block {
- let cumulus_test_client::BlockBuilderAndSupportData { block_builder, .. } = match at {
- Some(at) => match timestamp {
- Some(ts) => builder.init_block_builder_with_timestamp(at, None, sproof, ts),
- None => builder.init_block_builder_at(at, None, sproof),
- },
- None => builder.init_block_builder(None, sproof),
+ let cumulus_test_client::BlockBuilderAndSupportData { block_builder, .. } = {
+ let mut bb = builder.init_block_builder_builder().with_relay_sproof_builder(sproof);
+ if let Some(at) = at {
+ bb = bb.at(at);
+ }
+ if let Some(ts) = timestamp {
+ bb = bb.with_timestamp(ts);
+ }
+ bb.build()
};
let mut block = block_builder.build().unwrap().block;
@@ -570,7 +573,12 @@ async fn follow_finalized_does_not_stop_on_unknown_block() {
let unknown_block = {
let sproof = sproof_with_parent_by_hash(&client, block.hash());
- let block_builder = client.init_block_builder_at(block.hash(), None, sproof).block_builder;
+ let block_builder = client
+ .init_block_builder_builder()
+ .at(block.hash())
+ .with_relay_sproof_builder(sproof)
+ .build()
+ .block_builder;
block_builder.build().unwrap().block
};
@@ -625,7 +633,12 @@ async fn follow_new_best_sets_best_after_it_is_imported() {
let unknown_block = {
let sproof = sproof_with_parent_by_hash(&client, block.hash());
- let block_builder = client.init_block_builder_at(block.hash(), None, sproof).block_builder;
+ let block_builder = client
+ .init_block_builder_builder()
+ .at(block.hash())
+ .with_relay_sproof_builder(sproof)
+ .build()
+ .block_builder;
block_builder.build().unwrap().block
};
diff --git a/cumulus/client/proof-size-recording/Cargo.toml b/cumulus/client/proof-size-recording/Cargo.toml
new file mode 100644
index 0000000000000..1b3f206bbbef2
--- /dev/null
+++ b/cumulus/client/proof-size-recording/Cargo.toml
@@ -0,0 +1,19 @@
+[package]
+authors.workspace = true
+name = "cumulus-client-proof-size-recording"
+version = "0.1.0"
+edition.workspace = true
+description = "Storage proof size recording utilities."
+license = "GPL-3.0-or-later WITH Classpath-exception-2.0"
+homepage.workspace = true
+repository.workspace = true
+
+[lints]
+workspace = true
+
+[dependencies]
+codec = { workspace = true, default-features = true }
+sc-client-api = { workspace = true, default-features = true }
+sp-blockchain = { workspace = true, default-features = true }
+sp-runtime = { workspace = true, default-features = true }
+sp-trie = { workspace = true, default-features = true }
diff --git a/cumulus/client/proof-size-recording/src/lib.rs b/cumulus/client/proof-size-recording/src/lib.rs
new file mode 100644
index 0000000000000..a8c09f227ab1b
--- /dev/null
+++ b/cumulus/client/proof-size-recording/src/lib.rs
@@ -0,0 +1,117 @@
+// Copyright (C) Parity Technologies (UK) Ltd.
+// This file is part of Cumulus.
+// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
+
+// Cumulus 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.
+
+// Cumulus 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 .
+
+//! Proof size recording utilities.
+
+use codec::{Decode, Encode};
+use sc_client_api::{
+ backend::AuxStore,
+ client::{AuxDataOperations, FinalityNotification, PreCommitActions},
+};
+use sp_blockchain::{Error as ClientError, Result as ClientResult};
+use sp_runtime::traits::Block as BlockT;
+use sp_trie::proof_size_extension::RecordedProofSizeEstimations;
+use std::sync::Arc;
+
+const PROOF_SIZE_RECORDING_VERSION: &[u8] = b"cumulus_proof_size_recording_version";
+const PROOF_SIZE_RECORDING_CURRENT_VERSION: u32 = 1;
+
+/// The aux storage key used to store the proof size recordings for the given block hash.
+fn proof_size_recording_key(block_hash: H) -> Vec {
+ (b"cumulus_proof_size_recording", block_hash).encode()
+}
+
+fn load_decode(backend: &B, key: &[u8]) -> ClientResult>
+where
+ B: AuxStore,
+ T: Decode,
+{
+ let corrupt = |e: codec::Error| {
+ ClientError::Backend(format!("Proof size recording DB is corrupted. Decode error: {}", e))
+ };
+ match backend.get_aux(key)? {
+ None => Ok(None),
+ Some(t) => T::decode(&mut &t[..]).map(Some).map_err(corrupt),
+ }
+}
+
+/// Prepare aux storage key-value pairs for persisting proof size recordings.
+///
+/// Returns the key-value pairs that need to be written to the aux storage.
+pub fn prepare_proof_size_recording_aux_data(
+ block_hash: H,
+ recordings: Vec,
+) -> impl Iterator
- , Vec)> {
+ let current_version = PROOF_SIZE_RECORDING_CURRENT_VERSION.encode();
+ let key = proof_size_recording_key(block_hash);
+ let recordings = recordings.encode();
+
+ [(key, recordings), (PROOF_SIZE_RECORDING_VERSION.to_vec(), current_version)].into_iter()
+}
+
+/// Load the proof size recordings associated with a block.
+pub fn load_proof_size_recording(
+ backend: &B,
+ block_hash: H,
+) -> ClientResult> {
+ let version = load_decode::<_, u32>(backend, PROOF_SIZE_RECORDING_VERSION)?;
+
+ match version {
+ None => Ok(None),
+ Some(PROOF_SIZE_RECORDING_CURRENT_VERSION) => {
+ load_decode::<_, Vec>(backend, proof_size_recording_key(block_hash).as_slice())
+ .map(|recordings| recordings.map(Into::into))
+ },
+ Some(other) => Err(ClientError::Backend(format!(
+ "Unsupported proof size recording DB version: {:?}",
+ other
+ ))),
+ }
+}
+
+/// Cleanup auxiliary storage for finalized blocks.
+///
+/// This function removes proof size recordings for blocks that are no longer needed
+/// after finalization. It processes the finalized blocks and their stale heads to
+/// determine which recordings can be safely removed.
+fn aux_storage_cleanup(notification: &FinalityNotification) -> AuxDataOperations
+where
+ Block: BlockT,
+{
+ // Convert the hashes to deletion operations
+ notification
+ .stale_blocks
+ .iter()
+ .map(|b| (proof_size_recording_key(b.hash), None))
+ .collect()
+}
+
+/// Register a finality action for cleaning up proof size recordings.
+///
+/// This should be called during consensus initialization to automatically clean up
+/// proof size recordings when blocks are finalized.
+pub fn register_proof_size_recording_cleanup(client: Arc)
+where
+ C: PreCommitActions + 'static,
+ Block: BlockT,
+{
+ let on_finality = move |notification: &FinalityNotification| -> AuxDataOperations {
+ aux_storage_cleanup(notification)
+ };
+
+ client.register_finality_action(Box::new(on_finality));
+}
diff --git a/cumulus/client/service/Cargo.toml b/cumulus/client/service/Cargo.toml
index a300807d81498..11f965cc93cac 100644
--- a/cumulus/client/service/Cargo.toml
+++ b/cumulus/client/service/Cargo.toml
@@ -50,6 +50,7 @@ cumulus-client-collator = { workspace = true, default-features = true }
cumulus-client-consensus-common = { workspace = true, default-features = true }
cumulus-client-network = { workspace = true, default-features = true }
cumulus-client-pov-recovery = { workspace = true, default-features = true }
+cumulus-client-proof-size-recording = { workspace = true, default-features = true }
cumulus-primitives-core = { workspace = true, default-features = true }
cumulus-primitives-proof-size-hostfunction = { workspace = true, default-features = true }
cumulus-relay-chain-inprocess-interface = { workspace = true, default-features = true }
diff --git a/cumulus/client/service/src/lib.rs b/cumulus/client/service/src/lib.rs
index 4c8d3e2a827e8..3f7efee15ea1b 100644
--- a/cumulus/client/service/src/lib.rs
+++ b/cumulus/client/service/src/lib.rs
@@ -22,6 +22,7 @@
use cumulus_client_cli::CollatorOptions;
use cumulus_client_network::{AssumeSybilResistance, RequireSecondedInBlockAnnounce};
use cumulus_client_pov_recovery::{PoVRecovery, RecoveryDelayRange, RecoveryHandle};
+use cumulus_client_proof_size_recording::load_proof_size_recording;
use cumulus_primitives_core::{CollectCollationInfo, ParaId};
pub use cumulus_primitives_proof_size_hostfunction::storage_proof_size;
use cumulus_relay_chain_inprocess_interface::build_inprocess_relay_chain;
@@ -31,7 +32,8 @@ use futures::{channel::mpsc, StreamExt};
use polkadot_primitives::{CandidateEvent, CollatorPair, OccupiedCoreAssumption};
use prometheus::{Histogram, HistogramOpts, Registry};
use sc_client_api::{
- Backend as BackendT, BlockBackend, BlockchainEvents, Finalizer, ProofProvider, UsageProvider,
+ AuxStore, Backend as BackendT, BlockBackend, BlockchainEvents, Finalizer, ProofProvider,
+ UsageProvider,
};
use sc_consensus::{
import_queue::{ImportQueue, ImportQueueService},
@@ -56,7 +58,7 @@ use sp_runtime::{
traits::{Block as BlockT, BlockIdTo, Header},
SaturatedConversion, Saturating,
};
-use sp_trie::proof_size_extension::ProofSizeExt;
+use sp_trie::proof_size_extension::{ProofSizeExt, ReplayProofSizeProvider};
use std::{
sync::Arc,
time::{Duration, Instant},
@@ -622,13 +624,19 @@ impl ParachainTracingExecuteBlock {
impl