iota_indexer/models/
transactions.rs

1// Copyright (c) Mysten Labs, Inc.
2// Modifications Copyright (c) 2024 IOTA Stiftung
3// SPDX-License-Identifier: Apache-2.0
4
5use std::sync::Arc;
6
7use diesel::prelude::*;
8use iota_json_rpc_types::{
9    BalanceChange, IotaEvent, IotaExecutionStatus, IotaTransactionBlock,
10    IotaTransactionBlockEffects, IotaTransactionBlockEffectsAPI, IotaTransactionBlockEvents,
11    IotaTransactionBlockResponse, IotaTransactionBlockResponseOptions, ObjectChange,
12};
13use iota_package_resolver::{PackageStore, Resolver};
14use iota_sdk_types::{
15    Event, SenderSignedTransaction, TransactionDigest, TransactionEffects, TransactionEvents,
16    TypeTag,
17};
18use move_core_types::annotated_value::{MoveDatatypeLayout, MoveTypeLayout};
19#[cfg(feature = "shared_test_runtime")]
20use serde::Deserialize;
21
22use crate::{
23    errors::IndexerError,
24    schema::{optimistic_transactions, transactions, tx_global_order},
25    types::{IndexedBalanceChange, IndexedObjectChange, IndexedTransaction, IndexerResult},
26};
27
28#[derive(Clone, Debug, Queryable, Insertable, QueryableByName, Selectable)]
29#[diesel(table_name = tx_global_order)]
30pub struct TxGlobalOrder {
31    /// Sequence number of transaction according to checkpoint ordering.
32    /// Set after transaction is checkpoint-indexed.
33    pub tx_sequence_number: Option<i64>,
34    pub tx_digest: Vec<u8>,
35    /// Monotonically increasing number that represents the order
36    /// of execution of optimistic transactions.
37    ///
38    /// Checkpointed transactions use [`CHECKPOINT_TX_OPTIMISTIC_SEQ`] (-1).
39    /// Optimistic transactions should set this value to `None`,
40    /// so that it is auto-generated on the database.
41    #[diesel(deserialize_as = i64)]
42    pub optimistic_sequence_number: Option<i64>,
43}
44
45/// Value stored in `optimistic_sequence_number` for checkpointed
46/// transactions, to distinguish them from optimistic transactions
47/// which use positive auto-generated values.
48pub const CHECKPOINT_TX_OPTIMISTIC_SEQ: i64 = -1;
49
50impl From<&IndexedTransaction> for TxGlobalOrder {
51    fn from(tx: &IndexedTransaction) -> Self {
52        Self {
53            tx_sequence_number: Some(tx.tx_sequence_number as i64),
54            tx_digest: tx.tx_digest.into_bytes().to_vec(),
55            optimistic_sequence_number: Some(CHECKPOINT_TX_OPTIMISTIC_SEQ),
56        }
57    }
58}
59
60#[derive(Clone, Debug, Queryable, Insertable, QueryableByName, Selectable)]
61#[diesel(table_name = transactions)]
62#[cfg_attr(feature = "shared_test_runtime", derive(Deserialize))]
63pub struct StoredTransaction {
64    /// The index of the transaction in the global ordering that starts
65    /// from genesis.
66    pub tx_sequence_number: i64,
67    pub transaction_digest: Vec<u8>,
68    pub raw_transaction: Vec<u8>,
69    pub raw_effects: Vec<u8>,
70    pub checkpoint_sequence_number: i64,
71    pub timestamp_ms: i64,
72    pub object_changes: Vec<Option<Vec<u8>>>,
73    pub balance_changes: Vec<Option<Vec<u8>>>,
74    pub events: Vec<Option<Vec<u8>>>,
75    pub transaction_kind: i16,
76    pub success_command_count: i16,
77}
78
79#[derive(Clone, Debug, Queryable, Insertable, QueryableByName, Selectable)]
80#[diesel(table_name = optimistic_transactions)]
81pub struct OptimisticTransaction {
82    pub optimistic_sequence_number: i64,
83    pub transaction_digest: Vec<u8>,
84    pub raw_transaction: Vec<u8>,
85    pub raw_effects: Vec<u8>,
86    pub object_changes: Vec<Option<Vec<u8>>>,
87    pub balance_changes: Vec<Option<Vec<u8>>>,
88    pub events: Vec<Option<Vec<u8>>>,
89    pub transaction_kind: i16,
90    pub success_command_count: i16,
91}
92
93impl From<OptimisticTransaction> for StoredTransaction {
94    fn from(tx: OptimisticTransaction) -> Self {
95        StoredTransaction {
96            tx_sequence_number: tx.optimistic_sequence_number,
97            transaction_digest: tx.transaction_digest,
98            raw_transaction: tx.raw_transaction,
99            raw_effects: tx.raw_effects,
100            checkpoint_sequence_number: -1,
101            timestamp_ms: -1,
102            object_changes: tx.object_changes,
103            balance_changes: tx.balance_changes,
104            events: tx.events,
105            transaction_kind: tx.transaction_kind,
106            success_command_count: tx.success_command_count,
107        }
108    }
109}
110
111impl OptimisticTransaction {
112    pub fn from_stored(stored: StoredTransaction) -> Self {
113        OptimisticTransaction {
114            optimistic_sequence_number: stored.tx_sequence_number,
115            transaction_digest: stored.transaction_digest,
116            raw_transaction: stored.raw_transaction,
117            raw_effects: stored.raw_effects,
118            object_changes: stored.object_changes,
119            balance_changes: stored.balance_changes,
120            events: stored.events,
121            transaction_kind: stored.transaction_kind,
122            success_command_count: stored.success_command_count,
123        }
124    }
125
126    pub fn get_balance_len(&self) -> usize {
127        self.balance_changes.len()
128    }
129
130    pub fn get_balance_at_idx(&self, idx: usize) -> Option<Vec<u8>> {
131        self.balance_changes.get(idx).cloned().flatten()
132    }
133
134    pub fn get_object_len(&self) -> usize {
135        self.object_changes.len()
136    }
137
138    pub fn get_object_at_idx(&self, idx: usize) -> Option<Vec<u8>> {
139        self.object_changes.get(idx).cloned().flatten()
140    }
141
142    pub fn get_event_len(&self) -> usize {
143        self.events.len()
144    }
145
146    pub fn get_event_at_idx(&self, idx: usize) -> Option<Vec<u8>> {
147        self.events.get(idx).cloned().flatten()
148    }
149}
150
151pub type StoredTransactionEvents = Vec<Option<Vec<u8>>>;
152
153#[derive(Debug, Queryable)]
154pub struct TxSeq {
155    pub seq: i64,
156}
157
158impl Default for TxSeq {
159    fn default() -> Self {
160        Self { seq: -1 }
161    }
162}
163
164#[derive(Clone, Debug, Queryable)]
165pub struct StoredTransactionTimestamp {
166    pub tx_sequence_number: i64,
167    pub timestamp_ms: i64,
168}
169
170#[derive(Clone, Debug, Queryable)]
171pub struct StoredTransactionCheckpoint {
172    pub tx_sequence_number: i64,
173    pub checkpoint_sequence_number: i64,
174}
175
176#[derive(Clone, Debug, Queryable)]
177pub struct StoredTransactionSuccessCommandCount {
178    pub tx_sequence_number: i64,
179    pub checkpoint_sequence_number: i64,
180    pub success_command_count: i16,
181    pub timestamp_ms: i64,
182}
183
184impl From<&IndexedTransaction> for StoredTransaction {
185    fn from(tx: &IndexedTransaction) -> Self {
186        StoredTransaction {
187            tx_sequence_number: tx.tx_sequence_number as i64,
188            transaction_digest: tx.tx_digest.into_bytes().to_vec(),
189            raw_transaction: bcs::to_bytes(&tx.sender_signed_data).unwrap(),
190            raw_effects: bcs::to_bytes(&tx.effects).unwrap(),
191            checkpoint_sequence_number: tx.checkpoint_sequence_number as i64,
192            object_changes: tx
193                .object_changes
194                .iter()
195                .map(|oc| Some(bcs::to_bytes(&oc).unwrap()))
196                .collect(),
197            balance_changes: tx
198                .balance_change
199                .iter()
200                .map(|bc| Some(bcs::to_bytes(&bc).unwrap()))
201                .collect(),
202            events: tx
203                .events
204                .iter()
205                .map(|e| Some(bcs::to_bytes(&e).unwrap()))
206                .collect(),
207            timestamp_ms: tx.timestamp_ms as i64,
208            transaction_kind: tx.transaction_kind as i16,
209            success_command_count: tx.successful_tx_num as i16,
210        }
211    }
212}
213
214impl StoredTransaction {
215    pub fn get_balance_len(&self) -> usize {
216        self.balance_changes.len()
217    }
218
219    pub fn get_balance_at_idx(&self, idx: usize) -> Option<Vec<u8>> {
220        self.balance_changes.get(idx).cloned().flatten()
221    }
222
223    pub fn get_object_len(&self) -> usize {
224        self.object_changes.len()
225    }
226
227    pub fn get_object_at_idx(&self, idx: usize) -> Option<Vec<u8>> {
228        self.object_changes.get(idx).cloned().flatten()
229    }
230
231    pub fn get_event_len(&self) -> usize {
232        self.events.len()
233    }
234
235    pub fn get_event_at_idx(&self, idx: usize) -> Option<Vec<u8>> {
236        self.events.get(idx).cloned().flatten()
237    }
238
239    /// True for checkpointed transactions, False for optimistically indexed
240    /// transactions
241    pub fn is_checkpointed_transaction(&self) -> bool {
242        self.checkpoint_sequence_number >= 0
243    }
244
245    pub async fn try_into_iota_transaction_block_response(
246        self,
247        options: IotaTransactionBlockResponseOptions,
248        package_resolver: &Arc<Resolver<impl PackageStore>>,
249    ) -> IndexerResult<IotaTransactionBlockResponse> {
250        let options = options.clone();
251        let tx_digest =
252            TransactionDigest::from_bytes(self.transaction_digest.as_slice()).map_err(|e| {
253                IndexerError::PersistentStorageDataCorruption(format!(
254                    "Can't convert {:?} as tx_digest. Error: {e}",
255                    self.transaction_digest
256                ))
257            })?;
258
259        let timestamp_ms = self
260            .is_checkpointed_transaction()
261            .then_some(self.timestamp_ms as u64);
262        let checkpoint = self
263            .is_checkpointed_transaction()
264            .then_some(self.checkpoint_sequence_number as u64);
265
266        let transaction = if options.show_input {
267            let sender_signed_tx = self.try_into_sender_signed_transaction()?;
268            let tx_block = IotaTransactionBlock::try_from_with_package_resolver(
269                sender_signed_tx,
270                package_resolver,
271                tx_digest,
272            )
273            .await?;
274            Some(tx_block)
275        } else {
276            None
277        };
278
279        let effects = if options.show_effects {
280            Some(
281                self.try_into_iota_transaction_effects(package_resolver)
282                    .await?,
283            )
284        } else {
285            None
286        };
287
288        let raw_transaction = if options.show_raw_input {
289            self.raw_transaction
290        } else {
291            Default::default()
292        };
293
294        let events = if options.show_events {
295            let events = {
296                self
297                        .events
298                        .into_iter()
299                        .map(|event| match event {
300                            Some(event) => {
301                                let event: Event = bcs::from_bytes(&event).map_err(|e| {
302                                    IndexerError::PersistentStorageDataCorruption(format!(
303                                        "Can't convert event bytes into Event. tx_digest={tx_digest} Error: {e}"
304                                    ))
305                                })?;
306                                Ok(event)
307                            }
308                            None => Err(IndexerError::PersistentStorageDataCorruption(format!(
309                                "Event should not be null, tx_digest={tx_digest}"
310                            ))),
311                        })
312                        .collect::<Result<Vec<Event>, IndexerError>>()?
313            };
314            let tx_events = TransactionEvents(events);
315
316            Some(
317                tx_events_to_iota_tx_events(tx_events, package_resolver, tx_digest, timestamp_ms)
318                    .await?,
319            )
320        } else {
321            None
322        };
323
324        let object_changes = if options.show_object_changes {
325            let object_changes = {
326                self.object_changes.into_iter().map(|object_change| {
327                        match object_change {
328                            Some(object_change) => {
329                                let object_change: IndexedObjectChange = bcs::from_bytes(&object_change)
330                                    .map_err(|e| IndexerError::PersistentStorageDataCorruption(
331                                        format!("Can't convert object_change bytes into IndexedObjectChange. tx_digest={tx_digest} Error: {e}")
332                                    ))?;
333                                Ok(ObjectChange::from(object_change))
334                            }
335                            None => Err(IndexerError::PersistentStorageDataCorruption(format!("object_change should not be null, tx_digest={tx_digest}"))),
336                        }
337                    }).collect::<Result<Vec<ObjectChange>, IndexerError>>()?
338            };
339            Some(object_changes)
340        } else {
341            None
342        };
343
344        let balance_changes = if options.show_balance_changes {
345            let balance_changes = {
346                self.balance_changes.into_iter().map(|balance_change| {
347                        match balance_change {
348                            Some(balance_change) => {
349                                let balance_change: IndexedBalanceChange = bcs::from_bytes(&balance_change)
350                                    .map_err(|e| IndexerError::PersistentStorageDataCorruption(
351                                        format!("Can't convert balance_change bytes into IndexedBalanceChange. tx_digest={tx_digest} Error: {e}")
352                                    ))?;
353                                Ok(BalanceChange::from(balance_change))
354                            }
355                            None => Err(IndexerError::PersistentStorageDataCorruption(format!("balance_change should not be null, tx_digest={tx_digest}"))),
356                        }
357                    }).collect::<Result<Vec<BalanceChange>, IndexerError>>()?
358            };
359            Some(balance_changes)
360        } else {
361            None
362        };
363
364        let raw_effects = if options.show_raw_effects {
365            self.raw_effects
366        } else {
367            Default::default()
368        };
369
370        let errors = match effects.as_ref().map(|e| e.status()) {
371            Some(IotaExecutionStatus::Failure { error }) => vec![error.clone()],
372            _ => vec![],
373        };
374
375        Ok(IotaTransactionBlockResponse {
376            digest: tx_digest,
377            transaction,
378            raw_transaction,
379            effects,
380            events,
381            object_changes,
382            balance_changes,
383            timestamp_ms,
384            checkpoint,
385            confirmed_local_execution: None,
386            errors,
387            raw_effects,
388        })
389    }
390
391    pub fn try_into_sender_signed_transaction(&self) -> IndexerResult<SenderSignedTransaction> {
392        let sender_signed_tx: SenderSignedTransaction = bcs::from_bytes(&self.raw_transaction)
393            .map_err(|e| {
394                IndexerError::PersistentStorageDataCorruption(format!(
395                    "Can't convert raw_transaction of {} into SenderSignedTransaction. Error: {e}",
396                    self.tx_sequence_number
397                ))
398            })?;
399        Ok(sender_signed_tx)
400    }
401
402    pub async fn try_into_iota_transaction_effects(
403        &self,
404        package_resolver: &Arc<Resolver<impl PackageStore>>,
405    ) -> IndexerResult<IotaTransactionBlockEffects> {
406        let effects: TransactionEffects = bcs::from_bytes(&self.raw_effects).map_err(|e| {
407            IndexerError::PersistentStorageDataCorruption(format!(
408                "Can't convert raw_effects of {} into TransactionEffects. Error: {e}",
409                self.tx_sequence_number
410            ))
411        })?;
412        let effects =
413            IotaTransactionBlockEffects::from_native_with_clever_error(effects, package_resolver)
414                .await;
415        Ok(effects)
416    }
417
418    /// Check if this is the genesis transaction relying on the global ordering.
419    pub fn is_genesis(&self) -> bool {
420        self.tx_sequence_number == 0
421    }
422}
423
424pub fn stored_events_to_events(
425    stored_events: StoredTransactionEvents,
426) -> Result<Vec<Event>, IndexerError> {
427    stored_events
428        .into_iter()
429        .map(|event| match event {
430            Some(event) => {
431                let event: Event = bcs::from_bytes(&event).map_err(|e| {
432                    IndexerError::PersistentStorageDataCorruption(format!(
433                        "Can't convert event bytes into Event. Error: {e}",
434                    ))
435                })?;
436                Ok(event)
437            }
438            None => Err(IndexerError::PersistentStorageDataCorruption(
439                "Event should not be null".to_string(),
440            )),
441        })
442        .collect::<Result<Vec<Event>, IndexerError>>()
443}
444
445pub async fn tx_events_to_iota_tx_events(
446    mut tx_events: TransactionEvents,
447    package_resolver: &Arc<Resolver<impl PackageStore>>,
448    tx_digest: TransactionDigest,
449    timestamp: Option<u64>,
450) -> Result<IotaTransactionBlockEvents, IndexerError> {
451    let mut iota_event_futures = vec![];
452
453    for tx_event in tx_events.iter() {
454        let package_resolver_clone = package_resolver.clone();
455        let event_type = tx_event.struct_tag.clone();
456        iota_event_futures.push(tokio::task::spawn(async move {
457            let resolver = package_resolver_clone;
458            resolver
459                .type_layout(TypeTag::Struct(Box::new(event_type)))
460                .await
461        }));
462    }
463    let event_move_type_layouts = futures::future::join_all(iota_event_futures)
464        .await
465        .into_iter()
466        .collect::<Result<Vec<_>, _>>()?
467        .into_iter()
468        .collect::<Result<Vec<_>, _>>()
469        .map_err(|e| {
470            IndexerError::ResolveMoveStruct(format!(
471                "Failed to convert to iota event with Error: {e}",
472            ))
473        })?;
474    let event_move_datatype_layouts = event_move_type_layouts
475        .into_iter()
476        .filter_map(|move_type_layout| match move_type_layout {
477            MoveTypeLayout::Struct(s) => Some(MoveDatatypeLayout::Struct(s)),
478            MoveTypeLayout::Enum(e) => Some(MoveDatatypeLayout::Enum(e)),
479            _ => None,
480        })
481        .collect::<Vec<_>>();
482    assert!(tx_events.len() == event_move_datatype_layouts.len());
483    let iota_events = tx_events
484        .drain(..)
485        .enumerate()
486        .zip(event_move_datatype_layouts)
487        .map(|((seq, tx_event), move_datatype_layout)| {
488            IotaEvent::try_from(
489                tx_event,
490                tx_digest,
491                seq as u64,
492                timestamp,
493                move_datatype_layout,
494            )
495        })
496        .collect::<Result<Vec<_>, _>>()?;
497    let iota_tx_events = IotaTransactionBlockEvents { data: iota_events };
498    Ok(iota_tx_events)
499}
iota_indexer/models/transactions.rs

iota_indexer/models/
transactions.rs
