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

nautilus_dydx/grpc/
client.rs
