Chore/v2.108.0

Cargo.toml

@@ -1,12 +1,12 @@
 [package]
 name = "drift-gateway"
-version = "1.3.0"
+version = "1.3.1"
 edition = "2021"
 
 [dependencies]
 actix-web = "*"
 argh = "*"
-drift-rs = { git = "https://github.com/drift-labs/drift-rs", tag = "v1.0.0-alpha.6" }
+drift-rs = { git = "https://github.com/drift-labs/drift-rs", tag = "v1.0.0-alpha.7" }
 env_logger = "*"
 futures-util = "*"
 log = "*"

src/controller.rs

@@ -105,7 +105,7 @@ impl AppState {
     /// * `devnet` - whether to run against devnet or not
     /// * `wallet` - wallet to use for tx signing
     /// * `commitment` - Slot finalisation/commitement levels
-    /// * `default_subaccount_id` - by default all queries will use this subaccount
+    /// * `default_subaccount_id` - by default all queries will use this sub-account
     /// * `skip_tx_preflight` - submit txs without checking preflight results
     /// * `extra_rpcs` - list of additional RPC endpoints for tx submission
     pub async fn new(
@@ -180,7 +180,7 @@ impl AppState {
     ///
     /// * configured_markets - list of static markets provided by user
     ///
-    /// additional subscriptions will be included based on user's current positions (on default subaccouunt)
+    /// additional subscriptions will be included based on user's current positions (on default sub-account)
     pub(crate) async fn subscribe_market_data(&self, configured_markets: &[MarketId]) {
         let (spot, perps) = self
             .client
@@ -218,7 +218,7 @@ impl AppState {
     pub async fn get_sol_balance(&self) -> GatewayResult<SolBalanceResponse> {
         let balance = self
             .client
-            .inner()
+            .rpc()
             .get_balance(&self.wallet.inner().signer())
             .await
             .map_err(|err| ControllerError::Sdk(err.into()))?;
@@ -547,7 +547,7 @@ impl AppState {
 
         match self
             .client
-            .inner()
+            .rpc()
             .get_transaction_with_config(
                 &signature,
                 RpcTransactionConfig {
@@ -642,7 +642,7 @@ impl AppState {
         // submit to primary RPC first,
         let sig = self
             .client
-            .inner()
+            .rpc()
             .send_transaction_with_config(&tx, tx_config)
             .await
             .inspect(|s| {
@@ -659,7 +659,7 @@ impl AppState {
         // - retried at set intervals
         // - retried upto some given deadline
         // client should poll for the tx to confirm success
-        let primary_rpc = Arc::clone(&self.client);
+        let primary_rpc = Arc::clone(&self.client).rpc();
         let tx_signature = sig;
         let extra_rpcs = self.extra_rpcs.clone();
         tokio::spawn(async move {
@@ -674,11 +674,7 @@ impl AppState {
                 for rpc in extra_rpcs.iter() {
                     futs.push(rpc.send_transaction_with_config(&tx, tx_config));
                 }
-                futs.push(
-                    primary_rpc
-                        .inner()
-                        .send_transaction_with_config(&tx, tx_config),
-                );
+                futs.push(primary_rpc.send_transaction_with_config(&tx, tx_config));
 
                 while let Some(res) = futs.next().await {
                     match res {
@@ -693,11 +689,7 @@ impl AppState {
 
                 tokio::time::sleep(Duration::from_millis(400)).await;
 
-                if let Ok(Some(Ok(()))) = primary_rpc
-                    .inner()
-                    .get_signature_status(&tx_signature)
-                    .await
-                {
+                if let Ok(Some(Ok(()))) = primary_rpc.get_signature_status(&tx_signature).await {
                     confirmed = true;
                     info!(target: LOG_TARGET, "tx confirmed onchain: {tx_signature:?}");
                     break;

src/main.rs

@@ -285,7 +285,7 @@ async fn main() -> std::io::Result<()> {
     let client = Box::leak(Box::new(Arc::clone(state.client.borrow())));
     websocket::start_ws_server(
         format!("{}:{}", &config.host, config.ws_port).as_str(),
-        config.rpc_host.replace("http", "ws"),
+        client.ws(),
         state.wallet.inner().clone(),
         client.program_data(),
     )

src/websocket.rs

@@ -3,9 +3,8 @@
 use std::{collections::HashMap, ops::Neg, sync::Arc};
 
 use drift_rs::{
-    async_utils::retry_policy::{self},
     constants::ProgramData,
-    event_subscriber::{DriftEvent, EventSubscriber},
+    event_subscriber::{DriftEvent, EventSubscriber, PubsubClient},
     types::{MarketType, Order, OrderType, PositionDirection},
     Pubkey, Wallet,
 };
@@ -29,7 +28,7 @@ use crate::{
 /// Start the websocket server
 pub async fn start_ws_server(
     listen_address: &str,
-    ws_endpoint: String,
+    ws_client: Arc<PubsubClient>,
     wallet: Wallet,
     program_data: &'static ProgramData,
 ) {
@@ -42,7 +41,7 @@ pub async fn start_ws_server(
         while let Ok((stream, _)) = listener.accept().await {
             tokio::spawn(accept_connection(
                 stream,
-                ws_endpoint.clone(),
+                Arc::clone(&ws_client),
                 wallet.clone(),
                 program_data,
             ));
@@ -52,7 +51,7 @@ pub async fn start_ws_server(
 
 async fn accept_connection(
     stream: TcpStream,
-    ws_endpoint: String,
+    ws_client: Arc<PubsubClient>,
     wallet: Wallet,
     program_data: &'static ProgramData,
 ) {
@@ -102,23 +101,22 @@ async fn accept_connection(
                             }
                             info!(target: LOG_TARGET, "subscribing to events for: {}", request.sub_account_id);
 
+                            let sub_account_address =
+                                wallet.sub_account(request.sub_account_id as u16);
+                            let mut event_stream = EventSubscriber::subscribe(
+                                Arc::clone(&ws_client),
+                                sub_account_address,
+                            )
+                            .await
+                            .expect("ws connects");
+
                             let join_handle = tokio::spawn({
-                                let sub_account_address =
-                                    wallet.sub_account(request.sub_account_id as u16);
                                 let subscription_map = Arc::clone(&subscriptions);
                                 let sub_account_id = request.sub_account_id;
                                 let message_tx = message_tx.clone();
-                                let ws_endpoint = ws_endpoint.clone();
 
                                 async move {
                                     loop {
-                                        let mut event_stream = EventSubscriber::subscribe(
-                                            ws_endpoint.as_str(),
-                                            sub_account_address,
-                                            retry_policy::forever(5),
-                                        )
-                                        .await
-                                        .expect("ws connects");
                                         debug!(target: LOG_TARGET, "event stream connected: {sub_account_id:?}");
                                         while let Some(ref update) = event_stream.next().await {
                                             let (channel, data) = map_drift_event_for_account(