iota_json_rpc/

transaction_execution_api.rs

// Copyright (c) Mysten Labs, Inc.
// Modifications Copyright (c) 2024 IOTA Stiftung
// SPDX-License-Identifier: Apache-2.0

use std::{sync::Arc, time::Duration};

use async_trait::async_trait;
use fastcrypto::{encoding::Base64, traits::ToFromBytes};
use iota_core::{
    authority::AuthorityState, authority_client::NetworkAuthorityClient,
    transaction_orchestrator::TransactionOrchestrator,
};
use iota_json::IotaJsonValue;
use iota_json_rpc_api::{JsonRpcMetrics, WriteApiOpenRpc, WriteApiServer};
use iota_json_rpc_types::{
    DevInspectArgs, DevInspectResults, DryRunTransactionBlockResponse,
    ExecuteTransactionRequestType as ExecuteTransactionRequestTypeSchema, IotaExecutionStatus,
    IotaMoveViewCallResults, IotaTransactionBlock, IotaTransactionBlockEffects,
    IotaTransactionBlockEffectsAPI, IotaTransactionBlockEvents, IotaTransactionBlockResponse,
    IotaTransactionBlockResponseOptions, IotaTypeTag, MoveFunctionName,
};
use iota_metrics::spawn_monitored_task;
use iota_open_rpc::Module;
use iota_package_resolver::{
    Package, PackageStore, Resolver, error::Error as PackageResolverError,
};
use iota_protocol_config::Chain;
use iota_sdk_types::crypto::{Intent, IntentAppId, IntentMessage, IntentScope, IntentVersion};
use iota_transaction_builder::TransactionBuilder;
use iota_types::{
    base_types::{IotaAddress, ObjectID},
    digests::TransactionDigest,
    effects::{TransactionEffectsAPI, TransactionEffectsExt},
    iota_serde::BigInt,
    quorum_driver_types::{
        ExecuteTransactionRequestType, ExecuteTransactionRequestV1, ExecuteTransactionResponseV1,
    },
    signature::GenericSignature,
    storage::PostExecutionPackageResolver,
    transaction::{
        InputObjectKind, Transaction, TransactionData, TransactionDataAPI, TransactionKind,
    },
};
use jsonrpsee::{RpcModule, core::RpcResult};
use tracing::{Instrument, instrument};

use crate::{
    IotaRpcModule, ObjectProviderCache,
    authority_state::StateRead,
    error::{Error, IotaRpcInputError},
    get_balance_changes_from_effect, get_object_changes,
    logger::FutureWithTracing,
    transaction_builder_api::AuthorityStateDataReader,
};

#[derive(Clone)]
pub struct TransactionExecutionApi {
    state: Arc<dyn StateRead>,
    transaction_orchestrator: Arc<TransactionOrchestrator<NetworkAuthorityClient>>,
    metrics: Arc<JsonRpcMetrics>,
    transaction_builder: TransactionBuilder,
}

impl TransactionExecutionApi {
    pub fn new(
        state: Arc<AuthorityState>,
        transaction_orchestrator: Arc<TransactionOrchestrator<NetworkAuthorityClient>>,
        metrics: Arc<JsonRpcMetrics>,
    ) -> Self {
        let reader = Arc::new(AuthorityStateDataReader::new(state.clone()));
        Self {
            state,
            transaction_orchestrator,
            metrics,
            transaction_builder: TransactionBuilder::new(reader),
        }
    }

    pub fn convert_bytes<T: serde::de::DeserializeOwned>(
        &self,
        tx_bytes: Base64,
    ) -> Result<T, IotaRpcInputError> {
        let data: T = bcs::from_bytes(&tx_bytes.to_vec()?)?;
        Ok(data)
    }

