FM-137: Ethereum API filter methods

Cargo.toml

@@ -82,5 +82,5 @@ cid = { version = "0.8", features = ["serde-codec", "std"] }
 # Using the same tendermint-rs dependency as tower-abci. From both we are interested in v037 modules.
 tower-abci = { version = "0.7" }
 tendermint = { version = "0.31", features = ["secp256k1"] }
-tendermint-rpc = { version = "0.31", features = ["secp256k1", "http-client"] }
+tendermint-rpc = { version = "0.31", features = ["secp256k1", "http-client", "websocket-client"] }
 tendermint-proto = { version = "0.31" }

docker/ci.Dockerfile

@@ -21,9 +21,8 @@ CMD ["run"]
 
 STOPSIGNAL SIGTERM
 
-ENV FM_ABCI__HOST=0.0.0.0
-ENV FM_ETH__HTTP__HOST=0.0.0.0
-ENV FM_ETH__WS__HOST=0.0.0.0
+ENV FM_ABCI__LISTEN__HOST=0.0.0.0
+ENV FM_ETH__LISTEN__HOST=0.0.0.0
 
 COPY fendermint/app/config $FM_HOME_DIR/config
 COPY docker/.artifacts/bundle.car $FM_HOME_DIR/bundle.car

docker/local.Dockerfile

@@ -31,9 +31,8 @@ CMD ["run"]
 
 STOPSIGNAL SIGTERM
 
-ENV FM_ABCI__HOST=0.0.0.0
-ENV FM_ETH__HTTP__HOST=0.0.0.0
-ENV FM_ETH__WS__HOST=0.0.0.0
+ENV FM_ABCI__LISTEN__HOST=0.0.0.0
+ENV FM_ETH__LISTEN__HOST=0.0.0.0
 
 # We could build the actor bundles in the `builder` as well,
 # but we should be able to copy it from somewhere.

fendermint/app/Cargo.toml

@@ -22,6 +22,7 @@ prost = { workspace = true }
 rand_chacha = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
+serde_with = { workspace = true }
 tendermint = { workspace = true }
 tendermint-rpc = { workspace = true }
 tendermint-proto = { workspace = true }

fendermint/app/config/default.toml

@@ -4,29 +4,26 @@ data_dir = "data"
 builtin_actors_bundle = "bundle.car"
 
 [abci]
+bound = 1
+
+[abci.listen]
 # Only accept connections from Tendermint, assumed to be running locally.
 host = "127.0.0.1"
 # The default port where Tendermint is going to connect to the application.
 port = 26658
-bound = 1
 
 [db]
 # Keep unlimited history by default.
 state_hist_size = 0
 
-# Ethereum API facade.
+# Ethereum API facade
 [eth]
+# Maximum time allowed between polls for filter changes, in seconds, before the subscription is canceled.
+filter_timeout = 300
 
-# Ethereum API facade for JSON-RPC.
-[eth.http]
+[eth.listen]
 # Only accept local connections by default.
 host = "127.0.0.1"
-# The default port where the Ethereum JSON-RPC API will listen to connections.
+# The default port where the Ethereum API will listen to
+# JSON-RPC (POST) and WebSockets (GET) requests.
 port = 8545
-
-# Ethereum API facade for WebSockets.
-[eth.ws]
-# Only accept local connections by default.
-host = "127.0.0.1"
-# The default port where the Ethereum WebSocket API will listen to connections.
-port = 8546

fendermint/app/src/cmd/eth.rs

@@ -1,8 +1,8 @@
 // Copyright 2022-2023 Protocol Labs
 // SPDX-License-Identifier: Apache-2.0, MIT
 
