iota_sdk/apis/
quorum_driver.rs

1// Copyright (c) Mysten Labs, Inc.
2// Modifications Copyright (c) 2024 IOTA Stiftung
3// SPDX-License-Identifier: Apache-2.0
4
5use std::{
6    sync::Arc,
7    time::{Duration, Instant},
8};
9
10use iota_json_rpc_api::{ReadApiClient, WriteApiClient};
11use iota_json_rpc_types::{IotaTransactionBlockResponse, IotaTransactionBlockResponseOptions};
12use iota_sdk_types::TransactionDigest;
13use iota_types::{
14    quorum_driver_types::ExecuteTransactionRequestType, transaction::TransactionEnvelope,
15};
16
17use crate::{
18    RpcClient,
19    error::{Error, IotaRpcResult},
20    json_rpc_error,
21};
22
23const WAIT_FOR_LOCAL_EXECUTION_MIN_INTERVAL: Duration = Duration::from_millis(100);
24const WAIT_FOR_LOCAL_EXECUTION_MAX_INTERVAL: Duration = Duration::from_secs(2);
25
26/// Defines methods to execute transaction blocks and submit them to fullnodes.
27#[derive(Clone)]
28pub struct QuorumDriverApi {
29    api: Arc<RpcClient>,
30}
31
32impl QuorumDriverApi {
33    pub(crate) fn new(api: Arc<RpcClient>) -> Self {
34        Self { api }
35    }
36
37    /// Execute a transaction with a FullNode client.
38    ///
39    /// The request type defaults to
40    /// [`ExecuteTransactionRequestType::WaitForLocalExecution`] when
41    /// `options` require effects (see
42    /// [`IotaTransactionBlockResponseOptions::require_effects`]), and to
43    /// [`ExecuteTransactionRequestType::WaitForEffectsCert`] otherwise.
44    ///
45    /// Under `WaitForLocalExecution` the client polls the read API before
46    /// returning, whenever the node either does not confirm local execution
47    /// or fails with a transient error. If that poll times out, the call
48    /// returns whichever the node already gave it: the response, with
49    /// `confirmed_local_execution` left as the node reported it, or the
50    /// node's error.
51    ///
52    /// `checkpoint` and `timestamp_ms` are not populated on a response that
53    /// came from the execute call; only the read API sets them.
54    pub async fn execute_transaction_block(
55        &self,
56        tx: TransactionEnvelope,
57        options: IotaTransactionBlockResponseOptions,
58        request_type: impl Into<Option<ExecuteTransactionRequestType>>,
59    ) -> IotaRpcResult<IotaTransactionBlockResponse> {
60        let (tx_bytes, signatures) = tx.to_tx_bytes_and_signatures();
61        let request_type = request_type
62            .into()
63            .unwrap_or_else(|| options.default_execution_request_type());
64        let wait_for_local_execution = matches!(
65            request_type,
66            ExecuteTransactionRequestType::WaitForLocalExecution
67        );
68
69        let start = Instant::now();
70        let response = match self
71            .api
72            .http
73            .execute_transaction_block(
74                tx_bytes,
75                signatures,
76                Some(options.clone()),
77                Some(request_type.into()),
78            )
79            .await
80        {
81            Ok(response) => {
82                if !wait_for_local_execution || response.confirmed_local_execution == Some(true) {
83                    return Ok(response);
84                }
85                Ok(response)
86            }
87            Err(err) => {
88                if !wait_for_local_execution || !is_transient_error(&err) {
89                    return Err(err.into());
90                }
91                // A transient error carries no effects to fall back on, but the
92                // transaction may still land; poll for it rather than failing a
93                // call the network is going to finalize anyway.
94                Err(err)
95            }
96        };
97
98        // Both remaining cases wait for the transaction to become locally
99        // readable. A poll timeout is not itself a failure: the caller gets
100        // back whichever answer the node already gave.
101        let poll_response = self.wait_until_visible(*tx.digest(), &options, start).await;
102        match (response, poll_response) {
103            (Ok(mut response), Ok(_)) | (Err(_), Ok(mut response)) => {
104                response.confirmed_local_execution = Some(true);
105                Ok(response)
106            }
107            (Ok(response), Err(_)) => Ok(response),
108            (Err(e), Err(_)) => Err(e.into()),
109        }
110    }
111
112    /// Polls the read API until `digest` can be read back on the node that
113    /// served the request.
114    async fn wait_until_visible(
115        &self,
116        digest: TransactionDigest,
117        options: &IotaTransactionBlockResponseOptions,
118        start: Instant,
119    ) -> IotaRpcResult<IotaTransactionBlockResponse> {
120        // In simtests, fullnodes can stop receiving checkpoints for > 30s.
121        let wait_for_local_execution_timeout: Duration = if cfg!(msim) {
122            Duration::from_secs(120)
123        } else {
124            Duration::from_secs(60)
125        };
126        tokio::time::timeout(wait_for_local_execution_timeout, async {
127            let mut backoff = iota_common::backoff::ExponentialBackoff::new(
128                WAIT_FOR_LOCAL_EXECUTION_MIN_INTERVAL,
129                WAIT_FOR_LOCAL_EXECUTION_MAX_INTERVAL,
130            );
131            loop {
132                // Wait before the first request too, to leave time for the
133                // checkpoint containing the transaction to be certified,
134                // propagate to the full node, and get executed.
135                tokio::time::sleep(backoff.next().unwrap()).await;
136
137                if let Ok(poll_response) = self
138                    .api
139                    .http
140                    .get_transaction_block(digest, Some(options.clone()))
141                    .await
142                {
143                    return poll_response;
144                }
145            }
146        })
147        .await
148        .map_err(|_| Error::FailToConfirmTransactionStatus(digest, start.elapsed().as_secs()))
149    }
150}
151
152/// Whether `err` is a node-side error worth polling past rather than
153/// surfacing immediately.
154fn is_transient_error(err: &jsonrpsee::core::ClientError) -> bool {
155    match err {
156        jsonrpsee::core::ClientError::Call(object) => {
157            json_rpc_error::Error::from(jsonrpsee::core::ClientError::Call(object.clone()))
158                .is_transient_error()
159        }
160        _ => false,
161    }
162}
iota_sdk/apis/quorum_driver.rs

iota_sdk/apis/
quorum_driver.rs
