Improve Asset Scale conversions

crates/interledger-api/src/lib.rs

@@ -71,7 +71,6 @@ pub struct AccountDetails {
     pub amount_per_minute_limit: Option<u64>,
     pub packets_per_minute_limit: Option<u32>,
     pub settlement_engine_url: Option<String>,
-    pub settlement_engine_asset_scale: Option<u8>,
 }
 
 pub struct NodeApi<S, I> {

crates/interledger-service-util/src/exchange_rates_service.rs

@@ -90,9 +90,13 @@ where
                 .build()));
             };
 
+            // Note that this is not an overflow safe operation. Given realistic
+            // assumptions, the asset scale will be <=18 and >=1.
+            // It is doubtful that the exchange rate between two assets,
+            // multiplied by 18 would exceed std::f64::MAX
             let scaled_rate = rate.normalize_scale(ConvertDetails {
-                from: request.to.asset_scale(),
-                to: request.from.asset_scale(),
+                from: request.from.asset_scale(),
+                to: request.to.asset_scale(),
             });
             let outgoing_amount = (request.prepare.amount() as f64) * scaled_rate;
             request.prepare.set_amount(outgoing_amount as u64);

crates/interledger-settlement-engines/Cargo.toml

@@ -34,6 +34,7 @@ http = "0.1.17"
 clap = "2.32.0"
 clarity = { git =  "https://github.com/gakonst/clarity" }
 sha3 = "0.8.2"
+num-bigint = "0.2.2"
 
 [dev-dependencies]
 lazy_static = "1.3"
@@ -43,4 +44,4 @@ net2 = "0.2.33"
 os_type = "2.2.0"
 rand = "0.7.0"
 interledger = { path = "../interledger", version = "0.4.0" }
-interledger-packet = { path = "../interledger-packet", version = "0.2.1" }
+interledger-packet = { path = "../interledger-packet", version = "0.2.1" }

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

