pezkuwi-subxt/bridges/relays/utils/src/lib.rs

// Copyright 2019-2021 Parity Technologies (UK) Ltd.
// This file is part of Parity Bridges Common.

// Parity Bridges Common is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.

// Parity Bridges Common is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.

// You should have received a copy of the GNU General Public License
// along with Parity Bridges Common.  If not, see <http://www.gnu.org/licenses/>.

//! Utilities used by different relays.

pub use bp_runtime::HeaderId;
pub use error::Error;
pub use relay_loop::{relay_loop, relay_metrics};
pub use sp_runtime::traits::{UniqueSaturatedFrom, UniqueSaturatedInto};
use std::fmt::Debug;

use async_trait::async_trait;
use backoff::{backoff::Backoff, ExponentialBackoff};
use futures::future::{BoxFuture, FutureExt};
use std::time::Duration;
use thiserror::Error;

/// Default relay loop stall timeout. If transactions generated by relay are immortal, then
/// this timeout is used.
///
/// There are no any strict requirements on block time in Substrate. But we assume here that all
/// Substrate-based chains will be designed to produce relatively fast (compared to the slowest
/// blockchains) blocks. So 1 hour seems to be a good guess for (even congested) chains to mine
/// transaction, or remove it from the pool.
pub const STALL_TIMEOUT: Duration = Duration::from_secs(60 * 60);

/// Max delay after connection-unrelated error happened before we'll try the
/// same request again.
pub const MAX_BACKOFF_INTERVAL: Duration = Duration::from_secs(60);
/// Delay after connection-related error happened before we'll try
/// reconnection again.
pub const CONNECTION_ERROR_DELAY: Duration = Duration::from_secs(10);

pub mod error;
pub mod initialize;
pub mod metrics;
pub mod relay_loop;

/// Block number traits shared by all chains that relay is able to serve.
pub trait BlockNumberBase:
	'static
	+ From<u32>
	+ UniqueSaturatedInto<u64>
	+ Ord
	+ Clone
	+ Copy
	+ Default
	+ Send
	+ Sync
	+ std::fmt::Debug
	+ std::fmt::Display
	+ std::hash::Hash
	+ std::ops::Add<Output = Self>
	+ std::ops::Sub<Output = Self>
	+ num_traits::CheckedSub
	+ num_traits::Saturating
	+ num_traits::Zero
	+ num_traits::One
{
}

impl<T> BlockNumberBase for T where
	T: 'static
		+ From<u32>
		+ UniqueSaturatedInto<u64>
		+ Ord
		+ Clone
		+ Copy
		+ Default
		+ Send
		+ Sync
		+ std::fmt::Debug
		+ std::fmt::Display
		+ std::hash::Hash
		+ std::ops::Add<Output = Self>
		+ std::ops::Sub<Output = Self>
		+ num_traits::CheckedSub
		+ num_traits::Saturating
		+ num_traits::Zero
		+ num_traits::One
{
}

/// Macro that returns (client, Err(error)) tuple from function if result is Err(error).
#[macro_export]
macro_rules! bail_on_error {
	($result: expr) => {
		match $result {
			(client, Ok(result)) => (client, result),
			(client, Err(error)) => return (client, Err(error)),
		}
	};
}

/// Macro that returns (client, Err(error)) tuple from function if result is Err(error).
#[macro_export]
macro_rules! bail_on_arg_error {
	($result: expr, $client: ident) => {
		match $result {
			Ok(result) => result,
			Err(error) => return ($client, Err(error)),
		}
	};
}

/// Error type that can signal connection errors.
pub trait MaybeConnectionError {
	/// Returns true if error (maybe) represents connection error.
	fn is_connection_error(&self) -> bool;
}

/// Final status of the tracked transaction.
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum TrackedTransactionStatus<BlockId> {
	/// Transaction has been lost.
	Lost,
	/// Transaction has been mined and finalized at given block.
	Finalized(BlockId),
}

/// Transaction tracker.
#[async_trait]
pub trait TransactionTracker: Send {
	/// Header id, used by the chain.
	type HeaderId: Clone + Debug + Send;

	/// Wait until transaction is either finalized or invalidated/lost.
	async fn wait(self) -> TrackedTransactionStatus<Self::HeaderId>;
}

/// Future associated with `TransactionTracker`, monitoring the transaction status.
pub type TrackedTransactionFuture<T> =
	BoxFuture<'static, TrackedTransactionStatus<<T as TransactionTracker>::HeaderId>>;

/// Stringified error that may be either connection-related or not.
#[derive(Error, Debug)]
pub enum StringifiedMaybeConnectionError {
	/// The error is connection-related error.
	#[error("{0}")]
	Connection(String),
	/// The error is connection-unrelated error.
	#[error("{0}")]
	NonConnection(String),
}

impl StringifiedMaybeConnectionError {
	/// Create new stringified connection error.
	pub fn new(is_connection_error: bool, error: String) -> Self {
		if is_connection_error {
			StringifiedMaybeConnectionError::Connection(error)
		} else {
			StringifiedMaybeConnectionError::NonConnection(error)
		}
	}
}

