Runtime API: introduce candidates_pending_availability (#4027)

Fixes https://github.com/paritytech/polkadot-sdk/issues/3576

Required by elastic scaling collators.
Deprecates old API: `candidate_pending_availability`.

TODO:
- [x] PRDoc

---------

Signed-off-by: Andrei Sandu <andrei-mihail@parity.io>

polkadot/node/core/runtime-api/src/cache.rs

@@ -48,6 +48,7 @@ pub(crate) struct RequestResultCache {
 	validation_code: LruMap<(Hash, ParaId, OccupiedCoreAssumption), Option<ValidationCode>>,
 	validation_code_by_hash: LruMap<ValidationCodeHash, Option<ValidationCode>>,
 	candidate_pending_availability: LruMap<(Hash, ParaId), Option<CommittedCandidateReceipt>>,
+	candidates_pending_availability: LruMap<(Hash, ParaId), Vec<CommittedCandidateReceipt>>,
 	candidate_events: LruMap<Hash, Vec<CandidateEvent>>,
 	session_executor_params: LruMap<SessionIndex, Option<ExecutorParams>>,
 	session_info: LruMap<SessionIndex, SessionInfo>,
@@ -86,6 +87,7 @@ impl Default for RequestResultCache {
 			validation_code: LruMap::new(ByLength::new(DEFAULT_CACHE_CAP)),
 			validation_code_by_hash: LruMap::new(ByLength::new(DEFAULT_CACHE_CAP)),
 			candidate_pending_availability: LruMap::new(ByLength::new(DEFAULT_CACHE_CAP)),
+			candidates_pending_availability: LruMap::new(ByLength::new(DEFAULT_CACHE_CAP)),
 			candidate_events: LruMap::new(ByLength::new(DEFAULT_CACHE_CAP)),
 			session_executor_params: LruMap::new(ByLength::new(DEFAULT_CACHE_CAP)),
 			session_info: LruMap::new(ByLength::new(DEFAULT_CACHE_CAP)),
@@ -261,6 +263,21 @@ impl RequestResultCache {
 		self.candidate_pending_availability.insert(key, value);
 	}
 
+	pub(crate) fn candidates_pending_availability(
+		&mut self,
+		key: (Hash, ParaId),
+	) -> Option<&Vec<CommittedCandidateReceipt>> {
+		self.candidates_pending_availability.get(&key).map(|v| &*v)
+	}
+
+	pub(crate) fn cache_candidates_pending_availability(
+		&mut self,
+		key: (Hash, ParaId),
+		value: Vec<CommittedCandidateReceipt>,
+	) {
+		self.candidates_pending_availability.insert(key, value);
+	}
+
 	pub(crate) fn candidate_events(&mut self, relay_parent: &Hash) -> Option<&Vec<CandidateEvent>> {
 		self.candidate_events.get(relay_parent).map(|v| &*v)
 	}
@@ -591,4 +608,5 @@ pub(crate) enum RequestResult {
 	AsyncBackingParams(Hash, async_backing::AsyncBackingParams),
 	NodeFeatures(SessionIndex, NodeFeatures),
 	ClaimQueue(Hash, BTreeMap<CoreIndex, VecDeque<ParaId>>),
+	CandidatesPendingAvailability(Hash, ParaId, Vec<CommittedCandidateReceipt>),
 }

polkadot/node/core/runtime-api/src/lib.rs

@@ -133,6 +133,9 @@ where
 			CandidatePendingAvailability(relay_parent, para_id, candidate) => self
 				.requests_cache
 				.cache_candidate_pending_availability((relay_parent, para_id), candidate),
+			CandidatesPendingAvailability(relay_parent, para_id, candidates) => self
+				.requests_cache
+				.cache_candidates_pending_availability((relay_parent, para_id), candidates),
 			CandidateEvents(relay_parent, events) =>
 				self.requests_cache.cache_candidate_events(relay_parent, events),
 			SessionExecutorParams(_relay_parent, session_index, index) =>
@@ -252,6 +255,9 @@ where
 			Request::CandidatePendingAvailability(para, sender) =>
 				query!(candidate_pending_availability(para), sender)
 					.map(|sender| Request::CandidatePendingAvailability(para, sender)),
+			Request::CandidatesPendingAvailability(para, sender) =>
+				query!(candidates_pending_availability(para), sender)
+					.map(|sender| Request::CandidatesPendingAvailability(para, sender)),
 			Request::CandidateEvents(sender) =>
 				query!(candidate_events(), sender).map(|sender| Request::CandidateEvents(sender)),
 			Request::SessionExecutorParams(session_index, sender) => {
@@ -531,6 +537,12 @@ where
 			ver = 1,
 			sender
 		),
+		Request::CandidatesPendingAvailability(para, sender) => query!(
+			CandidatesPendingAvailability,
+			candidates_pending_availability(para),
+			ver = Request::CANDIDATES_PENDING_AVAILABILITY_RUNTIME_REQUIREMENT,
+			sender
+		),
 		Request::CandidateEvents(sender) => {
 			query!(CandidateEvents, candidate_events(), ver = 1, sender)
 		},

polkadot/node/core/runtime-api/src/tests.rs

@@ -47,6 +47,7 @@ struct MockSubsystemClient {
 	validation_outputs_results: HashMap<ParaId, bool>,
 	session_index_for_child: SessionIndex,
 	candidate_pending_availability: HashMap<ParaId, CommittedCandidateReceipt>,
+	candidates_pending_availability: HashMap<ParaId, Vec<CommittedCandidateReceipt>>,
 	dmq_contents: HashMap<ParaId, Vec<InboundDownwardMessage>>,
 	hrmp_channels: HashMap<ParaId, BTreeMap<ParaId, Vec<InboundHrmpMessage>>>,
 	validation_code_by_hash: HashMap<ValidationCodeHash, ValidationCode>,
@@ -140,6 +141,14 @@ impl RuntimeApiSubsystemClient for MockSubsystemClient {
 		Ok(self.candidate_pending_availability.get(&para_id).cloned())
 	}
 
+	async fn candidates_pending_availability(
+		&self,
+		_: Hash,
+		para_id: ParaId,
+	) -> Result<Vec<CommittedCandidateReceipt<Hash>>, ApiError> {
+		Ok(self.candidates_pending_availability.get(&para_id).cloned().unwrap_or_default())
+	}
+
 	async fn candidate_events(&self, _: Hash) -> Result<Vec<CandidateEvent<Hash>>, ApiError> {
 		Ok(self.candidate_events.clone())
 	}

polkadot/node/subsystem-types/src/messages.rs

@@ -670,7 +670,7 @@ pub enum RuntimeApiRequest {
 	/// Get validation code by its hash, either past, current or future code can be returned, as
 	/// long as state is still available.
 	ValidationCodeByHash(ValidationCodeHash, RuntimeApiSender<Option<ValidationCode>>),
-	/// Get a the candidate pending availability for a particular parachain by parachain / core
+	/// Get the candidate pending availability for a particular parachain by parachain / core
 	/// index
 	CandidatePendingAvailability(ParaId, RuntimeApiSender<Option<CommittedCandidateReceipt>>),
 	/// Get all events concerning candidates (backing, inclusion, time-out) in the parent of
@@ -739,6 +739,9 @@ pub enum RuntimeApiRequest {
 	/// Fetch the `ClaimQueue` from scheduler pallet
 	/// `V11`
 	ClaimQueue(RuntimeApiSender<BTreeMap<CoreIndex, VecDeque<ParaId>>>),
+	/// Get the candidates pending availability for a particular parachain
+	/// `V11`
+	CandidatesPendingAvailability(ParaId, RuntimeApiSender<Vec<CommittedCandidateReceipt>>),
 }
 
 impl RuntimeApiRequest {
@@ -776,6 +779,9 @@ impl RuntimeApiRequest {
 
 	/// `ClaimQueue`
 	pub const CLAIM_QUEUE_RUNTIME_REQUIREMENT: u32 = 11;
+
+	/// `candidates_pending_availability`
+	pub const CANDIDATES_PENDING_AVAILABILITY_RUNTIME_REQUIREMENT: u32 = 11;
 }
 
 /// A message to the Runtime API subsystem.

polkadot/node/subsystem-types/src/runtime_client.rs

@@ -333,6 +333,14 @@ pub trait RuntimeApiSubsystemClient {
 	// == v11: Claim queue ==
 	/// Fetch the `ClaimQueue` from scheduler pallet
 	async fn claim_queue(&self, at: Hash) -> Result<BTreeMap<CoreIndex, VecDeque<Id>>, ApiError>;
+
+	// == v11: Elastic scaling support ==
+	/// Get the receipts of all candidates pending availability for a `ParaId`.
+	async fn candidates_pending_availability(
+		&self,
+		at: Hash,
+		para_id: Id,
+	) -> Result<Vec<CommittedCandidateReceipt<Hash>>, ApiError>;
 }
 
 /// Default implementation of [`RuntimeApiSubsystemClient`] using the client.
@@ -428,6 +436,14 @@ where
 		self.client.runtime_api().candidate_pending_availability(at, para_id)
 	}
 
+	async fn candidates_pending_availability(
+		&self,
+		at: Hash,
+		para_id: Id,
+	) -> Result<Vec<CommittedCandidateReceipt<Hash>>, ApiError> {
+		self.client.runtime_api().candidates_pending_availability(at, para_id)
+	}
+
 	async fn candidate_events(&self, at: Hash) -> Result<Vec<CandidateEvent<Hash>>, ApiError> {
 		self.client.runtime_api().candidate_events(at)
 	}

polkadot/node/subsystem-util/src/lib.rs

@@ -296,6 +296,7 @@ specialize_requests! {
 	fn request_validation_code(para_id: ParaId, assumption: OccupiedCoreAssumption) -> Option<ValidationCode>; ValidationCode;
 	fn request_validation_code_by_hash(validation_code_hash: ValidationCodeHash) -> Option<ValidationCode>; ValidationCodeByHash;
 	fn request_candidate_pending_availability(para_id: ParaId) -> Option<CommittedCandidateReceipt>; CandidatePendingAvailability;
+	fn request_candidates_pending_availability(para_id: ParaId) -> Vec<CommittedCandidateReceipt>; CandidatesPendingAvailability;
 	fn request_candidate_events() -> Vec<CandidateEvent>; CandidateEvents;
 	fn request_session_info(index: SessionIndex) -> Option<SessionInfo>; SessionInfo;
 	fn request_validation_code_hash(para_id: ParaId, assumption: OccupiedCoreAssumption)

polkadot/primitives/src/runtime_api.rs

@@ -288,5 +288,10 @@ sp_api::decl_runtime_apis! {
 		/// Claim queue
 		#[api_version(11)]
 		fn claim_queue() -> BTreeMap<CoreIndex, VecDeque<ppp::Id>>;
+
+		/***** Added in v11 *****/
+		/// Elastic scaling support
+		#[api_version(11)]
+		fn candidates_pending_availability(para_id: ppp::Id) -> Vec<CommittedCandidateReceipt<Hash>>;
 	}
 }

polkadot/roadmap/implementers-guide/src/runtime-api/candidate-pending-availability.md

@@ -4,5 +4,8 @@ Get the receipt of a candidate pending availability. This returns `Some` for any
 `availability_cores` and `None` otherwise.
 
 ```rust
+// Deprectated.
 fn candidate_pending_availability(at: Block, ParaId) -> Option<CommittedCandidateReceipt>;
+// Use this one
+fn candidates_pending_availability(at: Block, ParaId) -> Vec<CommittedCandidateReceipt>;
 ```

polkadot/roadmap/implementers-guide/src/runtime/inclusion.md

@@ -154,6 +154,8 @@ All failed checks should lead to an unrecoverable error making the block invalid
   where the changes to the state are expected to be discarded directly after.
 * `candidate_pending_availability(ParaId) -> Option<CommittedCandidateReceipt>`: returns the `CommittedCandidateReceipt`
   pending availability for the para provided, if any.
+* `candidates_pending_availability(ParaId) -> Vec<CommittedCandidateReceipt>`: returns the `CommittedCandidateReceipt`s
+  pending availability for the para provided, if any.
 * `pending_availability(ParaId) -> Option<CandidatePendingAvailability>`: returns the metadata around the candidate
   pending availability for the para, if any.
 * `free_disputed(disputed: Vec<CandidateHash>) -> Vec<CoreIndex>`: Sweeps through all paras pending availability. If
@@ -164,10 +166,10 @@ These functions were formerly part of the UMP pallet:
 
 * `check_upward_messages(P: ParaId, Vec<UpwardMessage>)`:
     1. Checks that the parachain is not currently offboarding and error otherwise.
-    1. Checks that there are at most `config.max_upward_message_num_per_candidate` messages to be enqueued.
-    1. Checks that no message exceeds `config.max_upward_message_size`.
-    1. Checks that the total resulting queue size would not exceed `co`.
-    1. Verify that queuing up the messages could not result in exceeding the queue's footprint according to the config
+    2. Checks that there are at most `config.max_upward_message_num_per_candidate` messages to be enqueued.
+    3. Checks that no message exceeds `config.max_upward_message_size`.
+    4. Checks that the total resulting queue size would not exceed `co`.
+    5. Verify that queuing up the messages could not result in exceeding the queue's footprint according to the config
     items `config.max_upward_queue_count` and `config.max_upward_queue_size`. The queue's current footprint is provided
     in `well_known_keys` in order to facilitate oraclisation on to the para.
 

polkadot/runtime/parachains/src/inclusion/mod.rs

@@ -1104,6 +1104,24 @@ impl<T: Config> Pallet<T> {
 		})
 	}
 
+	/// Returns all the `CommittedCandidateReceipt` pending availability for the para provided, if
+	/// any.
+	pub(crate) fn candidates_pending_availability(
+		para: ParaId,
+	) -> Vec<CommittedCandidateReceipt<T::Hash>> {
+		<PendingAvailability<T>>::get(&para)
+			.map(|candidates| {
+				candidates
+					.into_iter()
+					.map(|candidate| CommittedCandidateReceipt {
+						descriptor: candidate.descriptor.clone(),
+						commitments: candidate.commitments.clone(),
+					})
+					.collect()
+			})
+			.unwrap_or_default()
+	}
+
 	/// Returns the metadata around the first candidate pending availability for the
 	/// para provided, if any.
 	pub(crate) fn pending_availability(

polkadot/runtime/parachains/src/runtime_api_impl/v10.rs

@@ -301,6 +301,10 @@ pub fn validation_code<T: initializer::Config>(
 }
 
 /// Implementation for the `candidate_pending_availability` function of the runtime API.
+#[deprecated(
+	note = "`candidate_pending_availability` will be removed. Use `candidates_pending_availability` to query 
+	all candidates pending availability"
+)]
 pub fn candidate_pending_availability<T: initializer::Config>(
 	para_id: ParaId,
 ) -> Option<CommittedCandidateReceipt<T::Hash>> {

polkadot/runtime/parachains/src/runtime_api_impl/vstaging.rs

@@ -16,8 +16,8 @@
 
 //! Put implementations of functions from staging APIs here.
 
-use crate::scheduler;
-use primitives::{CoreIndex, Id as ParaId};
+use crate::{inclusion, initializer, scheduler};
+use primitives::{CommittedCandidateReceipt, CoreIndex, Id as ParaId};
 use sp_runtime::traits::One;
 use sp_std::{
 	collections::{btree_map::BTreeMap, vec_deque::VecDeque},
@@ -41,3 +41,11 @@ pub fn claim_queue<T: scheduler::Config>() -> BTreeMap<CoreIndex, VecDeque<ParaI
 		})
 		.collect()
 }
+
+/// Returns all the candidates that are pending availability for a given `ParaId`.
+/// Deprecates `candidate_pending_availability` in favor of supporting elastic scaling.
+pub fn candidates_pending_availability<T: initializer::Config>(
+	para_id: ParaId,
+) -> Vec<CommittedCandidateReceipt<T::Hash>> {
+	<inclusion::Pallet<T>>::candidates_pending_availability(para_id)
+}

polkadot/runtime/rococo/src/lib.rs

@@ -1795,6 +1795,7 @@ sp_api::impl_runtime_apis! {
 		}
 
 		fn candidate_pending_availability(para_id: ParaId) -> Option<CommittedCandidateReceipt<Hash>> {
+			#[allow(deprecated)]
 			parachains_runtime_api_impl::candidate_pending_availability::<Runtime>(para_id)
 		}
 
@@ -1908,6 +1909,10 @@ sp_api::impl_runtime_apis! {
 		fn claim_queue() -> BTreeMap<CoreIndex, VecDeque<ParaId>> {
 			vstaging_parachains_runtime_api_impl::claim_queue::<Runtime>()
 		}
+
+		fn candidates_pending_availability(para_id: ParaId) -> Vec<CommittedCandidateReceipt<Hash>> {
+			vstaging_parachains_runtime_api_impl::candidates_pending_availability::<Runtime>(para_id)
+		}
 	}
 
 	#[api_version(3)]

polkadot/runtime/test-runtime/src/lib.rs

@@ -22,15 +22,19 @@
 
 use pallet_transaction_payment::FungibleAdapter;
 use parity_scale_codec::Encode;
-use sp_std::{collections::btree_map::BTreeMap, prelude::*};
+use sp_std::{
+	collections::{btree_map::BTreeMap, vec_deque::VecDeque},
+	prelude::*,
+};
 
 use polkadot_runtime_parachains::{
 	assigner_parachains as parachains_assigner_parachains,
 	configuration as parachains_configuration, disputes as parachains_disputes,
-	disputes::slashing as parachains_slashing, dmp as parachains_dmp, hrmp as parachains_hrmp,
-	inclusion as parachains_inclusion, initializer as parachains_initializer,
-	origin as parachains_origin, paras as parachains_paras,
-	paras_inherent as parachains_paras_inherent, runtime_api_impl::v10 as runtime_impl,
+	disputes::slashing as parachains_slashing,
+	dmp as parachains_dmp, hrmp as parachains_hrmp, inclusion as parachains_inclusion,
+	initializer as parachains_initializer, origin as parachains_origin, paras as parachains_paras,
+	paras_inherent as parachains_paras_inherent,
+	runtime_api_impl::{v10 as runtime_impl, vstaging as vstaging_parachains_runtime_api_impl},
 	scheduler as parachains_scheduler, session_info as parachains_session_info,
 	shared as parachains_shared,
 };
@@ -53,9 +57,9 @@ use pallet_transaction_payment::{FeeDetails, RuntimeDispatchInfo};
 use polkadot_runtime_parachains::reward_points::RewardValidatorsWithEraPoints;
 use primitives::{
 	slashing, AccountId, AccountIndex, Balance, BlockNumber, CandidateEvent, CandidateHash,
-	CommittedCandidateReceipt, CoreState, DisputeState, ExecutorParams, GroupRotationInfo,
-	Hash as HashT, Id as ParaId, InboundDownwardMessage, InboundHrmpMessage, Moment, Nonce,
-	OccupiedCoreAssumption, PersistedValidationData, ScrapedOnChainVotes,
+	CommittedCandidateReceipt, CoreIndex, CoreState, DisputeState, ExecutorParams,
+	GroupRotationInfo, Hash as HashT, Id as ParaId, InboundDownwardMessage, InboundHrmpMessage,
+	Moment, Nonce, OccupiedCoreAssumption, PersistedValidationData, ScrapedOnChainVotes,
 	SessionInfo as SessionInfoData, Signature, ValidationCode, ValidationCodeHash, ValidatorId,
 	ValidatorIndex, PARACHAIN_KEY_TYPE_ID,
 };
@@ -831,7 +835,7 @@ sp_api::impl_runtime_apis! {
 		}
 	}
 
