Don't have the router drop transactions which may have top-level transfers

The router will now match the top-level transfer so it isn't used as the
justification for the InInstruction it's handling. This allows the theoretical
case where a top-level transfer occurs (to any entity) and an internal call
performs a transfer to Serai.

Also uses a JoinSet for fetching transactions' top-level transfers in the ERC20
crate. This does add a dependency on tokio yet improves performance, and it's
scoped under serai-processor (which is always presumed to be tokio-based).
While we could instead import futures for join_all,
https://github.com/smol-rs/futures-lite/issues/6 summarizes why that wouldn't
be a good idea. While we could prefer async-executor over tokio's JoinSet,
JoinSet doesn't share the same issues as FuturesUnordered. That means our
question is solely if we want the async-executor executor or the tokio
executor, when we've already established the Serai processor is always presumed
to be tokio-based.

Cargo.lock

@@ -8748,6 +8748,7 @@ dependencies = [
  "alloy-sol-macro",
  "alloy-sol-types",
  "alloy-transport",
+ "tokio",
 ]
 
 [[package]]

processor/ethereum/erc20/Cargo.toml

@@ -26,3 +26,5 @@ alloy-rpc-types-eth = { version = "0.3", default-features = false }
 alloy-transport = { version = "0.3", default-features = false }
 alloy-simple-request-transport = { path = "../../../networks/ethereum/alloy-simple-request-transport", default-features = false }
 alloy-provider = { version = "0.3", default-features = false }
+
+tokio = { version = "1", default-features = false, features = ["rt"] }

processor/ethereum/erc20/src/lib.rs

@@ -13,6 +13,8 @@ use alloy_transport::{TransportErrorKind, RpcError};
 use alloy_simple_request_transport::SimpleRequest;
 use alloy_provider::{Provider, RootProvider};
 
+use tokio::task::JoinSet;
+
 #[rustfmt::skip]
 #[expect(warnings)]
 #[expect(needless_pass_by_value)]
@@ -27,7 +29,7 @@ pub use abi::IERC20::Transfer;
 
 /// A top-level ERC20 transfer
 #[derive(Clone, Debug)]
