feature: OpticsAgent trait and refactoring contract traits

rust/Cargo.lock

@@ -1231,6 +1231,7 @@ dependencies = [
  "ethers-middleware",
  "ethers-providers",
  "ethers-signers",
+ "futures-util",
  "optics-core",
  "serde 1.0.120",
  "serde_json",

rust/optics-base/Cargo.toml

@@ -24,4 +24,6 @@ thiserror = { version = "1.0.22", default-features = false }
 async-trait = { version = "0.1.42", default-features = false }
 url = { version = "2.2.0", default-features = false }
 
+futures-util = "0.3.12"
+
 color-eyre = "0.5.0"

rust/optics-base/src/abis/mod.rs

@@ -58,10 +58,6 @@ where
         Ok(receipt_opt.map(Into::into))
     }
 
-    fn origin_slip44(&self) -> u32 {
-        self.slip44
-    }
-
     async fn updater(&self) -> Result<H256, ChainCommunicationError> {
         Ok(self.contract.updater().call().await?.into())
     }
@@ -124,6 +120,10 @@ impl<M> Replica for ReplicaContract<M>
 where
     M: ethers_providers::Middleware + 'static,
 {
+    fn destination_slip44(&self) -> u32 {
+        self.slip44
+    }
+
     async fn next_pending(&self) -> Result<Option<(H256, U256)>, ChainCommunicationError> {
         let (pending, confirm_at) = self.contract.next_pending().call().await?;
 
@@ -214,10 +214,6 @@ where
         Ok(receipt_opt.map(Into::into))
     }
 
-    fn origin_slip44(&self) -> u32 {
-        self.slip44
-    }
-
     async fn updater(&self) -> Result<H256, ChainCommunicationError> {
         Ok(self.contract.updater().call().await?.into())
     }
@@ -280,6 +276,10 @@ impl<M> Home for HomeContract<M>
 where
     M: ethers_providers::Middleware + 'static,
 {
+    fn origin_slip44(&self) -> u32 {
+        self.slip44
+    }
+
     async fn raw_message_by_sequence(
         &self,
         destination: u32,

rust/optics-base/src/agent.rs

@@ -0,0 +1,56 @@
+use async_trait::async_trait;
+use color_eyre::{eyre::WrapErr, Result};
+use futures_util::future::{join_all, select_all};
+use std::sync::Arc;
+
+use crate::settings::Settings;
+use optics_core::traits::{Home, Replica};
+
+/// A trait for an application that runs on a replica and a reference to a
+/// home.
+#[async_trait]
+pub trait OpticsAgent: Send + Sync + std::fmt::Debug {
+    /// Run the agent with the given home and replica
+    async fn run(home: Arc<Box<dyn Home>>, replica: Box<dyn Replica>) -> Result<()>;
+
+    /// Run several agents
+    async fn run_many(home: Box<dyn Home>, replicas: Vec<Box<dyn Replica>>) -> Result<()> {
+        let home = Arc::new(home);
+
+        let mut replica_tasks: Vec<_> = replicas
+            .into_iter()
+            .map(|replica| Self::run(home.clone(), replica))
+            .collect();
+
+        loop {
+            let (_res, _, rem) = select_all(replica_tasks).await;
+            // TODO: report failure
+            replica_tasks = rem;
+            if replica_tasks.is_empty() {
+                break;
+            }
+        }
+
+        Ok(())
+    }
+
+    /// Run several agents based on the settings
+    async fn run_from_settings(settings: &Settings) -> Result<()> {
+        let home = settings
+            .home
+            .try_into_home()
+            .await
+            .wrap_err("failed to instantiate Home")?;
+
+        let replicas = join_all(settings.replicas.iter().map(|(k, v)| async move {
+            v.try_into_replica()
+                .await
+                .wrap_err_with(|| format!("Failed to instantiate replica named {}", k))
+        }))
+        .await
+        .into_iter()
+        .collect::<Result<Vec<_>>>()?;
+
+        Self::run_many(home, replicas).await
+    }
+}

rust/optics-base/src/main.rs

@@ -17,106 +17,31 @@ pub mod abis;
 /// Settings and configuration from file
 pub mod settings;
 
-use color_eyre::{
-    eyre::{eyre, WrapErr},
-    Result,
-};
-use std::collections::HashMap;
-use tokio::task::JoinHandle;
+/// Base trait for an agent
+pub mod agent;
 
-use optics_core::traits::{Home, Replica};
+use color_eyre::Result;
 
-/// The global app context.
-///
-/// We erase all type info to allow easier abstraction across chains without
-/// managing insanely large, annoying type systems.
-///
-/// Usually this will be bundled in a larger
-#[derive(Debug)]
-struct ChainConnections {
-    home: Box<dyn Home>,
-    replicas: HashMap<String, Box<dyn Replica>>,
-}
-
-#[derive(Debug)]
-struct App {
-    home: JoinHandle<Result<()>>,
-    replicas: HashMap<String, JoinHandle<Result<()>>>,
-}
-
-impl ChainConnections {
-    pub async fn try_from_settings(settings: &settings::Settings) -> Result<Self> {
-        let home = settings
-            .home
-            .try_into_home()
-            .await
-            .wrap_err("failed to instantiate Home")?;
-
-        let mut replicas = HashMap::new();
-        // TODO: parallelize if this becomes expensive
-        for (key, value) in settings.replicas.iter() {
-            replicas.insert(
-                key.clone(),
-                value
-                    .try_into_replica()
-                    .await
-                    .wrap_err_with(|| format!("Failed to instantiate replica named {}", key))?,
-            );
-        }
-
-        Ok(ChainConnections { home, replicas })
-    }
-}
-
-impl App {
-    pub async fn try_from_settings(settings: settings::Settings) -> Self {
-        let (tx, _) = tokio::sync::broadcast::channel::<()>(16);
-
-        let replicas = settings
-            .replicas
-            .into_iter()
-            .map(|(k, v)| {
-                let mut rx = tx.subscribe();
-                (
-                    k,
-                    tokio::spawn(async move {
-                        let replica = v.try_into_replica().await?;
+use crate::{agent::OpticsAgent, settings::Settings};
 
-                        loop {
-                            let _ = rx.recv().await?;
-                        }
-                        Ok(())
-                    }),
-                )
-            })
-            .collect();
-
-        let home = tokio::spawn(async move { Ok(()) });
-
-        Self { home, replicas }
-    }
-}
-
-async fn _main(settings: settings::Settings) -> Result<()> {
-    let app = ChainConnections::try_from_settings(&settings).await?;
-
-    let current = app.home.current_root().await.map_err(|e| eyre!(e))?;
-
-    println!("current is {}", &current);
-
-    Ok(())
+async fn _main<OA>(settings: Settings) -> Result<()>
+where
+    OA: OpticsAgent,
+{
+    OA::run_from_settings(&settings).await
 }
 
 fn main() -> Result<()> {
     color_eyre::install()?;
 
     let settings = settings::Settings::new().expect("!config");
+    dbg!(settings);
 
-    tokio::runtime::Builder::new_current_thread()
-        .enable_all()
-        .build()
-        .unwrap()
-        .block_on(_main(settings))?;
+    // tokio::runtime::Builder::new_current_thread()
+    //     .enable_all()
+    //     .build()
+    //     .unwrap()
+    //     .block_on(_main(settings))?;
 
     Ok(())
 }

rust/optics-base/src/settings/ethereum.rs

@@ -24,17 +24,17 @@ pub enum EthereumConnection {
 // Construct boxed contracts in a big "if-else" chain to handle multiple
 // combinations of middleware.
 macro_rules! construct_box_contract {
-    ($contract:ident, $slip44:expr, $address:expr, $provider:expr, $signer:expr) => {{
+    ($contract:ident, $origin_slip44:expr, $address:expr, $provider:expr, $signer:expr) => {{
         if let Some(signer) = $signer {
             let provider = ethers_middleware::SignerMiddleware::new($provider, signer);
             Box::new(crate::abis::$contract::at(
-                $slip44,
+                $origin_slip44,
                 $address,
                 provider.into(),
             ))
         } else {
             Box::new(crate::abis::$contract::at(
-                $slip44,
+                $origin_slip44,
                 $address,
                 $provider.into(),
             ))

rust/optics-core/src/traits/home.rs

@@ -4,6 +4,7 @@ use ethers_core::types::H256;
 
 use crate::{
     traits::{ChainCommunicationError, Common, TxOutcome},
+    utils::domain_hash,
     Decode, Message, SignedUpdate, Update,
 };
 
@@ -11,6 +12,14 @@ use crate::{
 /// chains
 #[async_trait]
 pub trait Home: Common + Send + Sync + std::fmt::Debug {
+    /// Return the slip44 ID
+    fn origin_slip44(&self) -> u32;
+
+    /// Return the domain hash
+    fn domain_hash(&self) -> H256 {
+        domain_hash(self.origin_slip44())
+    }
+
     /// Fetch the message to destination at the sequence number (or error).
     /// This should fetch events from the chain API.
     ///

rust/optics-core/src/traits/mod.rs

@@ -7,7 +7,7 @@ pub mod replica;
 use async_trait::async_trait;
 use ethers_core::types::{TransactionReceipt, H256};
 
-use crate::{utils::domain_hash, SignedUpdate};
+use crate::SignedUpdate;
 
 pub use home::*;
 pub use replica::*;
@@ -49,14 +49,6 @@ pub trait Common: Sync + Send + std::fmt::Debug {
     /// Get the status of a transaction
     async fn status(&self, txid: H256) -> Result<Option<TxOutcome>, ChainCommunicationError>;
 
-    /// Return the slip44 ID
-    fn origin_slip44(&self) -> u32;
-
-    /// Return the domain hash
-    fn domain_hash(&self) -> H256 {
-        domain_hash(self.origin_slip44())
-    }
-
     /// Fetch the current updater value
     async fn updater(&self) -> Result<H256, ChainCommunicationError>;
 

rust/optics-core/src/traits/replica.rs

@@ -9,6 +9,9 @@ use crate::{
 /// Interface for on-chain replicas
 #[async_trait]
 pub trait Replica: Common + Send + Sync + std::fmt::Debug {
+    /// Return the replica slip44 ID
+    fn destination_slip44(&self) -> u32;
+
     /// Return the pending root and time, if any
     async fn next_pending(&self) -> Result<Option<(H256, U256)>, ChainCommunicationError>;