Introduce basic slot-based collator (paritytech#4097)

Part of paritytech#3168 
On top of paritytech#3568

### Changes Overview
- Introduces a new collator variant in
`cumulus/client/consensus/aura/src/collators/slot_based/mod.rs`
- Two tasks are part of that module, one for block building and one for
collation building and submission.
- Introduces a new variant of `cumulus-test-runtime` which has 2s slot
duration, used for zombienet testing
- Zombienet tests for the new collator

**Note:** This collator is considered experimental and should only be
used for testing and exploration for now.

### Comparison with `lookahead` collator
- The new variant is slot based, meaning it waits for the next slot of
the parachain, then starts authoring
- The search for potential parents remains mostly unchanged from
lookahead
- As anchor, we use the current best relay parent
- In general, the new collator tends to be anchored to one relay parent
earlier. `lookahead` generally waits for a new relay block to arrive
before it attempts to build a block. This means the actual timing of
parachain blocks depends on when the relay block has been authored and
imported. With the slot-triggered approach we are authoring directly on
the slot boundary, were a new relay chain block has probably not yet
arrived.

### Limitations
- Overall, the current implementation focuses on the "happy path"
- We assume that we want to collate close to the tip of the relay chain.
It would be useful however to have some kind of configurable drift, so
that we could lag behind a bit.
paritytech#3965
- The collation task is pretty dumb currently. It checks if we have
cores scheduled and if yes, submits all the messages we have received
from the block builder until we have something submitted for every core.
Ideally we should do some extra checks, i.e. we do not need to submit if
the built block is already too old (build on a out of range relay
parent) or was authored with a relay parent that is not an ancestor of
the relay block we are submitting at.
paritytech#3966
- There is no throttling, we assume that we can submit _velocity_ blocks
every relay chain block. There should be communication between the
collator task and block-builder task.
- The parent search and ConsensusHook are not yet properly adjusted. The
parent search makes assumptions about the pending candidate which no
longer hold. paritytech#3967
- Custom triggers for block building not implemented.

---------

Co-authored-by: Davide Galassi <davxy@datawok.net>
Co-authored-by: Andrei Sandu <54316454+sandreim@users.noreply.github.com>
Co-authored-by: Bastian Köcher <git@kchr.de>
Co-authored-by: Javier Viola <363911+pepoviola@users.noreply.github.com>
Co-authored-by: command-bot <>

.gitlab/pipeline/zombienet/cumulus.yml

@@ -149,3 +149,27 @@ zombienet-cumulus-0007-full_node_warp_sync:
       --local-dir="${LOCAL_DIR}"
       --concurrency=1
       --test="0007-full_node_warp_sync.zndsl"
+
+zombienet-cumulus-0008-elastic_authoring:
+  extends:
+    - .zombienet-cumulus-common
+    - .zombienet-refs
+    - .zombienet-before-script
+    - .zombienet-after-script
+  script:
+    - /home/nonroot/zombie-net/scripts/ci/run-test-local-env-manager.sh
+      --local-dir="${LOCAL_DIR}"
+      --concurrency=1
+      --test="0008-elastic_authoring.zndsl"
+
+zombienet-cumulus-0009-elastic_pov_recovery:
+  extends:
+    - .zombienet-cumulus-common
+    - .zombienet-refs
+    - .zombienet-before-script
+    - .zombienet-after-script
+  script:
+    - /home/nonroot/zombie-net/scripts/ci/run-test-local-env-manager.sh
+      --local-dir="${LOCAL_DIR}"
+      --concurrency=1
+      --test="0009-elastic_pov_recovery.zndsl"

cumulus/client/consensus/aura/Cargo.toml

@@ -16,13 +16,15 @@ futures = { workspace = true }
 parking_lot = { workspace = true }
 tracing = { workspace = true, default-features = true }
 schnellru = { workspace = true }
+tokio = { workspace = true, features = ["macros"] }
 
 # Substrate
 sc-client-api = { workspace = true, default-features = true }
 sc-consensus = { workspace = true, default-features = true }
 sc-consensus-aura = { workspace = true, default-features = true }
 sc-consensus-babe = { workspace = true, default-features = true }
 sc-consensus-slots = { workspace = true, default-features = true }
+sc-utils = { workspace = true, default-features = true }
 sc-telemetry = { workspace = true, default-features = true }
 sp-api = { workspace = true, default-features = true }
 sp-application-crypto = { workspace = true, default-features = true }

cumulus/client/consensus/aura/src/collator.rs

@@ -156,26 +156,16 @@ where
 		Ok((paras_inherent_data, other_inherent_data))
 	}
 
-	/// Propose, seal, and import a block, packaging it into a collation.
-	///
-	/// Provide the slot to build at as well as any other necessary pre-digest logs,
-	/// the inherent data, and the proposal duration and PoV size limits.
-	///
-	/// The Aura pre-digest should not be explicitly provided and is set internally.
-	///
-	/// This does not announce the collation to the parachain network or the relay chain.
-	pub async fn collate(
+	/// Build and import a parachain block on the given parent header, using the given slot claim.
+	pub async fn build_block_and_import(
 		&mut self,
 		parent_header: &Block::Header,
 		slot_claim: &SlotClaim<P::Public>,
 		additional_pre_digest: impl Into<Option<Vec<DigestItem>>>,
 		inherent_data: (ParachainInherentData, InherentData),
 		proposal_duration: Duration,
 		max_pov_size: usize,
-	) -> Result<
-		Option<(Collation, ParachainBlockData<Block>, Block::Hash)>,
-		Box<dyn Error + Send + 'static>,
-	> {
+	) -> Result<Option<ParachainCandidate<Block>>, Box<dyn Error + Send + 'static>> {
 		let mut digest = additional_pre_digest.into().unwrap_or_default();
 		digest.push(slot_claim.pre_digest.clone());
 
@@ -205,7 +195,6 @@ where
 		)
 		.map_err(|e| e as Box<dyn Error + Send>)?;
 
