PubSub Mechanism - Follow up on #9994

Cargo.toml

@@ -85,6 +85,7 @@ members = [
 	"cumulus/pallets/parachain-system/proc-macro",
 	"cumulus/pallets/session-benchmarking",
 	"cumulus/pallets/solo-to-para",
+	"cumulus/pallets/subscriber",
 	"cumulus/pallets/weight-reclaim",
 	"cumulus/pallets/xcm",
 	"cumulus/pallets/xcmp-queue",
@@ -748,6 +749,7 @@ cumulus-pallet-aura-ext = { path = "cumulus/pallets/aura-ext", default-features
 cumulus-pallet-dmp-queue = { default-features = false, path = "cumulus/pallets/dmp-queue" }
 cumulus-pallet-parachain-system = { path = "cumulus/pallets/parachain-system", default-features = false }
 cumulus-pallet-parachain-system-proc-macro = { path = "cumulus/pallets/parachain-system/proc-macro", default-features = false }
+cumulus-pallet-subscriber = { path = "cumulus/pallets/subscriber", default-features = false }
 cumulus-pallet-session-benchmarking = { path = "cumulus/pallets/session-benchmarking", default-features = false }
 cumulus-pallet-solo-to-para = { path = "cumulus/pallets/solo-to-para", default-features = false }
 cumulus-pallet-weight-reclaim = { path = "cumulus/pallets/weight-reclaim", default-features = false }

cumulus/client/consensus/aura/src/collator.rs

@@ -177,6 +177,7 @@ where
 		parent_hash: Block::Hash,
 		timestamp: impl Into<Option<Timestamp>>,
 		relay_parent_descendants: Option<RelayParentData>,
+		relay_proof_request: cumulus_primitives_core::RelayProofRequest,
 		collator_peer_id: PeerId,
 	) -> Result<(ParachainInherentData, InherentData), Box<dyn Error + Send + Sync + 'static>> {
 		let paras_inherent_data = ParachainInherentDataProvider::create_at(
@@ -188,6 +189,7 @@ where
 				.map(RelayParentData::into_inherent_descendant_list)
 				.unwrap_or_default(),
 			Vec::new(),
+			relay_proof_request,
 			collator_peer_id,
 		)
 		.await;
@@ -224,6 +226,7 @@ where
 		validation_data: &PersistedValidationData,
 		parent_hash: Block::Hash,
 		timestamp: impl Into<Option<Timestamp>>,
+		relay_proof_request: cumulus_primitives_core::RelayProofRequest,
 		collator_peer_id: PeerId,
 	) -> Result<(ParachainInherentData, InherentData), Box<dyn Error + Send + Sync + 'static>> {
 		self.create_inherent_data_with_rp_offset(
@@ -232,6 +235,7 @@ where
 			parent_hash,
 			timestamp,
 			None,
+			relay_proof_request,
 			collator_peer_id,
 		)
 		.await

cumulus/client/consensus/aura/src/collators/basic.rs

@@ -238,6 +238,7 @@ where
 						&validation_data,
 						parent_hash,
 						claim.timestamp(),
+						Default::default(),
 						params.collator_peer_id,
 					)
 					.await

cumulus/client/consensus/aura/src/collators/lookahead.rs

@@ -36,7 +36,7 @@ use codec::{Codec, Encode};
 use cumulus_client_collator::service::ServiceInterface as CollatorServiceInterface;
 use cumulus_client_consensus_common::{self as consensus_common, ParachainBlockImportMarker};
 use cumulus_primitives_aura::AuraUnincludedSegmentApi;
-use cumulus_primitives_core::{CollectCollationInfo, PersistedValidationData};
+use cumulus_primitives_core::{CollectCollationInfo, KeyToIncludeInRelayProofApi, PersistedValidationData};
 use cumulus_relay_chain_interface::RelayChainInterface;
 use sp_consensus::Environment;
 
@@ -164,8 +164,10 @@ where
 		+ Send
 		+ Sync
 		+ 'static,
-	Client::Api:
-		AuraApi<Block, P::Public> + CollectCollationInfo<Block> + AuraUnincludedSegmentApi<Block>,
+	Client::Api: AuraApi<Block, P::Public>
+		+ CollectCollationInfo<Block>
+		+ AuraUnincludedSegmentApi<Block>
+		+ KeyToIncludeInRelayProofApi<Block>,
 	Backend: sc_client_api::Backend<Block> + 'static,
 	RClient: RelayChainInterface + Clone + 'static,
 	CIDP: CreateInherentDataProviders<Block, ()> + 'static,
@@ -216,8 +218,10 @@ where
 		+ Send
 		+ Sync
 		+ 'static,
-	Client::Api:
-		AuraApi<Block, P::Public> + CollectCollationInfo<Block> + AuraUnincludedSegmentApi<Block>,
+	Client::Api: AuraApi<Block, P::Public>
+		+ CollectCollationInfo<Block>
+		+ AuraUnincludedSegmentApi<Block>
+		+ KeyToIncludeInRelayProofApi<Block>,
 	Backend: sc_client_api::Backend<Block> + 'static,
 	RClient: RelayChainInterface + Clone + 'static,
 	CIDP: CreateInherentDataProviders<Block, ()> + 'static,