impl MaybeConnectionError for StringifiedMaybeConnectionError {
	fn is_connection_error(&self) -> bool {
		match *self {
			StringifiedMaybeConnectionError::Connection(_) => true,
			StringifiedMaybeConnectionError::NonConnection(_) => false,
		}
	}
}

/// Exponential backoff for connection-unrelated errors retries.
pub fn retry_backoff() -> ExponentialBackoff {
	ExponentialBackoff {
		// we do not want relayer to stop
		max_elapsed_time: None,
		max_interval: MAX_BACKOFF_INTERVAL,
		..Default::default()
	}
}

/// Compact format of IDs vector.
pub fn format_ids<Id: std::fmt::Debug>(mut ids: impl ExactSizeIterator<Item = Id>) -> String {
	const NTH_PROOF: &str = "we have checked len; qed";
	match ids.len() {
		0 => "<nothing>".into(),
		1 => format!("{:?}", ids.next().expect(NTH_PROOF)),
		2 => {
			let id0 = ids.next().expect(NTH_PROOF);
			let id1 = ids.next().expect(NTH_PROOF);
			format!("[{id0:?}, {id1:?}]")
		},
		len => {
			let id0 = ids.next().expect(NTH_PROOF);
			let id_last = ids.last().expect(NTH_PROOF);
			format!("{len}:[{id0:?} ... {id_last:?}]")
		},
	}
}

/// Stream that emits item every `timeout_ms` milliseconds.
pub fn interval(timeout: Duration) -> impl futures::Stream<Item = ()> {
	futures::stream::unfold((), move |_| async move {
		async_std::task::sleep(timeout).await;
		Some(((), ()))
	})
}

/// Which client has caused error.
#[derive(Debug, Eq, Clone, Copy, PartialEq)]
pub enum FailedClient {
	/// It is the source client who has caused error.
	Source,
	/// It is the target client who has caused error.
	Target,
	/// Both clients are failing, or we just encountered some other error that
	/// should be treated like that.
	Both,
}

/// Future process result.
#[derive(Debug, Clone, Copy)]
pub enum ProcessFutureResult {
	/// Future has been processed successfully.
	Success,
	/// Future has failed with non-connection error.
	Failed,
	/// Future has failed with connection error.
	ConnectionFailed,
}

impl ProcessFutureResult {
	/// Returns true if result is Success.
	pub fn is_ok(self) -> bool {
		match self {
			ProcessFutureResult::Success => true,
			ProcessFutureResult::Failed | ProcessFutureResult::ConnectionFailed => false,
		}
	}

	/// Returns `Ok(())` if future has succeeded.
	/// Returns `Err(failed_client)` otherwise.
	pub fn fail_if_error(self, failed_client: FailedClient) -> Result<(), FailedClient> {
		if self.is_ok() {
			Ok(())
		} else {
			Err(failed_client)
		}
	}

	/// Returns Ok(true) if future has succeeded.
	/// Returns Ok(false) if future has failed with non-connection error.
	/// Returns Err if future is `ConnectionFailed`.
	pub fn fail_if_connection_error(
		self,
		failed_client: FailedClient,
	) -> Result<bool, FailedClient> {
		match self {
			ProcessFutureResult::Success => Ok(true),
			ProcessFutureResult::Failed => Ok(false),
			ProcessFutureResult::ConnectionFailed => Err(failed_client),
		}
	}
}

/// Process result of the future from a client.
pub fn process_future_result<TResult, TError, TGoOfflineFuture>(
	result: Result<TResult, TError>,
	retry_backoff: &mut ExponentialBackoff,
	on_success: impl FnOnce(TResult),
	go_offline_future: &mut std::pin::Pin<&mut futures::future::Fuse<TGoOfflineFuture>>,
	go_offline: impl FnOnce(Duration) -> TGoOfflineFuture,
	error_pattern: impl FnOnce() -> String,
) -> ProcessFutureResult
where
	TError: std::fmt::Debug + MaybeConnectionError,
	TGoOfflineFuture: FutureExt,
{
	match result {
		Ok(result) => {
			on_success(result);
			retry_backoff.reset();
			ProcessFutureResult::Success
		},
		Err(error) if error.is_connection_error() => {
			log::error!(
				target: "bridge",
				"{}: {:?}. Going to restart",
				error_pattern(),
				error,
			);

			retry_backoff.reset();
			go_offline_future.set(go_offline(CONNECTION_ERROR_DELAY).fuse());
			ProcessFutureResult::ConnectionFailed
		},
		Err(error) => {
			let retry_delay = retry_backoff.next_backoff().unwrap_or(CONNECTION_ERROR_DELAY);
			log::error!(
				target: "bridge",
				"{}: {:?}. Retrying in {}",
				error_pattern(),
				error,
				retry_delay.as_secs_f64(),
			);

			go_offline_future.set(go_offline(retry_delay).fuse());
			ProcessFutureResult::Failed
		},
	}
}