fix(tests): return 0 if there were no leftovers found in the store

Author: gakonst

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

@@ -494,67 +494,66 @@ where
         amount: String,
         tx_hash: H256,
     ) -> impl Future<Item = (), Error = ()> {
-        let mut url = self.connector_url.clone();
-        let account_id_clone = account_id.clone();
         let engine_scale = self.asset_scale;
+        let mut url = self.connector_url.clone();
         url.path_segments_mut()
             .expect("Invalid connector URL")
             .push("accounts")
             .push(&account_id.clone())
             .push("settlements");
         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();
-            let amount = amount.clone();
-            let url = url.clone();
-
-            // settle for amount + leftovers
-            store.pop_leftovers(account_id.clone())
+
+        // settle for amount + leftovers
+        self.store
+            .pop_leftovers(account_id.clone())
             .and_then(move |leftovers| {
-                result(BigUint::from_str(&amount.clone()).map_err(move |err| {
+                debug!("POPPED LEFTOVERS {:?}", leftovers);
+                result(BigUint::from_str(&amount).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
-                        )
+                    debug!("Got uncredited amount {}", amount);
+                    let full_amount = amount + leftovers;
+                    let client = Client::new();
+                    debug!(
+                        "Notifying accounting system about full amount: {}",
+                        full_amount
+                    );
+                    let url = url.clone();
+                    let account_id_clone = account_id.clone();
+                    let full_amount_clone = full_amount.clone();
+
+                    let action = move || {
+                        let account_id = account_id.clone();
+                        let full_amount = full_amount.clone();
+                        client
+                            .post(url.clone())
+                            .header("Idempotency-Key", tx_hash.to_string())
+                            .json(&json!(Quantity::new(full_amount.clone(), engine_scale)))
+                            .send()
+                            .map_err(move |err| {
+                                error!(
+                                    "Error notifying Accounting System's account: {:?}, amount: {:?}: {:?}",
+                                    account_id.clone(), full_amount.clone(), err
+                                );
+                            })
+                    };
+
+                    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, full_amount_clone, tx_hash)
                     })
                     .and_then(move |response| {
                         trace!("Accounting system responded with {:?}", response);
                         Ok(()) // This call causes the type_length_error
                         // 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_clone, tx_hash)
-        })
     }
 
     fn process_connector_response(

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

@@ -91,7 +91,7 @@ impl LeftoversStore for TestStore {
             (*guard).insert(account_id, Zero::zero());
             Box::new(ok(l.clone()))
         } else {
-            Box::new(err(()))
+            Box::new(ok(Zero::zero()))
         }
     }
 }

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

@@ -9,6 +9,7 @@ use std::collections::HashMap;
 use std::str::FromStr;
 
 use crate::engines::ethereum_ledger::{EthereumAccount, EthereumAddresses, EthereumStore};
+use num_traits::Zero;
 use redis::{self, cmd, r#async::SharedConnection, ConnectionInfo, PipelineCommands, Value};
 
 use log::{error, trace};
@@ -116,6 +117,7 @@ impl LeftoversStore for EthereumLedgerRedisStore {
         account_id: String,
         leftovers: Self::AssetType,
     ) -> Box<dyn Future<Item = (), Error = ()> + Send> {
+        trace!("Saving leftovers {:?} {:?}", account_id, leftovers);
         let mut pipe = redis::pipe();
         pipe.set(format!("leftovers:{}", account_id), leftovers.to_string())
             .ignore();
@@ -130,18 +132,24 @@ impl LeftoversStore for EthereumLedgerRedisStore {
         &self,
         account_id: String,
     ) -> Box<dyn Future<Item = Self::AssetType, Error = ()> + Send> {
+        trace!("Loading leftovers {:?}", account_id);
         let mut pipe = redis::pipe();
         // Loads the value and resets it to 0
         pipe.getset(format!("leftovers:{}", account_id), 0);
         Box::new(
             pipe.query_async(self.connection.clone())
                 .map_err(move |err| error!("Error loading leftovers {:?}: ", err))
                 .and_then(move |(_conn, leftovers): (_, Vec<String>)| {
-                    // redis.rs returns a bulk value for some reason, length is always 1
-                    if let Ok(leftovers) = BigUint::from_str(&leftovers[0]) {
-                        Box::new(ok(leftovers))
+                    // redis.rs returns a bulk value for some reason, length is
+                    // always 1
+                    if leftovers.len() == 1 {
+                        if let Ok(leftovers) = BigUint::from_str(&leftovers[0]) {
+                            Box::new(ok(leftovers))
+                        } else {
+                            Box::new(ok(Zero::zero()))
+                        }
                     } else {
-                        Box::new(err(()))
+                        Box::new(ok(Zero::zero()))
                     }
                 }),
         )