-		let post_hash = sealed_importable.post_hash();
 		let block = Block::new(
 			sealed_importable.post_header(),
 			sealed_importable
@@ -220,11 +209,46 @@ where
 			.map_err(|e| Box::new(e) as Box<dyn Error + Send>)
 			.await?;
 
-		if let Some((collation, block_data)) = self.collator_service.build_collation(
-			parent_header,
-			post_hash,
-			ParachainCandidate { block, proof: proposal.proof },
-		) {
+		Ok(Some(ParachainCandidate { block, proof: proposal.proof }))
+	}
+
+	/// Propose, seal, import a block and packaging it into a collation.
+	///
+	/// Provide the slot to build at as well as any other necessary pre-digest logs,
+	/// the inherent data, and the proposal duration and PoV size limits.
+	///
+	/// The Aura pre-digest should not be explicitly provided and is set internally.
+	///
+	/// This does not announce the collation to the parachain network or the relay chain.
+	pub async fn collate(
+		&mut self,
+		parent_header: &Block::Header,
+		slot_claim: &SlotClaim<P::Public>,
+		additional_pre_digest: impl Into<Option<Vec<DigestItem>>>,
+		inherent_data: (ParachainInherentData, InherentData),
+		proposal_duration: Duration,
+		max_pov_size: usize,
+	) -> Result<
+		Option<(Collation, ParachainBlockData<Block>, Block::Hash)>,
+		Box<dyn Error + Send + 'static>,
+	> {
+		let maybe_candidate = self
+			.build_block_and_import(
+				parent_header,
+				slot_claim,
+				additional_pre_digest,
+				inherent_data,
+				proposal_duration,
+				max_pov_size,
+			)
+			.await?;
+
+		let Some(candidate) = maybe_candidate else { return Ok(None) };
+
+		let hash = candidate.block.header().hash();
+		if let Some((collation, block_data)) =
+			self.collator_service.build_collation(parent_header, hash, candidate)
+		{
 			tracing::info!(
 				target: crate::LOG_TARGET,
 				"PoV size {{ header: {}kb, extrinsics: {}kb, storage_proof: {}kb }}",
@@ -241,7 +265,7 @@ where
 				);
 			}
 
-			Ok(Some((collation, block_data, post_hash)))
+			Ok(Some((collation, block_data, hash)))
 		} else {
 			Err(Box::<dyn Error + Send + Sync>::from("Unable to produce collation")
 				as Box<dyn Error + Send>)

cumulus/client/consensus/aura/src/collators/basic.rs

@@ -41,7 +41,6 @@ use sc_consensus::BlockImport;
 use sp_api::{CallApiAt, ProvideRuntimeApi};
 use sp_application_crypto::AppPublic;
 use sp_blockchain::HeaderBackend;
-use sp_consensus::SyncOracle;
 use sp_consensus_aura::AuraApi;
 use sp_core::crypto::Pair;
 use sp_inherents::CreateInherentDataProviders;
@@ -53,7 +52,7 @@ use std::{sync::Arc, time::Duration};
 use crate::collator as collator_util;
 
 /// Parameters for [`run`].
-pub struct Params<BI, CIDP, Client, RClient, SO, Proposer, CS> {
+pub struct Params<BI, CIDP, Client, RClient, Proposer, CS> {
 	/// Inherent data providers. Only non-consensus inherent data should be provided, i.e.
 	/// the timestamp, slot, and paras inherents should be omitted, as they are set by this
 	/// collator.
@@ -64,8 +63,6 @@ pub struct Params<BI, CIDP, Client, RClient, SO, Proposer, CS> {
 	pub para_client: Arc<Client>,
 	/// A handle to the relay-chain client.
 	pub relay_client: RClient,
-	/// A chain synchronization oracle.
-	pub sync_oracle: SO,
 	/// The underlying keystore, which should contain Aura consensus keys.
 	pub keystore: KeystorePtr,
 	/// The collator key used to sign collations before submitting to validators.
@@ -89,8 +86,8 @@ pub struct Params<BI, CIDP, Client, RClient, SO, Proposer, CS> {
 }
 
 /// Run bare Aura consensus as a relay-chain-driven collator.
-pub fn run<Block, P, BI, CIDP, Client, RClient, SO, Proposer, CS>(
-	params: Params<BI, CIDP, Client, RClient, SO, Proposer, CS>,
+pub fn run<Block, P, BI, CIDP, Client, RClient, Proposer, CS>(
+	params: Params<BI, CIDP, Client, RClient, Proposer, CS>,
 ) -> impl Future<Output = ()> + Send + 'static
 where
 	Block: BlockT + Send,
@@ -108,7 +105,6 @@ where
 	CIDP: CreateInherentDataProviders<Block, ()> + Send + 'static,
 	CIDP::InherentDataProviders: Send,
 	BI: BlockImport<Block> + ParachainBlockImportMarker + Send + Sync + 'static,
-	SO: SyncOracle + Send + Sync + Clone + 'static,
 	Proposer: ProposerInterface<Block> + Send + Sync + 'static,
 	CS: CollatorServiceInterface<Block> + Send + Sync + 'static,
 	P: Pair,