Test Ether InInstructions

It had sequential async calls with complexity O(n), with a variety of redundant
calls. There was also a constant of... 4? 5? for each item. Now, the total
sequence depth is just 3-4.

Cargo.lock

@@ -9446,7 +9446,8 @@ dependencies = [
  "alloy-sol-macro",
  "alloy-sol-types",
  "alloy-transport",
- "tokio",
+ "futures-util",
+ "serai-processor-ethereum-primitives",
 ]
 
 [[package]]
@@ -9455,6 +9456,7 @@ version = "0.1.0"
 dependencies = [
  "alloy-consensus",
  "alloy-primitives",
+ "borsh",
  "group",
  "k256",
 ]
@@ -9475,10 +9477,13 @@ dependencies = [
  "alloy-sol-macro-input",
  "alloy-sol-types",
  "alloy-transport",
+ "borsh",
  "build-solidity-contracts",
  "ethereum-schnorr-contract",
+ "futures-util",
  "group",
  "k256",
+ "parity-scale-codec",
  "rand_core",
  "serai-client",
  "serai-ethereum-test-primitives",

processor/TODO/main.rs

@@ -1,26 +1,3 @@
-use messages::{
-  coordinator::{
-    SubstrateSignableId, PlanMeta, CoordinatorMessage as CoordinatorCoordinatorMessage,
-  },
-  CoordinatorMessage,
-};
-
-use serai_env as env;
-
-use message_queue::{Service, client::MessageQueue};
-
-mod db;
-pub use db::*;
-
-mod coordinator;
-pub use coordinator::*;
-
-mod multisigs;
-use multisigs::{MultisigEvent, MultisigManager};
-
-#[cfg(test)]
-mod tests;
-
 async fn handle_coordinator_msg<D: Db, N: Network, Co: Coordinator>(
   txn: &mut D::Transaction<'_>,
   network: &N,

processor/ethereum/TODO/tests/mod.rs

@@ -43,89 +43,3 @@ pub fn key_gen() -> (HashMap<Participant, ThresholdKeys<Secp256k1>>, PublicKey)
 
   (keys, public_key)
 }
-
-// TODO: Use a proper error here
-pub async fn send(
-  provider: &RootProvider<SimpleRequest>,
-  wallet: &k256::ecdsa::SigningKey,
-  mut tx: TxLegacy,
-) -> Option<TransactionReceipt> {
-  let verifying_key = *wallet.verifying_key().as_affine();
-  let address = Address::from(address(&verifying_key.into()));
-
-  // https://github.com/alloy-rs/alloy/issues/539
-  // let chain_id = provider.get_chain_id().await.unwrap();
-  // tx.chain_id = Some(chain_id);
-  tx.chain_id = None;
-  tx.nonce = provider.get_transaction_count(address).await.unwrap();
-  // 100 gwei
-  tx.gas_price = 100_000_000_000u128;
-
-  let sig = wallet.sign_prehash_recoverable(tx.signature_hash().as_ref()).unwrap();
-  assert_eq!(address, tx.clone().into_signed(sig.into()).recover_signer().unwrap());
-  assert!(
-    provider.get_balance(address).await.unwrap() >
-      ((U256::from(tx.gas_price) * U256::from(tx.gas_limit)) + tx.value)
-  );
-
-  let mut bytes = vec![];
-  tx.encode_with_signature_fields(&Signature::from(sig), &mut bytes);
-  let pending_tx = provider.send_raw_transaction(&bytes).await.ok()?;
-  pending_tx.get_receipt().await.ok()
-}
-
-pub async fn fund_account(
-  provider: &RootProvider<SimpleRequest>,
-  wallet: &k256::ecdsa::SigningKey,
-  to_fund: Address,
-  value: U256,
-) -> Option<()> {
-  let funding_tx =
-    TxLegacy { to: TxKind::Call(to_fund), gas_limit: 21_000, value, ..Default::default() };
-  assert!(send(provider, wallet, funding_tx).await.unwrap().status());
-
-  Some(())
-}
-
-// TODO: Use a proper error here
-pub async fn deploy_contract(
-  client: Arc<RootProvider<SimpleRequest>>,
-  wallet: &k256::ecdsa::SigningKey,
-  name: &str,
-) -> Option<Address> {
-  let hex_bin_buf = std::fs::read_to_string(format!("./artifacts/{name}.bin")).unwrap();
-  let hex_bin =
-    if let Some(stripped) = hex_bin_buf.strip_prefix("0x") { stripped } else { &hex_bin_buf };
-  let bin = Bytes::from_hex(hex_bin).unwrap();
-
-  let deployment_tx = TxLegacy {
-    chain_id: None,
-    nonce: 0,
-    // 100 gwei
-    gas_price: 100_000_000_000u128,
-    gas_limit: 1_000_000,
-    to: TxKind::Create,
-    value: U256::ZERO,
-    input: bin,
-  };
-
-  let deployment_tx = deterministically_sign(deployment_tx);
-
-  // Fund the deployer address
-  fund_account(
-    &client,
-    wallet,
-    deployment_tx.recover_signer().unwrap(),
-    U256::from(deployment_tx.tx().gas_limit) * U256::from(deployment_tx.tx().gas_price),
-  )
-  .await?;
-
-  let (deployment_tx, sig, _) = deployment_tx.into_parts();
-  let mut bytes = vec![];
-  deployment_tx.encode_with_signature_fields(&sig, &mut bytes);
-  let pending_tx = client.send_raw_transaction(&bytes).await.ok()?;
-  let receipt = pending_tx.get_receipt().await.ok()?;
-  assert!(receipt.status());
-
-  Some(receipt.contract_address.unwrap())
-}

processor/ethereum/TODO/tests/router.rs

@@ -1,185 +0,0 @@
-// TODO
-
-use std::{convert::TryFrom, sync::Arc, collections::HashMap};
-
-use rand_core::OsRng;
-
-use group::Group;
-use k256::ProjectivePoint;
-use frost::{
-  curve::Secp256k1,
-  Participant, ThresholdKeys,
-  algorithm::IetfSchnorr,
-  tests::{algorithm_machines, sign},
-};
-
-use alloy_core::primitives::{Address, U256};
-
-use alloy_simple_request_transport::SimpleRequest;
-use alloy_rpc_types_eth::BlockTransactionsKind;
-use alloy_rpc_client::ClientBuilder;
-use alloy_provider::{Provider, RootProvider};
-
-use alloy_node_bindings::{Anvil, AnvilInstance};
-
-use crate::{
-  crypto::*,
-  deployer::Deployer,
-  router::{Router, abi as router},
-  tests::{key_gen, send, fund_account},
-};
-
-async fn setup_test() -> (
-  AnvilInstance,
-  Arc<RootProvider<SimpleRequest>>,
-  u64,
-  Router,
-  HashMap<Participant, ThresholdKeys<Secp256k1>>,
-  PublicKey,
-) {
-  let anvil = Anvil::new().spawn();
-
-  let provider = RootProvider::new(
-    ClientBuilder::default().transport(SimpleRequest::new(anvil.endpoint()), true),
-  );
-  let chain_id = provider.get_chain_id().await.unwrap();
-  let wallet = anvil.keys()[0].clone().into();
-  let client = Arc::new(provider);
-
-  // Make sure the Deployer constructor returns None, as it doesn't exist yet
-  assert!(Deployer::new(client.clone()).await.unwrap().is_none());
-
-  // Deploy the Deployer
-  let tx = Deployer::deployment_tx();
-  fund_account(
-    &client,
-    &wallet,
-    tx.recover_signer().unwrap(),
-    U256::from(tx.tx().gas_limit) * U256::from(tx.tx().gas_price),
-  )
-  .await
-  .unwrap();
-
-  let (tx, sig, _) = tx.into_parts();
-  let mut bytes = vec![];
-  tx.encode_with_signature_fields(&sig, &mut bytes);
-
-  let pending_tx = client.send_raw_transaction(&bytes).await.unwrap();
-  let receipt = pending_tx.get_receipt().await.unwrap();
-  assert!(receipt.status());
-  let deployer =
-    Deployer::new(client.clone()).await.expect("network error").expect("deployer wasn't deployed");
-
-  let (keys, public_key) = key_gen();
-
-  // Verify the Router constructor returns None, as it doesn't exist yet
-  assert!(deployer.find_router(client.clone(), &public_key).await.unwrap().is_none());
-
-  // Deploy the router
-  let receipt = send(&client, &anvil.keys()[0].clone().into(), deployer.deploy_router(&public_key))
-    .await
-    .unwrap();
-  assert!(receipt.status());
-  let contract = deployer.find_router(client.clone(), &public_key).await.unwrap().unwrap();
-
-  (anvil, client, chain_id, contract, keys, public_key)
-}
-
-async fn latest_block_hash(client: &RootProvider<SimpleRequest>) -> [u8; 32] {
-  client
-    .get_block(client.get_block_number().await.unwrap().into(), BlockTransactionsKind::Hashes)
-    .await
-    .unwrap()
-    .unwrap()
-    .header
-    .hash
-    .0
-}
-
-#[tokio::test]
-async fn test_deploy_contract() {
-  let (_anvil, client, _, router, _, public_key) = setup_test().await;
-
-  let block_hash = latest_block_hash(&client).await;
-  assert_eq!(router.serai_key(block_hash).await.unwrap(), public_key);
-  assert_eq!(router.nonce(block_hash).await.unwrap(), U256::try_from(1u64).unwrap());
-  // TODO: Check it emitted SeraiKeyUpdated(public_key) at its genesis
-}
-
-pub fn hash_and_sign(
-  keys: &HashMap<Participant, ThresholdKeys<Secp256k1>>,
-  public_key: &PublicKey,
-  message: &[u8],
-) -> Signature {
-  let algo = IetfSchnorr::<Secp256k1, EthereumHram>::ietf();
-  let sig =
-    sign(&mut OsRng, &algo, keys.clone(), algorithm_machines(&mut OsRng, &algo, keys), message);
-
-  Signature::new(public_key, message, sig).unwrap()
-}
-
-#[tokio::test]
-async fn test_router_update_serai_key() {
-  let (anvil, client, chain_id, contract, keys, public_key) = setup_test().await;
-
-  let next_key = loop {
-    let point = ProjectivePoint::random(&mut OsRng);
-    let Some(next_key) = PublicKey::new(point) else { continue };
-    break next_key;
-  };
-
-  let message = Router::update_serai_key_message(
-    U256::try_from(chain_id).unwrap(),
-    U256::try_from(1u64).unwrap(),
-    &next_key,
-  );
-  let sig = hash_and_sign(&keys, &public_key, &message);
-
-  let first_block_hash = latest_block_hash(&client).await;
-  assert_eq!(contract.serai_key(first_block_hash).await.unwrap(), public_key);
-
-  let receipt =
-    send(&client, &anvil.keys()[0].clone().into(), contract.update_serai_key(&next_key, &sig))
-      .await
-      .unwrap();
-  assert!(receipt.status());
-
-  let second_block_hash = latest_block_hash(&client).await;
-  assert_eq!(contract.serai_key(second_block_hash).await.unwrap(), next_key);
-  // Check this does still offer the historical state
-  assert_eq!(contract.serai_key(first_block_hash).await.unwrap(), public_key);
-  // TODO: Check logs
-
-  println!("gas used: {:?}", receipt.gas_used);
-  // println!("logs: {:?}", receipt.logs);
-}
-
-#[tokio::test]
-async fn test_router_execute() {
-  let (anvil, client, chain_id, contract, keys, public_key) = setup_test().await;
-
-  let to = Address::from([0; 20]);
-  let value = U256::ZERO;
-  let tx = router::OutInstruction { to, value, calls: vec![] };
-  let txs = vec![tx];
-
-  let first_block_hash = latest_block_hash(&client).await;
-  let nonce = contract.nonce(first_block_hash).await.unwrap();
-  assert_eq!(nonce, U256::try_from(1u64).unwrap());
-
-  let message = Router::execute_message(U256::try_from(chain_id).unwrap(), nonce, txs.clone());
-  let sig = hash_and_sign(&keys, &public_key, &message);
-
-  let receipt =
-    send(&client, &anvil.keys()[0].clone().into(), contract.execute(&txs, &sig)).await.unwrap();
-  assert!(receipt.status());
-
-  let second_block_hash = latest_block_hash(&client).await;
-  assert_eq!(contract.nonce(second_block_hash).await.unwrap(), U256::try_from(2u64).unwrap());
-  // Check this does still offer the historical state
-  assert_eq!(contract.nonce(first_block_hash).await.unwrap(), U256::try_from(1u64).unwrap());
-  // TODO: Check logs
-
-  println!("gas used: {:?}", receipt.gas_used);
-  // println!("logs: {:?}", receipt.logs);
-}

processor/ethereum/erc20/Cargo.toml

@@ -27,4 +27,6 @@ alloy-transport = { version = "0.9", default-features = false }
 alloy-simple-request-transport = { path = "../../../networks/ethereum/alloy-simple-request-transport", default-features = false }
 alloy-provider = { version = "0.9", default-features = false }
 
-tokio = { version = "1", default-features = false, features = ["rt"] }
+ethereum-primitives = { package = "serai-processor-ethereum-primitives", path = "../primitives", default-features = false }
+
+futures-util = { version = "0.3", default-features = false, features = ["std"] }

processor/ethereum/erc20/contracts/IERC20.sol

@@ -18,3 +18,17 @@ interface IERC20 {
   function approve(address spender, uint256 value) external returns (bool);
   function allowance(address owner, address spender) external view returns (uint256);
 }
+
+interface SeraiIERC20 {
+  function transferWithInInstruction01BB244A8A(
+    address to,
+    uint256 value,
+    bytes calldata inInstruction
+  ) external returns (bool);
+  function transferFromWithInInstruction00081948E0(
+    address from,
+    address to,
+    uint256 value,
+    bytes calldata inInstruction
+  ) external returns (bool);
+}

processor/ethereum/erc20/src/lib.rs

@@ -2,22 +2,26 @@
 #![doc = include_str!("../README.md")]
 #![deny(missing_docs)]
 
-use std::{sync::Arc, collections::HashSet};
+use core::borrow::Borrow;
+use std::{sync::Arc, collections::HashMap};
 
-use alloy_core::primitives::{Address, B256, U256};
+use alloy_core::primitives::{Address, U256};
 
 use alloy_sol_types::{SolInterface, SolEvent};
 
-use alloy_rpc_types_eth::{Filter, TransactionTrait};
+use alloy_rpc_types_eth::{Log, Filter, TransactionTrait};
 use alloy_transport::{TransportErrorKind, RpcError};
 use alloy_simple_request_transport::SimpleRequest;
 use alloy_provider::{Provider, RootProvider};
 
-use tokio::task::JoinSet;
+use ethereum_primitives::LogIndex;
+
+use futures_util::stream::{StreamExt, FuturesUnordered};
 
 #[rustfmt::skip]
 #[expect(warnings)]
 #[expect(needless_pass_by_value)]
+#[expect(missing_docs)]
 #[expect(clippy::all)]
 #[expect(clippy::ignored_unit_patterns)]
 #[expect(clippy::redundant_closure_for_method_calls)]
@@ -25,15 +29,28 @@ mod abi {
   alloy_sol_macro::sol!("contracts/IERC20.sol");
 }
 use abi::IERC20::{IERC20Calls, transferCall, transferFromCall};
+use abi::SeraiIERC20::SeraiIERC20Calls;
 pub use abi::IERC20::Transfer;
+pub use abi::SeraiIERC20::{
+  transferWithInInstruction01BB244A8ACall as transferWithInInstructionCall,
+  transferFromWithInInstruction00081948E0Call as transferFromWithInInstructionCall,
+};
+
+#[cfg(test)]
+mod tests;
 
 /// A top-level ERC20 transfer
+///
+/// This does not include `token`, `to` fields. Those are assumed contextual to the creation of
+/// this.
 #[derive(Clone, Debug)]
 pub struct TopLevelTransfer {
   /// The ID of the event for this transfer.
-  pub id: ([u8; 32], u64),
+  pub id: LogIndex,
+  /// The hash of the transaction which caused this transfer.
+  pub transaction_hash: [u8; 32],
   /// The address which made the transfer.
-  pub from: [u8; 20],
+  pub from: Address,
   /// The amount transferred.
   pub amount: U256,
   /// The data appended after the call itself.
@@ -42,156 +59,187 @@ pub struct TopLevelTransfer {
 
 /// A view for an ERC20 contract.
 #[derive(Clone, Debug)]
-pub struct Erc20(Arc<RootProvider<SimpleRequest>>, Address);
+pub struct Erc20 {
+  provider: Arc<RootProvider<SimpleRequest>>,
+  address: Address,
+}
 impl Erc20 {
   /// Construct a new view of the specified ERC20 contract.
-  pub fn new(provider: Arc<RootProvider<SimpleRequest>>, address: [u8; 20]) -> Self {
+  pub fn new(provider: Arc<RootProvider<SimpleRequest>>, address: Address) -> Self {
-    Self(provider, Address::from(&address))
+    Self { provider, address }
   }
 
-  /// Match a transaction for its top-level transfer to the specified address (if one exists).
+  /// The filter for transfer logs of the specified ERC20, to the specified recipient.
-  pub async fn match_top_level_transfer(
+  pub fn transfer_filter(from_block: u64, to_block: u64, erc20: Address, to: Address) -> Filter {
+    let filter = Filter::new().from_block(from_block).to_block(to_block);
+    filter.address(erc20).event_signature(Transfer::SIGNATURE_HASH).topic2(to.into_word())
+  }
+
+  /// Yield the top-level transfer for the specified transaction (if one exists).
+  ///
+  /// The passed-in logs MUST be the logs for this transaction. The logs MUST be filtered to the
+  /// `Transfer` events of the intended token(s) and the intended `to` transferred to. These
+  /// properties are completely unchecked and assumed to be the case.
+  ///
+  /// This does NOT yield THE top-level transfer. If multiple `Transfer` events have identical
+  /// structure to the top-level transfer call, the earliest `Transfer` event present in the logs
+  /// is considered the top-level transfer.
+  // Yielding THE top-level transfer would require tracing the transaction execution and isn't
+  // worth the effort.
+  pub async fn top_level_transfer(
     provider: impl AsRef<RootProvider<SimpleRequest>>,
-    transaction_id: B256,
+    transaction_hash: [u8; 32],
-    to: Address,
+    mut transfer_logs: Vec<impl Borrow<Log>>,
   ) -> Result<Option<TopLevelTransfer>, RpcError<TransportErrorKind>> {
     // Fetch the transaction
     let transaction =
-      provider.as_ref().get_transaction_by_hash(transaction_id).await?.ok_or_else(|| {
+      provider.as_ref().get_transaction_by_hash(transaction_hash.into()).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.inner.input(), false) {
+    let Ok(call) = IERC20Calls::abi_decode(transaction.inner.input(), false) else {
+      return Ok(None);
+    };
+
     // Extract the top-level call's from/to/value
-      let (from, call_to, value) = match call {
+    let (from, 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 {
+    // Sort the logs to ensure the the earliest logs are first
-        return Ok(None);
+    transfer_logs.sort_by_key(|log| log.borrow().log_index);
+    // Find the log for this top-level transfer
+    for log in transfer_logs {
+      // Check the log is for the called contract
+      // This handles the edge case where we're checking if transfers of token X were top-level and
+      // a transfer of token Y (with equivalent structure) was top-level
+      if Some(log.borrow().address()) != transaction.inner.to() {
+        continue;
       }
 
-      // Fetch the transaction's logs
+      // Since the caller is responsible for filtering these to `Transfer` events, we can assume
-      let receipt =
+      // this is a non-compliant ERC20 or an error with the logs fetched. We assume ERC20
-        provider.as_ref().get_transaction_receipt(transaction_id).await?.ok_or_else(|| {
+      // compliance here, making this an RPC error
-          TransportErrorKind::Custom(
+      let log = log.borrow().log_decode::<Transfer>().map_err(|_| {
-            "node didn't have receipt for a transaction we were matching for a top-level transfer"
+        TransportErrorKind::Custom("log didn't include a valid transfer event".to_string().into())
-              .to_string()
-              .into(),
-          )
       })?;
 
-      // Find the log for this transfer
+      let block_hash = log.block_hash.ok_or_else(|| {
-      for log in receipt.inner.logs() {
+        TransportErrorKind::Custom("log didn't have its block hash set".to_string().into())
-        // If this log was emitted by a different contract, continue
+      })?;
-        if Some(log.address()) != transaction.inner.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
+      let log = log.inner.data;
-          .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
+      // Ensure the top-level transfer is equivalent to the transfer this log represents
-        // 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 = if let Ok(call) = SeraiIERC20Calls::abi_decode(transaction.inner.input(), true) {
-        let data = transaction.inner.input().as_ref()[encoded.len() ..].to_vec();
+        match call {
+          SeraiIERC20Calls::transferWithInInstruction01BB244A8A(
+            transferWithInInstructionCall { inInstruction, .. },
+          ) |
+          SeraiIERC20Calls::transferFromWithInInstruction00081948E0(
+            transferFromWithInInstructionCall { inInstruction, .. },
+          ) => Vec::from(inInstruction),
+        }
+      } else {
+        // We don't error here so this transfer is propagated up the stack, even without the
+        // InInstruction. In practice, Serai should acknowledge this and return it to the sender
+        vec![]
+      };
 
       return Ok(Some(TopLevelTransfer {
-          id: (*transaction_id, log_index),
+        id: LogIndex { block_hash: *block_hash, index_within_block: log_index },
-          from: *log.from.0,
+        transaction_hash,
+        from: log.from,
         amount: log.value,
         data,
       }));
     }
-    }
 
     Ok(None)
   }
 
-  /// Fetch all top-level transfers to the specified address.
+  /// Fetch all top-level transfers to the specified address for this token.
   ///
   /// The result of this function is unordered.
-  pub async fn top_level_transfers(
+  pub async fn top_level_transfers_unordered(
     &self,
-    block: u64,
+    from_block: u64,
+    to_block: u64,
     to: Address,
   ) -> Result<Vec<TopLevelTransfer>, RpcError<TransportErrorKind>> {
-    // Get all transfers within this block
+    // Get all transfers within these blocks
-    let filter = Filter::new().from_block(block).to_block(block).address(self.1);
+    let logs = self
-    let filter = filter.event_signature(Transfer::SIGNATURE_HASH);
+      .provider
-    let mut to_topic = [0; 32];
+      .get_logs(&Self::transfer_filter(from_block, to_block, self.address, to))
-    to_topic[12 ..].copy_from_slice(to.as_ref());
+      .await?;
-    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
+    // The logs, indexed by their transactions
-    // We now check each transaction for having a top-level transfer to the specified address
+    let mut transaction_logs = HashMap::new();
-    let tx_ids = logs
+    // Index the logs by their transactions
-      .into_iter()
+    for log in logs {
-      .map(|log| {
       // Double check the address which emitted this log
-        if log.address() != self.1 {
+      if log.address() != self.address {
         Err(TransportErrorKind::Custom(
           "node returned logs for a different address than requested".to_string().into(),
         ))?;
       }
+      // Double check the event signature for this log
+      if log.topics().first() != Some(&Transfer::SIGNATURE_HASH) {
+        Err(TransportErrorKind::Custom(
+          "node returned a log for a different topic than filtered to".to_string().into(),
+        ))?;
+      }
+      // Double check the `to` topic
+      if log.topics().get(2) != Some(&to.into_word()) {
+        Err(TransportErrorKind::Custom(
+          "node returned a transfer for a different `to` than filtered to".to_string().into(),
+        ))?;
+      }
 
-        log.transaction_hash.ok_or_else(|| {
+      let tx_id = log
+        .transaction_hash
+        .ok_or_else(|| {
           TransportErrorKind::Custom("log didn't specify its transaction hash".to_string().into())
-        })
+        })?
-      })
+        .0;
-      .collect::<Result<HashSet<_>, _>>()?;
 
-    let mut join_set = JoinSet::new();
+      transaction_logs.entry(tx_id).or_insert_with(|| Vec::with_capacity(1)).push(log);
-    for tx_id in tx_ids {
+    }
-      join_set.spawn(Self::match_top_level_transfer(self.0.clone(), tx_id, to));
+
+    // Use `FuturesUnordered` so these RPC calls run in parallel
+    let mut futures = FuturesUnordered::new();
+    for (tx_id, transfer_logs) in transaction_logs {
+      futures.push(Self::top_level_transfer(&self.provider, tx_id, transfer_logs));
     }
 
     let mut top_level_transfers = vec![];
-    while let Some(top_level_transfer) = join_set.join_next().await {
+    while let Some(top_level_transfer) = futures.next().await {
-      // This is an error if a task panics or aborts
+      match top_level_transfer {
-      // Panicking on a task panic is desired behavior, and we haven't aborted any tasks
-      match top_level_transfer.unwrap() {
         // Top-level transfer
         Ok(Some(top_level_transfer)) => top_level_transfers.push(top_level_transfer),
         // Not a top-level transfer
         Ok(None) => continue,
         // Failed to get this transaction's information so abort
-        Err(e) => {
+        Err(e) => Err(e)?,
-          join_set.abort_all();
-          Err(e)?
       }
     }
-    }
-
     Ok(top_level_transfers)
   }
 }

processor/ethereum/erc20/src/tests.rs

@@ -0,0 +1,13 @@
+use alloy_sol_types::SolCall;
+
+#[test]
+fn selector_collisions() {
+  assert_eq!(
+    crate::abi::IERC20::transferCall::SELECTOR,
+    crate::abi::SeraiIERC20::transferWithInInstruction01BB244A8ACall::SELECTOR
+  );
+  assert_eq!(
+    crate::abi::IERC20::transferFromCall::SELECTOR,
+    crate::abi::SeraiIERC20::transferFromWithInInstruction00081948E0Call::SELECTOR
+  );
+}

processor/ethereum/primitives/Cargo.toml

@@ -17,6 +17,8 @@ rustdoc-args = ["--cfg", "docsrs"]
 workspace = true
 
 [dependencies]
+borsh = { version = "1", default-features = false, features = ["std", "derive", "de_strict_order"] }
+
 group = { version = "0.13", default-features = false }
 k256 = { version = "^0.13.1", default-features = false, features = ["std", "arithmetic"] }
 

processor/ethereum/primitives/src/borsh.rs

@@ -0,0 +1,24 @@
+use ::borsh::{io, BorshSerialize, BorshDeserialize};
+
+use alloy_primitives::{U256, Address};
+
+/// Serialize a U256 with a borsh-compatible API.
+pub fn serialize_u256(value: &U256, writer: &mut impl io::Write) -> io::Result<()> {
+  let value: [u8; 32] = value.to_be_bytes();
+  value.serialize(writer)
+}
+
+/// Deserialize an address with a borsh-compatible API.
+pub fn deserialize_u256(reader: &mut impl io::Read) -> io::Result<U256> {
+  <[u8; 32]>::deserialize_reader(reader).map(|value| U256::from_be_bytes(value))
+}
+
+/// Serialize an address with a borsh-compatible API.
+pub fn serialize_address(address: &Address, writer: &mut impl io::Write) -> io::Result<()> {
+  <[u8; 20]>::from(address.0).serialize(writer)
+}
+
+/// Deserialize an address with a borsh-compatible API.
+pub fn deserialize_address(reader: &mut impl io::Read) -> io::Result<Address> {
+  <[u8; 20]>::deserialize_reader(reader).map(|address| Address(address.into()))
+}

processor/ethereum/primitives/src/lib.rs

@@ -2,12 +2,27 @@
 #![doc = include_str!("../README.md")]
 #![deny(missing_docs)]
 
+use ::borsh::{BorshSerialize, BorshDeserialize};
+
 use group::ff::PrimeField;
 use k256::Scalar;
 
 use alloy_primitives::PrimitiveSignature;
 use alloy_consensus::{SignableTransaction, Signed, TxLegacy};
 
+mod borsh;
+pub use borsh::*;
+
+/// An index of a log within a block.
+#[derive(Clone, Copy, PartialEq, Eq, Debug, BorshSerialize, BorshDeserialize)]
+#[borsh(crate = "::borsh")]
+pub struct LogIndex {
+  /// The hash of the block which produced this log.
+  pub block_hash: [u8; 32],
+  /// The index of this log within the execution of the block.
+  pub index_within_block: u64,
+}
+
 /// The Keccak256 hash function.
 pub fn keccak256(data: impl AsRef<[u8]>) -> [u8; 32] {
   alloy_primitives::keccak256(data.as_ref()).into()

processor/ethereum/router/Cargo.toml

@@ -17,6 +17,8 @@ rustdoc-args = ["--cfg", "docsrs"]
 workspace = true
 
 [dependencies]
+borsh = { version = "1", default-features = false, features = ["std", "derive", "de_strict_order"] }
+
 group = { version = "0.13", default-features = false }
 
 alloy-core = { version = "0.8", default-features = false }
@@ -37,8 +39,11 @@ ethereum-primitives = { package = "serai-processor-ethereum-primitives", path =
 ethereum-deployer = { package = "serai-processor-ethereum-deployer", path = "../deployer", default-features = false }
 erc20 = { package = "serai-processor-ethereum-erc20", path = "../erc20", default-features = false }
 
+scale = { package = "parity-scale-codec", version = "3", default-features = false, features = ["std"] }
 serai-client = { path = "../../../substrate/client", default-features = false, features = ["ethereum"] }
 
+futures-util = { version = "0.3", default-features = false, features = ["std"] }
+
 [build-dependencies]
 build-solidity-contracts = { path = "../../../networks/ethereum/build-contracts", default-features = false }
 

processor/ethereum/router/contracts/IRouter.sol

@@ -34,15 +34,16 @@ interface IRouterWithoutCollisions {
    *   An `OutInstruction` is considered as having succeeded if the call transferring ETH doesn't
    *   fail, the ERC20 transfer doesn't fail, and any executed code doesn't revert.
    */
-  event Executed(uint256 indexed nonce, bytes32 indexed messageHash, bytes results);
+  event Batch(uint256 indexed nonce, bytes32 indexed messageHash, bytes results);
 
   /// @notice Emitted when `escapeHatch` is invoked
   /// @param escapeTo The address to escape to
-  event EscapeHatch(address indexed escapeTo);
+  event EscapeHatch(uint256 indexed nonce, address indexed escapeTo);
 
   /// @notice Emitted when coins escape through the escape hatch
   /// @param coin The coin which escaped
-  event Escaped(address indexed coin);
+  /// @param amount The amount which escaped
+  event Escaped(address indexed coin, uint256 amount);
 
   /// @notice The key for Serai was invalid
   /// @dev This is incomplete and not always guaranteed to be thrown upon an invalid key
@@ -57,13 +58,17 @@ interface IRouterWithoutCollisions {
   /// @notice The call to an ERC20's `transferFrom` failed
   error TransferFromFailed();
 
-  /// @notice `execute` was re-entered
+  /// @notice A non-reentrant function was re-entered
-  error ReenteredExecute();
+  error Reentered();
 
   /// @notice An invalid address to escape to was specified.
   error InvalidEscapeAddress();
+  /// @notice The escape address wasn't a contract.
+  error EscapeAddressWasNotAContract();
   /// @notice Escaping when escape hatch wasn't invoked.
   error EscapeHatchNotInvoked();
+  /// @notice Escaping failed to transfer out.
+  error EscapeFailed();
 
   /// @notice Transfer coins into Serai with an instruction
   /// @param coin The coin to transfer in (address(0) if Ether)
@@ -122,7 +127,10 @@ interface IRouter is IRouterWithoutCollisions {
   }
 
   /// @title The type of destination
-  /// @dev A destination is either an address or a blob of code to deploy and call
+  /**
+   * @dev A destination is either an ABI-encoded address or an ABI-encoded `CodeDestination`
+   *   containing code to deploy (invoking its constructor).
+   */
   enum DestinationType {
     Address,
     Code

processor/ethereum/router/contracts/Router.sol

@@ -25,13 +25,11 @@ import "IRouter.sol";
 /// @author Luke Parker <lukeparker@serai.exchange>
 /// @notice Intakes coins for the Serai network and handles relaying batches of transfers out
 contract Router is IRouterWithoutCollisions {
+  /// @dev The code hash for a non-empty account without code
+  bytes32 constant ACCOUNT_WITHOUT_CODE_CODEHASH = keccak256("");
+
   /// @dev The address in transient storage used for the reentrancy guard
-  bytes32 constant EXECUTE_REENTRANCY_GUARD_SLOT = bytes32(
+  bytes32 constant REENTRANCY_GUARD_SLOT = bytes32(uint256(keccak256("ReentrancyGuard Router")) - 1);
-    /*
-      keccak256("ReentrancyGuard Router.execute") - 1
-    */
-    0xcf124a063de1614fedbd6b47187f98bf8873a1ae83da5c179a5881162f5b2401
-  );
 
   /**
    * @dev The next nonce used to determine the address of contracts deployed with CREATE. This is
@@ -65,6 +63,28 @@ contract Router is IRouterWithoutCollisions {
   /// @dev The address escaped to
   address private _escapedTo;
 
+  /// @dev Acquire the re-entrancy lock for the lifetime of this transaction
+  modifier nonReentrant() {
+    bytes32 reentrancyGuardSlot = REENTRANCY_GUARD_SLOT;
+    bytes32 priorEntered;
+    // slither-disable-next-line assembly
+    assembly {
+      priorEntered := tload(reentrancyGuardSlot)
+      tstore(reentrancyGuardSlot, 1)
+    }
+    if (priorEntered != bytes32(0)) {
+      revert Reentered();
+    }
+
+    _;
+
+    // Clear the re-entrancy guard to allow multiple transactions to non-re-entrant functions within
+    // a transaction
+    assembly {
+      tstore(reentrancyGuardSlot, 0)
+    }
+  }
+
   /// @dev Set the next Serai key. This does not read from/write to `_nextNonce`
   /// @param nonceUpdatedWith The nonce used to set the next key
   /// @param nextSeraiKeyVar The key to set as next
@@ -140,15 +160,16 @@ contract Router is IRouterWithoutCollisions {
     bytes32 signatureS;
 
     // slither-disable-next-line assembly
+    uint256 chainID = block.chainid;
     assembly {
       // Read the signature (placed after the function signature)
       signatureC := mload(add(message, 36))
       signatureS := mload(add(message, 68))
 
-      // Overwrite the signature challenge with the nonce
+      // Overwrite the signature challenge with the chain ID
-      mstore(add(message, 36), nonceUsed)
+      mstore(add(message, 36), chainID)
-      // Overwrite the signature response with 0
+      // Overwrite the signature response with the nonce
-      mstore(add(message, 68), 0)
+      mstore(add(message, 68), nonceUsed)
 
       // Calculate the message hash
       messageHash := keccak256(add(message, 32), messageLen)
@@ -405,6 +426,12 @@ contract Router is IRouterWithoutCollisions {
    *  fee.
    *
    *  The hex bytes are to cause a function selector collision with `IRouter.execute`.
+   *
+   *  Re-entrancy is prevented because we emit a bitmask of which `OutInstruction`s succeeded. Doing
+   *  that requires executing the `OutInstruction`s, which may re-enter here. While our application
+   *  of CEI with `verifySignature` prevents replays, re-entrancy would allow out-of-order
+   *  completion for the execution of batches (despite their in-order start of execution) which
+   *  isn't a headache worth dealing with.
    */
   // @param signature The signature by the current key for Serai's Ethereum validators
   // @param coin The coin all of these `OutInstruction`s are for
@@ -412,26 +439,7 @@ contract Router is IRouterWithoutCollisions {
   // @param outs The `OutInstruction`s to act on
   // Each individual call is explicitly metered to ensure there isn't a DoS here
   // slither-disable-next-line calls-loop,reentrancy-events
-  function execute4DE42904() external {
+  function execute4DE42904() external nonReentrant {
-    /*
-      Prevent re-entrancy.
-
-      We emit a bitmask of which `OutInstruction`s succeeded. Doing that requires executing the
-      `OutInstruction`s, which may re-enter here. While our application of CEI with verifySignature
-      prevents replays, re-entrancy would allow out-of-order execution of batches (despite their
-      in-order start of execution) which isn't a headache worth dealing with.
-    */
-    bytes32 executeReentrancyGuardSlot = EXECUTE_REENTRANCY_GUARD_SLOT;
-    bytes32 priorEntered;
-    // slither-disable-next-line assembly
-    assembly {
-      priorEntered := tload(executeReentrancyGuardSlot)
-      tstore(executeReentrancyGuardSlot, 1)
-    }
-    if (priorEntered != bytes32(0)) {
-      revert ReenteredExecute();
-    }
-
     (uint256 nonceUsed, bytes memory args, bytes32 message) = verifySignature(_seraiKey);
     (,, address coin, uint256 fee, IRouter.OutInstruction[] memory outs) =
       abi.decode(args, (bytes32, bytes32, address, uint256, IRouter.OutInstruction[]));
@@ -509,11 +517,11 @@ contract Router is IRouterWithoutCollisions {
     }
 
     /*
-      Emit execution with the status of all included events.
+      Emit batch execution with the status of all included events.
 
       This is an effect after interactions yet we have a reentrancy guard making this safe.
     */
-    emit Executed(nonceUsed, message, results);
+    emit Batch(nonceUsed, message, results);
 
     // Transfer the fee to the relayer
     transferOut(msg.sender, coin, fee);
@@ -529,13 +537,35 @@ contract Router is IRouterWithoutCollisions {
   // @param escapeTo The address to escape to
   function escapeHatchDCDD91CC() external {
     // Verify the signature
-    (, bytes memory args,) = verifySignature(_seraiKey);
+    (uint256 nonceUsed, bytes memory args,) = verifySignature(_seraiKey);
 
     (,, address escapeTo) = abi.decode(args, (bytes32, bytes32, address));
 
     if (escapeTo == address(0)) {
       revert InvalidEscapeAddress();
     }
+
+    /*
+      We could define the escape hatch as having its own confirmation flow, as new keys do, but new
+      contracts don't face all of the cryptographic concerns faced by new keys. New contracts also
+      would presumably be moved to after strict review, making the chance of specifying the wrong
+      contract incredibly unlikely.
+
+      The only check performed accordingly (with no confirmation flow) is that the new contract is
+      in fact a contract. This is done to confirm the contract was successfully deployed on this
+      blockchain.
+
+      This check is also comprehensive to the zero-address case, but this function doesn't have to
+      be perfectly optimized and it's better to explicitly handle that due to it being its own
+      invariant.
+    */
+    {
+      bytes32 codehash = escapeTo.codehash;
+      if ((codehash == bytes32(0)) || (codehash == ACCOUNT_WITHOUT_CODE_CODEHASH)) {
+        revert EscapeAddressWasNotAContract();
+      }
+    }
+
     /*
       We want to define the escape hatch so coins here now, and latently received, can be forwarded.
       If the last Serai key set could update the escape hatch, they could siphon off latently
@@ -546,7 +576,7 @@ contract Router is IRouterWithoutCollisions {
     }
 
     _escapedTo = escapeTo;
-    emit EscapeHatch(escapeTo);
+    emit EscapeHatch(nonceUsed, escapeTo);
   }
 
   /// @notice Escape coins after the escape hatch has been invoked
@@ -556,8 +586,6 @@ contract Router is IRouterWithoutCollisions {
       revert EscapeHatchNotInvoked();
     }
 
-    emit Escaped(coin);
-
     // Fetch the amount to escape
     uint256 amount = address(this).balance;
     if (coin != address(0)) {
@@ -565,7 +593,13 @@ contract Router is IRouterWithoutCollisions {
     }
 
     // Perform the transfer
-    transferOut(_escapedTo, coin, amount);
+    // While this can be re-entered to try escaping our balance twice, the outer call will fail
+    if (!transferOut(_escapedTo, coin, amount)) {
+      revert EscapeFailed();
+    }
+
+    // Since we successfully escaped this amount, emit the event for it
+    emit Escaped(coin, amount);
   }
 
   /// @notice Fetch the next nonce to use by an action published to this contract

processor/ethereum/router/src/tests/constants.rs

@@ -0,0 +1,21 @@
+use alloy_sol_types::SolCall;
+
+#[test]
+fn selector_collisions() {
+  assert_eq!(
+    crate::_irouter_abi::IRouter::confirmNextSeraiKeyCall::SELECTOR,
+    crate::_router_abi::Router::confirmNextSeraiKey34AC53ACCall::SELECTOR
+  );
+  assert_eq!(
+    crate::_irouter_abi::IRouter::updateSeraiKeyCall::SELECTOR,
+    crate::_router_abi::Router::updateSeraiKey5A8542A2Call::SELECTOR
+  );
+  assert_eq!(
+    crate::_irouter_abi::IRouter::executeCall::SELECTOR,
+    crate::_router_abi::Router::execute4DE42904Call::SELECTOR
+  );
+  assert_eq!(
+    crate::_irouter_abi::IRouter::escapeHatchCall::SELECTOR,
+    crate::_router_abi::Router::escapeHatchDCDD91CCCall::SELECTOR
+  );
+}

processor/ethereum/router/src/tests/mod.rs

@@ -10,51 +10,34 @@ use alloy_sol_types::SolCall;
 
 use alloy_consensus::TxLegacy;
 
-use alloy_rpc_types_eth::{BlockNumberOrTag, TransactionReceipt};
+#[rustfmt::skip]
+use alloy_rpc_types_eth::{BlockNumberOrTag, TransactionInput, TransactionRequest, TransactionReceipt};
 use alloy_simple_request_transport::SimpleRequest;
 use alloy_rpc_client::ClientBuilder;
-use alloy_provider::RootProvider;
+use alloy_provider::{Provider, RootProvider};
 
 use alloy_node_bindings::{Anvil, AnvilInstance};
 
+use scale::Encode;
+use serai_client::{
+  primitives::SeraiAddress,
+  in_instructions::primitives::{
+    InInstruction as SeraiInInstruction, RefundableInInstruction, Shorthand,
+  },
+};
+
+use ethereum_primitives::LogIndex;
 use ethereum_schnorr::{PublicKey, Signature};
 use ethereum_deployer::Deployer;
 
-use crate::{Coin, OutInstructions, Router};
+use crate::{
+  _irouter_abi::IRouterWithoutCollisions::{
+    self as IRouter, IRouterWithoutCollisionsErrors as IRouterErrors,
+  },
+  Coin, InInstruction, OutInstructions, Router, Executed, Escape,
+};
 
-mod read_write;
+mod constants;
-
-#[test]
-fn execute_reentrancy_guard() {
-  let hash = alloy_core::primitives::keccak256(b"ReentrancyGuard Router.execute");
-  assert_eq!(
-    alloy_core::primitives::hex::encode(
-      (U256::from_be_slice(hash.as_ref()) - U256::from(1u8)).to_be_bytes::<32>()
-    ),
-    // Constant from the Router contract
-    "cf124a063de1614fedbd6b47187f98bf8873a1ae83da5c179a5881162f5b2401",
-  );
-}
-
-#[test]
-fn selector_collisions() {
-  assert_eq!(
-    crate::_irouter_abi::IRouter::confirmNextSeraiKeyCall::SELECTOR,
-    crate::_router_abi::Router::confirmNextSeraiKey34AC53ACCall::SELECTOR
-  );
-  assert_eq!(
-    crate::_irouter_abi::IRouter::updateSeraiKeyCall::SELECTOR,
-    crate::_router_abi::Router::updateSeraiKey5A8542A2Call::SELECTOR
-  );
-  assert_eq!(
-    crate::_irouter_abi::IRouter::executeCall::SELECTOR,
-    crate::_router_abi::Router::execute4DE42904Call::SELECTOR
-  );
-  assert_eq!(
-    crate::_irouter_abi::IRouter::escapeHatchCall::SELECTOR,
-    crate::_router_abi::Router::escapeHatchDCDD91CCCall::SELECTOR
-  );
-}
 
 pub(crate) fn test_key() -> (Scalar, PublicKey) {
   loop {
@@ -66,13 +49,72 @@ pub(crate) fn test_key() -> (Scalar, PublicKey) {
   }
 }
 
-async fn setup_test(
+fn sign(key: (Scalar, PublicKey), msg: &[u8]) -> Signature {
-) -> (AnvilInstance, Arc<RootProvider<SimpleRequest>>, Router, (Scalar, PublicKey)) {
+  let nonce = Scalar::random(&mut OsRng);
-  let anvil = Anvil::new().spawn();
+  let c = Signature::challenge(ProjectivePoint::GENERATOR * nonce, &key.1, msg);
+  let s = nonce + (c * key.0);
+  Signature::new(c, s).unwrap()
+}
+
+/// Calculate the gas used by a transaction if none of its calldata's bytes were zero
+struct CalldataAgnosticGas;
+impl CalldataAgnosticGas {
+  fn calculate(tx: &TxLegacy, mut gas_used: u64) -> u64 {
+    const ZERO_BYTE_GAS_COST: u64 = 4;
+    const NON_ZERO_BYTE_GAS_COST: u64 = 16;
+    for b in &tx.input {
+      if *b == 0 {
+        gas_used += NON_ZERO_BYTE_GAS_COST - ZERO_BYTE_GAS_COST;
+      }
+    }
+    gas_used
+  }
+}
+
+struct RouterState {
+  next_key: Option<(Scalar, PublicKey)>,
+  key: Option<(Scalar, PublicKey)>,
+  next_nonce: u64,
+  escaped_to: Option<Address>,
+}
+
+struct Test {
+  #[allow(unused)]
+  anvil: AnvilInstance,
+  provider: Arc<RootProvider<SimpleRequest>>,
+  chain_id: U256,
+  router: Router,
+  state: RouterState,
+}
+
+impl Test {
+  async fn verify_state(&self) {
+    assert_eq!(
+      self.router.next_key(BlockNumberOrTag::Latest.into()).await.unwrap(),
+      self.state.next_key.map(|key| key.1)
+    );
+    assert_eq!(
+      self.router.key(BlockNumberOrTag::Latest.into()).await.unwrap(),
+      self.state.key.map(|key| key.1)
+    );
+    assert_eq!(
+      self.router.next_nonce(BlockNumberOrTag::Latest.into()).await.unwrap(),
+      self.state.next_nonce
+    );
+    assert_eq!(
+      self.router.escaped_to(BlockNumberOrTag::Latest.into()).await.unwrap(),
+      self.state.escaped_to,
+    );
+  }
+
+  async fn new() -> Self {
+    // The following is explicitly only evaluated against the cancun network upgrade at this time
+    let anvil = Anvil::new().arg("--hardfork").arg("cancun").spawn();
 
     let provider = Arc::new(RootProvider::new(
       ClientBuilder::default().transport(SimpleRequest::new(anvil.endpoint()), true),
     ));
+    let chain_id = U256::from(provider.get_chain_id().await.unwrap());
 
     let (private_key, public_key) = test_key();
     assert!(Router::new(provider.clone(), &public_key).await.unwrap().is_none());
@@ -81,96 +123,400 @@ async fn setup_test(
     let receipt = ethereum_test_primitives::publish_tx(&provider, Deployer::deployment_tx()).await;
     assert!(receipt.status());
 
-  // Get the TX to deploy the Router
     let mut tx = Router::deployment_tx(&public_key);
-  // Set a gas price (100 gwei)
+    tx.gas_limit = 1_100_000;
     tx.gas_price = 100_000_000_000;
-  // Sign it
     let tx = ethereum_primitives::deterministically_sign(tx);
-  // Publish it
     let receipt = ethereum_test_primitives::publish_tx(&provider, tx).await;
     assert!(receipt.status());
-  assert_eq!(Router::DEPLOYMENT_GAS, ((receipt.gas_used + 1000) / 1000) * 1000);
 
     let router = Router::new(provider.clone(), &public_key).await.unwrap().unwrap();
+    let state = RouterState {
+      next_key: Some((private_key, public_key)),
+      key: None,
+      // Nonce 0 should've been consumed by setting the next key to the key initialized with
+      next_nonce: 1,
+      escaped_to: None,
+    };
 
-  (anvil, provider, router, (private_key, public_key))
+    // Confirm nonce 0 was used as such
+    {
+      let block = receipt.block_number.unwrap();
+      let executed = router.executed(block, block).await.unwrap();
+      assert_eq!(executed.len(), 1);
+      assert_eq!(executed[0], Executed::NextSeraiKeySet { nonce: 0, key: public_key.eth_repr() });
+    }
+
+    let res = Test { anvil, provider, chain_id, router, state };
+    res.verify_state().await;
+    res
+  }
+
+  async fn call_and_decode_err(&self, tx: TxLegacy) -> IRouterErrors {
+    let call = TransactionRequest::default()
+      .to(self.router.address())
+      .input(TransactionInput::new(tx.input));
+    let call_err = self.provider.call(&call).await.unwrap_err();
+    call_err.as_error_resp().unwrap().as_decoded_error::<IRouterErrors>(true).unwrap()
+  }
+
+  fn confirm_next_serai_key_tx(&self) -> TxLegacy {
+    let msg = Router::confirm_next_serai_key_message(self.chain_id, self.state.next_nonce);
+    let sig = sign(self.state.next_key.unwrap(), &msg);
+
+    self.router.confirm_next_serai_key(&sig)
+  }
+
+  async fn confirm_next_serai_key(&mut self) {
+    let mut tx = self.confirm_next_serai_key_tx();
+    tx.gas_price = 100_000_000_000;
+    let tx = ethereum_primitives::deterministically_sign(tx);
+    let receipt = ethereum_test_primitives::publish_tx(&self.provider, tx.clone()).await;
+    assert!(receipt.status());
+    // Only check the gas is equal when writing to a previously unallocated storage slot, as this
+    // is the highest possible gas cost and what the constant is derived from
+    if self.state.key.is_none() {
+      assert_eq!(
+        CalldataAgnosticGas::calculate(tx.tx(), receipt.gas_used),
+        Router::CONFIRM_NEXT_SERAI_KEY_GAS,
+      );
+    } else {
+      assert!(
+        CalldataAgnosticGas::calculate(tx.tx(), receipt.gas_used) <
+          Router::CONFIRM_NEXT_SERAI_KEY_GAS
+      );
+    }
+
+    {
+      let block = receipt.block_number.unwrap();
+      let executed = self.router.executed(block, block).await.unwrap();
+      assert_eq!(executed.len(), 1);
+      assert_eq!(
+        executed[0],
+        Executed::SeraiKeyUpdated {
+          nonce: self.state.next_nonce,
+          key: self.state.next_key.unwrap().1.eth_repr()
+        }
+      );
+    }
+
+    self.state.next_nonce += 1;
+    self.state.key = self.state.next_key;
+    self.state.next_key = None;
+    self.verify_state().await;
+  }
+
+  fn update_serai_key_tx(&self) -> ((Scalar, PublicKey), TxLegacy) {
+    let next_key = test_key();
+
+    let msg = Router::update_serai_key_message(self.chain_id, self.state.next_nonce, &next_key.1);
+    let sig = sign(self.state.key.unwrap(), &msg);
+
+    (next_key, self.router.update_serai_key(&next_key.1, &sig))
+  }
+
+  async fn update_serai_key(&mut self) {
+    let (next_key, mut tx) = self.update_serai_key_tx();
+    tx.gas_price = 100_000_000_000;
+    let tx = ethereum_primitives::deterministically_sign(tx);
+    let receipt = ethereum_test_primitives::publish_tx(&self.provider, tx.clone()).await;
+    assert!(receipt.status());
+    assert_eq!(
+      CalldataAgnosticGas::calculate(tx.tx(), receipt.gas_used),
+      Router::UPDATE_SERAI_KEY_GAS,
+    );
+
+    {
+      let block = receipt.block_number.unwrap();
+      let executed = self.router.executed(block, block).await.unwrap();
+      assert_eq!(executed.len(), 1);
+      assert_eq!(
+        executed[0],
+        Executed::NextSeraiKeySet { nonce: self.state.next_nonce, key: next_key.1.eth_repr() }
+      );
+    }
+
+    self.state.next_nonce += 1;
+    self.state.next_key = Some(next_key);
+    self.verify_state().await;
+  }
+
+  fn eth_in_instruction_tx(&self) -> (Coin, U256, Shorthand, TxLegacy) {
+    let coin = Coin::Ether;
+    let amount = U256::from(1);
+    let shorthand = Shorthand::Raw(RefundableInInstruction {
+      origin: None,
+      instruction: SeraiInInstruction::Transfer(SeraiAddress([0xff; 32])),
+    });
+
+    let mut tx = self.router.in_instruction(coin, amount, &shorthand);
+    tx.gas_limit = 1_000_000;
+    tx.gas_price = 100_000_000_000;
+
+    (coin, amount, shorthand, tx)
+  }
+
+  fn escape_hatch_tx(&self, escape_to: Address) -> TxLegacy {
+    let msg = Router::escape_hatch_message(self.chain_id, self.state.next_nonce, escape_to);
+    let sig = sign(self.state.key.unwrap(), &msg);
+    self.router.escape_hatch(escape_to, &sig)
+  }
+
+  async fn escape_hatch(&mut self) {
+    let mut escape_to = [0; 20];
+    OsRng.fill_bytes(&mut escape_to);
+    let escape_to = Address(escape_to.into());
+
+    // Set the code of the address to escape to so it isn't flagged as a non-contract
+    let () = self.provider.raw_request("anvil_setCode".into(), (escape_to, [0])).await.unwrap();
+
+    let mut tx = self.escape_hatch_tx(escape_to);
+    tx.gas_price = 100_000_000_000;
+    let tx = ethereum_primitives::deterministically_sign(tx);
+    let receipt = ethereum_test_primitives::publish_tx(&self.provider, tx.clone()).await;
+    assert!(receipt.status());
+    assert_eq!(CalldataAgnosticGas::calculate(tx.tx(), receipt.gas_used), Router::ESCAPE_HATCH_GAS);
+
+    {
+      let block = receipt.block_number.unwrap();
+      let executed = self.router.executed(block, block).await.unwrap();
+      assert_eq!(executed.len(), 1);
+      assert_eq!(executed[0], Executed::EscapeHatch { nonce: self.state.next_nonce, escape_to });
+    }
+
+    self.state.next_nonce += 1;
+    self.state.escaped_to = Some(escape_to);
+    self.verify_state().await;
+  }
+
+  fn escape_tx(&self, coin: Coin) -> TxLegacy {
+    let mut tx = self.router.escape(coin);
+    tx.gas_limit = 100_000;
+    tx.gas_price = 100_000_000_000;
+    tx
+  }
 }
 
 #[tokio::test]
 async fn test_constructor() {
-  let (_anvil, _provider, router, key) = setup_test().await;
+  // `Test::new` internalizes all checks on initial state
-  assert_eq!(router.next_key(BlockNumberOrTag::Latest.into()).await.unwrap(), Some(key.1));
+  Test::new().await;
-  assert_eq!(router.key(BlockNumberOrTag::Latest.into()).await.unwrap(), None);
-  assert_eq!(router.next_nonce(BlockNumberOrTag::Latest.into()).await.unwrap(), 1);
-  assert_eq!(
-    router.escaped_to(BlockNumberOrTag::Latest.into()).await.unwrap(),
-    Address::from([0; 20])
-  );
-}
-
-async fn confirm_next_serai_key(
-  provider: &Arc<RootProvider<SimpleRequest>>,
-  router: &Router,
-  nonce: u64,
-  key: (Scalar, PublicKey),
-) -> TransactionReceipt {
-  let msg = Router::confirm_next_serai_key_message(nonce);
-
-  let nonce = Scalar::random(&mut OsRng);
-  let c = Signature::challenge(ProjectivePoint::GENERATOR * nonce, &key.1, &msg);
-  let s = nonce + (c * key.0);
-
-  let sig = Signature::new(c, s).unwrap();
-
-  let mut tx = router.confirm_next_serai_key(&sig);
-  tx.gas_price = 100_000_000_000;
-  let tx = ethereum_primitives::deterministically_sign(tx);
-  let receipt = ethereum_test_primitives::publish_tx(provider, tx).await;
-  assert!(receipt.status());
-  assert_eq!(Router::CONFIRM_NEXT_SERAI_KEY_GAS, ((receipt.gas_used + 1000) / 1000) * 1000);
-  receipt
 }
 
 #[tokio::test]
 async fn test_confirm_next_serai_key() {
-  let (_anvil, provider, router, key) = setup_test().await;
+  let mut test = Test::new().await;
-
+  // TODO: Check all calls fail at this time, including inInstruction
-  assert_eq!(router.next_key(BlockNumberOrTag::Latest.into()).await.unwrap(), Some(key.1));
+  test.confirm_next_serai_key().await;
-  assert_eq!(router.key(BlockNumberOrTag::Latest.into()).await.unwrap(), None);
-  assert_eq!(router.next_nonce(BlockNumberOrTag::Latest.into()).await.unwrap(), 1);
-
-  let receipt = confirm_next_serai_key(&provider, &router, 1, key).await;
-
-  assert_eq!(router.next_key(receipt.block_hash.unwrap().into()).await.unwrap(), None);
-  assert_eq!(router.key(receipt.block_hash.unwrap().into()).await.unwrap(), Some(key.1));
-  assert_eq!(router.next_nonce(receipt.block_hash.unwrap().into()).await.unwrap(), 2);
 }
 
 #[tokio::test]
 async fn test_update_serai_key() {
-  let (_anvil, provider, router, key) = setup_test().await;
+  let mut test = Test::new().await;
-  confirm_next_serai_key(&provider, &router, 1, key).await;
+  test.confirm_next_serai_key().await;
+  test.update_serai_key().await;
 
-  let update_to = test_key().1;
+  // Once we update to a new key, we should, of course, be able to continue to rotate keys
-  let msg = Router::update_serai_key_message(2, &update_to);
+  test.confirm_next_serai_key().await;
+}
 
-  let nonce = Scalar::random(&mut OsRng);
+#[tokio::test]
-  let c = Signature::challenge(ProjectivePoint::GENERATOR * nonce, &key.1, &msg);
+async fn test_eth_in_instruction() {
-  let s = nonce + (c * key.0);
+  let mut test = Test::new().await;
+  test.confirm_next_serai_key().await;
 
-  let sig = Signature::new(c, s).unwrap();
+  let (coin, amount, shorthand, tx) = test.eth_in_instruction_tx();
+
+  // This should fail if the value mismatches the amount
+  {
+    let mut tx = tx.clone();
+    tx.value = U256::ZERO;
+    assert!(matches!(
+      test.call_and_decode_err(tx).await,
+      IRouterErrors::AmountMismatchesMsgValue(IRouter::AmountMismatchesMsgValue {})
+    ));
+  }
 
-  let mut tx = router.update_serai_key(&update_to, &sig);
-  tx.gas_price = 100_000_000_000;
   let tx = ethereum_primitives::deterministically_sign(tx);
-  let receipt = ethereum_test_primitives::publish_tx(&provider, tx).await;
+  let receipt = ethereum_test_primitives::publish_tx(&test.provider, tx.clone()).await;
   assert!(receipt.status());
-  assert_eq!(Router::UPDATE_SERAI_KEY_GAS, ((receipt.gas_used + 1000) / 1000) * 1000);
 
-  assert_eq!(router.key(receipt.block_hash.unwrap().into()).await.unwrap(), Some(key.1));
+  let block = receipt.block_number.unwrap();
-  assert_eq!(router.next_key(receipt.block_hash.unwrap().into()).await.unwrap(), Some(update_to));
+  let in_instructions =
-  assert_eq!(router.next_nonce(receipt.block_hash.unwrap().into()).await.unwrap(), 3);
+    test.router.in_instructions_unordered(block, block, &HashSet::new()).await.unwrap();
+  assert_eq!(in_instructions.len(), 1);
+  assert_eq!(
+    in_instructions[0],
+    InInstruction {
+      id: LogIndex {
+        block_hash: *receipt.block_hash.unwrap(),
+        index_within_block: receipt.inner.logs()[0].log_index.unwrap(),
+      },
+      transaction_hash: **tx.hash(),
+      from: tx.recover_signer().unwrap(),
+      coin,
+      amount,
+      data: shorthand.encode(),
+    }
+  );
+}
+
+#[tokio::test]
+async fn test_erc20_in_instruction() {
+  todo!("TODO")
+}
+
+#[tokio::test]
+async fn test_eth_address_out_instruction() {
+  todo!("TODO")
+}
+
+#[tokio::test]
+async fn test_erc20_address_out_instruction() {
+  todo!("TODO")
+}
+
+#[tokio::test]
+async fn test_eth_code_out_instruction() {
+  todo!("TODO")
+}
+
+#[tokio::test]
+async fn test_erc20_code_out_instruction() {
+  todo!("TODO")
+}
+
+#[tokio::test]
+async fn test_escape_hatch() {
+  let mut test = Test::new().await;
+  test.confirm_next_serai_key().await;
+
+  // Queue another key so the below test cases can run
+  test.update_serai_key().await;
+
+  {
+    // The zero address should be invalid to escape to
+    assert!(matches!(
+      test.call_and_decode_err(test.escape_hatch_tx([0; 20].into())).await,
+      IRouterErrors::InvalidEscapeAddress(IRouter::InvalidEscapeAddress {})
+    ));
+    // Empty addresses should be invalid to escape to
+    assert!(matches!(
+      test.call_and_decode_err(test.escape_hatch_tx([1; 20].into())).await,
+      IRouterErrors::EscapeAddressWasNotAContract(IRouter::EscapeAddressWasNotAContract {})
+    ));
+    // Non-empty addresses without code should be invalid to escape to
+    let tx = ethereum_primitives::deterministically_sign(TxLegacy {
+      to: Address([1; 20].into()).into(),
+      gas_limit: 21_000,
+      gas_price: 100_000_000_000u128,
+      value: U256::from(1),
+      ..Default::default()
+    });
+    let receipt = ethereum_test_primitives::publish_tx(&test.provider, tx.clone()).await;
+    assert!(receipt.status());
+    assert!(matches!(
+      test.call_and_decode_err(test.escape_hatch_tx([1; 20].into())).await,
+      IRouterErrors::EscapeAddressWasNotAContract(IRouter::EscapeAddressWasNotAContract {})
+    ));
+
+    // Escaping at this point in time should fail
+    assert!(matches!(
+      test.call_and_decode_err(test.router.escape(Coin::Ether)).await,
+      IRouterErrors::EscapeHatchNotInvoked(IRouter::EscapeHatchNotInvoked {})
+    ));
+  }
+
+  // Invoke the escape hatch
+  test.escape_hatch().await;
+
+  // Now that the escape hatch has been invoked, all of the following calls should fail
+  {
+    assert!(matches!(
+      test.call_and_decode_err(test.update_serai_key_tx().1).await,
+      IRouterErrors::EscapeHatchInvoked(IRouter::EscapeHatchInvoked {})
+    ));
+    assert!(matches!(
+      test.call_and_decode_err(test.confirm_next_serai_key_tx()).await,
+      IRouterErrors::EscapeHatchInvoked(IRouter::EscapeHatchInvoked {})
+    ));
+    assert!(matches!(
+      test.call_and_decode_err(test.eth_in_instruction_tx().3).await,
+      IRouterErrors::EscapeHatchInvoked(IRouter::EscapeHatchInvoked {})
+    ));
+    // TODO execute
+    // We reject further attempts to update the escape hatch to prevent the last key from being
+    // able to switch from the honest escape hatch to siphoning via a malicious escape hatch (such
+    // as after the validators represented unstake)
+    assert!(matches!(
+      test.call_and_decode_err(test.escape_hatch_tx(test.state.escaped_to.unwrap())).await,
+      IRouterErrors::EscapeHatchInvoked(IRouter::EscapeHatchInvoked {})
+    ));
+  }
+
+  // Check the escape fn itself
+
+  // ETH
+  {
+    let () = test
+      .provider
+      .raw_request("anvil_setBalance".into(), (test.router.address(), 1))
+      .await
+      .unwrap();
+    let tx = ethereum_primitives::deterministically_sign(test.escape_tx(Coin::Ether));
+    let receipt = ethereum_test_primitives::publish_tx(&test.provider, tx.clone()).await;
+    assert!(receipt.status());
+
+    let block = receipt.block_number.unwrap();
+    assert_eq!(
+      test.router.escapes(block, block).await.unwrap(),
+      vec![Escape { coin: Coin::Ether, amount: U256::from(1) }],
+    );
+
+    assert!(test.provider.get_balance(test.router.address()).await.unwrap() == U256::from(0));
+    assert!(
+      test.provider.get_balance(test.state.escaped_to.unwrap()).await.unwrap() == U256::from(1)
+    );
+  }
+
+  // TODO ERC20 escape
+}
+
+/*
+  event InInstruction(
+    address indexed from, address indexed coin, uint256 amount, bytes instruction
+  );
+  event Batch(uint256 indexed nonce, bytes32 indexed messageHash, bytes results);
+  error InvalidSeraiKey();
+  error InvalidSignature();
+  error AmountMismatchesMsgValue();
+  error TransferFromFailed();
+  error Reentered();
+  error EscapeFailed();
+  function executeArbitraryCode(bytes memory code) external payable;
+  struct Signature {
+    bytes32 c;
+    bytes32 s;
+  }
+  enum DestinationType {
+    Address,
+    Code
+  }
+  struct CodeDestination {
+    uint32 gasLimit;
+    bytes code;
+  }
+  struct OutInstruction {
+    DestinationType destinationType;
+    bytes destination;
+    uint256 amount;
+  }
+  function execute(
+    Signature calldata signature,
+    address coin,
+    uint256 fee,
+    OutInstruction[] calldata outs
+  ) external;
 }
 
 #[tokio::test]
@@ -190,7 +536,7 @@ async fn test_eth_in_instruction() {
     gas_limit: 1_000_000,
     to: TxKind::Call(router.address()),
     value: amount,
-    input: crate::abi::inInstructionCall::new((
+    input: crate::_irouter_abi::inInstructionCall::new((
       [0; 20].into(),
       amount,
       in_instruction.clone().into(),
@@ -217,7 +563,10 @@ async fn test_eth_in_instruction() {
   assert_eq!(parsed_in_instructions.len(), 1);
   assert_eq!(
     parsed_in_instructions[0].id,
-    (<[u8; 32]>::from(receipt.block_hash.unwrap()), receipt.inner.logs()[0].log_index.unwrap())
+    LogIndex {
+      block_hash: *receipt.block_hash.unwrap(),
+      index_within_block: receipt.inner.logs()[0].log_index.unwrap(),
+    },
   );
   assert_eq!(parsed_in_instructions[0].from, signer);
   assert_eq!(parsed_in_instructions[0].coin, Coin::Ether);
@@ -225,11 +574,6 @@ async fn test_eth_in_instruction() {
   assert_eq!(parsed_in_instructions[0].data, in_instruction);
 }
 
-#[tokio::test]
-async fn test_erc20_in_instruction() {
-  todo!("TODO")
-}
-
 async fn publish_outs(
   provider: &RootProvider<SimpleRequest>,
   router: &Router,
@@ -273,68 +617,4 @@ async fn test_eth_address_out_instruction() {
 
   assert_eq!(router.next_nonce(receipt.block_hash.unwrap().into()).await.unwrap(), 3);
 }
-
+*/
-#[tokio::test]
-async fn test_erc20_address_out_instruction() {
-  todo!("TODO")
-}
-
-#[tokio::test]
-async fn test_eth_code_out_instruction() {
-  todo!("TODO")
-}
-
-#[tokio::test]
-async fn test_erc20_code_out_instruction() {
-  todo!("TODO")
-}
-
-async fn escape_hatch(
-  provider: &Arc<RootProvider<SimpleRequest>>,
-  router: &Router,
-  nonce: u64,
-  key: (Scalar, PublicKey),
-  escape_to: Address,
-) -> TransactionReceipt {
-  let msg = Router::escape_hatch_message(nonce, escape_to);
-
-  let nonce = Scalar::random(&mut OsRng);
-  let c = Signature::challenge(ProjectivePoint::GENERATOR * nonce, &key.1, &msg);
-  let s = nonce + (c * key.0);
-
-  let sig = Signature::new(c, s).unwrap();
-
-  let mut tx = router.escape_hatch(escape_to, &sig);
-  tx.gas_price = 100_000_000_000;
-  let tx = ethereum_primitives::deterministically_sign(tx);
-  let receipt = ethereum_test_primitives::publish_tx(provider, tx).await;
-  assert!(receipt.status());
-  assert_eq!(Router::ESCAPE_HATCH_GAS, ((receipt.gas_used + 1000) / 1000) * 1000);
-  receipt
-}
-
-async fn escape(
-  provider: &Arc<RootProvider<SimpleRequest>>,
-  router: &Router,
-  coin: Coin,
-) -> TransactionReceipt {
-  let mut tx = router.escape(coin.address());
-  tx.gas_price = 100_000_000_000;
-  let tx = ethereum_primitives::deterministically_sign(tx);
-  let receipt = ethereum_test_primitives::publish_tx(provider, tx).await;
-  assert!(receipt.status());
-  receipt
-}
-
-#[tokio::test]
-async fn test_escape_hatch() {
-  let (_anvil, provider, router, key) = setup_test().await;
-  confirm_next_serai_key(&provider, &router, 1, key).await;
-  let escape_to: Address = {
-    let mut escape_to = [0; 20];
-    OsRng.fill_bytes(&mut escape_to);
-    escape_to.into()
-  };
-  escape_hatch(&provider, &router, 2, key, escape_to).await;
-  escape(&provider, &router, Coin::Ether).await;
-}

processor/ethereum/router/src/tests/read_write.rs

@@ -1,85 +0,0 @@
-use rand_core::{RngCore, OsRng};
-
-use alloy_core::primitives::U256;
-
-use crate::{Coin, InInstruction, Executed};
-
-fn coins() -> [Coin; 2] {
-  [Coin::Ether, {
-    let mut erc20 = [0; 20];
-    OsRng.fill_bytes(&mut erc20);
-    Coin::Erc20(erc20.into())
-  }]
-}
-
-#[test]
-fn test_coin_read_write() {
-  for coin in coins() {
-    let mut res = vec![];
-    coin.write(&mut res).unwrap();
-    assert_eq!(coin, Coin::read(&mut res.as_slice()).unwrap());
-  }
-}
-
-#[test]
-fn test_in_instruction_read_write() {
-  for coin in coins() {
-    let instruction = InInstruction {
-      id: (
-        {
-          let mut tx_id = [0; 32];
-          OsRng.fill_bytes(&mut tx_id);
-          tx_id
-        },
-        OsRng.next_u64(),
-      ),
-      from: {
-        let mut from = [0; 20];
-        OsRng.fill_bytes(&mut from);
-        from
-      },
-      coin,
-      amount: U256::from_le_bytes({
-        let mut amount = [0; 32];
-        OsRng.fill_bytes(&mut amount);
-        amount
-      }),
-      data: {
-        let len = usize::try_from(OsRng.next_u64() % 65536).unwrap();
-        let mut data = vec![0; len];
-        OsRng.fill_bytes(&mut data);
-        data
-      },
-    };
-
-    let mut buf = vec![];
-    instruction.write(&mut buf).unwrap();
-    assert_eq!(InInstruction::read(&mut buf.as_slice()).unwrap(), instruction);
-  }
-}
-
-#[test]
-fn test_executed_read_write() {
-  for executed in [
-    Executed::SetKey {
-      nonce: OsRng.next_u64(),
-      key: {
-        let mut key = [0; 32];
-        OsRng.fill_bytes(&mut key);
-        key
-      },
-    },
-    Executed::Batch {
-      nonce: OsRng.next_u64(),
-      message_hash: {
-        let mut message_hash = [0; 32];
-        OsRng.fill_bytes(&mut message_hash);
-        message_hash
-      },
-    },
-  ] {
-    let mut res = vec![];
-    executed.write(&mut res).unwrap();
-    assert_eq!(executed, Executed::read(&mut res.as_slice()).unwrap());
-  }
-}

processor/ethereum/src/main.rs

@@ -6,11 +6,13 @@
 static ALLOCATOR: zalloc::ZeroizingAlloc<std::alloc::System> =
   zalloc::ZeroizingAlloc(std::alloc::System);
 
+use core::time::Duration;
 use std::sync::Arc;
 
+use alloy_core::primitives::U256;
 use alloy_simple_request_transport::SimpleRequest;
 use alloy_rpc_client::ClientBuilder;
-use alloy_provider::RootProvider;
+use alloy_provider::{Provider, RootProvider};
 
 use serai_client::validator_sets::primitives::Session;
 
@@ -62,10 +64,26 @@ async fn main() {
     ClientBuilder::default().transport(SimpleRequest::new(bin::url()), true),
   ));
 
+  let chain_id = {
+    let mut delay = Duration::from_secs(5);
+    loop {
+      match provider.get_chain_id().await {
+        Ok(chain_id) => break chain_id,
+        Err(e) => {
+          log::error!("failed to fetch the chain ID on boot: {e:?}");
+          tokio::time::sleep(delay).await;
+          delay = (delay + Duration::from_secs(5)).max(Duration::from_secs(120));
+        }
+      }
+    }
+  };
+
   bin::main_loop::<SetInitialKey, _, KeyGenParams, _>(
     db.clone(),
     Rpc { db: db.clone(), provider: provider.clone() },
-    Scheduler::<bin::Db>::new(SmartContract),
+    Scheduler::<bin::Db>::new(SmartContract {
+      chain_id: U256::from_le_slice(&chain_id.to_le_bytes()),
+    }),
     TransactionPublisher::new(db, provider, {
       let relayer_hostname = env::var("ETHEREUM_RELAYER_HOSTNAME")
         .expect("ethereum relayer hostname wasn't specified")

processor/ethereum/src/primitives/block.rs

@@ -32,6 +32,7 @@ impl primitives::BlockHeader for Epoch {
 #[derive(Clone, PartialEq, Eq, Debug)]
 pub(crate) struct FullEpoch {
   pub(crate) epoch: Epoch,
+  /// The unordered list of `InInstruction`s within this epoch
   pub(crate) instructions: Vec<EthereumInInstruction>,
   pub(crate) executed: Vec<Executed>,
 }
@@ -99,6 +100,7 @@ impl primitives::Block for FullEpoch {
       let Some(expected) =
         eventualities.active_eventualities.remove(executed.nonce().to_le_bytes().as_slice())
       else {
+        // TODO: Why is this a continue, not an assert?
         continue;
       };
       assert_eq!(

processor/ethereum/src/primitives/output.rs

@@ -81,8 +81,8 @@ impl ReceivedOutput<<Secp256k1 as Ciphersuite>::G, Address> for Output {
     match self {
       Output::Output { key: _, instruction } => {
         let mut id = [0; 40];
-        id[.. 32].copy_from_slice(&instruction.id.0);
+        id[.. 32].copy_from_slice(&instruction.id.block_hash);
-        id[32 ..].copy_from_slice(&instruction.id.1.to_le_bytes());
+        id[32 ..].copy_from_slice(&instruction.id.index_within_block.to_le_bytes());
         OutputId(id)
       }
       // Yet upon Eventuality completions, we report a Change output to ensure synchrony per the
@@ -97,7 +97,7 @@ impl ReceivedOutput<<Secp256k1 as Ciphersuite>::G, Address> for Output {
 
   fn transaction_id(&self) -> Self::TransactionId {
     match self {
-      Output::Output { key: _, instruction } => instruction.id.0,
+      Output::Output { key: _, instruction } => instruction.transaction_hash,
       Output::Eventuality { key: _, nonce } => {
         let mut id = [0; 32];
         id[.. 8].copy_from_slice(&nonce.to_le_bytes());
@@ -114,7 +114,7 @@ impl ReceivedOutput<<Secp256k1 as Ciphersuite>::G, Address> for Output {
 
   fn presumed_origin(&self) -> Option<Address> {
     match self {
-      Output::Output { key: _, instruction } => Some(Address::from(instruction.from)),
+      Output::Output { key: _, instruction } => Some(Address::Address(*instruction.from.0)),
       Output::Eventuality { .. } => None,
     }
   }
@@ -145,7 +145,7 @@ impl ReceivedOutput<<Secp256k1 as Ciphersuite>::G, Address> for Output {
       Output::Output { key, instruction } => {
         writer.write_all(&[0])?;
         writer.write_all(key.to_bytes().as_ref())?;
-        instruction.write(writer)
+        instruction.serialize(writer)
       }
       Output::Eventuality { key, nonce } => {
         writer.write_all(&[1])?;
@@ -164,7 +164,7 @@ impl ReceivedOutput<<Secp256k1 as Ciphersuite>::G, Address> for Output {
     Ok(match kind[0] {
       0 => {
         let key = Secp256k1::read_G(reader)?;
-        let instruction = EthereumInInstruction::read(reader)?;
+        let instruction = EthereumInInstruction::deserialize_reader(reader)?;
         Self::Output { key, instruction }
       }
       1 => {

processor/ethereum/src/primitives/transaction.rs

@@ -17,8 +17,8 @@ use crate::{output::OutputId, machine::ClonableTransctionMachine};
 
 #[derive(Clone, PartialEq, Debug)]
 pub(crate) enum Action {
-  SetKey { nonce: u64, key: PublicKey },
+  SetKey { chain_id: U256, nonce: u64, key: PublicKey },
-  Batch { nonce: u64, coin: Coin, fee: U256, outs: Vec<(Address, U256)> },
+  Batch { chain_id: U256, nonce: u64, coin: Coin, fee: U256, outs: Vec<(Address, U256)> },
 }
 
 #[derive(Clone, PartialEq, Eq, Debug)]
@@ -33,17 +33,25 @@ impl Action {
 
   pub(crate) fn message(&self) -> Vec<u8> {
     match self {
-      Action::SetKey { nonce, key } => Router::update_serai_key_message(*nonce, key),
+      Action::SetKey { chain_id, nonce, key } => {
-      Action::Batch { nonce, coin, fee, outs } => {
+        Router::update_serai_key_message(*chain_id, *nonce, key)
-        Router::execute_message(*nonce, *coin, *fee, OutInstructions::from(outs.as_ref()))
       }
+      Action::Batch { chain_id, nonce, coin, fee, outs } => Router::execute_message(
+        *chain_id,
+        *nonce,
+        *coin,
+        *fee,
+        OutInstructions::from(outs.as_ref()),
+      ),
     }
   }
 
   pub(crate) fn eventuality(&self) -> Eventuality {
     Eventuality(match self {
-      Self::SetKey { nonce, key } => Executed::SetKey { nonce: *nonce, key: key.eth_repr() },
+      Self::SetKey { chain_id: _, nonce, key } => {
-      Self::Batch { nonce, .. } => {
+        Executed::NextSeraiKeySet { nonce: *nonce, key: key.eth_repr() }
+      }
+      Self::Batch { chain_id: _, nonce, .. } => {
         Executed::Batch { nonce: *nonce, message_hash: keccak256(self.message()) }
       }
     })
@@ -77,6 +85,10 @@ impl SignableTransaction for Action {
       Err(io::Error::other("unrecognized Action type"))?;
     }
 
+    let mut chain_id = [0; 32];
+    reader.read_exact(&mut chain_id)?;
+    let chain_id = U256::from_be_bytes(chain_id);
+
     let mut nonce = [0; 8];
     reader.read_exact(&mut nonce)?;
     let nonce = u64::from_le_bytes(nonce);
@@ -88,10 +100,10 @@ impl SignableTransaction for Action {
         let key =
           PublicKey::from_eth_repr(key).ok_or_else(|| io::Error::other("invalid key in Action"))?;
 
-        Action::SetKey { nonce, key }
+        Action::SetKey { chain_id, nonce, key }
       }
       1 => {
-        let coin = Coin::read(reader)?;
+        let coin = borsh::from_reader(reader)?;
 
         let mut fee = [0; 32];
         reader.read_exact(&mut fee)?;
@@ -111,22 +123,24 @@ impl SignableTransaction for Action {
 
           outs.push((address, amount));
         }
-        Action::Batch { nonce, coin, fee, outs }
+        Action::Batch { chain_id, nonce, coin, fee, outs }
       }
       _ => unreachable!(),
     })
   }
   fn write(&self, writer: &mut impl io::Write) -> io::Result<()> {
     match self {
-      Self::SetKey { nonce, key } => {
+      Self::SetKey { chain_id, nonce, key } => {
         writer.write_all(&[0])?;
+        writer.write_all(&chain_id.to_be_bytes::<32>())?;
         writer.write_all(&nonce.to_le_bytes())?;
         writer.write_all(&key.eth_repr())
       }
-      Self::Batch { nonce, coin, fee, outs } => {
+      Self::Batch { chain_id, nonce, coin, fee, outs } => {
         writer.write_all(&[1])?;
+        writer.write_all(&chain_id.to_be_bytes::<32>())?;
         writer.write_all(&nonce.to_le_bytes())?;
-        coin.write(writer)?;
+        borsh::BorshSerialize::serialize(coin, writer)?;
         writer.write_all(&fee.as_le_bytes())?;
         writer.write_all(&u32::try_from(outs.len()).unwrap().to_le_bytes())?;
         for (address, amount) in outs {
@@ -167,9 +181,9 @@ impl primitives::Eventuality for Eventuality {
   }
 
   fn read(reader: &mut impl io::Read) -> io::Result<Self> {
-    Executed::read(reader).map(Self)
+    Ok(Self(borsh::from_reader(reader)?))
   }
   fn write(&self, writer: &mut impl io::Write) -> io::Result<()> {
-    self.0.write(writer)
+    borsh::BorshSerialize::serialize(&self.0, writer)
   }
 }

processor/ethereum/src/publisher.rs

@@ -88,8 +88,8 @@ impl<D: Db> signers::TransactionPublisher<Transaction> for TransactionPublisher<
       let nonce = tx.0.nonce();
       // Convert from an Action (an internal representation of a signable event) to a TxLegacy
       let tx = match tx.0 {
-        Action::SetKey { nonce: _, key } => router.update_serai_key(&key, &tx.1),
+        Action::SetKey { chain_id: _, nonce: _, key } => router.update_serai_key(&key, &tx.1),
-        Action::Batch { nonce: _, coin, fee, outs } => {
+        Action::Batch { chain_id: _, nonce: _, coin, fee, outs } => {
           router.execute(coin, fee, OutInstructions::from(outs.as_ref()), &tx.1)
         }
       };

processor/ethereum/src/rpc.rs

@@ -162,15 +162,19 @@ impl<D: Db> ScannerFeed for Rpc<D> {
         router: Router,
         block: Header,
       ) -> Result<(Vec<EthereumInInstruction>, Vec<Executed>), RpcError<TransportErrorKind>> {
-        let mut instructions = router.in_instructions(block.number, &HashSet::from(TOKENS)).await?;
+        let mut instructions = router
+          .in_instructions_unordered(block.number, block.number, &HashSet::from(TOKENS))
+          .await?;
 
         for token in TOKENS {
-          for TopLevelTransfer { id, from, amount, data } in Erc20::new(provider.clone(), **token)
+          for TopLevelTransfer { id, transaction_hash, from, amount, data } in
-            .top_level_transfers(block.number, router.address())
+            Erc20::new(provider.clone(), token)
+              .top_level_transfers_unordered(block.number, block.number, router.address())
               .await?
           {
             instructions.push(EthereumInInstruction {
               id,
+              transaction_hash,
               from,
               coin: EthereumCoin::Erc20(token),
               amount,
@@ -179,7 +183,7 @@ impl<D: Db> ScannerFeed for Rpc<D> {
           }
         }
 
-        let executed = router.executed(block.number).await?;
+        let executed = router.executed(block.number, block.number).await?;
 
         Ok((instructions, executed))
       }

processor/ethereum/src/scheduler.rs

@@ -36,7 +36,9 @@ fn balance_to_ethereum_amount(balance: Balance) -> U256 {
 }
 
 #[derive(Clone)]
-pub(crate) struct SmartContract;
+pub(crate) struct SmartContract {
+  pub(crate) chain_id: U256,
+}
 impl<D: Db> smart_contract_scheduler::SmartContract<Rpc<D>> for SmartContract {
   type SignableTransaction = Action;
 
@@ -46,8 +48,11 @@ impl<D: Db> smart_contract_scheduler::SmartContract<Rpc<D>> for SmartContract {
     _retiring_key: KeyFor<Rpc<D>>,
     new_key: KeyFor<Rpc<D>>,
   ) -> (Self::SignableTransaction, EventualityFor<Rpc<D>>) {
-    let action =
+    let action = Action::SetKey {
-      Action::SetKey { nonce, key: PublicKey::new(new_key).expect("rotating to an invald key") };
+      chain_id: self.chain_id,
+      nonce,
+      key: PublicKey::new(new_key).expect("rotating to an invald key"),
+    };
     (action.clone(), action.eventuality())
   }
 
@@ -133,6 +138,7 @@ impl<D: Db> smart_contract_scheduler::SmartContract<Rpc<D>> for SmartContract {
         }
 
         res.push(Action::Batch {
+          chain_id: self.chain_id,
           nonce,
           coin: coin_to_ethereum_coin(coin),
           fee: U256::try_from(total_gas).unwrap() * fee_per_gas,