fix(trade): Spawn trade executor in dedicated thread

Introduces a `Message::TradeError`, which allows to inform the app async about a failed trade execution.

CHANGELOG.md

@@ -7,14 +7,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+- Fix(validation): Show correct max counterparty collateral in validation message.
 - Fix(mobile): Calculate counterparty balance correctly when checking the validity of trade parameters.
+- Fix(trade): Spawn dedicated tokio task when executing trade
 
 ## [1.8.8] - 2024-02-14
 
 - Feat(mobile): Let user add a name to their profile for the leaderboard
 - Feat(coordinator): Allow to specify time range for leadership board
 - Fix: Prevent trade failures by ensuring that the funding transaction is not overpaying fees unexpectedly.
-- Fix(validation): Show correct max counterparty collateral in validation message.
 
 ## [1.8.7] - 2024-02-10
 

coordinator/src/node.rs

@@ -1,4 +1,5 @@
 use crate::db;
+use crate::message::OrderbookMessage;
 use crate::node::storage::NodeStorage;
 use crate::position::models::Position;
 use crate::position::models::PositionState;
@@ -29,6 +30,7 @@ use ln_dlc_node::WalletSettings;
 use rust_decimal::prelude::FromPrimitive;
 use rust_decimal::Decimal;
 use std::sync::Arc;
+use tokio::sync::mpsc;
 use tokio::sync::RwLock;
 use tracing::instrument;
 use trade::cfd::calculate_pnl;
@@ -119,10 +121,21 @@ impl Node {
         !usable_channels.is_empty()
     }
 
-    pub async fn trade(&self, params: TradeAndChannelParams) -> Result<()> {
-        let trade_executor =
-            TradeExecutor::new(self.inner.clone(), self.pool.clone(), self.settings.clone());
-        trade_executor.execute(&params).await;
+    pub async fn trade(
+        &self,
+        notifier: mpsc::Sender<OrderbookMessage>,
+        params: TradeAndChannelParams,
+    ) -> Result<()> {
+        let trade_executor = TradeExecutor::new(
+            self.inner.clone(),
+            self.pool.clone(),
+            self.settings.clone(),
+            notifier,
+        );
+
+        tokio::spawn(async move {
+            trade_executor.execute(&params).await;
+        });
 
         Ok(())
     }

coordinator/src/routes.rs

@@ -368,9 +368,13 @@ pub async fn post_trade(
     State(state): State<Arc<AppState>>,
     params: Json<TradeAndChannelParams>,
 ) -> Result<(), AppError> {
-    state.node.trade(params.0).await.map_err(|e| {
-        AppError::InternalServerError(format!("Could not handle trade request: {e:#}"))
-    })
+    state
+        .node
+        .trade(state.auth_users_notifier.clone(), params.0)
+        .await
+        .map_err(|e| {
+            AppError::InternalServerError(format!("Could not handle trade request: {e:#}"))
+        })
 }
 
 #[instrument(skip_all, err(Debug))]

coordinator/src/trade/mod.rs

@@ -1,6 +1,7 @@
 use crate::compute_relative_contracts;
 use crate::db;
 use crate::decimal_from_f32;
+use crate::message::OrderbookMessage;
 use crate::node::storage::NodeStorage;
 use crate::node::NodeSettings;
 use crate::orderbook::db::matches;
@@ -20,6 +21,7 @@ use bitcoin::hashes::hex::ToHex;
 use bitcoin::secp256k1::PublicKey;
 use commons::order_matching_fee_taker;
 use commons::MatchState;
+use commons::Message;
 use commons::OrderState;
 use commons::TradeAndChannelParams;
 use commons::TradeParams;
@@ -42,6 +44,7 @@ use rust_decimal::prelude::ToPrimitive;
 use rust_decimal::Decimal;
 use std::sync::Arc;
 use time::OffsetDateTime;
+use tokio::sync::mpsc;
 use tokio::sync::RwLock;
 use trade::cfd::calculate_long_liquidation_price;
 use trade::cfd::calculate_margin;
@@ -70,18 +73,21 @@ pub struct TradeExecutor {
     node: Arc<node::Node<CoordinatorTenTenOneStorage, NodeStorage>>,
     pool: Pool<ConnectionManager<PgConnection>>,
     settings: Arc<RwLock<NodeSettings>>,
+    notifier: mpsc::Sender<OrderbookMessage>,
 }
 
 impl TradeExecutor {
     pub fn new(
         node: Arc<node::Node<CoordinatorTenTenOneStorage, NodeStorage>>,
         pool: Pool<ConnectionManager<PgConnection>>,
         settings: Arc<RwLock<NodeSettings>>,
+        notifier: mpsc::Sender<OrderbookMessage>,
     ) -> Self {
         Self {
             node,
             pool,
             settings,
+            notifier,
         }
     }
 
@@ -112,6 +118,18 @@ impl TradeExecutor {
                 {
                     tracing::error!(%trader_id, %order_id, "Failed to update order and match: {e}");
                 };
+
+                let message = OrderbookMessage::TraderMessage {
+                    trader_id,
+                    message: Message::TradeError {
+                        order_id,
+                        error: format!("{e:#}"),
+                    },
+                    notification: None,
+                };
+                if let Err(e) = self.notifier.send(message).await {
+                    tracing::debug!("Failed to notify trader. Error: {e:#}");
+                }
             }
         };
     }

crates/commons/src/message.rs

@@ -58,6 +58,10 @@ pub enum Message {
         #[serde(with = "rust_decimal::serde::float")]
         execution_price: Decimal,
     },
+    TradeError {
+        order_id: Uuid,
+        error: String,
+    },
 }
 
 #[derive(Serialize, Clone, Deserialize, Debug)]
@@ -127,6 +131,9 @@ impl Display for Message {
             Message::CollaborativeRevert { .. } => {
                 write!(f, "LegacyCollaborativeRevert")
             }
+            Message::TradeError { .. } => {
+                write!(f, "TradeError")
+            }
         }
     }
 }

mobile/native/src/orderbook.rs

@@ -6,7 +6,10 @@ use crate::event::TaskStatus;
 use crate::health::ServiceStatus;
 use crate::ln_dlc;
 use crate::state;
+use crate::trade::order;
+use crate::trade::order::FailureReason;
 use crate::trade::position;
+use anyhow::anyhow;
 use anyhow::bail;
 use anyhow::Context;
 use anyhow::Result;
@@ -372,6 +375,14 @@ async fn handle_orderbook_message(
         msg @ Message::LimitOrderFilledMatches { .. } | msg @ Message::InvalidAuthentication(_) => {
             tracing::debug!(?msg, "Skipping message from orderbook");
         }
+        Message::TradeError { order_id, error } => {
+            order::handler::order_failed(
+                Some(order_id),
+                FailureReason::TradeResponse(error.clone()),
+                anyhow!("Coordinator failed to execute trade: {error}"),
+            )
+            .context("Could not set order to failed")?;
+        }
     };
 
     Ok(())