Have the Ethereum scheduler create Batches as necessary

Also introduces the fee logic, despite it being stubbed.

processor/ethereum/src/primitives/transaction.rs

@@ -18,7 +18,7 @@ use crate::{output::OutputId, machine::ClonableTransctionMachine};
 #[derive(Clone, PartialEq, Debug)]
 pub(crate) enum Action {
   SetKey { chain_id: U256, nonce: u64, key: PublicKey },
-  Batch { chain_id: U256, nonce: u64, outs: Vec<(Address, (Coin, U256))> },
+  Batch { chain_id: U256, nonce: u64, coin: Coin, fee_per_gas: U256, outs: Vec<(Address, U256)> },
 }
 
 #[derive(Clone, PartialEq, Eq, Debug)]
@@ -36,9 +36,13 @@ impl Action {
       Action::SetKey { chain_id, nonce, key } => {
         Router::update_serai_key_message(*chain_id, *nonce, key)
       }
-      Action::Batch { chain_id, nonce, outs } => {
-        Router::execute_message(*chain_id, *nonce, OutInstructions::from(outs.as_ref()))
-      }
+      Action::Batch { chain_id, nonce, coin, fee_per_gas, outs } => Router::execute_message(
+        *chain_id,
+        *nonce,
+        *coin,
+        *fee_per_gas,
+        OutInstructions::from(outs.as_ref()),
+      ),
     }
   }
 
@@ -47,13 +51,9 @@ impl Action {
       Self::SetKey { chain_id: _, nonce, key } => {
         Executed::SetKey { nonce: *nonce, key: key.eth_repr() }
       }
-      Self::Batch { chain_id, nonce, outs } => Executed::Batch {
+      Self::Batch { nonce, .. } => Executed::Batch {
         nonce: *nonce,
-        message_hash: keccak256(Router::execute_message(
-          *chain_id,
-          *nonce,
-          OutInstructions::from(outs.as_ref()),
-        )),
+        message_hash: keccak256(self.message()),
       },
     })
   }
@@ -104,6 +104,12 @@ impl SignableTransaction for Action {
         Action::SetKey { chain_id, nonce, key }
       }
       1 => {
+        let coin = Coin::read(reader)?;
+
+        let mut fee_per_gas = [0; 32];
+        reader.read_exact(&mut fee_per_gas)?;
+        let fee_per_gas = U256::from_le_bytes(fee_per_gas);
+
         let mut outs_len = [0; 4];
         reader.read_exact(&mut outs_len)?;
         let outs_len = usize::try_from(u32::from_le_bytes(outs_len)).unwrap();
@@ -111,15 +117,14 @@ impl SignableTransaction for Action {
         let mut outs = vec![];
         for _ in 0 .. outs_len {
           let address = borsh::from_reader(reader)?;
-          let coin = Coin::read(reader)?;
 
           let mut amount = [0; 32];
           reader.read_exact(&mut amount)?;
           let amount = U256::from_le_bytes(amount);
 
-          outs.push((address, (coin, amount)));
+          outs.push((address, amount));
         }
-        Action::Batch { chain_id, nonce, outs }
+        Action::Batch { chain_id, nonce, coin, fee_per_gas, outs }
       }
       _ => unreachable!(),
     })
@@ -132,14 +137,15 @@ impl SignableTransaction for Action {
         writer.write_all(&nonce.to_le_bytes())?;
         writer.write_all(&key.eth_repr())
       }
