Close BTP connection if a BTP token is provided (#611)

* feat(btp): add methods to close a btp connection

* feat(btp): add timeout to outgoing requests

* feat(btp): close connection with peer if they time out

* feat(api): close BTP connections when an account is removed or modified

Author: gakonst

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

@@ -192,6 +192,8 @@ where
         .boxed();
 
     // PUT /accounts/:username
+    let btp_clone = btp.clone();
+    let outgoing_handler_clone = outgoing_handler.clone();
     let put_account = warp::put()
         .and(warp::path("accounts"))
         .and(account_username_to_id.clone())
@@ -200,8 +202,15 @@ where
         .and(deserialize_json()) // warp::body::json() is not able to decode this!
         .and(with_store.clone())
         .and_then(move |id: Uuid, account_details: AccountDetails, store: S| {
-            let outgoing_handler = outgoing_handler.clone();
-            let btp = btp.clone();
+            let outgoing_handler = outgoing_handler_clone.clone();
+            let btp = btp_clone.clone();
+            if account_details.ilp_over_btp_incoming_token.is_some() {
+                // if the BTP token was provided, assume that it's different
+                // from the existing one and drop the connection
+                // the saved websocket connection
+                // a new one will be initialized in the `connect_to_external_services` call
+                btp.close_connection(&id);
+            }
             async move {
                 let account = store.update_account(id, account_details).await?;
                 connect_to_external_services(outgoing_handler, account.clone(), store, btp).await?;
@@ -253,29 +262,56 @@ where
         .boxed();
 
     // DELETE /accounts/:username
+    let btp_clone = btp.clone();
     let delete_account = warp::delete()
         .and(warp::path("accounts"))
         .and(account_username_to_id.clone())
         .and(warp::path::end())
         .and(admin_only)
         .and(with_store.clone())
-        .and_then(|id: Uuid, store: S| async move {
-            let account = store.delete_account(id).await?;
-            Ok::<Json, Rejection>(warp::reply::json(&account))
+        .and_then(move |id: Uuid, store: S| {
+            let btp = btp_clone.clone();
+            async move {
+                let account = store.delete_account(id).await?;
+                // close the btp connection (if any)
+                btp.close_connection(&id);
+                Ok::<Json, Rejection>(warp::reply::json(&account))
+            }
         })
         .boxed();
 
     // PUT /accounts/:username/settings
+    let outgoing_handler_clone = outgoing_handler;
     let put_account_settings = warp::put()
         .and(warp::path("accounts"))
         .and(admin_or_authorized_user_only.clone())
         .and(warp::path("settings"))
         .and(warp::path::end())
         .and(deserialize_json())
         .and(with_store.clone())
-        .and_then(|id: Uuid, settings: AccountSettings, store: S| async move {
-            let modified_account = store.modify_account_settings(id, settings).await?;
-            Ok::<Json, Rejection>(warp::reply::json(&modified_account))
+        .and_then(move |id: Uuid, settings: AccountSettings, store: S| {
+            let btp = btp.clone();
+            let outgoing_handler = outgoing_handler_clone.clone();
+            async move {
+                if settings.ilp_over_btp_incoming_token.is_some() {
+                    // if the BTP token was provided, assume that it's different
+                    // from the existing one and drop the connection
+                    // the saved websocket connection
+                    btp.close_connection(&id);
+                }
+                let modified_account = store.modify_account_settings(id, settings).await?;
+
+                // Since the account was modified, we should also try to
+                // connect to the new account:
+                connect_to_external_services(
+                    outgoing_handler,
+                    modified_account.clone(),
+                    store,
+                    btp,
+                )
+                .await?;
+                Ok::<Json, Rejection>(warp::reply::json(&modified_account))
+            }
         })
         .boxed();
 

crates/interledger-btp/src/lib.rs

@@ -166,9 +166,10 @@ mod client_server {
     async fn client_server_test() {
         let bind_addr = get_open_port();
 
+        let server_acc_id = Uuid::new_v4();
         let server_store = TestStore {
             accounts: Arc::new(vec![TestAccount {
-                id: Uuid::new_v4(),
+                id: server_acc_id,
                 ilp_over_btp_incoming_token: Some("test_auth_token".to_string()),
                 ilp_over_btp_outgoing_token: None,
                 ilp_over_btp_url: None,
@@ -260,6 +261,48 @@ mod client_server {
             })
             .await;
         assert!(res.is_ok());
+
+        btp_service.close_connection(&server_acc_id);
+        // after removing the connection this will fail
+        let mut btp_client_clone = btp_client.clone();
+        let res = btp_client_clone
+            .send_request(OutgoingRequest {
+                from: account.clone(),
+                to: account.clone(),
+                original_amount: 100,
+                prepare: PrepareBuilder {
+                    destination: Address::from_str("example.destination").unwrap(),
+                    amount: 100,
+                    execution_condition: &[0; 32],
+                    expires_at: SystemTime::now() + Duration::from_secs(30),
+                    data: b"test data",
+                }
+                .build(),
+            })
+            .await
+            .unwrap_err();
+        assert_eq!(res.code(), ErrorCode::R00_TRANSFER_TIMED_OUT);
+
+        // now that we have timed out, if we try sending again we'll see that we
+        // have no more connections with this user
+        let res = btp_client_clone
+            .send_request(OutgoingRequest {
+                from: account.clone(),
+                to: account.clone(),
+                original_amount: 100,
+                prepare: PrepareBuilder {
+                    destination: Address::from_str("example.destination").unwrap(),
+                    amount: 100,
+                    execution_condition: &[0; 32],
+                    expires_at: SystemTime::now() + Duration::from_secs(30),
+                    data: b"test data",
+                }
+                .build(),
+            })
+            .await
+            .unwrap_err();
+        assert_eq!(res.code(), ErrorCode::F02_UNREACHABLE);
+
         btp_service.close();
     }
 }

crates/interledger-btp/src/service.rs

@@ -26,6 +26,11 @@ const PING_INTERVAL: u64 = 30; // seconds
 static PING: Lazy<Message> = Lazy::new(|| Message::Ping(Vec::with_capacity(0)));
 static PONG: Lazy<Message> = Lazy::new(|| Message::Pong(Vec::with_capacity(0)));
 
+// Return a Reject timeout if the outgoing message future does not complete
+// within this timeout. This will probably happen if the peer closed the websocket
+// with us
+const SEND_MSG_TIMEOUT: Duration = Duration::from_secs(30);
+
 type IlpResultChannel = oneshot::Sender<Result<Fulfill, Reject>>;
 type IncomingRequestBuffer<A> = UnboundedReceiver<(A, u32, Prepare)>;
 
@@ -136,6 +141,11 @@ where
         }
     }
 
+    /// Deletes the websocket associated with the provided `account_id`
+    pub fn close_connection(&self, account_id: &Uuid) {
+        self.connections.write().remove(account_id);
+    }
+
     /// Close all of the open WebSocket connections
     // TODO is there some more automatic way of knowing when we should close the connections?
     // The problem is that the WS client can be a server too, so it's not clear when we are done with it
@@ -323,7 +333,30 @@ where
                 Ok(_) => {
                     let (sender, receiver) = oneshot::channel();
                     (*self.pending_outgoing.lock()).insert(request_id, sender);
-                    let result = receiver.await;
+
+                    // Wrap the receiver with a timeout to ensure we do not
+                    // wait too long if the other party has disconnected
+                    let result = tokio::time::timeout(SEND_MSG_TIMEOUT, receiver).await;
+
+                    let result = match result {
+                        Ok(packet) => packet,
+                        Err(err) => {
+                            error!("Request timed out. Did the peer disconnect? Err: {}", err);
+                            // Assume that such a long timeout means that the peer closed their
+                            // connection with us, so we'll remove the pending request and the websocket
+                            (*self.pending_outgoing.lock()).remove(&request_id);
+                            self.close_connection(&request.to.id());
+
+                            return Err(RejectBuilder {
+                                code: ErrorCode::R00_TRANSFER_TIMED_OUT,
+                                message: &[],
+                                triggered_by: Some(&ilp_address),
+                                data: &[],
+                            }
+                            .build());
+                        }
+                    };
+
                     // Drop the trigger here since we've gotten the response
                     // and don't need to keep the connections open if this was the
                     // last thing we were waiting for
@@ -390,6 +423,10 @@ where
     pub fn close(&self) {
         self.outgoing.close();
     }
+
+    pub fn close_connection(&self, account_id: &Uuid) {
+        self.outgoing.close_connection(account_id);
+    }
 }
 
 #[async_trait]