Tx Sitter Integration (#661)

Cargo.toml

@@ -76,6 +76,7 @@ tokio = { version = "1.17", features = [
 ] }
 tracing = "0.1"
 tracing-futures = "0.2"
+tx-sitter-client = { path = "crates/tx-sitter-client" }
 url = { version = "2.2", features = ["serde"] }
 # `ethers-rs` requires an older version of primitive-types.
 # But `ruint` supports the latest version. So we need to override it.

compose.yml

@@ -0,0 +1,36 @@
+version: '3'
+services:
+  chain:
+    image: ghcr.io/foundry-rs/foundry
+    ports:
+      - 8545:8545
+    command: ["anvil --host 0.0.0.0 --chain-id 31337 --block-time 2 --fork-url https://eth-sepolia.g.alchemy.com/v2/rbe0ZJX0TXJ7udcCB07HHggMt5x5Rcy9@4910128"]
+  tx-sitter-db:
+    image: postgres:latest
+    environment:
+      - POSTGRES_USER=postgres
+      - POSTGRES_PASSWORD=postgres
+      - POSTGRES_DB=tx-sitter
+  sequencer-db:
+    image: postgres:latest
+    environment:
+      - POSTGRES_USER=postgres
+      - POSTGRES_PASSWORD=postgres
+      - POSTGRES_DB=sequencer
+  tx-sitter:
+    image: ghcr.io/worldcoin/tx-sitter-monolith:dev
+    ports:
+      - 3000:3000
+    environment:
+      - TX_SITTER__SERVICE__ESCALATION_INTERVAL=1m
+      - TX_SITTER__DATABASE__KIND=connection_string
+      - TX_SITTER__DATABASE__CONNECTION_STRING=postgres://postgres:postgres@tx-sitter-db:5432/tx-sitter?sslmode=disable
+      - TX_SITTER__KEYS__KIND=local
+      - TX_SITTER__PREDEFINED__NETWORK__CHAIN_ID=31337
+      - TX_SITTER__PREDEFINED__NETWORK__HTTP_URL=http://chain:8545
+      - TX_SITTER__PREDEFINED__NETWORK__WS_URL=ws://chain:8545
+      - TX_SITTER__PREDEFINED__RELAYER__ID=1b908a34-5dc1-4d2d-a146-5eb46e975830
+      - TX_SITTER__PREDEFINED__RELAYER__CHAIN_ID=31337
+      - TX_SITTER__PREDEFINED__RELAYER__KEY_ID=d10607662a85424f02a33fb1e6d095bd0ac7154396ff09762e41f82ff2233aaa
+      - TX_SITTER__SERVER__HOST=0.0.0.0:3000
+      - TX_SITTER__SERVER__DISABLE_AUTH=true

crates/tx-sitter-client/Cargo.toml

@@ -0,0 +1,14 @@
+[package]
+name = "tx-sitter-client"
+version = "0.1.0"
+edition = "2021"
+publish = false
+
+[dependencies]
+anyhow = "1.0"
+ethers = { version = "2.0.10", features = [ ] }
+reqwest = "0.11.14"
+serde = { version = "1.0.154", features = ["derive"] }
+serde_json = "1.0.94"
+strum = { version = "0.25", features = ["derive"] }
+tracing = "0.1"

crates/tx-sitter-client/src/data.rs

@@ -0,0 +1,72 @@
+use ethers::types::{Address, Bytes, H256, U256};
+use serde::{Deserialize, Serialize};
+use strum::Display;
+
+mod decimal_u256;
+
+#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct SendTxRequest {
+    pub to:        Address,
+    #[serde(with = "decimal_u256")]
+    pub value:     U256,
+    #[serde(default)]
+    pub data:      Option<Bytes>,
+    #[serde(with = "decimal_u256")]
+    pub gas_limit: U256,
+    #[serde(default)]
+    pub priority:  TransactionPriority,
+    #[serde(default)]
+    pub tx_id:     Option<String>,
+}
+
+#[derive(Deserialize, Serialize, Debug, Clone, Copy, Default)]
+#[serde(rename_all = "camelCase")]
+pub enum TransactionPriority {
+    // 5th percentile
+    Slowest = 0,
+    // 25th percentile
+    Slow    = 1,
+    // 50th percentile
+    #[default]
+    Regular = 2,
+    // 75th percentile
+    Fast    = 3,
+    // 95th percentile
+    Fastest = 4,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct SendTxResponse {
+    pub tx_id: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct GetTxResponse {
+    pub tx_id:     String,
+    pub to:        Address,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub data:      Option<Bytes>,
+    #[serde(with = "decimal_u256")]
+    pub value:     U256,
+    #[serde(with = "decimal_u256")]
+    pub gas_limit: U256,
+    pub nonce:     u64,
+
+    // Sent tx data
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub tx_hash: Option<H256>,
+    pub status:  TxStatus,
+}
+
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, Display, PartialEq, Eq)]
+#[serde(rename_all = "camelCase")]
+#[strum(serialize_all = "camelCase")]
+pub enum TxStatus {
+    Unsent,
+    Pending,
+    Mined,
+    Finalized,
+}

