feat(rpc): implement Filecoin.EthGetMessageCidByTransactionHash (#4402)

.github/workflows/forest.yml

@@ -282,6 +282,32 @@ jobs:
           chmod +x ~/.cargo/bin/forest*
       - run: ./scripts/tests/calibnet_kademlia_check.sh
         timeout-minutes: '${{ fromJSON(env.SCRIPT_TIMEOUT_MINUTES) }}'
+  calibnet-eth-mapping-check:
+    needs:
+      - build-ubuntu
+    name: Calibnet eth mapping check
+    runs-on: ubuntu-latest
+    steps:
+      - run: lscpu
+      - uses: actions/cache@v4
+        with:
+          path: '${{ env.FIL_PROOFS_PARAMETER_CACHE }}'
+          key: proof-params-keys
+      - name: Checkout Sources
+        uses: actions/checkout@v4
+      - uses: actions/download-artifact@v4
+        with:
+          name: 'forest-${{ runner.os }}'
+          path: ~/.cargo/bin
+      - uses: actions/download-artifact@v4
+        with:
+          name: 'forest-${{ runner.os }}'
+          path: ~/.cargo/bin
+      - name: Set permissions
+        run: |
+          chmod +x ~/.cargo/bin/forest*
+      - run: ./scripts/tests/calibnet_eth_mapping_check.sh
+        timeout-minutes: '${{ fromJSON(env.SCRIPT_TIMEOUT_MINUTES) }}'
   db-migration-checks:
     needs:
       - build-ubuntu

CHANGELOG.md

@@ -41,6 +41,9 @@
 - [#4381](https://github.com/ChainSafe/forest/pull/4381) Add support for the
   `Filecoin.StateSectorPartition` RPC method.
 
+- [#4368](https://github.com/ChainSafe/forest/issues/4368) Add support for the
+  `Filecoin.EthGetMessageCidByTransactionHash` RPC method.
+
 - [#4167](https://github.com/ChainSafe/forest/issues/4167) Add support for the
   `Filecoin.EthGetBlockByHash` RPC method.
 

scripts/tests/calibnet_eth_mapping_check.sh

@@ -0,0 +1,73 @@
+#!/usr/bin/env bash
+# This script is checking the correctness of the ethereum mapping feature
+# It requires both the `forest` and `forest-cli` binaries to be in the PATH.
+
+set -eu
+
+source "$(dirname "$0")/harness.sh"
+
+forest_init
+
+FOREST_URL='http://127.0.0.1:2345/rpc/v1'
+
+NUM_TIPSETS=600
+
+echo "Get Ethereum block hashes and transactions hashes from the last $NUM_TIPSETS tipsets"
+
+OUTPUT=$($FOREST_CLI_PATH info show)
+
+HEAD_EPOCH=$(echo "$OUTPUT" | sed -n 's/.*epoch: \([0-9]*\).*/\1/p')
+EPOCH=$((HEAD_EPOCH - 1))
+
+ETH_BLOCK_HASHES=()
+ETH_TX_HASHES=()
+
+for ((i=0; i<=NUM_TIPSETS; i++)); do
+  EPOCH_HEX=$(printf "0x%x" $EPOCH)
+  JSON=$(curl -s -X POST "$FOREST_URL" \
+    -H 'Content-Type: application/json' \
+    --data "$(jq -n --arg epoch "$EPOCH_HEX" '{jsonrpc: "2.0", id: 1, method: "Filecoin.EthGetBlockByNumber", params: [$epoch, false]}')")
+
+
+  HASH=$(echo "$JSON" | jq -r '.result.hash')
+  ETH_BLOCK_HASHES+=("$HASH")
+
+  if [[ $(echo "$JSON" | jq -e '.result.transactions') != "null" ]]; then
+    TRANSACTIONS=$(echo "$JSON" | jq -r '.result.transactions[]')
+    for tx in $TRANSACTIONS; do
+        ETH_TX_HASHES+=("$tx")
+    done
+  else
+    echo "No transactions found for block hash: $EPOCH_HEX"
+  fi
+
+  EPOCH=$((EPOCH - 1))
+done
+
+ERROR=0
+echo "Testing Ethereum mapping"
+
+for hash in "${ETH_BLOCK_HASHES[@]}"; do
+  JSON=$(curl -s -X POST "$FOREST_URL" \
+    -H 'Content-Type: application/json' \
+    --data "$(jq -n --arg hash "$hash" '{jsonrpc: "2.0", id: 1, method: "Filecoin.EthGetBalance", params: ["0xff38c072f286e3b20b3954ca9f99c05fbecc64aa", $hash]}')")
+
+  if [[ $(echo "$JSON" | jq -e '.result') == "null" ]]; then
+    echo "Missing tipset key for hash $hash"
+    ERROR=1
+  fi
+done
+
+for hash in "${ETH_TX_HASHES[@]}"; do
+  JSON=$(curl -s -X POST "$FOREST_URL" \
+    -H 'Content-Type: application/json' \
+    --data "$(jq -n --arg hash "$hash" '{jsonrpc: "2.0", id: 1, method: "Filecoin.EthGetMessageCidByTransactionHash", params: [$hash]}')")
+
+  if [[ $(echo "$JSON" | jq -e '.result') == "null" ]]; then
+    echo "Missing cid for hash $hash"
+    ERROR=1
+  fi
+done
+
+echo "Done"
+exit $ERROR

src/chain/store/chain_store.rs

@@ -9,8 +9,8 @@ use crate::interpreter::BlockMessages;
 use crate::interpreter::VMTrace;
 use crate::libp2p_bitswap::{BitswapStoreRead, BitswapStoreReadWrite};
 use crate::message::{ChainMessage, Message as MessageTrait, SignedMessage};
-use crate::networks::ChainConfig;
-use crate::rpc::eth;
+use crate::networks::{ChainConfig, Height};
+use crate::rpc::eth::{self, eth_tx_from_signed_eth_message};
 use crate::shim::clock::ChainEpoch;
 use crate::shim::{
     address::Address, econ::TokenAmount, executor::Receipt, message::Message,
@@ -78,6 +78,9 @@ pub struct ChainStore<DB> {
 
     /// Ethereum mappings store
     eth_mappings: Arc<dyn EthMappingsStore + Sync + Send>,
+
+    /// Needed by the Ethereum mapping.
+    chain_config: Arc<ChainConfig>,
 }
 
 impl<DB> BitswapStoreRead for ChainStore<DB>
@@ -131,12 +134,13 @@ where
         let cs = Self {
             publisher,
             chain_index,
-            tipset_tracker: TipsetTracker::new(Arc::clone(&db), chain_config),
+            tipset_tracker: TipsetTracker::new(Arc::clone(&db), chain_config.clone()),
             db,
             settings,
             genesis_block_header,
             validated_blocks,
             eth_mappings,
+            chain_config,
         };
 
         Ok(cs)
@@ -162,10 +166,10 @@ where
     pub fn put_tipset(&self, ts: &Tipset) -> Result<(), Error> {
         persist_objects(self.blockstore(), ts.block_headers().iter())?;
 
-        self.put_tipset_key(ts.key())?;
-
         // Expand tipset to include other compatible blocks at the epoch.
         let expanded = self.expand_tipset(ts.min_ticket_block().clone())?;
+        self.put_tipset_key(expanded.key())?;
+
         self.update_heaviest(Arc::new(expanded))?;
         Ok(())
     }
@@ -177,6 +181,20 @@ where
         Ok(())
     }
 
+    /// Writes the delegated message `Cid`s to the blockstore for `EthAPI` queries.
+    pub fn put_delegated_message_hashes<'a>(
+        &self,
+        headers: impl Iterator<Item = &'a CachingBlockHeader>,
+    ) -> Result<(), Error> {
+        tracing::debug!("persist eth mapping");
+
+        // The messages will be ordered from most recent block to less recent
+        let delegated_messages = self.headers_delegated_messages(headers)?;
+
+        self.process_signed_messages(&delegated_messages)?;
+        Ok(())
+    }
+
     /// Reads the `TipsetKey` from the blockstore for `EthAPI` queries.
     pub fn get_required_tipset_key(&self, hash: &eth::Hash) -> Result<TipsetKey, Error> {
         let tsk = self
@@ -188,12 +206,19 @@ where
     }
 
     /// Writes with timestamp the `Hash` to `Cid` mapping to the blockstore for `EthAPI` queries.
-    pub fn put_mapping(&self, k: eth::Hash, v: Cid) -> Result<(), Error> {
-        let timestamp = chrono::Utc::now().timestamp() as u64;
+    pub fn put_mapping(&self, k: eth::Hash, v: Cid, timestamp: u64) -> Result<(), Error> {
         self.eth_mappings.write_obj(&k, &(v, timestamp))?;
         Ok(())
     }
 
+    /// Reads the `Cid` from the blockstore for `EthAPI` queries.
+    pub fn get_mapping(&self, hash: &eth::Hash) -> Result<Option<Cid>, Error> {
+        Ok(self
+            .eth_mappings
+            .read_obj::<(Cid, u64)>(hash)?
+            .map(|(cid, _)| cid))
+    }
+
     /// Expands tipset to tipset with all other headers in the same epoch using
     /// the tipset tracker.
     fn expand_tipset(&self, header: CachingBlockHeader) -> Result<Tipset, Error> {
@@ -354,6 +379,88 @@ where
             .map_err(|e| Error::Other(format!("Could not get tipset from keys {e:?}")))?;
         Ok((lbts, *next_ts.parent_state()))
     }
+
+    /// Filter [`SignedMessage`]'s to keep only the most recent ones, then write corresponding entries to the Ethereum mapping.
+    pub fn process_signed_messages(&self, messages: &[(SignedMessage, u64)]) -> anyhow::Result<()>
+    where
+        DB: fvm_ipld_blockstore::Blockstore,
+    {
+        let eth_txs: Vec<(eth::Hash, Cid, u64, usize)> = messages
+            .iter()
+            .enumerate()
+            .filter_map(|(i, (smsg, timestamp))| {
+                if let Ok(tx) = eth_tx_from_signed_eth_message(smsg, self.chain_config.eth_chain_id)
+                {
+                    if let Ok(hash) = tx.eth_hash() {
+                        // newest messages are the ones with lowest index
+                        Some((hash, smsg.cid().unwrap(), *timestamp, i))
+                    } else {
+                        None
+                    }
+                } else {
+                    None
+                }
+            })
+            .collect();
+        let filtered = filter_lowest_index(eth_txs);
+        let num_entries = filtered.len();
+
+        // write back
+        for (k, v, timestamp) in filtered.into_iter() {
+            tracing::trace!("Insert mapping {} => {}", k, v);
+            self.put_mapping(k, v, timestamp)?;
+        }
+        tracing::debug!("Wrote {} entries in Ethereum mapping", num_entries);
+        Ok(())
+    }
+
+    pub fn headers_delegated_messages<'a>(
+        &self,
+        headers: impl Iterator<Item = &'a CachingBlockHeader>,
+    ) -> anyhow::Result<Vec<(SignedMessage, u64)>>
+    where
+        DB: fvm_ipld_blockstore::Blockstore,
+    {
+        let mut delegated_messages = vec![];
+
+        // Hygge is the start of Ethereum support in the FVM (through the FEVM actor).
+        // Before this height, no notion of an Ethereum-like API existed.
+        let filtered_headers =
+            headers.filter(|bh| bh.epoch >= self.chain_config.epoch(Height::Hygge));
+
+        for bh in filtered_headers {
+            if let Ok((_, secp_cids)) = block_messages(self.blockstore(), bh) {
+                let mut messages: Vec<_> = secp_cids
+                    .into_iter()
+                    .filter(|msg| msg.is_delegated())
+                    .map(|m| (m, bh.timestamp))
+                    .collect();
+                delegated_messages.append(&mut messages);
+            }
+        }
+
+        Ok(delegated_messages)
+    }
+}
+
+fn filter_lowest_index(values: Vec<(eth::Hash, Cid, u64, usize)>) -> Vec<(eth::Hash, Cid, u64)> {
+    let map: HashMap<eth::Hash, (Cid, u64, usize)> = values.into_iter().fold(
+        HashMap::default(),
+        |mut acc, (hash, cid, timestamp, index)| {
+            acc.entry(hash)
+                .and_modify(|&mut (_, _, ref mut min_index)| {
+                    if index < *min_index {
+                        *min_index = index;
+                    }
+                })
+                .or_insert((cid, timestamp, index));
+            acc
+        },
+    );
+
+    map.into_iter()
+        .map(|(hash, (cid, timestamp, _))| (hash, cid, timestamp))
+        .collect()
 }
 
 /// Returns a Tuple of BLS messages of type `UnsignedMessage` and SECP messages

src/chain_sync/chain_muxer.rs

@@ -511,6 +511,9 @@ where
             block.persist(&chain_store.db)?;
         }
 
+        // This is needed for the Ethereum mapping
+        chain_store.put_tipset_key(tipset.key())?;
+
         // Update the peer head
         network
             .peer_manager()

src/chain_sync/tipset_syncer.rs

@@ -761,15 +761,20 @@ async fn sync_tipset_range<DB: Blockstore + Sync + Send + 'static>(
         return Err(why.into());
     };
 
