feat: async stream sender with rate enforcement

Co-Authored-By: Georgios Konstantopoulos <me@gakonst.com>

crates/ilp-node/src/node.rs

@@ -474,7 +474,7 @@ impl InterledgerNode {
             bytes05::Bytes::copy_from_slice(secret_seed.as_ref()),
             admin_auth_token,
             store.clone(),
-            incoming_service_api.clone(),
+            incoming_service_api,
             outgoing_service.clone(),
             btp.clone(), // btp client service!
         );

crates/ilp-node/tests/redis/test_helpers.rs

@@ -85,6 +85,7 @@ pub async fn send_money_to_username<T: Display + Debug>(
         .json(&json!({
             "receiver": format!("http://localhost:{}/accounts/{}/spsp", to_port, to_username),
             "source_amount": amount,
+            "slippage": 0.025 // allow up to 2.5% slippage
         }))
         .send()
         .map_err(|_| ())

crates/ilp-node/tests/redis/three_nodes.rs

@@ -106,6 +106,7 @@ async fn three_nodes() {
         "route_broadcast_interval": Some(200),
         "exchange_rate": {
             "poll_interval": 60000,
+            "spread": 0.02, // take a 2% spread
         },
     }))
     .expect("Error creating node2.");
@@ -119,7 +120,7 @@ async fn three_nodes() {
         "secret_seed": random_secret(),
         "route_broadcast_interval": Some(200),
         "exchange_rate": {
-            "poll_interval": 60000,
+            "poll_interval": 60000
         },
     }))
     .expect("Error creating node3.");
