sui-rpc: answer duplicate executions from the ledger

crates/sui-rpc/src/client/transaction_execution.rs

@@ -6,6 +6,7 @@ use crate::proto::sui::rpc::v2::ExecuteTransactionResponse;
 use crate::proto::sui::rpc::v2::ExecutionError;
 use crate::proto::sui::rpc::v2::GetEpochRequest;
 use crate::proto::sui::rpc::v2::GetTransactionRequest;
+use crate::proto::sui::rpc::v2::GetTransactionResponse;
 use crate::proto::sui::rpc::v2::SubscribeCheckpointsRequest;
 use futures::TryStreamExt;
 use prost_types::FieldMask;
@@ -80,6 +81,19 @@ impl Client {
     /// A `Result` containing the response if the transaction was executed and checkpoint confirmed,
     /// or an error that may still include the response if execution succeeded but checkpoint
     /// confirmation failed.
+    ///
+    /// # Duplicate submissions
+    /// Submitting a transaction that has already been executed is handled
+    /// gracefully. While the execution RPC is in flight the ledger is probed
+    /// for the transaction, and a transaction that is already in a checkpoint
+    /// is returned without waiting for execution to finish. Likewise, if
+    /// execution fails but the ledger shows the transaction in a checkpoint
+    /// (for example when a resubmission races the original submission), the
+    /// execution error is discarded and the committed transaction is
+    /// returned. In both cases the response is assembled from
+    /// `GetTransaction` using the request's read mask, so it carries the same
+    /// fields an execution response would, with `digest`, `checkpoint`, and
+    /// `timestamp` always populated.
     pub async fn execute_transaction_and_wait_for_checkpoint(
         &mut self,
         request: impl tonic::IntoRequest<ExecuteTransactionRequest>,
@@ -97,6 +111,27 @@ impl Client {
             Err(e) => return Err(ExecuteAndWaitError::ProtoConversionError(e)),
         };
 
+        // Read mask for answering from GetTransaction when execution cannot
+        // provide the response: a duplicate submission that already
+        // committed, or an execution error after the transaction landed.
+        // Both RPCs' masks select fields of `ExecutedTransaction`, so the
+        // caller's mask passes through unchanged; when the caller didn't set
+        // one, mirror ExecuteTransaction's documented default. `digest`,
+        // `checkpoint`, and `timestamp` are always included since this
+        // method's contract populates them.
+        let lookup_mask = {
+            let caller_paths = match &request.get_ref().read_mask {
+                Some(mask) => mask.paths.clone(),
+                None => vec!["effects.status".to_owned()],
+            };
+            FieldMask::from_paths(caller_paths.iter().map(String::as_str).chain([
+                "digest",
+                "checkpoint",
+                "timestamp",
+            ]))
+            .normalize()
+        };
+
         // Subscribe to checkpoint stream before execution to avoid missing the transaction.
         // Uses minimal read mask for efficiency since we only nee digest confirmation.
         // Once server-side filtering is available, we should filter by transaction digest to
@@ -139,14 +174,20 @@ impl Client {
             ))
         });
 
