nautilus_dydx/grpc/
client.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! gRPC client implementation for dYdX v4 protocol.
17//!
18//! This module provides the main gRPC client for interacting with dYdX v4 validator nodes.
19//! It handles transaction signing, broadcasting, and querying account state.
20
21use cosmrs::Tx;
22use nautilus_network::{ConnectionUrl, url_origin};
23use tonic::transport::Channel;
24
25use crate::{
26    error::DydxError,
27    proto::{
28        AccountAuthenticator, AccountPlusClient, GetAuthenticatorsRequest,
29        cosmos_sdk_proto::{
30            cosmos::{
31                auth::v1beta1::{
32                    BaseAccount, QueryAccountRequest, query_client::QueryClient as AuthClient,
33                },
34                bank::v1beta1::{QueryAllBalancesRequest, query_client::QueryClient as BankClient},
35                base::{
36                    tendermint::v1beta1::{
37                        Block, GetLatestBlockRequest, GetNodeInfoRequest, GetNodeInfoResponse,
38                        service_client::ServiceClient as BaseClient,
39                    },
40                    v1beta1::Coin,
41                },
42                tx::v1beta1::{
43                    BroadcastMode, BroadcastTxRequest, GetTxRequest, SimulateRequest,
44                    service_client::ServiceClient as TxClient,
45                },
46            },
47            traits::Message as ProstMessage,
48        },
49        dydxprotocol::{
50            clob::{ClobPair, QueryAllClobPairRequest, query_client::QueryClient as ClobClient},
51            perpetuals::{
52                Perpetual, QueryAllPerpetualsRequest, query_client::QueryClient as PerpetualsClient,
53            },
54            subaccounts::{
55                QueryGetSubaccountRequest, Subaccount as SubaccountInfo,
56                query_client::QueryClient as SubaccountsClient,
57            },
58        },
59    },
60};
61
62/// Transaction hash type (internally uses tendermint::Hash).
63pub type TxHash = String;
64
65/// Block height.
66#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
67pub struct Height(pub u32);
68
69/// gRPC client for dYdX v4 protocol operations.
70///
71/// This client handles:
72/// - Transaction signing and broadcasting.
73/// - Account query operations.
74/// - Order placement and management via Cosmos SDK messages.
75/// - Connection management and automatic failover to fallback nodes.
76#[derive(Debug, Clone)]
77pub struct DydxGrpcClient {
78    channel: Channel,
79    auth: AuthClient<Channel>,
80    bank: BankClient<Channel>,
81    base: BaseClient<Channel>,
82    tx: TxClient<Channel>,
83    clob: ClobClient<Channel>,
84    perpetuals: PerpetualsClient<Channel>,
85    subaccounts: SubaccountsClient<Channel>,
86    accountplus: AccountPlusClient<Channel>,
87    current_url: ConnectionUrl,
88}
89
90impl DydxGrpcClient {
91    /// Create a new gRPC client with a single URL.
92    ///
93    /// # Errors
94    ///
95    /// Returns an error if the gRPC connection cannot be established.
96    pub async fn new(grpc_url: String) -> Result<Self, DydxError> {
97        let mut endpoint = Channel::from_shared(grpc_url.clone())
98            .map_err(|e| DydxError::Config(format!("Invalid gRPC URL: {e}")))?
99            .connect_timeout(std::time::Duration::from_secs(10))
100            .timeout(std::time::Duration::from_secs(30));
101
102        // Enable TLS for HTTPS URLs (required for public gRPC nodes)
103        if grpc_url.starts_with("https://") {
104            let tls = tonic::transport::ClientTlsConfig::new().with_enabled_roots();
105            endpoint = endpoint
106                .tls_config(tls)
107                .map_err(|e| DydxError::Config(format!("TLS config failed: {e}")))?;
108        }
109
110        let channel = endpoint.connect().await.map_err(|e| {
111            DydxError::Grpc(Box::new(tonic::Status::unavailable(format!(
112                "Connection failed: {e}"
113            ))))
114        })?;
115
116        Ok(Self {
117            auth: AuthClient::new(channel.clone()),
118            bank: BankClient::new(channel.clone()),
119            base: BaseClient::new(channel.clone()),
120            tx: TxClient::new(channel.clone()),
121            clob: ClobClient::new(channel.clone()),
122            perpetuals: PerpetualsClient::new(channel.clone()),
123            subaccounts: SubaccountsClient::new(channel.clone()),
124            accountplus: AccountPlusClient::new(channel.clone()),
125            channel,
126            current_url: grpc_url.into(),
127        })
128    }
129
130    /// Create a new gRPC client with fallback support.
131    ///
132    /// Attempts to connect to each URL in the provided list until a successful
133    /// connection is established. This is useful for DEX environments where nodes
134    /// can fail and fallback options are needed.
135    ///
136    /// # Errors
137    ///
138    /// Returns an error if none of the provided URLs can establish a connection.
139    pub async fn new_with_fallback(grpc_urls: &[impl AsRef<str>]) -> Result<Self, DydxError> {
140        if grpc_urls.is_empty() {
141            return Err(DydxError::Config("No gRPC URLs provided".to_string()));
142        }
143
144        let mut last_error = None;
145
146        for (idx, url) in grpc_urls.iter().enumerate() {
147            let url_str = url.as_ref();
148            log::debug!(
149                "Attempting to connect to gRPC node: {} (attempt {}/{})",
150                url_origin(url_str),
151                idx + 1,
152                grpc_urls.len()
153            );
154
155            match Self::new(url_str.to_string()).await {
156                Ok(client) => {
157                    log::debug!(
158                        "Successfully connected to gRPC node: {}",
159                        url_origin(url_str)
160                    );
161                    return Ok(client);
162                }
163                Err(e) => {
164                    log::warn!("Failed to connect to gRPC node {}", url_origin(url_str));
165                    last_error = Some(e);
166                }
167            }
168        }
169
170        Err(last_error.unwrap_or_else(|| {
171            DydxError::Grpc(Box::new(tonic::Status::unavailable(
172                "All gRPC connection attempts failed".to_string(),
173            )))
174        }))
175    }
176
177    /// Reconnect to a different gRPC node from the fallback list.
178    ///
179    /// Attempts to establish a new connection to each URL in the provided list
180    /// until successful. This is useful when the current node fails and you need
181    /// to failover to a different validator node.
182    ///
183    /// # Errors
184    ///
185    /// Returns an error if none of the provided URLs can establish a connection.
186    pub async fn reconnect_with_fallback(
187        &mut self,
188        grpc_urls: &[impl AsRef<str>],
189    ) -> Result<(), DydxError> {
190        if grpc_urls.is_empty() {
191            return Err(DydxError::Config("No gRPC URLs provided".to_string()));
192        }
193
194        let mut last_error = None;
195
196        for (idx, url) in grpc_urls.iter().enumerate() {
197            let url_str = url.as_ref();
198
199            // Skip if it's the same URL we're currently connected to
200            if url_str == self.current_url.expose() {
201                log::debug!("Skipping current URL: {}", url_origin(url_str));
202                continue;
203            }
204
205            log::debug!(
206                "Attempting to reconnect to gRPC node: {} (attempt {}/{})",
207                url_origin(url_str),
208                idx + 1,
209                grpc_urls.len()
210            );
211
212            let mut endpoint = match Channel::from_shared(url_str.to_string())
213                .map_err(|e| DydxError::Config(format!("Invalid gRPC URL: {e}")))
214            {
215                Ok(ep) => ep
216                    .connect_timeout(std::time::Duration::from_secs(10))
217                    .timeout(std::time::Duration::from_secs(30)),
218                Err(e) => {
219                    last_error = Some(e);
220                    continue;
221                }
222            };
223
224            // Enable TLS for HTTPS URLs (required for public gRPC nodes)
225            if url_str.starts_with("https://") {
226                let tls = tonic::transport::ClientTlsConfig::new().with_enabled_roots();
227                endpoint = match endpoint.tls_config(tls) {
228                    Ok(ep) => ep,
229                    Err(e) => {
230                        last_error = Some(DydxError::Config(format!("TLS config failed: {e}")));
231                        continue;
232                    }
233                };
234            }
235
236            match endpoint.connect().await {
237                Ok(connected_channel) => {
238                    log::debug!(
239                        "Successfully reconnected to gRPC node: {}",
240                        url_origin(url_str)
241                    );
242
243                    // Update all service clients with the new channel
244                    self.channel = connected_channel.clone();
245                    self.auth = AuthClient::new(connected_channel.clone());
246                    self.bank = BankClient::new(connected_channel.clone());
247                    self.base = BaseClient::new(connected_channel.clone());
248                    self.tx = TxClient::new(connected_channel.clone());
249                    self.clob = ClobClient::new(connected_channel.clone());
250                    self.perpetuals = PerpetualsClient::new(connected_channel.clone());
251                    self.subaccounts = SubaccountsClient::new(connected_channel);
252                    self.current_url = url_str.into();
253
254                    return Ok(());
255                }
256                Err(e) => {
257                    log::warn!("Failed to reconnect to gRPC node {}", url_origin(url_str));
258                    last_error = Some(DydxError::Grpc(Box::new(tonic::Status::unavailable(
259                        format!("Connection failed: {e}"),
260                    ))));
261                }
262            }
263        }
264
265        Err(last_error.unwrap_or_else(|| {
266            DydxError::Grpc(Box::new(tonic::Status::unavailable(
267                "All gRPC reconnection attempts failed".to_string(),
268            )))
269        }))
270    }
271
272    /// Get the currently connected gRPC node URL.
273    #[must_use]
274    pub fn current_url(&self) -> &str {
275        self.current_url.expose()
276    }
277
278    /// Get the underlying gRPC channel.
279    ///
280    /// This can be used to create custom gRPC service clients.
281    #[must_use]
282    pub fn channel(&self) -> &Channel {
283        &self.channel
284    }
285
286    /// Query account information for a given address.
287    ///
288    /// Returns the account number and sequence number needed for transaction signing.
289    ///
290    /// # Errors
291    ///
292    /// Returns an error if the query fails or the account does not exist.
293    pub async fn query_address(&mut self, address: &str) -> Result<(u64, u64), anyhow::Error> {
294        let req = QueryAccountRequest {
295            address: address.to_string(),
296        };
297        let resp = self
298            .auth
299            .account(req)
300            .await?
301            .into_inner()
302            .account
303            .ok_or_else(|| {
304                anyhow::anyhow!("Query account request failure, account should exist")
305            })?;
306
307        let account = BaseAccount::decode(&*resp.value)?;
308        Ok((account.account_number, account.sequence))
309    }
310
311    /// Query for [an account](https://github.com/cosmos/cosmos-sdk/tree/main/x/auth#account-1)
312    /// by its address.
313    ///
314    /// # Errors
315    ///
316    /// Returns an error if the query fails or the account does not exist.
317    pub async fn get_account(&mut self, address: &str) -> Result<BaseAccount, anyhow::Error> {
318        let req = QueryAccountRequest {
319            address: address.to_string(),
320        };
321        let resp = self
322            .auth
323            .account(req)
324            .await?
325            .into_inner()
326            .account
327            .ok_or_else(|| {
328                anyhow::anyhow!("Query account request failure, account should exist")
329            })?;
330
331        Ok(BaseAccount::decode(&*resp.value)?)
332    }
333
334    /// Query for [account balances](https://github.com/cosmos/cosmos-sdk/tree/main/x/bank#allbalances)
335    /// by address for all denominations.
336    ///
337    /// # Errors
338    ///
339    /// Returns an error if the query fails.
340    pub async fn get_account_balances(
341        &mut self,
342        address: &str,
343    ) -> Result<Vec<Coin>, anyhow::Error> {
344        let req = QueryAllBalancesRequest {
345            address: address.to_string(),
346            resolve_denom: false,
347            pagination: None,
348        };
349        let balances = self.bank.all_balances(req).await?.into_inner().balances;
350        Ok(balances)
351    }
352
353    /// Query for authenticators registered for an account.
354    ///
355    /// Authenticators enable permissioned key trading, allowing API wallets
356    /// to sign transactions on behalf of a main account.
357    ///
358    /// # Errors
359    ///
360    /// Returns an error if the query fails.
361    pub async fn get_authenticators(
362        &mut self,
363        address: &str,
364    ) -> Result<Vec<AccountAuthenticator>, anyhow::Error> {
365        let req = GetAuthenticatorsRequest {
366            account: address.to_string(),
367        };
368        let resp = self.accountplus.get_authenticators(req).await?.into_inner();
369        Ok(resp.account_authenticators)
370    }
371
372    /// Query for node info.
373    ///
374    /// # Errors
375    ///
376    /// Returns an error if the query fails.
377    pub async fn get_node_info(&mut self) -> Result<GetNodeInfoResponse, anyhow::Error> {
378        let req = GetNodeInfoRequest {};
379        let info = self.base.get_node_info(req).await?.into_inner();
380        Ok(info)
381    }
382
383    /// Query for the latest block.
384    ///
385    /// # Errors
386    ///
387    /// Returns an error if the query fails.
388    pub async fn latest_block(&mut self) -> Result<Block, anyhow::Error> {
389        let req = GetLatestBlockRequest::default();
390        let latest_block = self
391            .base
392            .get_latest_block(req)
393            .await?
394            .into_inner()
395            .sdk_block
396            .ok_or_else(|| anyhow::anyhow!("The latest block is empty"))?;
397        Ok(latest_block)
398    }
399
400    /// Query for the latest block height.
401    ///
402    /// # Errors
403    ///
404    /// Returns an error if the query fails.
405    pub async fn latest_block_height(&mut self) -> Result<Height, anyhow::Error> {
406        let latest_block = self.latest_block().await?;
407        let header = latest_block
408            .header
409            .ok_or_else(|| anyhow::anyhow!("The block doesn't contain a header"))?;
410        let height = Height(header.height.try_into()?);
411        Ok(height)
412    }
413
414    /// Query for all perpetual markets.
415    ///
416    /// # Errors
417    ///
418    /// Returns an error if the query fails.
419    pub async fn get_perpetuals(&mut self) -> Result<Vec<Perpetual>, anyhow::Error> {
420        let req = QueryAllPerpetualsRequest { pagination: None };
421        let response = self.perpetuals.all_perpetuals(req).await?.into_inner();
422        Ok(response.perpetual)
423    }
424
425    /// Query for all CLOB pairs.
426    ///
427    /// # Errors
428    ///
429    /// Returns an error if the query fails.
430    pub async fn get_clob_pairs(&mut self) -> Result<Vec<ClobPair>, anyhow::Error> {
431        let req = QueryAllClobPairRequest { pagination: None };
432        let pairs = self.clob.clob_pair_all(req).await?.into_inner().clob_pair;
433        Ok(pairs)
434    }
435
436    /// Query for subaccount information.
437    ///
438    /// # Errors
439    ///
440    /// Returns an error if the query fails.
441    pub async fn get_subaccount(
442        &mut self,
443        address: &str,
444        number: u32,
445    ) -> Result<SubaccountInfo, anyhow::Error> {
446        let req = QueryGetSubaccountRequest {
447            owner: address.to_string(),
448            number,
449        };
450        let subaccount = self
451            .subaccounts
452            .subaccount(req)
453            .await?
454            .into_inner()
455            .subaccount
456            .ok_or_else(|| {
457                anyhow::anyhow!("Subaccount query response does not contain subaccount")
458            })?;
459        Ok(subaccount)
460    }
461
462    /// Simulate a transaction to estimate gas usage.
463    ///
464    /// # Errors
465    ///
466    /// Returns an error if simulation fails.
467    pub async fn simulate_tx(&mut self, tx_bytes: Vec<u8>) -> Result<u64, anyhow::Error> {
468        let req = SimulateRequest {
469            tx_bytes,
470            ..Default::default()
471        };
472        let gas_used = self
473            .tx
474            .simulate(req)
475            .await?
476            .into_inner()
477            .gas_info
478            .ok_or_else(|| anyhow::anyhow!("Simulation response does not contain gas info"))?
479            .gas_used;
480        Ok(gas_used)
481    }
482
483    /// Broadcast a signed transaction.
484    ///
485    /// # Errors
486    ///
487    /// Returns an error if broadcasting fails.
488    pub async fn broadcast_tx(&mut self, tx_bytes: Vec<u8>) -> Result<TxHash, anyhow::Error> {
489        let req = BroadcastTxRequest {
490            tx_bytes,
491            mode: BroadcastMode::Sync as i32,
492        };
493        let response = self.tx.broadcast_tx(req).await?.into_inner();
494
495        if let Some(tx_response) = response.tx_response {
496            if tx_response.code != 0 {
497                anyhow::bail!(
498                    "Transaction broadcast failed: code={}, log={}",
499                    tx_response.code,
500                    tx_response.raw_log
501                );
502            }
503            Ok(tx_response.txhash)
504        } else {
505            Err(anyhow::anyhow!(
506                "Broadcast response does not contain tx_response"
507            ))
508        }
509    }
510
511    /// Query transaction by hash.
512    ///
513    /// # Errors
514    ///
515    /// Returns an error if the query fails.
516    pub async fn get_tx(&mut self, hash: &str) -> Result<Tx, anyhow::Error> {
517        let req = GetTxRequest {
518            hash: hash.to_string(),
519        };
520        let response = self.tx.get_tx(req).await?.into_inner();
521
522        if let Some(tx) = response.tx {
523            // Convert through bytes since the types are incompatible
524            let tx_bytes = tx.encode_to_vec();
525            Tx::try_from(tx_bytes.as_slice()).map_err(|e| anyhow::anyhow!("{e}"))
526        } else {
527            anyhow::bail!("Transaction not found")
528        }
529    }
530}
531
532#[cfg(test)]
533mod tests {
534    use rstest::rstest;
535
536    use super::*;
537
538    #[rstest]
539    fn test_height_ordering() {
540        let h1 = Height(100);
541        let h2 = Height(200);
542        assert!(h1 < h2);
543        assert_eq!(h1, Height(100));
544    }
545
546    #[tokio::test]
547    async fn test_new_with_fallback_empty_urls() {
548        let result = DydxGrpcClient::new_with_fallback(&[] as &[&str]).await;
549        assert!(result.is_err());
550        if let Err(DydxError::Config(msg)) = result {
551            assert_eq!(msg, "No gRPC URLs provided");
552        } else {
553            panic!("Expected Config error");
554        }
555    }
556
557    #[tokio::test]
558    async fn test_new_with_fallback_invalid_urls() {
559        // Use malformed URLs that fail deterministically during parsing
560        let invalid_urls = vec!["http://", "http://[::1"];
561        let result = DydxGrpcClient::new_with_fallback(&invalid_urls).await;
562
563        assert!(matches!(result, Err(DydxError::Config(_))));
564    }
565}
566
567#[cfg(test)]
568mod logging_tests {
569    use super::*;
570
571    #[rstest::rstest]
572    #[case("http://node.example.com/grpc", "http://node.example.com")]
573    #[case(
574        "https://URL_SECRET_USER:URL_SECRET_PASSWORD@node.example.com:8443/URL_SECRET_PATH?token=URL_SECRET_QUERY",
575        "https://node.example.com:8443"
576    )]
577    #[tokio::test]
578    async fn test_debug_preserves_url_origin(#[case] url: &str, #[case] origin: &str) {
579        let channel = Channel::from_static("http://127.0.0.1:1").connect_lazy();
580
581        let client = DydxGrpcClient {
582            auth: AuthClient::new(channel.clone()),
583            bank: BankClient::new(channel.clone()),
584            base: BaseClient::new(channel.clone()),
585            tx: TxClient::new(channel.clone()),
586            clob: ClobClient::new(channel.clone()),
587            perpetuals: PerpetualsClient::new(channel.clone()),
588            subaccounts: SubaccountsClient::new(channel.clone()),
589            accountplus: AccountPlusClient::new(channel.clone()),
590            channel,
591            current_url: url.into(),
592        };
593
594        let debug = format!("{client:?}");
595        assert!(debug.contains(&format!("current_url: {origin:?}")));
596        assert!(!debug.contains("URL_SECRET"));
597        assert_eq!(client.current_url.expose(), url);
598    }
599}
nautilus_dydx/grpc/client.rs

nautilus_dydx/grpc/
client.rs
