drive_abci/rpc/
core.rs

1use crate::rpc::prefetch::CorePrefetcher;
2use dpp::dashcore::ephemerealdata::chain_lock::ChainLock;
3use dpp::dashcore::{Block, BlockHash, QuorumHash, Transaction, Txid};
4use dpp::dashcore::{Header, InstantLock};
5use dpp::dashcore_rpc::dashcore_rpc_json::{
6    AssetUnlockStatusResult, ExtendedQuorumDetails, ExtendedQuorumListResult, GetChainTipsResult,
7    MasternodeListDiff, MnSyncStatus, QuorumInfoResult, QuorumType, SoftforkInfo,
8};
9use dpp::dashcore_rpc::json::GetRawTransactionResult;
10use dpp::dashcore_rpc::{Auth, Client, Error, RpcApi};
11use dpp::prelude::TimestampMillis;
12use serde_json::Value;
13use std::collections::HashMap;
14use std::time::Duration;
15
16/// Information returned by QuorumListExtended
17pub type QuorumListExtendedInfo = HashMap<QuorumHash, ExtendedQuorumDetails>;
18
19/// Core height must be of type u32 (Platform heights are u64)
20pub type CoreHeight = u32;
21/// Core RPC interface
22#[cfg_attr(any(feature = "mocks", test), mockall::automock)]
23pub trait CoreRPCLike {
24    /// Get block hash by height
25    fn get_block_hash(&self, height: CoreHeight) -> Result<BlockHash, Error>;
26
27    /// Get block hash by height
28    fn get_block_header(&self, block_hash: &BlockHash) -> Result<Header, Error>;
29
30    /// Get block time of a chain locked core height
31    fn get_block_time_from_height(&self, height: CoreHeight) -> Result<TimestampMillis, Error>;
32
33    /// Get the best chain lock
34    fn get_best_chain_lock(&self) -> Result<ChainLock, Error>;
35
36    /// Submit a chain lock
37    fn submit_chain_lock(&self, chain_lock: &ChainLock) -> Result<u32, Error>;
38
39    /// Get transaction
40    fn get_transaction(&self, tx_id: &Txid) -> Result<Transaction, Error>;
41
42    /// Get asset unlock statuses
43    fn get_asset_unlock_statuses(
44        &self,
45        indices: &[u64],
46        core_chain_locked_height: u32,
47    ) -> Result<Vec<AssetUnlockStatusResult>, Error>;
48
49    /// Get transaction
50    fn get_transaction_extended_info(&self, tx_id: &Txid)
51        -> Result<GetRawTransactionResult, Error>;
52
53    /// Get optional transaction extended info
54    /// Returns None if transaction doesn't exists
55    fn get_optional_transaction_extended_info(
56        &self,
57        transaction_id: &Txid,
58    ) -> Result<Option<GetRawTransactionResult>, Error> {
59        match self.get_transaction_extended_info(transaction_id) {
60            Ok(transaction_info) => Ok(Some(transaction_info)),
61            // Return None if transaction with specified tx id is not present
62            Err(Error::JsonRpc(dpp::dashcore_rpc::jsonrpc::error::Error::Rpc(
63                dpp::dashcore_rpc::jsonrpc::error::RpcError {
64                    code: CORE_RPC_INVALID_ADDRESS_OR_KEY,
65                    ..
66                },
67            ))) => Ok(None),
68            Err(e) => Err(e),
69        }
70    }
71
72    /// Get block by hash
73    fn get_fork_info(&self, name: &str) -> Result<Option<SoftforkInfo>, Error>;
74
75    /// Get block by hash
76    fn get_block(&self, block_hash: &BlockHash) -> Result<Block, Error>;
77
78    /// Get block by hash in JSON format
79    fn get_block_json(&self, block_hash: &BlockHash) -> Result<Value, Error>;
80
81    /// Get chain tips
82    fn get_chain_tips(&self) -> Result<GetChainTipsResult, Error>;
83
84    /// Get list of quorums by type at a given height.
85    ///
86    /// See <https://dashcore.readme.io/v19.0.0/docs/core-api-ref-remote-procedure-calls-evo#quorum-listextended>
87    fn get_quorum_listextended(
88        &self,
89        height: Option<CoreHeight>,
90    ) -> Result<ExtendedQuorumListResult, Error>;
91
92    /// Get quorum information.
93    ///
94    /// See <https://dashcore.readme.io/v19.0.0/docs/core-api-ref-remote-procedure-calls-evo#quorum-info>
95    fn get_quorum_info(
96        &self,
97        quorum_type: QuorumType,
98        hash: &QuorumHash,
99        include_secret_key_share: Option<bool>,
100    ) -> Result<QuorumInfoResult, Error>;
101
102    /// Get the difference in masternode list, return masternodes as diff elements
103    fn get_protx_diff_with_masternodes(
104        &self,
105        base_block: Option<u32>,
106        block: u32,
107    ) -> Result<MasternodeListDiff, Error>;
108
109    // /// Get the detailed information about a deterministic masternode
110    // fn get_protx_info(&self, pro_tx_hash: &ProTxHash) -> Result<ProTxInfo, Error>;
111
112    /// Verify Instant Lock signature
113    /// If `max_height` is provided the chain lock will be verified
114    /// against quorums available at this height
115    fn verify_instant_lock(
116        &self,
117        instant_lock: &InstantLock,
118        max_height: Option<u32>,
119    ) -> Result<bool, Error>;
120
121    /// Verify a chain lock signature
122    fn verify_chain_lock(&self, chain_lock: &ChainLock) -> Result<bool, Error>;
123
124    /// Returns masternode sync status
125    fn masternode_sync_status(&self) -> Result<MnSyncStatus, Error>;
126
127    /// Sends raw transaction to the network
128    fn send_raw_transaction(&self, transaction: &[u8]) -> Result<Txid, Error>;
129}
130
131#[derive(Debug)]
132/// Default implementation of Dash Core RPC using DashCoreRPC client
133pub struct DefaultCoreRPC {
134    inner: Client,
135    /// Speculative fetcher for the next core height, on its own connection.
136    /// `None` when a second connection could not be opened.
137    prefetcher: Option<CorePrefetcher>,
138}
139
140// TODO: Create errors for these error codes in dashcore_rpc
141
142/// TX is invalid due to consensus rules
143pub const CORE_RPC_TX_CONSENSUS_ERROR: i32 = -26;
144/// Tx already broadcasted and included in the chain
145pub const CORE_RPC_TX_ALREADY_IN_CHAIN: i32 = -27;
146/// Client still warming up
147pub const CORE_RPC_ERROR_IN_WARMUP: i32 = -28;
148/// Dash is not connected
149pub const CORE_RPC_CLIENT_NOT_CONNECTED: i32 = -9;
150/// Still downloading initial blocks
151pub const CORE_RPC_CLIENT_IN_INITIAL_DOWNLOAD: i32 = -10;
152/// Parse error
153pub const CORE_RPC_PARSE_ERROR: i32 = -32700;
154/// Invalid address or key
155pub const CORE_RPC_INVALID_ADDRESS_OR_KEY: i32 = -5;
156/// Invalid, missing or duplicate parameter
157pub const CORE_RPC_INVALID_PARAMETER: i32 = -8;
158
159macro_rules! retry {
160    ($action:expr) => {{
161        /// Maximum number of retry attempts
162        const MAX_RETRIES: u32 = 4;
163        /// // Multiplier for Fibonacci sequence
164        const FIB_MULTIPLIER: u64 = 1;
165
166        fn fibonacci(n: u32) -> u64 {
167            match n {
168                0 => 0,
169                1 => 1,
170                _ => fibonacci(n - 1) + fibonacci(n - 2),
171            }
172        }
173
174        let mut last_err = None;
175        let result = (0..MAX_RETRIES).find_map(|i| {
176            match $action {
177                Ok(result) => Some(Ok(result)),
178                Err(e) => {
179                    match e {
180                        dpp::dashcore_rpc::Error::JsonRpc(
181                            // Retry on transport connection error
182                            dpp::dashcore_rpc::jsonrpc::error::Error::Transport(_)
183                            | dpp::dashcore_rpc::jsonrpc::error::Error::Rpc(
184                                // Retry on Core RPC "not ready" errors
185                                dpp::dashcore_rpc::jsonrpc::error::RpcError {
186                                    code:
187                                        CORE_RPC_ERROR_IN_WARMUP
188                                        | CORE_RPC_CLIENT_NOT_CONNECTED
189                                        | CORE_RPC_CLIENT_IN_INITIAL_DOWNLOAD,
190                                    ..
191                                },
192                            ),
193                        ) => {
194                            // Delay before next try
195                            last_err = Some(e);
196                            let delay = fibonacci(i + 2) * FIB_MULTIPLIER;
197                            std::thread::sleep(Duration::from_secs(delay));
198                            None
199                        }
200                        _ => Some(Err(e)),
201                    }
202                }
203            }
204        });
205
206        result.unwrap_or_else(|| Err(last_err.unwrap()))
207    }};
208}
209
210impl DefaultCoreRPC {
211    /// Create new instance
212    pub fn open(url: &str, username: String, password: String) -> Result<Self, Error> {
213        let prefetcher = CorePrefetcher::new(url, username.clone(), password.clone());
214        if prefetcher.is_none() {
215            tracing::warn!(
216                "could not open a second Core RPC connection; masternode and quorum updates will be fetched on the critical path"
217            );
218        }
219        Ok(DefaultCoreRPC {
220            inner: Client::new(url, Auth::UserPass(username, password))?,
221            prefetcher,
222        })
223    }
224}
225
226impl CoreRPCLike for DefaultCoreRPC {
227    fn get_block_hash(&self, height: u32) -> Result<BlockHash, Error> {
228        retry!(self.inner.get_block_hash(height))
229    }
230
231    fn get_block_header(&self, block_hash: &BlockHash) -> Result<Header, Error> {
232        retry!(self.inner.get_block_header(block_hash))
233    }
234
235    fn get_block_time_from_height(&self, height: CoreHeight) -> Result<TimestampMillis, Error> {
236        let block_hash = self.get_block_hash(height)?;
237        let block_header = self.get_block_header(&block_hash)?;
238        let block_time = block_header.time as u64 * 1000;
239        Ok(block_time)
240    }
241
242    fn get_best_chain_lock(&self) -> Result<ChainLock, Error> {
243        retry!(self.inner.get_best_chain_lock())
244    }
245
246    fn submit_chain_lock(&self, chain_lock: &ChainLock) -> Result<u32, Error> {
247        retry!(self.inner.submit_chain_lock(chain_lock))
248    }
249
250    fn get_transaction(&self, tx_id: &Txid) -> Result<Transaction, Error> {
251        retry!(self.inner.get_raw_transaction(tx_id, None))
252    }
253
254    fn get_transaction_extended_info(
255        &self,
256        tx_id: &Txid,
257    ) -> Result<GetRawTransactionResult, Error> {
258        retry!(self.inner.get_raw_transaction_info(tx_id, None))
259    }
260
261    fn get_fork_info(&self, name: &str) -> Result<Option<SoftforkInfo>, Error> {
262        retry!(self
263            .inner
264            .get_blockchain_info()
265            .map(|blockchain_info| blockchain_info.softforks.get(name).cloned()))
266    }
267
268    fn get_block(&self, block_hash: &BlockHash) -> Result<Block, Error> {
269        retry!(self.inner.get_block(block_hash))
270    }
271
272    fn get_block_json(&self, block_hash: &BlockHash) -> Result<Value, Error> {
273        retry!(self.inner.get_block_json(block_hash))
274    }
275
276    fn get_chain_tips(&self) -> Result<GetChainTipsResult, Error> {
277        retry!(self.inner.get_chain_tips())
278    }
279
280    fn get_quorum_listextended(
281        &self,
282        height: Option<CoreHeight>,
283    ) -> Result<ExtendedQuorumListResult, Error> {
284        // Block sync walks core heights in order, so the next call is almost
285        // always for height + 1. Take the speculative answer when it is for the
286        // height we were asked about, and start the next guess either way. The
287        // prefetcher declines a guess past the chain lock, so at the tip this is
288        // a no-op until Core locks the next block.
289        let prefetched = height
290            .zip(self.prefetcher.as_ref())
291            .and_then(|(height, prefetcher)| prefetcher.take_quorum_list(height));
292
293        let result = match prefetched {
294            Some(list) => Ok(list),
295            None => retry!(self.inner.get_quorum_listextended_reversed(height)),
296        };
297
298        if let (Ok(_), Some(height), Some(prefetcher)) = (&result, height, self.prefetcher.as_ref())
299        {
300            prefetcher.start_quorum_list(height + 1);
301        }
302
303        result
304    }
305
306    fn get_quorum_info(
307        &self,
308        quorum_type: QuorumType,
309        hash: &QuorumHash,
310        include_secret_key_share: Option<bool>,
311    ) -> Result<QuorumInfoResult, Error> {
312        retry!(self
313            .inner
314            .get_quorum_info_reversed(quorum_type, hash, include_secret_key_share))
315    }
316
317    fn get_protx_diff_with_masternodes(
318        &self,
319        base_block: Option<u32>,
320        block: u32,
321    ) -> Result<MasternodeListDiff, Error> {
322        let base = base_block.unwrap_or(1);
323
324        // Same reasoning as get_quorum_listextended: the next diff a syncing
325        // node asks for is from this block to the one after it.
326        let prefetched = self
327            .prefetcher
328            .as_ref()
329            .and_then(|prefetcher| prefetcher.take_protx_diff(base, block));
330
331        let result = match prefetched {
332            Some(diff) => Ok(diff),
333            None => retry!(self.inner.get_protx_listdiff(base, block)),
334        };
335
336        if let (Ok(_), Some(prefetcher)) = (&result, self.prefetcher.as_ref()) {
337            prefetcher.start_protx_diff(block, block + 1);
338        }
339
340        result
341    }
342
343    /// Verify Instant Lock signature
344    /// If `max_height` is provided the chain lock will be verified
345    /// against quorums available at this height
346    fn verify_instant_lock(
347        &self,
348        instant_lock: &InstantLock,
349        max_height: Option<u32>,
350    ) -> Result<bool, Error> {
351        let request_id = instant_lock.request_id()?.to_string();
352        let transaction_id = instant_lock.txid.to_string();
353        let signature = hex::encode(instant_lock.signature);
354
355        retry!(self
356            .inner
357            .get_verifyislock(&request_id, &transaction_id, &signature, max_height))
358    }
359
360    /// Verify a chain lock signature
361    fn verify_chain_lock(&self, chain_lock: &ChainLock) -> Result<bool, Error> {
362        let block_hash = chain_lock.block_hash.to_string();
363        let signature = hex::encode(chain_lock.signature);
364
365        retry!(self.inner.get_verifychainlock(
366            block_hash.as_str(),
367            &signature,
368            Some(chain_lock.block_height)
369        ))
370    }
371
372    /// Returns masternode sync status
373    fn masternode_sync_status(&self) -> Result<MnSyncStatus, Error> {
374        retry!(self.inner.mnsync_status())
375    }
376
377    fn send_raw_transaction(&self, transaction: &[u8]) -> Result<Txid, Error> {
378        retry!(self.inner.send_raw_transaction(transaction))
379    }
380
381    fn get_asset_unlock_statuses(
382        &self,
383        indices: &[u64],
384        core_chain_locked_height: u32,
385    ) -> Result<Vec<AssetUnlockStatusResult>, Error> {
386        retry!(self
387            .inner
388            .get_asset_unlock_statuses(indices, Some(core_chain_locked_height)))
389    }
390}
drive_abci/rpc/core.rs

drive_abci/rpc/
core.rs
