diff --git a/cumulus/client/consensus/common/src/lib.rs b/cumulus/client/consensus/common/src/lib.rs
index 08bceabb2bd4a49ac8917e82caf1b97cd9eac183..cebe34e7ea58828372a9261e3be94866e119546a 100644
--- a/cumulus/client/consensus/common/src/lib.rs
+++ b/cumulus/client/consensus/common/src/lib.rs
@@ -111,12 +111,15 @@ impl<B: BlockT> ParachainConsensus<B> for Box<dyn ParachainConsensus<B> + Send +
 
 /// Parachain specific block import.
 ///
-/// This is used to set `block_import_params.fork_choice` to `false` as long as the block origin is
-/// not `NetworkInitialSync`. The best block for parachains is determined by the relay chain.
-/// Meaning we will update the best block, as it is included by the relay-chain.
+/// Specialized block import for parachains. It supports to delay setting the best block until the
+/// relay chain has included a candidate in its best block. By default the delayed best block
+/// setting is disabled. The block import also monitors the imported blocks and prunes by default if
+/// there are too many blocks at the same height. Too many blocks at the same height can for example
+/// happen if the relay chain is rejecting the parachain blocks in the validation.
 pub struct ParachainBlockImport<Block: BlockT, BI, BE> {
 	inner: BI,
 	monitor: Option<SharedData<LevelMonitor<Block, BE>>>,
+	delayed_best_block: bool,
 }
 
 impl<Block: BlockT, BI, BE: Backend<Block>> ParachainBlockImport<Block, BI, BE> {
@@ -141,13 +144,27 @@ impl<Block: BlockT, BI, BE: Backend<Block>> ParachainBlockImport<Block, BI, BE>
 		let monitor =
 			level_limit.map(|level_limit| SharedData::new(LevelMonitor::new(level_limit, backend)));
 
-		Self { inner, monitor }
+		Self { inner, monitor, delayed_best_block: false }
+	}
+
+	/// Create a new instance which delays setting the best block.
+	///
+	/// The number of leaves per level limit is set to `LevelLimit::Default`.
+	pub fn new_with_delayed_best_block(inner: BI, backend: Arc<BE>) -> Self {
+		Self {
+			delayed_best_block: true,
+			..Self::new_with_limit(inner, backend, LevelLimit::Default)
+		}
 	}
 }
 
 impl<Block: BlockT, I: Clone, BE> Clone for ParachainBlockImport<Block, I, BE> {
 	fn clone(&self) -> Self {
-		ParachainBlockImport { inner: self.inner.clone(), monitor: self.monitor.clone() }
+		ParachainBlockImport {
+			inner: self.inner.clone(),
+			monitor: self.monitor.clone(),
+			delayed_best_block: self.delayed_best_block,
+		}
 	}
 }
 
@@ -182,11 +199,13 @@ where
 			params.finalized = true;
 		}
 
-		// Best block is determined by the relay chain, or if we are doing the initial sync
-		// we import all blocks as new best.
-		params.fork_choice = Some(sc_consensus::ForkChoiceStrategy::Custom(
-			params.origin == sp_consensus::BlockOrigin::NetworkInitialSync,
-		));
+		if self.delayed_best_block {
+			// Best block is determined by the relay chain, or if we are doing the initial sync
+			// we import all blocks as new best.
+			params.fork_choice = Some(sc_consensus::ForkChoiceStrategy::Custom(
+				params.origin == sp_consensus::BlockOrigin::NetworkInitialSync,
+			));
+		}
 
 		let maybe_lock = self.monitor.as_ref().map(|monitor_lock| {
 			let mut monitor = monitor_lock.shared_data_locked();
diff --git a/cumulus/client/consensus/common/src/tests.rs b/cumulus/client/consensus/common/src/tests.rs
index 9658a0add790da01ad9fe83f11956d1fbd88523a..597d1ab2acc2cff42d3230898c1129a7ba63b6f3 100644
--- a/cumulus/client/consensus/common/src/tests.rs
+++ b/cumulus/client/consensus/common/src/tests.rs
@@ -1124,7 +1124,8 @@ fn find_potential_parents_aligned_with_pending() {
 
 	let backend = Arc::new(Backend::new_test(1000, 1));
 	let client = Arc::new(TestClientBuilder::with_backend(backend.clone()).build());
-	let mut para_import = ParachainBlockImport::new(client.clone(), backend.clone());
+	let mut para_import =
+		ParachainBlockImport::new_with_delayed_best_block(client.clone(), backend.clone());
 
 	let relay_parent = relay_hash_from_block_num(10);
 	// Choose different relay parent for alternative chain to get new hashes.
@@ -1279,7 +1280,8 @@ fn find_potential_parents_aligned_no_pending() {
 
 	let backend = Arc::new(Backend::new_test(1000, 1));
 	let client = Arc::new(TestClientBuilder::with_backend(backend.clone()).build());
-	let mut para_import = ParachainBlockImport::new(client.clone(), backend.clone());
+	let mut para_import =
+		ParachainBlockImport::new_with_delayed_best_block(client.clone(), backend.clone());
 
 	let relay_parent = relay_hash_from_block_num(10);
 	// Choose different relay parent for alternative chain to get new hashes.