    #[expect(clippy::type_complexity)]
    fn prepare_execute_transaction_block(
        &self,
        tx_bytes: Base64,
        signatures: Vec<Base64>,
        opts: Option<IotaTransactionBlockResponseOptions>,
    ) -> Result<
        (
            ExecuteTransactionRequestV1,
            IotaTransactionBlockResponseOptions,
            IotaAddress,
            Vec<InputObjectKind>,
            Transaction,
            Option<IotaTransactionBlock>,
            Vec<u8>,
        ),
        IotaRpcInputError,
    > {
        let opts = opts.unwrap_or_default();
        let tx_data: TransactionData = self.convert_bytes(tx_bytes)?;
        let sender = tx_data.sender();
        let input_objs = tx_data.input_objects().unwrap_or_default();

        let mut sigs = Vec::new();
        for sig in signatures {
            sigs.push(GenericSignature::from_bytes(&sig.to_vec()?)?);
        }
        let txn = Transaction::from_generic_sig_data(tx_data, sigs);
        let raw_transaction = if opts.show_raw_input {
            bcs::to_bytes(txn.data())?
        } else {
            vec![]
        };
        let transaction = if opts.show_input {
            let epoch_store = self.state.load_epoch_store_one_call_per_task();

            Some(IotaTransactionBlock::try_from(
                txn.data().clone(),
                epoch_store.module_cache(),
                *txn.digest(),
            )?)
        } else {
            None
        };

        let request = ExecuteTransactionRequestV1 {
            transaction: txn.clone(),
            include_events: opts.show_events,
            include_input_objects: opts.show_balance_changes || opts.show_object_changes,
            include_output_objects: opts.show_balance_changes
                || opts.show_object_changes
                // In order to resolve events, we may need access to the newly published packages.
                || opts.show_events,
            include_auxiliary_data: false,
        };

        Ok((
            request,
            opts,
            sender,
            input_objs,
            txn,
            transaction,
            raw_transaction,
        ))
    }

    #[instrument("json_rpc_api_execute_transaction_block", level = "trace", skip_all)]
    async fn execute_transaction_block(
        &self,
        tx_bytes: Base64,
        signatures: Vec<Base64>,
        opts: Option<IotaTransactionBlockResponseOptions>,
        request_type: Option<ExecuteTransactionRequestType>,
    ) -> Result<IotaTransactionBlockResponse, Error> {
        let request_type =
            request_type.unwrap_or(ExecuteTransactionRequestType::WaitForEffectsCert);
        let (request, opts, sender, input_objs, txn, transaction, raw_transaction) =
            self.prepare_execute_transaction_block(tx_bytes, signatures, opts)?;
        let digest = *txn.digest();

        let transaction_orchestrator = self.transaction_orchestrator.clone();
        let orch_timer = self.metrics.orchestrator_latency_ms.start_timer();

        tracing::trace!(
            "Spawning transaction orchestrator task for transaction: {}",
            digest
        );
        let (response, is_executed_locally) = spawn_monitored_task!(
            transaction_orchestrator.execute_transaction_block(request, request_type, None)
        )
        .await?
        .map_err(Error::from)?;
        drop(orch_timer);

        self.handle_post_orchestration(
            response,
            is_executed_locally,
            opts,
            digest,
            input_objs,
            transaction,
            raw_transaction,
            sender,
        )
        .await
    }

