Skip to main content

miden_client_sqlite_store/
lib.rs

1//! SQLite-backed Store implementation for miden-client. This crate provides `SqliteStore` and its
2//! full implementation.
3//!
4//! [`SqliteStore`] enables the persistence of accounts, transactions, notes, block headers, and MMR
5//! nodes using an `SQLite` database.
6
7use std::boxed::Box;
8use std::collections::{BTreeMap, BTreeSet};
9use std::path::{Path, PathBuf};
10use std::rc::Rc;
11use std::string::{String, ToString};
12use std::time::Duration;
13use std::vec::Vec;
14
15use db_management::migration::SqliteMigrator;
16use db_management::pool_manager::{Pool, SqlitePoolManager};
17use deadpool::Runtime;
18use miden_client::Word;
19use miden_client::account::{
20    Account,
21    AccountCode,
22    AccountHeader,
23    AccountId,
24    AccountStorage,
25    Address,
26    StorageMapKey,
27    StorageSlotName,
28};
29use miden_client::asset::{Asset, AssetVault, AssetWitness};
30use miden_client::block::BlockHeader;
31use miden_client::crypto::{InOrderIndex, MmrPeaks};
32use miden_client::note::{BlockNumber, NoteScript, NoteTag, Nullifier};
33use miden_client::store::{
34    AccountRecord,
35    AccountStatus,
36    AccountStorageFilter,
37    BlockRelevance,
38    ClientAccountType,
39    InputNoteCursor,
40    InputNoteRecord,
41    NoteFilter,
42    OutputNoteRecord,
43    PartialBlockchainFilter,
44    SettingMutation,
45    SettingScope,
46    Store,
47    StoreError,
48    TransactionFilter,
49};
50use miden_client::sync::{NoteTagRecord, StateSyncUpdate};
51use miden_client::transaction::{TransactionRecord, TransactionStoreUpdate};
52use miden_client::utils::Serializable;
53use miden_protocol::Felt;
54use miden_protocol::account::StorageMapWitness;
55use miden_protocol::asset::AssetId;
56use rusqlite::Connection;
57use rusqlite::types::Value;
58use sql_error::SqlResultExt;
59
60use crate::account::rows::query_vault_assets;
61
62mod account;
63mod builder;
64mod chain_data;
65mod db_management;
66mod forest;
67mod note;
68mod settings;
69mod sql_error;
70mod sync;
71mod transaction;
72
73pub use builder::ClientBuilderSqliteExt;
74
75// SQLITE STORE
76// ================================================================================================
77
78/// `SQLite`-backed [`Store`] implementation.
79///
80/// Current table definitions are the result of applying every migration under `migrations/` in
81/// order.
82pub struct SqliteStore {
83    pub(crate) pool: Pool,
84    database_filepath: PathBuf,
85}
86
87impl SqliteStore {
88    // CONSTRUCTORS
89    // --------------------------------------------------------------------------------------------
90
91    /// Returns a new instance of [Store] instantiated with the specified configuration options.
92    pub async fn new(database_filepath: PathBuf) -> Result<Self, StoreError> {
93        if database_filepath.to_str().is_none() {
94            return Err(database_error(format!(
95                "database path is not valid UTF-8: {}",
96                database_filepath.display()
97            )));
98        }
99
100        let sqlite_pool_manager = SqlitePoolManager::new(database_filepath.clone());
101        let pool = Pool::builder(sqlite_pool_manager)
102            .wait_timeout(Some(Duration::from_secs(30)))
103            .runtime(Runtime::Tokio1)
104            .build()
105            .map_err(database_error)?;
106
107        Self::migrate(&pool, SqliteMigrator::client()).await?;
108
109        // Account SMT data is persisted in the forest tables and read on demand, so no state needs
110        // to be rebuilt here.
111        Ok(SqliteStore { pool, database_filepath })
112    }
113
114    /// Returns the path of the database file backing this store.
115    pub fn database_filepath(&self) -> &Path {
116        &self.database_filepath
117    }
118
119    /// Brings the database in `pool` up to the latest version of the schema `migration` builds.
120    ///
121    /// The upgrade is verified before it is committed, so a failure is rolled back by `SQLite` and
122    /// leaves the store exactly as it was.
123    async fn migrate(pool: &Pool, migration: &'static SqliteMigrator) -> Result<(), StoreError> {
124        let conn = pool.get().await.map_err(database_error)?;
125
126        conn.interact(move |conn| migration.apply(conn))
127            .await
128            .map_err(database_error)?
129            .map_err(database_error)
130    }
131
132    /// Interacts with the database by executing the provided function on a connection from the
133    /// pool.
134    ///
135    /// This function is a helper method which simplifies the process of making queries to the
136    /// database. It acquires a connection from the pool and executes the provided function,
137    /// returning the result.
138    async fn interact_with_connection<F, R>(&self, f: F) -> Result<R, StoreError>
139    where
140        F: FnOnce(&mut Connection) -> Result<R, StoreError> + Send + 'static,
141        R: Send + 'static,
142    {
143        self.pool
144            .get()
145            .await
146            .map_err(database_error)?
147            .interact(f)
148            .await
149            .map_err(database_error)?
150    }
151}
152
153// SQLite implementation of the Store trait
154//
155// To simplify, all implementations rely on inner SqliteStore functions that map 1:1 by name This
156// way, the actual implementations are grouped by entity types in their own sub-modules
157#[async_trait::async_trait]
158impl Store for SqliteStore {
159    fn identifier(&self) -> &str {
160        self.database_filepath
161            .to_str()
162            .expect("rejected by SqliteStore::new when not UTF-8")
163    }
164
165    fn get_current_timestamp(&self) -> Option<u64> {
166        Some(current_timestamp_u64())
167    }
168
169    async fn get_note_tags(&self) -> Result<Vec<NoteTagRecord>, StoreError> {
170        self.interact_with_connection(SqliteStore::get_note_tags).await
171    }
172
173    async fn get_unique_note_tags(&self) -> Result<BTreeSet<NoteTag>, StoreError> {
174        self.interact_with_connection(SqliteStore::get_unique_note_tags).await
175    }
176
177    async fn add_note_tag(&self, tag: NoteTagRecord) -> Result<bool, StoreError> {
178        self.interact_with_connection(move |conn| SqliteStore::add_note_tag(conn, tag))
179            .await
180    }
181
182    async fn remove_note_tag(&self, tag: NoteTagRecord) -> Result<usize, StoreError> {
183        self.interact_with_connection(move |conn| SqliteStore::remove_note_tag(conn, tag))
184            .await
185    }
186
187    async fn get_sync_height(&self) -> Result<BlockNumber, StoreError> {
188        self.interact_with_connection(SqliteStore::get_sync_height).await
189    }
190
191    async fn apply_state_sync(&self, state_sync_update: StateSyncUpdate) -> Result<(), StoreError> {
192        self.interact_with_connection(move |conn| {
193            SqliteStore::apply_state_sync(conn, state_sync_update)
194        })
195        .await
196    }
197
198    async fn get_transactions(
199        &self,
200        transaction_filter: TransactionFilter,
201    ) -> Result<Vec<TransactionRecord>, StoreError> {
202        self.interact_with_connection(move |conn| {
203            SqliteStore::get_transactions(conn, &transaction_filter)
204        })
205        .await
206    }
207
208    async fn apply_transaction(&self, tx_update: TransactionStoreUpdate) -> Result<(), StoreError> {
209        self.interact_with_connection(move |conn| SqliteStore::apply_transaction(conn, &tx_update))
210            .await
211    }
212
213    async fn apply_transaction_batch(
214        &self,
215        tx_updates: Vec<TransactionStoreUpdate>,
216    ) -> Result<(), StoreError> {
217        self.interact_with_connection(move |conn| {
218            SqliteStore::apply_transaction_batch(conn, &tx_updates)
219        })
220        .await
221    }
222
223    async fn get_input_notes(
224        &self,
225        filter: NoteFilter,
226    ) -> Result<Vec<InputNoteRecord>, StoreError> {
227        self.interact_with_connection(move |conn| SqliteStore::get_input_notes(conn, &filter))
228            .await
229    }
230
231    async fn get_output_notes(
232        &self,
233        note_filter: NoteFilter,
234    ) -> Result<Vec<OutputNoteRecord>, StoreError> {
235        self.interact_with_connection(move |conn| SqliteStore::get_output_notes(conn, &note_filter))
236            .await
237    }
238
239    async fn get_input_note_after(
240        &self,
241        filter: NoteFilter,
242        consumer: AccountId,
243        block_start: Option<BlockNumber>,
244        block_end: Option<BlockNumber>,
245        cursor: Option<InputNoteCursor>,
246    ) -> Result<Option<InputNoteRecord>, StoreError> {
247        self.interact_with_connection(move |conn| {
248            SqliteStore::get_input_note_after(
249                conn,
250                &filter,
251                consumer,
252                block_start,
253                block_end,
254                cursor,
255            )
256        })
257        .await
258    }
259
260    async fn upsert_input_notes(&self, notes: &[InputNoteRecord]) -> Result<(), StoreError> {
261        let notes = notes.to_vec();
262        self.interact_with_connection(move |conn| SqliteStore::upsert_input_notes(conn, &notes))
263            .await
264    }
265
266    async fn get_note_script(&self, script_root: Word) -> Result<NoteScript, StoreError> {
267        self.interact_with_connection(move |conn| SqliteStore::get_note_script(conn, script_root))
268            .await
269    }
270
271    async fn upsert_note_scripts(&self, note_scripts: &[NoteScript]) -> Result<(), StoreError> {
272        let note_scripts = note_scripts.to_vec();
273        self.interact_with_connection(move |conn| {
274            SqliteStore::upsert_note_scripts(conn, &note_scripts)
275        })
276        .await
277    }
278
279    async fn insert_block_header(
280        &self,
281        block_header: &BlockHeader,
282        nodes: &[(InOrderIndex, Word)],
283        has_client_notes: bool,
284    ) -> Result<(), StoreError> {
285        let block_header = block_header.clone();
286        let nodes = nodes.to_vec();
287        self.interact_with_connection(move |conn| {
288            SqliteStore::insert_block_header(conn, &block_header, &nodes, has_client_notes)
289        })
290        .await
291    }
292
293    async fn untrack_and_prune_irrelevant_blocks(
294        &self,
295        blocks_to_untrack: &[BlockNumber],
296        node_indices_to_remove: &[InOrderIndex],
297    ) -> Result<(), StoreError> {
298        let blocks_to_untrack = blocks_to_untrack.to_vec();
299        let node_indices_to_remove = node_indices_to_remove.to_vec();
300        self.interact_with_connection(move |conn| {
301            SqliteStore::untrack_and_prune_irrelevant_blocks(
302                conn,
303                &blocks_to_untrack,
304                &node_indices_to_remove,
305            )
306        })
307        .await
308    }
309
310    async fn prune_account_history(
311        &self,
312        account_id: AccountId,
313        up_to_nonce: Felt,
314    ) -> Result<usize, StoreError> {
315        self.interact_with_connection(move |conn| {
316            SqliteStore::prune_account_history(conn, account_id, up_to_nonce)
317        })
318        .await
319    }
320
321    async fn get_block_headers(
322        &self,
323        block_numbers: &BTreeSet<BlockNumber>,
324    ) -> Result<Vec<(BlockHeader, BlockRelevance)>, StoreError> {
325        let block_numbers = block_numbers.clone();
326        self.interact_with_connection(move |conn| {
327            SqliteStore::get_block_headers(conn, &block_numbers)
328        })
329        .await
330    }
331
332    async fn get_tracked_block_headers(&self) -> Result<Vec<BlockHeader>, StoreError> {
333        self.interact_with_connection(SqliteStore::get_tracked_block_headers).await
334    }
335
336    async fn get_tracked_block_header_numbers(&self) -> Result<BTreeSet<usize>, StoreError> {
337        self.interact_with_connection(SqliteStore::get_tracked_block_header_numbers)
338            .await
339    }
340
341    async fn get_partial_blockchain_nodes(
342        &self,
343        filter: PartialBlockchainFilter,
344    ) -> Result<BTreeMap<InOrderIndex, Word>, StoreError> {
345        self.interact_with_connection(move |conn| {
346            SqliteStore::get_partial_blockchain_nodes(conn, &filter)
347        })
348        .await
349    }
350
351    async fn get_current_blockchain_peaks(&self) -> Result<MmrPeaks, StoreError> {
352        self.interact_with_connection(SqliteStore::get_current_blockchain_peaks).await
353    }
354
355    async fn insert_account(
356        &self,
357        account: &Account,
358        initial_address: Address,
359        client_account_type: ClientAccountType,
360    ) -> Result<(), StoreError> {
361        let cloned_account = account.clone();
362
363        self.interact_with_connection(move |conn| {
364            SqliteStore::insert_account(
365                conn,
366                &cloned_account,
367                &initial_address,
368                client_account_type,
369            )
370        })
371        .await
372    }
373
374    async fn update_account(&self, account: &Account) -> Result<(), StoreError> {
375        let cloned_account = account.clone();
376
377        self.interact_with_connection(move |conn| {
378            SqliteStore::update_account(conn, &cloned_account)
379        })
380        .await
381    }
382
383    async fn get_account_ids(&self) -> Result<Vec<AccountId>, StoreError> {
384        self.interact_with_connection(SqliteStore::get_account_ids).await
385    }
386
387    async fn get_account_headers(&self) -> Result<Vec<(AccountHeader, AccountStatus)>, StoreError> {
388        self.interact_with_connection(SqliteStore::get_account_headers).await
389    }
390
391    async fn get_account_header(
392        &self,
393        account_id: AccountId,
394    ) -> Result<Option<(AccountHeader, AccountStatus)>, StoreError> {
395        self.interact_with_connection(move |conn| SqliteStore::get_account_header(conn, account_id))
396            .await
397    }
398
399    async fn get_account_header_by_commitment(
400        &self,
401        account_commitment: Word,
402    ) -> Result<Option<AccountHeader>, StoreError> {
403        self.interact_with_connection(move |conn| {
404            SqliteStore::get_account_header_by_commitment(conn, account_commitment)
405        })
406        .await
407    }
408
409    async fn get_account(
410        &self,
411        account_id: AccountId,
412    ) -> Result<Option<AccountRecord>, StoreError> {
413        self.interact_with_connection(move |conn| SqliteStore::get_account(conn, account_id))
414            .await
415    }
416
417    async fn get_account_code(
418        &self,
419        account_id: AccountId,
420    ) -> Result<Option<AccountCode>, StoreError> {
421        self.interact_with_connection(move |conn| {
422            SqliteStore::get_account_code_by_id(conn, account_id)
423        })
424        .await
425    }
426
427    async fn upsert_foreign_account_code(
428        &self,
429        account_id: AccountId,
430        code: AccountCode,
431    ) -> Result<(), StoreError> {
432        self.interact_with_connection(move |conn| {
433            SqliteStore::upsert_foreign_account_code(conn, account_id, &code)
434        })
435        .await
436    }
437
438    async fn get_foreign_account_code(
439        &self,
440        account_ids: Vec<AccountId>,
441    ) -> Result<BTreeMap<AccountId, AccountCode>, StoreError> {
442        self.interact_with_connection(move |conn| {
443            SqliteStore::get_foreign_account_code(conn, account_ids)
444        })
445        .await
446    }
447
448    async fn set_setting(
449        &self,
450        scope: SettingScope,
451        key: String,
452        value: Vec<u8>,
453    ) -> Result<(), StoreError> {
454        self.interact_with_connection(move |conn| {
455            SqliteStore::set_setting(conn, scope, &key, &value)
456        })
457        .await
458    }
459
460    async fn get_setting(
461        &self,
462        scope: SettingScope,
463        key: String,
464    ) -> Result<Option<Vec<u8>>, StoreError> {
465        self.interact_with_connection(move |conn| SqliteStore::get_setting(conn, scope, &key))
466            .await
467    }
468
469    async fn remove_setting(&self, scope: SettingScope, key: String) -> Result<bool, StoreError> {
470        self.interact_with_connection(move |conn| SqliteStore::remove_setting(conn, scope, &key))
471            .await
472    }
473
474    async fn list_setting_keys(&self, scope: SettingScope) -> Result<Vec<String>, StoreError> {
475        self.interact_with_connection(move |conn| SqliteStore::list_setting_keys(conn, scope))
476            .await
477    }
478
479    async fn apply_settings_mutations(
480        &self,
481        scope: SettingScope,
482        mutations: Vec<SettingMutation>,
483    ) -> Result<(), StoreError> {
484        self.interact_with_connection(move |conn| {
485            with_write_tx(conn, |tx| {
486                for mutation in &mutations {
487                    match mutation {
488                        SettingMutation::Set { key, value } => {
489                            SqliteStore::set_setting(tx, scope, key, value)?;
490                        },
491                        SettingMutation::Remove { key } => {
492                            SqliteStore::remove_setting(tx, scope, key)?;
493                        },
494                    }
495                }
496                Ok(())
497            })
498        })
499        .await
500    }
501
502    async fn get_unspent_input_note_nullifiers(&self) -> Result<Vec<Nullifier>, StoreError> {
503        self.interact_with_connection(SqliteStore::get_unspent_input_note_nullifiers)
504            .await
505    }
506
507    async fn get_account_vault(&self, account_id: AccountId) -> Result<AssetVault, StoreError> {
508        self.interact_with_connection(move |conn| SqliteStore::get_account_vault(conn, account_id))
509            .await
510    }
511
512    async fn get_account_assets(&self, account_id: AccountId) -> Result<Vec<Asset>, StoreError> {
513        self.interact_with_connection(move |conn| query_vault_assets(conn, account_id))
514            .await
515    }
516
517    async fn get_vault_asset_witnesses(
518        &self,
519        account_id: AccountId,
520        vault_root: Word,
521        asset_ids: BTreeSet<AssetId>,
522    ) -> Result<Vec<AssetWitness>, StoreError> {
523        self.interact_with_connection(move |conn| {
524            SqliteStore::get_vault_asset_witnesses(conn, account_id, vault_root, asset_ids)
525        })
526        .await
527    }
528
529    async fn get_account_asset(
530        &self,
531        account_id: AccountId,
532        asset_id: AssetId,
533    ) -> Result<Option<(Asset, AssetWitness)>, StoreError> {
534        self.interact_with_connection(move |conn| {
535            SqliteStore::get_account_asset(conn, account_id, asset_id)
536        })
537        .await
538    }
539
540    async fn get_account_map_item(
541        &self,
542        account_id: AccountId,
543        slot_name: StorageSlotName,
544        key: StorageMapKey,
545    ) -> Result<(Word, StorageMapWitness), StoreError> {
546        self.interact_with_connection(move |conn| {
547            SqliteStore::get_account_map_item(conn, account_id, slot_name, key)
548        })
549        .await
550    }
551
552    async fn get_account_storage(
553        &self,
554        account_id: AccountId,
555        filter: AccountStorageFilter,
556    ) -> Result<AccountStorage, StoreError> {
557        self.interact_with_connection(move |conn| {
558            SqliteStore::get_account_storage(conn, account_id, &filter)
559        })
560        .await
561    }
562
563    async fn get_addresses_by_account_id(
564        &self,
565        account_id: AccountId,
566    ) -> Result<Vec<Address>, StoreError> {
567        self.interact_with_connection(move |conn| {
568            SqliteStore::get_account_addresses(conn, account_id)
569        })
570        .await
571    }
572
573    async fn insert_address(
574        &self,
575        address: Address,
576        account_id: AccountId,
577    ) -> Result<(), StoreError> {
578        self.interact_with_connection(move |conn| {
579            SqliteStore::insert_address(conn, &address, account_id)
580        })
581        .await
582    }
583
584    async fn remove_address(&self, address: Address) -> Result<bool, StoreError> {
585        self.interact_with_connection(move |conn| SqliteStore::remove_address(conn, &address))
586            .await
587    }
588
589    async fn get_minimal_partial_account(
590        &self,
591        account_id: AccountId,
592    ) -> Result<Option<AccountRecord>, StoreError> {
593        self.interact_with_connection(move |conn| {
594            SqliteStore::get_minimal_partial_account(conn, account_id)
595        })
596        .await
597    }
598}
599
600// UTILS
601// ================================================================================================
602
603fn database_error(err: impl core::fmt::Display) -> StoreError {
604    StoreError::DatabaseError(err.to_string())
605}
606
607/// Returns the current UTC timestamp as `u64` (non-leap seconds since Unix epoch).
608pub(crate) fn current_timestamp_u64() -> u64 {
609    let now = chrono::Utc::now();
610    u64::try_from(now.timestamp()).expect("timestamp is always after epoch")
611}
612
613/// Gets a `u64` value from the database.
614///
615/// `Sqlite` uses `i64` as its internal representation format, and so when retrieving we need to
616/// make sure we cast as `u64` to get the original value
617pub(crate) fn column_value_as_u64<I: rusqlite::RowIndex>(
618    row: &rusqlite::Row<'_>,
619    index: I,
620) -> rusqlite::Result<u64> {
621    let value: i64 = row.get(index)?;
622    #[allow(
623        clippy::cast_sign_loss,
624        reason = "We store u64 as i64 as sqlite only allows the latter."
625    )]
626    Ok(value as u64)
627}
628
629/// Converts a `u64` into a [Value].
630///
631/// `Sqlite` uses `i64` as its internal representation format. Note that the `as` operator performs
632/// a lossless conversion from `u64` to `i64`.
633pub(crate) fn u64_to_value(v: u64) -> Value {
634    #[allow(
635        clippy::cast_possible_wrap,
636        reason = "We store u64 as i64 as sqlite only allows the latter."
637    )]
638    Value::Integer(v as i64)
639}
640
641/// Builds the value list for a `rarray(?)` parameter from serializable items, each stored as a BLOB
642/// of its canonical byte encoding.
643///
644/// Binding the list as a single table-valued parameter keeps the SQL text constant, so the prepared
645/// statement stays cacheable regardless of the list length (and the list is not subject to
646/// `SQLite`'s bound-parameter limit).
647pub(crate) fn blob_array<T: Serializable>(items: impl IntoIterator<Item = T>) -> Rc<Vec<Value>> {
648    Rc::new(items.into_iter().map(|item| Value::Blob(item.to_bytes())).collect())
649}
650
651/// Builds the value list for a `rarray(?)` parameter from `u64` values, stored as SQL INTEGERs
652/// through the same bit-cast as [`u64_to_value`].
653pub(crate) fn int_array(items: impl IntoIterator<Item = u64>) -> Rc<Vec<Value>> {
654    Rc::new(items.into_iter().map(u64_to_value).collect())
655}
656
657/// Builds the value list for a `rarray(?)` parameter from string values, stored as SQL TEXT.
658pub(crate) fn text_array(items: impl IntoIterator<Item = String>) -> Rc<Vec<Value>> {
659    Rc::new(items.into_iter().map(Value::Text).collect())
660}
661
662/// Runs `f` inside an `IMMEDIATE` rusqlite transaction. Commits on `Ok`, rolls back on `Err`.
663///
664/// The closure must write. An `IMMEDIATE` transaction takes the write lock at `BEGIN`, so a closure
665/// that reads and then writes cannot lose the lock upgrade in between. In WAL mode that upgrade
666/// fails with `SQLITE_BUSY_SNAPSHOT`, which the busy timeout does not retry.
667pub(crate) fn with_write_tx<R>(
668    conn: &mut Connection,
669    f: impl FnOnce(&rusqlite::Transaction<'_>) -> Result<R, StoreError>,
670) -> Result<R, StoreError> {
671    let tx = conn
672        .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)
673        .into_store_error()?;
674    let result = f(&tx)?;
675    tx.commit().into_store_error()?;
676    Ok(result)
677}
678
679// TESTS
680// ================================================================================================
681
682#[cfg(test)]
683pub mod tests {
684    use std::boxed::Box;
685    use std::sync::LazyLock;
686
687    use miden_client::store::Store;
688    use miden_client::testing::common::create_test_store_path;
689
690    use super::db_management::migration::SqliteMigrator;
691    use super::db_management::migration::tests::damaging_migration;
692    use super::db_management::pool_manager::SqlitePoolManager;
693    use super::{Pool, SqliteStore, StoreError, column_value_as_u64, u64_to_value, with_write_tx};
694
695    /// A migration set that changes the store and is then rejected, which is the failure the
696    /// rollback has to undo.
697    static DAMAGING_MIGRATION: LazyLock<SqliteMigrator> = LazyLock::new(damaging_migration);
698
699    #[tokio::test]
700    async fn failed_migration_leaves_the_store_as_it_was() {
701        let database_filepath = create_test_store_path();
702        drop(SqliteStore::new(database_filepath.clone()).await.unwrap());
703
704        let pool = Pool::builder(SqlitePoolManager::new(database_filepath.clone()))
705            .build()
706            .unwrap();
707        let err = SqliteStore::migrate(&pool, &DAMAGING_MIGRATION).await.unwrap_err();
708
709        assert!(
710            err.to_string().contains("produced a schema this client does not expect"),
711            "the migration should have been rejected, got {err}"
712        );
713        // Reopening verifies the schema, so it only succeeds if the dropped table is still there.
714        SqliteStore::new(database_filepath).await.unwrap();
715    }
716
717    fn assert_send_sync<T: Send + Sync>() {}
718
719    /// The write path bit-casts `u64` to `i64` and the read path must bit-cast it back, including
720    /// for values whose top bit is set (which are stored as negative SQL INTEGERs).
721    #[test]
722    fn u64_column_round_trip() {
723        let conn = rusqlite::Connection::open_in_memory().unwrap();
724        for value in [0u64, 1, 1 << 63, u64::MAX] {
725            let read: u64 = conn
726                .query_row("SELECT ?1", [u64_to_value(value)], |row| column_value_as_u64(row, 0))
727                .unwrap();
728            assert_eq!(read, value);
729        }
730    }
731
732    #[test]
733    fn with_write_tx_rolls_back_on_error() {
734        let mut conn = rusqlite::Connection::open_in_memory().unwrap();
735        conn.execute_batch("CREATE TABLE t (id INTEGER PRIMARY KEY);").unwrap();
736
737        let result = with_write_tx(&mut conn, |tx| {
738            tx.execute("INSERT INTO t (id) VALUES (1)", []).unwrap();
739            Err::<(), _>(StoreError::DatabaseError("forced failure".into()))
740        });
741        assert!(result.is_err());
742
743        let count: i64 = conn.query_row("SELECT COUNT(*) FROM t", [], |row| row.get(0)).unwrap();
744        assert_eq!(count, 0, "the insert must roll back when the closure errors");
745    }
746
747    #[test]
748    fn with_write_tx_commits_on_success() {
749        let mut conn = rusqlite::Connection::open_in_memory().unwrap();
750        conn.execute_batch("CREATE TABLE t (id INTEGER PRIMARY KEY);").unwrap();
751
752        with_write_tx(&mut conn, |tx| {
753            tx.execute("INSERT INTO t (id) VALUES (1)", []).unwrap();
754            Ok(())
755        })
756        .unwrap();
757
758        let count: i64 = conn.query_row("SELECT COUNT(*) FROM t", [], |row| row.get(0)).unwrap();
759        assert_eq!(count, 1);
760    }
761
762    #[test]
763    fn is_send_sync() {
764        assert_send_sync::<SqliteStore>();
765        assert_send_sync::<Box<dyn Store>>();
766    }
767
768    // Function that returns a `Send` future from a dynamic trait that must be `Sync`.
769    async fn dyn_trait_send_fut(store: Box<dyn Store>) {
770        // This wouldn't compile if `get_tracked_block_headers` doesn't return a `Send` future.
771        let res = store.get_tracked_block_headers().await;
772        assert!(res.is_ok());
773    }
774
775    #[tokio::test]
776    async fn future_is_send() {
777        let client = SqliteStore::new(create_test_store_path()).await.unwrap();
778        let client: Box<SqliteStore> = client.into();
779        tokio::task::spawn(async move { dyn_trait_send_fut(client).await });
780    }
781
782    pub(crate) async fn create_test_store() -> SqliteStore {
783        SqliteStore::new(create_test_store_path()).await.unwrap()
784    }
785}