feat: rebalancer service

Author: efbig

Cargo.lock

@@ -13,6 +13,7 @@ members = [
     "crates/rbuilder-config",
     "crates/rbuilder",
     "crates/rbuilder-operator",
+    "crates/rbuilder-rebalancer",
     "crates/reth-rbuilder",
     "crates/rbuilder/src/test_utils",
     "crates/rbuilder/src/telemetry/metrics_macros",

Cargo.toml

@@ -20,7 +20,7 @@ tokio-util.workspace = true
 tracing = "0.1.37"
 tracing-subscriber = { version = "0.3.18", features = ["env-filter", "json"] }
 log = "^0.4"
-clap = { version = "^3.2", features = ["derive"] }
+clap.workspace = true
 serde_json = { workspace = true }
 serde = { workspace = true }
 serde_with = { version = "3.9.0", features = ["time_0_3"] }
@@ -41,7 +41,7 @@ ssz_types = "^0.5"
 hex = "^0.4"
 derivative.workspace = true
 toml.workspace = true
-eyre = "0.6.12"
+eyre.workspace = true
 thiserror.workspace = true
 parking_lot.workspace = true
 strum = { version = "0.25", features = ["derive"] }

crates/bid-scraper/Cargo.toml

@@ -10,7 +10,7 @@ repository.workspace = true
 
 [dependencies]
 thiserror = "1.0.61"
-serde = { version = "1.0.203", features = ["derive"] }
+serde = { workspace = true, features = ["derive"] }
 serde_json = "1.0.117"
 serde_with = "3.9.0"
 rustc-hash = "2.0.0"

crates/eth-sparse-mpt/Cargo.toml

@@ -1,13 +1,26 @@
 use tracing_subscriber::EnvFilter;
 
 /// Logger configuration.
-#[derive(Debug, Clone)]
+#[derive(PartialEq, Eq, Clone, Debug, serde::Deserialize)]
 pub struct LoggerConfig {
     pub env_filter: String,
+    #[serde(default)]
     pub log_json: bool,
+    #[serde(default)]
     pub log_color: bool,
 }
 
+impl LoggerConfig {
+    /// Default logger configuration for development.
+    pub fn dev() -> Self {
+        Self {
+            env_filter: String::from("info"),
+            log_color: true,
+            log_json: false,
+        }
+    }
+}
+
 impl LoggerConfig {
     /// Initialize tracing subscriber based on the configuration.
     pub fn init_tracing(self) -> eyre::Result<()> {

crates/rbuilder-config/src/logger.rs

@@ -0,0 +1,32 @@
+[package]
+name = "rbuilder-rebalancer"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+homepage.workspace = true
+repository.workspace = true
+
+[dependencies]
+rbuilder-config.workspace = true
+
+alloy-primitives = { workspace = true, features = ["serde"] }
+alloy-eips.workspace = true
+alloy-consensus.workspace = true
+alloy-rpc-types-eth.workspace = true
+alloy-provider.workspace = true
+alloy-signer.workspace = true
+alloy-signer-local.workspace = true
+
+# rt
+tokio = { workspace = true, default-features = false, features = ["macros", "rt-multi-thread"] }
+futures.workspace = true
+
+# misc
+serde = { workspace = true, features = ["derive"] }
+serde_json.workspace = true
+clap.workspace = true
+eyre.workspace = true
+toml.workspace = true
+tracing.workspace = true
+reqwest.workspace = true

crates/rbuilder-rebalancer/Cargo.toml

@@ -0,0 +1,68 @@
+use alloy_provider::ProviderBuilder;
+use alloy_signer_local::PrivateKeySigner;
+use clap::Parser;
+use rbuilder_rebalancer::{config::RebalancerConfig, rebalancer::Rebalancer};
+use std::{path::PathBuf, str::FromStr, time::Duration};
+use tracing::{level_filters::LevelFilter, *};
+
+#[tokio::main]
+async fn main() {
+    if let Err(error) = Cli::parse().run().await {
+        eprintln!("Error: {error:?}");
+        std::process::exit(1);
+    }
+}
+
+#[derive(Parser)]
+struct Cli {
+    #[clap(env = "REBALANCER_CONFIG", help = "Config file path")]
+    config: PathBuf,
+}
+
+impl Cli {
+    async fn run(self) -> eyre::Result<()> {
+        let config = RebalancerConfig::parse_toml_file(&self.config)?;
+
+        config.logger.init_tracing()?;
+
+        if config.rules.is_empty() {
+            warn!("No rebalancing rules have been configured, rebalancer will be idling");
+        }
+
+        for rule in &config.rules {
+            if rule.destination_min_balance >= rule.destination_target_balance {
+                eyre::bail!("Invalid configuration for rule `{}`: minimum balance must be lower than the target", rule.description);
+            }
+
+            if !config.accounts.iter().any(|acc| acc.id == rule.source_id) {
+                eyre::bail!("Invalid configuration for rule `{}`: account entry is missing for source account {}", rule.description, rule.source_id);
+            }
+        }
+
+        let rpc_provider = ProviderBuilder::new().connect(&config.rpc_url).await?;
+        let transfer_max_priority_fee_per_gas =
+            config.transfer_max_priority_fee_per_gas.try_into().unwrap();
+        let accounts = config
+            .accounts
+            .into_iter()
+            .map(|account| {
+                let account = account.map_secret(|secret| {
+                    let secret = secret.value().expect("invalid env");
+                    PrivateKeySigner::from_str(&secret).expect("invalid private key")
+                });
+                (account.id.clone(), account)
+            })
+            .collect();
+        Rebalancer::new(
+            rpc_provider,
+            config.builder_url,
+            transfer_max_priority_fee_per_gas,
+            accounts,
+            config.rules,
+            Duration::from_secs(2),
+            Duration::from_secs(2),
+        )
+        .run()
+        .await
+    }
+}