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(crate) 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(crate) 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
205pub type StoredTransactionEvents = Vec<Option<Vec<u8>>>;
206
207#[derive(Debug, Queryable)]
208pub struct TxSeq {
209    pub seq: i64,
210}
211
212impl Default for TxSeq {
213    fn default() -> Self {
214        Self { seq: -1 }
215    }
216}
217
218#[derive(Clone, Debug, Queryable)]
219pub struct StoredTransactionTimestamp {
220    pub tx_sequence_number: i64,
221    pub timestamp_ms: i64,
222}
223
224#[derive(Clone, Debug, Queryable)]
225pub struct StoredTransactionCheckpoint {
226    pub tx_sequence_number: i64,
227    pub checkpoint_sequence_number: i64,
228}
229
230#[derive(Clone, Debug, Queryable)]
231pub struct StoredTransactionSuccessCommandCount {
232    pub tx_sequence_number: i64,
233    pub checkpoint_sequence_number: i64,
234    pub success_command_count: i16,
235    pub timestamp_ms: i64,
236}
237
238impl From<&IndexedTransaction> for StoredTransaction {
239    fn from(tx: &IndexedTransaction) -> Self {
240        StoredTransaction {
241            tx_sequence_number: tx.tx_sequence_number as i64,
242            transaction_digest: tx.tx_digest.into_inner().to_vec(),
243            raw_transaction: bcs::to_bytes(&tx.sender_signed_data).unwrap(),
244            raw_effects: bcs::to_bytes(&tx.effects).unwrap(),
245            checkpoint_sequence_number: tx.checkpoint_sequence_number as i64,
246            object_changes: tx
247                .object_changes
248                .iter()
249                .map(|oc| Some(bcs::to_bytes(&oc).unwrap()))
250                .collect(),
251            balance_changes: tx
252                .balance_change
253                .iter()
254                .map(|bc| Some(bcs::to_bytes(&bc).unwrap()))
255                .collect(),
256            events: tx
257                .events
258                .iter()
259                .map(|e| Some(bcs::to_bytes(&e).unwrap()))
260                .collect(),
261            timestamp_ms: tx.timestamp_ms as i64,
262            transaction_kind: tx.transaction_kind as i16,
263            success_command_count: tx.successful_tx_num as i16,
264        }
265    }
266}
267
268impl StoredTransaction {
269    pub fn get_balance_len(&self) -> usize {
270        self.balance_changes.len()
271    }
272
273    pub fn get_balance_at_idx(&self, idx: usize) -> Option<Vec<u8>> {
274        self.balance_changes.get(idx).cloned().flatten()
275    }
276
277    pub fn get_object_len(&self) -> usize {
278        self.object_changes.len()
279    }
280
281    pub fn get_object_at_idx(&self, idx: usize) -> Option<Vec<u8>> {
282        self.object_changes.get(idx).cloned().flatten()
283    }
284
285    pub fn get_event_len(&self) -> usize {
286        self.events.len()
287    }
288
289    pub fn get_event_at_idx(&self, idx: usize) -> Option<Vec<u8>> {
290        self.events.get(idx).cloned().flatten()
291    }
292
293    /// True for checkpointed transactions, False for optimistically indexed
294    /// transactions
295    pub fn is_checkpointed_transaction(&self) -> bool {
296        self.checkpoint_sequence_number >= 0
297    }
298
299    pub async fn try_into_iota_transaction_block_response(
300        self,
301        options: IotaTransactionBlockResponseOptions,
302        package_resolver: Arc<Resolver<impl PackageStore>>,
303    ) -> IndexerResult<IotaTransactionBlockResponse> {
304        let options = options.clone();
305        let tx_digest =
306            TransactionDigest::try_from(self.transaction_digest.as_slice()).map_err(|e| {
307                IndexerError::PersistentStorageDataCorruption(format!(
308                    "Can't convert {:?} as tx_digest. Error: {e}",
309                    self.transaction_digest
310                ))
311            })?;
312
313        let timestamp_ms = self
314            .is_checkpointed_transaction()
315            .then_some(self.timestamp_ms as u64);
316        let checkpoint = self
317            .is_checkpointed_transaction()
318            .then_some(self.checkpoint_sequence_number as u64);
319
320        let transaction = if options.show_input {
321            let sender_signed_data = self.try_into_sender_signed_data()?;
322            let tx_block = IotaTransactionBlock::try_from_with_package_resolver(
323                sender_signed_data,
324                package_resolver.clone(),
325                tx_digest,
326            )
327            .await?;
328            Some(tx_block)
329        } else {
330            None
331        };
332
333        let effects = options
334            .show_effects
335            .then(|| self.try_into_iota_transaction_effects())
336            .transpose()?;
337
338        let raw_transaction = if options.show_raw_input {
339            self.raw_transaction
340        } else {
341            Default::default()
342        };
343
344        let events = if options.show_events {
345            let events = {
346                self
347                        .events
348                        .into_iter()
349                        .map(|event| match event {
350                            Some(event) => {
351                                let event: Event = bcs::from_bytes(&event).map_err(|e| {
352                                    IndexerError::PersistentStorageDataCorruption(format!(
353                                        "Can't convert event bytes into Event. tx_digest={tx_digest:?} Error: {e}"
354                                    ))
355                                })?;
356                                Ok(event)
357                            }
358                            None => Err(IndexerError::PersistentStorageDataCorruption(format!(
359                                "Event should not be null, tx_digest={tx_digest:?}"
360                            ))),
361                        })
362                        .collect::<Result<Vec<Event>, IndexerError>>()?
363            };
364            let tx_events = TransactionEvents { data: events };
365
366            tx_events_to_iota_tx_events(tx_events, package_resolver, tx_digest, timestamp_ms)
367                .await?
368        } else {
369            None
370        };
371
372        let object_changes = if options.show_object_changes {
373            let object_changes = {
374                self.object_changes.into_iter().map(|object_change| {
375                        match object_change {
376                            Some(object_change) => {
377                                let object_change: IndexedObjectChange = bcs::from_bytes(&object_change)
378                                    .map_err(|e| IndexerError::PersistentStorageDataCorruption(
379                                        format!("Can't convert object_change bytes into IndexedObjectChange. tx_digest={tx_digest:?} Error: {e}")
380                                    ))?;
381                                Ok(ObjectChange::from(object_change))
382                            }
383                            None => Err(IndexerError::PersistentStorageDataCorruption(format!("object_change should not be null, tx_digest={tx_digest:?}"))),
384                        }
385                    }).collect::<Result<Vec<ObjectChange>, IndexerError>>()?
386            };
387            Some(object_changes)
388        } else {
389            None
390        };
391
392        let balance_changes = if options.show_balance_changes {
393            let balance_changes = {
394                self.balance_changes.into_iter().map(|balance_change| {
395                        match balance_change {
396                            Some(balance_change) => {
397                                let balance_change: BalanceChange = bcs::from_bytes(&balance_change)
398                                    .map_err(|e| IndexerError::PersistentStorageDataCorruption(
399                                        format!("Can't convert balance_change bytes into BalanceChange. tx_digest={tx_digest:?} Error: {e}")
400                                    ))?;
401                                Ok(balance_change)
402                            }
403                            None => Err(IndexerError::PersistentStorageDataCorruption(format!("object_change should not be null, tx_digest={tx_digest:?}"))),
404                        }
405                    }).collect::<Result<Vec<BalanceChange>, IndexerError>>()?
406            };
407            Some(balance_changes)
408        } else {
409            None
410        };
411
412        let raw_effects = if options.show_raw_effects {
413            self.raw_effects
414        } else {
415            Default::default()
416        };
417
418        Ok(IotaTransactionBlockResponse {
419            digest: tx_digest,
420            transaction,
421            raw_transaction,
422            effects,
423            events,
424            object_changes,
425            balance_changes,
426            timestamp_ms,
427            checkpoint,
428            confirmed_local_execution: None,
429            errors: vec![],
430            raw_effects,
431        })
432    }
433
434    fn try_into_sender_signed_data(&self) -> IndexerResult<SenderSignedData> {
435        let sender_signed_data: SenderSignedData =
436            bcs::from_bytes(&self.raw_transaction).map_err(|e| {
437                IndexerError::PersistentStorageDataCorruption(format!(
438                    "Can't convert raw_transaction of {} into SenderSignedData. Error: {e}",
439                    self.tx_sequence_number
440                ))
441            })?;
442        Ok(sender_signed_data)
443    }
444
445    pub fn try_into_iota_transaction_effects(&self) -> IndexerResult<IotaTransactionBlockEffects> {
446        let effects: TransactionEffects = bcs::from_bytes(&self.raw_effects).map_err(|e| {
447            IndexerError::PersistentStorageDataCorruption(format!(
448                "Can't convert raw_effects of {} into TransactionEffects. Error: {e}",
449                self.tx_sequence_number
450            ))
451        })?;
452        let effects = IotaTransactionBlockEffects::try_from(effects)?;
453        Ok(effects)
454    }
455
456    /// Check if this is the genesis transaction relying on the global ordering.
457    pub fn is_genesis(&self) -> bool {
458        self.tx_sequence_number == 0
459    }
460}
461
462pub fn stored_events_to_events(
463    stored_events: StoredTransactionEvents,
464) -> Result<Vec<Event>, IndexerError> {
465    stored_events
466        .into_iter()
467        .map(|event| match event {
468            Some(event) => {
469                let event: Event = bcs::from_bytes(&event).map_err(|e| {
470                    IndexerError::PersistentStorageDataCorruption(format!(
471                        "Can't convert event bytes into Event. Error: {e}",
472                    ))
473                })?;
474                Ok(event)
475            }
476            None => Err(IndexerError::PersistentStorageDataCorruption(
477                "Event should not be null".to_string(),
478            )),
479        })
480        .collect::<Result<Vec<Event>, IndexerError>>()
481}
482
483pub async fn tx_events_to_iota_tx_events(
484    tx_events: TransactionEvents,
485    package_resolver: Arc<Resolver<impl PackageStore>>,
486    tx_digest: TransactionDigest,
487    timestamp: Option<u64>,
488) -> Result<Option<IotaTransactionBlockEvents>, IndexerError> {
489    let mut iota_event_futures = vec![];
490    let tx_events_data_len = tx_events.data.len();
491    for tx_event in tx_events.data.clone() {
492        let package_resolver_clone = package_resolver.clone();
493        iota_event_futures.push(tokio::task::spawn(async move {
494            let resolver = package_resolver_clone;
495            resolver
496                .type_layout(TypeTag::Struct(Box::new(tx_event.type_.clone())))
497                .await
498        }));
499    }
500    let event_move_type_layouts = futures::future::join_all(iota_event_futures)
501        .await
502        .into_iter()
503        .collect::<Result<Vec<_>, _>>()?
504        .into_iter()
505        .collect::<Result<Vec<_>, _>>()
506        .map_err(|e| {
507            IndexerError::ResolveMoveStruct(format!(
508                "Failed to convert to iota event with Error: {e}",
509            ))
510        })?;
511    let event_move_datatype_layouts = event_move_type_layouts
512        .into_iter()
513        .filter_map(|move_type_layout| match move_type_layout {
514            MoveTypeLayout::Struct(s) => Some(MoveDatatypeLayout::Struct(s)),
515            MoveTypeLayout::Enum(e) => Some(MoveDatatypeLayout::Enum(e)),
516            _ => None,
517        })
518        .collect::<Vec<_>>();
519    assert!(tx_events_data_len == event_move_datatype_layouts.len());
520    let iota_events = tx_events
521        .data
522        .into_iter()
523        .enumerate()
524        .zip(event_move_datatype_layouts)
525        .map(|((seq, tx_event), move_datatype_layout)| {
526            IotaEvent::try_from(
527                tx_event,
528                tx_digest,
529                seq as u64,
530                timestamp,
531                move_datatype_layout,
532            )
533        })
534        .collect::<Result<Vec<_>, _>>()?;
535    let iota_tx_events = IotaTransactionBlockEvents { data: iota_events };
536    Ok(Some(iota_tx_events))
537}
iota_indexer/models/transactions.rs

iota_indexer/models/
transactions.rs
