Add endpoint for direct submission of Ethereum events

apps/Cargo.toml

@@ -128,6 +128,9 @@ web30 = "0.19.1"
 websocket = "0.26.2"
 winapi = "0.3.9"
 
+warp = "0.3.2"
+bytes = "1.1.0"
+
 [dev-dependencies]
 assert_matches = "1.5.0"
 namada = {path = "../shared", features = ["testing", "wasm-runtime"]}

apps/src/lib/config/ethereum.rs

@@ -1,3 +1,6 @@
+//! Configuration settings to do with the Ethereum bridge.
+#[allow(unused_imports)]
+use namada::types::ethereum_events::EthereumEvent;
 use serde::{Deserialize, Serialize};
 
 /// Default [Ethereum JSON-RPC](https://ethereum.org/en/developers/docs/apis/json-rpc/) endpoint used by the oracle
@@ -8,12 +11,20 @@ pub struct Config {
     /// The Ethereum JSON-RPC endpoint that the Ethereum event oracle will use
     /// to listen for events from the Ethereum bridge smart contracts
     pub oracle_rpc_endpoint: String,
+    /// If this is set to `true`, then instead of the oracle listening for
+    /// events at a Ethereum JSON-RPC endpoint, an endpoint will be exposed by
+    /// the ledger for submission of Borsh-serialized
+    /// [`EthereumEvent`]s
+    #[cfg(not(feature = "eth-fullnode"))]
+    pub oracle_event_endpoint: bool,
 }
 
 impl Default for Config {
     fn default() -> Self {
         Self {
             oracle_rpc_endpoint: DEFAULT_ORACLE_RPC_ENDPOINT.to_owned(),
+            #[cfg(not(feature = "eth-fullnode"))]
+            oracle_event_endpoint: false,
         }
     }
 }

apps/src/lib/node/ledger/ethereum_node/mod.rs

@@ -1,12 +1,8 @@
 pub mod events;
 pub mod oracle;
-#[cfg(feature = "eth-fullnode")]
-pub use oracle::{run_oracle, Oracle};
 pub mod test_tools;
 use std::ffi::OsString;
 
-#[cfg(not(feature = "eth-fullnode"))]
-pub use test_tools::mock_oracle::run_oracle;
 use thiserror::Error;
 use tokio::sync::oneshot::{Receiver, Sender};
 

apps/src/lib/node/ledger/ethereum_node/test_tools.rs

@@ -47,6 +47,69 @@ pub mod mock_oracle {
     }
 }
 
+#[cfg(not(feature = "eth-fullnode"))]
+pub mod event_endpoint {
+    use borsh::BorshDeserialize;
+    use namada::types::ethereum_events::EthereumEvent;
+    use tokio::sync::mpsc::UnboundedSender;
+
+    const ETHEREUM_EVENTS_ENDPOINT: ([u8; 4], u16) = ([127, 0, 0, 1], 3030);
+
+    /// The path to which Borsh-serialized Ethereum events should be submitted
+    const PATH: &str = "eth_events";
+
+    pub fn start_oracle(
+        sender: UnboundedSender<EthereumEvent>,
+    ) -> tokio::task::JoinHandle<()> {
+        tokio::spawn(async move {
+            use warp::Filter;
+
+            tracing::info!(
+                ?ETHEREUM_EVENTS_ENDPOINT,
+                "Ethereum event endpoint is starting"
+            );
+
+            let eth_events = warp::post()
+                .and(warp::path(PATH))
+                .and(warp::body::bytes())
+                .map(move |bytes: bytes::Bytes| {
+                    tracing::info!(len = bytes.len(), "Received request");
+                    let event = match EthereumEvent::try_from_slice(&bytes) {
+                        Ok(event) => event,
+                        Err(error) => {
+                            tracing::warn!(?error, "Couldn't handle request");
+                            return warp::reply::with_status(
+                                "Bad request",
+                                warp::http::StatusCode::BAD_REQUEST,
+                            );
+                        }
+                    };
+                    tracing::debug!("Serialized event - {:#?}", event);
+                    match sender.send(event) {
+                        Ok(()) => warp::reply::with_status(
+                            "OK",
+                            warp::http::StatusCode::OK,
+                        ),
+                        Err(error) => {
+                            tracing::warn!(?error, "Couldn't send event");
+                            warp::reply::with_status(
+                                "Internal server error",
+                                warp::http::StatusCode::INTERNAL_SERVER_ERROR,
+                            )
+                        }
+                    }
+                });
+
+            warp::serve(eth_events).run(ETHEREUM_EVENTS_ENDPOINT).await;
+
+            tracing::info!(
+                ?ETHEREUM_EVENTS_ENDPOINT,
+                "Ethereum event endpoint is no longer running"
+            );
+        })
+    }
+}
+
 #[cfg(all(test, feature = "eth-fullnode"))]
 pub mod mock_web3_client {
     use std::cell::RefCell;

apps/src/lib/node/ledger/mod.rs

@@ -340,8 +340,26 @@ async fn run_aux(config: config::Ledger, wasm_dir: PathBuf) {
             res
         });
 
-        let oracle =
-            ethereum_node::run_oracle(ethereum_url, eth_sender, abort_sender);
+        let oracle = {
+            #[cfg(not(feature = "eth-fullnode"))]
+            if config.ethereum.oracle_event_endpoint {
+                ethereum_node::test_tools::event_endpoint::start_oracle(
+                    eth_sender,
+                )
+            } else {
+                ethereum_node::test_tools::mock_oracle::run_oracle(
+                    ethereum_url,
+                    eth_sender,
+                    abort_sender,
+                )
+            }
+            #[cfg(feature = "eth-fullnode")]
+            ethereum_node::oracle::run_oracle(
+                ethereum_url,
+                eth_sender,
+                abort_sender,
+            )
+        };
 
         // Shutdown ethereum_node via a message to ensure that the child process
         // is properly cleaned-up.