Enforce current relay parent to be available (#11621)

I would suggest to encode our intention directly instead of working
around a symptom. This makes it much easier to reason about the code in
my opinion and should have less edge cases. In particular this change
will also wait for the current relay parent mid-parachain slot, which is
useful as otherwise we would build on an older than expected relay
parent, which could then again affect block confidence as the relay
parent might be out of scope already before the collation can land on
chain.

Also worth mentioning: As the original PR already shows, we have the
implicit assumption that the current relay parent has arrived after 1s
into the relay chain slot. This seems to be the case most of the time,
but not always - triggering the issue this PR is fixing. For best
performance we should consider bumping the slot offset some more. If I
understand correctly the error case we found was by a relay parent
coming late by only a couple of milliseconds - thus 1.5s might already
be plenty for the needed wait to almost never happen, but ideally we
should find a good value based on data from prod.

---------

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>
Co-authored-by: Alexandru Vasile <alexandru.vasile@parity.io>

Author: eskimor

cumulus/client/consensus/aura/src/collators/slot_based/block_builder_task.rs

@@ -39,9 +39,7 @@ use cumulus_primitives_core::{
 };
 use cumulus_relay_chain_interface::RelayChainInterface;
 use futures::prelude::*;
-use polkadot_primitives::{
-	Block as RelayBlock, CoreIndex, Hash as RelayHash, Header as RelayHeader, Id as ParaId,
-};
+use polkadot_primitives::{Block as RelayBlock, CoreIndex, Header as RelayHeader, Id as ParaId};
 use sc_client_api::{backend::AuxStore, BlockBackend, BlockOf, UsageProvider};
 use sc_consensus::BlockImport;
 use sc_consensus_aura::SlotDuration;
@@ -188,6 +186,18 @@ where
 			collator_util::Collator::<Block, P, _, _, _, _, _>::new(params)
 		};
 
+		let mut best_notifications = match relay_client.new_best_notification_stream().await {
+			Ok(s) => s,
+			Err(err) => {
+				tracing::error!(
+					target: LOG_TARGET,
+					?err,
+					"Failed to initialize consensus: no relay chain best block notification stream"
+				);
+				return;
+			},
+		};
+
 		let mut relay_chain_data_cache = RelayChainDataCache::new(relay_client.clone(), para_id);
 		let mut connection_helper = BackingGroupConnectionHelper::new(
 			keystore.clone(),
@@ -205,7 +215,19 @@ where
 				return;
 			};
 