crates/tx-sitter-client/src/data/decimal_u256.rs

@@ -0,0 +1,42 @@
+use ethers::types::U256;
+
+pub fn serialize<S>(u256: &U256, serializer: S) -> Result<S::Ok, S::Error>
+where
+    S: serde::Serializer,
+{
+    let s = u256.to_string();
+    serializer.serialize_str(&s)
+}
+
+pub fn deserialize<'de, D>(deserializer: D) -> Result<U256, D::Error>
+where
+    D: serde::Deserializer<'de>,
+{
+    let s: &str = serde::Deserialize::deserialize(deserializer)?;
+    let u256 = U256::from_dec_str(s).map_err(serde::de::Error::custom)?;
+    Ok(u256)
+}
+
+#[cfg(test)]
+mod tests {
+    use serde::{Deserialize, Serialize};
+
+    use super::*;
+
+    #[derive(Debug, Clone, Serialize, Deserialize)]
+    struct Test {
+        #[serde(with = "super")]
+        v: U256,
+    }
+
+    #[test]
+    fn test_u256_serde() {
+        let test = Test { v: U256::from(123) };
+
+        let s = serde_json::to_string(&test).unwrap();
+        assert_eq!(s, r#"{"v":"123"}"#);
+
+        let test: Test = serde_json::from_str(&s).unwrap();
+        assert_eq!(test.v, U256::from(123));
+    }
+}

crates/tx-sitter-client/src/lib.rs

@@ -0,0 +1,81 @@
+use data::{GetTxResponse, SendTxRequest, SendTxResponse, TxStatus};
+use reqwest::Response;
+use tracing::instrument;
+
+pub mod data;
+
+pub struct TxSitterClient {
+    client: reqwest::Client,
+    url:    String,
+}
+
+impl TxSitterClient {
+    pub fn new(url: impl ToString) -> Self {
+        Self {
+            client: reqwest::Client::new(),
+            url:    url.to_string(),
+        }
+    }
+
+    async fn json_post<T, R>(&self, url: &str, body: T) -> anyhow::Result<R>
+    where
+        T: serde::Serialize,
+        R: serde::de::DeserializeOwned,
+    {
+        let response = self.client.post(url).json(&body).send().await?;
+
+        let response = Self::validate_response(response).await?;
+
+        Ok(response.json().await?)
+    }
+
+    async fn json_get<R>(&self, url: &str) -> anyhow::Result<R>
+    where
+        R: serde::de::DeserializeOwned,
+    {
+        let response = self.client.get(url).send().await?;
+
+        let response = Self::validate_response(response).await?;
+
+        Ok(response.json().await?)
+    }
+
+    async fn validate_response(response: Response) -> anyhow::Result<Response> {
+        if !response.status().is_success() {
+            let status = response.status();
+            let body = response.text().await?;
+
+            tracing::error!("Response failed with status {} - {}", status, body);
+            return Err(anyhow::anyhow!(
+                "Response failed with status {status} - {body}"
+            ));
+        }
+
+        Ok(response)
+    }
+
+    #[instrument(skip(self))]
+    pub async fn send_tx(&self, req: &SendTxRequest) -> anyhow::Result<SendTxResponse> {
+        self.json_post(&format!("{}/tx", self.url), req).await
+    }
+
+    #[instrument(skip(self))]
+    pub async fn get_tx(&self, tx_id: &str) -> anyhow::Result<GetTxResponse> {
+        self.json_get(&format!("{}/tx/{}", self.url, tx_id)).await
+    }
+
+    #[instrument(skip(self))]
+    pub async fn get_txs(&self, tx_status: Option<TxStatus>) -> anyhow::Result<Vec<GetTxResponse>> {
+        let mut url = format!("{}/txs", self.url);
+
+        if let Some(tx_status) = tx_status {
+            url.push_str(&format!("?status={}", tx_status));
+        }
+
+        self.json_get(&url).await
+    }
+
+    pub fn rpc_url(&self) -> String {
+        format!("{}/rpc", self.url.clone())
+    }
+}

src/ethereum/mod.rs

@@ -10,13 +10,14 @@ use tracing::instrument;
 use url::Url;
 pub use write::TxError;
 
-use self::write::{TransactionId, WriteProvider};
+use self::write::TransactionId;
+use self::write_provider::WriteProvider;
 use crate::serde_utils::JsonStrWrapper;
 
 pub mod read;
 pub mod write;
 
-mod write_oz;
+mod write_provider;
 
 // TODO: Log and metrics for signer / nonces.
 #[derive(Clone, Debug, PartialEq, Parser)]
@@ -31,15 +32,15 @@ pub struct Options {
     pub secondary_providers: JsonStrWrapper<Vec<Url>>,
 
     #[clap(flatten)]
-    pub write_options: write_oz::Options,
+    pub write_options: write_provider::Options,
 }
 
 #[derive(Clone, Debug)]
 pub struct Ethereum {
     read_provider:            Arc<ReadProvider>,
     // Mapping of chain id to provider
     secondary_read_providers: HashMap<u64, Arc<ReadProvider>>,
-    write_provider:           Arc<dyn WriteProvider>,
+    write_provider:           Arc<WriteProvider>,
 }
 
 impl Ethereum {
@@ -57,8 +58,10 @@ impl Ethereum {
             );
         }
 
-        let write_provider: Arc<dyn WriteProvider> =
-            Arc::new(write_oz::Provider::new(read_provider.clone(), &options.write_options).await?);
+        let write_provider: Arc<WriteProvider> = Arc::new(
+            write_provider::WriteProvider::new(read_provider.clone(), &options.write_options)
+                .await?,
+        );
 
         Ok(Self {
             read_provider: Arc::new(read_provider),

src/ethereum/write/mod.rs

@@ -1,10 +1,8 @@
 use std::error::Error;
 use std::fmt;
 
-use async_trait::async_trait;
 use ethers::providers::ProviderError;
-use ethers::types::transaction::eip2718::TypedTransaction;
-use ethers::types::{Address, TransactionReceipt, H256};
+use ethers::types::{TransactionReceipt, H256};
 use thiserror::Error;
 
 #[derive(Clone, Debug)]
@@ -35,7 +33,7 @@ pub enum TxError {
     SendTimeout,
 
     #[error("Error sending transaction: {0}")]
-    Send(Box<dyn Error + Send + Sync + 'static>),
+    Send(anyhow::Error),
 
     #[error("Timeout while waiting for confirmations")]
     ConfirmationTimeout,
@@ -51,19 +49,7 @@ pub enum TxError {
 
     #[error("Error parsing transaction id: {0}")]
     Parse(Box<dyn Error + Send + Sync + 'static>),
-}
-
-#[async_trait]
-pub trait WriteProvider: Sync + Send + fmt::Debug {
-    async fn send_transaction(
-        &self,
-        tx: TypedTransaction,
-        only_once: bool,
-    ) -> Result<TransactionId, TxError>;
-
-    async fn fetch_pending_transactions(&self) -> Result<Vec<TransactionId>, TxError>;
-
-    async fn mine_transaction(&self, tx: TransactionId) -> Result<bool, TxError>;
 
-    fn address(&self) -> Address;
+    #[error("{0}")]
+    Other(anyhow::Error),
 }