Improve Asset Scale conversions (#192)

* fix(settlement-client): do not convert scale

The scale conversion will be performed on the engine. We must send our asset scale to the engine so that it knows how to handle the number we sent it.

* feat(settlement): Implement asset scale conversion for u256

* feat(settlement): Normalize the amount properly when being notified by the engine

* fix(eth-se): Scale the amount to settle for based on the body and engine scale

If we have configured the engine with asset scale 18, and the connector sends a body with scale 9 and amount 1, we should settle for 1e9

* test(eth-se): adjust test for proper scale conversion

* test(eth-xrp) Set ETHXRP exchange rate

* fix(crate): Remove settlement_engine_asset_scale from account

* improvement(settlement/engines): use BigUInt to handle big numbers

* fix(settlement): Convert asset scale properly

Previously the Convert trait implementation had an arithmetic error,
which was forcing us to use provide it parameters in the opposite order.

* feat(settlement/engine): Return None when idempotent data not found instead of Error

* fix(settlement): Make asset scale conversions overflow-safe

- If exchange rate normalization fails return a reject packet
- In all other places we use BigUint so the conversion should never fail,
but we still handle the error graciously
- prefer shadwing of variable name to avoid using the wrong variable
- clarify comment about SettlementEngineDetails

* improvement(exchange-rate): make the scale conversion after the exchange rate is applied

* test(exchange-rate): Add tests which verify that rate conversions fail correctly

Author: gakonst

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");