Always run on-demand headers relay in complex relay (#975)

* always run on-demand headers relay in complex relay

* fix compilation

bridges/relays/bin-substrate/src/cli/relay_headers.rs

@@ -103,7 +103,6 @@ impl RelayHeaders {
 				Finality::new(target_client.clone(), target_sign),
 				source_client,
 				target_client,
-				false,
 				metrics_params,
 			)
 			.await

bridges/relays/bin-substrate/src/finality_pipeline.rs

@@ -26,12 +26,12 @@ use sp_core::Bytes;
 use std::{fmt::Debug, marker::PhantomData, time::Duration};
 
 /// Default synchronization loop timeout.
-const STALL_TIMEOUT: Duration = Duration::from_secs(120);
+pub(crate) const STALL_TIMEOUT: Duration = Duration::from_secs(120);
 /// Default limit of recent finality proofs.
 ///
 /// Finality delay of 4096 blocks is unlikely to happen in practice in
 /// Substrate+GRANDPA based chains (good to know).
-const RECENT_FINALITY_PROOFS_LIMIT: usize = 4096;
+pub(crate) const RECENT_FINALITY_PROOFS_LIMIT: usize = 4096;
 
 /// Headers sync pipeline for Substrate <-> Substrate relays.
 pub trait SubstrateFinalitySyncPipeline: FinalitySyncPipeline {
@@ -119,7 +119,6 @@ pub async fn run<SourceChain, TargetChain, P>(
 	pipeline: P,
 	source_client: Client<SourceChain>,
 	target_client: Client<TargetChain>,
-	is_on_demand_task: bool,
 	metrics_params: MetricsParams,
 ) -> anyhow::Result<()>
 where
@@ -142,10 +141,9 @@ where
 	);
 
 	finality_relay::run(
-		FinalitySource::new(source_client),
+		FinalitySource::new(source_client, None),
 		SubstrateFinalityTarget::new(target_client, pipeline),
 		FinalitySyncParams {
-			is_on_demand_task,
 			tick: std::cmp::max(SourceChain::AVERAGE_BLOCK_INTERVAL, TargetChain::AVERAGE_BLOCK_INTERVAL),
 			recent_finality_proofs_limit: RECENT_FINALITY_PROOFS_LIMIT,
 			stall_timeout: STALL_TIMEOUT,

bridges/relays/bin-substrate/src/messages_source.rs

@@ -239,7 +239,7 @@ where
 
 	async fn require_target_header_on_source(&self, id: TargetHeaderIdOf<P>) {
 		if let Some(ref target_to_source_headers_relay) = self.target_to_source_headers_relay {
-			target_to_source_headers_relay.require_finalized_header(id);
+			target_to_source_headers_relay.require_finalized_header(id).await;
 		}
 	}
 }

bridges/relays/bin-substrate/src/messages_target.rs

@@ -226,7 +226,7 @@ where
 
 	async fn require_source_header_on_target(&self, id: SourceHeaderIdOf<P>) {
 		if let Some(ref source_to_target_headers_relay) = self.source_to_target_headers_relay {
-			source_to_target_headers_relay.require_finalized_header(id);
+			source_to_target_headers_relay.require_finalized_header(id).await;
 		}
 	}
 }

bridges/relays/bin-substrate/src/on_demand_headers.rs

@@ -16,39 +16,38 @@
 
 //! On-demand Substrate -> Substrate headers relay.
 
-use crate::finality_pipeline::{SubstrateFinalitySyncPipeline, SubstrateFinalityToSubstrate};
+use crate::finality_pipeline::{
+	SubstrateFinalitySyncPipeline, SubstrateFinalityToSubstrate, RECENT_FINALITY_PROOFS_LIMIT, STALL_TIMEOUT,
+};
 use crate::finality_target::SubstrateFinalityTarget;
 
+use async_std::sync::{Arc, Mutex};
 use bp_header_chain::justification::GrandpaJustification;
 use finality_relay::{
-	FinalitySyncPipeline, SourceClient as FinalitySourceClient, TargetClient as FinalityTargetClient,
-};
-use futures::{
-	channel::{mpsc, oneshot},
-	select, FutureExt, StreamExt,
+	FinalitySyncParams, FinalitySyncPipeline, SourceClient as FinalitySourceClient,
+	TargetClient as FinalityTargetClient,
 };
+use futures::{select, FutureExt};
 use num_traits::{CheckedSub, Zero};
 use relay_substrate_client::{
-	finality_source::FinalitySource as SubstrateFinalitySource, BlockNumberOf, Chain, Client, HashOf, HeaderIdOf,
-	SyncHeader,
+	finality_source::{FinalitySource as SubstrateFinalitySource, RequiredHeaderNumberRef},
+	BlockNumberOf, Chain, Client, HashOf, HeaderIdOf, SyncHeader,
 };
 use relay_utils::{
-	metrics::MetricsParams, relay_loop::Client as RelayClient, BlockNumberBase, FailedClient, HeaderId,
-	MaybeConnectionError,
+	metrics::MetricsParams, relay_loop::Client as RelayClient, BlockNumberBase, FailedClient, MaybeConnectionError,
 };
 use std::fmt::Debug;
 
 /// On-demand Substrate <-> Substrate headers relay.
 ///
-/// This relay may be started by messages whenever some other relay (e.g. messages relay) needs more
-/// headers to be relayed to continue its regular work. When enough headers are relayed, on-demand
-/// relay may be deactivated.
+/// This relay may be requested to sync more headers, whenever some other relay (e.g. messages relay) needs
+/// it to continue its regular work. When enough headers are relayed, on-demand stops syncing headers.
 #[derive(Clone)]
 pub struct OnDemandHeadersRelay<SourceChain: Chain> {
-	/// Background task name.
-	background_task_name: String,
-	/// Required headers to background sender.
-	required_header_tx: mpsc::Sender<HeaderId<SourceChain::Hash, SourceChain::BlockNumber>>,
+	/// Relay task name.
+	relay_task_name: String,
+	/// Shared reference to maximal required finalized header number.
+	required_header_number: RequiredHeaderNumberRef<SourceChain>,
 }
 
 impl<SourceChain: Chain> OnDemandHeadersRelay<SourceChain> {
@@ -75,49 +74,49 @@ impl<SourceChain: Chain> OnDemandHeadersRelay<SourceChain> {
 		SubstrateFinalityTarget<TargetChain, SubstrateFinalityToSubstrate<SourceChain, TargetChain, TargetSign>>:
 			FinalityTargetClient<SubstrateFinalityToSubstrate<SourceChain, TargetChain, TargetSign>>,
 	{
-		let (required_header_tx, required_header_rx) = mpsc::channel(1);
+		let required_header_number = Arc::new(Mutex::new(Zero::zero()));
+		let this = OnDemandHeadersRelay {
+			relay_task_name: on_demand_headers_relay_name::<SourceChain, TargetChain>(),
+			required_header_number: required_header_number.clone(),
+		};
 		async_std::task::spawn(async move {
 			background_task(
 				source_client,
 				target_client,
 				pipeline,
 				maximal_headers_difference,
-				required_header_rx,
+				required_header_number,
 			)
 			.await;
 		});
 
-		let background_task_name = format!(
-			"{}-background",
-			on_demand_headers_relay_name::<SourceChain, TargetChain>()
-		);
-		OnDemandHeadersRelay {
-			background_task_name,
-			required_header_tx,
-		}
+		this
 	}
 
 	/// Someone is asking us to relay given finalized header.
-	pub fn require_finalized_header(&self, header_id: HeaderIdOf<SourceChain>) {
-		if let Err(error) = self.required_header_tx.clone().try_send(header_id) {
-			log::error!(
+	pub async fn require_finalized_header(&self, header_id: HeaderIdOf<SourceChain>) {
+		let mut required_header_number = self.required_header_number.lock().await;
+		if header_id.0 > *required_header_number {
+			log::trace!(
 				target: "bridge",
-				"Failed to send require header id {:?} to {:?}: {:?}",
-				header_id,
-				self.background_task_name,
-				error,
+				"More {} headers required in {} relay. Going to sync up to the {}",
+				SourceChain::NAME,
+				self.relay_task_name,
+				header_id.0,
 			);
+
+			*required_header_number = header_id.0;
 		}
 	}
 }
 
-/// Background task that is responsible for starting and stopping headers relay when required.
+/// Background task that is responsible for starting headers relay.
 async fn background_task<SourceChain, TargetChain, TargetSign>(
 	source_client: Client<SourceChain>,
 	target_client: Client<TargetChain>,
 	pipeline: SubstrateFinalityToSubstrate<SourceChain, TargetChain, TargetSign>,
 	maximal_headers_difference: SourceChain::BlockNumber,
-	mut required_header_rx: mpsc::Receiver<HeaderIdOf<SourceChain>>,
+	required_header_number: RequiredHeaderNumberRef<SourceChain>,
 ) where
 	SourceChain: Chain + Debug,
 	SourceChain::BlockNumber: BlockNumberBase,
@@ -138,36 +137,19 @@ async fn background_task<SourceChain, TargetChain, TargetSign>(
 	let mut finality_source = SubstrateFinalitySource::<
 		_,
 		SubstrateFinalityToSubstrate<SourceChain, TargetChain, TargetSign>,
-	>::new(source_client.clone());
+	>::new(source_client.clone(), Some(required_header_number.clone()));
 	let mut finality_target = SubstrateFinalityTarget::new(target_client.clone(), pipeline.clone());
 
-	let mut active_headers_relay = None;
-	let mut required_header_number = Zero::zero();
-	let mut relay_exited_rx = futures::future::pending().left_future();
+	let mut restart_relay = true;
+	let finality_relay_task = futures::future::Fuse::terminated();
+	futures::pin_mut!(finality_relay_task);
 
 	loop {
-		// wait for next target block or for new required header
 		select! {
 			_ = async_std::task::sleep(TargetChain::AVERAGE_BLOCK_INTERVAL).fuse() => {},
-			required_header_id = required_header_rx.next() => {
-				match required_header_id {
-					Some(required_header_id) => {
-						if required_header_id.0 > required_header_number {
-							required_header_number = required_header_id.0;
-						}
-					},
-					None => {
-						// that's the only way to exit background task - to drop `required_header_tx`
-						break
-					},
-				}
-			},
-			_ = relay_exited_rx => {
-				// there could be a situation when we're receiving exit signals after we
-				// have already stopped relay or when we have already started new relay.
-				// but it isn't critical, because even if we'll accidentally stop new relay
-				// we'll restart it almost immediately
-				stop_on_demand_headers_relay(active_headers_relay.take()).await;
+			_ = finality_relay_task => {
+				// this should never happen in practice given the current code
+				restart_relay = true;
 			},
 		}
 
@@ -199,33 +181,84 @@ async fn background_task<SourceChain, TargetChain, TargetSign>(
 			continue;
 		}
 
-		// start or stop headers relay if required
-		let action = select_on_demand_relay_action::<SourceChain>(
+		// update required header
+		update_required_header_number_if_too_many_are_missing::<SourceChain>(
 			best_finalized_source_header_at_source.ok(),
 			best_finalized_source_header_at_target.ok(),
-			required_header_number,
 			maximal_headers_difference,
+			&required_header_number,
 			&relay_task_name,
-			active_headers_relay.is_some(),
-		);
-		match action {
-			OnDemandRelayAction::Start => {
-				let (relay_exited_tx, new_relay_exited_rx) = oneshot::channel();
-				active_headers_relay = start_on_demand_headers_relay(
-					relay_task_name.clone(),
-					relay_exited_tx,
-					source_client.clone(),
-					target_client.clone(),
-					pipeline.clone(),
-				);
-				if active_headers_relay.is_some() {
-					relay_exited_rx = new_relay_exited_rx.right_future();
-				}
-			}
-			OnDemandRelayAction::Stop => {
-				stop_on_demand_headers_relay(active_headers_relay.take()).await;
-			}
-			OnDemandRelayAction::None => (),
+		)
+		.await;
+
+		// start/restart relay
+		if restart_relay {
+			finality_relay_task.set(
+				finality_relay::run(
+					finality_source.clone(),
+					finality_target.clone(),
+					FinalitySyncParams {
+						tick: std::cmp::max(SourceChain::AVERAGE_BLOCK_INTERVAL, TargetChain::AVERAGE_BLOCK_INTERVAL),
+						recent_finality_proofs_limit: RECENT_FINALITY_PROOFS_LIMIT,
+						stall_timeout: STALL_TIMEOUT,
+					},
+					MetricsParams::disabled(),
+					futures::future::pending(),
+				)
+				.fuse(),
+			);
+
+			restart_relay = false;
+		}
+	}
+}
+
+/// If there are too many source headers missing at target, we ask for syncing more headers.
+async fn update_required_header_number_if_too_many_are_missing<C: Chain>(
+	best_finalized_source_header_at_source: Option<C::BlockNumber>,
+	best_finalized_source_header_at_target: Option<C::BlockNumber>,
+	maximal_headers_difference: C::BlockNumber,
+	required_header_number: &RequiredHeaderNumberRef<C>,
+	relay_task_name: &str,
+) {
+	let mut required_header_number = required_header_number.lock().await;
+
+	// if we have been unable to read header number from the target, then let's assume
+	// that it is the same as required header number. Otherwise we risk submitting
+	// unneeded transactions
+	let best_finalized_source_header_at_target =
+		best_finalized_source_header_at_target.unwrap_or(*required_header_number);
+
+	// if we have been unable to read header number from the source, then let's assume
+	// that it is the same as at the target
+	let best_finalized_source_header_at_source =
+		best_finalized_source_header_at_source.unwrap_or(best_finalized_source_header_at_target);
+
+	// if there are too many source headers missing from the target node, require some
+	// new headers at target
+	//
+	// why do we need that? When complex headers+messages relay is used, it'll normally only relay
+	// headers when there are undelivered messages/confirmations. But security model of the
+	// `pallet-bridge-grandpa` module relies on the fact that headers are synced in real-time and
+	// that it'll see authorities-change header before unbonding period will end for previous
+	// authorities set.
+	let current_headers_difference = best_finalized_source_header_at_source
+		.checked_sub(&best_finalized_source_header_at_target)
+		.unwrap_or_else(Zero::zero);
+	if current_headers_difference > maximal_headers_difference {
+		// if relay is already asked to sync headers, don't log anything
+		if *required_header_number <= best_finalized_source_header_at_target {
+			log::trace!(
+				target: "bridge",
+				"Too many {} headers missing at target in {} relay ({} vs {}). Going to sync up to the {}",
+				C::NAME,
+				relay_task_name,
+				best_finalized_source_header_at_source,
+				best_finalized_source_header_at_target,
+				best_finalized_source_header_at_source,
+			);
+
+			*required_header_number = best_finalized_source_header_at_source;
 		}
 	}
 }
@@ -236,21 +269,24 @@ async fn background_task<SourceChain, TargetChain, TargetSign>(
 async fn best_finalized_source_header_at_source<SourceChain: Chain, P>(
 	finality_source: &SubstrateFinalitySource<SourceChain, P>,
 	relay_task_name: &str,
-) -> Result<SourceChain::BlockNumber, <SubstrateFinalitySource<SourceChain, P> as RelayClient>::Error>
+) -> Result<SourceChain::BlockNumber, relay_substrate_client::Error>
 where
 	SubstrateFinalitySource<SourceChain, P>: FinalitySourceClient<P>,
 	P: FinalitySyncPipeline<Number = SourceChain::BlockNumber>,
 {
-	finality_source.best_finalized_block_number().await.map_err(|error| {
-		log::error!(
-			target: "bridge",
-			"Failed to read best finalized source header from source in {} relay: {:?}",
-			relay_task_name,
-			error,
-		);
+	finality_source
+		.on_chain_best_finalized_block_number()
+		.await
+		.map_err(|error| {
+			log::error!(
+				target: "bridge",
+				"Failed to read best finalized source header from source in {} relay: {:?}",
+				relay_task_name,
+				error,
+			);
 
-		error
-	})
+			error
+		})
 }
 
 /// Read best finalized source block number from target client.
@@ -279,130 +315,11 @@ where
 		})
 }
 
-/// What to do with the on-demand relay task?
-#[derive(Debug, PartialEq)]
-enum OnDemandRelayAction {
-	Start,
-	Stop,
-	None,
-}
-
-fn select_on_demand_relay_action<C: Chain>(
-	best_finalized_source_header_at_source: Option<C::BlockNumber>,
-	best_finalized_source_header_at_target: Option<C::BlockNumber>,
-	mut required_source_header_at_target: C::BlockNumber,
-	maximal_headers_difference: C::BlockNumber,
-	relay_task_name: &str,
-	is_active: bool,
-) -> OnDemandRelayAction {
-	// if we have been unable to read header number from the target, then let's assume
-	// that it is the same as required header number. Otherwise we risk submitting
-	// unneeded transactions
-	let best_finalized_source_header_at_target =
-		best_finalized_source_header_at_target.unwrap_or(required_source_header_at_target);
-
-	// if we have been unable to read header number from the source, then let's assume
-	// that it is the same as at the target
-	let best_finalized_source_header_at_source =
-		best_finalized_source_header_at_source.unwrap_or(best_finalized_source_header_at_target);
-
-	// if there are too many source headers missing from the target node, require some
-	// new headers at target
-	//
-	// why do we need that? When complex headers+messages relay is used, it'll normally only relay
-	// headers when there are undelivered messages/confirmations. But security model of the
-	// `pallet-bridge-grandpa` module relies on the fact that headers are synced in real-time and
-	// that it'll see authorities-change header before unbonding period will end for previous
-	// authorities set.
-	let current_headers_difference = best_finalized_source_header_at_source
-		.checked_sub(&best_finalized_source_header_at_target)
-		.unwrap_or_else(Zero::zero);
-	if current_headers_difference > maximal_headers_difference {
-		required_source_header_at_target = best_finalized_source_header_at_source;
-
-		// don't log if relay is already running
-		if !is_active {
-			log::trace!(
-				target: "bridge",
-				"Too many {} headers missing at target in {} relay ({} vs {}). Going to sync up to the {}",
-				C::NAME,
-				relay_task_name,
-				best_finalized_source_header_at_source,
-				best_finalized_source_header_at_target,
-				best_finalized_source_header_at_source,
-			);
-		}
-	}
-
-	// now let's select what to do with relay
-	let needs_to_be_active = required_source_header_at_target > best_finalized_source_header_at_target;
-	match (needs_to_be_active, is_active) {
-		(true, false) => OnDemandRelayAction::Start,
-		(false, true) => OnDemandRelayAction::Stop,
-		_ => OnDemandRelayAction::None,
-	}
-}
-
 /// On-demand headers relay task name.
 fn on_demand_headers_relay_name<SourceChain: Chain, TargetChain: Chain>() -> String {
 	format!("on-demand-{}-to-{}", SourceChain::NAME, TargetChain::NAME)
 }
 
-/// Start on-demand headers relay task.
-fn start_on_demand_headers_relay<SourceChain: Chain, TargetChain: Chain, TargetSign>(
-	task_name: String,
-	relay_exited_tx: oneshot::Sender<()>,
-	source_client: Client<SourceChain>,
-	target_client: Client<TargetChain>,
-	pipeline: SubstrateFinalityToSubstrate<SourceChain, TargetChain, TargetSign>,
-) -> Option<async_std::task::JoinHandle<()>>
-where
-	SourceChain::BlockNumber: BlockNumberBase,
-	SubstrateFinalityToSubstrate<SourceChain, TargetChain, TargetSign>: SubstrateFinalitySyncPipeline<
-		Hash = HashOf<SourceChain>,
-		Number = BlockNumberOf<SourceChain>,
-		Header = SyncHeader<SourceChain::Header>,
-		FinalityProof = GrandpaJustification<SourceChain::Header>,
-		TargetChain = TargetChain,
-	>,
-	TargetSign: 'static,
-{
-	let headers_relay_future =
-		crate::finality_pipeline::run(pipeline, source_client, target_client, true, MetricsParams::disabled());
-	let closure_task_name = task_name.clone();
-	async_std::task::Builder::new()
-		.name(task_name.clone())
-		.spawn(async move {
-			log::info!(target: "bridge", "Starting {} headers relay", closure_task_name);
-			let result = headers_relay_future.await;
-			log::trace!(target: "bridge", "{} headers relay has exited. Result: {:?}", closure_task_name, result);
-			let _ = relay_exited_tx.send(());
-		})
-		.map_err(|error| {
-			log::error!(
-				target: "bridge",
-				"Failed to start {} relay: {:?}",
-				task_name,
-				error,
-			);
-		})
-		.ok()
-}
-
-/// Stop on-demand headers relay task.
-async fn stop_on_demand_headers_relay(task: Option<async_std::task::JoinHandle<()>>) {
-	if let Some(task) = task {
-		let task_name = task
-			.task()
-			.name()
-			.expect("on-demand tasks are always started with name; qed")
-			.to_string();
-		log::trace!(target: "bridge", "Cancelling {} headers relay", task_name);
-		task.cancel().await;
-		log::info!(target: "bridge", "Cancelled {} headers relay", task_name);
-	}
-}
-
 #[cfg(test)]
 mod tests {
 	use super::*;
@@ -412,42 +329,17 @@ mod tests {
 	const AT_SOURCE: Option<bp_millau::BlockNumber> = Some(10);
 	const AT_TARGET: Option<bp_millau::BlockNumber> = Some(1);
 
-	#[test]
-	fn starts_relay_when_headers_are_required() {
-		assert_eq!(
-			select_on_demand_relay_action::<TestChain>(AT_SOURCE, AT_TARGET, 5, 100, "test", false),
-			OnDemandRelayAction::Start,
-		);
-
-		assert_eq!(
-			select_on_demand_relay_action::<TestChain>(AT_SOURCE, AT_TARGET, 5, 100, "test", true),
-			OnDemandRelayAction::None,
-		);
-	}
-
-	#[test]
-	fn starts_relay_when_too_many_headers_missing() {
-		assert_eq!(
-			select_on_demand_relay_action::<TestChain>(AT_SOURCE, AT_TARGET, 0, 5, "test", false),
-			OnDemandRelayAction::Start,
-		);
-
-		assert_eq!(
-			select_on_demand_relay_action::<TestChain>(AT_SOURCE, AT_TARGET, 0, 5, "test", true),
-			OnDemandRelayAction::None,
-		);
-	}
-
-	#[test]
-	fn stops_relay_if_required_header_is_synced() {
-		assert_eq!(
-			select_on_demand_relay_action::<TestChain>(AT_SOURCE, AT_TARGET, AT_TARGET.unwrap(), 100, "test", true),
-			OnDemandRelayAction::Stop,
-		);
-
-		assert_eq!(
-			select_on_demand_relay_action::<TestChain>(AT_SOURCE, AT_TARGET, AT_TARGET.unwrap(), 100, "test", false),
-			OnDemandRelayAction::None,
-		);
+	#[async_std::test]
+	async fn updates_required_header_when_too_many_headers_missing() {
+		let required_header_number = Arc::new(Mutex::new(0));
+		update_required_header_number_if_too_many_are_missing::<TestChain>(
+			AT_SOURCE,
+			AT_TARGET,
+			5,
+			&required_header_number,
+			"test",
+		)
+		.await;
+		assert_eq!(*required_header_number.lock().await, AT_SOURCE.unwrap());
 	}
 }