|
|
|
|
@ -1,7 +1,10 @@ |
|
|
|
|
use async_trait::async_trait; |
|
|
|
|
use color_eyre::{eyre::eyre, Result}; |
|
|
|
|
use color_eyre::{eyre::bail, Result}; |
|
|
|
|
use futures_util::future::select_all; |
|
|
|
|
use std::sync::Arc; |
|
|
|
|
use tokio::{ |
|
|
|
|
join, |
|
|
|
|
sync::Mutex, |
|
|
|
|
task::JoinHandle, |
|
|
|
|
time::{interval, Interval}, |
|
|
|
|
}; |
|
|
|
|
@ -16,67 +19,140 @@ use optics_core::traits::{Common, Replica}; |
|
|
|
|
|
|
|
|
|
use crate::settings::Settings; |
|
|
|
|
|
|
|
|
|
/// A relayer agent
|
|
|
|
|
#[derive(Debug)] |
|
|
|
|
pub struct Relayer { |
|
|
|
|
interval_seconds: u64, |
|
|
|
|
core: AgentCore, |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
impl AsRef<AgentCore> for Relayer { |
|
|
|
|
fn as_ref(&self) -> &AgentCore { |
|
|
|
|
&self.core |
|
|
|
|
} |
|
|
|
|
struct UpdatePoller { |
|
|
|
|
home: Arc<Homes>, |
|
|
|
|
replica: Arc<Replicas>, |
|
|
|
|
interval: Interval, |
|
|
|
|
semaphore: Mutex<()>, |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
#[allow(clippy::unit_arg)] |
|
|
|
|
impl Relayer { |
|
|
|
|
/// Instantiate a new relayer
|
|
|
|
|
pub fn new(interval_seconds: u64, core: AgentCore) -> Self { |
|
|
|
|
impl UpdatePoller { |
|
|
|
|
fn new(home: Arc<Homes>, replica: Arc<Replicas>, interval: Interval) -> Self { |
|
|
|
|
Self { |
|
|
|
|
interval_seconds, |
|
|
|
|
core, |
|
|
|
|
home, |
|
|
|
|
replica, |
|
|
|
|
interval, |
|
|
|
|
semaphore: Mutex::new(()), |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
#[tracing::instrument(err)] |
|
|
|
|
async fn poll_and_relay_update(home: Arc<Homes>, replica: Arc<Replicas>) -> Result<()> { |
|
|
|
|
// Get replica's current root
|
|
|
|
|
let old_root = replica.current_root().await?; |
|
|
|
|
async fn poll_and_relay_update(&self) -> Result<()> { |
|
|
|
|
// Get replica's current root.
|
|
|
|
|
// If the replica has a queue of pending updates, we use the last queue
|
|
|
|
|
// root instead
|
|
|
|
|
let (old_root_res, queue_end_res) = |
|
|
|
|
join!(self.replica.current_root(), self.replica.queue_end()); |
|
|
|
|
|
|
|
|
|
let old_root = { |
|
|
|
|
if let Some(end) = queue_end_res? { |
|
|
|
|
end |
|
|
|
|
} else { |
|
|
|
|
old_root_res? |
|
|
|
|
} |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
// Check for first signed update building off of the replica's current root
|
|
|
|
|
let signed_update_opt = home.signed_update_by_old_root(old_root).await?; |
|
|
|
|
let signed_update_opt = self.home.signed_update_by_old_root(old_root).await?; |
|
|
|
|
|
|
|
|
|
// If signed update exists, update replica's current root
|
|
|
|
|
if let Some(signed_update) = signed_update_opt { |
|
|
|
|
info!( |
|
|
|
|
"Have a signed update, dispatching to replica {}", |
|
|
|
|
replica.name() |
|
|
|
|
"Update for replica {}. Root {:?} to {:?}", |
|
|
|
|
self.replica.name(), |
|
|
|
|
&signed_update.update.previous_root, |
|
|
|
|
&signed_update.update.new_root, |
|
|
|
|
); |
|
|
|
|
replica.update(&signed_update).await?; |
|
|
|
|
|
|
|
|
|
let lock = self.semaphore.try_lock(); |
|
|
|
|
if lock.is_err() { |
|
|
|
|
return Ok(()); // tx in flight. just do nothing
|
|
|
|
|
} |
|
|
|
|
self.replica.update(&signed_update).await?; |
|
|
|
|
// lock dropped here
|
|
|
|
|
} else { |
|
|
|
|
info!("No update."); |
|
|
|
|
info!( |
|
|
|
|
"No update. Current root for replica {} is {:?}", |
|
|
|
|
self.replica.name(), |
|
|
|
|
old_root |
|
|
|
|
); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
Ok(()) |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
fn spawn(mut self) -> JoinHandle<Result<()>> { |
|
|
|
|
tokio::spawn(async move { |
|
|
|
|
loop { |
|
|
|
|
self.poll_and_relay_update().await?; |
|
|
|
|
self.interval.tick().await; |
|
|
|
|
} |
|
|
|
|
}) |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
#[derive(Debug)] |
|
|
|
|
struct ConfirmPoller { |
|
|
|
|
replica: Arc<Replicas>, |
|
|
|
|
interval: Interval, |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
impl ConfirmPoller { |
|
|
|
|
fn new(replica: Arc<Replicas>, interval: Interval) -> Self { |
|
|
|
|
Self { replica, interval } |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
#[tracing::instrument(err)] |
|
|
|
|
async fn poll_confirm(replica: Arc<Replicas>) -> Result<()> { |
|
|
|
|
async fn poll_confirm(&self) -> Result<()> { |
|
|
|
|
// Check for pending update that can be confirmed
|
|
|
|
|
let can_confirm = replica.can_confirm().await?; |
|
|
|
|
let can_confirm = self.replica.can_confirm().await?; |
|
|
|
|
|
|
|
|
|
// If valid pending update exists, confirm it
|
|
|
|
|
if can_confirm { |
|
|
|
|
info!("Can confirm. Confirming on replica {}", replica.name()); |
|
|
|
|
replica.confirm().await?; |
|
|
|
|
info!("Can confirm. Confirming on replica {}", self.replica.name()); |
|
|
|
|
// don't care if it succeeds
|
|
|
|
|
let _ = self.replica.confirm().await; |
|
|
|
|
} else { |
|
|
|
|
info!("Can't confirm"); |
|
|
|
|
info!("Can't confirm on replica {}", self.replica.name()); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
Ok(()) |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
fn spawn(mut self) -> JoinHandle<Result<()>> { |
|
|
|
|
tokio::spawn(async move { |
|
|
|
|
loop { |
|
|
|
|
self.poll_confirm().await?; |
|
|
|
|
self.interval.tick().await; |
|
|
|
|
} |
|
|
|
|
}) |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
/// A relayer agent
|
|
|
|
|
#[derive(Debug)] |
|
|
|
|
pub struct Relayer { |
|
|
|
|
interval_seconds: u64, |
|
|
|
|
core: AgentCore, |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
impl AsRef<AgentCore> for Relayer { |
|
|
|
|
fn as_ref(&self) -> &AgentCore { |
|
|
|
|
&self.core |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
#[allow(clippy::unit_arg)] |
|
|
|
|
impl Relayer { |
|
|
|
|
/// Instantiate a new relayer
|
|
|
|
|
pub fn new(interval_seconds: u64, core: AgentCore) -> Self { |
|
|
|
|
Self { |
|
|
|
|
interval_seconds, |
|
|
|
|
core, |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
fn interval(&self) -> Interval { |
|
|
|
|
interval(std::time::Duration::from_secs(self.interval_seconds)) |
|
|
|
|
} |
|
|
|
|
@ -101,130 +177,130 @@ impl OpticsAgent for Relayer { |
|
|
|
|
fn run(&self, name: &str) -> JoinHandle<Result<()>> { |
|
|
|
|
let replica_opt = self.replica_by_name(name); |
|
|
|
|
let home = self.home(); |
|
|
|
|
let mut interval = self.interval(); |
|
|
|
|
let i1 = self.interval(); |
|
|
|
|
let i2 = self.interval(); |
|
|
|
|
let name = name.to_owned(); |
|
|
|
|
|
|
|
|
|
tokio::spawn(async move { |
|
|
|
|
let replica = replica_opt.ok_or_else(|| eyre!("No replica named {}", name))?; |
|
|
|
|
|
|
|
|
|
loop { |
|
|
|
|
let (updated, confirmed) = tokio::join!( |
|
|
|
|
Self::poll_and_relay_update(home.clone(), replica.clone()), |
|
|
|
|
Self::poll_confirm(replica.clone()) |
|
|
|
|
); |
|
|
|
|
|
|
|
|
|
if let Err(ref e) = updated { |
|
|
|
|
tracing::error!("Error polling updates: {:?}", e) |
|
|
|
|
} |
|
|
|
|
if let Err(ref e) = confirmed { |
|
|
|
|
tracing::error!("Error polling confirms: {:?}", e) |
|
|
|
|
} |
|
|
|
|
updated?; |
|
|
|
|
confirmed?; |
|
|
|
|
interval.tick().await; |
|
|
|
|
if replica_opt.is_none() { |
|
|
|
|
bail!("No replica named {}", name); |
|
|
|
|
} |
|
|
|
|
}) |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
let replica = replica_opt.unwrap(); |
|
|
|
|
|
|
|
|
|
#[cfg(test)] |
|
|
|
|
mod test { |
|
|
|
|
use ethers::{core::types::H256, prelude::LocalWallet}; |
|
|
|
|
use std::sync::Arc; |
|
|
|
|
|
|
|
|
|
use super::*; |
|
|
|
|
use optics_core::{traits::TxOutcome, SignedUpdate, Update}; |
|
|
|
|
use optics_test::mocks::{MockHomeContract, MockReplicaContract}; |
|
|
|
|
|
|
|
|
|
#[tokio::test] |
|
|
|
|
async fn polls_and_relays_updates() { |
|
|
|
|
let signer: LocalWallet = |
|
|
|
|
"1111111111111111111111111111111111111111111111111111111111111111" |
|
|
|
|
.parse() |
|
|
|
|
.unwrap(); |
|
|
|
|
|
|
|
|
|
let first_root = H256::from([1; 32]); |
|
|
|
|
let second_root = H256::from([2; 32]); |
|
|
|
|
|
|
|
|
|
let signed_update = Update { |
|
|
|
|
home_domain: 1, |
|
|
|
|
previous_root: first_root, |
|
|
|
|
new_root: second_root, |
|
|
|
|
} |
|
|
|
|
.sign_with(&signer) |
|
|
|
|
.await |
|
|
|
|
.expect("!sign"); |
|
|
|
|
|
|
|
|
|
let mut mock_home = MockHomeContract::new(); |
|
|
|
|
let mut mock_replica = MockReplicaContract::new(); |
|
|
|
|
|
|
|
|
|
{ |
|
|
|
|
let signed_update = signed_update.clone(); |
|
|
|
|
// home.signed_update_by_old_root(first_root) called once and
|
|
|
|
|
// returns mock value signed_update
|
|
|
|
|
mock_home |
|
|
|
|
.expect__signed_update_by_old_root() |
|
|
|
|
.withf(move |r: &H256| *r == first_root) |
|
|
|
|
.times(1) |
|
|
|
|
.return_once(move |_| Ok(Some(signed_update))); |
|
|
|
|
} |
|
|
|
|
{ |
|
|
|
|
let signed_update = signed_update.clone(); |
|
|
|
|
// replica.current_root called once and returns mock value
|
|
|
|
|
// first_root
|
|
|
|
|
mock_replica |
|
|
|
|
.expect__current_root() |
|
|
|
|
.times(1) |
|
|
|
|
.returning(move || Ok(first_root)); |
|
|
|
|
// replica.update(signed_update) called once and returns
|
|
|
|
|
// mock default value
|
|
|
|
|
mock_replica |
|
|
|
|
.expect__update() |
|
|
|
|
.withf(move |s: &SignedUpdate| *s == signed_update) |
|
|
|
|
.times(1) |
|
|
|
|
.returning(|_| { |
|
|
|
|
Ok(TxOutcome { |
|
|
|
|
txid: H256::default(), |
|
|
|
|
executed: true, |
|
|
|
|
}) |
|
|
|
|
}); |
|
|
|
|
} |
|
|
|
|
let update_poller = UpdatePoller::new(home, replica.clone(), i1); |
|
|
|
|
let update_task = update_poller.spawn(); |
|
|
|
|
|
|
|
|
|
let mut home: Arc<Homes> = Arc::new(mock_home.into()); |
|
|
|
|
let mut replica: Arc<Replicas> = Arc::new(mock_replica.into()); |
|
|
|
|
Relayer::poll_and_relay_update(home.clone(), replica.clone()) |
|
|
|
|
.await |
|
|
|
|
.expect("Should have returned Ok(())"); |
|
|
|
|
let confirm_poller = ConfirmPoller::new(replica, i2); |
|
|
|
|
let confirm_task = confirm_poller.spawn(); |
|
|
|
|
|
|
|
|
|
let mock_home = Arc::get_mut(&mut home).unwrap(); |
|
|
|
|
mock_home.checkpoint(); |
|
|
|
|
let (res, _, _) = select_all(vec![confirm_task, update_task]).await; |
|
|
|
|
|
|
|
|
|
let mock_replica = Arc::get_mut(&mut replica).unwrap(); |
|
|
|
|
mock_replica.checkpoint(); |
|
|
|
|
} |
|
|
|
|
let res = res?; |
|
|
|
|
|
|
|
|
|
#[tokio::test] |
|
|
|
|
async fn confirms_updates() { |
|
|
|
|
let mut mock_replica = MockReplicaContract::new(); |
|
|
|
|
// replica.can_confirm called once and returns mock true
|
|
|
|
|
mock_replica |
|
|
|
|
.expect__can_confirm() |
|
|
|
|
.times(1) |
|
|
|
|
.returning(|| Ok(true)); |
|
|
|
|
// replica.confirm called once and returns mock default
|
|
|
|
|
mock_replica.expect__confirm().times(1).returning(|| { |
|
|
|
|
Ok(TxOutcome { |
|
|
|
|
txid: H256::default(), |
|
|
|
|
executed: true, |
|
|
|
|
}) |
|
|
|
|
}); |
|
|
|
|
|
|
|
|
|
let mut replica: Arc<Replicas> = Arc::new(mock_replica.into()); |
|
|
|
|
Relayer::poll_confirm(replica.clone()) |
|
|
|
|
.await |
|
|
|
|
.expect("Should have returned Ok(())"); |
|
|
|
|
|
|
|
|
|
let mock_replica = Arc::get_mut(&mut replica).unwrap(); |
|
|
|
|
mock_replica.checkpoint(); |
|
|
|
|
tracing::error!("Relayer error. {:?}", res); |
|
|
|
|
res |
|
|
|
|
}) |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
#[cfg(test)] |
|
|
|
|
mod test { |
|
|
|
|
// use ethers::{core::types::H256, prelude::LocalWallet};
|
|
|
|
|
// use std::sync::Arc;
|
|
|
|
|
|
|
|
|
|
// use super::*;
|
|
|
|
|
// use optics_core::{traits::TxOutcome, SignedUpdate, Update};
|
|
|
|
|
// use optics_test::mocks::{MockHomeContract, MockReplicaContract};
|
|
|
|
|
|
|
|
|
|
// #[tokio::test]
|
|
|
|
|
// async fn polls_and_relays_updates() {
|
|
|
|
|
// let signer: LocalWallet =
|
|
|
|
|
// "1111111111111111111111111111111111111111111111111111111111111111"
|
|
|
|
|
// .parse()
|
|
|
|
|
// .unwrap();
|
|
|
|
|
|
|
|
|
|
// let first_root = H256::from([1; 32]);
|
|
|
|
|
// let second_root = H256::from([2; 32]);
|
|
|
|
|
|
|
|
|
|
// let signed_update = Update {
|
|
|
|
|
// home_domain: 1,
|
|
|
|
|
// previous_root: first_root,
|
|
|
|
|
// new_root: second_root,
|
|
|
|
|
// }
|
|
|
|
|
// .sign_with(&signer)
|
|
|
|
|
// .await
|
|
|
|
|
// .expect("!sign");
|
|
|
|
|
|
|
|
|
|
// let mut mock_home = MockHomeContract::new();
|
|
|
|
|
// let mut mock_replica = MockReplicaContract::new();
|
|
|
|
|
|
|
|
|
|
// {
|
|
|
|
|
// let signed_update = signed_update.clone();
|
|
|
|
|
// // home.signed_update_by_old_root(first_root) called once and
|
|
|
|
|
// // returns mock value signed_update
|
|
|
|
|
// mock_home
|
|
|
|
|
// .expect__signed_update_by_old_root()
|
|
|
|
|
// .withf(move |r: &H256| *r == first_root)
|
|
|
|
|
// .times(1)
|
|
|
|
|
// .return_once(move |_| Ok(Some(signed_update)));
|
|
|
|
|
// }
|
|
|
|
|
// {
|
|
|
|
|
// let signed_update = signed_update.clone();
|
|
|
|
|
// // replica.current_root called once and returns mock value
|
|
|
|
|
// // first_root
|
|
|
|
|
// mock_replica
|
|
|
|
|
// .expect__current_root()
|
|
|
|
|
// .times(1)
|
|
|
|
|
// .returning(move || Ok(first_root));
|
|
|
|
|
// // replica.update(signed_update) called once and returns
|
|
|
|
|
// // mock default value
|
|
|
|
|
// mock_replica
|
|
|
|
|
// .expect__update()
|
|
|
|
|
// .withf(move |s: &SignedUpdate| *s == signed_update)
|
|
|
|
|
// .times(1)
|
|
|
|
|
// .returning(|_| {
|
|
|
|
|
// Ok(TxOutcome {
|
|
|
|
|
// txid: H256::default(),
|
|
|
|
|
// executed: true,
|
|
|
|
|
// })
|
|
|
|
|
// });
|
|
|
|
|
// }
|
|
|
|
|
|
|
|
|
|
// let mut home: Arc<Homes> = Arc::new(mock_home.into());
|
|
|
|
|
// let mut replica: Arc<Replicas> = Arc::new(mock_replica.into());
|
|
|
|
|
// Relayer::poll_and_relay_update(home.clone(), replica.clone())
|
|
|
|
|
// .await
|
|
|
|
|
// .expect("Should have returned Ok(())");
|
|
|
|
|
|
|
|
|
|
// let mock_home = Arc::get_mut(&mut home).unwrap();
|
|
|
|
|
// mock_home.checkpoint();
|
|
|
|
|
|
|
|
|
|
// let mock_replica = Arc::get_mut(&mut replica).unwrap();
|
|
|
|
|
// mock_replica.checkpoint();
|
|
|
|
|
// }
|
|
|
|
|
|
|
|
|
|
// #[tokio::test]
|
|
|
|
|
// async fn confirms_updates() {
|
|
|
|
|
// let mut mock_replica = MockReplicaContract::new();
|
|
|
|
|
// // replica.can_confirm called once and returns mock true
|
|
|
|
|
// mock_replica
|
|
|
|
|
// .expect__can_confirm()
|
|
|
|
|
// .times(1)
|
|
|
|
|
// .returning(|| Ok(true));
|
|
|
|
|
// // replica.confirm called once and returns mock default
|
|
|
|
|
// mock_replica.expect__confirm().times(1).returning(|| {
|
|
|
|
|
// Ok(TxOutcome {
|
|
|
|
|
// txid: H256::default(),
|
|
|
|
|
// executed: true,
|
|
|
|
|
// })
|
|
|
|
|
// });
|
|
|
|
|
|
|
|
|
|
// let mut replica: Arc<Replicas> = Arc::new(mock_replica.into());
|
|
|
|
|
// Relayer::poll_confirm(replica.clone())
|
|
|
|
|
// .await
|
|
|
|
|
// .expect("Should have returned Ok(())");
|
|
|
|
|
|
|
|
|
|
// let mock_replica = Arc::get_mut(&mut replica).unwrap();
|
|
|
|
|
// mock_replica.checkpoint();
|
|
|
|
|
// }
|
|
|
|
|
} |
|
|
|
|
|
James Prestwich
3 years ago