Make state commitment and tx commitment configurable

README.md

@@ -41,9 +41,21 @@ Options:
   --dev             run in devnet mode
   --host            gateway host address
   --port            gateway port
-  --delegate        use delegated signing mode, provide the delegator pubkey
+  --ws-port         gateway Ws port
+  --delegate        use delegated signing mode, provide the delegators pubkey
   --emulate         run the gateway in read-only mode for given authority pubkey
-  --help            display usage information
+  --tx-commitment   solana commitment level to use for transaction confirmation
+                    (default: confirmed)
+  --commitment      solana commitment level to use for state updates (default:
+                    confirmed)
+  --verbose         enable debug logging
+```
+
+## Logging
+use env var to enable logging
+```bash
+# enable logging
+RUST_LOG=gateway=debug drift-gateway ..
 ```
 
 ## API Examples

src/controller.rs

@@ -3,6 +3,7 @@ use std::{borrow::Cow, sync::Arc};
 use drift_sdk::{
     constants::{ProgramData, BASE_PRECISION},
     dlob::DLOBClient,
+    event_subscriber::CommitmentConfig,
     types::{
         Context, MarketId, MarketType, ModifyOrderParams, RpcSendTransactionConfig, SdkError,
         SdkResult, VersionedMessage,
@@ -41,6 +42,8 @@ pub struct AppState {
     pub wallet: Wallet,
     pub client: Arc<DriftClient<WsAccountProvider>>,
     dlob_client: DLOBClient,
+    /// Solana tx commitment level for preflight confirmation
+    tx_commitment: CommitmentConfig,
 }
 
 impl AppState {
@@ -55,14 +58,23 @@ impl AppState {
     pub fn default_sub_account(&self) -> Pubkey {
         self.wallet.default_sub_account()
     }
-    pub async fn new(endpoint: &str, devnet: bool, wallet: Wallet) -> Self {
+    pub async fn new(
+        endpoint: &str,
+        devnet: bool,
+        wallet: Wallet,
+        commitment: Option<(CommitmentConfig, CommitmentConfig)>,
+    ) -> Self {
+        let (state_commitment, tx_commitment) =
+            commitment.unwrap_or((CommitmentConfig::confirmed(), CommitmentConfig::confirmed()));
         let context = if devnet {
             Context::DevNet
         } else {
             Context::MainNet
         };
 
-        let account_provider = WsAccountProvider::new(endpoint).await.expect("ws connects");
+        let account_provider = WsAccountProvider::new_with_commitment(endpoint, state_commitment)
+            .await
+            .expect("ws connects");
         let client = DriftClient::new(context, account_provider)
             .await
             .expect("ok");
@@ -77,6 +89,7 @@ impl AppState {
             wallet,
             client: Arc::new(client),
             dlob_client: DLOBClient::new(dlob_endpoint),
+            tx_commitment,
         }
     }
 
@@ -317,6 +330,7 @@ impl AppState {
                 tx,
                 RpcSendTransactionConfig {
                     max_retries: Some(0),
+                    preflight_commitment: Some(self.tx_commitment.commitment),
                     ..Default::default()
                 },
             )

src/main.rs

@@ -9,7 +9,7 @@ use argh::FromArgs;
 use log::{debug, info, warn};
 
 use controller::{create_wallet, AppState, ControllerError};
-use drift_sdk::Pubkey;
+use drift_sdk::{types::CommitmentConfig, Pubkey};
 use serde_json::json;
 use std::{borrow::Borrow, str::FromStr, sync::Arc};
 use types::{
@@ -153,7 +153,14 @@ async fn get_sol_balance(controller: web::Data<AppState>) -> impl Responder {
 #[actix_web::main]
 async fn main() -> std::io::Result<()> {
     let config: GatewayConfig = argh::from_env();
-    env_logger::Builder::from_default_env().init();
+    let log_level = if config.verbose {
+        log::LevelFilter::Debug
+    } else {
+        log::LevelFilter::Info
+    };
+    env_logger::Builder::from_default_env()
+        .filter_module(LOG_TARGET, log_level)
+        .init();
     let secret_key = std::env::var("DRIFT_GATEWAY_KEY");
     let delegate = config
         .delegate
@@ -162,22 +169,36 @@ async fn main() -> std::io::Result<()> {
         .emulate
         .map(|ref x| Pubkey::from_str(x).expect("valid pubkey"));
     let wallet = create_wallet(secret_key.ok(), emulate, delegate);
-    let state = AppState::new(&config.rpc_host, config.dev, wallet).await;
+    let state_commitment = CommitmentConfig::from_str(&config.commitment)
+        .expect("one of: processed | confirmed | finalized");
+    let tx_commitment = CommitmentConfig::from_str(&config.tx_commitment)
+        .expect("one of: processed | confirmed | finalized");
+
+    let state = AppState::new(
+        &config.rpc_host,
+        config.dev,
+        wallet,
+        Some((state_commitment, tx_commitment)),
+    )
+    .await;
 
     info!(
+        target: LOG_TARGET,
         "🏛️ gateway listening at http://{}:{}",
         config.host, config.port
     );
 
     if delegate.is_some() {
         info!(
+            target: LOG_TARGET,
             "🪪: authority: {:?}, default sub-account: {:?}, 🔑 delegate: {:?}",
             state.authority(),
             state.default_sub_account(),
             state.signer(),
         );
     } else {
         info!(
+            target: LOG_TARGET,
             "🪪: authority: {:?}, default sub-account: {:?}",
             state.authority(),
             state.default_sub_account()
@@ -189,7 +210,7 @@ async fn main() -> std::io::Result<()> {
 
     let client = Box::leak(Box::new(Arc::clone(state.client.borrow())));
     websocket::start_ws_server(
-        format!("{}:1337", &config.host).as_str(),
+        format!("{}:{}", &config.host, config.ws_port).as_str(),
         config.rpc_host.replace("http", "ws"),
         state.wallet.clone(),
         client.program_data(),
@@ -198,7 +219,7 @@ async fn main() -> std::io::Result<()> {
 
     HttpServer::new(move || {
         App::new()
-            .wrap(Logger::new("%a | %s | %r | (%Dms)").log_target("gateway"))
+            .wrap(Logger::new("%a | %s | %r | (%Dms)").log_target(LOG_TARGET))
             .app_data(web::Data::new(state.clone()))
             .service(
                 web::scope("/v2")
@@ -218,29 +239,6 @@ async fn main() -> std::io::Result<()> {
     .await
 }
 
-#[derive(FromArgs)]
-/// Drift gateway server
-struct GatewayConfig {
-    /// the solana RPC URL
-    #[argh(positional)]
-    rpc_host: String,
-    /// run in devnet mode
-    #[argh(switch)]
-    dev: bool,
-    /// gateway host address
-    #[argh(option, default = "String::from(\"127.0.0.1\")")]
-    host: String,
-    /// gateway port
-    #[argh(option, default = "8080")]
-    port: u16,
-    /// use delegated signing mode, provide the delegator's pubkey
-    #[argh(option)]
-    delegate: Option<String>,
-    /// run the gateway in read-only mode for given authority pubkey
-    #[argh(option)]
-    emulate: Option<String>,
-}
-
 fn handle_result<T: std::fmt::Debug>(
     result: Result<T, ControllerError>,
 ) -> Either<HttpResponse, Json<T>> {
@@ -283,6 +281,41 @@ fn handle_deser_error<T>(err: serde_json::Error) -> Either<HttpResponse, Json<T>
     )))
 }
 
+#[derive(FromArgs)]
+/// Drift gateway server
+struct GatewayConfig {
+    /// the solana RPC URL
+    #[argh(positional)]
+    rpc_host: String,
+    /// run in devnet mode
+    #[argh(switch)]
+    dev: bool,
+    /// gateway host address
+    #[argh(option, default = "String::from(\"127.0.0.1\")")]
+    host: String,
+    /// gateway port
+    #[argh(option, default = "8080")]
+    port: u16,
+    /// gateway Ws port
+    #[argh(option, default = "1337")]
+    ws_port: u16,
+    /// use delegated signing mode, provide the delegator's pubkey
+    #[argh(option)]
+    delegate: Option<String>,
+    /// run the gateway in read-only mode for given authority pubkey
+    #[argh(option)]
+    emulate: Option<String>,
+    /// solana commitment level to use for transaction confirmation (default: confirmed)
+    #[argh(option, default = "String::from(\"confirmed\")")]
+    tx_commitment: String,
+    /// solana commitment level to use for state updates (default: confirmed)
+    #[argh(option, default = "String::from(\"confirmed\")")]
+    commitment: String,
+    #[argh(switch)]
+    /// enable debug logging
+    verbose: bool,
+}
+
 #[cfg(test)]
 mod tests {
     use actix_web::{http::Method, test, App};
@@ -303,7 +336,7 @@ mod tests {
 
     async fn setup_controller() -> AppState {
         let wallet = create_wallet(Some(get_seed()), None, None);
-        AppState::new(TEST_ENDPOINT, true, wallet).await
+        AppState::new(TEST_ENDPOINT, true, wallet, None).await
     }
 
     #[actix_web::test]

src/websocket.rs

@@ -10,7 +10,7 @@ use drift_sdk::{
     Pubkey, Wallet,
 };
 use futures_util::{SinkExt, StreamExt};
-use log::{debug, info};
+use log::info;
 use rust_decimal::Decimal;
 use serde::{Deserialize, Serialize, Serializer};
 use serde_json::json;
@@ -20,7 +20,10 @@ use tokio::{
 };
 use tokio_tungstenite::{accept_async, tungstenite::Message};
 
-use crate::types::{get_market_decimals, Market, PRICE_DECIMALS};
+use crate::{
+    types::{get_market_decimals, Market, PRICE_DECIMALS},
+    LOG_TARGET,
+};
 
 /// Start the websocket server
 pub async fn start_ws_server(
@@ -84,7 +87,7 @@ async fn accept_connection(
                                 // no double subs
                                 return;
                             }
-                            debug!("subscribing to events for: {}", request.sub_account_id);
+                            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);
@@ -125,7 +128,7 @@ async fn accept_connection(
                             stream_handle = Some(join_handle);
                         }
                         Method::Unsubscribe => {
-                            debug!("unsubscribing: {}", request.sub_account_id);
+                            info!(target: LOG_TARGET, "unsubscribing: {}", request.sub_account_id);
                             // TODO: support ending by channel, this ends all channels
                             if let Some(task) = stream_handle.take() {
                                 task.abort();