feat(engine): Add leftovers value when settling to connector

Author: gakonst

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

@@ -374,7 +374,7 @@ where
                         store
                             .load_account_id_from_address(addr)
                             .and_then(move |id| {
-                                self_clone.notify_connector(id.to_string(), amount, tx_hash)
+                                self_clone.notify_connector(id.to_string(), amount.to_string(), tx_hash)
                             })
                             .and_then(move |_| {
                                 // only save the transaction hash if the connector
@@ -457,7 +457,7 @@ where
                         Either::A(
                         store.load_account_id_from_address(addr)
                         .and_then(move |id| {
-                            self_clone.notify_connector(id.to_string(), amount, tx_hash)
+                            self_clone.notify_connector(id.to_string(), amount.to_string(), tx_hash)
                         })
                         .and_then(move |_| {
                             // only save the transaction hash if the connector
@@ -478,7 +478,7 @@ where
     fn notify_connector(
         &self,
         account_id: String,
-        amount: U256,
+        amount: String,
         tx_hash: H256,
     ) -> impl Future<Item = (), Error = ()> {
         let mut url = self.connector_url.clone();
@@ -489,45 +489,65 @@ where
             .push("accounts")
             .push(&account_id.clone())
             .push("settlements");
-        let client = Client::new();
         debug!("Making POST to {:?} {:?} about {:?}", url, amount, tx_hash);
         let self_clone = self.clone();
+        let store = self.store.clone();
+        let amount_clone = amount.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 store = store.clone();
             let account_id = account_id.clone();
             let account_id_clone2 = account_id.clone();
-            client
-                .post(url.clone())
-                .header("Idempotency-Key", tx_hash.to_string())
-                .json(&json!({ "amount": amount.to_string(), "scale" : engine_scale }))
-                .send()
-                .map_err(move |err| {
-                    error!(
-                        "Error notifying Accounting System's account: {:?}, amount: {:?}: {:?}",
-                        account_id, amount, err
-                    )
-                })
-                .and_then(move |response| {
-                    trace!("Accounting system responded with {:?}", response);
-                    self_clone.process_connector_response(account_id_clone2, response, amount)
+            let amount = amount.clone();
+            let url = url.clone();
+
+            // settle for amount + leftovers
+            store.load_leftovers(account_id.clone())
+            .and_then(move |leftovers| {
+                result(BigUint::from_str(&amount.clone()).map_err(move |err| {
+                    let error_msg = format!("Error converting to BigUint {:?}", err);
+                    error!("{:?}", error_msg);
+                }))
+                .and_then(move |amount| {
+                    Ok(amount + leftovers)
                 })
+            })
+            .and_then(move |full_amount| {
+                let client = Client::new();
+                let full_amount_clone = full_amount.clone();
+                client
+                    .post(url.clone())
+                    .header("Idempotency-Key", tx_hash.to_string())
+                    .json(&json!({ "amount": full_amount.clone().to_string(), "scale" : engine_scale }))
+                    .send()
+                    .map_err(move |err| {
+                        error!(
+                            "Error notifying Accounting System's account: {:?}, amount: {:?}: {:?}",
+                            account_id, full_amount_clone, err
+                        )
+                    })
+                    .and_then(move |response| {
+                        trace!("Accounting system responded with {:?}", response);
+                        self_clone.process_connector_response(account_id_clone2, response, full_amount.clone())
+                    })
+            })
         };
         Retry::spawn(
             ExponentialBackoff::from_millis(10).take(MAX_RETRIES),
             action,
         )
         .map_err(move |_| {
-            error!("Exceeded max retries when notifying connector about account {:?} for amount {:?} and transaction hash {:?}. Please check your API.", account_id_clone, amount, tx_hash)
+            error!("Exceeded max retries when notifying connector about account {:?} for amount {:?} and transaction hash {:?}. Please check your API.", account_id_clone, amount_clone, tx_hash)
         })
     }
 
     fn process_connector_response(
         &self,
         account_id: String,
         response: HttpResponse,
-        amount: U256,
+        engine_amount: BigUint,
     ) -> impl Future<Item = (), Error = ()> {
         let engine_scale = self.asset_scale;
         let store = self.store.clone();
@@ -551,18 +571,12 @@ where
                     })
                 })
                 .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())),
-                    ])
+                    result(BigUint::from_str(&quantity.amount))
                     .map_err(|err| {
-                        let error_msg = format!("Error converting to BigUints {:?}", err);
+                        let error_msg = format!("Error converting to BigUint {:?}", err);
                         error!("{:?}", error_msg);
                     })
