fix: use ethereum block in eth_subscribe newHeads

CHANGELOG.md

@@ -56,6 +56,8 @@
 - [#6327](https://github.com/ChainSafe/forest/pull/6327) Fixed: Forest returns 404 for all invalid api paths.
 - [#6354](https://github.com/ChainSafe/forest/pull/6354) Fixed: Correctly calculate the epoch range instead of directly using the look back limit value while searching for messages.
 
+- [#6400](https://github.com/ChainSafe/forest/issues/6400) Fixed `eth_subscribe` `newHeads` to return Ethereum block format instead of Filecoin block headers array.
+
 ## Forest v0.30.5 "Dulce de Leche"
 
 Non-mandatory release supporting new API methods and addressing a critical panic issue.

src/rpc/methods/chain.rs

@@ -17,7 +17,10 @@ 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::eth::Block as EthBlock;
+use crate::rpc::eth::{
+    EthLog, TxInfo, eth_logs_with_filter, types::ApiHeaders, types::EthFilterSpec,
+};
 use crate::rpc::f3::F3ExportLatestSnapshot;
 use crate::rpc::types::*;
 use crate::rpc::{ApiPaths, Ctx, EthEventHandler, Permission, RpcMethod, ServerError};
@@ -74,8 +77,8 @@ static CHAIN_EXPORT_LOCK: LazyLock<Mutex<Option<CancellationToken>>> =
 ///
 /// 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>,
+pub(crate) fn new_heads<DB: Blockstore + Send + Sync + 'static>(
+    data: Ctx<DB>,
 ) -> (Subscriber<ApiHeaders>, JoinHandle<()>) {
     let (sender, receiver) = broadcast::channel(HEAD_CHANNEL_CAPACITY);
 
@@ -84,7 +87,17 @@ pub(crate) fn new_heads<DB: Blockstore>(
     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()),
+                HeadChange::Apply(ts) => {
+                    // Convert the tipset to an Ethereum block with full transaction info
+                    // Note: In Filecoin's Eth RPC, a tipset maps to a single Ethereum block
+                    match EthBlock::from_filecoin_tipset(data.clone(), ts, TxInfo::Full).await {
+                        Ok(block) => ApiHeaders(block),
+                        Err(e) => {
+                            tracing::error!("Failed to convert tipset to eth block: {}", e);
+                            continue;
+                        }
+                    }
+                }
             };
             if let Err(e) = sender.send(headers) {
                 tracing::error!("Failed to send headers: {}", e);

src/rpc/methods/eth.rs

@@ -531,6 +531,21 @@ pub struct Block {
     pub uncles: Vec<EthHash>,
 }
 
+/// Specifies the level of detail for transactions in Ethereum blocks.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum TxInfo {
+    /// Return only transaction hashes
+    Hash,
+    /// Return full transaction objects
+    Full,
+}
+
+impl From<bool> for TxInfo {
+    fn from(full: bool) -> Self {
+        if full { TxInfo::Full } else { TxInfo::Hash }
+    }
+}
+
 impl Block {
     pub fn new(has_transactions: bool, tipset_len: usize) -> Self {
         Self {
@@ -545,6 +560,89 @@ impl Block {
             ..Default::default()
         }
     }
+
+    /// Creates a new Ethereum block from a Filecoin tipset, executing transactions if requested.
+    ///
+    /// Reference: <https://github.com/filecoin-project/lotus/blob/941455f1d23e73b9ee92a1a4ce745d8848969858/node/impl/eth/utils.go#L44>
+    pub async fn from_filecoin_tipset<DB: Blockstore + Send + Sync + 'static>(
+        ctx: Ctx<DB>,
+        tipset: crate::blocks::Tipset,
+        tx_info: TxInfo,
+    ) -> Result<Self> {
+        static ETH_BLOCK_CACHE: LazyLock<SizeTrackingLruCache<CidWrapper, Block>> =
+            LazyLock::new(|| {
+                const DEFAULT_CACHE_SIZE: NonZeroUsize = nonzero!(500usize);
+                let cache_size = std::env::var("FOREST_ETH_BLOCK_CACHE_SIZE")
+                    .ok()
+                    .and_then(|s| s.parse().ok())
+                    .unwrap_or(DEFAULT_CACHE_SIZE);
+                SizeTrackingLruCache::new_with_metrics("eth_block".into(), cache_size)
+            });
+
+        let block_cid = tipset.key().cid()?;
+        let mut block = if let Some(b) = ETH_BLOCK_CACHE.get_cloned(&block_cid.into()) {
+            b
+        } else {
+            let parent_cid = tipset.parents().cid()?;
+            let block_number = EthUint64(tipset.epoch() as u64);
+            let block_hash: EthHash = block_cid.into();
+
+            let (state_root, msgs_and_receipts) = execute_tipset(&ctx, &tipset).await?;
+
+            let state_tree = StateTree::new_from_root(ctx.store_owned(), &state_root)?;
+
+            let mut full_transactions = vec![];
+            let mut gas_used = 0;
+            for (i, (msg, receipt)) in msgs_and_receipts.iter().enumerate() {
+                let ti = EthUint64(i as u64);
+                gas_used += receipt.gas_used();
+                let smsg = match msg {
+                    ChainMessage::Signed(msg) => msg.clone(),
+                    ChainMessage::Unsigned(msg) => {
+                        let sig = Signature::new_bls(vec![]);
+                        SignedMessage::new_unchecked(msg.clone(), sig)
+                    }
+                };
+
+                let mut tx = new_eth_tx_from_signed_message(
+                    &smsg,
+                    &state_tree,
+                    ctx.chain_config().eth_chain_id,
+                )?;
+                tx.block_hash = block_hash.clone();
+                tx.block_number = block_number.clone();
+                tx.transaction_index = ti;
+                full_transactions.push(tx);
+            }
+
+            let b = Block {
+                hash: block_hash,
+                number: block_number,
+                parent_hash: parent_cid.into(),
+                timestamp: EthUint64(tipset.block_headers().first().timestamp),
+                base_fee_per_gas: tipset
+                    .block_headers()
+                    .first()
+                    .parent_base_fee
+                    .clone()
+                    .into(),
+                gas_used: EthUint64(gas_used),
+                transactions: Transactions::Full(full_transactions),
+                ..Block::new(!msgs_and_receipts.is_empty(), tipset.len())
+            };
+            ETH_BLOCK_CACHE.push(block_cid.into(), b.clone());
+            b
+        };
+
+        if tx_info == TxInfo::Hash
+            && let Transactions::Full(transactions) = &block.transactions
+        {
+            block.transactions =
+                Transactions::Hash(transactions.iter().map(|tx| tx.hash.to_string()).collect())
+        }
+
+        Ok(block)
+    }
 }
 
 lotus_json_with_self!(Block);
@@ -1492,84 +1590,6 @@ fn get_signed_message<DB: Blockstore>(ctx: &Ctx<DB>, message_cid: Cid) -> Result
     })
 }
 
