Revive the previous Filecoin.EthSubscribe PR

CHANGELOG.md

@@ -31,6 +31,8 @@
 
 - [#5859](https://github.com/ChainSafe/forest/pull/5859) Added size metrics for zstd frame cache and made max size configurable via `FOREST_ZSTD_FRAME_CACHE_DEFAULT_MAX_SIZE` environment variable.
 
+- [#4976](https://github.com/ChainSafe/forest/issues/4976) Add support for the `Filecoin.EthSubscribe` and `Filecoin.EthUnsubscribe` API methods to enable subscriptions to Ethereum event types: `heads` and `logs`.
+
 ### Changed
 
 - [#5869](https://github.com/ChainSafe/forest/pull/5869) Updated `forest-cli snapshot export` to print average speed.

Cargo.toml

@@ -108,7 +108,7 @@ integer-encoding = "4.0"
 ipld-core = { version = "0.4", features = ["serde", "arb"] }
 is-terminal = "0.4"
 itertools = "0.14"
-jsonrpsee = { version = "0.25", features = ["server", "ws-client", "http-client"] }
+jsonrpsee = { version = "0.25", features = ["server", "ws-client", "http-client", "macros"] }
 jsonwebtoken = "9"
 keccak-hash = "0.11"
 kubert-prometheus-process = "0.2"

src/rpc/channel.rs

@@ -367,8 +367,8 @@ impl RpcModule {
                                     Ok(msg) => {
                                         match create_notif_message(&sink, &msg) {
                                             Ok(msg) => {
-                                                // This fails only if the connection is closed
-                                                if sink.send(msg).await.is_err() {
+                                                if let Err(e) = sink.send(msg).await {
+                                                    tracing::error!("Failed to send message: {:?}", e);
                                                     break;
                                                 }
                                             }

src/rpc/methods/chain.rs

@@ -16,6 +16,7 @@ use crate::lotus_json::{HasLotusJson, LotusJson, lotus_json_with_self};
 #[cfg(test)]
 use crate::lotus_json::{assert_all_snapshots, assert_unchanged_via_json};
 use crate::message::{ChainMessage, SignedMessage};
+use crate::rpc::eth::{EthLog, eth_logs_with_filter, types::ApiHeaders, types::EthFilterSpec};
 use crate::rpc::types::{ApiTipsetKey, Event};
 use crate::rpc::{ApiPaths, Ctx, EthEventHandler, Permission, RpcMethod, ServerError};
 use crate::shim::clock::ChainEpoch;
@@ -45,6 +46,82 @@ use tokio::sync::{
     Mutex,
     broadcast::{self, Receiver as Subscriber},
 };
+use tokio::task::JoinHandle;
+
+const HEAD_CHANNEL_CAPACITY: usize = 10;
+
+/// Subscribes to head changes from the chain store and broadcasts new blocks.
+///
+/// # Notes
+///
+/// Spawns an internal `tokio` task that can be aborted anytime via the returned `JoinHandle`,
+/// allowing manual cleanup if needed.
+pub(crate) fn new_heads<DB: Blockstore>(
+    data: &crate::rpc::RPCState<DB>,
+) -> (Subscriber<ApiHeaders>, JoinHandle<()>) {
+    let (sender, receiver) = broadcast::channel(HEAD_CHANNEL_CAPACITY);
+
+    let mut subscriber = data.chain_store().publisher().subscribe();
+
+    let handle = tokio::spawn(async move {
+        while let Ok(v) = subscriber.recv().await {
+            let headers = match v {
+                HeadChange::Apply(ts) => ApiHeaders(ts.block_headers().clone().into()),
+            };
+            if let Err(e) = sender.send(headers) {
+                tracing::error!("Failed to send headers: {}", e);
+                break;
+            }
+        }
+    });
+
+    (receiver, handle)
+}
+
+/// Subscribes to head changes from the chain store and broadcasts new `Ethereum` logs.
+///
+/// # Notes
+///
+/// Spawns an internal `tokio` task that can be aborted anytime via the returned `JoinHandle`,
+/// allowing manual cleanup if needed.
+pub(crate) fn logs<DB: Blockstore + Sync + Send + 'static>(
+    ctx: &Ctx<DB>,
+    filter: Option<EthFilterSpec>,
+) -> (Subscriber<Vec<EthLog>>, JoinHandle<()>) {
+    let (sender, receiver) = broadcast::channel(HEAD_CHANNEL_CAPACITY);
+
+    let mut subscriber = ctx.chain_store().publisher().subscribe();
+
+    let ctx = ctx.clone();
+
+    let handle = tokio::spawn(async move {
+        while let Ok(v) = subscriber.recv().await {
+            match v {
+                HeadChange::Apply(ts) => {
+                    match eth_logs_with_filter(&ctx, &ts, filter.clone(), None).await {
+                        Ok(logs) => {
+                            if !logs.is_empty() {
+                                if let Err(e) = sender.send(logs) {
+                                    tracing::error!(
+                                        "Failed to send logs for tipset {}: {}",
+                                        ts.key(),
+                                        e
+                                    );
+                                    break;
+                                }
+                            }
+                        }
+                        Err(e) => {
+                            tracing::error!("Failed to fetch logs for tipset {}: {}", ts.key(), e);
+                        }
+                    }
+                }
+            }
+        }
+    });
+
+    (receiver, handle)
+}
 
 pub enum ChainGetMessage {}
 impl RpcMethod<1> for ChainGetMessage {
@@ -721,7 +798,7 @@ pub(crate) fn chain_notify<DB: Blockstore>(
     _params: Params<'_>,
     data: &crate::rpc::RPCState<DB>,
 ) -> Subscriber<Vec<ApiHeadChange>> {
-    let (sender, receiver) = broadcast::channel(100);
+    let (sender, receiver) = broadcast::channel(HEAD_CHANNEL_CAPACITY);
 
     // As soon as the channel is created, send the current tipset
     let current = data.chain_store().heaviest_tipset();

src/rpc/methods/eth.rs

@@ -4,6 +4,8 @@
 pub(crate) mod errors;
 mod eth_tx;
 pub mod filter;
+pub mod pubsub;
+pub(crate) mod pubsub_trait;
 mod trace;
 pub mod types;
 mod utils;
@@ -1325,7 +1327,7 @@ async fn new_eth_tx_receipt<DB: Blockstore + Send + Sync + 'static>(
     Ok(tx_receipt)
 }
 
-async fn eth_logs_for_block_and_transaction<DB: Blockstore + Send + Sync + 'static>(
+pub async fn eth_logs_for_block_and_transaction<DB: Blockstore + Send + Sync + 'static>(
     ctx: &Ctx<DB>,
     ts: &Arc<Tipset>,
     block_hash: &EthHash,
@@ -1336,22 +1338,33 @@ async fn eth_logs_for_block_and_transaction<DB: Blockstore + Send + Sync + 'stat
         ..Default::default()
     };
 
+    eth_logs_with_filter(ctx, ts, Some(spec), Some(tx_hash)).await
+}
+
+pub async fn eth_logs_with_filter<DB: Blockstore + Send + Sync + 'static>(
+    ctx: &Ctx<DB>,
+    ts: &Arc<Tipset>,
+    spec: Option<EthFilterSpec>,
+    tx_hash: Option<&EthHash>,
+) -> anyhow::Result<Vec<EthLog>> {
     let mut events = vec![];
     EthEventHandler::collect_events(
         ctx,
         ts,
-        Some(&spec),
+        spec.as_ref(),
         SkipEvent::OnUnresolvedAddress,
         &mut events,
     )
     .await?;
 
     let logs = eth_filter_logs_from_events(ctx, &events)?;
-    let out: Vec<_> = logs
-        .into_iter()
-        .filter(|log| &log.transaction_hash == tx_hash)
-        .collect();
-    Ok(out)
+    Ok(match tx_hash {
+        Some(hash) => logs
+            .into_iter()
+            .filter(|log| &log.transaction_hash == hash)
+            .collect(),
+        None => logs, // no tx hash, keep all logs
+    })
 }
 
 fn get_signed_message<DB: Blockstore>(ctx: &Ctx<DB>, message_cid: Cid) -> Result<SignedMessage> {
@@ -2639,6 +2652,56 @@ impl RpcMethod<1> for EthUninstallFilter {
     }
 }
 
+pub enum EthUnsubscribe {}
+impl RpcMethod<0> for EthUnsubscribe {
+    const NAME: &'static str = "Filecoin.EthUnsubscribe";
+    const NAME_ALIAS: Option<&'static str> = Some("eth_unsubscribe");
+    const PARAM_NAMES: [&'static str; 0] = [];
+    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
+    const PERMISSION: Permission = Permission::Read;
+    const SUBSCRIPTION: bool = true;
+
+    type Params = ();
+    type Ok = ();
+
+    // This method is a placeholder and is never actually called.
+    // Subscription handling is performed in [`pubsub.rs`](pubsub).
+    //
+    // We still need to implement the [`RpcMethod`] trait to expose method metadata
+    // like [`NAME`](Self::NAME), [`NAME_ALIAS`](Self::NAME_ALIAS), [`PERMISSION`](Self::PERMISSION), etc..
+    async fn handle(
+        _: Ctx<impl Blockstore + Send + Sync + 'static>,
+        (): Self::Params,
+    ) -> Result<Self::Ok, ServerError> {
+        Ok(())
+    }
+}
+
+pub enum EthSubscribe {}
+impl RpcMethod<0> for EthSubscribe {
+    const NAME: &'static str = "Filecoin.EthSubscribe";
+    const NAME_ALIAS: Option<&'static str> = Some("eth_subscribe");
+    const PARAM_NAMES: [&'static str; 0] = [];
+    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
+    const PERMISSION: Permission = Permission::Read;
+    const SUBSCRIPTION: bool = true;
+
+    type Params = ();
+    type Ok = ();
+
+    // This method is a placeholder and is never actually called.
+    // Subscription handling is performed in [`pubsub.rs`](pubsub).
+    //
+    // We still need to implement the [`RpcMethod`] trait to expose method metadata
+    // like [`NAME`](Self::NAME), [`NAME_ALIAS`](Self::NAME_ALIAS), [`PERMISSION`](Self::PERMISSION), etc..
+    async fn handle(
+        _: Ctx<impl Blockstore + Send + Sync + 'static>,
+        (): Self::Params,
+    ) -> Result<Self::Ok, ServerError> {
+        Ok(())
+    }
+}
+
 pub enum EthAddressToFilecoinAddress {}
 impl RpcMethod<1> for EthAddressToFilecoinAddress {
     const NAME: &'static str = "Filecoin.EthAddressToFilecoinAddress";