Adds Sqlite to db options

Cargo.toml

@@ -19,6 +19,7 @@ http = "1.3.1"
 hyper = "1.6.0"
 hyper-util = "0.1.15"
 reqwest = "0.12.22"
+rusqlite = "0.37.0"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1.0.140"
 serde_yaml_ng = "0.10"

crates/sigma-db/Cargo.toml

@@ -11,6 +11,7 @@ repository.workspace = true
 anyhow.workspace = true
 futures.workspace = true
 reqwest.workspace = true
+rusqlite.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 sigma-api.workspace = true

crates/sigma-db/src/lib.rs

@@ -10,23 +10,28 @@ use tokio::sync::{
 
 pub use memory::MemoryDb;
 pub use rqlite::Rqlite;
+pub use sqlite::Sqlite;
 
 mod memory;
 mod rqlite;
+mod sqlite;
 
 /// Maximum times to retry network calls.
 const MAX_RETRIES: usize = 10;
 
 pub enum Database {
     Memory(MemoryDb),
     Rqlite(Rqlite),
+    Sqlite(Sqlite),
 }
 
 #[derive(Debug, Error)]
 pub enum Error {
     #[error(transparent)]
     Rqlite(#[from] rqlite::Error),
     #[error(transparent)]
+    Sqlite(#[from] sqlite::Error),
+    #[error(transparent)]
     SetHeight(#[from] SendError<CurrentHeight>),
 }
 
@@ -37,10 +42,13 @@ pub async fn run(
     mut api_rx: Receiver<sigma_api::Request>,
     new_height: watch::Sender<CurrentHeight>,
 ) {
+    create_tables(&mut database)
+        .await
+        .expect("Failed to create tables");
     loop {
         let mut i = 0;
         let parent_hash = loop {
-            match get_current_hash(&database).await {
+            match get_current_hash(&mut database).await {
                 Ok(parent_hash) => {
                     break parent_hash;
                 }
@@ -74,6 +82,8 @@ pub async fn run(
                 }
             }
         }
+        // PERF: Note that requests are handled in serial which
+        // could be a performance bottleneck if there are many requests.
         let f1 = order_rx.recv();
         let f2 = api_rx.recv();
         futures::pin_mut!(f1, f2);
@@ -83,7 +93,7 @@ pub async fn run(
                 process_message(&mut database, message).await;
             }
             futures::future::Either::Right((Some(request), _)) => {
-                process_request(&database, request).await;
+                process_request(&mut database, request).await;
             }
             _ => {
                 tracing::error!("Crucial channel closed");
@@ -143,7 +153,7 @@ async fn process_message(database: &mut Database, message: sigma_order::Message)
     }
 }
 
-async fn process_request(database: &Database, request: sigma_api::Request) {
+async fn process_request(database: &mut Database, request: sigma_api::Request) {
     match request {
         sigma_api::Request::Event(EventReq { heights, response }) => {
             let events_res = get_events_range(database, heights).await;
@@ -174,6 +184,14 @@ async fn process_request(database: &Database, request: sigma_api::Request) {
     }
 }
 
+async fn create_tables(database: &mut Database) -> Result<(), Error> {
+    match database {
+        Database::Memory(_db) => Ok(()),
+        Database::Rqlite(db) => Ok(db.create_tables().await?),
+        Database::Sqlite(db) => Ok(db.create_tables().await?),
+    }
+}
+
 async fn add_event_and_hash(
     database: &mut Database,
     event: Signed<Event>,
@@ -185,32 +203,36 @@ async fn add_event_and_hash(
             Ok(())
         }
         Database::Rqlite(db) => Ok(db.add_event_and_hash(event, hash).await?),
+        Database::Sqlite(db) => Ok(db.add_event_and_hash(event, hash).await?),
     }
 }
 
 async fn get_events_range(
-    database: &Database,
+    database: &mut Database,
     range: Range<Height>,
 ) -> Result<Vec<Signed<Event>>, Error> {
     match database {
         Database::Memory(db) => Ok(db.get_events_range(range)),
-        Database::Rqlite(db) => db.get_events_range(range).await.map_err(Error::from),
+        Database::Rqlite(db) => Ok(db.get_events_range(range).await?),
+        Database::Sqlite(db) => Ok(db.get_events_range(range).await?),
     }
 }
 
 async fn get_hashes_range(
-    database: &Database,
+    database: &mut Database,
     range: Range<Height>,
 ) -> Result<Vec<Signed<Hash>>, Error> {
     match database {
         Database::Memory(db) => Ok(db.get_hashes_range(range)),
-        Database::Rqlite(db) => db.get_hashes_range(range).await.map_err(Error::from),
+        Database::Rqlite(db) => Ok(db.get_hashes_range(range).await?),
+        Database::Sqlite(db) => Ok(db.get_hashes_range(range).await?),
     }
 }
 
-async fn get_current_hash(database: &Database) -> Result<Hash, Error> {
+async fn get_current_hash(database: &mut Database) -> Result<Hash, Error> {
     match database {
         Database::Memory(db) => Ok(db.get_current_hash()),
         Database::Rqlite(db) => Ok(db.get_current_hash().await?),
+        Database::Sqlite(db) => Ok(db.get_current_hash().await?),
     }
 }