sui_rpc/client/v2/
transaction_execution.rs

1use futures::TryStreamExt;
2use std::fmt;
3use std::time::Duration;
4use tonic::Response;
5
6use crate::client::v2::Client;
7use crate::field::FieldMaskUtil;
8use crate::proto::sui::rpc::v2::ExecuteTransactionRequest;
9use crate::proto::sui::rpc::v2::ExecuteTransactionResponse;
10use crate::proto::sui::rpc::v2::ExecutionError;
11use crate::proto::sui::rpc::v2::GetEpochRequest;
12use crate::proto::sui::rpc::v2::SubscribeCheckpointsRequest;
13use crate::proto::TryFromProtoError;
14use prost_types::FieldMask;
15
16/// Error types that can occur when executing a transaction and waiting for checkpoint
17#[derive(Debug)]
18#[non_exhaustive]
19pub enum ExecuteAndWaitError {
20    /// RPC Error (actual tonic::Status from the client/server)
21    RpcError(tonic::Status),
22    /// Request is missing the required transaction field
23    MissingTransaction,
24    /// Failed to parse/convert the transaction for digest calculation
25    ProtoConversionError(TryFromProtoError),
26    /// Transaction executed but checkpoint wait timed out
27    CheckpointTimeout(Response<ExecuteTransactionResponse>),
28    /// Transaction executed but checkpoint stream had an error
29    CheckpointStreamError {
30        response: Response<ExecuteTransactionResponse>,
31        error: tonic::Status,
32    },
33}
34
35impl std::fmt::Display for ExecuteAndWaitError {
36    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
37        match self {
38            Self::RpcError(status) => write!(f, "RPC error: {status}"),
39            Self::MissingTransaction => {
40                write!(f, "Request is missing the required transaction field")
41            }
42            Self::ProtoConversionError(e) => write!(f, "Failed to convert transaction: {e}"),
43            Self::CheckpointTimeout(_) => {
44                write!(f, "Transaction executed but checkpoint wait timed out")
45            }
46            Self::CheckpointStreamError { error, .. } => {
47                write!(
48                    f,
49                    "Transaction executed but checkpoint stream had an error: {error}"
50                )
51            }
52        }
53    }
54}
55
56impl std::error::Error for ExecuteAndWaitError {
57    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
58        match self {
59            Self::RpcError(status) => Some(status),
60            Self::ProtoConversionError(e) => Some(e),
61            Self::CheckpointStreamError { error, .. } => Some(error),
62            Self::MissingTransaction => None,
63            Self::CheckpointTimeout(_) => None,
64        }
65    }
66}
67
68impl Client {
69    /// Executes a transaction and waits for it to be included in a checkpoint.
70    ///
71    /// This method provides "read your writes" consistency by executing the transaction
72    /// and waiting for it to appear in a checkpoint, which gauruntees indexes have been updated on
73    /// this node.
74    ///
75    /// # Arguments
76    /// * `request` - The transaction execution request (ExecuteTransactionRequest)
77    /// * `timeout` - Maximum time to wait for indexing confirmation
78    ///
79    /// # Returns
80    /// A `Result` containing the response if the transaction was executed and checkpoint confirmed,
81    /// or an error that may still include the response if execution succeeded but checkpoint
82    /// confirmation failed.
83    pub async fn execute_transaction_and_wait_for_checkpoint(
84        &mut self,
85        request: impl tonic::IntoRequest<ExecuteTransactionRequest>,
86        timeout: Duration,
87    ) -> Result<Response<ExecuteTransactionResponse>, ExecuteAndWaitError> {
88        // Subscribe to checkpoint stream before execution to avoid missing the transaction.
89        // Uses minimal read mask for efficiency since we only nee digest confirmation.
90        // Once server-side filtering is available, we should filter by transaction digest to
91        // further reduce bandwidth.
92        let mut checkpoint_stream = match self
93            .subscription_client()
94            .subscribe_checkpoints(
95                SubscribeCheckpointsRequest::default()
96                    .with_read_mask(FieldMask::from_str("transactions.digest,sequence_number")),
97            )
98            .await
99        {
100            Ok(stream) => stream.into_inner(),
101            Err(e) => return Err(ExecuteAndWaitError::RpcError(e)),
102        };
103
104        // Calculate digest from the input transaction to avoid relying on response read mask
105        let request = request.into_request();
106        let transaction = match request.get_ref().transaction_opt() {
107            Some(tx) => tx,
108            None => return Err(ExecuteAndWaitError::MissingTransaction),
109        };
110
111        let executed_txn_digest = match sui_sdk_types::Transaction::try_from(transaction) {
112            Ok(tx) => tx.digest().to_string(),
113            Err(e) => return Err(ExecuteAndWaitError::ProtoConversionError(e)),
114        };
115
116        let response = match self.execution_client().execute_transaction(request).await {
117            Ok(resp) => resp,
118            Err(e) => return Err(ExecuteAndWaitError::RpcError(e)),
119        };
120
121        // Wait for the transaction to appear in a checkpoint, at which point indexes will have been
122        // updated.
123        let timeout_future = tokio::time::sleep(timeout);
124        let checkpoint_future = async {
125            while let Some(response) = checkpoint_stream.try_next().await? {
126                let checkpoint = response.checkpoint();
127
128                for tx in checkpoint.transactions() {
129                    let digest = tx.digest();
130
131                    if digest == executed_txn_digest {
132                        return Ok(());
133                    }
134                }
135            }
136            Err(tonic::Status::aborted(
137                "checkpoint stream ended unexpectedly",
138            ))
139        };
140
141        tokio::select! {
142            result = checkpoint_future => {
143                match result {
144                    Ok(()) => Ok(response),
145                    Err(e) => Err(ExecuteAndWaitError::CheckpointStreamError { response, error: e })
146                }
147            },
148            _ = timeout_future => {
149                Err(ExecuteAndWaitError::CheckpointTimeout ( response))
150            }
151        }
152    }
153
154    /// Retrieves the current reference gas price from the latest epoch information.
155    ///
156    /// # Returns
157    /// The reference gas price as a `u64`
158    ///
159    /// # Errors
160    /// Returns an error if there is an RPC error when fetching the epoch information
161    pub async fn get_reference_gas_price(&mut self) -> Result<u64, tonic::Status> {
162        let request = GetEpochRequest::latest()
163            .with_read_mask(FieldMask::from_paths(["reference_gas_price"]));
164        let response = self.ledger_client().get_epoch(request).await?.into_inner();
165        Ok(response.epoch().reference_gas_price())
166    }
167}
168
169impl fmt::Display for ExecutionError {
170    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
171        let description = self.description.as_deref().unwrap_or("No description");
172        write!(
173            f,
174            "ExecutionError: Kind: {}, Description: {}",
175            self.kind().as_str_name(),
176            description
177        )
178    }
179}
sui_rpc/client/v2/transaction_execution.rs

sui_rpc/client/v2/
transaction_execution.rs
