Merge pull request #1 from worldcoin/dev

Initial PR

Cargo.toml

@@ -0,0 +1,76 @@
+[package]
+name = "service"
+version = "0.1.0"
+edition = "2021"
+
+[workspace]
+members = ["crates/*"]
+
+[dependencies]
+# Third Party
+## AWS
+aws-config = { version = "0.57.2" }
+aws-sdk-kms = "0.36.0"
+aws-smithy-types = "0.57.2"
+aws-types = "0.57.2"
+aws-credential-types = { version = "0.57.2", features = [
+    "hardcoded-credentials",
+] }
+
+## Other
+serde = "1.0.136"
+axum = { version = "0.6.20", features = ["headers"] }
+thiserror = "1.0.50"
+headers = "0.3.9"
+humantime = "2.1.0"
+humantime-serde = "1.1.1"
+hyper = "0.14.27"
+dotenv = "0.15.0"
+clap = { version = "4.3.0", features = ["env", "derive"] }
+ethers = { version = "2.0.11" }
+ethers-signers = { version = "2.0.11" }
+eyre = "0.6.5"
+hex = "0.4.3"
+hex-literal = "0.4.1"
+reqwest = { version = "0.11.13", default-features = false, features = [
+    "rustls-tls",
+] }
+serde_json = "1.0.91"
+strum = { version = "0.25.0", features = ["derive"] }
+tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
+tracing = { version = "0.1", features = ["log"] }
+tracing-subscriber = { version = "0.3", default-features = false, features = [
+    "env-filter",
+    "std",
+    "fmt",
+    "json",
+    "ansi",
+] }
+tower-http = { version = "0.4.4", features = ["trace"] }
+uuid = { version = "0.8", features = ["v4"] }
+futures = "0.3"
+chrono = "0.4"
+rand = "0.8.5"
+sha3 = "0.10.8"
+config = "0.13.3"
+toml = "0.8.8"
+sqlx = { version = "0.7.2", features = [
+    "runtime-tokio",
+    "tls-rustls",
+    "postgres",
+    "migrate",
+] }
+spki = "0.7.2"
+async-trait = "0.1.74"
+
+# Internal
+postgres-docker-utils = { path = "crates/postgres-docker-utils" }
+
+[dev-dependencies]
+test-case = "3.1.0"
+indoc = "2.0.3"
+fake-rpc = { path = "crates/fake-rpc" }
+
+[[bin]]
+name = "bootstrap"
+path = "src/main.rs"

README.md