-                    .and_then(move |amounts: Vec<BigUint>| {
-                        let connector_amount = amounts[0].clone();
-                        let engine_amount = amounts[1].clone();
+                    .and_then(move |connector_amount: BigUint| {
                         // Scale the amount settled by the
                         // connector back up to our scale
                         result(connector_amount.normalize_scale(ConvertDetails {

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

@@ -61,11 +61,12 @@ 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>>>,
+    pub leftovers: Arc<RwLock<HashMap<String, BigUint>>>,
 }
 
 use crate::stores::LeftoversStore;
 use num_bigint::BigUint;
+use num_traits::Zero;
 
 impl LeftoversStore for TestStore {
     fn save_leftovers(
@@ -74,9 +75,18 @@ impl LeftoversStore for TestStore {
         leftovers: BigUint,
     ) -> Box<Future<Item = (), Error = ()> + Send> {
         let mut guard = self.leftovers.write();
-        (*guard).insert(account_id, leftovers.to_string());
+        (*guard).insert(account_id, leftovers);
         Box::new(ok(()))
     }
+
+    fn load_leftovers(&self, account_id: String) -> Box<Future<Item = BigUint, Error = ()> + Send> {
+        let mut guard = self.leftovers.read();
+        if let Some(l) = guard.get(&account_id) {
+            Box::new(ok(l.clone()))
+        } else {
+            Box::new(err(()))
+        }
+    }
 }
 
 impl EthereumStore for TestStore {

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

@@ -1,4 +1,4 @@
-#![type_length_limit = "1739269"]
+#![type_length_limit = "3396805"]
 
 use clap::{value_t, App, Arg, SubCommand};
 use hex;

crates/interledger-settlement-engines/src/stores/mod.rs

@@ -37,4 +37,10 @@ pub trait LeftoversStore {
         account_id: String,
         leftovers: BigUint,
     ) -> Box<dyn Future<Item = (), Error = ()> + Send>;
+
+    /// Saves the leftover data
+    fn load_leftovers(
+        &self,
+        account_id: String,
+    ) -> Box<dyn Future<Item = BigUint, Error = ()> + Send>;
 }

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

@@ -6,6 +6,7 @@ use futures::{
 use ethereum_tx_sign::web3::types::{Address as EthAddress, H256, U256};
 use interledger_service::Account as AccountTrait;
 use std::collections::HashMap;
+use std::str::FromStr;
 
 use crate::engines::ethereum_ledger::{EthereumAccount, EthereumAddresses, EthereumStore};
 use redis::{self, cmd, r#async::SharedConnection, ConnectionInfo, PipelineCommands, Value};
@@ -122,6 +123,26 @@ impl LeftoversStore for EthereumLedgerRedisStore {
                 .and_then(move |(_conn, _ret): (_, Value)| Ok(())),
         )
     }
+
+    fn load_leftovers(
+        &self,
+        account_id: String,
+    ) -> Box<dyn Future<Item = BigUint, Error = ()> + Send> {
+        let mut pipe = redis::pipe();
+        pipe.get(format!("leftovers:{}", account_id))
+            .ignore();
+        Box::new(
+            pipe.query_async(self.connection.clone())
+                .map_err(move |err| error!("Error loading leftovers {:?}: ", err))
+                .and_then(move |(_conn, leftovers): (_, String)| {
+                    if let Ok(leftovers) = BigUint::from_str(&leftovers) {
+                        Box::new(ok(leftovers))
+                    } else {
+                        Box::new(err(()))
+                    }
+                })
+        )
+    }
 }
 
 impl EthereumStore for EthereumLedgerRedisStore {