chore: gRPC migration level 5 — transaction reads (#3224)

Migrates `get_transaction_with_options` and `get_events` to gRPC at migration level 5, and decouples Walrus from some `sui_sdk` response types.

### gRPC transaction reads (level 5)

Routes `get_transaction_with_options` through `LedgerService.GetTransaction` when `WALRUS_GRPC_MIGRATION_LEVEL >= 5`. The gRPC path handles call sites needing checkpoint, timestamp, raw_input, balance_changes, and/or events. Added conversion helpers for reconstructing `SenderSignedData` BCS, mapping gRPC `BalanceChange`/`Event` types, and building field masks. `query_transaction_blocks` stays JSON-RPC (no gRPC equivalent, only used in benchmarks).

### Protocol-agnostic types

Introduced `EventEnvelope`, `BalanceChange`, `TransactionResponse`, and `TransactionResponseOptions` in `walrus-sui/src/types/` to replace some `sui_sdk::rpc_types` at the Walrus API boundary. Converted all 24 `TryFrom<SuiEvent>` impls to `TryFrom<EventEnvelope>`. JSON-RPC boundaries convert to the new types at the call site.

### Consolidated `get_events`

`get_events` now routes through `get_transaction_with_options` for both gRPC and JSON-RPC paths, eliminating the separate `event_api().get_events()` dependency.

### Stack overflow fixes

Boxed the futures for `execute_transaction` and `get_transaction_with_options` to prevent stack overflows in deep async call chains.

Author: wbbradley

CLAUDE.md

@@ -181,6 +181,12 @@ Configured in `.config/nextest.toml`:
 - **simtest**: `num-cpus` threads, 30m timeout
 - **performance-test**: single-threaded, 15m timeout
 
+## gRPC protobuf definitions
+
+Sui Rust SDK houses the .proto files for the Sui gRPC API. Often these are stored locally at
+../sui-rust-sdk/crates/sui-rpc/vendored/proto/sui/rpc/v2/. If not, they are available at
+https://github.com/MystenLabs/sui-rust-sdk/tree/master/crates/sui-rpc/vendored/proto/sui/rpc/v2.
+
 ## Documentation
 
 Documentation files live under `docs/content/` and use MDX (Docusaurus) format. The site is built

crates/walrus-service/src/event/event_processor/bootstrap.rs

@@ -3,7 +3,6 @@
 
 //! Bootstrap module for getting the initial committee and checkpoint information.
 use anyhow::{Context as _, Result};
-use sui_sdk::rpc_types::SuiTransactionBlockResponseOptions;
 use sui_types::{
     base_types::ObjectID,
     committee::Committee,
@@ -26,7 +25,7 @@ pub async fn get_bootstrap_committee_and_checkpoint(
     system_pkg_id: ObjectID,
 ) -> Result<(Committee, VerifiedCheckpoint)> {
     let txn_digest = sui_client.get_previous_transaction(system_pkg_id).await?;
-    let txn_options = SuiTransactionBlockResponseOptions::new();
+    let txn_options = walrus_sui::types::TransactionResponseOptions::new();
     let txn = sui_client
         .get_transaction_with_options(txn_digest, txn_options)
         .await?;

crates/walrus-service/src/event/event_processor/checkpoint.rs

@@ -181,7 +181,8 @@ impl CheckpointProcessor {
                     None,
                     move_datatype_layout,
                 )?;
-                let contract_event: ContractEvent = sui_event.try_into()?;
+                let contract_event: ContractEvent =
+                    walrus_sui::types::EventEnvelope::from(sui_event).try_into()?;
                 let event_sequence_number = CheckpointEventPosition::new(
                     *checkpoint.checkpoint_summary.sequence_number(),
                     counter,

crates/walrus-sui/src/client/dual_client.rs

@@ -18,8 +18,10 @@ use sui_rpc::{
         BatchGetObjectsResponse,
         Bcs,
         DynamicField,
+        ExecutedTransaction,
         GetBalanceRequest,
         GetObjectRequest,
+        GetTransactionRequest,
         ListDynamicFieldsRequest,
         ListOwnedObjectsRequest,
         ListOwnedObjectsResponse,
@@ -32,7 +34,11 @@ use sui_sdk::{SuiClient, SuiClientBuilder};
 use sui_types::{
     TypeTag,
     base_types::{ObjectID, ObjectRef, SuiAddress},
+    crypto::ToFromBytes,
     digests::TransactionDigest,
+    event::EventID,
+    signature::GenericSignature,
+    transaction::{SenderSignedData, TransactionData},
 };
 use tonic::service::interceptor::InterceptedService;
 use walrus_core::ensure;
@@ -42,6 +48,7 @@ use crate::{
     coin::Coin,
     contracts::{AssociatedContractStruct, TypeOriginMap},
     dynamic_field_info::{DynamicFieldPage, dynamic_field_info_from_grpc},
+    types::{BalanceChange, EventEnvelope, TransactionResponse, TransactionResponseOptions},
 };
 
 /// The maximum number of objects to request in a single "batch" gRPC call.
@@ -160,6 +167,47 @@ impl DualClient {
             .context("parsing previous_transaction")?)
     }
 
+    /// Get a transaction response from the Sui network via gRPC.
+    ///
+    /// Supports requesting raw_input, balance_changes, and events via the options parameter.
+    /// Always includes digest, checkpoint, and timestamp.
+    pub async fn get_transaction_response_grpc(
+        &self,
+        digest: TransactionDigest,
+        options: &TransactionResponseOptions,
+    ) -> Result<TransactionResponse, SuiClientError> {
+        let field_mask = build_transaction_field_mask(options);
+        let executed_tx = self.get_transaction_raw(digest, field_mask).await?;
+        executed_transaction_to_response(digest, options, &executed_tx)
+    }
+
+    /// Low-level gRPC call to get a transaction with a specific field mask.
+    ///
+    /// Uses `Box::pin` to erase the future type and avoid `Send` recursion overflow
+    /// from the deep `ExecutedTransaction` proto type tree.
+    fn get_transaction_raw(
+        &self,
+        digest: TransactionDigest,
+        field_mask: FieldMask,
+    ) -> std::pin::Pin<
+        Box<
+            dyn std::future::Future<Output = Result<ExecutedTransaction, SuiClientError>>
+                + Send
+                + '_,
+        >,
+    > {
+        let mut grpc_client: GrpcClient = self.grpc_client.clone();
+        Box::pin(async move {
+            let sui_digest = sui_sdk_types::Digest::new(digest.into_inner());
+            let request = GetTransactionRequest::new(&sui_digest).with_read_mask(field_mask);
+            let response = grpc_client.ledger_client().get_transaction(request).await?;
+            Ok(response
+                .into_inner()
+                .transaction
+                .context("no transaction in get_transaction_response")?)
+        })
+    }
+
     /// Get the BCS representation of an object's contents and its type from the Sui network.
     pub async fn get_object_contents(
         &self,
@@ -657,6 +705,210 @@ fn convert_grpc_object_to_coin(object: Object) -> Result<Coin, anyhow::Error> {
     })
 }
 
+/// Build a [`FieldMask`] for `GetTransactionRequest` based on the requested options.
+///
+/// Always requests `digest`, `checkpoint`, and `timestamp`. Conditionally adds paths for
+/// `raw_input` (transaction BCS + signature BCS), `balance_changes`, and `events`.
+fn build_transaction_field_mask(options: &TransactionResponseOptions) -> FieldMask {
+    let mut paths: Vec<String> = vec![
+        ExecutedTransaction::path_builder().digest(),
+        ExecutedTransaction::path_builder().checkpoint(),
+        ExecutedTransaction::path_builder().timestamp(),
+    ];
+    if options.show_raw_input {
+        paths.push(
+            ExecutedTransaction::path_builder()
+                .transaction()
+                .bcs()
+                .finish(),
+        );
+        paths.push(
+            ExecutedTransaction::path_builder()
+                .signatures()
+                .bcs()
+                .finish(),
+        );
+    }
+    if options.show_balance_changes {
+        paths.push(
+            ExecutedTransaction::path_builder()
+                .balance_changes()
+                .finish(),
+        );
+    }
+    if options.show_events {
+        paths.push(
+            ExecutedTransaction::path_builder()
+                .events()
+                .events()
+                .finish(),
+        );
+    }
+    FieldMask::from_paths(&paths)
+}
+
+/// Convert a gRPC [`ExecutedTransaction`] into a [`TransactionResponse`].
+fn executed_transaction_to_response(
+    digest: TransactionDigest,
+    options: &TransactionResponseOptions,
+    executed_tx: &ExecutedTransaction,
+) -> Result<TransactionResponse, SuiClientError> {
+    let checkpoint = executed_tx.checkpoint;
+    let timestamp_ms = executed_tx
+        .timestamp
+        .as_ref()
+        .map(|ts| {
+            let seconds = u64::try_from(ts.seconds).context("negative timestamp seconds")?;
+            let nanos = u64::try_from(ts.nanos).context("negative timestamp nanos")?;
+            anyhow::Ok(seconds * 1000 + nanos / 1_000_000)
+        })
+        .transpose()?;
+
+    let raw_transaction = if options.show_raw_input {
+        reconstruct_raw_transaction(executed_tx)?
+    } else {
+        Vec::new()
+    };
+
+    let balance_changes = if options.show_balance_changes {
+        Some(
+            executed_tx
+                .balance_changes
+                .iter()
+                .map(convert_grpc_balance_change)
+                .collect::<Result<Vec<_>, _>>()?,
+        )
+    } else {
+        None
+    };
+
+    let events = if options.show_events {
+        executed_tx
+            .events
+            .as_ref()
+            .map(|events| {
+                events
+                    .events
+                    .iter()
+                    // Event index matches on-chain event_seq; protobuf repeated fields
+                    // preserve order and the gRPC server serializes from the same vector.
+                    .enumerate()
+                    .map(|(idx, event)| grpc_event_to_event_envelope(digest, idx, event))
+                    .collect::<Result<Vec<_>, _>>()
+            })
+            .transpose()?
+    } else {
+        None
+    };
+
+    Ok(TransactionResponse {
+        digest,
+        checkpoint,
+        timestamp_ms,
+        raw_transaction,
+        balance_changes,
+        events,
+    })
+}
+
+/// Reconstruct the BCS-encoded `SenderSignedData` (raw_transaction) from gRPC fields.
+fn reconstruct_raw_transaction(
+    executed_tx: &ExecutedTransaction,
+) -> Result<Vec<u8>, SuiClientError> {
+    let tx_bcs = executed_tx
+        .transaction
+        .as_ref()
+        .context("no transaction in executed_transaction")?
+        .bcs
+        .as_ref()
+        .context("no bcs in transaction")?;
+
+    let tx_data: TransactionData = bcs::from_bytes(
+        tx_bcs
+            .value
+            .as_ref()
+            .context("no value in transaction bcs")?,
+    )
+    .context("deserializing TransactionData from gRPC bcs")?;
+
+    let tx_signatures: Vec<GenericSignature> = executed_tx
+        .signatures
+        .iter()
+        .map(|sig| {
+            let sig_bcs = sig.bcs.as_ref().context("no bcs in user signature")?;
+            let sig_bytes = sig_bcs
+                .value
+                .as_ref()
+                .context("no value in signature bcs")?;
+            GenericSignature::from_bytes(sig_bytes)
+                .context("parsing GenericSignature from gRPC bcs")
+        })
+        .collect::<Result<Vec<_>, _>>()?;
+
+    let sender_signed_data = SenderSignedData::new(tx_data, tx_signatures);
+    Ok(bcs::to_bytes(&sender_signed_data).context("serializing SenderSignedData")?)
+}
+
+/// Convert a gRPC `BalanceChange` to our protocol-agnostic [`BalanceChange`].
+fn convert_grpc_balance_change(
+    change: &sui_rpc::proto::sui::rpc::v2::BalanceChange,
+) -> Result<BalanceChange, SuiClientError> {
+    let address: SuiAddress = change
+        .address
+        .as_ref()
+        .context("no address in balance change")?
+        .parse()
+        .context("parsing balance change address")?;
+    let coin_type: TypeTag = change
+        .coin_type
+        .as_ref()
+        .context("no coin_type in balance change")?
+        .parse()
+        .context("parsing balance change coin_type")?;
+    let amount: i128 = change
+        .amount
+        .as_ref()
+        .context("no amount in balance change")?
+        .parse()
+        .context("parsing balance change amount")?;
+    Ok(BalanceChange {
+        address,
+        coin_type,
+        amount,
+    })
+}
+
+/// Convert a gRPC `Event` to an [`EventEnvelope`].
+fn grpc_event_to_event_envelope(
+    tx_digest: TransactionDigest,
+    event_seq: usize,
+    event: &sui_rpc::proto::sui::rpc::v2::Event,
+) -> Result<EventEnvelope, SuiClientError> {
+    let type_: StructTag = event
+        .event_type
+        .as_ref()
+        .context("no event_type in event")?
+        .parse()
+        .context("parsing event type")?;
+    let bcs = event
+        .contents
+        .as_ref()
+        .context("no contents in event")?
+        .value
+        .as_ref()
+        .context("no value in event contents")?
+        .to_vec();
+
+    Ok(EventEnvelope {
+        id: EventID {
+            tx_digest,
+            event_seq: u64::try_from(event_seq).context("event_seq overflow")?,
+        },
+        type_,
+        bcs,
+    })
+}
+
 async fn batch_get_objects<T>(
     mut grpc_client: GrpcClient,
     object_ids: &[ObjectID],

crates/walrus-sui/src/client/read_client.rs

@@ -20,7 +20,6 @@ use sui_sdk::{
     apis::EventApi,
     rpc_types::{
         EventFilter,
-        SuiEvent,
         SuiObjectData,
         SuiObjectDataFilter,
         SuiObjectDataOptions,
@@ -1509,7 +1508,7 @@ async fn poll_for_events<U>(
     mut last_event: Option<EventID>,
 ) -> Result<()>
 where
-    U: TryFrom<SuiEvent> + Send + Sync + Debug + 'static,
+    U: TryFrom<crate::types::EventEnvelope> + Send + Sync + Debug + 'static,
 {
     // The actual interval with which we poll, increases if there is an RPC error
     let mut polling_interval = initial_polling_interval;
@@ -1536,9 +1535,10 @@ where
                         event_id = ?event.id,
                         event_type = ?event.type_
                     );
+                    let envelope = crate::types::EventEnvelope::from(event);
 
                     let continue_or_exit = async move {
-                        let event_obj = match event.try_into() {
+                        let event_obj = match envelope.try_into() {
                             Ok(event_obj) => event_obj,
                             Err(_) => {
                                 tracing::error!("could not convert event");