split command line into separte module

Cargo.toml

@@ -8,6 +8,7 @@ members = [
   "rate-counter",
   "redis-rate-limiter",
   "web3_proxy",
+  "web3_proxy_cli",
 ]
 resolver = "2"
 

Dockerfile

@@ -78,7 +78,7 @@ RUN --mount=type=cache,target=/root/.cargo/git \
 FROM rust as rust_with_env
 
 # changing our features doesn't change any of the steps above
-ENV WEB3_PROXY_FEATURES "deadlock_detection,rdkafka-src"
+ENV WEB3_PROXY_FEATURES "rdkafka-src"
 
 # copy the app
 COPY . .
@@ -130,7 +130,7 @@ RUN --mount=type=cache,target=/root/.cargo/git \
     --frozen \
     --no-default-features \
     --offline \
-    --path ./web3_proxy \
+    --path ./web3_proxy_cli \
     --root /usr/local \
     ; \
     /usr/local/bin/web3_proxy_cli --help | grep 'Usage: web3_proxy_cli'

README.md

@@ -50,7 +50,7 @@ RUST_LOG=web3_proxy=trace,info cargo nextest run
 Run more tests:
 
 ```
-RUST_LOG=web3_proxy=trace,info cargo nextest run --features tests-needing-docker
+RUST_LOG=web3_proxy=trace,info cargo nextest run --features tests-needing-docker,tests_utils
 ```
 
 ## Mysql

web3_proxy/Cargo.toml

@@ -1,18 +1,17 @@
 [package]
 name = "web3_proxy"
-version = "1.42.1"
+version = "1.42.2"
 edition = "2021"
-default-run = "web3_proxy_cli"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [features]
-default = ["deadlock_detection"]
-deadlock_detection = ["parking_lot/deadlock_detection"]
+default = []
+
 mimalloc = ["dep:mimalloc"]
-tokio-console = ["dep:tokio-console", "dep:console-subscriber"]
 rdkafka-src = ["rdkafka/cmake-build", "rdkafka/libz", "rdkafka/ssl-vendored", "rdkafka/zstd-pkg-config"]
 tests-needing-docker = []
+tokio-console = ["dep:tokio-console", "dep:console-subscriber"]
 
 [dependencies]
 deferred-rate-limiter = { path = "../deferred-rate-limiter" }
@@ -74,8 +73,6 @@ once_cell = { version = "1.18.0" }
 ordered-float = {version = "3.7.0" }
 pagerduty-rs = { version = "0.1.6", default-features = false, features = ["async", "rustls", "sync"] }
 parking_lot = { version = "0.12.1", features = ["arc_lock", "nightly"] }
-prettytable = "0.10.0"
-proctitle = "0.1.1"
 rdkafka = { version = "0.33.2", features = ["tracing"] }
 regex = "1.9.1"
 reqwest = { version = "0.11.18", default-features = false, features = ["deflate", "gzip", "json", "tokio-rustls"] }
@@ -97,7 +94,6 @@ toml = "0.7.6"
 tower = { version = "0.4.13", features = ["timeout", "tracing"] }
 tower-http = { version = "0.4.3", features = ["cors", "normalize-path", "sensitive-headers", "trace"] }
 tracing = "0.1"
-tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 ulid = { version = "1.0.0", features = ["rand", "uuid", "serde"] }
 url = { version = "2.4.0" }
 uuid = { version = "1.4.1", default-features = false, features = ["fast-rng", "v4", "zerocopy"] }

web3_proxy/src/lib.rs

@@ -16,12 +16,13 @@ pub mod globals;
 pub mod http_params;
 pub mod jsonrpc;
 pub mod pagerduty;
+pub mod prelude;
 pub mod premium;
 pub mod prometheus;
 pub mod referral_code;
 pub mod relational_db;
 pub mod response_cache;
 pub mod rpcs;
 pub mod stats;
-pub mod sub_commands;
+pub mod test_utils;
 pub mod user_token;