-pub struct TopLevelErc20Transfer {
+pub struct TopLevelTransfer {
   /// The transaction ID which effected this transfer.
   pub id: [u8; 32],
   /// The address which made the transfer.
@@ -38,6 +40,14 @@ pub struct TopLevelErc20Transfer {
   pub data: Vec<u8>,
 }
 
+/// A transaction with a top-level transfer, matched to the log index of the transfer.
+pub struct MatchedTopLevelTransfer {
+  /// The transfer.
+  pub transfer: TopLevelTransfer,
+  /// The log index of the transfer.
+  pub log_index: u64,
+}
+
 /// A view for an ERC20 contract.
 #[derive(Clone, Debug)]
 pub struct Erc20(Arc<RootProvider<SimpleRequest>>, Address);
@@ -47,12 +57,104 @@ impl Erc20 {
     Self(provider, Address::from(&address))
   }
 
-  /// Fetch all top-level transfers to the specified ERC20.
+  /// Match a transaction for its top-level transfer to the specified address (if one exists).
+  pub async fn match_top_level_transfer(
+    provider: impl AsRef<RootProvider<SimpleRequest>>,
+    transaction_id: B256,
+    to: Address,
+  ) -> Result<Option<MatchedTopLevelTransfer>, RpcError<TransportErrorKind>> {
+    // Fetch the transaction
+    let transaction =
+      provider.as_ref().get_transaction_by_hash(transaction_id).await?.ok_or_else(|| {
+        TransportErrorKind::Custom(
+          "node didn't have the transaction which emitted a log it had".to_string().into(),
+        )
+      })?;
+
+    // If this is a top-level call...
+    // Don't validate the encoding as this can't be re-encoded to an identical bytestring due
+    // to the `InInstruction` appended after the call itself
+    if let Ok(call) = IERC20Calls::abi_decode(&transaction.input, false) {
+      // Extract the top-level call's from/to/value
+      let (from, call_to, value) = match call {
+        IERC20Calls::transfer(transferCall { to, value }) => (transaction.from, to, value),
+        IERC20Calls::transferFrom(transferFromCall { from, to, value }) => (from, to, value),
+        // Treat any other function selectors as unrecognized
+        _ => return Ok(None),
+      };
+      // If this isn't a transfer to the expected address, return None
+      if call_to != to {
+        return Ok(None);
+      }
+
+      // Fetch the transaction's logs
+      let receipt =
+        provider.as_ref().get_transaction_receipt(transaction_id).await?.ok_or_else(|| {
+          TransportErrorKind::Custom(
+            "node didn't have receipt for a transaction we were matching for a top-level transfer"
+              .to_string()
+              .into(),
+          )
+        })?;
+
+      // Find the log for this transfer
+      for log in receipt.inner.logs() {
+        // If this log was emitted by a different contract, continue
+        if Some(log.address()) != transaction.to {
+          continue;
+        }
+
+        // Check if this is actually a transfer log
+        // https://github.com/alloy-rs/core/issues/589
+        if log.topics().first() != Some(&Transfer::SIGNATURE_HASH) {
+          continue;
+        }
+
+        let log_index = log.log_index.ok_or_else(|| {
+          TransportErrorKind::Custom("log didn't have its index set".to_string().into())
+        })?;
+        let log = log
+          .log_decode::<Transfer>()
+          .map_err(|e| {
+            TransportErrorKind::Custom(format!("failed to decode Transfer log: {e:?}").into())
+          })?
+          .inner
+          .data;
+
+        // Ensure the top-level transfer is equivalent to the transfer this log represents. Since
+        // we can't find the exact top-level transfer without tracing the call, we just rule the
+        // first equivalent transfer as THE top-level transfer
+        if !((log.from == from) && (log.to == to) && (log.value == value)) {
+          continue;
+        }
+
+        // Read the data appended after
+        let encoded = call.abi_encode();
+        let data = transaction.input.as_ref()[encoded.len() ..].to_vec();
+
+        return Ok(Some(MatchedTopLevelTransfer {
+          transfer: TopLevelTransfer {
+            // Since there's only one top-level transfer per TX, set the ID to the TX ID
+            id: *transaction_id,
+            from: *log.from.0,
+            amount: log.value,
+            data,
+          },
+          log_index,
+        }));
+      }
+    }
+
+    Ok(None)
+  }
+
+  /// Fetch all top-level transfers to the specified address.
   pub async fn top_level_transfers(
     &self,
     block: u64,
     to: Address,
-  ) -> Result<Vec<TopLevelErc20Transfer>, RpcError<TransportErrorKind>> {
+  ) -> Result<Vec<TopLevelTransfer>, RpcError<TransportErrorKind>> {
+    // Get all transfers within this block
     let filter = Filter::new().from_block(block).to_block(block).address(self.1);
     let filter = filter.event_signature(Transfer::SIGNATURE_HASH);
     let mut to_topic = [0; 32];
@@ -60,83 +162,46 @@ impl Erc20 {
     let filter = filter.topic2(B256::from(to_topic));
     let logs = self.0.get_logs(&filter).await?;
 
-    /*
+    // These logs are for all transactions which performed any transfer
-      A set of all transactions we've handled a transfer from. This handles the edge case where a
+    // We now check each transaction for having a top-level transfer to the specified address
-      top-level transfer T somehow triggers another transfer T', with equivalent contents, within
+    let tx_ids = logs
-      the same transaction. We only want to report one transfer as only one is top-level.
+      .into_iter()
-    */
+      .map(|log| {
-    let mut handled = HashSet::new();
+        // Double check the address which emitted this log
+        if log.address() != self.1 {
+          Err(TransportErrorKind::Custom(
+            "node returned logs for a different address than requested".to_string().into(),
+          ))?;
+        }
+
+        log.transaction_hash.ok_or_else(|| {
+          TransportErrorKind::Custom("log didn't specify its transaction hash".to_string().into())
+        })
+      })
+      .collect::<Result<HashSet<_>, _>>()?;
+
+    let mut join_set = JoinSet::new();
+    for tx_id in tx_ids {
+      join_set.spawn(Self::match_top_level_transfer(self.0.clone(), tx_id, to));
+    }
 
     let mut top_level_transfers = vec![];
-    for log in logs {
+    while let Some(top_level_transfer) = join_set.join_next().await {
-      // Double check the address which emitted this log
+      // This is an error if a task panics or aborts
-      if log.address() != self.1 {
+      // Panicking on a task panic is desired behavior, and we haven't aborted any tasks
-        Err(TransportErrorKind::Custom(
+      match top_level_transfer.unwrap() {
-          "node returned logs for a different address than requested".to_string().into(),
+        // Top-level transfer
-        ))?;
+        Ok(Some(top_level_transfer)) => top_level_transfers.push(top_level_transfer.transfer),
-      }
+        // Not a top-level transfer
-
+        Ok(None) => continue,
-      let tx_id = log.transaction_hash.ok_or_else(|| {
+        // Failed to get this transaction's information so abort
-        TransportErrorKind::Custom("log didn't specify its transaction hash".to_string().into())
+        Err(e) => {
-      })?;
+          join_set.abort_all();
-      let tx = self.0.get_transaction_by_hash(tx_id).await?.ok_or_else(|| {
+          Err(e)?
-        TransportErrorKind::Custom(
-          "node didn't have the transaction which emitted a log it had".to_string().into(),
-        )
-      })?;
-
-      // If this is a top-level call...
-      if tx.to == Some(self.1) {
-        // And we recognize the call...
-        // Don't validate the encoding as this can't be re-encoded to an identical bytestring due
-        // to the InInstruction appended
-        if let Ok(call) = IERC20Calls::abi_decode(&tx.input, false) {
-          // Extract the top-level call's from/to/value
-          let (from, call_to, value) = match call {
-            IERC20Calls::transfer(transferCall { to: call_to, value }) => (tx.from, call_to, value),
-            IERC20Calls::transferFrom(transferFromCall { from, to: call_to, value }) => {
-              (from, call_to, value)
-            }
-            // Treat any other function selectors as unrecognized
-            _ => continue,
-          };
-
-          let log = log
-            .log_decode::<Transfer>()
-            .map_err(|e| {
-              TransportErrorKind::Custom(format!("failed to decode Transfer log: {e:?}").into())
-            })?
-            .inner
-            .data;
-
-          // Ensure the top-level transfer is equivalent, and this presumably isn't a log for an
-          // internal transfer
-          if (log.from != from) || (call_to != to) || (value != log.value) {
-            continue;
-          }
-
-          // Now that the top-level transfer is confirmed to be equivalent to the log, ensure it's
-          // the only log we handle
-          if handled.contains(&tx_id) {
-            continue;
-          }
-          handled.insert(tx_id);
-
-          // Read the data appended after
-          let encoded = call.abi_encode();
-          let data = tx.input.as_ref()[encoded.len() ..].to_vec();
-
-          // Push the transfer
-          top_level_transfers.push(TopLevelErc20Transfer {
-            // Since we'll only handle one log for this TX, set the ID to the TX ID
-            id: *tx_id,
-            from: *log.from.0,
-            amount: log.value,
-            data,
-          });
         }
       }
     }
+
     Ok(top_level_transfers)
   }
 }

processor/ethereum/router/src/lib.rs

@@ -25,7 +25,7 @@ use alloy_provider::{Provider, RootProvider};
 
 use ethereum_schnorr::{PublicKey, Signature};
 use ethereum_deployer::Deployer;
-use erc20::Transfer;
+use erc20::{Transfer, Erc20};
 
 use serai_client::{primitives::Amount, networks::ethereum::Address as SeraiAddress};
 
@@ -346,11 +346,6 @@ impl Router {
       let tx_hash = log.transaction_hash.ok_or_else(|| {
         TransportErrorKind::Custom("log didn't have its transaction hash set".to_string().into())
       })?;
-      let tx = self.0.get_transaction_by_hash(tx_hash).await?.ok_or_else(|| {
-        TransportErrorKind::Custom(
-          "node didn't have a transaction it had the logs of".to_string().into(),
-        )
-      })?;
 
       let log = log
         .log_decode::<InInstructionEvent>()
@@ -371,23 +366,6 @@ impl Router {
           continue;
         }
 
-        /*
-          If this also counts as a top-level transfer of a token, drop it.
-
-          This event will only exist if there's an ERC20 which has some form of programmability
-          (`onTransferFrom`), and when a top-level transfer was made, that hook made its own call
-          into the Serai router.
-
-          If such an ERC20 exists, Serai would parse it as a top-level transfer and as a router
-          InInstruction. While no such ERC20 is planned to be integrated, this enures we don't
-          allow a double-spend on that premise.
-
-          TODO: See below note.
-        */
-        if tx.to == Some(token.into()) {
-          continue;
-        }
-
         // Get all logs for this TX
         let receipt = self.0.get_transaction_receipt(tx_hash).await?.ok_or_else(|| {
           TransportErrorKind::Custom(
@@ -397,9 +375,14 @@ impl Router {
         let tx_logs = receipt.inner.logs();
 
         /*
-          TODO: If this is also a top-level transfer, drop the log from the top-level transfer and
+          The transfer which causes an InInstruction event won't be a top-level transfer.
-          only iterate over the rest of the logs.
+          Accordingly, when looking for the matching transfer, disregard the top-level transfer (if
+          one exists).
         */
+        if let Some(matched) = Erc20::match_top_level_transfer(&self.0, tx_hash, self.1).await? {
+          // Mark this log index as used so it isn't used again
+          transfer_check.insert(matched.log_index);
+        }
 
         // Find a matching transfer log
         let mut found_transfer = false;
@@ -409,6 +392,7 @@ impl Router {
               "log in transaction receipt didn't have its log index set".to_string().into(),
             )
           })?;
+
           // Ensure we didn't already use this transfer to check a distinct InInstruction event
           if transfer_check.contains(&log_index) {
             continue;
@@ -420,7 +404,7 @@ impl Router {
           }
           // Check if this is a transfer log
           // https://github.com/alloy-rs/core/issues/589
-          if tx_log.topics()[0] != Transfer::SIGNATURE_HASH {
+          if tx_log.topics().first() != Some(&Transfer::SIGNATURE_HASH) {
             continue;
           }
           let Ok(transfer) = Transfer::decode_log(&tx_log.inner.clone(), true) else { continue };