@@ -7,7 +7,7 @@ use futures::{
 use hyper::{Response, StatusCode};
 use interledger_settlement::Quantity;
 use interledger_settlement::{IdempotentData, IdempotentStore};
-use log::trace;
+use log::error;
 use ring::digest::{digest, SHA256};
 use tokio::executor::spawn;
 use tower_web::{net::ConnectionStream, ServiceBuilder};
@@ -77,19 +77,23 @@ impl_web! {
             &self,
             idempotency_key: String,
             input_hash: [u8; 32],
-        ) -> impl Future<Item = (StatusCode, Bytes), Error = String> {
+        ) -> impl Future<Item = Option<(StatusCode, Bytes)>, Error = String> {
             self.store
                 .load_idempotent_data(idempotency_key.clone())
                 .map_err(move |_| {
-                    let error_msg = "Couldn't load idempotent data".to_owned();
-                    trace!("{}", error_msg);
+                    let error_msg = format!("Couldn't load idempotent data for idempotency key {:?}", idempotency_key);
+                    error!("{}", error_msg);
                     error_msg
                 })
-                .and_then(move |ret: IdempotentData| {
-                    if ret.2 == input_hash {
-                        Ok((ret.0, ret.1))
+                .and_then(move |ret: Option<IdempotentData>| {
+                    if let Some(ret) = ret {
+                        if ret.2 == input_hash {
+                            Ok(Some((ret.0, ret.1)))
+                        } else {
+                            Ok(Some((StatusCode::from_u16(409).unwrap(), Bytes::from(&b"Provided idempotency key is tied to other input"[..]))))
+                        }
                     } else {
-                        Ok((StatusCode::from_u16(409).unwrap(), Bytes::from(&b"Provided idempotency key is tied to other input"[..])))
+                        Ok(None)
                     }
                 })
         }
@@ -104,8 +108,8 @@ impl_web! {
                 Either::A(
                     self.check_idempotency(idempotency_key.clone(), input_hash)
                     .map_err(|err| Response::builder().status(502).body(err).unwrap())
-                    .then(move |ret: Result<(StatusCode, Bytes), Response<String>>| {
-                        if let Ok(ret) = ret {
+                    .and_then(move |ret: Option<(StatusCode, Bytes)>| {
+                        if let Some(ret) = ret {
                             let resp = Response::builder().status(ret.0).body(String::from_utf8_lossy(&ret.1).to_string()).unwrap();
                             if ret.0.is_success() {
                                 return Either::A(Either::A(ok(resp)))

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

@@ -16,6 +16,7 @@ use ethereum_tx_sign::web3::{
 use hyper::StatusCode;
 use interledger_store_redis::RedisStoreBuilder;
 use log::info;
+use num_bigint::BigUint;
 use redis::IntoConnectionInfo;
 use reqwest::r#async::{Client, Response as HttpResponse};
 use ring::{digest, hmac};
@@ -35,7 +36,7 @@ use uuid::Uuid;
 
 use crate::stores::redis_ethereum_ledger::*;
 use crate::{ApiResponse, CreateAccount, SettlementEngine, SettlementEngineApi};
-use interledger_settlement::Quantity;
+use interledger_settlement::{Convert, ConvertDetails, Quantity};
 
 const MAX_RETRIES: usize = 10;
 const ETH_CREATE_ACCOUNT_PREFIX: &[u8] = b"ilp-ethl-create-account-message";
@@ -480,7 +481,7 @@ where
     ) -> impl Future<Item = (), Error = ()> {
         let mut url = self.connector_url.clone();
         let account_id_clone = account_id.clone();
-        let asset_scale = self.asset_scale;
+        let engine_scale = self.asset_scale;
         url.path_segments_mut()
             .expect("Invalid connector URL")
             .push("accounts")
@@ -493,7 +494,7 @@ where
             client
                 .post(url.clone())
                 .header("Idempotency-Key", tx_hash.to_string())
-                .json(&json!({ "amount": amount.to_string(), "scale" : asset_scale }))
+                .json(&json!({ "amount": amount.to_string(), "scale" : engine_scale }))
                 .send()
                 .map_err(move |err| {
                     error!(
@@ -757,12 +758,29 @@ where
         body: Quantity,
     ) -> Box<dyn Future<Item = ApiResponse, Error = ApiResponse> + Send> {
         let self_clone = self.clone();
+        let engine_scale = self.asset_scale;
         Box::new(
-            result(U256::from_dec_str(&body.amount).map_err(move |err| {
+            result(BigUint::from_str(&body.amount).map_err(move |err| {
                 let error_msg = format!("Error converting to U256 {:?}", err);
                 error!("{:?}", error_msg);
                 (StatusCode::from_u16(400).unwrap(), error_msg)
             }))
+            .and_then(move |amount_from_connector| {
+                // If we receive a Quantity { amount: "1", scale: 9},
+                // we must normalize it to our engine's scale
+                let amount = amount_from_connector.normalize_scale(ConvertDetails {
+                    from: body.scale,
+                    to: engine_scale,
+                });
+
+                // Typecast from num_bigint::BigUInt because we're using
+                // ethereum_types::U256 for all rust-web3 related functionality
+                result(U256::from_dec_str(&amount.to_string()).map_err(move |err| {
+                    let error_msg = format!("Error converting to U256 {:?}", err);
+                    error!("{:?}", error_msg);
+                    (StatusCode::from_u16(400).unwrap(), error_msg)
+                }))
+            })
             .and_then(move |amount| {
                 self_clone
                     .load_account(account_id)
@@ -929,7 +947,7 @@ mod tests {
 
         let bob_mock = mockito::mock("POST", "/accounts/42/settlements")
             .match_body(mockito::Matcher::JsonString(
-                "{\"amount\": \"100\", \"scale\": 18 }".to_string(),
+                "{\"amount\": \"100000000000\", \"scale\": 18 }".to_string(),
             ))
             .with_status(200)
             .with_body("OK".to_string())
@@ -953,7 +971,7 @@ mod tests {
         );
 
         let ret =
-            block_on(alice_engine.send_money(bob.id.to_string(), Quantity::new(100, 6))).unwrap();
+            block_on(alice_engine.send_money(bob.id.to_string(), Quantity::new(100, 9))).unwrap();
         assert_eq!(ret.0.as_u16(), 200);
         assert_eq!(ret.1, "OK");
 

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

@@ -176,14 +176,14 @@ impl IdempotentStore for TestStore {
     fn load_idempotent_data(
         &self,
         idempotency_key: String,
-    ) -> Box<dyn Future<Item = IdempotentData, Error = ()> + Send> {
+    ) -> Box<dyn Future<Item = Option<IdempotentData>, Error = ()> + Send> {
         let cache = self.cache.read();
         if let Some(data) = cache.get(&idempotency_key) {
             let mut guard = self.cache_hits.write();
             *guard += 1; // used to test how many times this branch gets executed
-            Box::new(ok((data.0, Bytes::from(data.1.clone()), data.2)))
+            Box::new(ok(Some((data.0, Bytes::from(data.1.clone()), data.2))))
         } else {
-            Box::new(err(()))
+            Box::new(ok(None))
         }
     }
 

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

@@ -13,11 +13,10 @@ pub type IdempotentEngineData = (StatusCode, Bytes, [u8; 32]);
 pub trait IdempotentEngineStore {
     /// Returns the API response that was saved when the idempotency key was used
     /// Also returns a hash of the input data which resulted in the response
-    /// Returns error if the data was not found
     fn load_idempotent_data(
         &self,
         idempotency_key: String,
-    ) -> Box<dyn Future<Item = IdempotentEngineData, Error = ()> + Send>;
+    ) -> Box<dyn Future<Item = Option<IdempotentEngineData>, Error = ()> + Send>;
 
     /// Saves the data that was passed along with the api request for later
     /// The store MUST also save a hash of the input, so that it errors out on requests

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

@@ -43,7 +43,7 @@ impl IdempotentEngineStore for EngineRedisStore {
     fn load_idempotent_data(
         &self,
         idempotency_key: String,
-    ) -> Box<dyn Future<Item = IdempotentEngineData, Error = ()> + Send> {
+    ) -> Box<dyn Future<Item = Option<IdempotentEngineData>, Error = ()> + Send> {
         let idempotency_key_clone = idempotency_key.clone();
         Box::new(
             cmd("HGETALL")
@@ -65,13 +65,13 @@ impl IdempotentEngineStore for EngineRedisStore {
                             trace!("Loaded idempotency key {:?} - {:?}", idempotency_key, ret);
                             let mut input_hash: [u8; 32] = Default::default();
                             input_hash.copy_from_slice(input_hash_slice.as_ref());
-                            Ok((
+                            Ok(Some((
                                 StatusCode::from_str(status_code).unwrap(),
                                 Bytes::from(data.clone()),
                                 input_hash,
-                            ))
+                            )))
                         } else {
-                            Err(())
+                            Ok(None)
                         }
                     },
                 ),
@@ -137,20 +137,23 @@ mod tests {
                         .load_idempotent_data(IDEMPOTENCY_KEY.clone())
                         .map_err(|err| eprintln!("Redis error: {:?}", err))
                         .and_then(move |data1| {
-                            assert_eq!(data1, (StatusCode::OK, Bytes::from("TEST"), input_hash));
+                            assert_eq!(
+                                data1.unwrap(),
+                                (StatusCode::OK, Bytes::from("TEST"), input_hash)
+                            );
                             let _ = context;
 
                             store
                                 .load_idempotent_data("asdf".to_string())
                                 .map_err(|err| eprintln!("Redis error: {:?}", err))
-                                .and_then(move |_data2| {
-                                    assert_eq!(_data2.0, StatusCode::OK);
+                                .and_then(move |data2| {
+                                    assert!(data2.is_none());
                                     let _ = context;
                                     Ok(())
                                 })
                         })
                 })
         }))
-        .unwrap_err() // the second idempotent load fails
+        .unwrap()
     }
 }

crates/interledger-settlement-engines/tests/eth_ledger_settlement.rs

@@ -16,9 +16,7 @@ mod redis_helpers;
 use redis_helpers::*;
 
 mod test_helpers;
-use test_helpers::{
-    create_account, get_balance, send_money, start_eth_engine, start_ganache, ETH_DECIMALS,
-};
+use test_helpers::{create_account, get_balance, send_money, start_eth_engine, start_ganache};
 
 #[test]
 /// In this test we have Alice and Bob who have peered with each other and run
@@ -29,6 +27,7 @@ use test_helpers::{
 /// transactions, and once the transaction has sufficient confirmations it
 /// lets Bob's connector know about it, so that it adjusts their credit.
 fn eth_ledger_settlement() {
+    let eth_decimals = 9;
     // Nodes 1 and 2 are peers, Node 2 is the parent of Node 3
     let _ = env_logger::try_init();
     let context = TestContext::new();
@@ -76,7 +75,7 @@ fn eth_ledger_settlement() {
                     .insert_account(AccountDetails {
                         ilp_address: Address::from_str("example.alice").unwrap(),
                         asset_code: "ETH".to_string(),
-                        asset_scale: ETH_DECIMALS,
+                        asset_scale: eth_decimals,
                         btp_incoming_token: None,
                         btp_uri: None,
                         http_endpoint: None,
@@ -93,13 +92,12 @@ fn eth_ledger_settlement() {
                         packets_per_minute_limit: None,
                         amount_per_minute_limit: None,
                         settlement_engine_url: None,
-                        settlement_engine_asset_scale: None,
                     })
                     .and_then(move |_| {
                         node1_clone.insert_account(AccountDetails {
                             ilp_address: Address::from_str("example.bob").unwrap(),
                             asset_code: "ETH".to_string(),
-                            asset_scale: ETH_DECIMALS,
+                            asset_scale: eth_decimals,
                             btp_incoming_token: None,
                             btp_uri: None,
                             http_endpoint: Some(format!("http://localhost:{}/ilp", node2_http)),
@@ -119,7 +117,6 @@ fn eth_ledger_settlement() {
                                 "http://localhost:{}",
                                 node1_engine
                             )),
-                            settlement_engine_asset_scale: Some(ETH_DECIMALS),
                         })
                     })
                     .and_then(move |_| node1.serve())
@@ -146,7 +143,7 @@ fn eth_ledger_settlement() {
                     .insert_account(AccountDetails {
                         ilp_address: Address::from_str("example.bob").unwrap(),
                         asset_code: "ETH".to_string(),
-                        asset_scale: ETH_DECIMALS,
+                        asset_scale: eth_decimals,
                         btp_incoming_token: None,
                         btp_uri: None,
                         http_endpoint: None,
@@ -163,14 +160,13 @@ fn eth_ledger_settlement() {
                         packets_per_minute_limit: None,
                         amount_per_minute_limit: None,
                         settlement_engine_url: None,
-                        settlement_engine_asset_scale: None,
                     })
                     .and_then(move |_| {
                         node2
                             .insert_account(AccountDetails {
                                 ilp_address: Address::from_str("example.alice").unwrap(),
                                 asset_code: "ETH".to_string(),
-                                asset_scale: ETH_DECIMALS,
+                                asset_scale: eth_decimals,
                                 btp_incoming_token: None,
                                 btp_uri: None,
                                 http_endpoint: Some(format!("http://localhost:{}/ilp", node1_http)),
@@ -190,7 +186,6 @@ fn eth_ledger_settlement() {
                                     "http://localhost:{}",
                                     node2_engine
                                 )),
-                                settlement_engine_asset_scale: Some(ETH_DECIMALS),
                             })
                             .and_then(move |_| node2.serve())
                     })
@@ -223,29 +218,29 @@ fn eth_ledger_settlement() {
                         .and_then(move |_| send1)
                         .and_then(move |_| {
                             get_balance(1, node1_http, "bob").and_then(move |ret| {
-                                assert_eq!(ret, "{\"balance\":\"10\"}");
+                                assert_eq!(ret, 10);
                                 get_balance(1, node2_http, "alice").and_then(move |ret| {
-                                    assert_eq!(ret, "{\"balance\":\"-10\"}");
+                                    assert_eq!(ret, -10);
                                     Ok(())
                                 })
                             })
                         })
                         .and_then(move |_| send2)
                         .and_then(move |_| {
                             get_balance(1, node1_http, "bob").and_then(move |ret| {
-                                assert_eq!(ret, "{\"balance\":\"30\"}");
+                                assert_eq!(ret, 30);
                                 get_balance(1, node2_http, "alice").and_then(move |ret| {
-                                    assert_eq!(ret, "{\"balance\":\"-30\"}");
+                                    assert_eq!(ret, -30);
                                     Ok(())
                                 })
                             })
                         })
                         .and_then(move |_| send3)
                         .and_then(move |_| {
                             get_balance(1, node1_http, "bob").and_then(move |ret| {
-                                assert_eq!(ret, "{\"balance\":\"70\"}");
+                                assert_eq!(ret, 70);
                                 get_balance(1, node2_http, "alice").and_then(move |ret| {
-                                    assert_eq!(ret, "{\"balance\":\"-70\"}");
+                                    assert_eq!(ret, -70);
                                     Ok(())
                                 })
                             })
@@ -261,9 +256,9 @@ fn eth_ledger_settlement() {
                             // Since the credit connection reached -71, and the
                             // settle_to is -10, a 61 Wei transaction is made.
                             get_balance(1, node1_http, "bob").and_then(move |ret| {
-                                assert_eq!(ret, "{\"balance\":\"10\"}");
+                                assert_eq!(ret, 10);
                                 get_balance(1, node2_http, "alice").and_then(move |ret| {
-                                    assert_eq!(ret, "{\"balance\":\"-10\"}");
+                                    assert_eq!(ret, -10);
                                     ganache_pid.kill().unwrap();
                                     Ok(())
                                 })