web3_proxy/src/prelude.rs

@@ -0,0 +1,34 @@
+pub use anyhow;
+pub use argh;
+pub use chrono;
+pub use entities;
+pub use ethers;
+pub use ethers::prelude::rand;
+pub use fdlimit;
+pub use futures;
+pub use glob;
+pub use hashbrown;
+pub use http;
+pub use influxdb2;
+pub use migration;
+pub use migration::sea_orm;
+pub use moka;
+pub use num;
+pub use ordered_float;
+pub use pagerduty_rs;
+pub use parking_lot;
+pub use rdkafka;
+pub use reqwest;
+pub use rmp_serde;
+pub use rust_decimal;
+pub use sentry;
+pub use sentry_tracing;
+pub use serde;
+pub use serde_inline_default;
+pub use serde_json;
+pub use tokio;
+pub use toml;
+pub use tracing;
+pub use ulid;
+pub use url;
+pub use uuid;

web3_proxy/src/stats/mod.rs

@@ -617,3 +617,73 @@ impl RpcQueryStats {
         Ok(x)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use crate::test_utils::TestInflux;
+    use crate::{caches::UserBalanceCache, stats::StatBuffer};
+    use moka::future::Cache;
+    use tokio::sync::{broadcast, mpsc};
+
+    #[cfg_attr(not(feature = "tests-needing-docker"), ignore)]
+    #[test_log::test(tokio::test)]
+    async fn test_two_buffers() {
+        let i = TestInflux::spawn().await;
+
+        let billing_period_seconds = 86400 * 7;
+        let chain_id = 999_001_999;
+        let db_save_interval_seconds = 60;
+        let influxdb_bucket = Some(i.bucket.clone());
+        let influxdb_client = Some(i.client.clone());
+        let rpc_secret_key_cache = Cache::builder().build();
+        let tsdb_save_interval_seconds = 30;
+        let user_balance_cache: UserBalanceCache = Cache::builder().build().into();
+
+        let (shutdown_sender, shutdown_receiver_1) = broadcast::channel(1);
+        let shutdown_receiver_2 = shutdown_sender.subscribe();
+
+        let (flush_sender_1, flush_receiver_1) = mpsc::channel(1);
+        let (flush_sender_2, flush_receiver_2) = mpsc::channel(1);
+
+        let buffer_1 = StatBuffer::try_spawn(
+            billing_period_seconds,
+            chain_id,
+            db_save_interval_seconds,
+            influxdb_bucket.clone(),
+            influxdb_client.clone(),
+            rpc_secret_key_cache.clone(),
+            user_balance_cache.clone(),
+            shutdown_receiver_1,
+            tsdb_save_interval_seconds,
+            flush_sender_1,
+            flush_receiver_1,
+            1,
+        )
+        .unwrap()
+        .unwrap();
+
+        let buffer_2 = StatBuffer::try_spawn(
+            billing_period_seconds,
+            chain_id,
+            db_save_interval_seconds,
+            influxdb_bucket,
+            influxdb_client,
+            rpc_secret_key_cache,
+            user_balance_cache,
+            shutdown_receiver_2,
+            tsdb_save_interval_seconds,
+            flush_sender_2,
+            flush_receiver_2,
+            2,
+        )
+        .unwrap()
+        .unwrap();
+
+        // TODO: send things to the buffers
+
+        shutdown_sender.send(()).unwrap();
+
+        buffer_1.background_handle.await.unwrap().unwrap();
+        buffer_2.background_handle.await.unwrap().unwrap();
+    }
+}

web3_proxy/tests/common/anvil.rs → web3_proxy/src/test_utils/anvil.rs

@@ -1,12 +1,12 @@
 // TODO: option to spawn in a dedicated thread?
 // TODO: option to subscribe to another anvil and copy blocks
 
+use crate::rpcs::provider::EthersHttpProvider;
 use ethers::{
     signers::LocalWallet,
     utils::{Anvil, AnvilInstance},
 };
 use tracing::info;