    #[instrument(level = "trace", skip_all)]
    async fn handle_post_orchestration(
        &self,
        response: ExecuteTransactionResponseV1,
        is_executed_locally: bool,
        opts: IotaTransactionBlockResponseOptions,
        digest: TransactionDigest,
        input_objs: Vec<InputObjectKind>,
        transaction: Option<IotaTransactionBlock>,
        raw_transaction: Vec<u8>,
        sender: IotaAddress,
    ) -> Result<IotaTransactionBlockResponse, Error> {
        let _post_orch_timer = self.metrics.post_orchestrator_latency_ms.start_timer();

        let events = if opts.show_events {
            tracing::trace!("Resolving events");
            let epoch_store = self.state.load_epoch_store_one_call_per_task();
            let backing_package_store = PostExecutionPackageResolver::new(
                self.state.get_backing_package_store().clone(),
                &response.output_objects,
            );
            let mut layout_resolver = epoch_store
                .executor()
                .type_layout_resolver(Box::new(backing_package_store));
            Some(IotaTransactionBlockEvents::try_from(
                response.events.unwrap_or_default(),
                digest,
                None,
                layout_resolver.as_mut(),
            )?)
        } else {
            None
        };

        // Skip cache (and downstream balance/object_changes) when the validator
        // returned no input/output objects — e.g. the already-executed early-return.
        // Without this guard, cache misses fall through to a provider lookup that
        // races with local state and returns "version higher than latest".
        let object_cache = if (opts.show_balance_changes || opts.show_object_changes)
            && (response.input_objects.is_some() || response.output_objects.is_some())
        {
            let mut object_cache = ObjectProviderCache::new(self.state.clone());
            if let Some(input_objects) = response.input_objects {
                object_cache.insert_objects_into_cache(input_objects);
            }
            if let Some(output_objects) = response.output_objects {
                object_cache.insert_objects_into_cache(output_objects);
            }
            Some(object_cache)
        } else {
            None
        };

        let balance_changes = match &object_cache {
            Some(object_cache) if opts.show_balance_changes => Some(
                get_balance_changes_from_effect(
                    object_cache,
                    &response.effects.effects,
                    input_objs,
                    None,
                )
                .instrument(tracing::trace_span!("resolving balance changes"))
                .await?,
            ),
            _ => None,
        };

        let object_changes = match &object_cache {
            Some(object_cache) if opts.show_object_changes => Some(
                get_object_changes(
                    object_cache,
                    sender,
                    response.effects.effects.modified_at_versions(),
                    response.effects.effects.all_changed_objects(),
                    response.effects.effects.all_removed_objects(),
                )
                .instrument(tracing::trace_span!("resolving object changes"))
                .await?,
            ),
            _ => None,
        };

        let raw_effects = if opts.show_raw_effects {
            bcs::to_bytes(&response.effects.effects)?
        } else {
            vec![]
        };
        let resolver = Resolver::new(self.clone());

        let effects = if opts.show_effects {
            Some(
                IotaTransactionBlockEffects::from_native_with_clever_error(
                    response.effects.effects,
                    &resolver,
                )
                .await,
            )
        } else {
            None
        };

        let errors = match effects.as_ref().map(|e| e.status()) {
            Some(IotaExecutionStatus::Failure { error }) => vec![error.clone()],
            _ => vec![],
        };

        Ok(IotaTransactionBlockResponse {
            digest,
            transaction,
            raw_transaction,
            effects,
            events,
            object_changes,
            balance_changes,
            timestamp_ms: None,
            confirmed_local_execution: Some(is_executed_locally),
            checkpoint: None,
            errors,
            raw_effects,
        })
    }

    pub fn prepare_dry_run_transaction_block(
        &self,
        tx_bytes: Base64,
    ) -> Result<(TransactionData, TransactionDigest, Vec<InputObjectKind>), IotaRpcInputError> {
        let tx_data: TransactionData = self.convert_bytes(tx_bytes)?;
        let input_objs = tx_data.input_objects()?;
        let intent_msg = IntentMessage::new(
            Intent {
                version: IntentVersion::V0,
                scope: IntentScope::TransactionData,
                app_id: IntentAppId::Iota,
            },
            tx_data,
        );
        let txn_digest = TransactionDigest::new(intent_msg.value.digest().into_inner());
        Ok((intent_msg.value, txn_digest, input_objs))
    }

    async fn dry_run_transaction_block(
        &self,
        tx_bytes: Base64,
    ) -> Result<DryRunTransactionBlockResponse, Error> {
        let (txn_data, txn_digest, input_objs) =
            self.prepare_dry_run_transaction_block(tx_bytes)?;
        let sender = txn_data.sender();

        // Use spawn_blocking since dry_exec_transaction is a long-running synchronous
        // operation
        let state = self.state.clone();
        let (resp, written_objects, transaction_effects, mock_gas) =
            tokio::task::spawn_blocking(move || {
                state.dry_exec_transaction(txn_data.clone(), txn_digest)
            })
            .await
            .map_err(Error::from)??;

        let object_cache = ObjectProviderCache::new_with_cache(self.state.clone(), written_objects);
        let balance_changes = get_balance_changes_from_effect(
            &object_cache,
            &transaction_effects,
            input_objs,
            mock_gas,
        )
        .await?;
        let object_changes = get_object_changes(
            &object_cache,
            sender,
            transaction_effects.modified_at_versions(),
            transaction_effects.all_changed_objects(),
            transaction_effects.all_removed_objects(),
        )
        .await?;

        let resolver = Resolver::new(self.clone());
        let effects = IotaTransactionBlockEffects::from_native_with_clever_error(
            transaction_effects,
            &resolver,
        )
        .await;

        Ok(DryRunTransactionBlockResponse {
            effects,
            events: resp.events,
            object_changes,
            balance_changes,
            input: resp.input,
            suggested_gas_price: resp.suggested_gas_price,
            execution_error_source: resp.execution_error_source,
        })
    }
}