@@ -131,6 +132,14 @@ async fn three_nodes() {
     create_account_on_node(node1_http, bob_on_alice, "admin")
         .await
         .unwrap();
+    let client = reqwest::Client::new();
+    client
+        .put(&format!("http://localhost:{}/rates", node1_http))
+        .header("Authorization", "Bearer admin")
+        .json(&json!({"ABC": 1, "XYZ": 2.01}))
+        .send()
+        .await
+        .unwrap();
 
     node2.serve().await.unwrap();
     create_account_on_node(node2_http, alice_on_bob, "admin")
@@ -139,8 +148,6 @@ async fn three_nodes() {
     create_account_on_node(node2_http, charlie_on_bob, "admin")
         .await
         .unwrap();
-    // Also set exchange rates
-    let client = reqwest::Client::new();
     client
         .put(&format!("http://localhost:{}/rates", node2_http))
         .header("Authorization", "Bearer admin")
@@ -156,6 +163,13 @@ async fn three_nodes() {
     create_account_on_node(node3_http, bob_on_charlie, "admin")
         .await
         .unwrap();
+    client
+        .put(&format!("http://localhost:{}/rates", node3_http))
+        .header("Authorization", "Bearer admin")
+        .json(&json!({"ABC": 1, "XYZ": 2}))
+        .send()
+        .await
+        .unwrap();
 
     delay(1000).await;
 
@@ -167,14 +181,16 @@ async fn three_nodes() {
         ])
     };
 
-    // Node 1 sends 1000 to Node 3. However, Node1's scale is 9,
+    // Node 1 sends 1,000,000 to Node 3. However, Node1's scale is 9,
     // while Node 3's scale is 6. This means that Node 3 will
-    // see 1000x less. In addition, the conversion rate is 2:1
-    // for 3's asset, so he will receive 2 total.
+    // see 1000x less. Node 2's rate is 2:1, minus a 2% spread.
+    // Node 1 has however set the rate to 2.01:1. The payment suceeds because
+    // the end delivered result to Node 3 is 1960 which is slightly
+    // over 0.975 * 2010, which is within the 2.5% slippage set by the sender.
     let receipt = send_money_to_username(
         node1_http,
         node3_http,
-        1000,
+        1_000_000,
         "charlie_on_c",
         "alice_on_a",
         "default account holder",
@@ -191,40 +207,45 @@ async fn three_nodes() {
         .to
         .to_string()
         .starts_with("example.bob.charlie_on_b.charlie_on_c."));
-    assert_eq!(receipt.sent_asset_code, "XYZ");
-    assert_eq!(receipt.sent_asset_scale, 9);
-    assert_eq!(receipt.sent_amount, 1000);
-    assert_eq!(receipt.delivered_asset_code.unwrap(), "ABC");
-    assert_eq!(receipt.delivered_amount, 2);
-    assert_eq!(receipt.delivered_asset_scale.unwrap(), 6);
+    assert_eq!(receipt.source_asset_code, "XYZ");
+    assert_eq!(receipt.source_asset_scale, 9);
+    assert_eq!(receipt.source_amount, 1_000_000);
+    assert_eq!(receipt.sent_amount, 1_000_000);
+    assert_eq!(receipt.in_flight_amount, 0);
+    assert_eq!(receipt.delivered_amount, 1960);
+    assert_eq!(receipt.destination_asset_code.unwrap(), "ABC");
+    assert_eq!(receipt.destination_asset_scale.unwrap(), 6);
     let ret = get_balances().await;
     let ret: Vec<_> = ret.into_iter().map(|r| r.unwrap()).collect();
-    // -1000 divided by asset scale 9
+    // -1000000 divided by asset scale 9
     assert_eq!(
         ret[0],
         BalanceData {
             asset_code: "XYZ".to_owned(),
-            balance: -1e-6
+            balance: -1_000_000.0 / 1e9
         }
     );
-    // 2 divided by asset scale 6
+    // 1960 divided by asset scale 6
     assert_eq!(
         ret[1],
         BalanceData {
             asset_code: "ABC".to_owned(),
-            balance: 2e-6
+            balance: 1960.0 / 1e6
         }
     );
-    // 2 divided by asset scale 6
+    // 1960 divided by asset scale 6
     assert_eq!(
         ret[2],
         BalanceData {
             asset_code: "ABC".to_owned(),
-            balance: 2e-6
+            balance: 1960.0 / 1e6
         }
     );
 
-    // Charlie sends to Alice
+    // Node 3 sends 1,000 to Node 1. However, Node 1's scale is 9,
+    // while Node 3's scale is 6. This means that Node 1 will
+    // see 1000x more. Node 2's rate is 1:2, minus a 2% spread.
+    // Node 3 should receive 495,500 units total.
     let receipt = send_money_to_username(
         node3_http,
         node1_http,
@@ -242,36 +263,38 @@ async fn three_nodes() {
         "Payment receipt incorrect (2)"
     );
     assert!(receipt.to.to_string().starts_with("example.alice"));
-    assert_eq!(receipt.sent_asset_code, "ABC");
-    assert_eq!(receipt.sent_asset_scale, 6);
+    assert_eq!(receipt.source_asset_code, "ABC");
+    assert_eq!(receipt.source_asset_scale, 6);
+    assert_eq!(receipt.source_amount, 1000);
     assert_eq!(receipt.sent_amount, 1000);
-    assert_eq!(receipt.delivered_asset_code.unwrap(), "XYZ");
-    assert_eq!(receipt.delivered_amount, 500_000);
-    assert_eq!(receipt.delivered_asset_scale.unwrap(), 9);
+    assert_eq!(receipt.in_flight_amount, 0);
+    assert_eq!(receipt.delivered_amount, 490_000);
+    assert_eq!(receipt.destination_asset_code.unwrap(), "XYZ");
+    assert_eq!(receipt.destination_asset_scale.unwrap(), 9);
     let ret = get_balances().await;
     let ret: Vec<_> = ret.into_iter().map(|r| r.unwrap()).collect();
-    // 499,000 divided by asset scale 9
+    // (490,000 - 1,000,000) divided by asset scale 9
     assert_eq!(
         ret[0],
         BalanceData {
             asset_code: "XYZ".to_owned(),
-            balance: 499e-6
+            balance: (490_000.0 - 1_000_000.0) / 1e9
         }
     );
-    // -998 divided by asset scale 6
+    // (1,960 - 1,000) divided by asset scale 6
     assert_eq!(
         ret[1],
         BalanceData {
             asset_code: "ABC".to_owned(),
-            balance: -998e-6
+            balance: (1960.0 - 1000.0) / 1e6
         }
     );
-    // -998 divided by asset scale 6
+    // (1,960 - 1,000) divided by asset scale 6
     assert_eq!(
         ret[2],
         BalanceData {
             asset_code: "ABC".to_owned(),
-            balance: -998e-6
+            balance: (1960.0 - 1000.0) / 1e6
         }
     );
 }

crates/interledger-api/src/routes/accounts.rs

@@ -28,11 +28,20 @@ use warp::{self, reply::Json, Filter, Rejection};
 
 pub const BEARER_TOKEN_START: usize = 7;
 
+const fn get_default_max_slippage() -> f64 {
+    0.01
+}
+
 #[derive(Deserialize, Debug)]
 struct SpspPayRequest {
     receiver: String,
     #[serde(deserialize_with = "number_or_string")]
     source_amount: u64,
+    #[serde(
+        deserialize_with = "number_or_string",
+        default = "get_default_max_slippage"
+    )]
+    slippage: f64,
 }
 
 pub fn accounts_api<I, O, S, A, B>(
@@ -339,14 +348,17 @@ where
         .and(warp::path::end())
         .and(deserialize_json())
         .and(with_incoming_handler)
+        .and(with_store.clone())
         .and_then(
-            move |account: A, pay_request: SpspPayRequest, incoming_handler: I| {
+            move |account: A, pay_request: SpspPayRequest, incoming_handler: I, store: S| {
                 async move {
                     let receipt = pay(
                         incoming_handler,
                         account.clone(),
+                        store,
                         &pay_request.receiver,
                         pay_request.source_amount,
+                        pay_request.slippage,
                     )
                     .map_err(|err| {
                         let msg = format!("Error sending SPSP payment: {}", err);

crates/interledger-api/src/routes/node_settings.rs

@@ -7,7 +7,6 @@ use interledger_packet::Address;
 use interledger_rates::ExchangeRateStore;
 use interledger_router::RouterStore;
 use interledger_service::{Account, AccountStore, AddressStore, Username};
-use interledger_service_util::ExchangeRateStore;
 use interledger_settlement::core::{types::SettlementAccount, SettlementClient};
 use log::{error, trace};
 use secrecy::{ExposeSecret, SecretString};

crates/interledger-rates/Cargo.toml

@@ -1,12 +1,11 @@
 [package]
 name = "interledger-rates"
 version = "0.4.0"
-authors = ["Kincaid O'Neil <kincaidoneil@users.noreply.github.com>"]
+authors = ["Evan Schwartz <evan@ripple.com>"]
+description = "Exchange rate utilities"
+license = "Apache-2.0"
 edition = "2018"
-
-# TODO Add other metadata here!
-
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+repository = "https://github.com/interledger-rs/interledger-rs"
 
 [dependencies]
 async-trait = "0.1.22"

crates/interledger-rates/src/lib.rs

@@ -61,7 +61,7 @@ pub struct ExchangeRateFetcher<S> {
     consecutive_failed_polls: Arc<AtomicU32>,
     failed_polls_before_invalidation: u32,
     store: S,
-    client: Client, // TODO What is a Client?
+    client: Client,
 }
 
 impl<S> ExchangeRateFetcher<S>

crates/interledger-spsp/Cargo.toml

@@ -9,6 +9,7 @@ repository = "https://github.com/interledger-rs/interledger-rs"
 
 [dependencies]
 interledger-packet = { path = "../interledger-packet", version = "^0.4.0", features = ["serde"], default-features = false }
+interledger-rates = { path = "../interledger-rates", version = "^0.4.0", default-features = false }
 interledger-service = { path = "../interledger-service", version = "^0.4.0", default-features = false }
 interledger-stream = { path = "../interledger-stream", version = "^0.4.0", default-features = false }