+        // The concurrent futures below each need a service client, and a
+        // single `&mut self` cannot back all of them at once, so give each
+        // its own client over the shared channel.
+        let mut execution_client = self.execution_client();
+        let mut post_exec_lookup_client = self.ledger_client();
+        let mut probe_client = self.ledger_client();
+
         // Execute, then query the fullnode directly to see if it already has
         // the txn in a checkpoint. This is to handle the case where an
         // already executed transaction is sent multiple times.
         let mut exec_and_check = Box::pin(async {
-            let response = self.execution_client().execute_transaction(request).await?;
+            let response = execution_client.execute_transaction(request).await?;
 
-            let already_checkpointed = match self
-                .ledger_client()
+            let already_checkpointed = match post_exec_lookup_client
                 .get_transaction(
                     GetTransactionRequest::default()
                         .with_digest(&executed_txn_digest)
@@ -164,22 +205,66 @@ impl Client {
             Ok::<_, tonic::Status>((response, already_checkpointed))
         });
 
-        // Drive execution and the scan together. The scan can complete first
-        // (for example, when a duplicate of an already executed transaction
-        // lands in a checkpoint mid-execution), so remember its outcome; the
-        // guard keeps a completed scan from being polled again.
+        // Probe the ledger while execution is in flight: a resubmission of a
+        // transaction that already committed can be answered from the ledger
+        // without waiting for, or succeeding at, execution.
+        let mut probe = Box::pin(async {
+            probe_client
+                .get_transaction(
+                    GetTransactionRequest::default()
+                        .with_digest(&executed_txn_digest)
+                        .with_read_mask(lookup_mask.clone()),
+                )
+                .await
+        });
+
+        // Drive execution, the scan, and the probe together. The scan can
+        // complete first (for example, when a duplicate of an already
+        // executed transaction lands in a checkpoint mid-execution), so
+        // remember its outcome; the guards keep completed futures from being
+        // polled again. A probe that finds the transaction in a checkpoint
+        // resolves the call on the spot; any other probe outcome (not found,
+        // not yet checkpointed, or an RPC error) means execution has to
+        // provide the answer.
         let mut scan_result = None;
-        let (mut response, already_checkpointed) = loop {
+        let mut probe_done = false;
+        let exec_result = loop {
             tokio::select! {
-                exec = &mut exec_and_check => {
-                    match exec {
-                        Ok(ok) => break ok,
-                        Err(e) => return Err(ExecuteAndWaitError::RpcError(e)),
-                    }
-                }
+                exec = &mut exec_and_check => break exec,
                 result = &mut scan, if scan_result.is_none() => {
                     scan_result = Some(result);
                 }
+                result = &mut probe, if !probe_done => {
+                    probe_done = true;
+                    if let Ok(lookup) = result
+                        && lookup.get_ref().transaction().checkpoint_opt().is_some()
+                    {
+                        return Ok(lookup_into_execute_response(lookup));
+                    }
+                }
+            }
+        };
+
+        let (mut response, already_checkpointed) = match exec_result {
+            Ok(ok) => ok,
+            Err(error) => {
+                // Execution can fail for a transaction that nonetheless
+                // committed, for example when a resubmission races the
+                // original submission. Consult the ledger before surfacing
+                // the error.
+                drop(probe);
+                if let Ok(lookup) = probe_client
+                    .get_transaction(
+                        GetTransactionRequest::default()
+                            .with_digest(&executed_txn_digest)
+                            .with_read_mask(lookup_mask),
+                    )
+                    .await
+                    && lookup.get_ref().transaction().checkpoint_opt().is_some()
+                {
+                    return Ok(lookup_into_execute_response(lookup));
+                }
+                return Err(ExecuteAndWaitError::RpcError(error));
             }
         };
 
@@ -232,6 +317,16 @@ impl Client {
     }
 }
 
+/// Builds the response for a transaction answered from the ledger instead of
+/// from execution.
+fn lookup_into_execute_response(
+    response: Response<GetTransactionResponse>,
+) -> Response<ExecuteTransactionResponse> {
+    Response::new(ExecuteTransactionResponse {
+        transaction: response.into_inner().transaction,
+    })
+}
+
 impl fmt::Display for ExecutionError {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         let description = self.description.as_deref().unwrap_or("No description");

crates/sui-rpc/tests/execute_and_wait.rs

@@ -9,8 +9,11 @@
 //! client's body watchdog — turning a successful execution into a spurious
 //! `CheckpointStreamError`.
 
+use std::sync::Arc;
+use std::sync::Mutex;
 use std::time::Duration;
 
+use prost_types::FieldMask;
 use proto::ledger_service_server::LedgerService;
 use proto::ledger_service_server::LedgerServiceServer;
 use proto::subscription_service_server::SubscriptionService;
@@ -31,9 +34,34 @@ struct MockServer {
     digest_at_seq: u64,
     /// How long `ExecuteTransaction` takes.
     exec_delay: Duration,
+    /// When set, `ExecuteTransaction` fails with this status after
+    /// `exec_delay`.
+    exec_error: Option<tonic::Status>,
     /// When set, `GetTransaction` reports the transaction as already
     /// included in this checkpoint (the duplicate-submission shortcut).
     shortcut_checkpoint: Option<u64>,
+    /// The 1-based `GetTransaction` call from which `shortcut_checkpoint`
+    /// applies; earlier calls report the transaction as not yet
+    /// checkpointed.
+    shortcut_from_lookup: usize,
+    /// Every `GetTransaction` request received, in order.
+    lookups: Arc<Mutex<Vec<proto::GetTransactionRequest>>>,
+}
+
+impl MockServer {
+    fn new(digest: String) -> Self {
+        Self {
+            digest,
+            // Far enough out that the subscription scan never resolves
+            // within a test's wait timeout.
+            digest_at_seq: 10_000,
+            exec_delay: Duration::ZERO,
+            exec_error: None,
+            shortcut_checkpoint: None,
+            shortcut_from_lookup: 1,
+            lookups: Arc::default(),
+        }
+    }
 }
 
 #[tonic::async_trait]
@@ -43,6 +71,9 @@ impl TransactionExecutionService for MockServer {
         _request: tonic::Request<proto::ExecuteTransactionRequest>,
     ) -> Result<tonic::Response<proto::ExecuteTransactionResponse>, tonic::Status> {
         tokio::time::sleep(self.exec_delay).await;
+        if let Some(status) = &self.exec_error {
+            return Err(status.clone());
+        }
         Ok(tonic::Response::new(
             proto::ExecuteTransactionResponse::default(),
         ))
@@ -53,10 +84,17 @@ impl TransactionExecutionService for MockServer {
 impl LedgerService for MockServer {
     async fn get_transaction(
         &self,
-        _request: tonic::Request<proto::GetTransactionRequest>,
+        request: tonic::Request<proto::GetTransactionRequest>,
     ) -> Result<tonic::Response<proto::GetTransactionResponse>, tonic::Status> {
+        let lookup_count = {
+            let mut lookups = self.lookups.lock().unwrap();
+            lookups.push(request.into_inner());
+            lookups.len()
+        };
         let mut response = proto::GetTransactionResponse::default();
-        if let Some(checkpoint) = self.shortcut_checkpoint {
+        if let Some(checkpoint) = self.shortcut_checkpoint
+            && lookup_count >= self.shortcut_from_lookup
+        {
             let mut transaction = proto::ExecutedTransaction::default();
             transaction.digest = Some(self.digest.clone());
             transaction.checkpoint = Some(checkpoint);
@@ -150,10 +188,9 @@ async fn subscription_survives_slow_execution() {
     let (transaction, digest) = test_transaction();
     let digest_at_seq = 50;
     let addr = spawn_mock_server(MockServer {
-        digest,
         digest_at_seq,
         exec_delay: Duration::from_secs(4),
-        shortcut_checkpoint: None,
+        ..MockServer::new(digest)
     })
     .await;
 
@@ -178,10 +215,8 @@ async fn subscription_survives_slow_execution() {
 async fn already_checkpointed_transaction_short_circuits() {
     let (transaction, digest) = test_transaction();
     let addr = spawn_mock_server(MockServer {
-        digest,
-        digest_at_seq: 10_000,
-        exec_delay: Duration::ZERO,
         shortcut_checkpoint: Some(7),
+        ..MockServer::new(digest)
     })
     .await;
 
@@ -195,6 +230,106 @@ async fn already_checkpointed_transaction_short_circuits() {
     assert_eq!(response.get_ref().transaction().checkpoint(), 7);
 }
 
+/// A resubmission of a transaction that is already in a checkpoint is
+/// answered by the ledger probe that races execution: the call returns long
+/// before the execution RPC completes, and the lookup carries the caller's
+/// read mask plus the fields this method always populates.
+#[tokio::test(flavor = "multi_thread")]
+async fn committed_duplicate_returns_before_execution_finishes() {
+    let (transaction, digest) = test_transaction();
+    let lookups = Arc::new(Mutex::new(Vec::new()));
+    let addr = spawn_mock_server(MockServer {
+        exec_delay: Duration::from_secs(60),
+        shortcut_checkpoint: Some(7),
+        lookups: lookups.clone(),
+        ..MockServer::new(digest.clone())
+    })
+    .await;
+
+    let mut client = Client::new(format!("http://{addr}")).expect("client");
+
+    let request = proto::ExecuteTransactionRequest::default()
+        .with_transaction(transaction)
+        .with_read_mask(FieldMask {
+            paths: vec!["effects.status".to_owned()],
+        });
+    let start = std::time::Instant::now();
+    let response = client
+        .execute_transaction_and_wait_for_checkpoint(request, Duration::from_secs(5))
+        .await
+        .expect("committed duplicate resolved from the ledger probe");
+    assert!(
+        start.elapsed() < Duration::from_secs(30),
+        "the probe must resolve the call without waiting for execution"
+    );
+    assert_eq!(response.get_ref().transaction().checkpoint(), 7);
+    assert_eq!(response.get_ref().transaction().digest(), digest);
+
+    let probe_mask = lookups.lock().unwrap()[0]
+        .read_mask
+        .clone()
+        .expect("probe read mask");
+    for path in ["effects.status", "digest", "checkpoint", "timestamp"] {
+        assert!(
+            probe_mask.paths.iter().any(|p| p == path),
+            "probe read mask is missing `{path}`: {probe_mask:?}"
+        );
+    }
+}
+
+/// An execution error for a transaction that nonetheless committed (for
+/// example a resubmission racing the original submission) is downgraded to
+/// success by a final ledger lookup. The probe still sees the transaction as
+/// not checkpointed here, so only the post-error lookup can resolve the call.
+#[tokio::test(flavor = "multi_thread")]
+async fn execution_error_with_committed_transaction_succeeds() {
+    let (transaction, digest) = test_transaction();
+    let addr = spawn_mock_server(MockServer {
+        exec_delay: Duration::from_millis(500),
+        exec_error: Some(tonic::Status::aborted("transaction already being executed")),
+        shortcut_checkpoint: Some(9),
+        shortcut_from_lookup: 2,
+        ..MockServer::new(digest.clone())
+    })
+    .await;
+
+    let mut client = Client::new(format!("http://{addr}")).expect("client");
+
+    let request = proto::ExecuteTransactionRequest::default().with_transaction(transaction);
+    let response = client
+        .execute_transaction_and_wait_for_checkpoint(request, Duration::from_secs(5))
+        .await
+        .expect("execution error downgraded for a committed transaction");
+    assert_eq!(response.get_ref().transaction().checkpoint(), 9);
+    assert_eq!(response.get_ref().transaction().digest(), digest);
+}
+
+/// An execution error for a transaction the ledger has no checkpoint for
+/// still surfaces as `RpcError`.
+#[tokio::test(flavor = "multi_thread")]
+async fn execution_error_without_commit_surfaces() {
+    let (transaction, digest) = test_transaction();
+    let addr = spawn_mock_server(MockServer {
+        exec_error: Some(tonic::Status::internal("boom")),
+        ..MockServer::new(digest)
+    })
+    .await;
+
+    let mut client = Client::new(format!("http://{addr}")).expect("client");
+
+    let request = proto::ExecuteTransactionRequest::default().with_transaction(transaction);
+    let err = client
+        .execute_transaction_and_wait_for_checkpoint(request, Duration::from_secs(5))
+        .await
+        .expect_err("execution error with no committed transaction");
+    match err {
+        sui_rpc::client::ExecuteAndWaitError::RpcError(status) => {
+            assert_eq!(status.code(), tonic::Code::Internal);
+        }
+        other => panic!("expected RpcError, got {other:?}"),
+    }
+}
+
 /// A request without a transaction fails fast, before any network work.
 #[tokio::test(flavor = "multi_thread")]
 async fn missing_transaction_fails_without_rpc() {