#[async_trait]
impl WriteApiServer for TransactionExecutionApi {
    #[instrument(skip(self))]
    async fn execute_transaction_block(
        &self,
        tx_bytes: Base64,
        signatures: Vec<Base64>,
        opts: Option<IotaTransactionBlockResponseOptions>,
        request_type: Option<ExecuteTransactionRequestTypeSchema>,
    ) -> RpcResult<IotaTransactionBlockResponse> {
        self.execute_transaction_block(tx_bytes, signatures, opts, request_type.map(Into::into))
            .trace_timeout(Duration::from_secs(10))
            .await
    }

    /// Calls a move view function.
    #[instrument(skip(self))]
    async fn view_function_call(
        &self,
        function_name: String,
        type_args: Option<Vec<IotaTypeTag>>,
        arguments: Vec<IotaJsonValue>,
    ) -> RpcResult<IotaMoveViewCallResults> {
        let chain = self
            .state
            .get_chain_identifier()
            .map_err(Error::from)?
            .chain();
        if !matches!(chain, Chain::Unknown) {
            return Err(Error::UnsupportedFeature(format!(
                "View function calls not supported yet on {}",
                chain.as_str()
            ))
            .into());
        }
        let MoveFunctionName {
            package,
            module,
            function,
        } = function_name.as_str().parse().map_err(Error::from)?;
        let sender = IotaAddress::ZERO;
        let tx_kind = self
            .transaction_builder
            .move_view_call_tx_kind(
                package,
                &module,
                &function,
                type_args.unwrap_or_default(),
                arguments,
            )
            .await
            .map_err(Error::from)?;
        let tx_bytes = Base64::from_bytes(&bcs::to_bytes(&tx_kind).map_err(Error::from)?);
        let dev_inspect_results = self
            .dev_inspect_transaction_block(sender, tx_bytes, None, None, None)
            .await?;
        Ok(
            IotaMoveViewCallResults::from_dev_inspect_results(self.clone(), dev_inspect_results)
                .await
                .map_err(Error::from)?,
        )
    }

    #[instrument(skip(self, sender_address), fields(sender_address = %sender_address))]
    async fn dev_inspect_transaction_block(
        &self,
        sender_address: IotaAddress,
        tx_bytes: Base64,
        gas_price: Option<BigInt<u64>>,
        _epoch: Option<BigInt<u64>>,
        additional_args: Option<DevInspectArgs>,
    ) -> RpcResult<DevInspectResults> {
        async move {
            let DevInspectArgs {
                gas_sponsor,
                gas_budget,
                gas_objects,
                show_raw_txn_data_and_effects,
                skip_checks,
            } = additional_args.unwrap_or_default();
            let tx_kind: TransactionKind = self.convert_bytes(tx_bytes)?;
            self.state
                .dev_inspect_transaction_block(
                    sender_address,
                    tx_kind,
                    gas_price.map(|i| *i),
                    gas_budget,
                    gas_sponsor,
                    gas_objects,
                    show_raw_txn_data_and_effects,
                    skip_checks,
                )
                .await
                .map_err(Error::from)
        }
        .trace()
        .await
    }

    #[instrument(skip(self))]
    async fn dry_run_transaction_block(
        &self,
        tx_bytes: Base64,
    ) -> RpcResult<DryRunTransactionBlockResponse> {
        self.dry_run_transaction_block(tx_bytes).trace().await
    }
}

impl IotaRpcModule for TransactionExecutionApi {
    fn rpc(self) -> RpcModule<Self> {
        self.into_rpc()
    }

    fn rpc_doc_module() -> Module {
        WriteApiOpenRpc::module_doc()
    }
}

#[async_trait]
impl PackageStore for TransactionExecutionApi {
    async fn fetch(&self, id: IotaAddress) -> Result<Arc<Package>, PackageResolverError> {
        let backing_store = self.state.get_backing_package_store();
        match backing_store.get_package_object(&ObjectID::new(id.into_bytes())) {
            Ok(Some(pkg)) => Ok(Arc::new(Package::read_from_package(pkg.move_package())?)),
            Ok(None) => Err(PackageResolverError::PackageNotFound(id)),
            Err(e) => Err(PackageResolverError::Store {
                store: "Node",
                source: Arc::new(e),
            }),
        }
    }
}