feat: refactored coroutine pipeline so nothing can get stuck (#166)

* big refactor (that's broken)

* feat: refactored coroutine pipeline so nothing can get stuck

- added lots of gucci comments
- nearly lost my mind understanding async rust

* aligned some terminology

Author: CluEleSsUK

bin/onlyswaps-verifier/src/app.rs

@@ -0,0 +1,82 @@
+use crate::chain_state::NetworkBus;
+use crate::channel_manager::TaskManager;
+use crate::config::AppConfig;
+use crate::control_plane::DefaultControlPlane;
+use crate::retry_runtime::RetryScheduler;
+use crate::verification_events::EventManagement;
+use futures::{StreamExt, stream};
+use std::sync::Arc;
+
+pub(crate) struct App {}
+
+impl App {
+    pub async fn start(app_config: &AppConfig) -> anyhow::Result<()> {
+        // the `network_bus` manages access to all the chains at once for pulling state or submitting txs
+        let network_bus = Arc::new(NetworkBus::new(app_config).await?);
+
+        // the `control_plane` provides the app phases each request goes through; consider it the
+        // inner circle of the domain-driven-design nested circles.
+        // Each validation goes through the following phases:
+        // - receive                           (event appears from the chain)
+        // - resolve                           (we resolve the state relating to that requestId)
+        // - evaluate                          (we evaluate the resolved state to see if it's a complete swap)
+        // - sign                              (we sign the necessary parameters of verified swaps)
+        // - submit                            (we ship the payload back to the contract)
+        // and optionally error handling       (we can stage many of the errors for retry)
+        let control_plane = DefaultControlPlane::new(app_config, network_bus.clone()).await?;
+
+        // the `task_manager` connects all the above phases to one another and the outside world
+        // and manages sending errors on the right channels. It abstracts movement between
+        // phases to make the control plane easier to test
+        let task_manager = TaskManager::new(control_plane);
+
+        // `EventManagement` sets up some streaming-related structs; there are some weird semantics
+        // around dropping that make it a nightmare to actually hide anything away in its implementation,
+        // hence why so much of the streaming logic is in this function :(
+        let EventManagement {
+            omnievent,
+            event_ids,
+        } = EventManagement::new(app_config).await?;
+        //
+        let live_stream = omnievent
+            .get_ethereum_multi_event_stream(event_ids.clone())
+            .await?
+            .filter_map(|maybe_event| async move {
+                match maybe_event {
+                    Ok(event) => match event.data.try_into() {
+                        Ok(verification) => Some(verification),
+                        _ => {
+                            tracing::warn!("received an invalid RPC event");
+                            None
+                        }
+                    },
+                    _ => None,
+                }
+            });
+
+        // the `retry_scheduler` allows errors at any phase to drop back in at the relevant stage.
+        // e.g. if an RPC is down during submission, it may be possible to just resubmit the
+        // verified signature in a short while rather than pull all the state again
+        let retry_scheduler = RetryScheduler::new(app_config);
+        let retry_tx = retry_scheduler.tx();
+        let retry_stream = retry_scheduler.into_stream();
+
+        // we get outstanding pending verifiations that might have accumulated while our
+        // node was down. It's not a 'stream' in the traditional sense, hence we have to do some
+        // iter -> chain magic below
+        let pending_verifications = network_bus
+            .fetch_pending_verifications()
+            .await
+            .unwrap_or_default();
+
+        let live_streams = stream::select(live_stream, retry_stream);
+        let stream = stream::iter(pending_verifications).chain(live_streams);
+
+        // in order to allow the `task_manager` to send retries back 'up the funnel', we pass in a
+        // tx "Sender" into its `run` method.
+        task_manager.run(retry_tx, Box::pin(stream)).await;
+
+        // if the `run` function ever ends, it's night night
+        anyhow::bail!("onlyswaps closed unexpectedly")
+    }
+}

bin/onlyswaps-verifier/src/chain_state.rs

@@ -1,6 +1,6 @@
 use crate::chain_state_pending::{RequestId, Verification, extract_pending_verifications};
-use crate::config::TimeoutConfig;
-use crate::signing::VerifiedSwap;
+use crate::config::{AppConfig, TimeoutConfig};
+use crate::signing::SignedVerification;
 use alloy::network::EthereumWallet;
 use alloy::primitives::{Address, Bytes, FixedBytes};
 use alloy::providers::{DynProvider, Provider, ProviderBuilder, WsConnect};
@@ -33,16 +33,14 @@ pub(crate) struct Network<P> {
 }
 
 impl NetworkBus<DynProvider> {
-    pub async fn new(
-        eth_private_key: impl Into<Arc<PrivateKeySigner>>,
-        network_configs: &[NetworkConfig],
-        timeout_config: &TimeoutConfig,
-    ) -> anyhow::Result<Self> {
-        let private_key = eth_private_key.into();
+    pub async fn new(app_config: &AppConfig) -> anyhow::Result<Self> {
+        let private_key = PrivateKeySigner::from_slice(app_config.eth_private_key.as_slice())?;
+        let private_key = Arc::new(private_key);
         let mut networks = HashMap::new();
 
-        for config in network_configs.iter() {
-            let network = Network::new(private_key.clone(), config, timeout_config.clone()).await?;
+        for config in app_config.networks.iter() {
+            let network =
+                Network::new(private_key.clone(), config, app_config.timeout.clone()).await?;
             networks.insert(config.chain_id, network);
         }
 
@@ -84,7 +82,7 @@ impl NetworkBus<DynProvider> {
 
     pub(crate) async fn submit_verification(
         &self,
-        verified_swap: &VerifiedSwap,
+        verified_swap: &SignedVerification,
     ) -> anyhow::Result<()> {
         let chain_id: u64 = verified_swap.src_chain_id.try_into()?;
         let transport = self.networks.get(&chain_id).ok_or(anyhow!(
@@ -98,11 +96,10 @@ impl NetworkBus<DynProvider> {
 
 impl Network<DynProvider> {
     pub async fn new(
-        signer: impl Into<Arc<PrivateKeySigner>>,
+        signer: Arc<PrivateKeySigner>,
         config: &NetworkConfig,
         timeout_config: TimeoutConfig,
     ) -> anyhow::Result<Self> {
-        let signer = signer.into();
         let url = config.rpc_url.clone();
         let own_addr = signer.address();
         let provider = ProviderBuilder::new()
@@ -178,7 +175,10 @@ impl<P: Provider> Network<P> {
             .await?)
     }
 
-    pub async fn submit_verified_swap(&self, verified_swap: &VerifiedSwap) -> anyhow::Result<()> {
+    pub async fn submit_verified_swap(
+        &self,
+        verified_swap: &SignedVerification,
+    ) -> anyhow::Result<()> {
         // nodes can be configured not to write the signature to save gas
         if !self.should_write {
             return Ok(());
@@ -196,7 +196,7 @@ impl<P: Provider> Network<P> {
         }
     }
 
-    async fn rebalance(&self, verified_swap: &VerifiedSwap) -> anyhow::Result<()> {
+    async fn rebalance(&self, verified_swap: &SignedVerification) -> anyhow::Result<()> {
         let tx = self
             .router
             .rebalanceSolver(

bin/onlyswaps-verifier/src/chain_state_pending.rs

@@ -88,7 +88,7 @@ enum Status {
 }
 pub type RequestId = FixedBytes<32>;
 
-#[derive(Debug, PartialEq, Eq, Hash)]
+#[derive(Debug, PartialEq, Eq, Hash, Clone)]
 pub struct Verification<ID> {
     pub dest_chain_id: u64,
     pub request_id: ID,

bin/onlyswaps-verifier/src/channel_manager.rs

@@ -0,0 +1,185 @@
+use crate::chain_state_pending::{RequestId, Verification};
+use crate::control_plane::{ControlPlane, ResolvedState, VerificationError};
+use crate::retry_runtime::RetrySender;
+use crate::signing::SignedVerification;
+use futures::Stream;
+use std::pin::Pin;
+use std::sync::Arc;
+use tokio::sync::mpsc::unbounded_channel;
+use tokio::task::JoinSet;
+use tokio_stream::StreamExt;
+
+pub(crate) struct TaskManager<CP> {
+    control_plane: Arc<CP>,
+}
+
+impl<CP: ControlPlane> TaskManager<CP>
+where
+    CP: Send + Sync + 'static,
+{
+    pub fn new(control_plane: impl Into<Arc<CP>>) -> Self {
+        Self {
+            control_plane: control_plane.into(),
+        }
+    }
+
+    // `run` spawns a coroutine for each of the steps of processing, and each
+    // request also gets its own sub(?)-coroutine to ensure that no slow RPC or
+    // dodgy request can block the pipeline for other valid ones
+    pub async fn run(
+        &self,
+        retry_tx: RetrySender,
+        mut event_stream: Pin<Box<impl Stream<Item = Verification<RequestId>> + Send + 'static>>,
+    ) {
+        tracing::info!("starting channel manager");
+        let mut tasks = JoinSet::new();
+        let control_plane = self.control_plane.clone();
+
+        let (tx_verifications, mut rx_verifications) =
+            unbounded_channel::<Verification<RequestId>>();
+        let (tx_resolve, mut rx_resolve) = unbounded_channel::<ResolvedState>();
+        let (tx_sign, mut rx_sign) = unbounded_channel::<ResolvedState>();
+        let (tx_submit, mut rx_submit) = unbounded_channel::<SignedVerification>();
+        let (tx_done, mut rx_done) = unbounded_channel::<SignedVerification>();
+        let (tx_err, mut rx_err) = unbounded_channel::<VerificationError>();
+
+        // 'receive' step
+        {
+            let tx_verifications = tx_verifications.clone();
+            tasks.spawn(async move {
+                while let Some(verification) = event_stream.next().await {
+                    tx_verifications
+                        .send(verification)
+                        .expect("failed to send verification on channel");
+                }
+            });
+        }
+
+        // 'resolve' step
+        {
+            let control_plane = control_plane.clone();
+            let tx_resolve = tx_resolve.clone();
+            let tx_err = tx_err.clone();
+
+            // resolve
+            tasks.spawn(async move {
+                while let Some(verification) = rx_verifications.recv().await {
+                    let control_plane = control_plane.clone();
+                    let tx_resolve = tx_resolve.clone();
+                    let tx_err = tx_err.clone();
+
+                    tokio::spawn(async move {
+                        match control_plane.resolve_state(verification.clone()).await {
+                            Ok(v) => tx_resolve.send(v).expect("error writing on channel"),
+                            Err(_) => tx_err
+                                .send(VerificationError::Resolve(verification))
+                                .expect("error writing on channel"),
+                        }
+                    });
+                }
+            });
+        }
+
+        // 'evaluate' step
+        {
+            let control_plane = control_plane.clone();
+            let tx_err = tx_err.clone();
+
+            tasks.spawn(async move {
+                while let Some(valid) = rx_resolve.recv().await {
+                    let control_plane = control_plane.clone();
+                    let tx_sign = tx_sign.clone();
+                    let tx_err = tx_err.clone();
+
+                    tokio::spawn(async move {
+                        match control_plane.evaluate_state(valid.clone()).await {
+                            Ok(v) => tx_sign.send(v).expect("error writing on channel"),
+                            Err(_) => tx_err
+                                .send(VerificationError::Evaluate(valid))
+                                .expect("error writing on channel"),
+                        }
+                    });
+                }
+            });
+        }
+
+        // 'sign' step
+        {
+            let control_plane = control_plane.clone();
+            let tx_err = tx_err.clone();
+
+            tasks.spawn(async move {
+                while let Some(state) = rx_sign.recv().await {
+                    let control_plane = control_plane.clone();
+                    let tx_submit = tx_submit.clone();
+                    let tx_err = tx_err.clone();
+
+                    tokio::spawn(async move {
+                        match control_plane.sign_state(state.clone()).await {
+                            Ok(v) => tx_submit.send(v).expect("error writing on channel"),
+                            Err(_) => tx_err
+                                .send(VerificationError::Sign(state))
+                                .expect("error writing on channel"),
+                        }
+                    });
+                }
+            });
+        }
+
+        // 'submit' step
+        {
+            let control_plane = control_plane.clone();
+            let tx_done = tx_done.clone();
+            let tx_err = tx_err.clone();
+
+            tasks.spawn(async move {
+                while let Some(signed_verification) = rx_submit.recv().await {
+                    let control_plane = control_plane.clone();
+                    let tx_done = tx_done.clone();
+                    let tx_err = tx_err.clone();
+
+                    tokio::spawn(async move {
+                        match control_plane
+                            .submit_state(signed_verification.clone())
+                            .await
+                        {
+                            Ok(v) => tx_done.send(v).expect("error writing on channel"),
+                            Err(_) => tx_err
+                                .send(VerificationError::Submit(signed_verification))
+                                .expect("error writing on channel"),
+                        }
+                    });
+                }
+            });
+        }
+
+        // 'done' step
+        tasks.spawn(async move {
+            while let Some(state) = rx_done.recv().await {
+                tracing::info!(
+                    chain_id = state.src_chain_id.to_string(),
+                    request_id = state.request_id.to_string(),
+                    "verification completed successfully"
+                );
+            }
+        });
+
+        // 'error handling' step
+        {
+            let control_plane = control_plane.clone();
+            let retry_tx = retry_tx.clone();
+
+            tasks.spawn(async move {
+                while let Some(err) = rx_err.recv().await {
+                    control_plane.handle_error(&err, retry_tx.clone()).await;
+                }
+            });
+        }
+
+        // join all the things so we don't cede control back to the caller
+        tracing::info!("all tasks started");
+        if tasks.join_next().await.is_some() {
+            panic!("a task manager task ended, but they should run forever")
+        }
+    }
+}