use crate::{error::Error, Chain, Client};
use async_trait::async_trait;
use sp_version::RuntimeVersion;
use std::{
fmt::Display,
time::{Duration, Instant},
};
#[async_trait]
pub trait Environment<C>: Send + Sync + 'static {
type Error: Display + Send + Sync + 'static;
async fn runtime_version(&mut self) -> Result<RuntimeVersion, Self::Error>;
fn now(&self) -> Instant {
Instant::now()
}
async fn sleep(&mut self, duration: Duration) {
async_std::task::sleep(duration).await
}
async fn abort(&mut self) {
std::process::abort();
}
}
pub fn abort_on_spec_version_change<C: Chain>(
mut env: impl Environment<C>,
expected_spec_version: u32,
) {
async_std::task::spawn(async move {
log::info!(
target: "bridge-guard",
"Starting spec_version guard for {}. Expected spec_version: {}",
C::NAME,
expected_spec_version,
);
loop {
let actual_spec_version = env.runtime_version().await;
match actual_spec_version {
Ok(version) if version.spec_version == expected_spec_version => (),
Ok(version) => {
log::error!(
target: "bridge-guard",
"{} runtime spec version has changed from {} to {}. Aborting relay",
C::NAME,
expected_spec_version,
version.spec_version,
);
env.abort().await;
},
Err(error) => log::warn!(
target: "bridge-guard",
"Failed to read {} runtime version: {}. Relay may need to be stopped manually",
C::NAME,
error,
),
}
env.sleep(conditions_check_delay::<C>()).await;
}
});
}
fn conditions_check_delay<C: Chain>() -> Duration {
C::AVERAGE_BLOCK_INTERVAL * (10 + rand::random::<u32>() % 10)
}
#[async_trait]
impl<C: Chain, Clnt: Client<C>> Environment<C> for Clnt {
type Error = Error;
async fn runtime_version(&mut self) -> Result<RuntimeVersion, Self::Error> {
Client::<C>::runtime_version(self).await
}
}
#[cfg(test)]
pub(crate) mod tests {
use super::*;
use crate::test_chain::TestChain;
use futures::{
channel::mpsc::{unbounded, UnboundedReceiver, UnboundedSender},
future::FutureExt,
stream::StreamExt,
SinkExt,
};
pub struct TestEnvironment {
pub runtime_version_rx: UnboundedReceiver<RuntimeVersion>,
pub slept_tx: UnboundedSender<()>,
pub aborted_tx: UnboundedSender<()>,
}
#[async_trait]
impl Environment<TestChain> for TestEnvironment {
type Error = Error;
async fn runtime_version(&mut self) -> Result<RuntimeVersion, Self::Error> {
Ok(self.runtime_version_rx.next().await.unwrap_or_default())
}
async fn sleep(&mut self, _duration: Duration) {
let _ = self.slept_tx.send(()).await;
}
async fn abort(&mut self) {
let _ = self.aborted_tx.send(()).await;
async_std::task::sleep(Duration::from_secs(60)).await;
}
}
#[test]
fn aborts_when_spec_version_is_changed() {
async_std::task::block_on(async {
let (
(mut runtime_version_tx, runtime_version_rx),
(slept_tx, mut slept_rx),
(aborted_tx, mut aborted_rx),
) = (unbounded(), unbounded(), unbounded());
abort_on_spec_version_change(
TestEnvironment { runtime_version_rx, slept_tx, aborted_tx },
0,
);
runtime_version_tx
.send(RuntimeVersion { spec_version: 42, ..Default::default() })
.await
.unwrap();
aborted_rx.next().await;
assert!(slept_rx.next().now_or_never().is_none());
});
}
#[test]
fn does_not_aborts_when_spec_version_is_unchanged() {
async_std::task::block_on(async {
let (
(mut runtime_version_tx, runtime_version_rx),
(slept_tx, mut slept_rx),
(aborted_tx, mut aborted_rx),
) = (unbounded(), unbounded(), unbounded());
abort_on_spec_version_change(
TestEnvironment { runtime_version_rx, slept_tx, aborted_tx },
42,
);
runtime_version_tx
.send(RuntimeVersion { spec_version: 42, ..Default::default() })
.await
.unwrap();
slept_rx.next().await;
assert!(aborted_rx.next().now_or_never().is_none());
});
}
}