Merge commit 'f9c24ef0db390c355241445af37a5c7999a7dc66' into hc-bump-bridges-subtree-take-2

polkadot/bridges/relays/utils/src/initialize.rs

@@ -16,7 +16,11 @@
 
 //! Relayer initialization functions.
 
-use std::{fmt::Display, io::Write};
+use std::{cell::RefCell, fmt::Display, io::Write};
+
+async_std::task_local! {
+	pub(crate) static LOOP_NAME: RefCell<String> = RefCell::new(String::default());
+}
 
 /// Initialize relay environment.
 pub fn initialize_relay() {
@@ -43,20 +47,56 @@ pub fn initialize_logger(with_timestamp: bool) {
 				Either::Right(ansi_term::Colour::Fixed(8).bold().paint(timestamp))
 			};
 
-			writeln!(buf, "{} {} {} {}", timestamp, log_level, log_target, record.args(),)
+			writeln!(
+				buf,
+				"{}{} {} {} {}",
+				loop_name_prefix(),
+				timestamp,
+				log_level,
+				log_target,
+				record.args(),
+			)
 		});
 	} else {
 		builder.format(move |buf, record| {
 			let log_level = color_level(record.level());
 			let log_target = color_target(record.target());
 
-			writeln!(buf, "{} {} {}", log_level, log_target, record.args(),)
+			writeln!(
+				buf,
+				"{}{} {} {}",
+				loop_name_prefix(),
+				log_level,
+				log_target,
+				record.args(),
+			)
 		});
 	}
 
 	builder.init();
 }
 