-			let Ok(relay_best_hash) = relay_client.best_block_hash().await else {
+			// Wait for the best relay block to be from the current relay
+			// chain slot. If propagation exceeded `slot_offset`, this
+			// blocks until a new-best notification arrives.
+			// See: https://github.com/paritytech/polkadot-sdk/pull/11453
+			let Some(relay_best_header) = wait_for_current_relay_block(
+				&relay_client,
+				&mut relay_chain_data_cache,
+				&mut best_notifications,
+				slot_offset,
+				relay_chain_slot_duration,
+			)
+			.await
+			else {
 				tracing::warn!(target: crate::LOG_TARGET, "Unable to fetch latest relay chain block hash.");
 				continue;
 			};
@@ -221,7 +243,7 @@ where
 
 			let Ok(Some(rp_data)) = offset_relay_parent_find_descendants(
 				&mut relay_chain_data_cache,
-				relay_best_hash,
+				relay_best_header,
 				relay_parent_offset,
 			)
 			.await
@@ -507,6 +529,84 @@ fn adjust_para_to_relay_parent_slot(
 	Some(para_slot)
 }
 
+/// Returns `true` if the best relay chain block is from the current relay chain
+/// slot. Uses the wall clock adjusted by `slot_offset`.
+fn is_best_relay_block_current(
+	best_relay_slot: u64,
+	slot_offset: Duration,
+	relay_chain_slot_duration: Duration,
+) -> bool {
+	let now = super::slot_timer::duration_now().saturating_sub(slot_offset);
+	is_best_relay_block_current_at(best_relay_slot, now, relay_chain_slot_duration)
+}
+
+/// Pure logic for the relay block freshness check, taking the current time as
+/// a parameter for testability.
+fn is_best_relay_block_current_at(
+	best_relay_slot: u64,
+	now: Duration,
+	relay_chain_slot_duration: Duration,
+) -> bool {
+	let current_relay_slot = now.as_millis() as u64 / relay_chain_slot_duration.as_millis() as u64;
+	best_relay_slot >= current_relay_slot
+}
+
+/// Wait until the best relay chain block is from the current relay chain slot.
+///
+/// If the current best block is already current, returns its hash immediately.
+/// Otherwise waits for a new-best notification and re-checks. This ensures
+/// the collator doesn't build on a stale relay parent when relay block
+/// propagation exceeds `slot_offset` at a slot boundary.
+///
+/// Returns the best relay block hash, or `None` on error.
+pub(crate) async fn wait_for_current_relay_block<RelayClient>(
+	relay_client: &RelayClient,
+	relay_chain_data_cache: &mut RelayChainDataCache<RelayClient>,
+	best_notifications: &mut (impl Stream<Item = RelayHeader> + Unpin),
+	slot_offset: Duration,
+	relay_chain_slot_duration: Duration,
+) -> Option<RelayHeader>
+where
+	RelayClient: RelayChainInterface + Clone + 'static,
+{
+	let relay_best_hash = relay_client.best_block_hash().await.ok()?;
+	let mut first_best_header = Some(
+		relay_chain_data_cache
+			.get_mut_relay_chain_data(relay_best_hash)
+			.await
+			.ok()
+			.map(|d| d.relay_parent_header.clone())?,
+	);
+
+	loop {
+		// Drain buffered notifications.
+		while let Some(maybe_header) = best_notifications.next().now_or_never() {
+			first_best_header = Some(maybe_header?);
+		}
+
+		let best_header = match first_best_header.take() {
+			Some(h) => h,
+			None => best_notifications.next().await?, // Block until one arrives.
+		};
+
+		let best_slot = sc_consensus_babe::find_pre_digest::<RelayBlock>(&best_header)
+			.map(|d| d.slot())
+			.ok()?;
+
+		if is_best_relay_block_current(*best_slot, slot_offset, relay_chain_slot_duration) {
+			return Some(best_header);
+		}
+
+		tracing::debug!(
+			target: LOG_TARGET,
+			?relay_best_hash,
+			relay_best_num = %best_header.number(),
+			?best_slot,
+			"Best relay block is stale, waiting for fresh one."
+		);
+	}
+}
+
 /// Finds a relay chain parent block at a specified offset from the best block, collecting its
 /// descendants.
 ///
@@ -518,21 +618,13 @@ fn adjust_para_to_relay_parent_slot(
 /// offset, collecting all blocks in between to maintain the chain of ancestry.
 pub(crate) async fn offset_relay_parent_find_descendants<RelayClient>(
 	relay_chain_data_cache: &mut RelayChainDataCache<RelayClient>,
-	relay_best_block: RelayHash,
+	mut relay_header: RelayHeader,
 	relay_parent_offset: u32,
 ) -> Result<Option<RelayParentData>, ()>
 where
 	RelayClient: RelayChainInterface + Clone + 'static,
 {
-	let Ok(mut relay_header) = relay_chain_data_cache
-		.get_mut_relay_chain_data(relay_best_block)
-		.await
-		.map(|d| d.relay_parent_header.clone())
-	else {
-		tracing::error!(target: LOG_TARGET, ?relay_best_block, "Unable to fetch best relay chain block header.");
-		return Err(());
-	};
-
+	let relay_best_block = relay_header.hash();
 	if relay_parent_offset == 0 {
 		return Ok(Some(RelayParentData::new(relay_header)));
 	}
@@ -668,3 +760,91 @@ pub(crate) async fn determine_core<H: HeaderT, RI: RelayChainInterface + 'static
 		number_of_cores: cores_at_offset.len() as u16,
 	}))
 }
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	const RELAY_SLOT_DURATION: Duration = Duration::from_secs(6);
+
+	/// Simulate the wall clock at a specific point within a relay slot.
+	///
+	/// `relay_slot` is the current relay chain slot number, `ms_into_slot` is
+	/// how far into that slot we are (0..6000).
+	fn now_at(relay_slot: u64, ms_into_slot: u64) -> Duration {
+		Duration::from_millis(relay_slot * 6000 + ms_into_slot)
+	}
+
+	// ---------------------------------------------------------------
+	// Tests for `is_best_relay_block_current_at`
+	// ---------------------------------------------------------------
+
+	#[test]
+	fn best_block_in_current_slot_is_current() {
+		// Wall clock in slot 804, best block from slot 804 → current.
+		assert!(is_best_relay_block_current_at(804, now_at(804, 500), RELAY_SLOT_DURATION));
+	}
+
+	#[test]
+	fn best_block_in_previous_slot_is_stale() {
+		// Wall clock in slot 805, best block from slot 804 → stale.
+		assert!(!is_best_relay_block_current_at(804, now_at(805, 500), RELAY_SLOT_DURATION));
+	}
+
+	#[test]
+	fn the_bug_scenario_best_block_stale_at_slot_boundary() {
+		// THE BUG: wall clock just crossed into slot 805 (17ms in),
+		// but best relay block is still from slot 804. Stale.
+		assert!(!is_best_relay_block_current_at(804, now_at(805, 17), RELAY_SLOT_DURATION));
+	}
+
+	#[test]
+	fn best_block_current_after_new_relay_block_arrives() {
+		// New relay block (slot 805) arrives. Wall clock in slot 805.
+		assert!(is_best_relay_block_current_at(805, now_at(805, 500), RELAY_SLOT_DURATION));
+	}
+
+	#[test]
+	fn best_block_from_future_slot_is_current() {
+		// Should not happen, but must not panic.
+		assert!(is_best_relay_block_current_at(810, now_at(805, 0), RELAY_SLOT_DURATION));
+	}
+
+	#[test]
+	fn stale_at_exact_slot_boundary() {
+		// Exactly at the start of slot 805.
+		// Best from 804 → stale (804 < 805).
+		assert!(!is_best_relay_block_current_at(804, now_at(805, 0), RELAY_SLOT_DURATION));
+		// Best from 805 → current.
+		assert!(is_best_relay_block_current_at(805, now_at(805, 0), RELAY_SLOT_DURATION));
+	}
+
+	#[test]
+	fn current_at_end_of_slot() {
+		// 5999ms into slot 804 — still in slot 804.
+		// Best from 804 → current.
+		assert!(is_best_relay_block_current_at(804, now_at(804, 5999), RELAY_SLOT_DURATION));
+	}
+
+	#[test]
+	fn no_wait_needed_during_normal_building() {
+		// During elastic scaling in slot 804: best is from 804,
+		// wall clock is mid-slot 804. No wait needed.
+		for ms in (0..6000).step_by(500) {
+			assert!(
+				is_best_relay_block_current_at(804, now_at(804, ms), RELAY_SLOT_DURATION),
+				"Should be current at {}ms into slot 804",
+				ms
+			);
+		}
+	}
+
+	#[test]
+	fn wait_needed_when_slot_advances() {
+		// Wall clock moves to slot 805, best still from 804.
+		// This is the race condition — must detect as stale.
+		assert!(!is_best_relay_block_current_at(804, now_at(805, 0), RELAY_SLOT_DURATION));
+		assert!(!is_best_relay_block_current_at(804, now_at(805, 17), RELAY_SLOT_DURATION));
+		assert!(!is_best_relay_block_current_at(804, now_at(805, 500), RELAY_SLOT_DURATION));
+	}
+}

cumulus/client/consensus/aura/src/collators/slot_based/slot_timer.rs

@@ -140,7 +140,7 @@ fn compute_time_until_next_slot_change(
 }
 
 /// Returns current duration since Unix epoch.
-fn duration_now() -> Duration {
+pub(super) fn duration_now() -> Duration {
 	use std::time::SystemTime;
 	let now = SystemTime::now();
 	now.duration_since(SystemTime::UNIX_EPOCH).unwrap_or_else(|e| {