feature(engines): Add trait for saving leftover values

Implement the trait for the eth engine

Author: gakonst

crates/interledger-settlement-engines/Cargo.toml

@@ -35,6 +35,7 @@ clap = "2.32.0"
 clarity = { git =  "https://github.com/gakonst/clarity" }
 sha3 = "0.8.2"
 num-bigint = "0.2.2"
+num-traits = "0.2.8"
 
 [dev-dependencies]
 lazy_static = "1.3"

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

@@ -17,6 +17,7 @@ use hyper::StatusCode;
 use interledger_store_redis::RedisStoreBuilder;
 use log::info;
 use num_bigint::BigUint;
+use num_traits::Zero;
 use redis::IntoConnectionInfo;
 use reqwest::r#async::{Client, Response as HttpResponse};
 use ring::{digest, hmac};
@@ -489,7 +490,11 @@ where
             .push("settlements");
         let client = Client::new();
         debug!("Making POST to {:?} {:?} about {:?}", url, amount, tx_hash);
+        let self_clone = self.clone();
         let action = move || {
+            // need to make 2 clones, one to own the variables in the function
+            // and one for the retry closure..
+            let self_clone = self_clone.clone();
             let account_id = account_id.clone();
             client
                 .post(url.clone())
@@ -504,11 +509,7 @@ where
                 })
                 .and_then(move |response| {
                     trace!("Accounting system responded with {:?}", response);
-                    if response.status().is_success() {
-                        Ok(())
-                    } else {
-                        Err(())
-                    }
+                    self_clone.process_connector_response(account_id_clone2, response, amount)
                 })
         };
         Retry::spawn(
@@ -520,6 +521,68 @@ where
         })
     }
 
+    fn process_connector_response(
+        &self,
+        account_id: String,
+        response: HttpResponse,
+        amount: U256,
+    ) -> impl Future<Item = (), Error = ()> {
+        let engine_scale = self.asset_scale;
+        let store = self.store.clone();
+        if !response.status().is_success() {
+            return Either::A(err(()));
+        }
+        Either::B(
+            response
+                .into_body()
+                .concat2()
+                .map_err(|err| {
+                    let err = format!("Couldn't retrieve body {:?}", err);
+                    error!("{}", err);
+                })
+                .and_then(move |body| {
+                    // Get the amount stored by the connector and
+                    // check for any precision loss / overflow
+                    serde_json::from_slice::<Quantity>(&body).map_err(|err| {
+                        let err = format!("Couldn't parse body {:?} into Quantity {:?}", body, err);
+                        error!("{}", err);
+                    })
+                })
+                .and_then(move |quantity: Quantity| {
+                    // Convert both to BigUInts so we can do arithmetic
+                    join_all(vec![
+                        result(BigUint::from_str(&quantity.amount)),
+                        result(BigUint::from_str(&amount.to_string())),
+                    ])
+                    .map_err(|err| {
+                        let error_msg = format!("Error converting to BigUints {:?}", err);
+                        error!("{:?}", error_msg);
+                    })
+                    .and_then(move |amounts: Vec<BigUint>| {
+                        let connector_amount = amounts[0].clone();
+                        let engine_amount = amounts[1].clone();
+                        // Scale the amount settled by the
+                        // connector back up to our scale
+                        result(connector_amount.normalize_scale(ConvertDetails {
+                            from: quantity.scale,
+                            to: engine_scale,
+                        }))
+                        .and_then(move |scaled_connector_amount| {
+                            let diff: BigUint = engine_amount - scaled_connector_amount;
+                            if diff > Zero::zero() {
+                                // connector settled less than we
+                                // instructed it to, so we must save
+                                // the difference for the leftovers
+                                Either::A(store.save_leftovers(account_id, diff))
+                            } else {
+                                Either::B(ok(()))
+                            }
+                        })
+                    })
+                }),
+        )
+    }
+
     /// Helper function which submits an Ethereum ledger transaction to `to` for `amount`.
     /// If called with `token_address`, it makes an ERC20 transaction instead.
     /// Due to the lack of an API to create and sign the transaction

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

@@ -61,6 +61,22 @@ pub struct TestStore {
     pub last_observed_block: Arc<RwLock<U256>>,
     pub saved_hashes: Arc<RwLock<HashMap<H256, bool>>>,
     pub cache_hits: Arc<RwLock<u64>>,
+    pub leftovers: Arc<RwLock<HashMap<String, String>>>,
+}
+
+use crate::stores::LeftoversStore;
+use num_bigint::BigUint;
+
+impl LeftoversStore for TestStore {
+    fn save_leftovers(
+        &self,
+        account_id: String,
+        leftovers: BigUint,
+    ) -> Box<Future<Item = (), Error = ()> + Send> {
+        let mut guard = self.leftovers.write();
+        (*guard).insert(account_id, leftovers.to_string());
+        Box::new(ok(()))
+    }
 }
 
 impl EthereumStore for TestStore {

crates/interledger-settlement-engines/src/main.rs

@@ -1,3 +1,5 @@
+#![type_length_limit = "1739269"]
+
 use clap::{value_t, App, Arg, SubCommand};
 use hex;
 use std::str::FromStr;

crates/interledger-settlement-engines/src/stores/redis_ethereum_ledger/store.rs

@@ -105,6 +105,23 @@ impl EthereumLedgerRedisStore {
     }
 }
 
+impl LeftoversStore for EthereumLedgerRedisStore {
+    fn save_leftovers(
+        &self,
+        account_id: String,
+        leftovers: BigUint,
+    ) -> Box<dyn Future<Item = (), Error = ()> + Send> {
+        let mut pipe = redis::pipe();
+        pipe.set(format!("leftovers:{}", account_id), leftovers.to_string())
+            .ignore();
+        Box::new(
+            pipe.query_async(self.connection.clone())
+                .map_err(move |err| error!("Error saving leftovers {:?}: {:?}", leftovers, err))
+                .and_then(move |(_conn, _ret): (_, Value)| Ok(())),
+        )
+    }
+}
+
 impl EthereumStore for EthereumLedgerRedisStore {
     type Account = Account;