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

iota_indexer/models/
transactions.rs
