> + Member + Codec,
+{
+ let relay_slot: Slot = get_relay_slot(internal_scheduling_parent_header)?;
+
+ // Authorities and slot duration are read at `authority_anchor` (the walked segment's tip),
+ // which is on the same fork as the blocks we are resubmitting.
+ let para_slot_duration = crate::slot_duration_at(&**para_client, authority_anchor).ok()?;
+ let para_slot_duration_ms = para_slot_duration.as_millis();
+ let para_slot = relay_slot_to_para_slot(relay_slot, para_slot_duration_ms)?;
+
+ let mut runtime_api = para_client.runtime_api();
+ runtime_api.set_call_context(sp_core::traits::CallContext::Onchain { import: false });
+ let authorities = runtime_api.authorities(authority_anchor).ok()?;
+
+ // claim_slot returns the author public key only if we control its key in the keystore.
+ let author_pub =
+ aura_internal::claim_slot::(Slot::from(para_slot), &authorities, keystore).await?;
+
+ let signature = keystore
+ .sign_with(
+ <
::Public as AppCrypto>::ID,
+ <
::Public as AppCrypto>::CRYPTO_ID,
+ author_pub.as_slice(),
+ &payload.encode(),
+ )
+ .ok()
+ .flatten()?;
+ let signature: [u8; 64] = signature.try_into().ok()?;
+
+ Some(SignedSchedulingInfo { payload, signature })
+}
+
+/// Map a relay-chain slot to the parachain slot used for eligible-author selection. Mirrors the
+/// formula used by the on-chain verifier in
+/// . Returns `None` for a zero slot
+/// duration so the caller skips signing rather than dividing by zero.
+fn relay_slot_to_para_slot(relay_slot: Slot, para_slot_duration_ms: u64) -> Option {
+ if para_slot_duration_ms == 0 {
+ return None;
+ }
+ u64::from(relay_slot)
+ .saturating_mul(RELAY_CHAIN_SLOT_DURATION_MILLIS)
+ .checked_div(para_slot_duration_ms)
+}
+
+/// Convert a libp2p [`PeerId`] into the on-chain [`ApprovedPeerId`] (a 64-byte-bounded vector).
+fn peer_id_to_approved(peer_id: PeerId) -> ApprovedPeerId {
+ ApprovedPeerId::truncate_from(peer_id.to_bytes())
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn relay_slot_to_para_slot_matches_verifier_formula() {
+ // 6s relay slot, 2s para slots: three para slots per relay slot.
+ assert_eq!(relay_slot_to_para_slot(Slot::from(10), 2_000), Some(30));
+ // Equal durations: 1:1 mapping.
+ assert_eq!(relay_slot_to_para_slot(Slot::from(7), 6_000), Some(7));
+ // 6s para slot duration with a 12s "double" duration: floor division.
+ assert_eq!(relay_slot_to_para_slot(Slot::from(5), 12_000), Some(2));
+ // Zero duration is guarded rather than panicking.
+ assert_eq!(relay_slot_to_para_slot(Slot::from(1), 0), None);
+ }
+
+ #[test]
+ fn peer_id_round_trips_through_approved_peer_id() {
+ let peer_id = PeerId::random();
+ // PeerId encodings are ~38 bytes, well under the 64-byte bound, so no truncation.
+ assert_eq!(peer_id_to_approved(peer_id).into_inner(), peer_id.to_bytes());
+ }
+}
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 cb1e3cf9985dc..9639d9c0fc683 100644
--- a/cumulus/client/consensus/aura/src/collators/slot_based/tests.rs
+++ b/cumulus/client/consensus/aura/src/collators/slot_based/tests.rs
@@ -506,6 +506,16 @@ impl RelayChainInterface for TestRelayClient {
unimplemented!("Not needed for test")
}
+ async fn ancestor_relay_parent_info(
+ &self,
+ _at: RelayHash,
+ _session_index: SessionIndex,
+ _relay_parent: RelayHash,
+ ) -> RelayChainResult>>
+ {
+ unimplemented!("Not needed for test")
+ }
+
async fn node_features(&self, _at: RelayHash) -> RelayChainResult {
Ok(NodeFeatures::default())
}
@@ -621,6 +631,8 @@ impl RelayChainDataCache {
claim_queue: claim_queue_snapshot,
max_pov_size: 1024 * 1024,
node_features,
+ max_relay_parent_session_age: 0,
+ session_index: 0,
};
self.insert_test_data(relay_parent_hash, data);
diff --git a/cumulus/client/consensus/aura/src/collators/slot_based/unincluded_segment.rs b/cumulus/client/consensus/aura/src/collators/slot_based/unincluded_segment.rs
new file mode 100644
index 0000000000000..f716b4ffc598a
--- /dev/null
+++ b/cumulus/client/consensus/aura/src/collators/slot_based/unincluded_segment.rs
@@ -0,0 +1,307 @@
+// 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 .
+
+//! Reconstruct the parachain's unincluded segment from local node state.
+//!
+//! Walks the canonical fork (per the parachain backend's fork choice) from the relay-chain
+//! reported included block forward to the parachain's best block. The result is the
+//! ordered list of parablock headers that have been authored locally but not yet included
+//! on the relay chain.
+//!
+//! For V4 resubmission, each block's relay parent is checked against the ancestry of the
+//! `scheduling_parent` we are building on; blocks whose relay parent is not an ancestor
+//! are dropped, producing a potentially gapped list. The ancestry walk respects the
+//! relay-chain `max_relay_parent_session_age` runtime limit, so older sessions are not
+//! considered valid relay parents.
+
+use super::CollatorMessage;
+use crate::collators::slot_based::relay_chain_data_cache::RelayChainDataCache;
+use cumulus_primitives_core::{
+ relay_chain::Hash as RelayHash, ParaId, PersistedValidationData, SchedulingProof,
+};
+use cumulus_relay_chain_interface::RelayChainInterface;
+use polkadot_primitives::{
+ Block as RelayBlock, CoreIndex, Header as RelayHeader, ValidationCodeHash,
+};
+use sc_client_api::Backend;
+use sp_api::StorageProof;
+use sp_blockchain::HeaderBackend;
+use sp_runtime::traits::{Block as BlockT, Header as HeaderT};
+use std::time::Instant;
+
+const LOG_TARGET: &str = "consensus::slot_based::unincluded_segment";
+
+/// Reconstruct the unincluded segment as the canonical chain from the relay-chain reported
+/// included block (exclusive) up to the parachain's best block (inclusive).
+///
+/// Parablocks whose relay parent is not within the relay ancestry of `scheduling_parent`
+/// (bounded by the relay chain's `max_relay_parent_session_age`) are dropped, so the returned
+/// chain may contain gaps in block numbers.
+///
+/// Returns headers ordered oldest first (just after the included block) to newest (best).
+pub async fn reconstruct_unincluded_segment(
+ scheduling_parent: RelayHeader,
+ para_id: ParaId,
+ para_backend: &B,
+ relay_client: &RClient,
+ relay_chain_data_cache: &mut RelayChainDataCache,
+) -> Vec
+where
+ Block: BlockT,
+ B: Backend,
+ RClient: RelayChainInterface + Clone + 'static,
+{
+ let scheduling_parent_hash = scheduling_parent.hash();
+ let Some((included_header, included_hash)) =
+ cumulus_client_consensus_common::fetch_included_from_relay_chain::(
+ relay_client,
+ para_backend,
+ para_id,
+ scheduling_parent_hash,
+ )
+ .await
+ .ok()
+ .flatten()
+ else {
+ return Vec::new();
+ };
+ let included_rp = rp_from_digest::(&included_header);
+
+ let best_hash = para_backend.blockchain().info().best_hash;
+ let Some(segment_newest_first) = walk_back(para_backend, best_hash, included_hash) else {
+ return Vec::new();
+ };
+
+ let rp_ancestry =
+ build_scheduling_parent_ancestry(scheduling_parent, included_rp, relay_chain_data_cache)
+ .await;
+
+ let mut filtered: Vec = segment_newest_first
+ .into_iter()
+ .filter(|h| rp_in_ancestry::(h, &rp_ancestry))
+ .collect();
+ filtered.reverse();
+ filtered
+}
+
+/// Walk parent pointers from `tip_hash` until `stop_hash` (the included block), collecting
+/// headers along the way. Returns headers newest-first (excluding `stop_hash`).
+///
+/// Returns `None` if any header on the path is missing (which would mean the local backend is
+/// inconsistent with the relay-chain reported included block).
+fn walk_back(
+ para_backend: &B,
+ tip_hash: Block::Hash,
+ stop_hash: Block::Hash,
+) -> Option>
+where
+ Block: BlockT,
+ B: Backend,
+{
+ let mut acc = Vec::new();
+ let mut cursor = tip_hash;
+ while cursor != stop_hash {
+ let header = match para_backend.blockchain().header(cursor) {
+ Ok(Some(h)) => h,
+ _ => {
+ tracing::warn!(
+ target: LOG_TARGET,
+ hash = ?cursor,
+ "Missing parachain header while walking back to included block.",
+ );
+ return None;
+ },
+ };
+ cursor = *header.parent_hash();
+ acc.push(header);
+ }
+ Some(acc)
+}
+
+/// Collect the relay-chain ancestry of `scheduling_parent` that is admissible as a parablock
+/// relay parent. The walk stops at the earliest of:
+/// * the relay parent of the included head (inclusive) — anything older cannot be the relay
+/// parent of an unincluded block in practice;
+/// * `max_relay_parent_session_age` sessions back, matching the relay-chain runtime limit;
+/// * the relay-chain genesis.
+///
+/// Session boundaries are detected via BABE epoch-change digests on the walked relay headers,
+/// matching the rule applied in `offset_relay_parent_find_descendants`. Returns the admissible
+/// ancestor hashes newest-first, including `scheduling_parent` itself.
+async fn build_scheduling_parent_ancestry(
+ scheduling_parent: RelayHeader,
+ included_rp: Option,
+ relay_chain_data_cache: &mut RelayChainDataCache,
+) -> Vec
+where
+ RClient: RelayChainInterface + Clone + 'static,
+{
+ let max_relay_parent_session_age = relay_chain_data_cache
+ .get_by_hash(scheduling_parent.hash())
+ .await
+ .map(|d| d.max_relay_parent_session_age)
+ .unwrap_or(0);
+ let mut ancestry = Vec::new();
+ let mut current = scheduling_parent;
+ let mut session_age: u32 = 0;
+ loop {
+ // Session boundary: the current header carries an epoch-change digest, so the *parent*
+ // belongs to an older session. We've already accepted `current` itself before bumping.
+ if sc_consensus_babe::contains_epoch_change::(¤t) {
+ session_age = session_age.saturating_add(1);
+ }
+ if session_age > max_relay_parent_session_age {
+ break;
+ }
+
+ let current_hash = current.hash();
+ ancestry.push(current_hash);
+
+ // Stop once we've recorded the relay parent of the included head: nothing older can be
+ // the relay parent of an unincluded block.
+ if Some(current_hash) == included_rp {
+ break;
+ }
+
+ if current.number == 0 {
+ break;
+ }
+
+ let parent_hash = *current.parent_hash();
+ match relay_chain_data_cache.get_by_hash(parent_hash).await {
+ Ok(data) => current = data.relay_header.clone(),
+ Err(()) => break,
+ }
+ }
+ ancestry
+}
+
+/// Extract the relay parent hash that a parablock built against, from its digest.
+fn rp_from_digest(header: &Block::Header) -> Option {
+ cumulus_primitives_core::extract_relay_parent(header.digest())
+}
+
+/// Whether the parablock's relay parent is within the precomputed relay ancestry.
+fn rp_in_ancestry(header: &Block::Header, ancestry: &[RelayHash]) -> bool {
+ match rp_from_digest::(header) {
+ Some(rp) => ancestry.iter().any(|h| *h == rp),
+ None => false,
+ }
+}
+
+// ---------------------------------------------------------------------------
+// Segment shapes used by the collation task to assemble multi-entry submissions.
+//
+// The on-wire channel payload from the builder is `super::CollatorMessage` — always one bundle.
+// Everything multi-entry lives below and never crosses the channel.
+// ---------------------------------------------------------------------------
+
+/// One built bundle's payload, stripped of core/proof metadata.
+pub struct SegmentEntry {
+ pub relay_parent: RelayHash,
+ pub parent_header: Block::Header,
+ pub blocks: Vec,
+ pub proof: StorageProof,
+ pub validation_code_hash: ValidationCodeHash,
+ pub validation_data: PersistedValidationData,
+}
+
+impl From> for SegmentEntry {
+ fn from(msg: CollatorMessage) -> Self {
+ Self {
+ relay_parent: msg.relay_parent,
+ parent_header: msg.parent_header,
+ blocks: msg.blocks,
+ proof: msg.proof,
+ validation_code_hash: msg.validation_code_hash,
+ validation_data: msg.validation_data,
+ }
+ }
+}
+
+/// Walked + loaded prior unincluded segment, *unsigned*. Lives between `HydrateSegment` and
+/// `FinalizeSegment`. Non-empty by construction.
+pub struct HydratedSegment {
+ pub entries: Vec>,
+ /// Newest unincluded block; anchor for the authority/slot-duration lookup at signing time.
+ pub authority_anchor: Block::Hash,
+}
+
+/// Signed prior segment held by the collation task until either the matching first-core
+/// `CollatorMessage` arrives (merge then submit) or `deadline` elapses (submit alone).
+pub struct PriorSegment {
+ pub scheduling_proof: SchedulingProof,
+ pub core_index: CoreIndex,
+ pub deadline: Instant,
+ pub entries: Vec>,
+}
+
+impl PriorSegment {
+ /// First-core blocks arrived: append, then submit under the prior's signed proof.
+ pub fn merge_first_core(mut self, msg: CollatorMessage) -> Segment {
+ debug_assert_eq!(self.core_index, msg.core_index);
+ self.entries.push(msg.into());
+ Segment {
+ scheduling_proof: Some(self.scheduling_proof),
+ core_index: self.core_index,
+ entries: self.entries,
+ }
+ }
+
+ /// Deadline path: the first core built nothing this slot.
+ pub fn into_segment(self) -> Segment {
+ Segment {
+ scheduling_proof: Some(self.scheduling_proof),
+ core_index: self.core_index,
+ entries: self.entries,
+ }
+ }
+}
+
+/// Submit-ready segment for one core.
+pub struct Segment {
+ pub scheduling_proof: Option,
+ pub core_index: CoreIndex,
+ pub entries: Vec>,
+}
+
+impl Segment {
+ pub fn singleton(msg: CollatorMessage) -> Self {
+ let CollatorMessage {
+ scheduling_proof,
+ core_index,
+ relay_parent,
+ parent_header,
+ blocks,
+ proof,
+ validation_code_hash,
+ validation_data,
+ } = msg;
+ Self {
+ scheduling_proof,
+ core_index,
+ entries: vec![SegmentEntry {
+ relay_parent,
+ parent_header,
+ blocks,
+ proof,
+ validation_code_hash,
+ validation_data,
+ }],
+ }
+ }
+}
diff --git a/cumulus/client/consensus/common/Cargo.toml b/cumulus/client/consensus/common/Cargo.toml
index c65e05c02266a..558dcf1699626 100644
--- a/cumulus/client/consensus/common/Cargo.toml
+++ b/cumulus/client/consensus/common/Cargo.toml
@@ -54,3 +54,4 @@ sp-tracing = { workspace = true, default-features = true }
# Cumulus
cumulus-test-client = { workspace = true }
cumulus-test-relay-sproof-builder = { workspace = true, default-features = true }
+substrate-test-runtime = { workspace = true }
diff --git a/cumulus/client/consensus/common/src/finality.rs b/cumulus/client/consensus/common/src/finality.rs
new file mode 100644
index 0000000000000..92c16de2a35b3
--- /dev/null
+++ b/cumulus/client/consensus/common/src/finality.rs
@@ -0,0 +1,162 @@
+// 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 .
+
+//! Shared helpers for finality-driven aux-storage cleanup.
+
+use sc_client_api::HeaderBackend;
+use sp_runtime::traits::{Block as BlockT, Header as _};
+
+const LOG_TARGET: &str = "consensus::common::finality";
+
+/// Compute the previously-finalized block hash from a tree route and a fallback parent hash.
+///
+/// This is the parent of the first block in the tree route, or the supplied `fallback_parent`
+/// (typically the parent hash of the just-finalized header) when the tree route is empty or its
+/// first block's header can't be loaded.
+///
+/// Taking the inputs directly rather than a `FinalityNotification` keeps this function unit-
+/// testable from outside `sc-client-api` (which has a private `unpin_handle` field).
+pub fn old_finalized_hash(
+ client: &C,
+ tree_route: &[Block::Hash],
+ fallback_parent: Block::Hash,
+) -> Block::Hash
+where
+ C: HeaderBackend,
+ Block: BlockT,
+{
+ let Some(first) = tree_route.first() else {
+ return fallback_parent;
+ };
+
+ match client.header(*first) {
+ Ok(Some(header)) => *header.parent_hash(),
+ Ok(None) => {
+ tracing::warn!(
+ target: LOG_TARGET,
+ ?first,
+ "tree_route head header missing; falling back to notification parent hash",
+ );
+ fallback_parent
+ },
+ Err(error) => {
+ tracing::warn!(
+ target: LOG_TARGET,
+ ?first,
+ ?error,
+ "tree_route head header lookup failed; falling back to notification parent hash",
+ );
+ fallback_parent
+ },
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use sp_blockchain::Result as ClientResult;
+
+ type Block = substrate_test_runtime::Block;
+ type Hash = ::Hash;
+
+ // Minimal `HeaderBackend` mock for the `old_finalized_hash_*` tests.
+ //
+ // `lookup` is `None` ⇒ `header()` returns `Ok(None)` (simulates a missing header).
+ struct MockHeaderBackend {
+ lookup: Option<(Hash, substrate_test_runtime::Header)>,
+ }
+
+ impl HeaderBackend for MockHeaderBackend {
+ fn header(
+ &self,
+ hash: ::Hash,
+ ) -> ClientResult::Header>> {
+ Ok(self.lookup.as_ref().and_then(|(h, hdr)| (*h == hash).then(|| hdr.clone())))
+ }
+
+ fn info(&self) -> sc_client_api::blockchain::Info {
+ unimplemented!()
+ }
+
+ fn status(
+ &self,
+ _hash: ::Hash,
+ ) -> ClientResult {
+ unimplemented!()
+ }
+
+ fn number(
+ &self,
+ _hash: ::Hash,
+ ) -> ClientResult>> {
+ unimplemented!()
+ }
+
+ fn hash(
+ &self,
+ _number: sp_runtime::traits::NumberFor,
+ ) -> ClientResult::Hash>> {
+ unimplemented!()
+ }
+ }
+
+ #[test]
+ fn old_finalized_hash_with_empty_tree_route() {
+ let client = MockHeaderBackend { lookup: None };
+ let fallback = Hash::repeat_byte(0x01);
+
+ let old_hash = old_finalized_hash::<_, Block>(&client, &[], fallback);
+ assert_eq!(
+ old_hash, fallback,
+ "empty tree_route should fall through to the supplied parent"
+ );
+ }
+
+ #[test]
+ fn old_finalized_hash_with_tree_route() {
+ use substrate_test_runtime::Header as TestHeader;
+
+ let expected_old = Hash::repeat_byte(0x01);
+ let tree_block = Hash::repeat_byte(0x02);
+
+ let header = TestHeader {
+ parent_hash: expected_old,
+ number: 2,
+ state_root: Default::default(),
+ extrinsics_root: Default::default(),
+ digest: Default::default(),
+ };
+
+ let client = MockHeaderBackend { lookup: Some((tree_block, header)) };
+ let fallback = Hash::repeat_byte(0xFF);
+
+ let old_hash = old_finalized_hash::<_, Block>(&client, &[tree_block], fallback);
+ assert_eq!(old_hash, expected_old, "should resolve to parent of first tree_route block");
+ }
+
+ #[test]
+ fn old_finalized_hash_falls_back_when_header_missing() {
+ // Non-empty tree_route, but the header lookup returns None — the function should fall back
+ // to `fallback_parent` rather than panicking or returning a wrong hash.
+ let client = MockHeaderBackend { lookup: None };
+ let tree_block = Hash::repeat_byte(0x02);
+ let fallback = Hash::repeat_byte(0xAA);
+
+ let old_hash = old_finalized_hash::<_, Block>(&client, &[tree_block], fallback);
+ assert_eq!(old_hash, fallback, "missing header should fall back to the supplied parent");
+ }
+}
diff --git a/cumulus/client/consensus/common/src/lib.rs b/cumulus/client/consensus/common/src/lib.rs
index d7d135f98a842..c4aa295a714f4 100644
--- a/cumulus/client/consensus/common/src/lib.rs
+++ b/cumulus/client/consensus/common/src/lib.rs
@@ -30,12 +30,14 @@ use sp_timestamp::Timestamp;
use std::{sync::Arc, time::Duration};
+mod finality;
mod level_monitor;
mod parachain_consensus;
mod parent_search;
#[cfg(test)]
mod tests;
+pub use finality::old_finalized_hash;
pub use parent_search::*;
pub use cumulus_relay_chain_streams::finalized_heads;
diff --git a/cumulus/client/consensus/common/src/parent_search.rs b/cumulus/client/consensus/common/src/parent_search.rs
index 3d7729f3b98a0..0676a03736923 100644
--- a/cumulus/client/consensus/common/src/parent_search.rs
+++ b/cumulus/client/consensus/common/src/parent_search.rs
@@ -79,30 +79,59 @@ pub async fn find_parent_for_building(
) -> Result>, RelayChainError> {
tracing::trace!(target: LOG_TARGET, "Parent search parameters: {params:?}");
- // Get the included block.
- let Some((included_header, included_hash)) =
- fetch_included_from_relay_chain(relay_client, backend, params.para_id, params.relay_parent)
+ let Some(ParentSearchStart { included_header, start_hash, start_header }) =
+ pick_parent_search_start(relay_client, backend, params.para_id, params.relay_parent)
.await?
else {
return Ok(None);
};
- // Fetch the pending block if one exists.
+ // Build up the ancestry record of the relay chain to compare against.
+ let rp_ancestry =
+ build_relay_parent_ancestry(params.ancestry_lookback, params.relay_parent, relay_client)
+ .await?;
+
+ // Search for the deepest valid parent starting from the pending/included block.
+ let best_parent_header =
+ find_deepest_valid_parent(start_hash, start_header, backend, &rp_ancestry);
+
+ Ok(Some(ParentSearchResult { included_header, best_parent_header }))
+}
+
+/// Anchor for a parent search: the truly-included head plus the chosen `start` block (pending
+/// availability if reported and locally known and a descendant of included, otherwise included).
+pub struct ParentSearchStart {
+ pub included_header: B::Header,
+ pub start_hash: B::Hash,
+ pub start_header: B::Header,
+}
+
+/// Pick the start block for a parent search at `relay_parent`. Prefers pending availability over
+/// the truly-included head. Returns `Ok(None)` when the included head is unknown locally, when a
+/// reported pending head is unusable (not locally known, undecodable, or not a descendant of
+/// included), or when any required relay-chain call fails to produce a head.
+pub async fn pick_parent_search_start(
+ relay_client: &impl RelayChainInterface,
+ backend: &impl Backend,
+ para_id: ParaId,
+ relay_parent: RelayHash,
+) -> Result>, RelayChainError> {
+ let Some((included_header, included_hash)) =
+ fetch_included_from_relay_chain(relay_client, backend, para_id, relay_parent).await?
+ else {
+ return Ok(None);
+ };
+
+ // Fetch the pending block if one exists. `OccupiedCoreAssumption::Included` enacts the
+ // candidate pending availability before returning, so the head we get is what the next
+ // block should be built on top of.
let maybe_pending = {
- // Fetch the most recent pending header from the relay chain. We use
- // `OccupiedCoreAssumption::Included` so the candidate pending availability gets enacted
- // before being returned to us.
let pending_header = relay_client
- .persisted_validation_data(
- params.relay_parent,
- params.para_id,
- OccupiedCoreAssumption::Included,
- )
+ .persisted_validation_data(relay_parent, para_id, OccupiedCoreAssumption::Included)
.await?
.and_then(|p| B::Header::decode(&mut &p.parent_head.0[..]).ok())
.filter(|x| x.hash() != included_hash);
- // If the pending block is not locally known, we can't proceed.
if let Some(header) = pending_header {
let pending_hash = header.hash();
let Ok(Some(header)) = backend.blockchain().header(pending_hash) else {
@@ -119,10 +148,8 @@ pub async fn find_parent_for_building(
}
};
- // Determine the starting point for the search.
let (start_hash, start_header) = match &maybe_pending {
Some((pending_header, pending_hash)) => {
- // Verify pending is a descendant of included.
let route =
sp_blockchain::tree_route(backend.blockchain(), included_hash, *pending_hash)?;
if !route.retracted().is_empty() {
@@ -137,20 +164,11 @@ pub async fn find_parent_for_building(
None => (included_hash, included_header.clone()),
};
- // Build up the ancestry record of the relay chain to compare against.
- let rp_ancestry =
- build_relay_parent_ancestry(params.ancestry_lookback, params.relay_parent, relay_client)
- .await?;
-
- // Search for the deepest valid parent starting from the pending/included block.
- let best_parent_header =
- find_deepest_valid_parent(start_hash, start_header, backend, &rp_ancestry);
-
- Ok(Some(ParentSearchResult { included_header, best_parent_header }))
+ Ok(Some(ParentSearchStart { included_header, start_hash, start_header }))
}
/// Fetch the included block from the relay chain.
-async fn fetch_included_from_relay_chain(
+pub async fn fetch_included_from_relay_chain(
relay_client: &impl RelayChainInterface,
backend: &impl Backend,
para_id: ParaId,
diff --git a/cumulus/client/consensus/common/src/tests.rs b/cumulus/client/consensus/common/src/tests.rs
index 0f7e273e456e2..c4eb2dfaf9f0c 100644
--- a/cumulus/client/consensus/common/src/tests.rs
+++ b/cumulus/client/consensus/common/src/tests.rs
@@ -309,6 +309,16 @@ impl RelayChainInterface for Relaychain {
unimplemented!("Not needed for test")
}
+ async fn ancestor_relay_parent_info(
+ &self,
+ _at: PHash,
+ _session_index: SessionIndex,
+ _relay_parent: PHash,
+ ) -> RelayChainResult>>
+ {
+ unimplemented!("Not needed for test")
+ }
+
async fn node_features(&self, _at: Hash) -> RelayChainResult {
unimplemented!("Not needed for test")
}
diff --git a/cumulus/client/network/src/tests.rs b/cumulus/client/network/src/tests.rs
index f673c32b7682f..755cd5a0bbec3 100644
--- a/cumulus/client/network/src/tests.rs
+++ b/cumulus/client/network/src/tests.rs
@@ -368,6 +368,16 @@ impl RelayChainInterface for DummyRelayChainInterface {
unimplemented!("Not needed for test")
}
+ async fn ancestor_relay_parent_info(
+ &self,
+ _at: PHash,
+ _session_index: SessionIndex,
+ _relay_parent: PHash,
+ ) -> RelayChainResult>>
+ {
+ unimplemented!("Not needed for test")
+ }
+
async fn node_features(&self, _at: PHash) -> RelayChainResult {
unimplemented!("Not needed for test")
}
diff --git a/cumulus/client/pov-recovery/src/tests.rs b/cumulus/client/pov-recovery/src/tests.rs
index d05e3e3442168..e28c0423e396e 100644
--- a/cumulus/client/pov-recovery/src/tests.rs
+++ b/cumulus/client/pov-recovery/src/tests.rs
@@ -22,8 +22,9 @@ use cumulus_primitives_core::relay_chain::{
BlockId, CandidateCommitments, CandidateDescriptorV2, CoreIndex, CoreState,
};
use cumulus_relay_chain_interface::{
- ChildInfo, InboundDownwardMessage, InboundHrmpMessage, OccupiedCoreAssumption, PHash, PHeader,
- PersistedValidationData, RelayChainResult, StorageValue, ValidationCodeHash, ValidatorId,
+ BlockNumber, ChildInfo, InboundDownwardMessage, InboundHrmpMessage, OccupiedCoreAssumption,
+ PHash, PHeader, PersistedValidationData, RelayChainResult, SessionIndex, StorageValue,
+ ValidationCodeHash, ValidatorId,
};
use cumulus_test_client::runtime::{Block, Header};
use futures::{channel::mpsc, SinkExt, Stream};
@@ -533,6 +534,16 @@ impl RelayChainInterface for Relaychain {
unimplemented!("Not needed for test");
}
+ async fn ancestor_relay_parent_info(
+ &self,
+ _at: PHash,
+ _session_index: SessionIndex,
+ _relay_parent: PHash,
+ ) -> RelayChainResult>>
+ {
+ unimplemented!("Not needed for test");
+ }
+
async fn node_features(&self, _at: PHash) -> RelayChainResult {
unimplemented!("Not needed for test");
}
diff --git a/cumulus/client/relay-chain-inprocess-interface/src/lib.rs b/cumulus/client/relay-chain-inprocess-interface/src/lib.rs
index 552a98e30d0c9..ff6fad4969a57 100644
--- a/cumulus/client/relay-chain-inprocess-interface/src/lib.rs
+++ b/cumulus/client/relay-chain-inprocess-interface/src/lib.rs
@@ -34,7 +34,7 @@ use cumulus_primitives_core::{
InboundDownwardMessage, ParaId, PersistedValidationData,
};
use cumulus_relay_chain_interface::{
- ChildInfo, RelayChainError, RelayChainInterface, RelayChainResult,
+ ChildInfo, RelayChainError, RelayChainInterface, RelayChainResult, RelayParentInfo,
};
use futures::{FutureExt, Stream, StreamExt};
use polkadot_primitives::{CandidateEvent, NodeFeatures};
@@ -353,6 +353,18 @@ impl RelayChainInterface for RelayChainInProcessInterface {
Ok(self.full_client.runtime_api().max_relay_parent_session_age(at)?)
}
+ async fn ancestor_relay_parent_info(
+ &self,
+ at: PHash,
+ session_index: SessionIndex,
+ relay_parent: PHash,
+ ) -> RelayChainResult>> {
+ Ok(self
+ .full_client
+ .runtime_api()
+ .ancestor_relay_parent_info(at, session_index, relay_parent)?)
+ }
+
async fn node_features(&self, at: PHash) -> RelayChainResult {
Ok(self.full_client.runtime_api().node_features(at)?)
}
diff --git a/cumulus/client/relay-chain-interface/src/lib.rs b/cumulus/client/relay-chain-interface/src/lib.rs
index d76a43428c7ec..9581356c02efd 100644
--- a/cumulus/client/relay-chain-interface/src/lib.rs
+++ b/cumulus/client/relay-chain-interface/src/lib.rs
@@ -34,6 +34,7 @@ use sp_api::ApiError;
use cumulus_primitives_core::relay_chain::{
BlockId, CandidateEvent, Hash as RelayHash, NodeFeatures,
};
+pub use cumulus_primitives_core::relay_chain::vstaging::RelayParentInfo;
pub use cumulus_primitives_core::{
relay_chain::{
BlockNumber, CommittedCandidateReceiptV2 as CommittedCandidateReceipt, CoreIndex,
@@ -264,6 +265,17 @@ pub trait RelayChainInterface: Send + Sync {
async fn max_relay_parent_session_age(&self, at: RelayHash) -> RelayChainResult;
+ /// Look up info for `relay_parent` (an ancestor of `at`) in session `session_index`.
+ /// Returns `None` if `relay_parent` is not in `at`'s allowed relay parents for that
+ /// session. A block is never in its own allowed list, so calling this with
+ /// `relay_parent == at` always yields `None`.
+ async fn ancestor_relay_parent_info(
+ &self,
+ at: RelayHash,
+ session_index: SessionIndex,
+ relay_parent: RelayHash,
+ ) -> RelayChainResult>>;
+
async fn node_features(&self, at: RelayHash) -> RelayChainResult;
}
@@ -441,6 +453,15 @@ where
(**self).max_relay_parent_session_age(at).await
}
+ async fn ancestor_relay_parent_info(
+ &self,
+ at: RelayHash,
+ session_index: SessionIndex,
+ relay_parent: RelayHash,
+ ) -> RelayChainResult>> {
+ (**self).ancestor_relay_parent_info(at, session_index, relay_parent).await
+ }
+
async fn node_features(&self, at: RelayHash) -> RelayChainResult {
(**self).node_features(at).await
}
diff --git a/cumulus/client/relay-chain-rpc-interface/src/lib.rs b/cumulus/client/relay-chain-rpc-interface/src/lib.rs
index 12e5478dd850f..208d42ae54147 100644
--- a/cumulus/client/relay-chain-rpc-interface/src/lib.rs
+++ b/cumulus/client/relay-chain-rpc-interface/src/lib.rs
@@ -27,7 +27,7 @@ use cumulus_primitives_core::{
};
use cumulus_relay_chain_interface::{
BlockNumber, ChildInfo, CoreIndex, CoreState, PHeader, RelayChainError, RelayChainInterface,
- RelayChainResult,
+ RelayChainResult, RelayParentInfo,
};
use futures::{FutureExt, Stream, StreamExt};
use polkadot_overseer::Handle;
@@ -311,6 +311,17 @@ impl RelayChainInterface for RelayChainRpcInterface {
self.rpc_client.parachain_host_max_relay_parent_session_age(at).await
}
+ async fn ancestor_relay_parent_info(
+ &self,
+ at: RelayHash,
+ session_index: SessionIndex,
+ relay_parent: RelayHash,
+ ) -> RelayChainResult>> {
+ self.rpc_client
+ .parachain_host_ancestor_relay_parent_info(at, session_index, relay_parent)
+ .await
+ }
+
async fn node_features(&self, at: RelayHash) -> RelayChainResult {
self.rpc_client.parachain_host_node_features(at).await
}
diff --git a/cumulus/client/unincluded-segment-store/Cargo.toml b/cumulus/client/unincluded-segment-store/Cargo.toml
new file mode 100644
index 0000000000000..70636b33421d2
--- /dev/null
+++ b/cumulus/client/unincluded-segment-store/Cargo.toml
@@ -0,0 +1,26 @@
+[package]
+authors.workspace = true
+name = "cumulus-client-unincluded-segment-store"
+version = "0.1.0"
+edition.workspace = true
+description = "Per-block proof store for unincluded segment resubmission."
+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 }
+cumulus-client-consensus-common = { 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 }
+
+[dev-dependencies]
+hex = { workspace = true, default-features = true }
+sc-client-db = { workspace = true, default-features = true }
+substrate-test-runtime = { workspace = true }
+tempfile = { workspace = true }
diff --git a/cumulus/client/unincluded-segment-store/src/lib.rs b/cumulus/client/unincluded-segment-store/src/lib.rs
new file mode 100644
index 0000000000000..873416134cf31
--- /dev/null
+++ b/cumulus/client/unincluded-segment-store/src/lib.rs
@@ -0,0 +1,440 @@
+// 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 .
+
+//! Per-block proof store for unincluded segment resubmission.
+//!
+//! Persists `(time, StorageProof)` per imported parablock, keyed by parablock hash. Data is
+//! pruned on parachain finality.
+
+use codec::{Decode, Encode};
+use cumulus_client_consensus_common::old_finalized_hash;
+use sc_client_api::{
+ backend::AuxStore,
+ client::{AuxDataOperations, FinalityNotification, PreCommitActions},
+ HeaderBackend,
+};
+use sp_blockchain::{Error as ClientError, Result as ClientResult};
+use sp_runtime::traits::{Block as BlockT, Header as _};
+use sp_trie::StorageProof;
+use std::{
+ marker::PhantomData,
+ sync::Arc,
+ time::{SystemTime, UNIX_EPOCH},
+};
+
+const STORE_VERSION_KEY: &[u8] = b"cumulus_unincluded_segment_store_version";
+const STORE_CURRENT_VERSION: u32 = 1;
+const STORE_ENTRY_PREFIX: &[u8] = b"cumulus_unincluded_segment_store";
+
+/// Return the current Unix milliseconds timestamp.
+pub fn now_unix_ms() -> u64 {
+ SystemTime::now()
+ .duration_since(UNIX_EPOCH)
+ .expect("system clock is before UNIX epoch; qed")
+ .as_millis() as u64
+}
+
+/// Entry stored in aux storage for each unincluded parablock.
+#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)]
+pub struct StoredEntry {
+ /// Unix millis at the moment block_import started.
+ pub time_ms: u64,
+ /// The storage proof captured at block_import.
+ pub proof: StorageProof,
+}
+
+fn entry_key(block_hash: H) -> Vec {
+ (STORE_ENTRY_PREFIX, block_hash).encode()
+}
+
+/// Per-block proof store backed by `AuxStore`.
+pub struct UnincludedSegmentStore {
+ backend: Arc,
+ _marker: PhantomData Block>,
+}
+
+impl Clone for UnincludedSegmentStore {
+ fn clone(&self) -> Self {
+ Self { backend: self.backend.clone(), _marker: PhantomData }
+ }
+}
+
+impl UnincludedSegmentStore {
+ /// Create a new store over `backend`.
+ pub fn new(backend: Arc) -> Self {
+ Self { backend, _marker: PhantomData }
+ }
+}
+
+/// Build the aux-data key/value pairs to commit alongside a block.
+///
+/// The caller should push these into `BlockImportParams::auxiliary` so they commit in the
+/// same DB transaction as the block. Stateless — no backend access required.
+pub fn prepare_aux_data(
+ block_hash: Block::Hash,
+ time_ms: u64,
+ proof: &StorageProof,
+) -> impl Iterator- , Vec)> {
+ let encoded_entry = (time_ms, proof).encode();
+ let encoded_version = STORE_CURRENT_VERSION.encode();
+
+ [(entry_key(block_hash), encoded_entry), (STORE_VERSION_KEY.to_vec(), encoded_version)]
+ .into_iter()
+}
+
+impl UnincludedSegmentStore {
+ /// Load the entry stored for `block_hash`, if any.
+ pub fn load(&self, block_hash: Block::Hash) -> ClientResult> {
+ let version = self.decode_aux::(STORE_VERSION_KEY)?;
+
+ match version {
+ None => Ok(None),
+ Some(STORE_CURRENT_VERSION) => self.decode_aux(entry_key(block_hash).as_slice()),
+ Some(other) => Err(ClientError::Backend(format!(
+ "Unsupported unincluded segment store DB version: {:?}",
+ other
+ ))),
+ }
+ }
+
+ fn decode_aux(&self, key: &[u8]) -> ClientResult> {
+ match self.backend.get_aux(key)? {
+ None => Ok(None),
+ Some(t) => T::decode(&mut &t[..]).map(Some).map_err(|e| {
+ ClientError::Backend(format!(
+ "Unincluded segment store DB is corrupted. Decode error: {}",
+ e
+ ))
+ }),
+ }
+ }
+}
+
+impl UnincludedSegmentStore
+where
+ Block: BlockT,
+ B: PreCommitActions + HeaderBackend + 'static,
+{
+ /// Register a finality hook that prunes entries for the just-finalized chain, the
+ /// intermediate tree route, the prior finalized head, and any stale forks.
+ ///
+ /// TODO(#12034): also prune entries whose relay-parent session is older than
+ /// `max_relay_parent_session_age` relative to the current session. Session info will be
+ /// sourced from a separate session-data cache (see #11624), not from per-entry storage.
+ pub fn register_cleanup(&self) {
+ let client = self.backend.clone();
+ let on_finality = move |notification: &FinalityNotification| -> AuxDataOperations {
+ let old_finalized = old_finalized_hash::<_, Block>(
+ &*client,
+ ¬ification.tree_route,
+ *notification.header.parent_hash(),
+ );
+
+ finality_cleanup_ops::(
+ notification.hash,
+ old_finalized,
+ ¬ification.tree_route,
+ notification.stale_blocks.iter().map(|b| b.hash),
+ )
+ };
+
+ self.backend.register_finality_action(Box::new(on_finality));
+ }
+}
+
+/// Compute aux storage cleanup operations.
+///
+/// Emits deletes for stale-fork blocks, intermediate tree-route blocks, the pre-finality head,
+/// and the just-finalized block itself. Once a block is finalized it is no longer in any
+/// unincluded segment, so its proof entry is dead weight.
+fn finality_cleanup_ops(
+ just_finalized_hash: Block::Hash,
+ old_finalized_hash: Block::Hash,
+ tree_route: &[Block::Hash],
+ stale_block_hashes: impl IntoIterator
- ,
+) -> AuxDataOperations {
+ let stale_iter = stale_block_hashes.into_iter();
+
+ let mut ops = Vec::with_capacity(stale_iter.size_hint().0 + tree_route.len() + 2);
+ ops.extend(stale_iter.map(|hash| (entry_key(hash), None)));
+ ops.extend(tree_route.iter().map(|hash| (entry_key(hash), None)));
+ ops.push((entry_key(old_finalized_hash), None));
+ ops.push((entry_key(just_finalized_hash), None));
+
+ ops
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use sc_client_api::backend::AuxStore;
+
+ type Block = substrate_test_runtime::Block;
+ type Hash = ::Hash;
+ type TestBackend = sc_client_api::in_mem::Backend;
+ type Store = UnincludedSegmentStore;
+
+ fn create_test_entry(time_ms: u64) -> StoredEntry {
+ StoredEntry { time_ms, proof: StorageProof::new(vec![vec![1, 2, 3], vec![4, 5, 6]]) }
+ }
+
+ fn new_store() -> (Arc, Store) {
+ let backend = Arc::new(TestBackend::new());
+ let store = Store::new(backend.clone());
+ (backend, store)
+ }
+
+ fn write_via_store(backend: &Arc, hash: Hash, entry: &StoredEntry) {
+ let pairs: Vec<_> = prepare_aux_data::(hash, entry.time_ms, &entry.proof).collect();
+ let insert_pairs: Vec<_> =
+ pairs.iter().map(|(k, v)| (k.as_slice(), v.as_slice())).collect();
+ AuxStore::insert_aux(&**backend, &insert_pairs, &[]).expect("aux insert should succeed");
+ }
+
+ #[test]
+ fn prepare_produces_expected_key_value_pairs() {
+ let hash = Hash::repeat_byte(0xAB);
+ let time_ms = 1234567890u64;
+ let proof = StorageProof::new(vec![vec![10, 20, 30]]);
+
+ let pairs: Vec<_> = prepare_aux_data::(hash, time_ms, &proof).collect();
+
+ assert_eq!(pairs.len(), 2);
+
+ let expected_key = (STORE_ENTRY_PREFIX, hash).encode();
+ assert_eq!(pairs[0].0, expected_key);
+
+ let decoded_entry =
+ StoredEntry::decode(&mut pairs[0].1.as_slice()).expect("entry should decode");
+ assert_eq!(decoded_entry.time_ms, time_ms);
+ assert_eq!(decoded_entry.proof, proof);
+
+ assert_eq!(pairs[1].0, STORE_VERSION_KEY.to_vec());
+ let decoded_version =
+ u32::decode(&mut pairs[1].1.as_slice()).expect("version should decode");
+ assert_eq!(decoded_version, STORE_CURRENT_VERSION);
+ }
+
+ #[test]
+ fn load_returns_none_when_no_entry_exists() {
+ let (_backend, store) = new_store();
+ let hash = Hash::repeat_byte(0xEF);
+
+ assert_eq!(store.load(hash).expect("load should succeed"), None);
+ }
+
+ #[test]
+ fn cleanup_combines_all_categories() {
+ let stale_1 = Hash::repeat_byte(0xAA);
+ let stale_2 = Hash::repeat_byte(0xBB);
+ let route_1 = Hash::repeat_byte(0xC1);
+ let route_2 = Hash::repeat_byte(0xC2);
+ let old_finalized = Hash::repeat_byte(0xF0);
+ let just_finalized = Hash::repeat_byte(0xFF);
+
+ let ops = finality_cleanup_ops::(
+ just_finalized,
+ old_finalized,
+ &[route_1, route_2],
+ [stale_1, stale_2],
+ );
+
+ let keys: Vec<_> = ops.iter().map(|(k, _)| k.clone()).collect();
+
+ assert!(keys.contains(&entry_key(stale_1)));
+ assert!(keys.contains(&entry_key(stale_2)));
+ assert!(keys.contains(&entry_key(route_1)));
+ assert!(keys.contains(&entry_key(route_2)));
+ assert!(keys.contains(&entry_key(old_finalized)));
+ assert!(keys.contains(&entry_key(just_finalized)));
+
+ assert!(ops.iter().all(|(_, v)| v.is_none()));
+ }
+
+ #[test]
+ fn cleanup_handles_empty_inputs() {
+ let just_finalized = Hash::repeat_byte(0xFF);
+ let old_finalized = Hash::repeat_byte(0xF0);
+
+ let ops = finality_cleanup_ops::(
+ just_finalized,
+ old_finalized,
+ &[],
+ std::iter::empty::(),
+ );
+
+ assert_eq!(ops.len(), 2);
+ assert!(ops.iter().all(|(_, v)| v.is_none()));
+ }
+
+ #[test]
+ fn stored_entry_encoding_hex_snapshot() {
+ let entry =
+ StoredEntry { time_ms: 1234567890u64, proof: StorageProof::new(vec![vec![1, 2, 3]]) };
+
+ let encoded = entry.encode();
+ // Snapshot of the SCALE encoding. If this assertion fires, the on-disk format of
+ // `StoredEntry` has changed and existing aux entries written by older builds will fail
+ // to decode — bump `STORE_CURRENT_VERSION` and add a migration before updating this
+ // snapshot.
+ let encoded_hex = hex::encode(&encoded);
+ // time_ms = 1234567890 little-endian u64 = d2 02 96 49 00 00 00 00
+ // proof = Vec> with one element [1,2,3]: outer len 1 (SCALE compact = 04),
+ // inner len 3 (compact = 0c), bytes 01 02 03
+ let expected_hex = "d20296490000000004 0c 010203".replace(' ', "");
+ assert_eq!(encoded_hex, expected_hex, "StoredEntry encoding changed!");
+
+ let decoded = StoredEntry::decode(&mut encoded.as_slice()).expect("decode should succeed");
+ assert_eq!(entry, decoded);
+ }
+
+ #[test]
+ fn decode_corrupted_entry_body() {
+ let (backend, store) = new_store();
+ let hash = Hash::repeat_byte(0xAB);
+
+ // Write correct version.
+ let version_encoded = STORE_CURRENT_VERSION.encode();
+ AuxStore::insert_aux(&*backend, &[(STORE_VERSION_KEY, version_encoded.as_slice())], &[])
+ .expect("aux insert should succeed");
+
+ // Write bogus entry body.
+ let key = entry_key(hash);
+ let bogus_data = vec![0xFF, 0xAA, 0xBB];
+ AuxStore::insert_aux(&*backend, &[(&key[..], bogus_data.as_slice())], &[])
+ .expect("aux insert should succeed");
+
+ let result = store.load(hash);
+ assert!(result.is_err());
+ let err_msg = result.unwrap_err().to_string();
+ assert!(
+ err_msg.contains("DB is corrupted") && err_msg.contains("Decode error"),
+ "unexpected error: {}",
+ err_msg
+ );
+ }
+
+ #[test]
+ fn end_to_end_write_cleanup_load() {
+ let (backend, store) = new_store();
+
+ let hash1 = Hash::repeat_byte(0x01);
+ let hash2 = Hash::repeat_byte(0x02);
+ let hash3 = Hash::repeat_byte(0x03);
+
+ let entry1 = create_test_entry(1000);
+ let entry2 = create_test_entry(2000);
+ let entry3 = create_test_entry(3000);
+
+ write_via_store(&backend, hash1, &entry1);
+ write_via_store(&backend, hash2, &entry2);
+ write_via_store(&backend, hash3, &entry3);
+
+ assert_eq!(store.load(hash1).expect("load"), Some(entry1));
+ assert_eq!(store.load(hash2).expect("load"), Some(entry2));
+ assert_eq!(store.load(hash3).expect("load"), Some(entry3.clone()));
+
+ // Generate cleanup that deletes hash1 (just-finalized) and hash2 (in tree route).
+ let ops = finality_cleanup_ops::(
+ hash1,
+ Hash::repeat_byte(0xF0),
+ &[hash2],
+ std::iter::empty::(),
+ );
+ let delete_keys: Vec<_> =
+ ops.iter().filter_map(|(k, v)| v.is_none().then(|| k.as_slice())).collect();
+
+ AuxStore::insert_aux(&*backend, &[], &delete_keys).expect("delete should succeed");
+
+ assert_eq!(store.load(hash1).expect("load"), None, "hash1 should be deleted");
+ assert_eq!(store.load(hash2).expect("load"), None, "hash2 should be deleted");
+ assert_eq!(store.load(hash3).expect("load"), Some(entry3), "hash3 should survive");
+ }
+
+ #[test]
+ fn entries_survive_disk_restart() {
+ use sc_client_db::{
+ Backend as DbBackend, BlocksPruning, DatabaseSettings, DatabaseSource, PruningMode,
+ };
+
+ fn with_backend(
+ path: &std::path::Path,
+ f: impl FnOnce(&Arc>) -> R,
+ ) -> R {
+ let backend = Arc::new(
+ DbBackend::::new(
+ DatabaseSettings {
+ trie_cache_maximum_size: Some(16 * 1024 * 1024),
+ state_pruning: Some(PruningMode::ArchiveAll),
+ blocks_pruning: BlocksPruning::KeepAll,
+ pruning_filters: Default::default(),
+ source: DatabaseSource::ParityDb { path: path.to_path_buf() },
+ metrics_registry: None,
+ },
+ 0,
+ )
+ .expect("open backend"),
+ );
+ let result = f(&backend);
+ // `backend` (and any clones held by the closure) drop here, closing parity-db.
+ result
+ }
+
+ let tmp = tempfile::tempdir().expect("tempdir");
+ let path = tmp.path();
+
+ let hash_a = Hash::repeat_byte(0x10);
+ let hash_b = Hash::repeat_byte(0x20);
+ let entry_a = create_test_entry(10_000);
+ let entry_b = create_test_entry(20_000);
+
+ // Write `a` and `b` via the same path block-import uses, then close.
+ with_backend(path, |backend| {
+ let pairs: Vec<_> = prepare_aux_data::(hash_a, entry_a.time_ms, &entry_a.proof)
+ .chain(prepare_aux_data::(hash_b, entry_b.time_ms, &entry_b.proof))
+ .collect();
+ let refs: Vec<_> = pairs.iter().map(|(k, v)| (k.as_slice(), v.as_slice())).collect();
+ AuxStore::insert_aux(&**backend, &refs, &[]).expect("aux insert");
+ });
+
+ // Restart: confirm both entries survived, then apply a finality-style delete of `a`.
+ with_backend(path, |backend| {
+ let store = UnincludedSegmentStore::::new(backend.clone());
+ assert_eq!(store.load(hash_a).expect("load a"), Some(entry_a.clone()));
+ assert_eq!(store.load(hash_b).expect("load b"), Some(entry_b.clone()));
+
+ let ops = finality_cleanup_ops::(
+ hash_a,
+ Hash::repeat_byte(0xF0),
+ &[],
+ std::iter::empty::(),
+ );
+ let delete_keys: Vec<_> =
+ ops.iter().filter_map(|(k, v)| v.is_none().then(|| k.as_slice())).collect();
+ AuxStore::insert_aux(&**backend, &[], &delete_keys).expect("delete");
+ });
+
+ // Restart: the delete must have persisted; `b` must still be there.
+ with_backend(path, |backend| {
+ let store = UnincludedSegmentStore::::new(backend.clone());
+ assert_eq!(store.load(hash_a).expect("load a"), None, "hash_a delete must persist");
+ assert_eq!(store.load(hash_b).expect("load b"), Some(entry_b));
+ });
+
+ // `tmp` drops here, recursively removing the parity-db directory.
+ }
+}
diff --git a/cumulus/pallets/aura-ext/Cargo.toml b/cumulus/pallets/aura-ext/Cargo.toml
index ad633865e0fb4..1721446d4db82 100644
--- a/cumulus/pallets/aura-ext/Cargo.toml
+++ b/cumulus/pallets/aura-ext/Cargo.toml
@@ -22,17 +22,18 @@ pallet-aura = { workspace = true }
pallet-timestamp = { workspace = true }
sp-application-crypto = { workspace = true }
sp-consensus-aura = { workspace = true }
+sp-consensus-babe = { workspace = true }
sp-runtime = { workspace = true }
# Cumulus
cumulus-pallet-parachain-system = { workspace = true }
+cumulus-primitives-core = { workspace = true }
[dev-dependencies]
rstest = { workspace = true }
# Cumulus
cumulus-pallet-parachain-system = { workspace = true, default-features = true }
-cumulus-primitives-core = { workspace = true, default-features = true }
cumulus-primitives-proof-size-hostfunction = { workspace = true, default-features = true }
cumulus-test-relay-sproof-builder = { workspace = true, default-features = true }
@@ -49,6 +50,7 @@ default = ["std"]
std = [
"codec/std",
"cumulus-pallet-parachain-system/std",
+ "cumulus-primitives-core/std",
"frame-support/std",
"frame-system/std",
"pallet-aura/std",
@@ -56,6 +58,7 @@ std = [
"scale-info/std",
"sp-application-crypto/std",
"sp-consensus-aura/std",
+ "sp-consensus-babe/std",
"sp-runtime/std",
]
try-runtime = [
diff --git a/cumulus/pallets/aura-ext/src/lib.rs b/cumulus/pallets/aura-ext/src/lib.rs
index 5fcc09b8ef420..47ed329a29978 100644
--- a/cumulus/pallets/aura-ext/src/lib.rs
+++ b/cumulus/pallets/aura-ext/src/lib.rs
@@ -40,10 +40,12 @@ use sp_runtime::traits::{Block as BlockT, Header as HeaderT, LazyBlock};
pub mod consensus_hook;
pub mod migration;
+pub mod signature_verifier;
#[cfg(test)]
mod test;
pub use consensus_hook::FixedVelocityConsensusHook;
+pub use signature_verifier::AuraSchedulingVerifier;
type Aura