Asynchronous backing PR (#2300)

* Update substrate & polkadot * min changes to make async backing compile * (async backing) parachain-system: track limitations for unincluded blocks (#2438) * unincluded segment draft * read para head from storage proof * read_para_head -> read_included_para_head * Provide pub interface * add errors * fix unincluded segment update * BlockTracker -> Ancestor * add a dmp limit * Read para head depending on the storage switch * doc comments * storage items docs * add a sanity check on block initialize * Check watermark * append to the segment on block finalize * Move segment update into set_validation_data * Resolve para head todo * option watermark * fix comment * Drop dmq check * fix weight * doc-comments on inherent invariant * Remove TODO * add todo * primitives tests * pallet tests * doc comments * refactor unincluded segment length into a ConsensusHook (#2501) * refactor unincluded segment length into a ConsensusHook * add docs * refactor bandwidth_out calculation Co-authored-by: Chris Sosnin <48099298+slumber@users.noreply.github.com> * test for limits from impl * fmt * make tests compile * update comment * uncomment test * fix collator test by adding parent to state proof * patch HRMP watermark rules for unincluded segment * get consensus-common tests to pass, using unincluded segment * fix unincluded segment tests * get all tests passing * fmt * rustdoc CI * aura-ext: limit the number of authored blocks per slot (#2551) * aura_ext consensus hook * reverse dependency * include weight into hook * fix tests * remove stray println Co-authored-by: Chris Sosnin <48099298+slumber@users.noreply.github.com> * fix test warning * fix doc link --------- Co-authored-by: Chris Sosnin <48099298+slumber@users.noreply.github.com> Co-authored-by: Chris Sosnin <chris125_@live.com> * parachain-system: ignore go ahead signal once upgrade is processed (#2594) * handle goahead signal for unincluded segment * doc comment * add test * parachain-system: drop processed messages from inherent data (#2590) * implement `drop_processed_messages` * drop messages based on relay parent number * adjust tests * drop changes to mqc * fix comment * drop test * drop more dead code * clippy * aura-ext: check slot in consensus hook and remove all `CheckInherents` logic (#2658) * aura-ext: check slot in consensus hook * convert relay chain slot * Make relay chain slot duration generic * use fixed velocity hook for pallets with aura * purge timestamp inherent * fix warning * adjust runtime tests * fix slots in tests * Make `xcm-emulator` test pass for new consensus hook (#2722) * add pallets on_initialize * tests pass * add AuraExt on_init * ".git/.scripts/commands/fmt/fmt.sh" --------- Co-authored-by: command-bot <> --------- Co-authored-by: Ignacio Palacios <ignacio.palacios.santos@gmail.com> * update polkadot git refs * CollationGenerationConfig closure is now optional (#2772) * CollationGenerationConfig closure is now optional * fix test * propagate network-protocol-staging feature (#2899) * Feature Flagging Consensus Hook Type Parameter (#2911) * First pass * fmt * Added as default feature in tomls * Changed to direct dependency feature * Dealing with clippy error * Update pallets/parachain-system/src/lib.rs Co-authored-by: asynchronous rob <rphmeier@gmail.com> --------- Co-authored-by: asynchronous rob <rphmeier@gmail.com> * fmt * bump deps and remove warning * parachain-system: update RelevantMessagingState according to the unincluded segment (#2948) * mostly address 2471 with a bug introduced * adjust relevant messaging state after computing total * fmt * max -> min * fix test implementation of xcmp source * add test * fix test message sending logic * fix + test * add more to unincluded segment test * fmt --------- Co-authored-by: Chris Sosnin <chris125_@live.com> * Integrate new Aura / Parachain Consensus Logic in Parachain-Template / Polkadot-Parachain (#2864) * add a comment * refactor client/service utilities * deprecate start_collator * update parachain-template * update test-service in the same way * update polkadot-parachain crate * fmt * wire up new SubmitCollation message * some runtime utilities for implementing unincluded segment runtime APIs * allow parachains to configure their level of sybil-resistance when starting the network * make aura-ext compile * update to specify sybil resistance levels * fmt * specify relay chain slot duration in milliseconds * update Aura to explicitly produce Send futures also, make relay_chain_slot_duration a Duration * add authoring duration to basic collator and document params * integrate new basic collator into parachain-template * remove assert_send used for testing * basic-aura: only author when parent included * update polkadot-parachain-bin * fmt * some fixes * fixes * add a RelayNumberMonotonicallyIncreases * add a utility function for initializing subsystems * some logging for timestamp adjustment * fmt * some fixes for lookahead collator * add a log * update `find_potential_parents` to account for sessions * bound the loop * restore & deprecate old start_collator and start_full_node functions. * remove unnecessary await calls * fix warning * clippy * more clippy * remove unneeded logic * ci * update comment Co-authored-by: Marcin S. <marcin@bytedude.com> * (async backing) restore `CheckInherents` for backwards-compatibility (#2977) * bring back timestamp * Restore CheckInherents * revert to empty CheckInherents * make CheckInherents optional * attempt * properly end system blocks * add some more comments * ignore failing system parachain tests * update refs after main feature branch merge * comment out the offending tests because CI runs ignored tests * fix warnings * fmt * revert to polkadot master * cargo update -p polkadot-primitives -p sp-io --------- Co-authored-by: asynchronous rob <rphmeier@gmail.com> Co-authored-by: Ignacio Palacios <ignacio.palacios.santos@gmail.com> Co-authored-by: Bradley Olson <34992650+BradleyOlson64@users.noreply.github.com> Co-authored-by: Marcin S. <marcin@bytedude.com> Co-authored-by: eskimor <eskimor@users.noreply.github.com> Co-authored-by: Andronik <write@reusable.software>
2026-05-30 05:51:02 +00:00 · 2023-08-18 19:57:34 +03:00
parent 4f699c70a4
commit 6ef1117685
105 changed files with 4707 additions and 2359 deletions
@@ -75,16 +75,18 @@ substrate-state-trie-migration-rpc = { git = "https://github.com/paritytech/subs

 # Polkadot
 # Use rococo-native as this is currently the default "local" relay chain
-polkadot-cli = { git = "https://github.com/paritytech/polkadot", branch = "master", features = ["rococo-native"] }
+polkadot-cli = { git = "https://github.com/paritytech/polkadot", features = ["rococo-native"] , branch = "master" }
 polkadot-primitives = { git = "https://github.com/paritytech/polkadot", branch = "master" }
 polkadot-service = { git = "https://github.com/paritytech/polkadot", branch = "master" }
 xcm = { git = "https://github.com/paritytech/polkadot", branch = "master" }

 # Cumulus
 cumulus-client-cli = { path = "../client/cli" }
+cumulus-client-collator = { path = "../client/collator" }
 cumulus-client-consensus-aura = { path = "../client/consensus/aura" }
 cumulus-client-consensus-relay-chain = { path = "../client/consensus/relay-chain" }
 cumulus-client-consensus-common = { path = "../client/consensus/common" }
+cumulus-client-consensus-proposer = { path = "../client/consensus/proposer" }
 cumulus-client-service = { path = "../client/service" }
 cumulus-primitives-core = { path = "../primitives/core" }
 cumulus-primitives-parachain-inherent = { path = "../primitives/parachain-inherent" }
@@ -16,19 +16,25 @@

 use codec::Codec;
 use cumulus_client_cli::CollatorOptions;
-use cumulus_client_consensus_aura::{AuraConsensus, BuildAuraConsensusParams, SlotProportion};
+use cumulus_client_collator::service::CollatorService;
+use cumulus_client_consensus_aura::collators::basic::{
+	self as basic_aura, Params as BasicAuraParams,
+};
 use cumulus_client_consensus_common::{
 	ParachainBlockImport as TParachainBlockImport, ParachainCandidate, ParachainConsensus,
 };
+use cumulus_client_consensus_proposer::Proposer;
+#[allow(deprecated)]
+use cumulus_client_service::old_consensus;
 use cumulus_client_service::{
-	build_network, build_relay_chain_interface, prepare_node_config, start_collator,
-	start_full_node, BuildNetworkParams, StartCollatorParams, StartFullNodeParams,
+	build_network, build_relay_chain_interface, prepare_node_config, start_relay_chain_tasks,
+	BuildNetworkParams, CollatorSybilResistance, DARecoveryProfile, StartRelayChainTasksParams,
 };
 use cumulus_primitives_core::{
 	relay_chain::{Hash as PHash, PersistedValidationData},
 	ParaId,
 };
-use cumulus_relay_chain_interface::RelayChainInterface;
+use cumulus_relay_chain_interface::{OverseerHandle, RelayChainInterface};
 use sp_core::Pair;

 use jsonrpsee::RpcModule;
@@ -54,6 +60,8 @@ use sp_runtime::{app_crypto::AppCrypto, traits::Header as HeaderT};
 use std::{marker::PhantomData, sync::Arc, time::Duration};
 use substrate_prometheus_endpoint::Registry;

+use polkadot_primitives::CollatorPair;
+
 #[cfg(not(feature = "runtime-benchmarks"))]
 type HostFunctions = sp_io::SubstrateHostFunctions;

@@ -325,14 +333,15 @@ where
 /// This is the actual implementation that is abstract over the executor and the runtime api for
 /// shell nodes.
 #[sc_tracing::logging::prefix_logs_with("Parachain")]
-async fn start_shell_node_impl<RuntimeApi, RB, BIQ, BIC>(
+async fn start_shell_node_impl<RuntimeApi, RB, BIQ, SC>(
 	parachain_config: Configuration,
 	polkadot_config: Configuration,
 	collator_options: CollatorOptions,
+	sybil_resistance_level: CollatorSybilResistance,
 	para_id: ParaId,
 	rpc_ext_builder: RB,
 	build_import_queue: BIQ,
-	build_consensus: BIC,
+	start_consensus: SC,
 	hwbench: Option<sc_sysinfo::HwBench>,
 ) -> sc_service::error::Result<(TaskManager, Arc<ParachainClient<RuntimeApi>>)>
 where
@@ -353,7 +362,7 @@ where
 		Option<TelemetryHandle>,
 		&TaskManager,
 	) -> Result<sc_consensus::DefaultImportQueue<Block>, sc_service::Error>,
-	BIC: FnOnce(
+	SC: FnOnce(
 		Arc<ParachainClient<RuntimeApi>>,
 		ParachainBlockImport<RuntimeApi>,
 		Option<&Registry>,
@@ -363,8 +372,12 @@ where
 		Arc<sc_transaction_pool::FullPool<Block, ParachainClient<RuntimeApi>>>,
 		Arc<SyncingService<Block>>,
 		KeystorePtr,
-		bool,
-	) -> Result<Box<dyn ParachainConsensus<Block>>, sc_service::Error>,
+		Duration,
+		ParaId,
+		CollatorPair,
+		OverseerHandle,
+		Arc<dyn Fn(Hash, Option<Vec<u8>>) + Send + Sync>,
+	) -> Result<(), sc_service::Error>,
 {
 	let parachain_config = prepare_node_config(parachain_config);

@@ -387,7 +400,6 @@ where
 	.await
 	.map_err(|e| sc_service::Error::Application(Box::new(e) as Box<_>))?;

-	let force_authoring = parachain_config.force_authoring;
 	let validator = parachain_config.role.is_authority();
 	let prometheus_registry = parachain_config.prometheus_registry().cloned();
 	let transaction_pool = params.transaction_pool.clone();
@@ -404,6 +416,7 @@ where
 			spawn_handle: task_manager.spawn_handle(),
 			relay_chain_interface: relay_chain_interface.clone(),
 			import_queue: params.import_queue,
+			sybil_resistance_level,
 		})
 		.await?;

@@ -452,8 +465,25 @@ where
 		.overseer_handle()
 		.map_err(|e| sc_service::Error::Application(Box::new(e)))?;

+	start_relay_chain_tasks(StartRelayChainTasksParams {
+		client: client.clone(),
+		announce_block: announce_block.clone(),
+		para_id,
+		relay_chain_interface: relay_chain_interface.clone(),
+		task_manager: &mut task_manager,
+		da_recovery_profile: if validator {
+			DARecoveryProfile::Collator
+		} else {
+			DARecoveryProfile::FullNode
+		},
+		import_queue: import_queue_service,
+		relay_chain_slot_duration,
+		recovery_handle: Box::new(overseer_handle.clone()),
+		sync_service: sync_service.clone(),
+	})?;
+
 	if validator {
-		let parachain_consensus = build_consensus(
+		start_consensus(
 			client.clone(),
 			block_import,
 			prometheus_registry.as_ref(),
@@ -463,42 +493,12 @@ where
 			transaction_pool,
 			sync_service.clone(),
 			params.keystore_container.keystore(),
-			force_authoring,
+			relay_chain_slot_duration,
+			para_id,
+			collator_key.expect("Command line arguments do not allow this. qed"),
+			overseer_handle,
+			announce_block,
 		)?;
-
-		let spawner = task_manager.spawn_handle();
-
-		let params = StartCollatorParams {
-			para_id,
-			block_status: client.clone(),
-			announce_block,
-			client: client.clone(),
-			task_manager: &mut task_manager,
-			relay_chain_interface,
-			spawner,
-			parachain_consensus,
-			import_queue: import_queue_service,
-			collator_key: collator_key.expect("Command line arguments do not allow this. qed"),
-			relay_chain_slot_duration,
-			recovery_handle: Box::new(overseer_handle),
-			sync_service,
-		};
-
-		start_collator(params).await?;
-	} else {
-		let params = StartFullNodeParams {
-			client: client.clone(),
-			announce_block,
-			task_manager: &mut task_manager,
-			para_id,
-			relay_chain_interface,
-			relay_chain_slot_duration,
-			import_queue: import_queue_service,
-			recovery_handle: Box::new(overseer_handle),
-			sync_service,
-		};
-
-		start_full_node(params)?;
 	}

 	start_network.start_network();
@@ -510,14 +510,15 @@ where
 ///
 /// This is the actual implementation that is abstract over the executor and the runtime api.
 #[sc_tracing::logging::prefix_logs_with("Parachain")]
-async fn start_node_impl<RuntimeApi, RB, BIQ, BIC>(
+async fn start_node_impl<RuntimeApi, RB, BIQ, SC>(
 	parachain_config: Configuration,
 	polkadot_config: Configuration,
 	collator_options: CollatorOptions,
+	sybil_resistance_level: CollatorSybilResistance,
 	para_id: ParaId,
 	_rpc_ext_builder: RB,
 	build_import_queue: BIQ,
-	build_consensus: BIC,
+	start_consensus: SC,
 	hwbench: Option<sc_sysinfo::HwBench>,
 ) -> sc_service::error::Result<(TaskManager, Arc<ParachainClient<RuntimeApi>>)>
 where
@@ -539,7 +540,7 @@ where
 		Option<TelemetryHandle>,
 		&TaskManager,
 	) -> Result<sc_consensus::DefaultImportQueue<Block>, sc_service::Error>,
-	BIC: FnOnce(
+	SC: FnOnce(
 		Arc<ParachainClient<RuntimeApi>>,
 		ParachainBlockImport<RuntimeApi>,
 		Option<&Registry>,
@@ -549,8 +550,12 @@ where
 		Arc<sc_transaction_pool::FullPool<Block, ParachainClient<RuntimeApi>>>,
 		Arc<SyncingService<Block>>,
 		KeystorePtr,
-		bool,
-	) -> Result<Box<dyn ParachainConsensus<Block>>, sc_service::Error>,
+		Duration,
+		ParaId,
+		CollatorPair,
+		OverseerHandle,
+		Arc<dyn Fn(Hash, Option<Vec<u8>>) + Send + Sync>,
+	) -> Result<(), sc_service::Error>,
 {
 	let parachain_config = prepare_node_config(parachain_config);

@@ -572,7 +577,6 @@ where
 	.await
 	.map_err(|e| sc_service::Error::Application(Box::new(e) as Box<_>))?;

-	let force_authoring = parachain_config.force_authoring;
 	let validator = parachain_config.role.is_authority();
 	let prometheus_registry = parachain_config.prometheus_registry().cloned();
 	let transaction_pool = params.transaction_pool.clone();
@@ -589,6 +593,7 @@ where
 			spawn_handle: task_manager.spawn_handle(),
 			relay_chain_interface: relay_chain_interface.clone(),
 			import_queue: params.import_queue,
+			sybil_resistance_level,
 		})
 		.await?;

@@ -649,8 +654,26 @@ where
 	let overseer_handle = relay_chain_interface
 		.overseer_handle()
 		.map_err(|e| sc_service::Error::Application(Box::new(e)))?;
+
+	start_relay_chain_tasks(StartRelayChainTasksParams {
+		client: client.clone(),
+		announce_block: announce_block.clone(),
+		para_id,
+		relay_chain_interface: relay_chain_interface.clone(),
+		task_manager: &mut task_manager,
+		da_recovery_profile: if validator {
+			DARecoveryProfile::Collator
+		} else {
+			DARecoveryProfile::FullNode
+		},
+		import_queue: import_queue_service,
+		relay_chain_slot_duration,
+		recovery_handle: Box::new(overseer_handle.clone()),
+		sync_service: sync_service.clone(),
+	})?;
+
 	if validator {
-		let parachain_consensus = build_consensus(
+		start_consensus(
 			client.clone(),
 			block_import,
 			prometheus_registry.as_ref(),
@@ -660,42 +683,12 @@ where
 			transaction_pool,
 			sync_service.clone(),
 			params.keystore_container.keystore(),
-			force_authoring,
+			relay_chain_slot_duration,
+			para_id,
+			collator_key.expect("Command line arguments do not allow this. qed"),
+			overseer_handle,
+			announce_block,
 		)?;
-
-		let spawner = task_manager.spawn_handle();
-
-		let params = StartCollatorParams {
-			para_id,
-			block_status: client.clone(),
-			announce_block,
-			client: client.clone(),
-			task_manager: &mut task_manager,
-			relay_chain_interface: relay_chain_interface.clone(),
-			spawner,
-			parachain_consensus,
-			import_queue: import_queue_service,
-			collator_key: collator_key.expect("Command line arguments do not allow this. qed"),
-			relay_chain_slot_duration,
-			recovery_handle: Box::new(overseer_handle),
-			sync_service,
-		};
-
-		start_collator(params).await?;
-	} else {
-		let params = StartFullNodeParams {
-			client: client.clone(),
-			announce_block,
-			task_manager: &mut task_manager,
-			para_id,
-			relay_chain_interface,
-			relay_chain_slot_duration,
-			import_queue: import_queue_service,
-			recovery_handle: Box::new(overseer_handle),
-			sync_service,
-		};
-
-		start_full_node(params)?;
 	}

 	start_network.start_network();
@@ -756,6 +749,7 @@ pub async fn start_rococo_parachain_node(
 		parachain_config,
 		polkadot_config,
 		collator_options,
+		CollatorSybilResistance::Resistant, // Aura
 		para_id,
 		|_| Ok(RpcModule::new(())),
 		rococo_parachain_build_import_queue,
@@ -768,7 +762,11 @@ pub async fn start_rococo_parachain_node(
 		 transaction_pool,
 		 sync_oracle,
 		 keystore,
-		 force_authoring| {
+		 relay_chain_slot_duration,
+		 para_id,
+		 collator_key,
+		 overseer_handle,
+		 announce_block| {
 			let slot_duration = cumulus_client_consensus_aura::slot_duration(&*client)?;

 			let proposer_factory = sc_basic_authorship::ProposerFactory::with_proof_recording(
@@ -778,53 +776,47 @@ pub async fn start_rococo_parachain_node(
 				prometheus_registry,
 				telemetry.clone(),
 			);
+			let proposer = Proposer::new(proposer_factory);

-			Ok(AuraConsensus::build::<sp_consensus_aura::sr25519::AuthorityPair, _, _, _, _, _, _>(
-				BuildAuraConsensusParams {
-					proposer_factory,
-					create_inherent_data_providers: move |_, (relay_parent, validation_data)| {
-						let relay_chain_interface = relay_chain_interface.clone();
+			let collator_service = CollatorService::new(
+				client.clone(),
+				Arc::new(task_manager.spawn_handle()),
+				announce_block,
+				client.clone(),
+			);

-						async move {
-							let parachain_inherent =
-							cumulus_primitives_parachain_inherent::ParachainInherentData::create_at(
-								relay_parent,
-								&relay_chain_interface,
-								&validation_data,
-								para_id,
-							).await;
+			let params = BasicAuraParams {
+				create_inherent_data_providers: move |_, ()| async move { Ok(()) },
+				block_import,
+				para_client: client,
+				relay_client: relay_chain_interface,
+				sync_oracle,
+				keystore,
+				collator_key,
+				para_id,
+				overseer_handle,
+				slot_duration,
+				relay_chain_slot_duration,
+				proposer,
+				collator_service,
+				// Very limited proposal time.
+				authoring_duration: Duration::from_millis(500),
+			};

-							let timestamp = sp_timestamp::InherentDataProvider::from_system_time();
+			let fut = basic_aura::run::<
+				Block,
+				sp_consensus_aura::sr25519::AuthorityPair,
+				_,
+				_,
+				_,
+				_,
+				_,
+				_,
+				_,
+			>(params);
+			task_manager.spawn_essential_handle().spawn("aura", None, fut);

-							let slot =
-							sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_slot_duration(
-								*timestamp,
-								slot_duration,
-							);
-
-							let parachain_inherent = parachain_inherent.ok_or_else(|| {
-								Box::<dyn std::error::Error + Send + Sync>::from(
-									"Failed to create parachain inherent",
-								)
-							})?;
-
-							Ok((slot, timestamp, parachain_inherent))
-						}
-					},
-					block_import,
-					para_client: client,
-					backoff_authoring_blocks: Option::<()>::None,
-					sync_oracle,
-					keystore,
-					force_authoring,
-					slot_duration,
-					// We got around 500ms for proposing
-					block_proposal_slot_portion: SlotProportion::new(1f32 / 24f32),
-					// And a maximum of 750ms if slots are skipped
-					max_block_proposal_slot_portion: Some(SlotProportion::new(1f32 / 16f32)),
-					telemetry,
-				},
-			))
+			Ok(())
 		},
 		hwbench,
 	)
@@ -880,6 +872,7 @@ where
 		parachain_config,
 		polkadot_config,
 		collator_options,
+		CollatorSybilResistance::Unresistant, // free-for-all consensus
 		para_id,
 		|_| Ok(RpcModule::new(())),
 		shell_build_import_queue,
@@ -890,18 +883,22 @@ where
 		 task_manager,
 		 relay_chain_interface,
 		 transaction_pool,
-		 _,
-		 _,
-		 _| {
+		 _sync_oracle,
+		 _keystore,
+		 _relay_chain_slot_duration,
+		 para_id,
+		 collator_key,
+		 overseer_handle,
+		 announce_block| {
 			let proposer_factory = sc_basic_authorship::ProposerFactory::with_proof_recording(
 				task_manager.spawn_handle(),
-				client,
+				client.clone(),
 				transaction_pool,
 				prometheus_registry,
 				telemetry,
 			);

-			Ok(cumulus_client_consensus_relay_chain::build_relay_chain_consensus(
+			let free_for_all = cumulus_client_consensus_relay_chain::build_relay_chain_consensus(
 				cumulus_client_consensus_relay_chain::BuildRelayChainConsensusParams {
 					para_id,
 					proposer_factory,
@@ -926,7 +923,24 @@ where
 						}
 					},
 				},
-			))
+			);
+
+			let spawner = task_manager.spawn_handle();
+
+			// Required for free-for-all consensus
+			#[allow(deprecated)]
+			old_consensus::start_collator_sync(old_consensus::StartCollatorParams {
+				para_id,
+				block_status: client.clone(),
+				announce_block,
+				overseer_handle,
+				spawner,
+				key: collator_key,
+				parachain_consensus: free_for_all,
+				runtime_api: client.clone(),
+			});
+
+			Ok(())
 		},
 		hwbench,
 	)
@@ -1127,6 +1141,7 @@ where
 		parachain_config,
 		polkadot_config,
 		collator_options,
+		CollatorSybilResistance::Resistant, // Aura
 		para_id,
 		|_| Ok(RpcModule::new(())),
 		aura_build_import_queue::<_, AuraId>,
@@ -1139,123 +1154,52 @@ where
 		 transaction_pool,
 		 sync_oracle,
 		 keystore,
-		 force_authoring| {
-			let spawn_handle = task_manager.spawn_handle();
-			let client2 = client.clone();
-			let block_import2 = block_import.clone();
-			let transaction_pool2 = transaction_pool.clone();
-			let telemetry2 = telemetry.clone();
-			let prometheus_registry2 = prometheus_registry.map(|r| (*r).clone());
-			let relay_chain_for_aura = relay_chain_interface.clone();
-
-			let aura_consensus = BuildOnAccess::Uninitialized(Some(Box::new(move || {
-				let slot_duration =
-					cumulus_client_consensus_aura::slot_duration(&*client2).unwrap();
-
-				let proposer_factory = sc_basic_authorship::ProposerFactory::with_proof_recording(
-					spawn_handle,
-					client2.clone(),
-					transaction_pool2,
-					prometheus_registry2.as_ref(),
-					telemetry2.clone(),
-				);
-
-				AuraConsensus::build::<<AuraId as AppCrypto>::Pair, _, _, _, _, _, _>(
-					BuildAuraConsensusParams {
-						proposer_factory,
-						create_inherent_data_providers:
-							move |_, (relay_parent, validation_data)| {
-								let relay_chain_for_aura = relay_chain_for_aura.clone();
-								async move {
-									let parachain_inherent =
-										cumulus_primitives_parachain_inherent::ParachainInherentData::create_at(
-											relay_parent,
-											&relay_chain_for_aura,
-											&validation_data,
-											para_id,
-										).await;
-
-									let timestamp =
-										sp_timestamp::InherentDataProvider::from_system_time();
-
-									let slot =
-										sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_slot_duration(
-											*timestamp,
-											slot_duration,
-										);
-
-									let parachain_inherent =
-										parachain_inherent.ok_or_else(|| {
-											Box::<dyn std::error::Error + Send + Sync>::from(
-												"Failed to create parachain inherent",
-											)
-										})?;
-
-									Ok((slot, timestamp, parachain_inherent))
-								}
-							},
-						block_import: block_import2,
-						para_client: client2,
-						backoff_authoring_blocks: Option::<()>::None,
-						sync_oracle,
-						keystore,
-						force_authoring,
-						slot_duration,
-						// We got around 500ms for proposing
-						block_proposal_slot_portion: SlotProportion::new(1f32 / 24f32),
-						// And a maximum of 750ms if slots are skipped
-						max_block_proposal_slot_portion: Some(SlotProportion::new(1f32 / 16f32)),
-						telemetry: telemetry2,
-					},
-				)
-			})));
+		 relay_chain_slot_duration,
+		 para_id,
+		 collator_key,
+		 overseer_handle,
+		 announce_block| {
+			let slot_duration = cumulus_client_consensus_aura::slot_duration(&*client)?;

 			let proposer_factory = sc_basic_authorship::ProposerFactory::with_proof_recording(
 				task_manager.spawn_handle(),
 				client.clone(),
 				transaction_pool,
 				prometheus_registry,
-				telemetry,
+				telemetry.clone(),
+			);
+			let proposer = Proposer::new(proposer_factory);
+
+			let collator_service = CollatorService::new(
+				client.clone(),
+				Arc::new(task_manager.spawn_handle()),
+				announce_block,
+				client.clone(),
 			);

-			let relay_chain_consensus =
-				cumulus_client_consensus_relay_chain::build_relay_chain_consensus(
-					cumulus_client_consensus_relay_chain::BuildRelayChainConsensusParams {
-						para_id,
-						proposer_factory,
-						block_import,
-						relay_chain_interface: relay_chain_interface.clone(),
-						create_inherent_data_providers:
-							move |_, (relay_parent, validation_data)| {
-								let relay_chain_interface = relay_chain_interface.clone();
-								async move {
-									let parachain_inherent =
-									cumulus_primitives_parachain_inherent::ParachainInherentData::create_at(
-										relay_parent,
-										&relay_chain_interface,
-										&validation_data,
-										para_id,
-									).await;
-									let parachain_inherent =
-										parachain_inherent.ok_or_else(|| {
-											Box::<dyn std::error::Error + Send + Sync>::from(
-												"Failed to create parachain inherent",
-											)
-										})?;
-									Ok(parachain_inherent)
-								}
-							},
-					},
-				);
+			let params = BasicAuraParams {
+				create_inherent_data_providers: move |_, ()| async move { Ok(()) },
+				block_import,
+				para_client: client,
+				relay_client: relay_chain_interface,
+				sync_oracle,
+				keystore,
+				collator_key,
+				para_id,
+				overseer_handle,
+				slot_duration,
+				relay_chain_slot_duration,
+				proposer,
+				collator_service,
+				// Very limited proposal time.
+				authoring_duration: Duration::from_millis(500),
+			};

-			let parachain_consensus = Box::new(WaitForAuraConsensus {
-				client,
-				aura_consensus: Arc::new(Mutex::new(aura_consensus)),
-				relay_chain_consensus: Arc::new(Mutex::new(relay_chain_consensus)),
-				_phantom: PhantomData,
-			});
+			let fut =
+				basic_aura::run::<Block, <AuraId as AppCrypto>::Pair, _, _, _, _, _, _, _>(params);
+			task_manager.spawn_essential_handle().spawn("aura", None, fut);

-			Ok(parachain_consensus)
+			Ok(())
 		},
 		hwbench,
 	)
@@ -1263,14 +1207,15 @@ where
 }

 #[sc_tracing::logging::prefix_logs_with("Parachain")]
-async fn start_contracts_rococo_node_impl<RuntimeApi, RB, BIQ, BIC>(
+async fn start_contracts_rococo_node_impl<RuntimeApi, RB, BIQ, SC>(
 	parachain_config: Configuration,
 	polkadot_config: Configuration,
 	collator_options: CollatorOptions,
+	sybil_resistance_level: CollatorSybilResistance,
 	para_id: ParaId,
 	_rpc_ext_builder: RB,
 	build_import_queue: BIQ,
-	build_consensus: BIC,
+	start_consensus: SC,
 	hwbench: Option<sc_sysinfo::HwBench>,
 ) -> sc_service::error::Result<(TaskManager, Arc<ParachainClient<RuntimeApi>>)>
 where
@@ -1292,7 +1237,7 @@ where
 		Option<TelemetryHandle>,
 		&TaskManager,
 	) -> Result<sc_consensus::DefaultImportQueue<Block>, sc_service::Error>,
-	BIC: FnOnce(
+	SC: FnOnce(
 		Arc<ParachainClient<RuntimeApi>>,
 		ParachainBlockImport<RuntimeApi>,
 		Option<&Registry>,
@@ -1302,8 +1247,12 @@ where
 		Arc<sc_transaction_pool::FullPool<Block, ParachainClient<RuntimeApi>>>,
 		Arc<SyncingService<Block>>,
 		KeystorePtr,
-		bool,
-	) -> Result<Box<dyn ParachainConsensus<Block>>, sc_service::Error>,
+		Duration,
+		ParaId,
+		CollatorPair,
+		OverseerHandle,
+		Arc<dyn Fn(Hash, Option<Vec<u8>>) + Send + Sync>,
+	) -> Result<(), sc_service::Error>,
 {
 	let parachain_config = prepare_node_config(parachain_config);

@@ -1325,7 +1274,6 @@ where
 	.await
 	.map_err(|e| sc_service::Error::Application(Box::new(e) as Box<_>))?;

-	let force_authoring = parachain_config.force_authoring;
 	let validator = parachain_config.role.is_authority();
 	let prometheus_registry = parachain_config.prometheus_registry().cloned();
 	let transaction_pool = params.transaction_pool.clone();
@@ -1342,6 +1290,7 @@ where
 			spawn_handle: task_manager.spawn_handle(),
 			relay_chain_interface: relay_chain_interface.clone(),
 			import_queue: params.import_queue,
+			sybil_resistance_level,
 		})
 		.await?;

@@ -1401,8 +1350,26 @@ where
 	let overseer_handle = relay_chain_interface
 		.overseer_handle()
 		.map_err(|e| sc_service::Error::Application(Box::new(e)))?;
+
+	start_relay_chain_tasks(StartRelayChainTasksParams {
+		client: client.clone(),
+		announce_block: announce_block.clone(),
+		para_id,
+		relay_chain_interface: relay_chain_interface.clone(),
+		task_manager: &mut task_manager,
+		da_recovery_profile: if validator {
+			DARecoveryProfile::Collator
+		} else {
+			DARecoveryProfile::FullNode
+		},
+		import_queue: import_queue_service,
+		relay_chain_slot_duration,
+		recovery_handle: Box::new(overseer_handle.clone()),
+		sync_service: sync_service.clone(),
+	})?;
+
 	if validator {
-		let parachain_consensus = build_consensus(
+		start_consensus(
 			client.clone(),
 			block_import,
 			prometheus_registry.as_ref(),
@@ -1412,42 +1379,12 @@ where
 			transaction_pool,
 			sync_service.clone(),
 			params.keystore_container.keystore(),
-			force_authoring,
+			relay_chain_slot_duration,
+			para_id,
+			collator_key.expect("Command line arguments do not allow this. qed"),
+			overseer_handle,
+			announce_block,
 		)?;
-
-		let spawner = task_manager.spawn_handle();
-
-		let params = StartCollatorParams {
-			para_id,
-			block_status: client.clone(),
-			announce_block,
-			client: client.clone(),
-			task_manager: &mut task_manager,
-			relay_chain_interface,
-			spawner,
-			parachain_consensus,
-			import_queue: import_queue_service,
-			collator_key: collator_key.expect("Command line arguments do not allow this. qed"),
-			relay_chain_slot_duration,
-			recovery_handle: Box::new(overseer_handle),
-			sync_service,
-		};
-
-		start_collator(params).await?;
-	} else {
-		let params = StartFullNodeParams {
-			client: client.clone(),
-			announce_block,
-			task_manager: &mut task_manager,
-			para_id,
-			relay_chain_interface,
-			relay_chain_slot_duration,
-			import_queue: import_queue_service,
-			recovery_handle: Box::new(overseer_handle),
-			sync_service,
-		};
-
-		start_full_node(params)?;
 	}

 	start_network.start_network();
@@ -1508,6 +1445,7 @@ pub async fn start_contracts_rococo_node(
 		parachain_config,
 		polkadot_config,
 		collator_options,
+		CollatorSybilResistance::Resistant, // Aura
 		para_id,
 		|_| Ok(RpcModule::new(())),
 		contracts_rococo_build_import_queue,
@@ -1520,7 +1458,11 @@ pub async fn start_contracts_rococo_node(
 		 transaction_pool,
 		 sync_oracle,
 		 keystore,
-		 force_authoring| {
+		 relay_chain_slot_duration,
+		 para_id,
+		 collator_key,
+		 overseer_handle,
+		 announce_block| {
 			let slot_duration = cumulus_client_consensus_aura::slot_duration(&*client)?;

 			let proposer_factory = sc_basic_authorship::ProposerFactory::with_proof_recording(
@@ -1530,52 +1472,47 @@ pub async fn start_contracts_rococo_node(
 				prometheus_registry,
 				telemetry.clone(),
 			);
+			let proposer = Proposer::new(proposer_factory);

-			Ok(AuraConsensus::build::<sp_consensus_aura::sr25519::AuthorityPair, _, _, _, _, _, _>(
-				BuildAuraConsensusParams {
-					proposer_factory,
-					create_inherent_data_providers: move |_, (relay_parent, validation_data)| {
-						let relay_chain_interface = relay_chain_interface.clone();
-						async move {
-							let parachain_inherent =
-								cumulus_primitives_parachain_inherent::ParachainInherentData::create_at(
-									relay_parent,
-									&relay_chain_interface,
-									&validation_data,
-									para_id,
-								).await;
+			let collator_service = CollatorService::new(
+				client.clone(),
+				Arc::new(task_manager.spawn_handle()),
+				announce_block,
+				client.clone(),
+			);

-							let timestamp = sp_timestamp::InherentDataProvider::from_system_time();
+			let params = BasicAuraParams {
+				create_inherent_data_providers: move |_, ()| async move { Ok(()) },
+				block_import,
+				para_client: client,
+				relay_client: relay_chain_interface,
+				sync_oracle,
+				keystore,
+				collator_key,
+				para_id,
+				overseer_handle,
+				slot_duration,
+				relay_chain_slot_duration,
+				proposer,
+				collator_service,
+				// Very limited proposal time.
+				authoring_duration: Duration::from_millis(500),
+			};

-							let slot =
-								sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_slot_duration(
-									*timestamp,
-									slot_duration,
-								);
+			let fut = basic_aura::run::<
+				Block,
+				sp_consensus_aura::sr25519::AuthorityPair,
+				_,
+				_,
+				_,
+				_,
+				_,
+				_,
+				_,
+			>(params);
+			task_manager.spawn_essential_handle().spawn("aura", None, fut);

-							let parachain_inherent = parachain_inherent.ok_or_else(|| {
-								Box::<dyn std::error::Error + Send + Sync>::from(
-									"Failed to create parachain inherent",
-								)
-							})?;
-
-							Ok((slot, timestamp, parachain_inherent))
-						}
-					},
-					block_import,
-					para_client: client,
-					backoff_authoring_blocks: Option::<()>::None,
-					sync_oracle,
-					keystore,
-					force_authoring,
-					slot_duration,
-					// We got around 500ms for proposing
-					block_proposal_slot_portion: SlotProportion::new(1f32 / 24f32),
-					// And a maximum of 750ms if slots are skipped
-					max_block_proposal_slot_portion: Some(SlotProportion::new(1f32 / 16f32)),
-					telemetry,
-				},
-			))
+			Ok(())
 		},
 		hwbench,
 	)