-use fendermint_rpc::client::http_client;
-use tendermint_rpc::HttpClient;
+use fendermint_rpc::client::ws_client;
+use tendermint_rpc::WebSocketClient;
 
 use crate::{
     cmd,
@@ -13,15 +13,21 @@ use crate::{
 cmd! {
   EthArgs(self, settings: EthSettings) {
     match self.command.clone() {
-      EthCommands::Run { url, proxy_url } => {
-        let client = http_client(url, proxy_url)?;
-        run(settings, client).await
+      EthCommands::Run { url, proxy_url: _ } => {
+        let (client, driver) = ws_client(url).await?;
+        let driver_handle = tokio::spawn(async move { driver.run().await });
+
+        let result = run(settings, client).await;
+
+        // Await the driver's termination to ensure proper connection closure.
+        let _ = driver_handle.await;
+        result
       }
     }
   }
 }
 
 /// Run the Ethereum
-async fn run(settings: EthSettings, client: HttpClient) -> anyhow::Result<()> {
-    fendermint_eth_api::listen(settings.http.addr(), client).await
+async fn run(settings: EthSettings, client: WebSocketClient) -> anyhow::Result<()> {
+    fendermint_eth_api::listen(settings.listen.addr(), client, settings.filter_timeout).await
 }

fendermint/app/src/cmd/run.rs

@@ -58,7 +58,7 @@ async fn run(settings: Settings) -> anyhow::Result<()> {
 
     // Run the ABCI server.
     server
-        .listen(settings.abci.listen_addr())
+        .listen(settings.abci.listen.addr())
         .await
         .map_err(|e| anyhow!("error listening: {e}"))?;
 

fendermint/app/src/options/eth.rs

@@ -18,8 +18,8 @@ pub enum EthCommands {
         #[arg(
             long,
             short,
-            default_value = "http://127.0.0.1:26657",
-            env = "TENDERMINT_RPC_URL"
+            default_value = "ws://127.0.0.1:26657/websocket",
+            env = "TENDERMINT_WS_URL"
         )]
         url: Url,
 

fendermint/app/src/settings.rs

@@ -3,22 +3,31 @@
 
 use config::{Config, ConfigError, Environment, File};
 use serde::Deserialize;
-use std::path::{Path, PathBuf};
+use serde_with::{serde_as, DurationSeconds};
+use std::{
+    path::{Path, PathBuf},
+    time::Duration,
+};
 
 #[derive(Debug, Deserialize)]
-pub struct AbciSettings {
+pub struct Address {
     pub host: String,
     pub port: u32,
-    /// Queue size for each ABCI component.
-    pub bound: usize,
 }
 
-impl AbciSettings {
-    pub fn listen_addr(&self) -> String {
+impl Address {
+    pub fn addr(&self) -> String {
         format!("{}:{}", self.host, self.port)
     }
 }
 
+#[derive(Debug, Deserialize)]
+pub struct AbciSettings {
+    pub listen: Address,
+    /// Queue size for each ABCI component.
+    pub bound: usize,
+}
+
 #[derive(Debug, Deserialize)]
 pub struct DbSettings {
     /// Length of the app state history to keep in the database before pruning; 0 means unlimited.
@@ -27,25 +36,13 @@ pub struct DbSettings {
     pub state_hist_size: u64,
 }
 
-#[derive(Debug, Deserialize)]
-pub struct Address {
-    pub host: String,
-    pub port: u32,
-}
-
-impl Address {
-    pub fn addr(&self) -> String {
-        format!("{}:{}", self.host, self.port)
-    }
-}
-
 /// Ethereum API facade settings.
+#[serde_as]
 #[derive(Debug, Deserialize)]
 pub struct EthSettings {
-    /// Listen address for JSON-RPC
-    pub http: Address,
-    /// Listen address for WebSockets
-    pub ws: Address,
+    pub listen: Address,
+    #[serde_as(as = "DurationSeconds<u64>")]
+    pub filter_timeout: Duration,
 }
 
 #[derive(Debug, Deserialize)]

fendermint/app/src/tmconv.rs

@@ -112,6 +112,24 @@ pub fn to_begin_block(ret: FvmApplyRet) -> response::BeginBlock {
 }
 
 /// Convert events to key-value pairs.
+///
+///
+/// Fot the EVM, they are returned like so:
+///
+/// ```text
+/// StampedEvent { emitter: 103,
+///  event: ActorEvent { entries: [
+///    Entry { flags: FLAG_INDEXED_VALUE, key: "t1", value: RawBytes { 5820ddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef } },
+///    Entry { flags: FLAG_INDEXED_VALUE, key: "t2", value: RawBytes { 54ff00000000000000000000000000000000000065 } },
+///    Entry { flags: FLAG_INDEXED_VALUE, key: "t3", value: RawBytes { 54ff00000000000000000000000000000000000066 } },
+///    Entry { flags: FLAG_INDEXED_VALUE, key: "d", value: RawBytes { 582000000000000000000000000000000000000000000000000000000000000007d0 } }] } }
+/// ```
+///
+/// The values are:
+/// * "t1" will be the cbor encoded keccak-256 hash of the event signature Transfer(address,address,uint256)
+/// * "t2" will be the first indexed argument, i.e. _from  (cbor encoded byte array; needs padding to 32 bytes to work with ethers)
+/// * "t3" will be the second indexed argument, i.e. _to (cbor encoded byte array; needs padding to 32 bytes to work with ethers)
+/// * "d" is a cbor encoded byte array of all the remaining arguments
 pub fn to_events(kind: &str, stamped_events: Vec<StampedEvent>) -> Vec<Event> {
     stamped_events
         .into_iter()

fendermint/eth/api/Cargo.toml

@@ -21,6 +21,7 @@ serde_json = { workspace = true }
 tracing = { workspace = true }
 tendermint = { workspace = true }
 tendermint-rpc = { workspace = true }
+tokio = { workspace = true }
 
 cid = { workspace = true }
 fvm_shared = { workspace = true }
@@ -42,7 +43,6 @@ tracing-subscriber = { workspace = true }
 quickcheck = { workspace = true }
 quickcheck_macros = { workspace = true }
 thiserror = { workspace = true }
-tokio = { workspace = true }
 
 fendermint_testing = { path = "../../testing", features = ["arb"] }
 fendermint_vm_message = { path = "../../vm/message", features = ["arb"] }