@@ -392,12 +396,15 @@ where
 
 				// Build and announce collations recursively until
 				// `can_build_upon` fails or building a collation fails.
+				let relay_proof_request = super::get_relay_proof_request(&*params.para_client, parent_hash);
+
 				let (parachain_inherent_data, other_inherent_data) = match collator
 					.create_inherent_data(
 						relay_parent,
 						&validation_data,
 						parent_hash,
 						slot_claim.timestamp(),
+						relay_proof_request,
 						params.collator_peer_id,
 					)
 					.await

cumulus/client/consensus/aura/src/collators/mod.rs

@@ -25,7 +25,9 @@ use crate::collator::SlotClaim;
 use codec::Codec;
 use cumulus_client_consensus_common::{self as consensus_common, ParentSearchParams};
 use cumulus_primitives_aura::{AuraUnincludedSegmentApi, Slot};
-use cumulus_primitives_core::{relay_chain::Header as RelayHeader, BlockT};
+use cumulus_primitives_core::{
+	relay_chain::Header as RelayHeader, BlockT, KeyToIncludeInRelayProofApi,
+};
 use cumulus_relay_chain_interface::{OverseerHandle, RelayChainInterface};
 use polkadot_node_subsystem::messages::{CollatorProtocolMessage, RuntimeApiRequest};
 use polkadot_node_subsystem_util::runtime::ClaimQueueSnapshot;
@@ -662,6 +664,34 @@ mod tests {
 	}
 }
 
+/// Fetches relay chain storage proof requests from the parachain runtime.
+///
+/// Queries the runtime API to determine which relay chain storage keys
+/// (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<Block, Client>(
+	client: &Client,
+	parent_hash: Block::Hash,
+) -> cumulus_primitives_core::RelayProofRequest
+where
+	Block: BlockT,
+	Client: ProvideRuntimeApi<Block>,
+	Client::Api: KeyToIncludeInRelayProofApi<Block>,
+{
+	client
+		.runtime_api()
+		.keys_to_prove(parent_hash)
+		.unwrap_or_else(|e| {
+			tracing::warn!(
+				target: crate::LOG_TARGET,
+				error = ?e,
+				"Failed to fetch relay proof requests from runtime, using empty request"
+			);
+			Default::default()
+		})
+}
+
 /// Holds a relay parent and its descendants.
 pub struct RelayParentData {
 	/// The relay parent block header

cumulus/client/consensus/aura/src/collators/slot_based/block_builder_task.rs

@@ -359,14 +359,14 @@ where
 				relay_parent_storage_root: *relay_parent_header.state_root(),
 				max_pov_size: *max_pov_size,
 			};
-
 			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),
+					Default::default(),
 					collator_peer_id,
 				)
 				.await

cumulus/client/consensus/aura/src/collators/slot_based/tests.rs

@@ -566,6 +566,15 @@ impl RelayChainInterface for TestRelayClient {
 		unimplemented!("Not needed for test")
 	}
 
+	async fn prove_child_read(
+		&self,
+		_: RelayHash,
+		_: &cumulus_relay_chain_interface::ChildInfo,
+		_: &[Vec<u8>],
+	) -> RelayChainResult<sc_client_api::StorageProof> {
+		unimplemented!("Not needed for test")
+	}
+
 	async fn wait_for_block(&self, _: RelayHash) -> RelayChainResult<()> {
 		unimplemented!("Not needed for test")
 	}

cumulus/client/parachain-inherent/src/lib.rs

