sui_indexer_alt_reader/
kv_loader.rs

1// Copyright (c) Mysten Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4use std::collections::HashMap;
5use std::sync::Arc;
6
7use anyhow::Context;
8use async_graphql::dataloader::DataLoader;
9use prometheus::Registry;
10use sui_rpc::proto::sui::rpc::v2 as grpc;
11use sui_types::base_types::ObjectID;
12use sui_types::crypto::AuthorityQuorumSignInfo;
13use sui_types::digests::CheckpointDigest;
14use sui_types::digests::TransactionDigest;
15use sui_types::digests::TransactionEffectsDigest;
16use sui_types::effects::TransactionEffects;
17use sui_types::effects::TransactionEffectsAPI;
18use sui_types::effects::TransactionEvents;
19use sui_types::event::Event;
20use sui_types::message_envelope::Message;
21use sui_types::messages_checkpoint::CheckpointContents;
22use sui_types::messages_checkpoint::CheckpointSummary;
23use sui_types::object::Object;
24use sui_types::signature::GenericSignature;
25use sui_types::transaction::TransactionData;
26use tonic::transport::Uri;
27
28use crate::alpha_ledger_grpc_reader::AlphaLedgerGrpcReader;
29use crate::checkpoints::CheckpointKey;
30use crate::error::Error;
31use crate::events::TransactionEventsKey;
32use crate::ledger_grpc_reader::CheckpointedTransaction;
33use crate::ledger_grpc_reader::LedgerGrpcArgs;
34use crate::ledger_grpc_reader::LedgerGrpcReader;
35use crate::ledger_grpc_reader::MAX_BATCH_GET_OBJECTS;
36use crate::ledger_grpc_reader::MAX_BATCH_GET_TRANSACTIONS;
37use crate::objects::VersionedObjectKey;
38use crate::transactions::ProtoEffectsKey;
39use crate::transactions::TransactionKey;
40use crate::transactions::TransactionTimestampKey;
41
42/// Arguments for configuring KV store access via the Ledger gRPC service.
43#[derive(clap::Args, Debug, Clone, Default)]
44pub struct KvArgs {
45    /// Maximum gRPC decoding message size for KV responses, in bytes.
46    #[arg(long)]
47    pub kv_max_decoding_message_size: Option<usize>,
48
49    /// gRPC endpoint URL for the ledger service (e.g., archive.mainnet.sui.io)
50    #[arg(long)]
51    pub ledger_grpc_url: Option<Uri>,
52
53    /// Whether the configured ledger gRPC service serves the List APIs (e.g. bitmap-backed
54    /// transaction pagination). When unset, treated as `false`.
55    #[arg(long, alias = "experimental-query-apis")]
56    pub enable_list_apis: Option<bool>,
57
58    /// Time spent waiting for a request to the kv store to complete, in milliseconds.
59    #[arg(long)]
60    pub kv_statement_timeout_ms: Option<u64>,
61}
62
63/// A loader for point lookups against the Ledger gRPC service.
64/// Supported lookups:
65/// - Objects by id and version
66/// - Checkpoints by sequence number
67/// - Transactions by digest
68#[derive(Clone)]
69pub struct KvLoader(Arc<DataLoader<LedgerGrpcReader>>);
70
71/// A wrapper for the contents of a transaction, either from Ledger gRPC or just executed.
72#[allow(clippy::large_enum_variant)]
73#[derive(Clone)]
74pub enum TransactionContents {
75    LedgerGrpc(CheckpointedTransaction),
76    ExecutedTransaction(ExecutedTransactionData),
77}
78
79/// Transaction data from a gRPC execution or streaming response.
80#[derive(Clone)]
81pub struct ExecutedTransactionData {
82    pub effects: Box<TransactionEffects>,
83    pub events: Vec<Arc<Event>>,
84    pub transaction_data: Box<TransactionData>,
85    pub signatures: Vec<GenericSignature>,
86    pub balance_changes: Vec<grpc::BalanceChange>,
87    /// The proto TransactionEffects from gRPC, if available.
88    /// Contains fully-rendered effects with object types and clever errors.
89    pub proto_effects: Option<grpc::TransactionEffects>,
90    /// The proto Transaction from gRPC, if available.
91    /// Contains the fully-rendered transaction.
92    pub proto_transaction: Option<grpc::Transaction>,
93    /// Checkpoint timestamp. Set for streamed/checkpointed transactions, None for mutations.
94    pub timestamp_ms: Option<u64>,
95    /// Checkpoint sequence number. Set for streamed/checkpointed transactions, None for mutations.
96    pub cp_sequence_number: Option<u64>,
97}
98
99impl KvArgs {
100    /// `max_batch_get_transactions`/`max_batch_get_objects` may lower
101    /// `LedgerGrpcReader`'s batch-chunking size below `MAX_BATCH_GET_TRANSACTIONS`/
102    /// `MAX_BATCH_GET_OBJECTS`, never raise it above — a larger value would just be
103    /// rejected by the ledger gRPC/KV-RPC service, so it's clamped rather
104    /// than passed through.
105    pub async fn ledger_grpc_reader(
106        &self,
107        prefix: Option<&str>,
108        registry: &Registry,
109        max_batch_get_transactions: Option<usize>,
110        max_batch_get_objects: Option<usize>,
111    ) -> anyhow::Result<Option<LedgerGrpcReader>> {
112        let Some(ledger_grpc_url) = self.ledger_grpc_url.as_ref() else {
113            return Ok(None);
114        };
115
116        Ok(Some(
117            LedgerGrpcReader::new(
118                ledger_grpc_url.clone(),
119                self.ledger_grpc_args(),
120                prefix,
121                registry,
122                max_batch_get_transactions
123                    .unwrap_or(MAX_BATCH_GET_TRANSACTIONS)
124                    .min(MAX_BATCH_GET_TRANSACTIONS),
125                max_batch_get_objects
126                    .unwrap_or(MAX_BATCH_GET_OBJECTS)
127                    .min(MAX_BATCH_GET_OBJECTS),
128            )
129            .await?,
130        ))
131    }
132
133    /// Construct a streaming list reader when the operator has opted in via
134    /// `enable_list_apis` AND a ledger gRPC URL is configured. Returns `None`
135    /// otherwise. Reuses the same channel settings as the v2 `ledger_grpc_reader`.
136    pub async fn alpha_ledger_grpc_reader(
137        &self,
138        prefix: Option<&str>,
139        registry: &Registry,
140    ) -> anyhow::Result<Option<AlphaLedgerGrpcReader>> {
141        if !self.enable_list_apis.unwrap_or(false) {
142            return Ok(None);
143        }
144        let Some(ledger_grpc_url) = self.ledger_grpc_url.as_ref() else {
145            return Ok(None);
146        };
147
148        Ok(Some(
149            AlphaLedgerGrpcReader::new(
150                ledger_grpc_url.clone(),
151                self.ledger_grpc_args(),
152                prefix,
153                registry,
154            )
155            .await?,
156        ))
157    }
158
159    fn ledger_grpc_args(&self) -> LedgerGrpcArgs {
160        LedgerGrpcArgs::new(
161            self.kv_statement_timeout_ms,
162            self.kv_max_decoding_message_size,
163        )
164    }
165}
166
167impl KvLoader {
168    pub fn new(ledger_grpc: LedgerGrpcReader) -> Self {
169        Self(Arc::new(ledger_grpc.as_data_loader()))
170    }
171
172    pub async fn load_one_object(
173        &self,
174        id: ObjectID,
175        version: u64,
176    ) -> Result<Option<Object>, Error> {
177        self.0.load_one(VersionedObjectKey(id, version)).await
178    }
179
180    pub async fn load_many_objects(
181        &self,
182        keys: Vec<VersionedObjectKey>,
183    ) -> Result<HashMap<VersionedObjectKey, Object>, Error> {
184        self.0.load_many(keys).await
185    }
186
187    pub async fn load_one_checkpoint(
188        &self,
189        sequence_number: u64,
190    ) -> Result<
191        Option<(
192            CheckpointSummary,
193            CheckpointContents,
194            AuthorityQuorumSignInfo<true>,
195        )>,
196        Error,
197    > {
198        self.0.load_one(CheckpointKey(sequence_number)).await
199    }
200
201    /// Resolve a checkpoint digest to its sequence number. Returns `None` if the digest is not
202    /// found. Used by `Query.checkpoint(digest:)` to translate a caller-supplied digest into the
203    /// sequence number that downstream resolvers consume.
204    ///
205    /// Calls the reader directly rather than through the `DataLoader`, since the ledger gRPC
206    /// backend only supports single-digest lookup — `DataLoader` can't add real batching here; it
207    /// would only fan keys out into N parallel backend requests.
208    pub async fn load_one_checkpoint_seq_by_digest(
209        &self,
210        digest: CheckpointDigest,
211    ) -> Result<Option<u64>, Error> {
212        self.0
213            .loader()
214            .checkpoint_seq_by_digest(digest)
215            .await
216            .map_err(Error::from)
217    }
218
219    pub async fn load_one_transaction(
220        &self,
221        digest: TransactionDigest,
222    ) -> Result<Option<TransactionContents>, Error> {
223        Ok(self
224            .0
225            .load_one(TransactionKey(digest))
226            .await?
227            .map(TransactionContents::LedgerGrpc))
228    }
229
230    pub async fn load_one_transaction_timestamp(
231        &self,
232        digest: TransactionDigest,
233    ) -> Result<Option<u64>, Error> {
234        self.0.load_one(TransactionTimestampKey(digest)).await
235    }
236
237    /// Load a transaction's effects as rendered by the ledger service.
238    pub async fn load_one_rendered_effects(
239        &self,
240        digest: TransactionDigest,
241    ) -> Result<Option<grpc::TransactionEffects>, Error> {
242        self.0.load_one(ProtoEffectsKey(digest)).await
243    }
244
245    pub async fn load_many_transaction_events(
246        &self,
247        digests: Vec<TransactionDigest>,
248    ) -> Result<HashMap<TransactionDigest, grpc::ExecutedTransaction>, Arc<Error>> {
249        let keys = digests
250            .iter()
251            .map(|d| TransactionEventsKey(*d))
252            .collect::<Vec<_>>();
253
254        Ok(self
255            .0
256            .load_many(keys)
257            .await?
258            .into_iter()
259            .map(|(key, data)| (key.0, data))
260            .collect())
261    }
262
263    pub async fn load_many_transactions(
264        &self,
265        digests: Vec<TransactionDigest>,
266    ) -> Result<HashMap<TransactionDigest, TransactionContents>, Arc<Error>> {
267        let keys = digests
268            .iter()
269            .map(|d| TransactionKey(*d))
270            .collect::<Vec<_>>();
271
272        Ok(self
273            .0
274            .load_many(keys)
275            .await?
276            .into_iter()
277            .map(|(key, txn)| (key.0, TransactionContents::LedgerGrpc(txn)))
278            .collect())
279    }
280}
281
282impl TransactionContents {
283    pub fn from_executed_transaction(
284        executed_transaction: &grpc::ExecutedTransaction,
285        transaction_data: TransactionData,
286        signatures: Vec<GenericSignature>,
287    ) -> anyhow::Result<Self> {
288        // Parse effects from BCS
289        let effects: TransactionEffects = executed_transaction
290            .effects
291            .as_ref()
292            .and_then(|effects| effects.bcs.as_ref())
293            .context("Effects BCS should be present")?
294            .deserialize()
295            .context("Effects BCS should be valid")?;
296
297        // Parse events from BCS if present, defaulting to empty when absent.
298        let events: Vec<Arc<Event>> = executed_transaction
299            .events
300            .as_ref()
301            .and_then(|events| events.bcs.as_ref())
302            .map(|bcs| bcs.deserialize().context("Events BCS should be valid"))
303            .transpose()?
304            .map(|events: TransactionEvents| events.data.into_iter().map(Arc::new).collect())
305            .unwrap_or_default();
306
307        let balance_changes = executed_transaction.balance_changes.clone();
308
309        // Store the proto effects and transaction for JSON serialization
310        let proto_effects = executed_transaction.effects.clone();
311        let proto_transaction = executed_transaction.transaction.clone();
312
313        Ok(Self::ExecutedTransaction(ExecutedTransactionData {
314            effects: Box::new(effects),
315            events,
316            transaction_data: Box::new(transaction_data),
317            signatures,
318            balance_changes,
319            proto_effects,
320            proto_transaction,
321            timestamp_ms: None,
322            cp_sequence_number: None,
323        }))
324    }
325
326    /// A minimal instance whose `digest()` returns `digest`, for tests that only need identity. All
327    /// other accessors resolve to empty or absent values.
328    #[cfg(feature = "testing")]
329    pub fn for_test(digest: TransactionDigest) -> Self {
330        let mut effects = TransactionEffects::default();
331        *effects.transaction_digest_mut_for_testing() = digest;
332
333        let pt = sui_types::programmable_transaction_builder::ProgrammableTransactionBuilder::new()
334            .finish();
335        let transaction_data = TransactionData::new_programmable(
336            sui_types::base_types::SuiAddress::ZERO,
337            vec![],
338            pt,
339            0,
340            0,
341        );
342
343        Self::ExecutedTransaction(ExecutedTransactionData {
344            effects: Box::new(effects),
345            events: vec![],
346            transaction_data: Box::new(transaction_data),
347            signatures: vec![],
348            balance_changes: vec![],
349            proto_effects: None,
350            proto_transaction: None,
351            timestamp_ms: None,
352            cp_sequence_number: None,
353        })
354    }
355
356    pub fn data(&self) -> anyhow::Result<TransactionData> {
357        match self {
358            Self::LedgerGrpc(txn) => Ok(txn.transaction_data.as_ref().clone()),
359            Self::ExecutedTransaction(tx) => Ok(tx.transaction_data.as_ref().clone()),
360        }
361    }
362
363    pub fn digest(&self) -> anyhow::Result<TransactionDigest> {
364        match self {
365            Self::LedgerGrpc(txn) => Ok(*txn.effects.as_ref().transaction_digest()),
366            Self::ExecutedTransaction(tx) => Ok(*tx.effects.as_ref().transaction_digest()),
367        }
368    }
369
370    pub fn effects_digest(&self) -> anyhow::Result<TransactionEffectsDigest> {
371        match self {
372            Self::LedgerGrpc(txn) => Ok(txn.effects.digest()),
373            Self::ExecutedTransaction(tx) => Ok(tx.effects.digest()),
374        }
375    }
376
377    pub fn signatures(&self) -> anyhow::Result<Vec<GenericSignature>> {
378        match self {
379            Self::LedgerGrpc(txn) => Ok(txn.signatures.clone()),
380            Self::ExecutedTransaction(tx) => Ok(tx.signatures.clone()),
381        }
382    }
383
384    pub fn effects(&self) -> anyhow::Result<TransactionEffects> {
385        match self {
386            Self::LedgerGrpc(txn) => Ok(txn.effects.as_ref().clone()),
387            Self::ExecutedTransaction(tx) => Ok(tx.effects.as_ref().clone()),
388        }
389    }
390
391    /// Returns the events for this transaction. Each `Event` is wrapped in an `Arc` so
392    /// callers fanning the same transaction out to multiple consumers (e.g., subscription
393    /// resolvers serving different subscribers) share the underlying event allocation
394    /// rather than each performing a deep clone.
395    pub fn events(&self) -> anyhow::Result<Vec<Arc<Event>>> {
396        fn wrap(events: Vec<Event>) -> Vec<Arc<Event>> {
397            events.into_iter().map(Arc::new).collect()
398        }
399        match self {
400            Self::LedgerGrpc(txn) => Ok(wrap(txn.events.clone().unwrap_or_default())),
401            Self::ExecutedTransaction(tx) => Ok(tx.events.clone()),
402        }
403    }
404
405    pub fn balance_changes(&self) -> &[grpc::BalanceChange] {
406        match self {
407            Self::ExecutedTransaction(tx) => &tx.balance_changes,
408            Self::LedgerGrpc(txn) => &txn.balance_changes,
409        }
410    }
411
412    /// The proto TransactionEffects cached from a gRPC execution or streaming response, if any.
413    pub fn cached_proto_effects(&self) -> Option<&grpc::TransactionEffects> {
414        match self {
415            Self::ExecutedTransaction(tx) => tx.proto_effects.as_ref(),
416            Self::LedgerGrpc(_) => None,
417        }
418    }
419
420    /// Returns the proto TransactionEffects.
421    ///
422    /// Prefers the proto cached from an execution or streaming response (rendered by the fullnode).
423    /// Otherwise retrieve the rendered effects from kv.
424    pub async fn proto_effects(
425        &self,
426        kv_loader: &KvLoader,
427    ) -> anyhow::Result<grpc::TransactionEffects> {
428        if let Some(proto) = self.cached_proto_effects() {
429            return Ok(proto.clone());
430        }
431
432        // TODO: fullnode-rendered protos also include clever error rendering, which sui-kv-rpc does
433        // not implement (though it has the package resolver required).
434
435        match kv_loader
436            .load_one_rendered_effects(self.digest()?)
437            .await
438            .context("Failed to fetch rendered effects")?
439        {
440            Some(proto) => Ok(proto),
441            None => Ok(self.effects()?.into()),
442        }
443    }
444
445    /// Returns the proto Transaction.
446    ///
447    /// For ExecutedTransaction, returns the cached proto from gRPC.
448    /// For other sources, converts native transaction to proto.
449    pub fn proto_transaction(&self) -> anyhow::Result<grpc::Transaction> {
450        match self {
451            Self::ExecutedTransaction(tx) => {
452                // Use cached proto if available, otherwise convert from native
453                if let Some(proto) = &tx.proto_transaction {
454                    Ok(proto.clone())
455                } else {
456                    Ok(self.data()?.into())
457                }
458            }
459            Self::LedgerGrpc(_) => Ok(self.data()?.into()),
460        }
461    }
462
463    pub fn raw_transaction(&self) -> anyhow::Result<Vec<u8>> {
464        match self {
465            Self::LedgerGrpc(txn) => bcs::to_bytes(txn.transaction_data.as_ref())
466                .context("Failed to serialize transaction"),
467            Self::ExecutedTransaction(tx) => bcs::to_bytes(tx.transaction_data.as_ref())
468                .context("Failed to serialize transaction"),
469        }
470    }
471
472    pub fn raw_effects(&self) -> anyhow::Result<Vec<u8>> {
473        match self {
474            Self::LedgerGrpc(txn) => {
475                bcs::to_bytes(txn.effects.as_ref()).context("Failed to serialize effects")
476            }
477            Self::ExecutedTransaction(tx) => {
478                bcs::to_bytes(tx.effects.as_ref()).context("Failed to serialize effects")
479            }
480        }
481    }
482
483    pub fn timestamp_ms(&self) -> Option<u64> {
484        match self {
485            Self::LedgerGrpc(txn) => txn.timestamp_ms,
486            Self::ExecutedTransaction(tx) => tx.timestamp_ms,
487        }
488    }
489
490    pub fn cp_sequence_number(&self) -> Option<u64> {
491        match self {
492            Self::LedgerGrpc(txn) => txn.cp_sequence_number,
493            Self::ExecutedTransaction(tx) => tx.cp_sequence_number,
494        }
495    }
496}
sui_indexer_alt_reader/kv_loader.rs

sui_indexer_alt_reader/
kv_loader.rs