-    //  Sync and validate messages from the tipsets
+    // Persist tipset keys
+    for ts in parent_tipsets.iter() {
+        chain_store.put_tipset_key(ts.key())?;
+    }
+
+    // Sync and validate messages from the tipsets
     tracker.write().set_stage(SyncStage::Messages);
     if let Err(why) = sync_messages_check_state(
         tracker.clone(),
         state_manager,
         network,
         chain_store.clone(),
         &bad_block_cache,
-        parent_tipsets,
+        parent_tipsets.clone(),
         &genesis,
         InvalidBlockStrategy::Forgiving,
     )
@@ -780,6 +785,9 @@ async fn sync_tipset_range<DB: Blockstore + Sync + Send + 'static>(
         return Err(why);
     };
 
+    // Call only once messages persisted
+    chain_store.put_delegated_message_hashes(headers.into_iter())?;
+
     // At this point the head is synced and it can be set in the store as the
     // heaviest
     debug!(
@@ -961,6 +969,9 @@ async fn sync_tipset<DB: Blockstore + Sync + Send + 'static>(
         proposed_head.block_headers().iter(),
     )?;
 
+    // Persist tipset key
+    chain_store.put_tipset_key(proposed_head.key())?;
+
     // Sync and validate messages from the tipsets
     if let Err(e) = sync_messages_check_state(
         // Include a dummy WorkerState
@@ -979,6 +990,9 @@ async fn sync_tipset<DB: Blockstore + Sync + Send + 'static>(
         return Err(e);
     }
 
+    // Call only once messages persisted
+    chain_store.put_delegated_message_hashes(proposed_head.block_headers().iter())?;
+
     // Add the tipset to the store. The tipset will be expanded with other blocks
     // with the same [epoch, parents] before updating the heaviest Tipset in
     // the store.

src/cli_shared/cli/client.rs

@@ -75,6 +75,8 @@ pub struct Client {
     pub token_exp: Duration,
     /// Load actors from the bundle file (possibly generating it if it doesn't exist)
     pub load_actors: bool,
+    /// `TTL` to set for Ethereum `Hash` to `Cid` entries or `None` to never reclaim them.
+    pub eth_mapping_ttl: Option<u32>,
 }
 
 impl Default for Client {
@@ -103,6 +105,7 @@ impl Default for Client {
             ),
             token_exp: Duration::try_seconds(5184000).expect("Infallible"), // 60 Days = 5184000 Seconds
             load_actors: true,
+            eth_mapping_ttl: None,
         }
     }
 }