Add availability distribution subsystem

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

@@ -270,7 +270,7 @@ where
 	let block_id = BlockId::Hash(hash);
 
 	if backend.blockchain().status(block_id)? == BlockStatus::InChain {
-		return Ok(BlockCheckStatus::InChain);
+		return Ok(BlockCheckStatus::InChain)
 	}
 
 	let listener = client.import_notification_stream();

client/relay-chain-minimal-node/Cargo.toml

@@ -14,12 +14,13 @@ polkadot-node-subsystem-util = { git = "https://github.com/paritytech/polkadot",
 polkadot-node-network-protocol = { git = "https://github.com/paritytech/polkadot", branch = "master" }
 polkadot-network-bridge = { git = "https://github.com/paritytech/polkadot", branch = "master" }
 polkadot-node-core-av-store = { git = "https://github.com/paritytech/polkadot", branch = "master" }
+polkadot-availability-distribution = { git = "https://github.com/paritytech/polkadot", branch = "master" }
 
 
 # substrate deps
 sc-authority-discovery = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sc-client-api = { git = "https://github.com/paritytech/substrate", branch = "master" }
-sc-client-db = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sc-keystore = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sc-network = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sc-network-sync = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sc-network-common = { git = "https://github.com/paritytech/substrate", branch = "master" }
@@ -34,7 +35,6 @@ sp-blockchain = { git = "https://github.com/paritytech/substrate", branch = "mas
 sp-consensus-babe = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sp-consensus = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sp-runtime = { git = "https://github.com/paritytech/substrate", branch = "master" }
-sp-state-machine = { git = "https://github.com/paritytech/substrate", branch = "master" }
 
 # cumulus deps
 cumulus-relay-chain-interface = { path = "../relay-chain-interface" }

client/relay-chain-minimal-node/src/blockchain_rpc_client.rs

@@ -55,6 +55,8 @@ impl BlockChainRpcClient {
 	}
 }
 
+// Implementation required by Availability-Distribution subsystem
+// but never called in our case.
 impl AuxStore for BlockChainRpcClient {
 	fn insert_aux<
 		'a,
@@ -64,14 +66,14 @@ impl AuxStore for BlockChainRpcClient {
 		D: IntoIterator<Item = &'a &'b [u8]>,
 	>(
 		&self,
-		insert: I,
-		delete: D,
+		_insert: I,
+		_delete: D,
 	) -> sp_blockchain::Result<()> {
-		todo!()
+		unimplemented!()
 	}
 
-	fn get_aux(&self, key: &[u8]) -> sp_blockchain::Result<Option<Vec<u8>>> {
-		todo!()
+	fn get_aux(&self, _key: &[u8]) -> sp_blockchain::Result<Option<Vec<u8>>> {
+		unimplemented!()
 	}
 }
 

client/relay-chain-minimal-node/src/collator_overseer.rs

@@ -15,11 +15,17 @@
 // along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
 
 use lru::LruCache;
+use polkadot_availability_distribution::{
+	AvailabilityDistributionSubsystem, IncomingRequestReceivers,
+};
 use polkadot_node_core_av_store::Config;
 use polkadot_node_network_protocol::{
 	peer_set::PeerSetProtocolNames,
 	request_response::{
-		v1::{AvailableDataFetchingRequest, CollationFetchingRequest},
+		v1::{
+			AvailableDataFetchingRequest, ChunkFetchingRequest, CollationFetchingRequest,
+			PoVFetchingRequest,
+		},
 		IncomingRequestReceiver, ReqProtocolNames,
 	},
 };
@@ -38,6 +44,7 @@ use polkadot_service::{
 	Error, OverseerConnector,
 };
 use sc_authority_discovery::Service as AuthorityDiscoveryService;
+use sc_keystore::LocalKeystore;
 use sc_network::NetworkStateInfo;
 
 use std::sync::Arc;
@@ -52,36 +59,42 @@ use sp_runtime::traits::Block as BlockT;
 
 /// Arguments passed for overseer construction.
 pub(crate) struct CollatorOverseerGenArgs<'a> {
-	/// Set of initial relay chain leaves to track.
-	pub leaves: Vec<BlockInfo>,
 	/// Runtime client generic, providing the `ProvieRuntimeApi` trait besides others.
 	pub runtime_client: Arc<BlockChainRpcClient>,
 	/// Underlying network service implementation.
 	pub network_service: Arc<sc_network::NetworkService<Block, PHash>>,
 	/// Underlying authority discovery service.
 	pub authority_discovery_service: AuthorityDiscoveryService,
+	// Receiver for collation request protocol
 	pub collation_req_receiver: IncomingRequestReceiver<CollationFetchingRequest>,
+	// Receiver for PoV request protocol
+	pub pov_req_receiver: IncomingRequestReceiver<PoVFetchingRequest>,
+	// Receiver for chunk request protocol
+	pub chunk_req_receiver: IncomingRequestReceiver<ChunkFetchingRequest>,
+	// Receiver for availability request protocol
 	pub available_data_req_receiver: IncomingRequestReceiver<AvailableDataFetchingRequest>,
 	/// Prometheus registry, commonly used for production systems, less so for test.
 	pub registry: Option<&'a Registry>,
 	/// Task spawner to be used throughout the overseer and the APIs it provides.
 	pub spawner: sc_service::SpawnTaskHandle,
 	/// Determines the behavior of the collator.
 	pub collator_pair: CollatorPair,
+	/// Request response protocols
 	pub req_protocol_names: ReqProtocolNames,
 	pub peer_set_protocol_names: PeerSetProtocolNames,
+	/// Config for the availability store
 	pub availability_config: Config,
 	/// The underlying key value store for the parachains.
 	pub parachains_db: Arc<dyn polkadot_node_subsystem_util::database::Database>,
 }
 
 pub(crate) struct CollatorOverseerGen;
+
 impl CollatorOverseerGen {
 	pub(crate) fn generate<'a>(
 		&self,
 		connector: OverseerConnector,
 		CollatorOverseerGenArgs {
-			leaves,
 			runtime_client,
 			network_service,
 			authority_discovery_service,
@@ -94,6 +107,8 @@ impl CollatorOverseerGen {
 			req_protocol_names,
 			peer_set_protocol_names,
 			parachains_db,
+			pov_req_receiver,
+			chunk_req_receiver,
 		}: CollatorOverseerGenArgs<'a>,
 	) -> Result<
 		(
@@ -102,12 +117,17 @@ impl CollatorOverseerGen {
 		),
 		Error,
 	> {
+		let leaves = Vec::new();
 		let metrics = <OverseerMetrics as MetricsTrait>::register(registry)?;
-
+		let keystore = Arc::new(LocalKeystore::in_memory());
 		let spawner = SpawnGlue(spawner);
 		let network_bridge_metrics: NetworkBridgeMetrics = Metrics::register(registry)?;
 		let builder = Overseer::builder()
-			.availability_distribution(DummySubsystem)
+			.availability_distribution(AvailabilityDistributionSubsystem::new(
+				keystore.clone(),
+				IncomingRequestReceivers { pov_req_receiver, chunk_req_receiver },
+				Metrics::register(registry)?,
+			))
 			.availability_recovery(AvailabilityRecoverySubsystem::with_chunks_only(
 				available_data_req_receiver,
 				Metrics::register(registry)?,
@@ -183,6 +203,7 @@ pub(crate) fn spawn_overseer(
 	relay_chain_rpc_client: Arc<BlockChainRpcClient>,
 ) -> Result<polkadot_overseer::Handle, polkadot_service::Error> {
 	let overseer_gen = CollatorOverseerGen;
+
 	let (overseer, overseer_handle) = overseer_gen
 		.generate(OverseerConnector::default(), overseer_args)
 		.map_err(|e| {

client/relay-chain-minimal-node/src/lib.rs

@@ -18,16 +18,14 @@ use collator_overseer::{CollatorOverseerGenArgs, NewMinimalNode};
 
 use cumulus_relay_chain_interface::RelayChainError;
 use polkadot_network_bridge::{peer_sets_info, IsAuthority};
-use polkadot_node_core_av_store::Config as AvailabilityConfig;
 use polkadot_node_network_protocol::{
 	peer_set::PeerSetProtocolNames,
-	request_response::{IncomingRequest, ReqProtocolNames},
+	request_response::{self, IncomingRequest, ReqProtocolNames},
 };
 use polkadot_node_subsystem_util::metrics::prometheus::Registry;
 use polkadot_primitives::v2::CollatorPair;
 
 use sc_authority_discovery::Service as AuthorityDiscoveryService;
-use sc_client_db::DatabaseSource;
 use sc_network::{Event, NetworkService};
 use sc_network_common::service::NetworkEventStream;
 use std::sync::Arc;
@@ -120,7 +118,6 @@ pub async fn new_minimal_relay_chain(
 
 	let peer_set_protocol_names =
 		PeerSetProtocolNames::new(genesis_hash, config.chain_spec.fork_id());
-
 	{
 		let is_authority = if role.is_authority() { IsAuthority::Yes } else { IsAuthority::No };
 		config
@@ -130,13 +127,8 @@ pub async fn new_minimal_relay_chain(
 	}
 
 	let request_protocol_names = ReqProtocolNames::new(genesis_hash, config.chain_spec.fork_id());
-	let (collation_req_receiver, cfg) =
-		IncomingRequest::get_config_receiver(&request_protocol_names);
-	config.network.request_response_protocols.push(cfg);
-	let (available_data_req_receiver, cfg) =
-		IncomingRequest::get_config_receiver(&request_protocol_names);
-	config.network.request_response_protocols.push(cfg);
-
+	let (collation_req_receiver, available_data_req_receiver, pov_req_receiver, chunk_req_receiver) =
+		build_request_response_protocol_receivers(&request_protocol_names, &mut config);
 	let (network, network_starter) =
 		network::build_collator_network(network::BuildCollatorNetworkParams {
 			config: &config,
@@ -145,8 +137,6 @@ pub async fn new_minimal_relay_chain(
 			genesis_hash,
 		})?;
 
-	let active_leaves = Vec::new();
-
 	let authority_discovery_service = build_authority_discovery_service(
 		&task_manager,
 		relay_chain_rpc_client.clone(),
@@ -156,9 +146,8 @@ pub async fn new_minimal_relay_chain(
 	);
 
 	let parachains_db = open_database(&config.database)?;
-	let availability_config = AvailabilityConfig { col_data: 0, col_meta: 1 };
+
 	let overseer_args = CollatorOverseerGenArgs {
-		leaves: active_leaves,
 		runtime_client: relay_chain_rpc_client.clone(),
 		network_service: network.clone(),
 		authority_discovery_service,
@@ -170,7 +159,9 @@ pub async fn new_minimal_relay_chain(
 		req_protocol_names: request_protocol_names,
 		peer_set_protocol_names,
 		parachains_db,
-		availability_config,
+		availability_config: polkadot_service::create_availability_config(),
+		pov_req_receiver,
+		chunk_req_receiver,
 	};
 
 	let overseer_handle = collator_overseer::spawn_overseer(
@@ -183,3 +174,25 @@ pub async fn new_minimal_relay_chain(
 
 	Ok(NewMinimalNode { task_manager, overseer_handle, network })
 }
+
+fn build_request_response_protocol_receivers(
+	request_protocol_names: &ReqProtocolNames,
+	config: &mut Configuration,
+) -> (
+	request_response::IncomingRequestReceiver<request_response::v1::CollationFetchingRequest>,
+	request_response::IncomingRequestReceiver<request_response::v1::AvailableDataFetchingRequest>,
+	request_response::IncomingRequestReceiver<request_response::v1::PoVFetchingRequest>,
+	request_response::IncomingRequestReceiver<request_response::v1::ChunkFetchingRequest>,
+) {
+	let (collation_req_receiver, cfg) =
+		IncomingRequest::get_config_receiver(request_protocol_names);
+	config.network.request_response_protocols.push(cfg);
+	let (available_data_req_receiver, cfg) =
+		IncomingRequest::get_config_receiver(request_protocol_names);
+	config.network.request_response_protocols.push(cfg);
+	let (pov_req_receiver, cfg) = IncomingRequest::get_config_receiver(request_protocol_names);
+	config.network.request_response_protocols.push(cfg);
+	let (chunk_req_receiver, cfg) = IncomingRequest::get_config_receiver(request_protocol_names);
+	config.network.request_response_protocols.push(cfg);
+	(collation_req_receiver, available_data_req_receiver, pov_req_receiver, chunk_req_receiver)
+}

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

@@ -171,7 +171,7 @@ impl RelayChainInterface for RelayChainRpcInterface {
 		let mut head_stream = self.rpc_client.get_imported_heads_stream().await?;
 
 		if self.rpc_client.chain_get_header(Some(wait_for_hash)).await?.is_some() {
-			return Ok(());
+			return Ok(())
 		}
 
 		let mut timeout = futures_timer::Delay::new(Duration::from_secs(TIMEOUT_IN_SECONDS)).fuse();

test/service/src/lib.rs

@@ -194,7 +194,7 @@ async fn build_relay_chain_interface(
 		.await?;
 		task_manager.add_child(collator_node.task_manager);
 		tracing::info!("Using relay chain minimal node.");
-		return Ok(Arc::new(RelayChainRpcInterface::new(client, collator_node.overseer_handle)));
+		return Ok(Arc::new(RelayChainRpcInterface::new(client, collator_node.overseer_handle)))
 	}
 
 	let relay_chain_full_node = polkadot_test_service::new_full(