Ethereum relayer server

Causes send test to pass for the processor.

Cargo.toml

@@ -38,8 +38,11 @@ members = [
   "crypto/schnorrkel",
 
   "coins/bitcoin",
+
   "coins/ethereum/alloy-simple-request-transport",
   "coins/ethereum",
+  "coins/ethereum/relayer",
+
   "coins/monero/generators",
   "coins/monero",
 

coins/ethereum/relayer/Cargo.toml

@@ -0,0 +1,30 @@
+[package]
+name = "serai-ethereum-relayer"
+version = "0.1.0"
+description = "A relayer for Serai's Ethereum transactions"
+license = "AGPL-3.0-only"
+repository = "https://github.com/serai-dex/serai/tree/develop/coins/ethereum/relayer"
+authors = ["Luke Parker <[email protected]>"]
+keywords = []
+edition = "2021"
+publish = false
+
+[package.metadata.docs.rs]
+all-features = true
+rustdoc-args = ["--cfg", "docsrs"]
+
+[lints]
+workspace = true
+
+[dependencies]
+log = { version = "0.4", default-features = false, features = ["std"] }
+env_logger = { version = "0.10", default-features = false, features = ["humantime"] }
+
+tokio = { version = "1", default-features = false, features = ["rt", "time", "io-util", "net", "macros"] }
+
+serai-env = { path = "../../../common/env" }
+serai-db = { path = "../../../common/db" }
+
+[features]
+parity-db = ["serai-db/parity-db"]
+rocksdb = ["serai-db/rocksdb"]

coins/ethereum/relayer/LICENSE

@@ -0,0 +1,15 @@
+AGPL-3.0-only license
+
+Copyright (c) 2023-2024 Luke Parker
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU Affero General Public License Version 3 as
+published by the Free Software Foundation.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+GNU Affero General Public License for more details.
+
+You should have received a copy of the GNU Affero General Public License
+along with this program. If not, see <http://www.gnu.org/licenses/>.

coins/ethereum/relayer/README.md

@@ -0,0 +1,4 @@
+# Ethereum Transaction Relayer
+
+This server collects Ethereum router commands to be published, offering an RPC
+to fetch them.

coins/ethereum/relayer/src/main.rs

@@ -0,0 +1,100 @@
+pub(crate) use tokio::{
+  io::{AsyncReadExt, AsyncWriteExt},
+  net::TcpListener,
+};
+
+use serai_db::{Get, DbTxn, Db as DbTrait};
+
+#[tokio::main(flavor = "current_thread")]
+async fn main() {
+  // Override the panic handler with one which will panic if any tokio task panics
+  {
+    let existing = std::panic::take_hook();
+    std::panic::set_hook(Box::new(move |panic| {
+      existing(panic);
+      const MSG: &str = "exiting the process due to a task panicking";
+      println!("{MSG}");
+      log::error!("{MSG}");
+      std::process::exit(1);
+    }));
+  }
+
+  if std::env::var("RUST_LOG").is_err() {
+    std::env::set_var("RUST_LOG", serai_env::var("RUST_LOG").unwrap_or_else(|| "info".to_string()));
+  }
+  env_logger::init();
+
+  log::info!("Starting Ethereum relayer server...");
+
+  // Open the DB
+  #[allow(unused_variables, unreachable_code)]
+  let db = {
+    #[cfg(all(feature = "parity-db", feature = "rocksdb"))]
+    panic!("built with parity-db and rocksdb");
+    #[cfg(all(feature = "parity-db", not(feature = "rocksdb")))]
+    let db =
+      serai_db::new_parity_db(&serai_env::var("DB_PATH").expect("path to DB wasn't specified"));
+    #[cfg(feature = "rocksdb")]
+    let db =
+      serai_db::new_rocksdb(&serai_env::var("DB_PATH").expect("path to DB wasn't specified"));
+    db
+  };
+
+  // Start command recipience server
+  // This should not be publicly exposed
+  // TODO: Add auth
+  tokio::spawn({
+    let db = db.clone();
+    async move {
+      // 5132 ^ ((b'E' << 8) | b'R')
+      let server = TcpListener::bind("0.0.0.0:20830").await.unwrap();
+      loop {
+        let (mut socket, _) = server.accept().await.unwrap();
+        let db = db.clone();
+        tokio::spawn(async move {
+          let mut db = db.clone();
+          loop {
+            let Ok(msg_len) = socket.read_u32_le().await else { break };
+            let mut buf = vec![0; usize::try_from(msg_len).unwrap()];
+            let Ok(_) = socket.read_exact(&mut buf).await else { break };
+
+            if buf.len() < 5 {
+              break;
+            }
+            let nonce = u32::from_le_bytes(buf[.. 4].try_into().unwrap());
+            let mut txn = db.txn();
+            txn.put(nonce.to_le_bytes(), &buf[4 ..]);
+            txn.commit();
+
+            let Ok(()) = socket.write_all(&[1]).await else { break };
+
+            log::info!("received signed command #{nonce}");
+          }
+        });
+      }
+    }
+  });
+
+  // Start command fetch server
+  // 5132 ^ ((b'E' << 8) | b'R') + 1
+  let server = TcpListener::bind("0.0.0.0:20831").await.unwrap();
+  loop {
+    let (mut socket, _) = server.accept().await.unwrap();
+    let db = db.clone();
+    tokio::spawn(async move {
+      let db = db.clone();
+      loop {
+        // Nonce to get the router comamnd for
+        let mut buf = vec![0; 4];
+        let Ok(_) = socket.read_exact(&mut buf).await else { break };
+
+        let command = db.get(&buf[.. 4]).unwrap_or(vec![]);
+        let Ok(()) = socket.write_all(&u32::try_from(command.len()).unwrap().to_le_bytes()).await
+        else {
+          break;
+        };
+        let Ok(()) = socket.write_all(&command).await else { break };
+      }
+    });
+  }
+}

deny.toml

@@ -44,6 +44,7 @@ exceptions = [
   { allow = ["AGPL-3.0"], name = "serai-env" },
 
   { allow = ["AGPL-3.0"], name = "ethereum-serai" },
+  { allow = ["AGPL-3.0"], name = "serai-ethereum-relayer" },
 
   { allow = ["AGPL-3.0"], name = "serai-message-queue" },
 

orchestration/dev/coins/ethereum-relayer/.folder

@@ -0,0 +1,11 @@
+#!/bin/sh
+
+RPC_USER="${RPC_USER:=serai}"
+RPC_PASS="${RPC_PASS:=seraidex}"
+
+# Run Monero
+monerod --non-interactive --regtest --offline --fixed-difficulty=1 \
+  --no-zmq --rpc-bind-ip=0.0.0.0 --rpc-bind-port=18081 --confirm-external-bind \
+  --rpc-access-control-origins "*" --disable-rpc-ban \
+  --rpc-login=$RPC_USER:$RPC_PASS \
+  $1

orchestration/src/ethereum_relayer.rs

@@ -0,0 +1,39 @@
+use std::path::Path;
+
+use crate::{Network, Os, mimalloc, os, build_serai_service, write_dockerfile};
+
+pub fn ethereum_relayer(orchestration_path: &Path, network: Network) {
+  let setup = mimalloc(Os::Debian).to_string() +
+    &build_serai_service("", network.release(), network.db(), "serai-ethereum-relayer");
+
+  let env_vars = [
+    ("DB_PATH", "/volume/ethereum-relayer-db".to_string()),
+    ("RUST_LOG", "info,serai_ethereum_relayer=trace".to_string()),
+  ];
+  let mut env_vars_str = String::new();
+  for (env_var, value) in env_vars {
+    env_vars_str += &format!(r#"{env_var}=${{{env_var}:="{value}"}} "#);
+  }
+
+  let run_ethereum_relayer = format!(
+    r#"
+# Copy the relayer server binary and relevant license
+COPY --from=builder --chown=ethereumrelayer /serai/bin/serai-ethereum-relayer /bin
+
+# Run ethereum-relayer
+EXPOSE 20830
+EXPOSE 20831
+CMD {env_vars_str} serai-ethereum-relayer
+"#
+  );
+
+  let run = os(Os::Debian, "", "ethereumrelayer") + &run_ethereum_relayer;
+  let res = setup + &run;
+
+  let mut ethereum_relayer_path = orchestration_path.to_path_buf();
+  ethereum_relayer_path.push("coins");
+  ethereum_relayer_path.push("ethereum-relayer");
+  ethereum_relayer_path.push("Dockerfile");
+
+  write_dockerfile(ethereum_relayer_path, &res);
+}

orchestration/src/main.rs

@@ -32,6 +32,9 @@ use mimalloc::mimalloc;
 mod coins;
 use coins::*;
 
+mod ethereum_relayer;
+use ethereum_relayer::ethereum_relayer;
+
 mod message_queue;
 use message_queue::message_queue;
 
@@ -280,6 +283,8 @@ fn dockerfiles(network: Network) {
   let ethereum_key = infrastructure_keys.remove("ethereum").unwrap();
   let monero_key = infrastructure_keys.remove("monero").unwrap();
 
+  ethereum_relayer(&orchestration_path, network);
+
   message_queue(
     &orchestration_path,
     network,
@@ -363,6 +368,7 @@ fn start(network: Network, services: HashSet<String>) {
     let name = match service.as_ref() {
       "serai" => "serai",
       "coordinator" => "coordinator",
+      "ethereum-relayer" => "ethereum-relayer",
       "message-queue" => "message-queue",
       "bitcoin-daemon" => "bitcoin",
       "bitcoin-processor" => "bitcoin-processor",
@@ -495,6 +501,10 @@ fn start(network: Network, services: HashSet<String>) {
             command
           }
         }
+        "ethereum-relayer" => {
+          // Expose the router command fetch server
+          command.arg("-p").arg("20831:20831")
+        }
         "monero" => {
           // Expose the RPC for tests
           if network == Network::Dev {
@@ -561,6 +571,9 @@ Commands:
     - `message-queue`
     - `bitcoin-daemon`
     - `bitcoin-processor`
+    - `ethereum-daemon`
+    - `ethereum-processor`
+    - `ethereum-relayer`
     - `monero-daemon`
     - `monero-processor`
     - `monero-wallet-rpc` (if "dev")
@@ -593,6 +606,9 @@ Commands:
     Some("start") => {
       let mut services = HashSet::new();
       for arg in args {
+        if arg == "ethereum-processor" {
+          services.insert("ethereum-relayer".to_string());
+        }
         if let Some(ext_network) = arg.strip_suffix("-processor") {
           services.insert(ext_network.to_string() + "-daemon");
         }

orchestration/src/processor.rs

@@ -41,24 +41,32 @@ RUN apt install -y ca-certificates
   const RPC_PASS: &str = "seraidex";
   // TODO: Isolate networks
   let hostname = format!("serai-{}-{coin}", network.label());
-  let port = match coin {
-    "bitcoin" => 8332,
-    "ethereum" => 8545,
-    "monero" => 18081,
-    _ => panic!("unrecognized external network"),
-  };
+  let port = format!(
+    "{}",
+    match coin {
+      "bitcoin" => 8332,
+      "ethereum" => 8545,
+      "monero" => 18081,
+      _ => panic!("unrecognized external network"),
+    }
+  );
 
-  let env_vars = [
+  let mut env_vars = vec![
     ("MESSAGE_QUEUE_RPC", format!("serai-{}-message-queue", network.label())),
     ("MESSAGE_QUEUE_KEY", hex::encode(coin_key.to_repr())),
     ("ENTROPY", hex::encode(entropy.as_ref())),
     ("NETWORK", coin.to_string()),
     ("NETWORK_RPC_LOGIN", format!("{RPC_USER}:{RPC_PASS}")),
     ("NETWORK_RPC_HOSTNAME", hostname),
-    ("NETWORK_RPC_PORT", format!("{port}")),
+    ("NETWORK_RPC_PORT", port),
     ("DB_PATH", "/volume/processor-db".to_string()),
     ("RUST_LOG", "info,serai_processor=debug".to_string()),
   ];
+  if coin == "ethereum" {
+    env_vars
+      .push(("ETHEREUM_RELAYER_HOSTNAME", format!("serai-{}-ethereum-relayer", network.label())));
+    env_vars.push(("ETHEREUM_RELAYER_PORT", "20830".to_string()));
+  }
   let mut env_vars_str = String::new();
   for (env_var, value) in env_vars {
     env_vars_str += &format!(r#"{env_var}=${{{env_var}:="{value}"}} "#);

orchestration/testnet/coins/ethereum-relayer/.folder

@@ -0,0 +1,11 @@
+#!/bin/sh
+
+RPC_USER="${RPC_USER:=serai}"
+RPC_PASS="${RPC_PASS:=seraidex}"
+
+# Run Monero
+monerod --non-interactive --regtest --offline --fixed-difficulty=1 \
+  --no-zmq --rpc-bind-ip=0.0.0.0 --rpc-bind-port=18081 --confirm-external-bind \
+  --rpc-access-control-origins "*" --disable-rpc-ban \
+  --rpc-login=$RPC_USER:$RPC_PASS \
+  $1

processor/src/main.rs

@@ -748,7 +748,15 @@ async fn main() {
     #[cfg(feature = "bitcoin")]
     NetworkId::Bitcoin => run(db, Bitcoin::new(url).await, coordinator).await,
     #[cfg(feature = "ethereum")]
-    NetworkId::Ethereum => run(db.clone(), Ethereum::new(db, url).await, coordinator).await,
+    NetworkId::Ethereum => {
+      let relayer_hostname = env::var("ETHEREUM_RELAYER_HOSTNAME")
+        .expect("ethereum relayer hostname wasn't specified")
+        .to_string();
+      let relayer_port =
+        env::var("ETHEREUM_RELAYER_PORT").expect("ethereum relayer port wasn't specified");
+      let relayer_url = relayer_hostname + ":" + &relayer_port;
+      run(db.clone(), Ethereum::new(db, url, relayer_url).await, coordinator).await
+    }
     #[cfg(feature = "monero")]
     NetworkId::Monero => run(db, Monero::new(url).await, coordinator).await,
     _ => panic!("spawning a processor for an unsupported network"),