-pub async fn block_from_filecoin_tipset<DB: Blockstore + Send + Sync + 'static>(
-    data: Ctx<DB>,
-    tipset: Tipset,
-    full_tx_info: bool,
-) -> Result<Block> {
-    static ETH_BLOCK_CACHE: LazyLock<SizeTrackingLruCache<CidWrapper, Block>> =
-        LazyLock::new(|| {
-            const DEFAULT_CACHE_SIZE: NonZeroUsize = nonzero!(500usize);
-            let cache_size = std::env::var("FOREST_ETH_BLOCK_CACHE_SIZE")
-                .ok()
-                .and_then(|s| s.parse().ok())
-                .unwrap_or(DEFAULT_CACHE_SIZE);
-            SizeTrackingLruCache::new_with_metrics("eth_block".into(), cache_size)
-        });
-
-    let block_cid = tipset.key().cid()?;
-    let mut block = if let Some(b) = ETH_BLOCK_CACHE.get_cloned(&block_cid.into()) {
-        b
-    } else {
-        let parent_cid = tipset.parents().cid()?;
-        let block_number = EthUint64(tipset.epoch() as u64);
-        let block_hash: EthHash = block_cid.into();
-
-        let (state_root, msgs_and_receipts) = execute_tipset(&data, &tipset).await?;
-
-        let state_tree = StateTree::new_from_root(data.store_owned(), &state_root)?;
-
-        let mut full_transactions = vec![];
-        let mut gas_used = 0;
-        for (i, (msg, receipt)) in msgs_and_receipts.iter().enumerate() {
-            let ti = EthUint64(i as u64);
-            gas_used += receipt.gas_used();
-            let smsg = match msg {
-                ChainMessage::Signed(msg) => msg.clone(),
-                ChainMessage::Unsigned(msg) => {
-                    let sig = Signature::new_bls(vec![]);
-                    SignedMessage::new_unchecked(msg.clone(), sig)
-                }
-            };
-
-            let mut tx = new_eth_tx_from_signed_message(
-                &smsg,
-                &state_tree,
-                data.chain_config().eth_chain_id,
-            )?;
-            tx.block_hash = block_hash.clone();
-            tx.block_number = block_number.clone();
-            tx.transaction_index = ti;
-            full_transactions.push(tx);
-        }
-
-        let b = Block {
-            hash: block_hash,
-            number: block_number,
-            parent_hash: parent_cid.into(),
-            timestamp: EthUint64(tipset.block_headers().first().timestamp),
-            base_fee_per_gas: tipset
-                .block_headers()
-                .first()
-                .parent_base_fee
-                .clone()
-                .into(),
-            gas_used: EthUint64(gas_used),
-            transactions: Transactions::Full(full_transactions),
-            ..Block::new(!msgs_and_receipts.is_empty(), tipset.len())
-        };
-        ETH_BLOCK_CACHE.push(block_cid.into(), b.clone());
-        b
-    };
-
-    if !full_tx_info && let Transactions::Full(transactions) = &block.transactions {
-        block.transactions =
-            Transactions::Hash(transactions.iter().map(|tx| tx.hash.to_string()).collect())
-    }
-
-    Ok(block)
-}
-
 pub enum EthGetBlockByHash {}
 impl RpcMethod<2> for EthGetBlockByHash {
     const NAME: &'static str = "Filecoin.EthGetBlockByHash";
@@ -1590,8 +1610,9 @@ impl RpcMethod<2> for EthGetBlockByHash {
             BlockNumberOrHash::from_block_hash(block_hash),
             ResolveNullTipset::TakeOlder,
         )?;
-        let block = block_from_filecoin_tipset(ctx, ts, full_tx_info).await?;
-        Ok(block)
+        Block::from_filecoin_tipset(ctx, ts, full_tx_info.into())
+            .await
+            .map_err(ServerError::from)
     }
 }
 
