PubSub Mechanism - Addresses #606

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

@@ -486,6 +486,14 @@ impl RelayChainInterface for TestRelayClient {
 		unimplemented!("Not needed for test")
 	}
 
+	async fn retrieve_subscribed_published_data(
+		&self,
+		_: ParaId,
+		_: RelayHash,
+	) -> RelayChainResult<BTreeMap<ParaId, Vec<(Vec<u8>, Vec<u8>)>>> {
+		unimplemented!("Not needed for test")
+	}
+
 	async fn persisted_validation_data(
 		&self,
 		_: RelayHash,

cumulus/client/consensus/common/src/tests.rs

@@ -123,6 +123,14 @@ impl RelayChainInterface for Relaychain {
 		unimplemented!("Not needed for test")
 	}
 
+	async fn retrieve_subscribed_published_data(
+		&self,
+		_: ParaId,
+		_: PHash,
+	) -> RelayChainResult<BTreeMap<ParaId, Vec<(Vec<u8>, Vec<u8>)>>> {
+		unimplemented!("Not needed for test")
+	}
+
 	async fn persisted_validation_data(
 		&self,
 		hash: PHash,

cumulus/client/network/src/tests.rs

@@ -140,6 +140,14 @@ impl RelayChainInterface for DummyRelayChainInterface {
 		Ok(BTreeMap::new())
 	}
 
+	async fn retrieve_subscribed_published_data(
+		&self,
+		_: ParaId,
+		_: PHash,
+	) -> RelayChainResult<BTreeMap<ParaId, Vec<(Vec<u8>, Vec<u8>)>>> {
+		Ok(BTreeMap::new())
+	}
+
 	async fn persisted_validation_data(
 		&self,
 		_: PHash,

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

@@ -136,6 +136,9 @@ async fn collect_relay_storage_proof(
 		relevant_keys.push(relay_well_known_keys::NEXT_AUTHORITIES.to_vec());
 	}
 
+	// Include broadcaster published data roots
+	relevant_keys.push(relay_well_known_keys::BROADCASTER_PUBLISHED_DATA_ROOTS.to_vec());
+
 	// Add additional relay state keys
 	let unique_keys: Vec<Vec<u8>> = additional_relay_state_keys
 		.into_iter()
@@ -223,13 +226,27 @@ impl ParachainInherentDataProvider {
 			})
 			.ok()?;
 
+		let published_data = relay_chain_interface
+			.retrieve_subscribed_published_data(para_id, relay_parent)
+			.await
+			.map_err(|e| {
+				tracing::error!(
+					target: LOG_TARGET,
+					relay_parent = ?relay_parent,
+					error = ?e,
+					"An error occurred during requesting subscribed published data.",
+				);
+			})
+			.unwrap_or_default();
+
 		Some(ParachainInherentData {
 			downward_messages,
 			horizontal_messages,
 			validation_data: validation_data.clone(),
 			relay_chain_state,
 			relay_parent_descendants,
 			collator_peer_id,
+			published_data,
 		})
 	}
 }

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

@@ -242,6 +242,7 @@ impl<R: Send + Sync + GenerateRandomness<u64>> InherentDataProvider
 			relay_chain_state: proof,
 			relay_parent_descendants: Default::default(),
 			collator_peer_id: None,
+			published_data: Default::default(),
 		};
 
 		parachain_inherent_data.provide_inherent_data(inherent_data).await

cumulus/client/pov-recovery/src/tests.rs

@@ -364,6 +364,14 @@ impl RelayChainInterface for Relaychain {
 		unimplemented!("Not needed for test")
 	}
 
+	async fn retrieve_subscribed_published_data(
+		&self,
+		_: ParaId,
+		_: PHash,
+	) -> RelayChainResult<BTreeMap<ParaId, Vec<(Vec<u8>, Vec<u8>)>>> {
+		unimplemented!("Not needed for test")
+	}
+
 	async fn persisted_validation_data(
 		&self,
 		_: PHash,

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

@@ -106,6 +106,14 @@ impl RelayChainInterface for RelayChainInProcessInterface {
 			.inbound_hrmp_channels_contents(relay_parent, para_id)?)
 	}
 
