sui_indexer_alt_reader/
transactions.rs

1// Copyright (c) Mysten Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4use std::collections::HashMap;
5
6use anyhow::Context;
7use prost_types::FieldMask;
8use sui_rpc::field::FieldMaskUtil;
9use sui_rpc::proto::proto_to_timestamp_ms;
10use sui_rpc::proto::sui::rpc::v2 as proto;
11use sui_types::digests::TransactionDigest;
12
13use crate::error::Error;
14use crate::ledger_grpc_reader::CheckpointedTransaction;
15use crate::ledger_grpc_reader::ChunkedLoader;
16use crate::ledger_grpc_reader::LedgerGrpcReader;
17
18/// Key for fetching transaction contents (TransactionData, Effects, and Events) by digest.
19#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
20pub struct TransactionKey(pub TransactionDigest);
21
22/// Key for fetching just the checkpoint timestamp of a transaction by digest.
23#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
24pub struct TransactionTimestampKey(pub TransactionDigest);
25
26/// Key for fetching a transaction's effects as rendered by the server, which carries additional
27/// information that cannot be derived from the effects BCS client-side (object type annotations,
28/// runtime-loaded objects).
29#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
30pub struct ProtoEffectsKey(pub TransactionDigest);
31
32#[async_trait::async_trait]
33impl ChunkedLoader<TransactionKey> for LedgerGrpcReader {
34    type Value = CheckpointedTransaction;
35    type Error = Error;
36
37    fn chunk_size(&self) -> usize {
38        self.max_batch_get_transactions()
39    }
40
41    async fn load_chunk(
42        &self,
43        keys: &[TransactionKey],
44    ) -> Result<HashMap<TransactionKey, CheckpointedTransaction>, Error> {
45        let digests = keys.iter().map(|key| key.0.to_string()).collect();
46
47        let mut request = proto::BatchGetTransactionsRequest::default();
48        request.digests = digests;
49        request.read_mask = Some(CheckpointedTransaction::read_mask());
50
51        let batch_response = self.batch_get_transactions(request).await?;
52
53        let mut results = HashMap::new();
54        for tx_result in batch_response.transactions {
55            if let Some(proto::get_transaction_result::Result::Transaction(executed)) =
56                tx_result.result
57            {
58                let transaction = CheckpointedTransaction::try_from(&executed)?;
59                results.insert(
60                    TransactionKey(transaction.transaction_data.digest()),
61                    transaction,
62                );
63            }
64        }
65        Ok(results)
66    }
67}
68
69#[async_trait::async_trait]
70impl ChunkedLoader<TransactionTimestampKey> for LedgerGrpcReader {
71    type Value = u64;
72    type Error = Error;
73
74    fn chunk_size(&self) -> usize {
75        self.max_batch_get_transactions()
76    }
77
78    async fn load_chunk(
79        &self,
80        keys: &[TransactionTimestampKey],
81    ) -> Result<HashMap<TransactionTimestampKey, u64>, Error> {
82        let digests = keys.iter().map(|key| key.0.to_string()).collect();
83
84        let mut request = proto::BatchGetTransactionsRequest::default();
85        request.digests = digests;
86        request.read_mask = Some(FieldMask::from_paths(["digest", "timestamp"]));
87
88        let batch_response = self.batch_get_transactions(request).await?;
89
90        let mut results = HashMap::new();
91        for tx_result in batch_response.transactions {
92            let Some(proto::get_transaction_result::Result::Transaction(executed)) =
93                tx_result.result
94            else {
95                continue;
96            };
97
98            let digest: TransactionDigest = executed
99                .digest
100                .as_deref()
101                .context("BatchGetTransactions response missing digest")?
102                .parse()
103                .context("Failed to parse transaction digest")?;
104
105            // Transactions served by the ledger service are always checkpointed, but tolerate a
106            // missing timestamp by treating the transaction as not found.
107            let Some(timestamp) = executed.timestamp else {
108                continue;
109            };
110            let timestamp_ms = proto_to_timestamp_ms(timestamp)
111                .map_err(|e| anyhow::anyhow!("Failed to parse timestamp: {}", e))?;
112
113            results.insert(TransactionTimestampKey(digest), timestamp_ms);
114        }
115        Ok(results)
116    }
117}
118
119#[async_trait::async_trait]
120impl ChunkedLoader<ProtoEffectsKey> for LedgerGrpcReader {
121    type Value = proto::TransactionEffects;
122    type Error = Error;
123
124    fn chunk_size(&self) -> usize {
125        self.max_batch_get_transactions()
126    }
127
128    async fn load_chunk(
129        &self,
130        keys: &[ProtoEffectsKey],
131    ) -> Result<HashMap<ProtoEffectsKey, Self::Value>, Error> {
132        if keys.is_empty() {
133            return Ok(HashMap::new());
134        }
135
136        let digests = keys.iter().map(|key| key.0.to_string()).collect();
137
138        let mut request = proto::BatchGetTransactionsRequest::default();
139        request.digests = digests;
140        request.read_mask = Some(FieldMask::from_paths(["digest", "effects"]));
141
142        let batch_response = self.batch_get_transactions(request).await?;
143
144        let mut results = HashMap::new();
145        for tx_result in batch_response.transactions {
146            let Some(proto::get_transaction_result::Result::Transaction(executed)) =
147                tx_result.result
148            else {
149                continue;
150            };
151
152            let digest: TransactionDigest = executed
153                .digest
154                .as_deref()
155                .context("BatchGetTransactions response missing digest")?
156                .parse()
157                .context("Failed to parse transaction digest")?;
158
159            let Some(effects) = executed.effects else {
160                continue;
161            };
162
163            results.insert(ProtoEffectsKey(digest), effects);
164        }
165        Ok(results)
166    }
167}
168
169#[cfg(test)]
170mod tests {
171    use async_graphql::dataloader::Loader;
172
173    use super::*;
174    use crate::ledger_grpc_reader::test_support::assert_chunked;
175    use crate::ledger_grpc_reader::test_support::mock_reader;
176
177    #[tokio::test]
178    async fn transaction_load_chunks_oversized_batches() {
179        let (reader, mock, server) = mock_reader().await;
180        let limit = reader.max_batch_get_transactions();
181
182        let keys: Vec<TransactionKey> = (0..limit + 50)
183            .map(|_| TransactionKey(TransactionDigest::random()))
184            .collect();
185
186        let result = reader.load(&keys).await.expect("load should succeed");
187        assert!(result.is_empty());
188
189        let expected: Vec<String> = keys.iter().map(|key| key.0.to_string()).collect();
190        assert_chunked(mock.transaction_batches(), limit, &expected);
191
192        server.abort();
193    }
194
195    #[tokio::test]
196    async fn transaction_timestamp_load_chunks_oversized_batches() {
197        let (reader, mock, server) = mock_reader().await;
198        let limit = reader.max_batch_get_transactions();
199
200        let keys: Vec<TransactionTimestampKey> = (0..limit + 50)
201            .map(|_| TransactionTimestampKey(TransactionDigest::random()))
202            .collect();
203
204        let result = reader.load(&keys).await.expect("load should succeed");
205        assert!(result.is_empty());
206
207        let expected: Vec<String> = keys.iter().map(|key| key.0.to_string()).collect();
208        assert_chunked(mock.transaction_batches(), limit, &expected);
209
210        server.abort();
211    }
212}
sui_indexer_alt_reader/transactions.rs

sui_indexer_alt_reader/
transactions.rs