+/// Initialize relay loop. Must only be called once per every loop task.
+pub(crate) fn initialize_loop(loop_name: String) {
+	LOOP_NAME.with(|g_loop_name| *g_loop_name.borrow_mut() = loop_name);
+}
+
+/// Returns loop name prefix to use in logs. The prefix is initialized with the `initialize_loop` call.
+fn loop_name_prefix() -> String {
+	// try_with to avoid panic outside of async-std task context
+	LOOP_NAME
+		.try_with(|loop_name| {
+			// using borrow is ok here, because loop is only initialized once (=> borrow_mut will only be called once)
+			let loop_name = loop_name.borrow();
+			if loop_name.is_empty() {
+				String::new()
+			} else {
+				format!("[{}] ", loop_name)
+			}
+		})
+		.unwrap_or_else(|_| String::new())
+}
+
 enum Either<A, B> {
 	Left(A),
 	Right(B),

polkadot/bridges/relays/utils/src/relay_loop.rs

@@ -26,9 +26,9 @@ pub const RECONNECT_DELAY: Duration = Duration::from_secs(10);
 
 /// Basic blockchain client from relay perspective.
 #[async_trait]
-pub trait Client: Clone + Send + Sync {
+pub trait Client: 'static + Clone + Send + Sync {
 	/// Type of error this clients returns.
-	type Error: Debug + MaybeConnectionError;
+	type Error: 'static + Debug + MaybeConnectionError + Send + Sync;
 
 	/// Try to reconnect to source node.
 	async fn reconnect(&mut self) -> Result<(), Self::Error>;
@@ -38,6 +38,7 @@ pub trait Client: Clone + Send + Sync {
 pub fn relay_loop<SC, TC>(source_client: SC, target_client: TC) -> Loop<SC, TC, ()> {
 	Loop {
 		reconnect_delay: RECONNECT_DELAY,
+		spawn_loop_task: true,
 		source_client,
 		target_client,
 		loop_metric: None,
@@ -49,6 +50,7 @@ pub fn relay_metrics(prefix: Option<String>, params: MetricsParams) -> LoopMetri
 	LoopMetrics {
 		relay_loop: Loop {
 			reconnect_delay: RECONNECT_DELAY,
+			spawn_loop_task: true,
 			source_client: (),
 			target_client: (),
 			loop_metric: None,
@@ -63,6 +65,7 @@ pub fn relay_metrics(prefix: Option<String>, params: MetricsParams) -> LoopMetri
 /// Generic relay loop.
 pub struct Loop<SC, TC, LM> {
 	reconnect_delay: Duration,
+	spawn_loop_task: bool,
 	source_client: SC,
 	target_client: TC,
 	loop_metric: Option<LM>,
@@ -84,11 +87,23 @@ impl<SC, TC, LM> Loop<SC, TC, LM> {
 		self
 	}
 
+	/// Set spawn-dedicated-loop-task flag.
+	///
+	/// If `true` (default), separate async task is spawned to run relay loop. This is the default
+	/// behavior for all loops. If `false`, then loop is executed as a part of the current
+	/// task. The `false` is used for on-demand tasks, which are cancelled from time to time
+	/// and there's already a dedicated on-demand task for running such loops.
+	pub fn spawn_loop_task(mut self, spawn_loop_task: bool) -> Self {
+		self.spawn_loop_task = spawn_loop_task;
+		self
+	}
+
 	/// Start building loop metrics using given prefix.
 	pub fn with_metrics(self, prefix: Option<String>, params: MetricsParams) -> LoopMetrics<SC, TC, ()> {
 		LoopMetrics {
 			relay_loop: Loop {
 				reconnect_delay: self.reconnect_delay,
+				spawn_loop_task: self.spawn_loop_task,
 				source_client: self.source_client,
 				target_client: self.target_client,
 				loop_metric: None,
@@ -105,63 +120,47 @@ impl<SC, TC, LM> Loop<SC, TC, LM> {
 	/// This function represents an outer loop, which in turn calls provided `run_loop` function to do
 	/// actual job. When `run_loop` returns, this outer loop reconnects to failed client (source,
 	/// target or both) and calls `run_loop` again.
-	pub async fn run<R, F>(mut self, run_loop: R) -> Result<(), String>
+	pub async fn run<R, F>(mut self, loop_name: String, run_loop: R) -> Result<(), String>
 	where
-		R: Fn(SC, TC, Option<LM>) -> F,
-		F: Future<Output = Result<(), FailedClient>>,
-		SC: Client,
-		TC: Client,
-		LM: Clone,
+		R: 'static + Send + Fn(SC, TC, Option<LM>) -> F,
+		F: 'static + Send + Future<Output = Result<(), FailedClient>>,
+		SC: 'static + Client,
+		TC: 'static + Client,
+		LM: 'static + Send + Clone,
 	{
-		loop {
-			let result = run_loop(
-				self.source_client.clone(),
-				self.target_client.clone(),
-				self.loop_metric.clone(),
-			)
-			.await;
+		let spawn_loop_task = self.spawn_loop_task;
+		let run_loop_task = async move {
+			crate::initialize::initialize_loop(loop_name);
 
-			match result {
-				Ok(()) => break,
-				Err(failed_client) => loop {
-					async_std::task::sleep(self.reconnect_delay).await;
-					if failed_client == FailedClient::Both || failed_client == FailedClient::Source {
-						match self.source_client.reconnect().await {
-							Ok(()) => (),
-							Err(error) => {
-								log::warn!(
-									target: "bridge",
-									"Failed to reconnect to source client. Going to retry in {}s: {:?}",
-									self.reconnect_delay.as_secs(),
-									error,
-								);
-								continue;
-							}
-						}
-					}
-					if failed_client == FailedClient::Both || failed_client == FailedClient::Target {
-						match self.target_client.reconnect().await {
-							Ok(()) => (),
-							Err(error) => {
-								log::warn!(
-									target: "bridge",
-									"Failed to reconnect to target client. Going to retry in {}s: {:?}",
-									self.reconnect_delay.as_secs(),
-									error,
-								);
-								continue;
-							}
-						}
-					}
+			loop {
+				let loop_metric = self.loop_metric.clone();
+				let future_result = run_loop(self.source_client.clone(), self.target_client.clone(), loop_metric);
+				let result = future_result.await;
 
-					break;
-				},
+				match result {
+					Ok(()) => break,
+					Err(failed_client) => {
+						reconnect_failed_client(
+							failed_client,
+							self.reconnect_delay,
+							&mut self.source_client,
+							&mut self.target_client,
+						)
+						.await
+					}
+				}
+
+				log::debug!(target: "bridge", "Restarting relay loop");
 			}
 
-			log::debug!(target: "bridge", "Restarting relay loop");
-		}
+			Ok(())
+		};
 
-		Ok(())
+		if spawn_loop_task {
+			async_std::task::spawn(run_loop_task).await
+		} else {
+			run_loop_task.await
+		}
 	}
 }
 
@@ -237,6 +236,7 @@ impl<SC, TC, LM> LoopMetrics<SC, TC, LM> {
 
 		Ok(Loop {
 			reconnect_delay: self.relay_loop.reconnect_delay,
+			spawn_loop_task: self.relay_loop.spawn_loop_task,
 			source_client: self.relay_loop.source_client,
 			target_client: self.relay_loop.target_client,
 			loop_metric: self.loop_metric,
@@ -244,6 +244,48 @@ impl<SC, TC, LM> LoopMetrics<SC, TC, LM> {
 	}
 }
 
+/// Deal with the client who has returned connection error.
+pub async fn reconnect_failed_client(
+	failed_client: FailedClient,
+	reconnect_delay: Duration,
+	source_client: &mut impl Client,
+	target_client: &mut impl Client,
+) {
+	loop {
+		async_std::task::sleep(reconnect_delay).await;
+		if failed_client == FailedClient::Both || failed_client == FailedClient::Source {
+			match source_client.reconnect().await {
+				Ok(()) => (),
+				Err(error) => {
+					log::warn!(
+						target: "bridge",
+						"Failed to reconnect to source client. Going to retry in {}s: {:?}",
+						reconnect_delay.as_secs(),
+						error,
+					);
+					continue;
+				}
+			}
+		}
+		if failed_client == FailedClient::Both || failed_client == FailedClient::Target {
+			match target_client.reconnect().await {
+				Ok(()) => (),
+				Err(error) => {
+					log::warn!(
+						target: "bridge",
+						"Failed to reconnect to target client. Going to retry in {}s: {:?}",
+						reconnect_delay.as_secs(),
+						error,
+					);
+					continue;
+				}
+			}
+		}
+
+		break;
+	}
+}
+
 /// Create new registry with global metrics.
 fn create_metrics_registry(prefix: Option<String>) -> Registry {
 	match prefix {