Add on demand assignment provider pallet

node/core/backing/src/lib.rs

@@ -48,7 +48,7 @@ use polkadot_node_subsystem_util::{
 	request_validators, Validator,
 };
 use polkadot_primitives::{
-	BackedCandidate, CandidateCommitments, CandidateHash, CandidateReceipt, CollatorId,
+	BackedCandidate, CandidateCommitments, CandidateHash, CandidateReceipt,
 	CommittedCandidateReceipt, CoreIndex, CoreState, Hash, Id as ParaId, PvfExecTimeoutKind,
 	SigningContext, ValidatorId, ValidatorIndex, ValidatorSignature, ValidityAttestation,
 };
@@ -354,7 +354,7 @@ async fn handle_active_leaves_update<Context>(
 			let group_index = group_rotation_info.group_for_core(core_index, n_cores);
 			if let Some(g) = validator_groups.get(group_index.0 as usize) {
 				if validator.as_ref().map_or(false, |v| g.contains(&v.index())) {
-					assignment = Some((scheduled.para_id, scheduled.collator));
+					assignment = Some(scheduled.para_id);
 				}
 				groups.insert(scheduled.para_id, g.clone());
 			}
@@ -363,15 +363,15 @@ async fn handle_active_leaves_update<Context>(
 
 	let table_context = TableContext { groups, validators, validator };
 
-	let (assignment, required_collator) = match assignment {
+	let assignment = match assignment {
 		None => {
 			assignments_span.add_string_tag("assigned", "false");
-			(None, None)
+			None
 		},
-		Some((assignment, required_collator)) => {
+		Some(assignment) => {
 			assignments_span.add_string_tag("assigned", "true");
 			assignments_span.add_para_id(assignment);
-			(Some(assignment), required_collator)
+			Some(assignment)
 		},
 	};
 
@@ -381,7 +381,6 @@ async fn handle_active_leaves_update<Context>(
 	let job = CandidateBackingJob {
 		parent,
 		assignment,
-		required_collator,
 		issued_statements: HashSet::new(),
 		awaiting_validation: HashSet::new(),
 		fallbacks: HashMap::new(),
@@ -412,8 +411,6 @@ struct CandidateBackingJob<Context> {
 	parent: Hash,
 	/// The `ParaId` assigned to this validator
 	assignment: Option<ParaId>,
-	/// The collator required to author the candidate, if any.
-	required_collator: Option<CollatorId>,
 	/// Spans for all candidates that are not yet backable.
 	unbacked_candidates: HashMap<CandidateHash, jaeger::Span>,
 	/// We issued `Seconded`, `Valid` or `Invalid` statements on about these candidates.
@@ -911,21 +908,6 @@ impl<Context> CandidateBackingJob<Context> {
 		candidate: &CandidateReceipt,
 		pov: Arc<PoV>,
 	) -> Result<(), Error> {
-		// Check that candidate is collated by the right collator.
-		if self
-			.required_collator
-			.as_ref()
-			.map_or(false, |c| c != &candidate.descriptor().collator)
-		{
-			// Break cycle - bounded as there is only one candidate to
-			// second per block.
-			ctx.send_unbounded_message(CollatorProtocolMessage::Invalid(
-				self.parent,
-				candidate.clone(),
-			));
-			return Ok(())
-		}
-
 		let candidate_hash = candidate.hash();
 		let mut span = self.get_unbacked_validation_child(
 			root_span,
@@ -1169,25 +1151,13 @@ impl<Context> CandidateBackingJob<Context> {
 			return Ok(())
 		}
 
-		let descriptor = attesting.candidate.descriptor().clone();
-
 		gum::debug!(
 			target: LOG_TARGET,
 			candidate_hash = ?candidate_hash,
 			candidate_receipt = ?attesting.candidate,
 			"Kicking off validation",
 		);
 
-		// Check that candidate is collated by the right collator.
-		if self.required_collator.as_ref().map_or(false, |c| c != &descriptor.collator) {
-			// If not, we've got the statement in the table but we will
-			// not issue validation work for it.
-			//
-			// Act as though we've issued a statement.
-			self.issued_statements.insert(candidate_hash);
-			return Ok(())
-		}
-
 		let bg_sender = ctx.sender().clone();
 		let pov = PoVData::FetchFromValidator {
 			from_validator: attesting.from_validator,

node/core/backing/src/tests.rs

@@ -31,8 +31,8 @@ use polkadot_node_subsystem::{
 };
 use polkadot_node_subsystem_test_helpers as test_helpers;
 use polkadot_primitives::{
-	CandidateDescriptor, CollatorId, GroupRotationInfo, HeadData, PersistedValidationData,
-	PvfExecTimeoutKind, ScheduledCore,
+	CandidateDescriptor, GroupRotationInfo, HeadData, PersistedValidationData, PvfExecTimeoutKind,
+	ScheduledCore,
 };
 use sp_application_crypto::AppCrypto;
 use sp_keyring::Sr25519Keyring;
@@ -97,14 +97,10 @@ impl Default for TestState {
 		let group_rotation_info =
 			GroupRotationInfo { session_start_block: 0, group_rotation_frequency: 100, now: 1 };
 
-		let thread_collator: CollatorId = Sr25519Keyring::Two.public().into();
 		let availability_cores = vec![
 			CoreState::Scheduled(ScheduledCore { para_id: chain_a, collator: None }),
 			CoreState::Scheduled(ScheduledCore { para_id: chain_b, collator: None }),
-			CoreState::Scheduled(ScheduledCore {
-				para_id: thread_a,
-				collator: Some(thread_collator.clone()),
-			}),
+			CoreState::Scheduled(ScheduledCore { para_id: thread_a, collator: None }),
 		];
 
 		let mut head_data = HashMap::new();
@@ -1190,66 +1186,11 @@ fn backing_works_after_failed_validation() {
 	});
 }
 
-// Test that a `CandidateBackingMessage::Second` issues validation work
-// and in case validation is successful issues a `StatementDistributionMessage`.
-#[test]
-fn backing_doesnt_second_wrong_collator() {
-	let mut test_state = TestState::default();
-	test_state.availability_cores[0] = CoreState::Scheduled(ScheduledCore {
-		para_id: ParaId::from(1),
-		collator: Some(Sr25519Keyring::Bob.public().into()),
-	});
-
-	test_harness(test_state.keystore.clone(), |mut virtual_overseer| async move {
-		test_startup(&mut virtual_overseer, &test_state).await;
-
-		let pov = PoV { block_data: BlockData(vec![42, 43, 44]) };
-
-		let expected_head_data = test_state.head_data.get(&test_state.chain_ids[0]).unwrap();
-
-		let pov_hash = pov.hash();
-		let candidate = TestCandidateBuilder {
-			para_id: test_state.chain_ids[0],
-			relay_parent: test_state.relay_parent,
-			pov_hash,
-			head_data: expected_head_data.clone(),
-			erasure_root: make_erasure_root(&test_state, pov.clone()),
-			..Default::default()
-		}
-		.build();
-
-		let second = CandidateBackingMessage::Second(
-			test_state.relay_parent,
-			candidate.to_plain(),
-			pov.clone(),
-		);
-
-		virtual_overseer.send(FromOrchestra::Communication { msg: second }).await;
-
-		assert_matches!(
-			virtual_overseer.recv().await,
-			AllMessages::CollatorProtocol(
-				CollatorProtocolMessage::Invalid(parent, c)
-			) if parent == test_state.relay_parent && c == candidate.to_plain() => {
-			}
-		);
-
-		virtual_overseer
-			.send(FromOrchestra::Signal(OverseerSignal::ActiveLeaves(
-				ActiveLeavesUpdate::stop_work(test_state.relay_parent),
-			)))
-			.await;
-		virtual_overseer
-	});
-}
-
 #[test]
 fn validation_work_ignores_wrong_collator() {
 	let mut test_state = TestState::default();
-	test_state.availability_cores[0] = CoreState::Scheduled(ScheduledCore {
-		para_id: ParaId::from(1),
-		collator: Some(Sr25519Keyring::Bob.public().into()),
-	});
+	test_state.availability_cores[0] =
+		CoreState::Scheduled(ScheduledCore { para_id: ParaId::from(1), collator: None });
 
 	test_harness(test_state.keystore.clone(), |mut virtual_overseer| async move {
 		test_startup(&mut virtual_overseer, &test_state).await;

node/network/collator-protocol/src/validator_side/mod.rs

@@ -921,6 +921,7 @@ async fn process_incoming_peer_message<Context>(
 				.span_per_relay_parent
 				.get(&relay_parent)
 				.map(|s| s.child("advertise-collation"));
+
 			if !state.view.contains(&relay_parent) {
 				gum::debug!(
 					target: LOG_TARGET,

node/service/src/chain_spec.rs

@@ -211,8 +211,7 @@ fn default_parachains_host_configuration(
 		max_pov_size: MAX_POV_SIZE,
 		max_head_data_size: 32 * 1024,
 		group_rotation_frequency: 20,
-		chain_availability_period: 4,
-		thread_availability_period: 4,
+		paras_availability_period: 4,
 		max_upward_queue_count: 8,
 		max_upward_queue_size: 1024 * 1024,
 		max_downward_message_size: 1024 * 1024,
@@ -223,10 +222,8 @@ fn default_parachains_host_configuration(
 		hrmp_channel_max_capacity: 8,
 		hrmp_channel_max_total_size: 8 * 1024,
 		hrmp_max_parachain_inbound_channels: 4,
-		hrmp_max_parathread_inbound_channels: 4,
 		hrmp_channel_max_message_size: 1024 * 1024,
 		hrmp_max_parachain_outbound_channels: 4,
-		hrmp_max_parathread_outbound_channels: 4,
 		hrmp_max_message_num_per_candidate: 5,
 		dispute_period: 6,
 		no_show_slots: 2,

node/test/service/src/chain_spec.rs

@@ -175,8 +175,7 @@ fn polkadot_testnet_genesis(
 				max_pov_size: MAX_POV_SIZE,
 				max_head_data_size: 32 * 1024,
 				group_rotation_frequency: 20,
-				chain_availability_period: 4,
-				thread_availability_period: 4,
+				paras_availability_period: 4,
 				no_show_slots: 10,
 				minimum_validation_upgrade_delay: 5,
 				..Default::default()

primitives/src/lib.rs

@@ -56,7 +56,8 @@ pub use v5::{
 	UpgradeRestriction, UpwardMessage, ValidDisputeStatementKind, ValidationCode,
 	ValidationCodeHash, ValidatorId, ValidatorIndex, ValidatorSignature, ValidityAttestation,
 	ValidityError, ASSIGNMENT_KEY_TYPE_ID, LOWEST_PUBLIC_ID, MAX_CODE_SIZE, MAX_HEAD_DATA_SIZE,
-	MAX_POV_SIZE, PARACHAINS_INHERENT_IDENTIFIER, PARACHAIN_KEY_TYPE_ID,
+	MAX_POV_SIZE, ON_DEMAND_DEFAULT_QUEUE_MAX_SIZE, PARACHAINS_INHERENT_IDENTIFIER,
+	PARACHAIN_KEY_TYPE_ID,
 };
 
 #[cfg(feature = "std")]

primitives/src/v5/mod.rs

@@ -384,6 +384,11 @@ pub const MAX_HEAD_DATA_SIZE: u32 = 1 * 1024 * 1024;
 // NOTE: This value is used in the runtime so be careful when changing it.
 pub const MAX_POV_SIZE: u32 = 5 * 1024 * 1024;
 
+/// Default queue size we use for the on-demand order book.
+///
+/// Can be adjusted in configuration.
+pub const ON_DEMAND_DEFAULT_QUEUE_MAX_SIZE: u32 = 10_000;
+
 // The public key of a keypair used by a validator for determining assignments
 /// to approve included parachain candidates.
 mod assignment_app {
@@ -812,29 +817,63 @@ pub struct ParathreadClaim(pub Id, pub Option<CollatorId>);
 pub struct ParathreadEntry {
 	/// The claim.
 	pub claim: ParathreadClaim,
-	/// Number of retries.
+	/// Number of retries
 	pub retries: u32,
 }
 
+/// An assignment for a parachain scheduled to be backed and included in a relay chain block.
+#[derive(Clone, Encode, Decode, PartialEq, TypeInfo, RuntimeDebug)]
+pub struct Assignment {
+	/// Assignment's ParaId
+	pub para_id: Id,
+}
+
+impl Assignment {
+	/// Create a new `Assignment`.
+	pub fn new(para_id: Id) -> Self {
+		Self { para_id }
+	}
+}
+
+/// An entry tracking a paras
+#[derive(Clone, Encode, Decode, TypeInfo, PartialEq, RuntimeDebug)]
+pub struct ParasEntry<N = BlockNumber> {
+	/// The `Assignment`
+	pub assignment: Assignment,
+	/// The number of times the entry has timed out in availability.
+	pub availability_timeouts: u32,
+	/// The block height where this entry becomes invalid.
+	pub ttl: N,
+}
+
+impl<N> ParasEntry<N> {
+	/// Return `Id` from the underlying `Assignment`.
+	pub fn para_id(&self) -> Id {
+		self.assignment.para_id
+	}
+
+	/// Create a new `ParasEntry`.
+	pub fn new(assignment: Assignment, now: N) -> Self {
+		ParasEntry { assignment, availability_timeouts: 0, ttl: now }
+	}
+}
+
 /// What is occupying a specific availability core.
 #[derive(Clone, Encode, Decode, TypeInfo, RuntimeDebug)]
 #[cfg_attr(feature = "std", derive(PartialEq))]
-pub enum CoreOccupied {
+pub enum CoreOccupied<N> {
 	/// The core is not occupied.
 	Free,
-	/// A parathread.
-	Parathread(ParathreadEntry),
-	/// A parachain.
-	Parachain(Id),
+	/// A paras.
+	Paras(ParasEntry<N>),
 }
 
-impl CoreOccupied {
+impl<N> CoreOccupied<N> {
 	/// Is core free?
 	pub fn is_free(&self) -> bool {
 		match self {
 			Self::Free => true,
-			Self::Parachain(_) => false,
-			Self::Parathread(_) => false,
+			Self::Paras(_) => false,
 		}
 	}
 }
@@ -968,7 +1007,9 @@ impl<H, N> OccupiedCore<H, N> {
 pub struct ScheduledCore {
 	/// The ID of a para scheduled.
 	pub para_id: Id,
-	/// The collator required to author the block, if any.
+	/// DEPRECATED: see: https://github.com/paritytech/polkadot/issues/7575
+	///
+	/// Will be removed in a future version.
 	pub collator: Option<CollatorId>,
 }
 

runtime/kusama/src/lib.rs

@@ -39,6 +39,7 @@ use scale_info::TypeInfo;
 use sp_std::{cmp::Ordering, collections::btree_map::BTreeMap, prelude::*};
 
 use 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,
@@ -1166,11 +1167,12 @@ impl parachains_paras_inherent::Config for Runtime {
 	type WeightInfo = weights::runtime_parachains_paras_inherent::WeightInfo<Runtime>;
 }
 
-impl runtime_parachains::scheduler_parachains::Config for Runtime {}
 impl parachains_scheduler::Config for Runtime {
-	type AssignmentProvider = runtime_parachains::scheduler_parachains::Pallet<Runtime>;
+	type AssignmentProvider = ParaAssignmentProvider;
 }
 
+impl parachains_assigner_parachains::Config for Runtime {}
+
 impl parachains_initializer::Config for Runtime {
 	type Randomness = pallet_babe::RandomnessFromOneEpochAgo<Runtime>;
 	type ForceOrigin = EnsureRoot<AccountId>;
@@ -1473,6 +1475,7 @@ construct_runtime! {
 		ParaSessionInfo: parachains_session_info::{Pallet, Storage} = 61,
 		ParasDisputes: parachains_disputes::{Pallet, Call, Storage, Event<T>} = 62,
 		ParasSlashing: parachains_slashing::{Pallet, Call, Storage, ValidateUnsigned} = 63,
+		ParaAssignmentProvider: parachains_assigner_parachains::{Pallet, Storage} = 64,
 
 		// Parachain Onboarding Pallets. Start indices at 70 to leave room.
 		Registrar: paras_registrar::{Pallet, Call, Storage, Event<T>} = 70,

runtime/kusama/src/weights/runtime_parachains_configuration.rs

@@ -140,4 +140,6 @@ impl<T: frame_system::Config> runtime_parachains::configuration::WeightInfo for
 			.saturating_add(T::DbWeight::get().reads(3))
 			.saturating_add(T::DbWeight::get().writes(1))
 	}
+
+	fn set_config_with_perbill() -> Weight { todo!() }
 }