miden_client_sqlite_store/
transaction.rs

1#![allow(clippy::items_after_statements)]
2
3use std::rc::Rc;
4use std::vec::Vec;
5
6use miden_client::Word;
7use miden_client::note::ToInputNoteCommitments;
8use miden_client::store::{StoreError, TransactionFilter};
9use miden_client::transaction::{
10    TransactionDetails,
11    TransactionId,
12    TransactionRecord,
13    TransactionStatus,
14    TransactionStatusVariant,
15    TransactionStoreUpdate,
16};
17use miden_client::utils::{Deserializable as _, Serializable as _};
18use rusqlite::types::Value;
19use rusqlite::{Connection, Transaction, params};
20
21use super::SqliteStore;
22use super::note::apply_note_updates_tx;
23use super::sync::add_note_tag_tx;
24use crate::forest::{ScopedAccountForest, SqliteForestBackend};
25use crate::sql_error::SqlResultExt;
26use crate::{blob_array, insert_sql, proto, subst, with_write_tx};
27
28pub(crate) const UPSERT_TRANSACTION_QUERY: &str = insert_sql!(
29    transactions {
30        id,
31        details,
32        script_root,
33        status_variant,
34        status
35    } | REPLACE
36);
37
38pub(crate) const INSERT_TRANSACTION_SCRIPT_QUERY: &str =
39    insert_sql!(transaction_scripts { script_root, script } | IGNORE);
40
41/// The column aliases match the names that [`SqliteStore::get_transactions`] reads.
42const TRANSACTIONS_BASE_QUERY: &str = "SELECT \
43     tx.id AS id, \
44     script.script AS script, \
45     tx.details AS details, \
46     tx.status AS status \
47     FROM transactions AS tx \
48     LEFT JOIN transaction_scripts AS script ON tx.script_root = script.script_root";
49
50/// Returns the transactions query for a [`TransactionFilter`], and the value list it binds.
51///
52/// Only [`TransactionFilter::Ids`] binds a parameter. The ids are bound as one `rarray(?)` value,
53/// so the SQL text stays constant for any number of ids.
54fn transaction_filter_to_query(filter: &TransactionFilter) -> (String, Option<Rc<Vec<Value>>>) {
55    match filter {
56        TransactionFilter::All => (TRANSACTIONS_BASE_QUERY.to_string(), None),
57        TransactionFilter::Uncommitted => (
58            format!(
59                "{TRANSACTIONS_BASE_QUERY} WHERE tx.status_variant = {}",
60                TransactionStatusVariant::Pending as u8
61            ),
62            None,
63        ),
64        TransactionFilter::Ids(ids) => (
65            format!("{TRANSACTIONS_BASE_QUERY} WHERE tx.id IN rarray(?)"),
66            Some(blob_array(ids)),
67        ),
68    }
69}
70
71// TRANSACTIONS
72// ================================================================================================
73
74impl SqliteStore {
75    /// Retrieves tracked transactions, filtered by [`TransactionFilter`].
76    pub fn get_transactions(
77        conn: &mut Connection,
78        filter: &TransactionFilter,
79    ) -> Result<Vec<TransactionRecord>, StoreError> {
80        let (query, id_list) = transaction_filter_to_query(filter);
81
82        conn.prepare(&query)
83            .into_store_error()?
84            .query_map(rusqlite::params_from_iter(id_list), |row| {
85                Ok((
86                    row.get::<_, Vec<u8>>("id")?,
87                    row.get::<_, Option<Vec<u8>>>("script")?,
88                    row.get::<_, Vec<u8>>("details")?,
89                    row.get::<_, Vec<u8>>("status")?,
90                ))
91            })
92            .into_store_error()?
93            .map(|result| {
94                let (id, script, details, status) = result.into_store_error()?;
95                Ok(TransactionRecord {
96                    id: TransactionId::read_from_bytes(&id)?,
97                    details: proto::decode_unchecked(&details)?,
98                    script: script.map(|script| proto::decode_unchecked(&script)).transpose()?,
99                    status: proto::decode_unchecked(&status)?,
100                })
101            })
102            .collect::<Result<Vec<TransactionRecord>, _>>()
103    }
104
105    /// Inserts a transaction and updates the current state based on the `tx_result` changes.
106    ///
107    /// SQL writes and forest mutations go through the same rusqlite transaction, so they commit or
108    /// roll back atomically.
109    pub(crate) fn apply_transaction(
110        conn: &mut Connection,
111        tx_update: &TransactionStoreUpdate,
112    ) -> Result<(), StoreError> {
113        with_write_tx(conn, |tx| {
114            let mut forest = ScopedAccountForest::new(SqliteForestBackend::new(tx))?;
115            Self::apply_transaction_in_txn(tx, &mut forest, tx_update)
116        })
117    }
118
119    /// Applies a batch of [`TransactionStoreUpdate`]s atomically. Either every update in the slice
120    /// is persisted or none are. Executes in order inside a single [`rusqlite::Transaction`].
121    pub(crate) fn apply_transaction_batch(
122        conn: &mut Connection,
123        tx_updates: &[TransactionStoreUpdate],
124    ) -> Result<(), StoreError> {
125        with_write_tx(conn, |tx| {
126            let mut forest = ScopedAccountForest::new(SqliteForestBackend::new(tx))?;
127            for update in tx_updates {
128                Self::apply_transaction_in_txn(tx, &mut forest, update)?;
129            }
130            Ok(())
131        })
132    }
133
134    /// Applies a transaction's store update within the provided rusqlite transaction. Does NOT
135    /// commit — caller is responsible for commit/rollback.
136    ///
137    /// The storage-map-root pre-read is performed via the transaction so that each call sees writes
138    /// made by prior calls within the same outer transaction.
139    pub(crate) fn apply_transaction_in_txn(
140        db_tx: &Transaction<'_>,
141        smt_forest: &mut ScopedAccountForest<'_, '_>,
142        tx_update: &TransactionStoreUpdate,
143    ) -> Result<(), StoreError> {
144        let executed_transaction = tx_update.executed_transaction();
145        let account_patch = executed_transaction.account_patch();
146
147        // Build transaction record
148        let nullifiers: Vec<Word> = executed_transaction
149            .input_notes()
150            .iter()
151            .map(|x| x.nullifier().as_word())
152            .collect();
153
154        let output_notes = executed_transaction.output_notes();
155
156        let details = TransactionDetails {
157            account_id: executed_transaction.account_id(),
158            init_account_state: executed_transaction.initial_account().initial_commitment(),
159            final_account_state: executed_transaction.final_account().to_commitment(),
160            input_note_nullifiers: nullifiers,
161            output_notes: output_notes.clone(),
162            block_num: executed_transaction.block_header().block_num(),
163            submission_height: tx_update.submission_height(),
164            expiration_block_num: executed_transaction.expiration_block_num(),
165            creation_timestamp: super::current_timestamp_u64(),
166        };
167
168        let transaction_record = TransactionRecord::new(
169            executed_transaction.id(),
170            details,
171            executed_transaction.tx_args().tx_script().cloned(),
172            TransactionStatus::Pending,
173        );
174
175        // Insert transaction data
176        upsert_transaction_record(db_tx, &transaction_record)?;
177
178        // Account Data
179        Self::apply_account_patch(
180            db_tx,
181            smt_forest,
182            &executed_transaction.initial_account().into(),
183            executed_transaction.final_account(),
184            account_patch,
185        )?;
186
187        // Note Updates
188        apply_note_updates_tx(db_tx, tx_update.note_updates())?;
189
190        // Note tags
191        for tag_record in tx_update.new_tags() {
192            add_note_tag_tx(db_tx, tag_record)?;
193        }
194
195        Ok(())
196    }
197}
198
199/// Updates the transaction record in the database, inserting it if it doesn't exist.
200pub(crate) fn upsert_transaction_record(
201    tx: &Transaction<'_>,
202    transaction: &TransactionRecord,
203) -> Result<(), StoreError> {
204    let script_root = transaction.script.as_ref().map(|script| script.root().to_bytes());
205
206    if let Some(script) = &transaction.script {
207        tx.execute(INSERT_TRANSACTION_SCRIPT_QUERY, params![script_root, proto::encode(script)])
208            .into_store_error()?;
209    }
210
211    tx.execute(
212        UPSERT_TRANSACTION_QUERY,
213        params![
214            transaction.id.to_bytes(),
215            proto::encode(&transaction.details),
216            script_root,
217            transaction.status.variant() as u8,
218            proto::encode(&transaction.status),
219        ],
220    )
221    .into_store_error()?;
222
223    Ok(())
224}
225
226// TESTS
227// ================================================================================================
228
229#[cfg(test)]
230mod tests {
231    use miden_client::store::TransactionFilter;
232    use miden_client::transaction::{
233        DiscardCause,
234        RawOutputNotes,
235        TransactionDetails,
236        TransactionId,
237        TransactionRecord,
238        TransactionStatus,
239    };
240    use miden_client::{Felt, Word, ZERO};
241    use miden_protocol::account::AccountId;
242    use miden_protocol::block::BlockNumber;
243    use miden_protocol::testing::account_id::ACCOUNT_ID_REGULAR_PRIVATE_ACCOUNT_UPDATABLE_CODE;
244    use rusqlite::Connection;
245
246    use super::{SqliteStore, transaction_filter_to_query, upsert_transaction_record};
247    use crate::db_management::migration::SqliteMigrator;
248
249    /// Builds a script-less transaction record with the given status.
250    fn create_transaction_record(index: u64, status: TransactionStatus) -> TransactionRecord {
251        const BLOCK_NUM: u32 = 5;
252
253        let account_id =
254            AccountId::try_from(ACCOUNT_ID_REGULAR_PRIVATE_ACCOUNT_UPDATABLE_CODE).unwrap();
255        let details = TransactionDetails {
256            account_id,
257            init_account_state: Word::default(),
258            final_account_state: Word::default(),
259            input_note_nullifiers: vec![],
260            output_notes: RawOutputNotes::new(vec![]).unwrap(),
261            block_num: BlockNumber::from(BLOCK_NUM),
262            submission_height: BlockNumber::from(BLOCK_NUM),
263            expiration_block_num: BlockNumber::from(BLOCK_NUM + 1),
264            creation_timestamp: 0,
265        };
266
267        let id = TransactionId::from_raw([Felt::new_unchecked(index), ZERO, ZERO, ZERO].into());
268
269        TransactionRecord::new(id, details, None, status)
270    }
271
272    fn create_test_connection(records: &[TransactionRecord]) -> Connection {
273        let mut conn = Connection::open_in_memory().unwrap();
274        SqliteMigrator::client().apply(&mut conn).unwrap();
275
276        let db_tx = conn.transaction().unwrap();
277        for record in records {
278            upsert_transaction_record(&db_tx, record).unwrap();
279        }
280        db_tx.commit().unwrap();
281
282        conn
283    }
284
285    /// Returns the `detail` column of every step of the query plan for `query`.
286    fn query_plan(conn: &Connection, query: &str) -> Vec<String> {
287        let mut stmt = conn.prepare(&format!("EXPLAIN QUERY PLAN {query}")).unwrap();
288        stmt.query_map([], |row| row.get::<_, String>(3))
289            .unwrap()
290            .collect::<Result<Vec<_>, _>>()
291            .unwrap()
292    }
293
294    #[test]
295    fn uncommitted_returns_only_pending_transactions() {
296        let pending = create_transaction_record(1, TransactionStatus::Pending);
297        let committed = create_transaction_record(
298            2,
299            TransactionStatus::Committed {
300                block_number: BlockNumber::from(6u32),
301                commit_timestamp: 0,
302            },
303        );
304        let discarded =
305            create_transaction_record(3, TransactionStatus::Discarded(DiscardCause::Expired));
306
307        let mut conn = create_test_connection(&[pending.clone(), committed, discarded]);
308
309        let records =
310            SqliteStore::get_transactions(&mut conn, &TransactionFilter::Uncommitted).unwrap();
311
312        let ids: Vec<_> = records.iter().map(|record| record.id).collect();
313        assert_eq!(ids, vec![pending.id]);
314    }
315
316    #[test]
317    fn uncommitted_is_served_by_the_pending_transactions_index() {
318        let conn = create_test_connection(&[]);
319
320        let (query, _) = transaction_filter_to_query(&TransactionFilter::Uncommitted);
321        let plan = query_plan(&conn, &query).join("\n");
322
323        // Every entry of the partial index is a pending transaction, so the search never touches a
324        // committed or discarded row.
325        assert!(
326            plan.contains("SEARCH tx USING INDEX idx_transactions_pending (status_variant=?)"),
327            "pending transactions must be read from the partial index: {plan}"
328        );
329    }
330}
miden_client_sqlite_store/transaction.rs

miden_client_sqlite_store/
transaction.rs