@@ -1,2 +1,20 @@
-# tx-sitter-monolith
-Transaction sitter. But a glorious monolith.
+# Tx Sitter Monolith
+
+A monolithized version of the [tx-sitter](https://github.com/worldcoin/tx-sitter-aws/).
+
+## Testing locally
+
+Copy `.env.example` to `.env` or set `RUST_LOG=info,service=debug` to have logging.
+
+1. Spin up the database `docker run --rm -e POSTGRES_HOST_AUTH_METHOD=trust -p 5432:5432 postgres`
+2. Spin up the chain `anvil --chain-id 31337 --block-time 2`
+3. Start the service `cargo run`
+
+This will use the `config.toml` configuration.
+
+If you have [nushell](https://www.nushell.sh/) installed, `nu manual_test.nu` can be run to execute a basic test.
+
+## Running tests
+While you obviously can run tests with `cargo test --workspace` some tests take quite a long time (due to spinning up an anvil node, sending txs, etc.).
+
+Therefore I recommend [cargo-nextest](https://nexte.st/) as it runs all the tests in parallel. Once installed `cargo nextest run --workspace` can be used instead.

TODO.md

@@ -0,0 +1,15 @@
+# TODO
+
+1. [ ] Handling reorgs
+2. [ ] Per network settings (i.e. max inflight txs, max gas price)
+3. [ ] Multiple RPCs support
+4. [ ] Cross-network dependencies (i.e. Optimism depends on L1 gas cost)
+5. [ ] Transaction priority
+6. [ ] Metrics
+7. [ ] Tracing (add telemetry-batteries)
+8. [ ] Automated testing
+   1. [x] Basic
+   2. [ ] Basic with contracts
+   3. [ ] Escalation testing
+   4. [ ] Reorg testing (how?!?)
+9.  [ ] Parallelization in a few places

config.toml

@@ -0,0 +1,20 @@
+[service]
+escalation_interval = "1m"
+
+[server]
+host = "127.0.0.1:3000"
+disable_auth = true
+
+[rpc]
+rpcs = ["http://127.0.0.1:8545"]
+
+[database]
+connection_string = "postgres://postgres:[email protected]:5432/database"
+
+[keys]
+kind = "local"
+
+# Example KMS configuration
+# [keys]
+# kind = "kms"
+# region = "us-east-1"

crates/fake-rpc/Cargo.toml

@@ -0,0 +1,52 @@
+[package]
+name = "fake-rpc"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+serde = "1.0.136"
+axum = { version = "0.6.20", features = ["headers"] }
+thiserror = "1.0.50"
+headers = "0.3.9"
+humantime = "2.1.0"
+humantime-serde = "1.1.1"
+hyper = "0.14.27"
+dotenv = "0.15.0"
+clap = { version = "4.3.0", features = ["env", "derive"] }
+ethers = { version = "2.0.11" }
+ethers-signers = { version = "2.0.11" }
+eyre = "0.6.5"
+hex = "0.4.3"
+hex-literal = "0.4.1"
+reqwest = { version = "0.11.13", default-features = false, features = [
+    "rustls-tls",
+] }
+serde_json = "1.0.91"
+strum = { version = "0.25.0", features = ["derive"] }
+tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
+tracing = { version = "0.1", features = ["log"] }
+tracing-subscriber = { version = "0.3", default-features = false, features = [
+    "env-filter",
+    "std",
+    "fmt",
+    "json",
+    "ansi",
+] }
+tower-http = { version = "0.4.4", features = ["trace"] }
+uuid = { version = "0.8", features = ["v4"] }
+futures = "0.3"
+chrono = "0.4"
+rand = "0.8.5"
+sha3 = "0.10.8"
+config = "0.13.3"
+toml = "0.8.8"
+sqlx = { version = "0.7.2", features = [
+    "runtime-tokio",
+    "tls-rustls",
+    "postgres",
+    "migrate",
+] }
+spki = "0.7.2"
+async-trait = "0.1.74"

crates/fake-rpc/src/lib.rs

@@ -0,0 +1,144 @@
+use std::net::{Ipv4Addr, SocketAddr};
+use std::sync::atomic::AtomicBool;
+use std::sync::Arc;
+
+use axum::extract::State;
+use axum::routing::{post, IntoMakeService};
+use axum::{Json, Router};
+use ethers::utils::{Anvil, AnvilInstance};
+use hyper::server::conn::AddrIncoming;
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+use tokio::sync::Mutex;
+
+pub struct DoubleAnvil {
+    main_anvil: Mutex<AnvilInstance>,
+    reference_anvil: Mutex<AnvilInstance>,
+    held_back_txs: Mutex<Vec<JsonRpcReq>>,
+
+    auto_advance: AtomicBool,
+}
+
+impl DoubleAnvil {
+    pub async fn drop_txs(&self) -> eyre::Result<()> {
+        let mut held_back_txs = self.held_back_txs.lock().await;
+        held_back_txs.clear();
+        Ok(())
+    }
+
+    pub async fn advance(&self) -> eyre::Result<()> {
+        let mut held_back_txs = self.held_back_txs.lock().await;
+
+        for req in held_back_txs.drain(..) {
+            tracing::info!(?req, "eth_sendRawTransaction");
+
+            let response = reqwest::Client::new()
+                .post(&self.main_anvil.lock().await.endpoint())
+                .json(&req)
+                .send()
+                .await
+                .unwrap();
+
+            let resp = response.json::<Value>().await.unwrap();
+
+            tracing::info!(?resp, "eth_sendRawTransaction.response");
+        }
+
+        Ok(())
+    }
+
+    pub fn set_auto_advance(&self, auto_advance: bool) {
+        self.auto_advance
+            .store(auto_advance, std::sync::atomic::Ordering::SeqCst);
+    }
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+struct JsonRpcReq {
+    pub id: u64,
+    pub jsonrpc: String,
+    pub method: String,
+    #[serde(default)]
+    pub params: Vec<Value>,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct JsonRpcResponse {
+    pub id: u64,
+    pub jsonrpc: String,
+    pub result: Value,
+}
+
+async fn advance(State(anvil): State<Arc<DoubleAnvil>>) {
+    anvil.advance().await.unwrap();
+}
+
+async fn rpc(
+    State(anvil): State<Arc<DoubleAnvil>>,
+    Json(req): Json<JsonRpcReq>,
+) -> Json<Value> {
+    let method = req.method.as_str();
+    let anvil_instance = match method {
+        "eth_sendRawTransaction" => {
+            anvil.held_back_txs.lock().await.push(req.clone());
+
+            if anvil.auto_advance.load(std::sync::atomic::Ordering::SeqCst) {
+                anvil.advance().await.unwrap();
+            }
+
+            anvil.reference_anvil.lock().await
+        }
+        "eth_getTransactionReceipt" => anvil.main_anvil.lock().await,
+        "eth_getTransactionByHash" => anvil.reference_anvil.lock().await,
+        _ => anvil.main_anvil.lock().await,
+    };
+
+    tracing::info!(?req, "{}", method);
+
+    let response = reqwest::Client::new()
+        .post(&anvil_instance.endpoint())
+        .json(&req)
+        .send()
+        .await
+        .unwrap();
+
+    let resp = response.json::<Value>().await.unwrap();
+
+    tracing::info!(?resp, "{}.response", method);
+
+    Json(resp)
+}
+
+pub async fn serve(
+    port: u16,
+) -> (
+    Arc<DoubleAnvil>,
+    axum::Server<AddrIncoming, IntoMakeService<Router>>,
+) {
+    let main_anvil = Anvil::new().spawn();
+    let reference_anvil = Anvil::new().spawn();
+
+    tracing::info!("Main anvil instance: {}", main_anvil.endpoint());
+    tracing::info!("Reference anvil instance: {}", reference_anvil.endpoint());
+
+    let state = Arc::new(DoubleAnvil {
+        main_anvil: Mutex::new(main_anvil),
+        reference_anvil: Mutex::new(reference_anvil),
+        held_back_txs: Mutex::new(Vec::new()),
+        auto_advance: AtomicBool::new(true),
+    });
+
+    let router = Router::new()
+        .route("/", post(rpc))
+        .route("/advance", post(advance))
+        .with_state(state.clone())
+        .layer(tower_http::trace::TraceLayer::new_for_http());
+
+    let host = Ipv4Addr::new(127, 0, 0, 1);
+    let socket_addr = SocketAddr::new(host.into(), port);
+
+    let server =
+        axum::Server::bind(&socket_addr).serve(router.into_make_service());
+
+    (state, server)
+}

crates/fake-rpc/src/main.rs

@@ -0,0 +1,30 @@
+use clap::Parser;
+use tracing_subscriber::layer::SubscriberExt;
+use tracing_subscriber::util::SubscriberInitExt;
+use tracing_subscriber::EnvFilter;
+
+#[derive(Debug, Clone, Parser)]
+struct Args {
+    #[clap(short, long, default_value = "8545")]
+    port: u16,
+}
+
+#[tokio::main]
+async fn main() -> eyre::Result<()> {
+    dotenv::dotenv().ok();
+
+    tracing_subscriber::registry()
+        .with(tracing_subscriber::fmt::layer().pretty().compact())
+        .with(EnvFilter::from_default_env())
+        .init();
+
+    let args = Args::parse();
+
+    let (_app, server) = fake_rpc::serve(args.port).await;
+
+    tracing::info!("Serving fake RPC at {}", server.local_addr());
+
+    server.await?;
+
+    Ok(())
+}

crates/postgres-docker-utils/Cargo.toml

@@ -0,0 +1,10 @@
+[package]
+name = "postgres-docker-utils"
+version = "0.1.0"
+edition = "2021"
+publish = false
+
+[dependencies]
+eyre = "0.6.8"
+test-case = "3.1.0"
+tokio = { version = "1.0", features = ["full"] }