-use web3_proxy::rpcs::provider::EthersHttpProvider;
 
 /// on drop, the anvil instance will be shut down
 pub struct TestAnvil {
@@ -15,7 +15,6 @@ pub struct TestAnvil {
 }
 
 impl TestAnvil {
-    #[allow(unused)]
     pub async fn spawn(chain_id: u64) -> Self {
         info!(?chain_id);
 
@@ -30,7 +29,6 @@ impl TestAnvil {
         Self { instance, provider }
     }
 
-    #[allow(unused)]
     pub fn wallet(&self, id: usize) -> LocalWallet {
         self.instance.keys()[id].clone().into()
     }

web3_proxy/src/test_utils/create_provider_with_rpc_key.rs

@@ -0,0 +1,10 @@
+use crate::rpcs::provider::EthersHttpProvider;
+use ulid::Ulid;
+use url::Url;
+
+pub async fn create_provider_for_user(url: &Url, user_secret_key: &Ulid) -> EthersHttpProvider {
+    // Then generate a provider
+    let proxy_endpoint = format!("{}rpc/{}", url, user_secret_key);
+
+    EthersHttpProvider::try_from(proxy_endpoint).unwrap()
+}

web3_proxy/tests/common/influx.rs → web3_proxy/src/test_utils/influx.rs

@@ -1,12 +1,12 @@
-use ethers::prelude::rand::{self, distributions::Alphanumeric, Rng};
-use influxdb2::Client;
-use std::process::Command as SyncCommand;
-use std::time::Duration;
-use tokio::{
+use crate::prelude::influxdb2::Client;
+use crate::prelude::rand::{self, distributions::Alphanumeric, Rng};
+use crate::prelude::tokio::{
     net::TcpStream,
     process::Command as AsyncCommand,
     time::{sleep, Instant},
 };
+use std::process::Command as SyncCommand;
+use std::time::Duration;
 use tracing::{info, trace};
 
 /// on drop, the mysql docker container will be shut down
@@ -21,7 +21,6 @@ pub struct TestInflux {
 }
 
 impl TestInflux {
-    #[allow(unused)]
     pub async fn spawn() -> Self {
         let random: String = rand::thread_rng()
             .sample_iter(&Alphanumeric)
@@ -136,7 +135,7 @@ impl TestInflux {
 
         // create the TestInflux as soon as the url is known
         // when this is dropped, the docker container will be stopped
-        let mut test_influx = Self {
+        let test_influx = Self {
             host: influx_host,
             org: org.to_string(),
             token: admin_token.to_string(),

web3_proxy/src/test_utils/lib.rs

@@ -0,0 +1,3 @@
+fn main() {
+    println!("Hello, world!");
+}

web3_proxy/src/test_utils/mod.rs

@@ -0,0 +1,8 @@
+pub mod anvil;
+pub mod create_provider_with_rpc_key;
+pub mod influx;
+pub mod mysql;
+
+pub use self::anvil::TestAnvil;
+pub use self::influx::TestInflux;
+pub use self::mysql::TestMysql;

web3_proxy/tests/common/mysql.rs → web3_proxy/src/test_utils/mysql.rs

@@ -1,3 +1,4 @@
+use crate::relational_db::{connect_db, get_migrated_db};
 use ethers::prelude::rand::{self, distributions::Alphanumeric, Rng};
 use migration::sea_orm::DatabaseConnection;
 use std::process::Command as SyncCommand;
@@ -8,7 +9,6 @@ use tokio::{
     time::{sleep, Instant},
 };
 use tracing::{info, trace, warn};
-use web3_proxy::relational_db::{connect_db, get_migrated_db};
 
 /// on drop, the mysql docker container will be shut down
 pub struct TestMysql {
@@ -17,7 +17,6 @@ pub struct TestMysql {
 }
 
 impl TestMysql {
-    #[allow(unused)]
     pub async fn spawn() -> Self {
         let password: String = rand::thread_rng()
             .sample_iter(&Alphanumeric)