-	#[api_version(10)]
+	#[api_version(11)]
 	impl primitives::runtime_api::ParachainHost<Block> for Runtime {
 		fn validators() -> Vec<ValidatorId> {
 			runtime_impl::validators::<Runtime>()
@@ -879,6 +883,7 @@ sp_api::impl_runtime_apis! {
 		}
 
 		fn candidate_pending_availability(para_id: ParaId) -> Option<CommittedCandidateReceipt<Hash>> {
+			#[allow(deprecated)]
 			runtime_impl::candidate_pending_availability::<Runtime>(para_id)
 		}
 
@@ -983,6 +988,14 @@ sp_api::impl_runtime_apis! {
 		fn node_features() -> primitives::NodeFeatures {
 			runtime_impl::node_features::<Runtime>()
 		}
+
+		fn claim_queue() -> BTreeMap<CoreIndex, VecDeque<ParaId>> {
+			vstaging_parachains_runtime_api_impl::claim_queue::<Runtime>()
+		}
+
+		fn candidates_pending_availability(para_id: ParaId) -> Vec<CommittedCandidateReceipt<Hash>> {
+			vstaging_parachains_runtime_api_impl::candidates_pending_availability::<Runtime>(para_id)
+		}
 	}
 
 	impl beefy_primitives::BeefyApi<Block, BeefyId> for Runtime {

polkadot/runtime/westend/src/lib.rs

@@ -1853,6 +1853,7 @@ sp_api::impl_runtime_apis! {
 		}
 
 		fn candidate_pending_availability(para_id: ParaId) -> Option<CommittedCandidateReceipt<Hash>> {
+			#[allow(deprecated)]
 			parachains_runtime_api_impl::candidate_pending_availability::<Runtime>(para_id)
 		}
 
@@ -1966,6 +1967,10 @@ sp_api::impl_runtime_apis! {
 		fn claim_queue() -> BTreeMap<CoreIndex, VecDeque<ParaId>> {
 			vstaging_parachains_runtime_api_impl::claim_queue::<Runtime>()
 		}
+
+		fn candidates_pending_availability(para_id: ParaId) -> Vec<CommittedCandidateReceipt<Hash>> {
+			vstaging_parachains_runtime_api_impl::candidates_pending_availability::<Runtime>(para_id)
+		}
 	}
 
 	impl beefy_primitives::BeefyApi<Block, BeefyId> for Runtime {