@@ -1615,8 +1636,9 @@ impl RpcMethod<2> for EthGetBlockByNumber {
             block_param,
             ResolveNullTipset::TakeOlder,
         )?;
-        let block = block_from_filecoin_tipset(ctx, ts, full_tx_info).await?;
-        Ok(block)
+        Block::from_filecoin_tipset(ctx, ts, full_tx_info.into())
+            .await
+            .map_err(ServerError::from)
     }
 }
 
@@ -1637,8 +1659,9 @@ impl RpcMethod<2> for EthGetBlockByNumberV2 {
     ) -> Result<Self::Ok, ServerError> {
         let ts = tipset_by_block_number_or_hash_v2(&ctx, block_param, ResolveNullTipset::TakeOlder)
             .await?;
-        let block = block_from_filecoin_tipset(ctx, ts, full_tx_info).await?;
-        Ok(block)
+        Block::from_filecoin_tipset(ctx, ts, full_tx_info.into())
+            .await
+            .map_err(ServerError::from)
     }
 }
 

src/rpc/methods/eth/pubsub.rs

@@ -123,7 +123,7 @@ where
         accepted_sink: jsonrpsee::SubscriptionSink,
         ctx: Arc<RPCState<DB>>,
     ) {
-        let (subscriber, handle) = chain::new_heads(&ctx);
+        let (subscriber, handle) = chain::new_heads(ctx);
         tokio::spawn(async move {
             handle_subscription(subscriber, accepted_sink, handle).await;
         });

src/rpc/methods/eth/types.rs

@@ -2,7 +2,6 @@
 // SPDX-License-Identifier: Apache-2.0, MIT
 
 use super::*;
-use crate::blocks::CachingBlockHeader;
 use crate::rpc::eth::pubsub_trait::LogFilter;
 use anyhow::ensure;
 use get_size2::GetSize;
@@ -423,7 +422,7 @@ pub struct FilterID(EthHash);
 lotus_json_with_self!(FilterID);
 
 #[derive(Clone, Serialize, Deserialize, PartialEq, Debug)]
-pub struct ApiHeaders(#[serde(with = "crate::lotus_json")] pub Vec<CachingBlockHeader>);
+pub struct ApiHeaders(#[serde(with = "crate::lotus_json")] pub Block);
 
 lotus_json_with_self!(ApiHeaders);