+	async fn retrieve_subscribed_published_data(
+		&self,
+		para_id: ParaId,
+		relay_parent: PHash,
+	) -> RelayChainResult<BTreeMap<ParaId, Vec<(Vec<u8>, Vec<u8>)>>> {
+		Ok(self.full_client.runtime_api().get_subscribed_data(relay_parent, para_id)?)
+	}
+
 	async fn header(&self, block_id: BlockId) -> RelayChainResult<Option<PHeader>> {
 		let hash = match block_id {
 			BlockId::Hash(hash) => hash,

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

@@ -151,6 +151,14 @@ pub trait RelayChainInterface: Send + Sync {
 		relay_parent: PHash,
 	) -> RelayChainResult<BTreeMap<ParaId, Vec<InboundHrmpMessage>>>;
 
+	/// Returns published data from all subscribed publishers for the parachain we are collating
+	/// for.
+	async fn retrieve_subscribed_published_data(
+		&self,
+		para_id: ParaId,
+		relay_parent: PHash,
+	) -> RelayChainResult<BTreeMap<ParaId, Vec<(Vec<u8>, Vec<u8>)>>>;
+
 	/// Yields the persisted validation data for the given `ParaId` along with an assumption that
 	/// should be used if the para currently occupies a core.
 	///
@@ -273,6 +281,14 @@ where
 		(**self).retrieve_all_inbound_hrmp_channel_contents(para_id, relay_parent).await
 	}
 
+	async fn retrieve_subscribed_published_data(
+		&self,
+		para_id: ParaId,
+		relay_parent: PHash,
+	) -> RelayChainResult<BTreeMap<ParaId, Vec<(Vec<u8>, Vec<u8>)>>> {
+		(**self).retrieve_subscribed_published_data(para_id, relay_parent).await
+	}
+
 	async fn persisted_validation_data(
 		&self,
 		block_id: PHash,

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

@@ -82,6 +82,14 @@ impl RelayChainInterface for RelayChainRpcInterface {
 			.await
 	}
 
+	async fn retrieve_subscribed_published_data(
+		&self,
+		para_id: ParaId,
+		relay_parent: RelayHash,
+	) -> RelayChainResult<BTreeMap<ParaId, Vec<(Vec<u8>, Vec<u8>)>>> {
+		self.rpc_client.parachain_host_get_subscribed_data(para_id, relay_parent).await
+	}
+
 	async fn header(&self, block_id: BlockId) -> RelayChainResult<Option<PHeader>> {
 		let hash = match block_id {
 			BlockId::Hash(hash) => hash,

cumulus/client/relay-chain-rpc-interface/src/rpc_client.rs

@@ -598,6 +598,20 @@ impl RelayChainRpcClient {
 		.await
 	}
 
+	/// Get published data from all subscribed publishers for a parachain.
+	pub async fn parachain_host_get_subscribed_data(
+		&self,
+		para_id: ParaId,
+		at: RelayHash,
+	) -> Result<BTreeMap<ParaId, Vec<(Vec<u8>, Vec<u8>)>>, RelayChainError> {
+		self.call_remote_runtime_function(
+			"ParachainHost_get_subscribed_data",
+			at,
+			Some(para_id),
+		)
+		.await
+	}
+
 	/// Get all the pending inbound messages in the downward message queue for a para.
 	pub async fn parachain_host_dmq_contents(
 		&self,

cumulus/pallets/parachain-system/src/benchmarking.rs

@@ -54,6 +54,59 @@ mod benchmarks {
 		assert_eq!(LastDmqMqcHead::<T>::get().head(), head);
 	}
 
+	/// Benchmark processing published data from the broadcaster pallet.
+	///
+	/// - `p`: Number of publishers with changed data
+	/// - `k`: Number of key-value pairs per publisher
+	/// - `v`: Size of each value in bytes
+	#[benchmark]
+	fn process_published_data(
+		p: Linear<1, 100>,
+		k: Linear<1, 16>,
+		v: Linear<1, 1024>,
+	) {
+		use alloc::collections::BTreeMap;
+
+		// Populate storage with existing data to maximize clear_prefix cost
+		for i in 0..p {
+			let para_id = ParaId::from(1000 + i);
+			for j in 0..k {
+				PublishedData::<T>::insert(
+					para_id,
+					vec![j as u8; 32],
+					vec![0u8; v as usize],
+				);
+			}
+		}
+
+		// Store initial roots
+		let initial_roots: BTreeMap<ParaId, Vec<u8>> = (0..p)
+			.map(|i| (ParaId::from(1000 + i), vec![0xBB; 32]))
+			.collect();
+		PreviousPublishedDataRoots::<T>::put(initial_roots);
+
+		// Prepare new data with changed roots
+		let mut published_data = BTreeMap::new();
+		let mut current_roots = Vec::new();
+
+		for i in 0..p {
+			let para_id = ParaId::from(1000 + i);
+			let entries: Vec<(Vec<u8>, Vec<u8>)> = (0..k)
+				.map(|j| (vec![j as u8; 32], vec![1u8; v as usize]))
+				.collect();
+			published_data.insert(para_id, entries);
+			current_roots.push((para_id, vec![0xAA; 32]));
+		}
+
+		#[block]
+		{
+			Pallet::<T>::process_published_data(&published_data, &current_roots);
+		}
+
+		// Verify storage updated
+		assert_eq!(PreviousPublishedDataRoots::<T>::get().len(), p as usize);
+	}
+
 	/// Re-implements an easy version of the `MessageQueueChain` for testing purposes.
 	fn mqp_head(msgs: &Vec<InboundDownwardMessage>) -> RelayHash {
 		let mut head = Default::default();