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

iota_indexer/models/
transactions.rs
