iota_storage/
write_path_pending_tx_log.rs

1// Copyright (c) Mysten Labs, Inc.
2// Modifications Copyright (c) 2024 IOTA Stiftung
3// SPDX-License-Identifier: Apache-2.0
4
5//! WritePathPendingTransactionLog is used in the transaction write path (e.g.
6//! in TransactionOrchestrator) for transaction submission processing. It helps
7//! to achieve:
8//! 1. At one time, a transaction is only processed once.
9//! 2. When Fullnode crashes and restarts, the pending transaction will be loaded and retried.
10
11use std::path::PathBuf;
12
13use iota_sdk_types::{SenderSignedTransaction, TransactionDigest};
14use iota_types::{
15    crypto::EmptySignInfo,
16    error::{IotaError, IotaResult},
17    message_envelope::TrustedEnvelope,
18    transaction::VerifiedTransaction,
19};
20use tracing::instrument;
21use typed_store::{
22    DBMapUtils,
23    rocks::{DBMap, MetricConf},
24    traits::Map,
25};
26
27use crate::mutex_table::MutexTable;
28
29pub type IsFirstRecord = bool;
30const NUM_SHARDS: usize = 4096;
31
32#[derive(DBMapUtils)]
33struct WritePathPendingTransactionTable {
34    logs: DBMap<TransactionDigest, TrustedEnvelope<SenderSignedTransaction, EmptySignInfo>>,
35}
36
37pub struct WritePathPendingTransactionLog {
38    pending_transactions: WritePathPendingTransactionTable,
39    mutex_table: MutexTable<TransactionDigest>,
40}
41
42impl WritePathPendingTransactionLog {
43    pub fn new(path: PathBuf) -> Self {
44        let pending_transactions = WritePathPendingTransactionTable::open_tables_read_write(
45            path,
46            MetricConf::new("pending_tx_log"),
47            None,
48            None,
49        );
50        Self {
51            pending_transactions,
52            mutex_table: MutexTable::new(NUM_SHARDS),
53        }
54    }
55
56    // Returns whether the table currently has this transaction in record.
57    // If not, write the transaction and return true; otherwise return false.
58    // Because the record will be cleaned up when the transaction finishes,
59    // even when it returns true, the callsite of this function should check
60    // the transaction status before doing anything, to avoid duplicates.
61    #[instrument(level = "trace", skip_all, fields(tx_digest = ?tx.digest()))]
62    pub async fn write_pending_transaction_maybe(
63        &self,
64        tx: &VerifiedTransaction,
65    ) -> IotaResult<IsFirstRecord> {
66        let tx_digest = tx.digest();
67        let _guard = self.mutex_table.acquire_lock(*tx_digest);
68        if self.pending_transactions.logs.contains_key(tx_digest)? {
69            Ok(false)
70        } else {
71            self.pending_transactions
72                .logs
73                .insert(tx_digest, tx.serializable_ref())?;
74            Ok(true)
75        }
76    }
77
78    // This function does not need to be behind a lock because:
79    // 1. there could be more than one callsite but the deletion is idempotent.
80    // 2. it does not race with the insert (`write_pending_transaction_maybe`) in a way that we
81    //    care. 2.a. for one transaction, `finish_transaction` shouldn't predate
82    //    `write_pending_transaction_maybe`. 2.b  for concurrent requests of one transaction, a call
83    //    to this function may happen in between hence making the second request thinks it is the
84    //    first record. It's preventable by checking this transaction again after the call of
85    //    `write_pending_transaction_maybe`.
86    pub fn finish_transaction(&self, tx: &TransactionDigest) -> IotaResult {
87        let mut write_batch = self.pending_transactions.logs.batch();
88        write_batch.delete_batch(&self.pending_transactions.logs, std::iter::once(tx))?;
89        write_batch.write().map_err(IotaError::from)
90    }
91
92    pub fn load_all_pending_transactions(&self) -> IotaResult<Vec<VerifiedTransaction>> {
93        Ok(self
94            .pending_transactions
95            .logs
96            .safe_iter()
97            .map(|item| item.map(|(_tx_digest, tx)| VerifiedTransaction::from(tx)))
98            .collect::<Result<Vec<_>, _>>()?)
99    }
100}
101
102#[cfg(test)]
103mod tests {
104    use std::collections::HashSet;
105
106    use anyhow;
107    use iota_types::utils::create_fake_transaction;
108
109    use super::*;
110
111    #[tokio::test]
112    async fn test_pending_tx_log_basic() -> anyhow::Result<()> {
113        let tmp_dir = iota_common::tempdir();
114        let pending_txes = WritePathPendingTransactionLog::new(tmp_dir.path().to_path_buf());
115        let tx = VerifiedTransaction::new_unchecked(create_fake_transaction());
116        let tx_digest = *tx.digest();
117        assert!(
118            pending_txes
119                .write_pending_transaction_maybe(&tx)
120                .await
121                .unwrap()
122        );
123        // The second write will return false
124        assert!(
125            !pending_txes
126                .write_pending_transaction_maybe(&tx)
127                .await
128                .unwrap()
129        );
130
131        let loaded_txes = pending_txes.load_all_pending_transactions()?;
132        assert_eq!(vec![tx], loaded_txes);
133
134        pending_txes.finish_transaction(&tx_digest).unwrap();
135        let loaded_txes = pending_txes.load_all_pending_transactions()?;
136        assert!(loaded_txes.is_empty());
137
138        // It's ok to finish an already finished transaction
139        pending_txes.finish_transaction(&tx_digest).unwrap();
140
141        // Test writing and finishing more transactions
142        let txes: Vec<_> = (0..10)
143            .map(|_| VerifiedTransaction::new_unchecked(create_fake_transaction()))
144            .collect();
145        for tx in txes.iter().take(10) {
146            assert!(
147                pending_txes
148                    .write_pending_transaction_maybe(tx)
149                    .await
150                    .unwrap()
151            );
152        }
153        let loaded_tx_digests: HashSet<_> = pending_txes
154            .load_all_pending_transactions()?
155            .iter()
156            .map(|t| *t.digest())
157            .collect();
158        assert_eq!(
159            txes.iter().map(|t| *t.digest()).collect::<HashSet<_>>(),
160            loaded_tx_digests
161        );
162
163        for tx in txes.iter().take(5) {
164            pending_txes.finish_transaction(tx.digest()).unwrap();
165        }
166        let loaded_tx_digests: HashSet<_> = pending_txes
167            .load_all_pending_transactions()?
168            .iter()
169            .map(|t| *t.digest())
170            .collect();
171        assert_eq!(
172            txes.iter()
173                .skip(5)
174                .map(|t| *t.digest())
175                .collect::<HashSet<_>>(),
176            loaded_tx_digests
177        );
178
179        Ok(())
180    }
181}
iota_storage/write_path_pending_tx_log.rs

iota_storage/
write_path_pending_tx_log.rs
