Ethereum Settlement Engine improvements

crates/interledger-api/src/routes/accounts.rs

@@ -12,7 +12,7 @@ use reqwest::r#async::Client;
 use serde::Serialize;
 use serde_json::{json, Value};
 use std::str::FromStr;
-use tokio_retry::{strategy::ExponentialBackoff, Retry};
+use tokio_retry::{strategy::FixedInterval, Retry};
 use tower_web::{impl_web, Response};
 use url::Url;
 
@@ -108,7 +108,7 @@ impl_web! {
                                     }
                                 })
                             };
-                            Retry::spawn(ExponentialBackoff::from_millis(10).take(MAX_RETRIES), action)
+                            Retry::spawn(FixedInterval::from_millis(2000).take(MAX_RETRIES), action)
                             .map_err(|_| Response::builder().status(500).body(()).unwrap())
                             .and_then(move |_| {
                                 Ok(json!(account))

crates/interledger-settlement-engines/src/engines/ethereum_ledger/test_helpers.rs

@@ -59,7 +59,6 @@ pub struct TestStore {
     #[allow(clippy::all)]
     pub cache: Arc<RwLock<HashMap<String, (StatusCode, String, [u8; 32])>>>,
     pub last_observed_block: Arc<RwLock<U256>>,
-    pub last_observed_balance: Arc<RwLock<U256>>,
     pub saved_hashes: Arc<RwLock<HashMap<H256, bool>>>,
     pub cache_hits: Arc<RwLock<u64>>,
 }
@@ -100,25 +99,17 @@ impl EthereumStore for TestStore {
         Box::new(ok(v))
     }
 
-    fn save_recently_observed_data(
+    fn save_recently_observed_block(
         &self,
         block: U256,
-        balance: U256,
     ) -> Box<dyn Future<Item = (), Error = ()> + Send> {
         let mut guard = self.last_observed_block.write();
         *guard = block;
-        let mut guard = self.last_observed_balance.write();
-        *guard = balance;
         Box::new(ok(()))
     }
 
-    fn load_recently_observed_data(
-        &self,
-    ) -> Box<dyn Future<Item = (U256, U256), Error = ()> + Send> {
-        Box::new(ok((
-            *self.last_observed_block.read(),
-            *self.last_observed_balance.read(),
-        )))
+    fn load_recently_observed_block(&self) -> Box<dyn Future<Item = U256, Error = ()> + Send> {
+        Box::new(ok(*self.last_observed_block.read()))
     }
 
     fn load_account_id_from_address(
@@ -135,16 +126,24 @@ impl EthereumStore for TestStore {
         Box::new(ok(d))
     }
 
-    fn check_tx_credited(&self, tx_hash: H256) -> Box<dyn Future<Item = bool, Error = ()> + Send> {
-        let mut hashes = self.saved_hashes.write();
-        // if hash exists error, else store it
+    fn check_if_tx_processed(
+        &self,
+        tx_hash: H256,
+    ) -> Box<dyn Future<Item = bool, Error = ()> + Send> {
+        let hashes = self.saved_hashes.read();
+        // if hash exists then return error
         if hashes.get(&tx_hash).is_some() {
             Box::new(ok(true))
         } else {
-            (*hashes).insert(tx_hash, true);
             Box::new(ok(false))
         }
     }
+
+    fn mark_tx_processed(&self, tx_hash: H256) -> Box<dyn Future<Item = (), Error = ()> + Send> {
+        let mut hashes = self.saved_hashes.write();
+        (*hashes).insert(tx_hash, true);
+        Box::new(ok(()))
+    }
 }
 
 impl AccountStore for TestStore {
@@ -235,7 +234,6 @@ impl TestStore {
             address_to_id: Arc::new(RwLock::new(address_to_id)),
             cache: Arc::new(RwLock::new(HashMap::new())),
             cache_hits: Arc::new(RwLock::new(0)),
-            last_observed_balance: Arc::new(RwLock::new(U256::from(0))),
             last_observed_block: Arc::new(RwLock::new(U256::from(0))),
             saved_hashes: Arc::new(RwLock::new(HashMap::new())),
         }

crates/interledger-settlement-engines/src/engines/ethereum_ledger/types.rs

@@ -46,18 +46,15 @@ pub trait EthereumStore {
         account_ids: Vec<<Self::Account as Account>::AccountId>,
     ) -> Box<dyn Future<Item = Vec<Addresses>, Error = ()> + Send>;
 
-    /// Saves the latest block number and account balance, up to which all
+    /// Saves the latest block number, up to which all
     /// transactions have been communicated to the connector
-    fn save_recently_observed_data(
+    fn save_recently_observed_block(
         &self,
         block: U256,
-        balance: U256,
     ) -> Box<dyn Future<Item = (), Error = ()> + Send>;
 
-    /// Loads the latest saved block number and account balance
-    fn load_recently_observed_data(
-        &self,
-    ) -> Box<dyn Future<Item = (U256, U256), Error = ()> + Send>;
+    /// Loads the latest saved block number
+    fn load_recently_observed_block(&self) -> Box<dyn Future<Item = U256, Error = ()> + Send>;
 
     /// Retrieves the account id associated with the provided addresses pair.
     /// Note that an account with the same `own_address` but different ERC20
@@ -68,9 +65,15 @@ pub trait EthereumStore {
         eth_address: Addresses,
     ) -> Box<dyn Future<Item = <Self::Account as Account>::AccountId, Error = ()> + Send>;
 
-    /// Errors out if the transaction hash has already been stored before,
-    /// otherwise saves the transaction hash in the store.
-    fn check_tx_credited(&self, tx_hash: H256) -> Box<dyn Future<Item = bool, Error = ()> + Send>;
+    /// Returns true if the transaction has already been processed and saved in
+    /// the store.
+    fn check_if_tx_processed(
+        &self,
+        tx_hash: H256,
+    ) -> Box<dyn Future<Item = bool, Error = ()> + Send>;
+
+    /// Saves the transaction hash in the store.
+    fn mark_tx_processed(&self, tx_hash: H256) -> Box<dyn Future<Item = (), Error = ()> + Send>;
 }
 
 /// Implement this trait for datatypes which can be used to sign an Ethereum

crates/interledger-settlement-engines/src/engines/ethereum_ledger/utils.rs

@@ -1,23 +1,28 @@
 use ethabi::Token;
 use ethereum_tx_sign::{
-    web3::types::{Address, Transaction, U256},
+    web3::{
+        api::Web3,
+        futures::future::Future,
+        transports::Http,
+        types::{Address, BlockNumber, FilterBuilder, Transaction, H160, H256, U256},
+    },
     RawTransaction,
 };
+use log::error;
+use std::str::FromStr;
+
+/// This is the result of keccak256("Transfer(address,address,to)"), which is
+/// used to filter through Ethereum ERC20 Transfer events in transaction receipts.
+const TRANSFER_EVENT_FILTER: &str =
+    "ddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef";
 
 // Helper function which is used to construct an Ethereum transaction sending
-// `value` tokens to `to`. The account's nonce is required since Ethereum uses
-// an account based model with nonces for replay protection. If a
-// `token_address` is provided, then an ERC20 transaction is created instead for
-// that token. Ethereum transactions cost 21000 Gas, while ERC20 transactions
-// cost at most 70k (can tighten the gas limit, but 70k is safe if the address
-// is indeed an ERC20 token.
-// TODO: pass it gas_price as a parameter which is calculated from `web3.eth().gas_price()`
-pub fn make_tx(
-    to: Address,
-    value: U256,
-    nonce: U256,
-    token_address: Option<Address>,
-) -> RawTransaction {
+// `value` tokens to `to`. If a `token_address` is provided, then an ERC20
+// transaction  is created instead for that token. The `nonce`, `gas` and
+// `gas_price` fields are set to 0 and are expected to be set with the values
+// returned by the corresponding `eth_getTransactionCount`, `eth_estimateGas`,
+// `eth_gasPrice` calls to an Ethereum node.
+pub fn make_tx(to: Address, value: U256, token_address: Option<Address>) -> RawTransaction {
     if let Some(token_address) = token_address {
         // Ethereum contract transactions format:
         // [transfer function selector][`to` padded ][`value` padded]
@@ -29,26 +34,97 @@ pub fn make_tx(
         data.extend(ethabi::encode(&[Token::Address(to), Token::Uint(value)]));
         RawTransaction {
             to: Some(token_address),
-            nonce,
+            nonce: U256::from(0),
             data,
-            gas: 70000.into(), // ERC20 transactions cost approximately 40k gas.
-            gas_price: 20000.into(),
+            gas: U256::from(0),
+            gas_price: U256::from(0),
             value: U256::zero(),
         }
     } else {
         // Ethereum account transaction:
         // The receiver is `to`, and the data field is left empty.
         RawTransaction {
             to: Some(to),
-            nonce,
+            nonce: U256::from(0),
             data: vec![],
-            gas: 21000.into(),
-            gas_price: 20000.into(),
+            gas: U256::from(0),
+            gas_price: U256::from(0),
             value,
         }
     }
 }
 
+#[derive(Clone, Copy, Debug)]
+pub struct ERC20Transfer {
+    pub tx_hash: H256,
+    pub from: Address,
+    pub to: Address,
+    pub amount: U256,
+}
+
+/// Filters out transactions where the `from` and `to` fields match the provides
+/// addreses.
+pub fn transfer_logs(
+    web3: Web3<Http>,
+    contract_address: Address,
+    from: Option<Address>,
+    to: Option<Address>,
+    from_block: BlockNumber,
+    to_block: BlockNumber,
+) -> impl Future<Item = Vec<ERC20Transfer>, Error = ()> {
+    let from = if let Some(from) = from {
+        Some(vec![H256::from(from)])
+    } else {
+        None
+    };
+    let to = if let Some(to) = to {
+        Some(vec![H256::from(to)])
+    } else {
+        None
+    };
+
+    // create a filter for Transfer events from `from_block` until `to_block
+    // that filters all events indexed by `from` and `to`.
+    let filter = FilterBuilder::default()
+        .from_block(from_block)
+        .to_block(to_block)
+        .address(vec![contract_address])
+        .topics(
+            Some(vec![H256::from(
+                // keccak256("transfer(address,address,to)")
+                TRANSFER_EVENT_FILTER,
+            )]),
+            from,
+            to,
+            None,
+        )
+        .build();
+
+    // Make an eth_getLogs call to the Ethereum node
+    web3.eth()
+        .logs(filter)
+        .map_err(move |err| error!("Got error when fetching transfer logs{:?}", err))
+        .and_then(move |logs| {
+            let mut ret = Vec::new();
+            for log in logs {
+                // NOTE: From/to are indexed events.
+                // Amount is parsed directly from the data field.
+                let indexed = log.topics;
+                let from = H160::from(indexed[1]);
+                let to = H160::from(indexed[2]);
+                let data = log.data;
+                let amount = U256::from_str(&hex::encode(data.0)).unwrap();
+                ret.push(ERC20Transfer {
+                    tx_hash: log.transaction_hash.unwrap(),
+                    from,
+                    to,
+                    amount,
+                });
+            }
+            Ok(ret)
+        })
+}
+
 // TODO: Extend this function to inspect the data field of a
 // transaction, so that it supports contract wallets such as the Gnosis Multisig
 // etc. There is no need to implement any ERC20 functionality here since these
@@ -71,9 +147,8 @@ mod tests {
         // https://etherscan.io/tx/0x6fd1b68f02f4201a38662647b7f09170b159faec6af4825ae509beefeb8e8130
         let to = "c92be489639a9c61f517bd3b955840fa19bc9b7c".parse().unwrap();
         let value = "16345785d8a0000".into();
-        let nonce = 1.into();
         let token_address = Some("B8c77482e45F1F44dE1745F52C74426C631bDD52".into());
-        let tx = make_tx(to, value, nonce, token_address);
+        let tx = make_tx(to, value, token_address);
         assert_eq!(tx.to, token_address);
         assert_eq!(tx.value, U256::from(0));
         assert_eq!(hex::encode(tx.data), "a9059cbb000000000000000000000000c92be489639a9c61f517bd3b955840fa19bc9b7c000000000000000000000000000000000000000000000000016345785d8a0000")
@@ -83,9 +158,8 @@ mod tests {
     fn test_eth_make_tx() {
         let to = "c92be489639a9c61f517bd3b955840fa19bc9b7c".parse().unwrap();
         let value = "16345785d8a0000".into();
-        let nonce = 1.into();
         let token_address = None;
-        let tx = make_tx(to, value, nonce, token_address);
+        let tx = make_tx(to, value, token_address);
         assert_eq!(tx.to, Some(to));
         assert_eq!(tx.value, value);
         let empty: Vec<u8> = Vec::new();