Merge pull request #28 from sergerad/main

Support multiple monitors in rbtr with async actor

Author: sergerad

Cargo.toml

@@ -6,19 +6,20 @@ members = [
     "crates/tokens",
     "crates/uniswapv3pool",
     "crates/lib",
-    "bin/rbtr", "tests",
+    "bin/rbtr",
+    "tests",
 ]
 
 [workspace.dependencies]
-uniswap-v3-sdk = { version = "2.2.0", features = ["extensions", "std"] }
-uniswap-sdk-core = "3.0.0"
+uniswap-v3-sdk = { version = "2.6.0", features = ["extensions", "std"] }
+uniswap-sdk-core = "3.2.0"
 eyre = "0.6.12"
-alloy = { version = "0.5", features = ["contract"] }
+alloy = { version = "0.6", features = ["contract"] }
 alloy-primitives = "0.8"
 alloy-sol-types = "0.8"
-tokio = { version = "1.40", features = ["full"] }
+tokio = { version = "1.41.1", features = ["full"] }
 toml = "0.8.19"
-serde = { version = "1.0.214", features = ["derive"] }
+serde = { version = "1.0.215", features = ["derive"] }
 
 tokens = { path = "./crates/tokens" }
 uniswapv3pool = { path = "./crates/uniswapv3pool" }

bin/rbtr/Cargo.toml

@@ -9,8 +9,11 @@ serde = { workspace = true }
 eyre = { workspace = true }
 alloy = { workspace = true }
 tokio = { workspace = true }
+uniswap-v3-sdk = { workspace = true }
 tokio-util = "0.7.12"
 actix = "0.13.5"
 ctrlc = "3.1.0"
+uniswapv3pool.workspace = true
+duration-string = { version = "0.4.0", features = ["serde"] }
 
 lib = { path = "../../crates/lib" }

bin/rbtr/Config.toml

@@ -1,4 +1,13 @@
-url_one = "https://rpc.immutable.com"
-url_two = "https://rpc.immutable.com"
-token_one = "0xD67cc11151dBccCC424A16F8963ece3D0539BD61"
-token_two = "0xD67cc11151dBccCC424A16F8963ece3D0539BD61"
+tick_rate = "5s"
+
+[[monitors]]
+rpc_url = "https://rpc.immutable.com"
+factory = "0x56c2162254b0E4417288786eE402c2B41d4e181e"
+token_one = "0x52A6c53869Ce09a731CD772f245b97A4401d3348"
+token_two = "0x3A0C2Ba54D6CBd3121F01b96dFd20e99D1696C9D"
+
+[[monitors]]
+rpc_url = "https://rpc.immutable.com"
+factory = "0x56c2162254b0E4417288786eE402c2B41d4e181e"
+token_one = "0x52A6c53869Ce09a731CD772f245b97A4401d3348"
+token_two = "0x3A0C2Ba54D6CBd3121F01b96dFd20e99D1696C9D"

bin/rbtr/src/actors/messages.rs

@@ -1,4 +1,5 @@
 use actix::prelude::*;
+use uniswapv3pool::UniswapV3PoolSdk;
 
 /// Control is a message to control the operation of an Actor.
 #[derive(Debug, Message)]
