Add sqlite database framework and simple tests

.gitignore

@@ -29,3 +29,6 @@ mutants.out.old/
 # OSX
 **/.DS_Store
 
+# Local database file for testing
+*.db
+

Cargo.toml

@@ -44,6 +44,7 @@ sha2 = "0.10"
 snafu = "0.8"
 surf-disco = "0.9"
 tagged-base64 = "0.4"
+tempfile = "3.7"
 tide-disco = "0.9"
 thiserror = "2.0"
 tokio = "1"
@@ -55,6 +56,7 @@ url = "2.5"
 vbs = "0.1"
 vec1 = "1.12"
 tracing-subscriber = "0.3"
+sqlx = { version = "^0.8", features = ["postgres", "sqlite", "macros"] }
 
 [workspace.lints.rust]
 unexpected_cfgs = { level = "warn", check-cfg = [

crates/shared/Cargo.toml

@@ -29,7 +29,9 @@ quick_cache = { workspace = true }
 rand = { workspace = true }
 serde = { workspace = true }
 sha2 = { workspace = true }
+sqlx = { workspace = true }
 surf-disco = { workspace = true }
+tempfile = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }

crates/shared/src/lib.rs

@@ -3,6 +3,7 @@
 pub mod block;
 pub mod coordinator;
 pub mod error;
+pub mod persistence;
 pub mod state;
 #[cfg_attr(coverage_nightly, coverage(off))]
 pub mod testing;

crates/shared/src/persistence/mod.rs

@@ -0,0 +1,39 @@
+use anyhow::Context;
+use async_trait::async_trait;
+use std::path::{Path, PathBuf};
+use std::time::Instant;
+
+pub mod sqlite;
+
+/// The trait BuilderPersistence defined needed functions to maintain persistence of builder-related data
+#[async_trait]
+pub trait BuilderPersistence {
+    /// Append a transaction to persistence mempool
+    async fn append(&self, tx_data: Vec<u8>) -> Result<(), sqlx::Error>;
+    /// Load all the transactions whose `created_at` is before or equal to `before_instant`
+    async fn load(&self, before_instant: Instant) -> Result<Vec<Vec<u8>>, sqlx::Error>;
+    /// Remove a transaction from the persistence mempool
+    async fn remove(&self, tx: Vec<u8>) -> Result<(), sqlx::Error>;
+}
+
+/// build sqlite database path, if not exist, will create one
+pub fn build_sqlite_path(path: &Path) -> anyhow::Result<PathBuf> {
+    let sub_dir = path.join("sqlite");
+
+    // if `sqlite` sub dir does not exist then create it
+    if !sub_dir.exists() {
+        std::fs::create_dir_all(&sub_dir)
+            .with_context(|| format!("failed to create directory: {:?}", sub_dir))?;
+    }
+
+    // Return the full path to the SQLite database file
+    let db_path = sub_dir.join("database.sqlite");
+
+    // Ensure the file exists (create it if it doesn’t)
+    if !db_path.exists() {
+        std::fs::File::create(&db_path)
+            .with_context(|| format!("Failed to create SQLite database file: {:?}", db_path))?;
+    }
+
+    Ok(db_path)
+}

crates/shared/src/persistence/sqlite.rs

@@ -0,0 +1,222 @@
+#[allow(unused_imports)]
+use super::build_sqlite_path;
+use super::BuilderPersistence;
+use anyhow::Result;
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+use sqlx::Row;
+use sqlx::SqlitePool;
+use std::time::{Instant, SystemTime};
+
+/// Struct of transaction database in sqlite
+#[derive(Debug)]
+pub struct SqliteTxnDb {
+    pool: SqlitePool,
+}
+impl SqliteTxnDb {
+    /// New a `SqliteTxnDb` by calling with `database_url`
+    pub async fn new(database_url: String) -> Result<Self, sqlx::Error> {
+        let pool = SqlitePool::connect(&database_url).await?;
+        // it will handle the default CURRENT_TIMESTAMP automatically and assign to transaction's created_at
+        sqlx::query(
+            r#"
+            CREATE TABLE IF NOT EXISTS transactions (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                tx_data BLOB NOT NULL,
+                created_at DATETIME DEFAULT CURRENT_TIMESTAMP
+            );
+            "#,
+        )
+        .execute(&pool)
+        .await?;
+        Ok(Self { pool })
+    }
+
+    /// Clear all the data in this `SqliteTxnDb` database
+    pub async fn clear(&self) -> Result<(), sqlx::Error> {
+        // Execute a SQL statement to delete all rows from the `transactions` table
+        sqlx::query("DELETE FROM transactions")
+            .execute(&self.pool)
+            .await?;
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl BuilderPersistence for SqliteTxnDb {
+    async fn append(&self, tx_data: Vec<u8>) -> Result<(), sqlx::Error> {
+        sqlx::query(
+            r#"
+            INSERT INTO transactions (tx_data) VALUES (?);
+            "#,
+        )
+        .bind(tx_data)
+        .execute(&self.pool)
+        .await?;
+        Ok(())
+    }
+
+    async fn load(&self, before_instant: Instant) -> Result<Vec<Vec<u8>>, sqlx::Error> {
+        // Convert Instant to SystemTime
+        let now = SystemTime::now();
+        let elapsed = before_instant.elapsed();
+        let target_time = now - elapsed;
+
+        // Convert SystemTime to a format SQLite understands (RFC 3339)
+        let target_timestamp = DateTime::<Utc>::from(target_time)
+            .naive_utc()
+            .format("%Y-%m-%d %H:%M:%S")
+            .to_string();
+
+        let rows = sqlx::query(
+            r#"
+            SELECT id, tx_data, created_at FROM transactions
+            WHERE created_at <= ? ;
+            "#,
+        )
+        .bind(target_timestamp)
+        .fetch_all(&self.pool)
+        .await?;
+
+        let tx_data_list = rows
+            .into_iter()
+            .map(|row| row.get::<Vec<u8>, _>("tx_data"))
+            .collect();
+        Ok(tx_data_list)
+    }
+
+    async fn remove(&self, tx_data: Vec<u8>) -> Result<(), sqlx::Error> {
+        let result = sqlx::query(
+            r#"
+            DELETE FROM transactions WHERE tx_data = ?;
+            "#,
+        )
+        .bind(tx_data)
+        .execute(&self.pool)
+        .await?;
+
+        if result.rows_affected() > 0 {
+            Ok(())
+        } else {
+            Err(sqlx::Error::RowNotFound)
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::build_sqlite_path;
+    use super::BuilderPersistence;
+    use super::SqliteTxnDb;
+    use std::time::Instant;
+    use tempfile::tempdir;
+
+    /// This test checks we can set up sqlite properly
+    /// and can do basic append() and load()
+    #[tokio::test]
+    async fn test_persistence_append_and_load_txn() {
+        // Create a temporary directory
+        let tmp_dir = tempdir().expect("In test_persistence_append_and_load_txn, should be able to create a temporary directory.");
+        // Construct the database path
+        let db_path = build_sqlite_path(tmp_dir.path()).expect("In test_persistence_append_and_load_txn, should be able to create a temporary database file.");
+        let database_url = format!("sqlite://{}", db_path.to_str().expect("In test_persistence_append_and_load_txn, should be able to convert temporary database file path to string."));
+        // Initialize the database
+        let db = SqliteTxnDb::new(database_url).await.expect(
+            "In test_persistence_append_and_load_txn, it should be able to initiate a sqlite db.",
+        );
+
+        db.clear()
+            .await
+            .expect("In test_persistence_remove_txn, it should be able to clear all transactions.");
+
+        // Append a few transactions
+        let test_tx_data_list = vec![vec![1, 2, 3], vec![4, 5, 6]];
+        for tx in test_tx_data_list.clone() {
+            db.append(tx).await.expect("In test_persistence_append_and_load_txn, there shouldn't be any error when doing append");
+        }
+
+        // Set before_instant to the current time
+        let before_instant = Instant::now();
+
+        // Simulate some delay
+        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+
+        // Append one more transaction
+        db.append(vec![7, 8, 9]).await.expect("In test_persistence_append_and_load_txn, there shouldn't be any error when doing append");
+
+        // Load transactions before before_instant
+        let tx_data_list = db.load(before_instant).await.expect(
+            "In test_persistence_append_and_load_txn, it should be able to load some transactions.",
+        );
+        tracing::debug!("Transaction data before timeout: {:?}", tx_data_list);
+
+        assert_eq!(tx_data_list, test_tx_data_list);
+
+        db.clear()
+            .await
+            .expect("In test_persistence_remove_txn, it should be able to clear all transactions.");
+    }
+
+    #[tokio::test]
+    /// This test checks we can remove transaction from database properly
+    async fn test_persistence_remove_txn() {
+        // Create a temporary directory
+        let tmp_dir = tempdir().expect(
+            "In test_persistence_remove_txn, should be able to create a temporary directory.",
+        );
+        // Construct the database path
+        let db_path = build_sqlite_path(tmp_dir.path()).expect(
+            "In test_persistence_remove_txn, should be able to create a temporary database file.",
+        );
+        let database_url = format!("sqlite://{}", db_path.to_str().expect("In test_persistence_remove_txn, should be able to convert temporary database file path to string."));
+        // Initialize the database
+        let db = SqliteTxnDb::new(database_url)
+            .await
+            .expect("In test_persistence_remove_txn, it should be able to initiate a sqlite db.");
+
+        db.clear()
+            .await
+            .expect("In test_persistence_remove_txn, it should be able to clear all transactions.");
+
+        // Append some transactions
+        let test_tx_data_list = vec![vec![1, 2, 3], vec![4, 5, 6], vec![7, 8, 9]];
+        for tx in test_tx_data_list.clone() {
+            db.append(tx).await.expect(
+                "In test_persistence_remove_txn, there shouldn't be any error when doing append",
+            );
+        }
+
+        // Load all transactions
+        let mut all_transactions = db
+            .load(Instant::now())
+            .await
+            .expect("In test_persistence_remove_txn, it should be able to load some transactions.");
+        tracing::debug!("All transactions before removal: {:?}", all_transactions);
+        assert_eq!(all_transactions, test_tx_data_list);
+
+        // Remove a specific transaction
+        let tx_to_remove = test_tx_data_list[1].clone();
+        if let Err(e) = db.remove(tx_to_remove).await {
+            panic!("Failed to remove transaction: {}", e);
+        } else {
+            tracing::debug!("Transaction removed.");
+        }
+
+        // Load all transactions after removal
+
+        let remaining_transactions = db
+            .load(Instant::now())
+            .await
+            .expect("In test_persistence_remove_txn, it should be able to load some transactions.");
+        tracing::debug!(
+            "All transactions after removal: {:?}",
+            remaining_transactions
+        );
+        all_transactions.remove(1);
+        assert_eq!(remaining_transactions, all_transactions);
+
+        db.clear()
+            .await
+            .expect("In test_persistence_remove_txn, it should be able to clear all transactions.");
+    }
+}