@@ -30,6 +30,8 @@ pub use cumulus_primitives_parachain_inherent::{ParachainInherentData, INHERENT_
 use cumulus_relay_chain_interface::RelayChainInterface;
 pub use mock::{MockValidationDataInherentDataProvider, MockXcmConfig};
 use sc_network_types::PeerId;
+use sp_state_machine::StorageProof;
+use sp_storage::ChildInfo;
 
 const LOG_TARGET: &str = "parachain-inherent";
 
@@ -157,6 +159,84 @@ async fn collect_relay_storage_proof(
 		.ok()
 }
 
+/// Collect storage proofs for relay chain data.
+///
+/// Generates proofs for both top-level relay chain storage and child trie data.
+/// Top-level keys are proven directly. Child trie roots are automatically included
+/// from their standard storage locations (`:child_storage:default:` + identifier).
+///
+/// Returns a merged proof combining all requested data, or `None` if there are no requests.
+async fn collect_relay_storage_proofs(
+	relay_chain_interface: &impl RelayChainInterface,
+	relay_parent: PHash,
+	relay_proof_request: cumulus_primitives_core::RelayProofRequest,
+) -> Option<StorageProof> {
+	use cumulus_primitives_core::RelayStorageKey;
+
+	let cumulus_primitives_core::RelayProofRequest { keys } = relay_proof_request;
+
+	if keys.is_empty() {
+		return None;
+	}
+
+	let mut combined_proof: Option<StorageProof> = None;
+
+	// Group keys by storage type
+	let mut top_keys = Vec::new();
+	let mut child_keys: std::collections::BTreeMap<Vec<u8>, Vec<Vec<u8>>> =
+		std::collections::BTreeMap::new();
+
+	for key in keys {
+		match key {
+			RelayStorageKey::Top(k) => top_keys.push(k),
+			RelayStorageKey::Child { storage_key, key } => {
+				child_keys.entry(storage_key).or_default().push(key);
+			},
+		}
+	}
+
+	// Collect top-level storage proofs
+	if !top_keys.is_empty() {
+		match relay_chain_interface.prove_read(relay_parent, &top_keys).await {
+			Ok(top_proof) => {
+				combined_proof = Some(top_proof);
+			},
+			Err(e) => {
+				tracing::error!(
+					target: LOG_TARGET,
+					relay_parent = ?relay_parent,
+					error = ?e,
+					"Cannot obtain top-level storage proof from relay chain.",
+				);
+			},
+		}
+	}
+
+	// Collect child trie proofs
+	for (storage_key, data_keys) in child_keys {
+		let child_info = ChildInfo::new_default(&storage_key);
+		match relay_chain_interface.prove_child_read(relay_parent, &child_info, &data_keys).await {
+			Ok(child_proof) => {
+				combined_proof = match combined_proof {
+					None => Some(child_proof),
+					Some(existing) => Some(StorageProof::merge([existing, child_proof])),
+				};
+			},
+			Err(e) => {
+				tracing::error!(
+					target: LOG_TARGET,
+					relay_parent = ?relay_parent,
+					child_trie_id = ?child_info.storage_key(),
+					error = ?e,
+					"Cannot obtain child trie proof from relay chain.",
+				);
+			},
+		}
+	}
+
+	combined_proof
+}
+
 pub struct ParachainInherentDataProvider;
 
 impl ParachainInherentDataProvider {
@@ -170,6 +250,7 @@ impl ParachainInherentDataProvider {
 		para_id: ParaId,
 		relay_parent_descendants: Vec<RelayHeader>,
 		additional_relay_state_keys: Vec<Vec<u8>>,
+		relay_proof_request: cumulus_primitives_core::RelayProofRequest,
 		collator_peer_id: PeerId,
 	) -> Option<ParachainInherentData> {
 		let collator_peer_id = ApprovedPeerId::try_from(collator_peer_id.to_bytes())
@@ -188,7 +269,7 @@ impl ParachainInherentDataProvider {
 			.iter()
 			.skip(1)
 			.any(sc_consensus_babe::contains_epoch_change::<RelayBlock>);
-		let relay_chain_state = collect_relay_storage_proof(
+		let mut relay_chain_state = collect_relay_storage_proof(
 			relay_chain_interface,
 			para_id,
 			relay_parent,
@@ -198,6 +279,14 @@ impl ParachainInherentDataProvider {
 		)
 		.await?;
 
+		// Collect additional requested storage proofs (top-level and child tries)
+		if let Some(additional_proofs) =
+			collect_relay_storage_proofs(relay_chain_interface, relay_parent, relay_proof_request)
+				.await
+		{
+			relay_chain_state = StorageProof::merge([relay_chain_state, additional_proofs]);
+		}
+
 		let downward_messages = relay_chain_interface
 			.retrieve_dmq_contents(para_id, relay_parent)
 			.await

cumulus/client/relay-chain-inprocess-interface/src/lib.rs

@@ -240,6 +240,18 @@ impl RelayChainInterface for RelayChainInProcessInterface {
 			.map_err(RelayChainError::StateMachineError)
 	}
 
+	async fn prove_child_read(
+		&self,
+		relay_parent: PHash,
+		child_info: &cumulus_relay_chain_interface::ChildInfo,
+		child_keys: &[Vec<u8>],
+	) -> RelayChainResult<StorageProof> {
+		let state_backend = self.backend.state_at(relay_parent, TrieCacheContext::Untrusted)?;
+
+		sp_state_machine::prove_child_read(state_backend, child_info, child_keys)
+			.map_err(RelayChainError::StateMachineError)
+	}
+
 	/// Wait for a given relay chain block in an async way.
 	///
 	/// The caller needs to pass the hash of a block it waits for and the function will return when

cumulus/client/relay-chain-interface/Cargo.toml

@@ -21,6 +21,7 @@ sc-network = { workspace = true, default-features = true }
 sp-api = { workspace = true, default-features = true }
 sp-blockchain = { workspace = true, default-features = true }
 sp-state-machine = { workspace = true, default-features = true }
+sp-storage = { workspace = true, default-features = true }
 sp-version = { workspace = true }
 
 async-trait = { workspace = true }