@@ -12,8 +13,8 @@ pub enum Control {
 #[derive(Debug, Message)]
 #[rtype(result = "()")]
 #[allow(dead_code)]
-pub struct PriceDiff {
-    pub diff: f64,
+pub struct PoolUpdate {
+    pub pool: UniswapV3PoolSdk,
 }
 
 /// Message to subscribe to some other message.

bin/rbtr/src/actors/mod.rs

@@ -1,5 +1,5 @@
 mod messages;
-pub use messages::{Control, PriceDiff, Subscribe};
+pub use messages::{Control, PoolUpdate, Subscribe};
 
 mod monitor;
 pub use monitor::Monitor;

bin/rbtr/src/actors/monitor.rs

@@ -1,18 +1,35 @@
-use crate::{PriceDiff, Subscribe};
+use crate::{PoolUpdate, Subscribe};
 use actix::prelude::*;
+use lib::prelude::{Address, Provider, RootProvider};
+use uniswap_v3_sdk::prelude::FeeAmount;
+use uniswapv3pool::UniswapV3PoolSdk;
 
 use super::Control;
 
 /// Monitor is the actor responsible for monitoring on-chain state
 /// and notifying subscribers of changes.
+#[derive(Clone)]
 pub struct Monitor {
-    price_diff_subs: Vec<Recipient<PriceDiff>>,
+    price_diff_subs: Vec<Recipient<PoolUpdate>>,
+    provider: RootProvider,
+    factory: Address,
+    token_0: Address,
+    token_1: Address,
 }
 
 impl Monitor {
-    pub fn new() -> Self {
+    pub fn new(
+        provider: RootProvider,
+        factory: Address,
+        token_0: Address,
+        token_1: Address,
+    ) -> Self {
         Monitor {
             price_diff_subs: vec![],
+            provider,
+            factory,
+            token_0,
+            token_1,
         }
     }
 }
@@ -21,10 +38,10 @@ impl Actor for Monitor {
     type Context = Context<Self>;
 }
 
-impl Handler<Subscribe<PriceDiff>> for Monitor {
+impl Handler<Subscribe<PoolUpdate>> for Monitor {
     type Result = ();
 
-    fn handle(&mut self, msg: Subscribe<PriceDiff>, _: &mut Self::Context) {
+    fn handle(&mut self, msg: Subscribe<PoolUpdate>, _: &mut Self::Context) {
         self.price_diff_subs.push(msg.0);
     }
 }
@@ -39,10 +56,25 @@ impl Handler<Control> for Monitor {
                 ctx.stop();
             }
             Control::Tick => {
-                for sub in &self.price_diff_subs {
-                    // TODO: handle error
-                    sub.do_send(PriceDiff { diff: 0.1 });
+                let self_clone = self.clone();
+                async move {
+                    let pool = UniswapV3PoolSdk::from_pool_key(
+                        self_clone.provider.get_chain_id().await.unwrap(),
+                        self_clone.factory,
+                        self_clone.token_0,
+                        self_clone.token_1,
+                        FeeAmount::LOW, // TODO: parameterize
+                        self_clone.provider,
+                        None,
+                    )
+                    .await
+                    .unwrap();
+                    for sub in self_clone.price_diff_subs {
+                        sub.do_send(PoolUpdate { pool: pool.clone() });
+                    }
                 }
+                .into_actor(self)
+                .wait(ctx);
             }
         }
     }

bin/rbtr/src/actors/resolver.rs

@@ -1,6 +1,6 @@
 use actix::prelude::*;
 
-use super::{Control, PriceDiff};
+use super::{Control, PoolUpdate};
 
 /// Resolver is the actor responsible for handling all data and events
 /// derived from on-chain state that other actors may provide.
@@ -16,10 +16,10 @@ impl Actor for Resolver {
     type Context = Context<Self>;
 }
 
-impl Handler<PriceDiff> for Resolver {
+impl Handler<PoolUpdate> for Resolver {
     type Result = ();
 
-    fn handle(&mut self, msg: PriceDiff, _: &mut Self::Context) -> Self::Result {
+    fn handle(&mut self, msg: PoolUpdate, _: &mut Self::Context) -> Self::Result {
         println!("Price diff: {msg:?}");
     }
 }

bin/rbtr/src/config.rs

@@ -1,24 +1,58 @@
+use duration_string::DurationString;
 use eyre::{Context, Result};
 use lib::prelude::*;
 use serde::Deserialize;
-use std::{fs, path::Path};
+use std::{fs, path::Path, time::Duration};
 
 #[derive(Deserialize)]
 struct RawConfig {
-    url_one: String,
-    url_two: String,
+    tick_rate: DurationString,
+    monitors: Vec<RawMonitorConfig>,
+}
+
+#[derive(Deserialize)]
+struct RawMonitorConfig {
+    rpc_url: String,
+    factory: Address,
     token_one: Address,
     token_two: Address,
 }
 
-// TODO: remove allow when this is used
-#[allow(dead_code)]
 #[derive(Debug)]
 pub struct Config {
-    provider_one: RootProvider,
-    provider_two: RootProvider,
-    token_address_one: Address,
-    token_address_two: Address,
+    pub tick_rate: Duration,
+    pub monitors: Vec<MonitorConfig>,
+}
+
+impl From<RawConfig> for Config {
+    fn from(raw: RawConfig) -> Self {
+        let monitors = raw.monitors.into_iter().map(MonitorConfig::from).collect();
+        Self {
+            tick_rate: raw.tick_rate.into(),
+            monitors,
+        }
+    }
+}
+
+#[derive(Debug)]
+pub struct MonitorConfig {
+    pub provider: RootProvider,
+    pub factory: Address,
+    pub token_one: Address,
+    pub token_two: Address,
+}
+
+impl From<RawMonitorConfig> for MonitorConfig {
+    fn from(raw: RawMonitorConfig) -> Self {
+        let rpc_url = Url::parse(&raw.rpc_url).unwrap();
+        let provider = ProviderBuilder::new().on_http(rpc_url);
+        Self {
+            provider,
+            factory: raw.factory,
+            token_one: raw.token_one,
+            token_two: raw.token_two,
+        }
+    }
 }
 
 impl Config {
@@ -29,19 +63,6 @@ impl Config {
         let raw =
             toml::from_str::<RawConfig>(&file_str).wrap_err("failed to parse config from toml")?;
 
-        // Parse URLs and create the providers.
-        let url_one =
-            Url::parse(&raw.url_one).wrap_err(format!("failed to parse url: {}", raw.url_one))?;
-        let url_two =
-            Url::parse(&raw.url_two).wrap_err(format!("failed to parse url: {}", raw.url_two))?;
-        let provider_one = ProviderBuilder::new().on_http(url_one);
-        let provider_two = ProviderBuilder::new().on_http(url_two);
-
-        Ok(Self {
-            provider_one,
-            provider_two,
-            token_address_one: raw.token_one,
-            token_address_two: raw.token_two,
-        })
+        Ok(raw.into())
     }
 }

bin/rbtr/src/main.rs

@@ -1,5 +1,4 @@
 mod config;
-use std::time::Duration;
 
 use actix::prelude::*;
 use config::Config;
@@ -28,23 +27,33 @@ fn main() -> Result<()> {
 
     // Block the main thread and run the actor system.
     system.block_on(async {
-        // Set up actors and their subscriptions
+        // Set up actors and their subscriptions.
         let resolver = Resolver::new().start();
-        let monitor = Monitor::new().start();
-        let subscribe = Subscribe::<PriceDiff>(resolver.clone().recipient());
-        monitor
-            .send(subscribe)
-            .await
-            .expect("Failed to set initial subscription to Monitor");
+
+        // Setup up the monitors.
+        let mut monitors = vec![];
+        for monitor_config in config.monitors.iter() {
+            let monitor = Monitor::new(
+                monitor_config.provider.clone(),
+                monitor_config.factory,
+                monitor_config.token_one,
+                monitor_config.token_two,
+            )
+            .start();
+            subscribe_to_monitor(monitor.clone(), resolver.clone()).await;
+            monitors.push(monitor);
+        }
 
         // Drive the actors.
-        let mut interval = tokio::time::interval(Duration::from_secs(5));
+        let mut interval = tokio::time::interval(config.tick_rate);
         loop {
             select! {
                 // Shutdown.
                 _ = cancel.cancelled() => {
-                    if let Err(e) = monitor.send(Control::Stop).await {
-                        eprintln!("Failed to send Stop to Monitor: {e}");
+                    for monitor in monitors.into_iter() {
+                        if let Err(e) = monitor.send(Control::Stop).await {
+                            eprintln!("Failed to send Stop to Monitor: {e}");
+                        }
                     }
                     if let Err(e) = resolver.send(Control::Stop).await {
                         eprintln!("Failed to send Stop to Resolver: {e}");
@@ -54,8 +63,10 @@ fn main() -> Result<()> {
                 }
                 // Tick.
                 _ = interval.tick() => {
-                    if let Err(e) = monitor.send(Control::Tick).await {
-                        eprintln!("Failed to send Tick to Monitor: {e}");
+                    for monitor in monitors.iter() {
+                        if let Err(e) = monitor.send(Control::Tick).await {
+                            eprintln!("Failed to send Tick to Monitor: {e}");
+                        }
                     }
                 }
             }
@@ -67,3 +78,11 @@ fn main() -> Result<()> {
     println!("Shutting down");
     Ok(())
 }
+
+async fn subscribe_to_monitor(monitor: Addr<Monitor>, resolver: Addr<Resolver>) {
+    let subscribe = Subscribe::<PoolUpdate>(resolver.recipient());
+    monitor
+        .send(subscribe)
+        .await
+        .expect("Failed to set subscription to Monitor");
+}

crates/uniswapv3pool/src/univ3sdk.rs

@@ -14,6 +14,7 @@ use eyre::Result;
 
 use lib::prelude::*;
 
+#[derive(Debug, Clone)]
 pub struct UniswapV3PoolSdk {
     pub pool: Pool,
     pub tick_data_provider: EphemeralTickDataProvider,