-      Self::Batch { chain_id, nonce, outs } => {
+      Self::Batch { chain_id, nonce, coin, fee_per_gas, outs } => {
         writer.write_all(&[1])?;
         writer.write_all(&chain_id.as_le_bytes())?;
         writer.write_all(&nonce.to_le_bytes())?;
+        coin.write(writer)?;
+        writer.write_all(&fee_per_gas.as_le_bytes())?;
         writer.write_all(&u32::try_from(outs.len()).unwrap().to_le_bytes())?;
-        for (address, (coin, amount)) in outs {
+        for (address, amount) in outs {
           borsh::BorshSerialize::serialize(address, writer)?;
-          coin.write(writer)?;
           writer.write_all(&amount.as_le_bytes())?;
         }
         Ok(())

processor/ethereum/src/publisher.rs

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

processor/ethereum/src/scheduler.rs

@@ -1,6 +1,11 @@
+use std::collections::HashMap;
+
 use alloy_core::primitives::U256;
 
-use serai_client::primitives::{NetworkId, Coin, Balance};
+use serai_client::{
+  primitives::{NetworkId, Coin, Balance},
+  networks::ethereum::Address,
+};
 
 use serai_db::Db;
 
@@ -53,27 +58,86 @@ impl<D: Db> smart_contract_scheduler::SmartContract<Rpc<D>> for SmartContract {
 
   fn fulfill(
     &self,
-    nonce: u64,
+    mut nonce: u64,
     _key: KeyFor<Rpc<D>>,
     payments: Vec<Payment<AddressFor<Rpc<D>>>>,
   ) -> Vec<(Self::SignableTransaction, EventualityFor<Rpc<D>>)> {
-    let mut outs = Vec::with_capacity(payments.len());
+    // Sort by coin
+    let mut outs = HashMap::<_, _>::new();
     for payment in payments {
-      outs.push((
-        payment.address().clone(),
-        (
-          coin_to_ethereum_coin(payment.balance().coin),
-          balance_to_ethereum_amount(payment.balance()),
-        ),
-      ));
+      let coin = payment.balance().coin;
+      outs
+        .entry(coin)
+        .or_insert_with(|| Vec::with_capacity(1))
+        .push((payment.address().clone(), balance_to_ethereum_amount(payment.balance())));
     }
 
-    // TODO: Per-batch gas limit
-    // TODO: Create several batches
-    // TODO: Handle fees
-    let action = Action::Batch { chain_id: self.chain_id, nonce, outs };
+    let mut res = vec![];
+    for coin in [Coin::Ether, Coin::Dai] {
+      let Some(outs) = outs.remove(&coin) else { continue };
+      assert!(!outs.is_empty());
 
-    vec![(action.clone(), action.eventuality())]
+      let fee_per_gas: U256 = todo!("TODO");
+
+      // The gas required to perform any interaction with the Router.
+      const BASE_GAS: u32 = 0; // TODO
+
+      // The gas required to handle an additional payment to an address, in the worst case.
+      const ADDRESS_PAYMENT_GAS: u32 = 0; // TODO
+
+      // The gas required to handle an additional payment to an smart contract, in the worst case.
+      // This does not include the explicit gas budget defined within the address specification.
+      const CONTRACT_PAYMENT_GAS: u32 = 0; // TODO
+
+      // The maximum amount of gas for a batch.
+      const BATCH_GAS_LIMIT: u32 = 10_000_000;
+
+      // Split these outs into batches, respecting BATCH_GAS_LIMIT
+      let mut batches = vec![vec![]];
+      let mut current_gas = BASE_GAS;
+      for out in outs {
+        let payment_gas = match out.0 {
+          Address::Address(_) => ADDRESS_PAYMENT_GAS,
+          Address::Contract(deployment) => CONTRACT_PAYMENT_GAS + deployment.gas_limit(),
+        };
+        if (current_gas + payment_gas) > BATCH_GAS_LIMIT {
+          assert!(!batches.last().unwrap().is_empty());
+          batches.push(vec![]);
+          current_gas = BASE_GAS;
+        }
+        batches.last_mut().unwrap().push(out);
+        current_gas += payment_gas;
+      }
+
+      // Push each batch onto the result
+      for outs in batches {
+        let base_gas = BASE_GAS.div_ceil(u32::try_from(outs.len()).unwrap());
+        // Deduce the fee from each out
+        for out in &mut outs {
+          let payment_gas = base_gas +
+            match out.0 {
+              Address::Address(_) => ADDRESS_PAYMENT_GAS,
+              Address::Contract(deployment) => CONTRACT_PAYMENT_GAS + deployment.gas_limit(),
+            };
+
+          let payment_gas_cost = fee_per_gas * U256::try_from(payment_gas).unwrap();
+          out.1 -= payment_gas_cost;
+        }
+
+        res.push(Action::Batch {
+          chain_id: self.chain_id,
+          nonce,
+          coin: coin_to_ethereum_coin(coin),
+          fee_per_gas,
+          outs,
+        });
+        nonce += 1;
+      }
+    }
+    // Ensure we handled all payments we're supposed to
+    assert!(outs.is_empty());
+
+    res.into_iter().map(|action| (action.clone(), action.eventuality())).collect()
   }
 }
 

substrate/client/src/networks/ethereum.rs

@@ -5,13 +5,18 @@ use borsh::{BorshSerialize, BorshDeserialize};
 
 use crate::primitives::{MAX_ADDRESS_LEN, ExternalAddress};
 
+/// THe maximum amount of gas an address is allowed to specify as its gas limit.
+///
+/// Payments to an address with a gas limit which exceed this value will be dropped entirely.
+pub const ADDRESS_GAS_LIMIT: u32 = 950_000;
+
 #[derive(Clone, PartialEq, Eq, Debug, BorshSerialize, BorshDeserialize)]
 pub struct ContractDeployment {
   /// The gas limit to use for this contract's execution.
   ///
   /// THis MUST be less than the Serai gas limit. The cost of it will be deducted from the amount
   /// transferred.
-  gas: u32,
+  gas_limit: u32,
   /// The initialization code of the contract to deploy.
   ///
   /// This contract will be deployed (executing the initialization code). No further calls will
@@ -21,17 +26,23 @@ pub struct ContractDeployment {
 
 /// A contract to deploy, enabling executing arbitrary code.
 impl ContractDeployment {
-  pub fn new(gas: u32, code: Vec<u8>) -> Option<Self> {
+  pub fn new(gas_limit: u32, code: Vec<u8>) -> Option<Self> {
+    // Check the gas limit is less the address gas limit
+    if gas_limit > ADDRESS_GAS_LIMIT {
+      None?;
+    }
+
     // The max address length, minus the type byte, minus the size of the gas
     const MAX_CODE_LEN: usize = (MAX_ADDRESS_LEN as usize) - (1 + core::mem::size_of::<u32>());
     if code.len() > MAX_CODE_LEN {
       None?;
     }
-    Some(Self { gas, code })
+
+    Some(Self { gas_limit, code })
   }
 
-  pub fn gas(&self) -> u32 {
-    self.gas
+  pub fn gas_limit(&self) -> u32 {
+    self.gas_limit
   }
   pub fn code(&self) -> &[u8] {
     &self.code
@@ -66,12 +77,18 @@ impl TryFrom<ExternalAddress> for Address {
         Address::Address(address)
       }
       1 => {
-        let mut gas = [0xff; 4];
-        reader.read_exact(&mut gas).map_err(|_| ())?;
-        // The code is whatever's left since the ExternalAddress is a delimited container of
-        // appropriately bounded length
+        let mut gas_limit = [0xff; 4];
+        reader.read_exact(&mut gas_limit).map_err(|_| ())?;
         Address::Contract(ContractDeployment {
-          gas: u32::from_le_bytes(gas),
+          gas_limit: {
+            let gas_limit = u32::from_le_bytes(gas_limit);
+            if gas_limit > ADDRESS_GAS_LIMIT {
+              Err(())?;
+            }
+            gas_limit
+          },
+          // The code is whatever's left since the ExternalAddress is a delimited container of
+          // appropriately bounded length
           code: reader.to_vec(),
         })
       }
@@ -87,9 +104,9 @@ impl From<Address> for ExternalAddress {
         res.push(0);
         res.extend(&address);
       }
-      Address::Contract(ContractDeployment { gas, code }) => {
+      Address::Contract(ContractDeployment { gas_limit, code }) => {
         res.push(1);
-        res.extend(&gas.to_le_bytes());
+        res.extend(&gas_limit.to_le_bytes());
         res.extend(&code);
       }
     }