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,15 +90,217 @@ where
                 .build()));
             };
 
-            let scaled_rate = rate.normalize_scale(ConvertDetails {
-                from: request.to.asset_scale(),
-                to: request.from.asset_scale(),
+            // Can we overflow here?
+            let outgoing_amount = (request.prepare.amount() as f64) * rate;
+            let outgoing_amount = outgoing_amount.normalize_scale(ConvertDetails {
+                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);
-            trace!("Converted incoming amount of: {} {} (scale {}) from account {} to outgoing amount of: {} {} (scale {}) for account {}", request.original_amount, request.from.asset_code(), request.from.asset_scale(), request.from.id(), outgoing_amount, request.to.asset_code(), request.to.asset_scale(), request.to.id());
+
+            match outgoing_amount {
+                Ok(outgoing_amount) => {
+                    // The conversion succeeded, but the produced f64
+                    // is larger than the maximum value for a u64.
+                    // When it gets cast to a u64, it will end up being 0.
+                    if outgoing_amount != 0.0 && outgoing_amount as u64 == 0 {
+                        return Box::new(err(RejectBuilder {
+                            code: ErrorCode::F08_AMOUNT_TOO_LARGE,
+                            message: format!(
+                                "Could not cast outgoing amount to u64 {}",
+                                outgoing_amount,
+                            )
+                            .as_bytes(),
+                            triggered_by: Some(&self.ilp_address),
+                            data: &[],
+                        }
+                        .build()));
+                    }
+                    request.prepare.set_amount(outgoing_amount as u64);
+                    trace!("Converted incoming amount of: {} {} (scale {}) from account {} to outgoing amount of: {} {} (scale {}) for account {}",
+                        request.original_amount, request.from.asset_code(), request.from.asset_scale(), request.from.id(),
+                        outgoing_amount, request.to.asset_code(), request.to.asset_scale(), request.to.id());
+                }
+                Err(_) => {
+                    // This branch gets executed when the `Convert` trait
+                    // returns an error. Happens due to float
+                    // multiplication overflow .
+                    // (float overflow in Rust produces +inf)
+                    return Box::new(err(RejectBuilder {
+                        code: ErrorCode::F08_AMOUNT_TOO_LARGE,
+                        message: format!(
+                            "Could not convert exchange rate from {}:{} to: {}:{}. Got incoming amount: {}",
+                            request.from.asset_code(),
+                            request.from.asset_scale(),
+                            request.to.asset_code(),
+                            request.to.asset_scale(),
+                            request.prepare.amount(),
+                        )
+                        .as_bytes(),
+                        triggered_by: Some(&self.ilp_address),
+                        data: &[],
+                    }
+                    .build()));
+                }
+            }
         }
 
         Box::new(self.next.send_request(request))
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use futures::{future::ok, Future};
+    use interledger_ildcp::IldcpAccount;
+    use interledger_packet::{Address, FulfillBuilder, PrepareBuilder};
+    use interledger_service::{outgoing_service_fn, Account};
+    use std::collections::HashMap;
+    use std::str::FromStr;
+    use std::{
+        sync::{Arc, Mutex},
+        time::SystemTime,
+    };
+
+    #[test]
+    fn exchange_rate_ok() {
+        let ret = exchange_rate(100, 1, 1.0, 1, 2.0);
+        assert_eq!(ret.1[0].prepare.amount(), 200);
+
+        let ret = exchange_rate(1_000_000, 1, 3.0, 1, 2.0);
+        assert_eq!(ret.1[0].prepare.amount(), 666_666);
+    }
+
+    #[test]
+    fn exchange_conversion_error() {
+        // rejects f64 that does not fit in u64
+        let ret = exchange_rate(std::u64::MAX, 1, 1.0, 1, 2.0);
+        let reject = ret.0.unwrap_err();
+        assert_eq!(reject.code(), ErrorCode::F08_AMOUNT_TOO_LARGE);
+        assert!(reject.message().starts_with(b"Could not cast"));
+
+        // `Convert` errored
+        let ret = exchange_rate(std::u64::MAX, 1, 1.0, 255, std::f64::MAX);
+        let reject = ret.0.unwrap_err();
+        assert_eq!(reject.code(), ErrorCode::F08_AMOUNT_TOO_LARGE);
+        assert!(reject.message().starts_with(b"Could not convert"));
+    }
+
+    // Instantiates an exchange rate service and returns the fulfill/reject
+    // packet and the outgoing request after performing an asset conversion
+    fn exchange_rate(
+        amount: u64,
+        scale1: u8,
+        rate1: f64,
+        scale2: u8,
+        rate2: f64,
+    ) -> (Result<Fulfill, Reject>, Vec<OutgoingRequest<TestAccount>>) {
+        let requests = Arc::new(Mutex::new(Vec::new()));
+        let requests_clone = requests.clone();
+        let outgoing = outgoing_service_fn(move |request| {
+            requests_clone.lock().unwrap().push(request);
+            Box::new(ok(FulfillBuilder {
+                fulfillment: &[0; 32],
+                data: b"hello!",
+            }
+            .build()))
+        });
+        let mut service = test_service(rate1, rate2, outgoing);
+        let result = service
+            .send_request(OutgoingRequest {
+                from: TestAccount::new("ABC".to_owned(), scale1),
+                to: TestAccount::new("XYZ".to_owned(), scale2),
+                original_amount: amount,
+                prepare: PrepareBuilder {
+                    destination: Address::from_str("example.destination").unwrap(),
+                    amount,
+                    expires_at: SystemTime::now(),
+                    execution_condition: &[1; 32],
+                    data: b"hello",
+                }
+                .build(),
+            })
+            .wait();
+
+        let reqs = requests.lock().unwrap();
+        (result, reqs.clone())
+    }
+
+    #[derive(Debug, Clone)]
+    struct TestAccount {
+        ilp_address: Address,
+        asset_code: String,
+        asset_scale: u8,
+    }
+    impl TestAccount {
+        fn new(asset_code: String, asset_scale: u8) -> Self {
+            TestAccount {
+                ilp_address: Address::from_str("example.alice").unwrap(),
+                asset_code,
+                asset_scale,
+            }
+        }
+    }
+
+    impl Account for TestAccount {
+        type AccountId = u64;
+
+        fn id(&self) -> u64 {
+            0
+        }
+    }
+
+    impl IldcpAccount for TestAccount {
+        fn asset_code(&self) -> &str {
+            &self.asset_code
+        }
+
+        fn asset_scale(&self) -> u8 {
+            self.asset_scale
+        }
+
+        fn client_address(&self) -> &Address {
+            &self.ilp_address
+        }
+    }
+
+    #[derive(Debug, Clone)]
+    struct TestStore {
+        rates: HashMap<Vec<String>, (f64, f64)>,
+    }
+
+    impl ExchangeRateStore for TestStore {
+        fn get_exchange_rates(&self, asset_codes: &[&str]) -> Result<Vec<f64>, ()> {
+            let mut ret = Vec::new();
+            let key = vec![asset_codes[0].to_owned(), asset_codes[1].to_owned()];
+            let v = self.rates.get(&key);
+            if let Some(v) = v {
+                ret.push(v.0);
+                ret.push(v.1);
+            } else {
+                return Err(());
+            }
+            Ok(ret)
+        }
+    }
+
+    fn test_store(rate1: f64, rate2: f64) -> TestStore {
+        let mut rates = HashMap::new();
+        rates.insert(vec!["ABC".to_owned(), "XYZ".to_owned()], (rate1, rate2));
+        TestStore { rates }
+    }
+
+    fn test_service(
+        rate1: f64,
+        rate2: f64,
+        handler: impl OutgoingService<TestAccount> + Clone + Send + Sync,
+    ) -> ExchangeRateService<
+        TestStore,
+        impl OutgoingService<TestAccount> + Clone + Send + Sync,
+        TestAccount,
+    > {
+        let store = test_store(rate1, rate2);
+        ExchangeRateService::new(Address::from_str("example.bob").unwrap(), store, handler)
+    }
+
+}

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,37 @@ 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| {
-                let error_msg = format!("Error converting to U256 {:?}", err);
+            result(BigUint::from_str(&body.amount).map_err(move |err| {
+                let error_msg = format!("Error converting to BigUint {:?}", 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,
+                });
+
+                result(amount)
+                    .map_err(move |err| {
+                        let error_msg = format!("Error scaling amount: {:?}", err);
+                        error!("{:?}", error_msg);
+                        (StatusCode::from_u16(400).unwrap(), error_msg)
+                    })
+                    .and_then(move |amount| {
+                        // 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 +955,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 +979,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");