wallet: Replace sqlx with rusqlite.

bin/faucetd/Cargo.toml

@@ -20,13 +20,13 @@ darkfi-money-contract = {path = "../../src/contract/money", features = ["no-entr
 easy-parallel = "3.3.0"
 log = "0.4.19"
 rand = "0.8.5"
+rusqlite = "0.29.0"
 serde_json = "1.0.96"
 signal-hook = "0.3.15"
 signal-hook-async-std = "0.2.2"
 simplelog = "0.12.1"
 sled = "0.34.7"
 smol = "1.3.0"
-sqlx = {version = "0.6.3", features = ["runtime-async-std-rustls", "sqlite"]}
 url = "2.4.0"
 
 # Argument parsing

bin/faucetd/src/main.rs

@@ -29,12 +29,6 @@ use async_std::{
 };
 use async_trait::async_trait;
 use chrono::Utc;
-use darkfi::{
-    runtime::vm_runtime::SMART_CONTRACT_ZKAS_DB_NAME,
-    tx::Transaction,
-    zk::{halo2::Field, proof::ProvingKey, vm::ZkCircuit, vm_heap::empty_witnesses},
-    zkas::ZkBinary,
-};
 use darkfi_money_contract::{
     client::{
         transfer_v1::TransferCallBuilder, MONEY_KEYS_COL_IS_DEFAULT, MONEY_KEYS_COL_PUBLIC,
@@ -58,7 +52,6 @@ use rand::rngs::OsRng;
 use serde_json::{json, Value};
 use signal_hook::consts::{SIGHUP, SIGINT, SIGQUIT, SIGTERM};
 use signal_hook_async_std::Signals;
-use sqlx::Row;
 use structopt_toml::{serde::Deserialize, structopt::StructOpt, StructOptToml};
 use url::Url;
 
@@ -83,12 +76,16 @@ use darkfi::{
         },
         server::{listen_and_serve, RequestHandler},
     },
+    runtime::vm_runtime::SMART_CONTRACT_ZKAS_DB_NAME,
+    tx::Transaction,
     util::{
         async_util::sleep,
         parse::decode_base10,
         path::{expand_path, get_config_path},
     },
-    wallet::{walletdb::init_wallet, WalletPtr},
+    wallet::{WalletDb, WalletPtr},
+    zk::{halo2::Field, proof::ProvingKey, vm::ZkCircuit, vm_heap::empty_witnesses},
+    zkas::ZkBinary,
     Error, Result,
 };
 
@@ -300,17 +297,23 @@ impl Faucetd {
 
         // Get a wallet connection
         info!("Acquiring wallet connection");
-        let mut conn = wallet.conn.acquire().await?;
+        let conn = wallet.conn.lock().await;
 
         info!("Initializing wallet schema");
-        sqlx::query(wallet_schema).execute(&mut conn).await?;
+        conn.execute(wallet_schema, rusqlite::params![])?;
 
         let query = format!("SELECT * FROM {}", MONEY_TREE_COL_TREE);
-        let merkle_tree = match sqlx::query(&query).fetch_one(&mut conn).await {
-            Ok(t) => {
+        let merkle_tree = conn.query_row(&query, [], |row| {
+            let tree_bytes: Vec<u8> = row.get(MONEY_TREE_COL_TREE)?;
+            Ok(deserialize(&tree_bytes).unwrap())
+        });
+
+        let merkle_tree = match merkle_tree {
+            Ok(v) => {
                 info!("Merkle tree already exists");
-                deserialize(t.get(MONEY_TREE_COL_TREE))?
+                v
             }
+
             Err(_) => {
                 let tree = MerkleTree::new(100);
                 let tree_bytes = serialize(&tree);
@@ -318,7 +321,8 @@ impl Faucetd {
                     "DELETE FROM {}; INSERT INTO {} ({}) VALUES (?1)",
                     MONEY_TREE_TABLE, MONEY_TREE_TABLE, MONEY_TREE_COL_TREE
                 );
-                sqlx::query(&query).bind(tree_bytes).execute(&mut conn).await?;
+
+                conn.execute(&query, rusqlite::params![tree_bytes])?;
                 info!("Successfully initialized Merkle tree");
                 tree
             }
@@ -328,23 +332,31 @@ impl Faucetd {
     }
 
     async fn initialize_keypair(wallet: WalletPtr) -> Result<Keypair> {
-        let mut conn = wallet.conn.acquire().await?;
+        let conn = wallet.conn.lock().await;
 
         let query = format!(
             "SELECT {}, {} FROM {};",
             MONEY_KEYS_COL_PUBLIC, MONEY_KEYS_COL_SECRET, MONEY_KEYS_TABLE
         );
-        let keypair = match sqlx::query(&query).fetch_one(&mut conn).await {
-            Ok(row) => {
-                let public = deserialize(row.get(MONEY_KEYS_COL_PUBLIC))?;
-                let secret = deserialize(row.get(MONEY_KEYS_COL_SECRET))?;
-                Keypair { public, secret }
-            }
+
+        let keypair = conn.query_row(&query, [], |row| {
+            let public_bytes: Vec<u8> = row.get("public")?;
+            let secret_bytes: Vec<u8> = row.get("secret")?;
+
+            let public = deserialize(&public_bytes).unwrap();
+            let secret = deserialize(&secret_bytes).unwrap();
+
+            Ok(Keypair { public, secret })
+        });
+
+        let keypair = match keypair {
+            Ok(k) => k,
             Err(_) => {
                 let keypair = Keypair::random(&mut OsRng);
                 let is_default = 0;
                 let public_bytes = serialize(&keypair.public);
                 let secret_bytes = serialize(&keypair.secret);
+
                 let query = format!(
                     "INSERT INTO {} ({}, {}, {}) VALUES (?1, ?2, ?3)",
                     MONEY_KEYS_TABLE,
@@ -352,13 +364,8 @@ impl Faucetd {
                     MONEY_KEYS_COL_PUBLIC,
                     MONEY_KEYS_COL_SECRET
                 );
-                sqlx::query(&query)
-                    .bind(is_default)
-                    .bind(public_bytes)
-                    .bind(secret_bytes)
-                    .execute(&mut conn)
-                    .await?;
 
+                conn.execute(&query, rusqlite::params![is_default, public_bytes, secret_bytes])?;
                 info!("Wrote keypair to wallet");
                 keypair
             }
@@ -581,10 +588,7 @@ impl Faucetd {
         }
 
         // Broadcast transaction to the network.
-        if let Err(e) = self.sync_p2p.broadcast(tx.clone()).await {
-            error!("airdrop(): Failed broadcasting transaction: {}", e);
-            return JsonError::new(InternalError, None, id).into()
-        };
+        self.sync_p2p.broadcast(&tx).await;
 
         // Add/Update this airdrop into the hashmap
         let mut map = self.airdrop_map.lock().await;
@@ -658,7 +662,7 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'_>>) -> Result<()> {
     let signals_task = task::spawn(handle_signals(signals, cfg_path.clone(), term_tx));
 
     // Initialize or load wallet
-    let wallet = init_wallet(&args.wallet_path, &args.wallet_pass).await?;
+    let wallet = WalletDb::new(Some(expand_path(&args.wallet_path)?), &args.wallet_pass).await?;
 
     // Initialize or open sled database
     let db_path =
@@ -803,9 +807,5 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'_>>) -> Result<()> {
     let flushed_bytes = sled_db.flush_async().await?;
     info!("Flushed {} bytes", flushed_bytes);
 
-    info!("Closing wallet connection...");
-    wallet.conn.close().await;
-    info!("Closed wallet connection");
-
     Ok(())
 }

src/consensus/state.rs

@@ -24,7 +24,6 @@ use darkfi_sdk::{
 use darkfi_serial::{deserialize, serialize, SerialDecodable, SerialEncodable};
 use log::info;
 use rand::{thread_rng, Rng};
-use sqlx::Row;
 
 use super::{
     constants,
@@ -253,17 +252,22 @@ impl ConsensusState {
 
         // Retrieve coin from wallet
         // NOTE: In future this will be retrieved from the money contract.
-        // Get a wallet connection
-        let mut conn = self.wallet.conn.acquire().await?;
 
         // Execute the query and see if we find any rows
         let query_str = format!("SELECT * FROM {}", constants::CONSENSUS_COIN_TABLE);
-        let mut query = sqlx::query(&query_str);
-        let coin = match query.fetch_one(&mut conn).await {
-            Ok(row) => {
-                let bytes: Vec<u8> = row.try_get(constants::CONSENSUS_COIN_COL)?;
-                deserialize(&bytes)?
-            }
+        let wallet_conn = self.wallet.conn.lock().await;
+        let mut stmt = wallet_conn.prepare(&query_str)?;
+
+        let coin = stmt.query_row((), |row| {
+            let bytes: Vec<u8> = row.get(constants::CONSENSUS_COIN_COL)?;
+            let coin = deserialize(&bytes).unwrap();
+            Ok(coin)
+        });
+
+        stmt.finalize()?;
+
+        let coin = match coin {
+            Ok(c) => c,
             Err(_) => {
                 // If no records are found, we generate a new coin and save it to the database
                 info!(target: "consensus::state", "create_coins(): No LeadCoin was found in DB, generating new one...");
@@ -286,12 +290,12 @@ impl ConsensusState {
                     constants::CONSENSUS_COIN_TABLE,
                     constants::CONSENSUS_COIN_COL
                 );
-                query = sqlx::query(&query_str);
-                query = query.bind(serialize(&c));
-                query.execute(&mut conn).await?;
+                let mut stmt = wallet_conn.prepare(&query_str)?;
+                stmt.execute([serialize(&c)])?;
                 c
             }
         };
+
         info!(target: "consensus::state", "create_coins(): Will use LeadCoin with value: {}", coin.value);
         coins.push(coin);
 
@@ -872,7 +876,7 @@ mod tests {
     #[async_std::test]
     async fn calc_sigmas_test() -> Result<()> {
         // Generate dummy state
-        let wallet = WalletDb::new("sqlite::memory:", "foo").await?;
+        let wallet = WalletDb::new(None, "foo").await?;
         let sled_db = sled::Config::new().temporary(true).open()?;
         let blockchain = Blockchain::new(&sled_db)?;
         let state = ConsensusState::new(

src/consensus/validator.rs

@@ -685,16 +685,15 @@ impl ValidatorState {
             let derived = c.derive_coin(&mut state_checkpoint.coins_tree, derived_blind);
             // Update consensus coin in wallet
             // NOTE: In future this will be redundant as consensus coins will live in the money contract.
-            // Get a wallet connection
-            let mut conn = self.wallet.conn.acquire().await?;
             let query_str = format!(
                 "UPDATE {} SET {} = ?1",
                 constants::CONSENSUS_COIN_TABLE,
                 constants::CONSENSUS_COIN_COL
             );
-            let mut query = sqlx::query(&query_str);
-            query = query.bind(serialize(&derived));
-            query.execute(&mut conn).await?;
+            let wallet_conn = self.wallet.conn.lock().await;
+            let mut stmt = wallet_conn.prepare(&query_str)?;
+            stmt.execute([serialize(&derived)])?;
+            stmt.finalize()?;
 
             state_checkpoint.coins[idx] = derived;
         }

src/consensus/wallet.rs

@@ -20,7 +20,6 @@ use async_trait::async_trait;
 use darkfi_sdk::crypto::{Keypair, PublicKey, SecretKey};
 use darkfi_serial::deserialize;
 use log::debug;
-use sqlx::Row;
 
 use crate::{wallet::WalletDb, Result};
 
@@ -36,17 +35,21 @@ pub trait ConsensusWallet {
 impl ConsensusWallet for WalletDb {
     async fn get_default_keypair(&self) -> Result<Keypair> {
         debug!(target: "consensus::wallet", "Returning default keypair");
-        let mut conn = self.conn.acquire().await?;
 
-        let row = sqlx::query(&format!(
+        let wallet_conn = self.conn.lock().await;
+        let mut stmt = wallet_conn.prepare(&format!(
             "SELECT * FROM {} WHERE {} = 1",
             CONSENSUS_KEYS_TABLE, CONSENSUS_KEYS_COLUMN_IS_DEFAULT
-        ))
-        .fetch_one(&mut conn)
-        .await?;
-
-        let public: PublicKey = deserialize(row.get("public"))?;
-        let secret: SecretKey = deserialize(row.get("secret"))?;
+        ))?;
+
+        let (public, secret): (PublicKey, SecretKey) = stmt.query_row((), |row| {
+            let p_bytes: Vec<u8> = row.get("public")?;
+            let s_bytes: Vec<u8> = row.get("secret")?;
+            let public = deserialize(&p_bytes).unwrap();
+            let secret = deserialize(&s_bytes).unwrap();
+            Ok((public, secret))
+        })?;
+        stmt.finalize()?;
 
         Ok(Keypair { secret, public })
     }

src/wallet/walletdb.rs

@@ -16,19 +16,17 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::{path::Path, str::FromStr, time::Duration};
+use std::path::PathBuf;
 
-use async_std::{fs::create_dir_all, sync::Arc};
-use log::{debug, error, info, LevelFilter};
-use sqlx::{
-    sqlite::{SqliteConnectOptions, SqliteJournalMode},
-    ConnectOptions, SqlitePool,
-};
+use async_std::sync::{Arc, Mutex};
+use log::{debug, info};
+use rusqlite::Connection;
 
-use crate::{util::path::expand_path, Error, Result};
+use crate::Result;
 
 pub type WalletPtr = Arc<WalletDb>;
 
+/*
 /// Helper function to initialize `WalletPtr`
 pub async fn init_wallet(wallet_path: &str, wallet_pass: &str) -> Result<WalletPtr> {
     let expanded = expand_path(wallet_path)?;
@@ -36,6 +34,7 @@ pub async fn init_wallet(wallet_path: &str, wallet_pass: &str) -> Result<WalletP
     let wallet = WalletDb::new(&wallet_path, wallet_pass).await?;
     Ok(wallet)
 }
+*/
 
 /// Types we want to allow to query from the SQL wallet
 #[repr(u8)]
@@ -70,46 +69,29 @@ impl From<u8> for QueryType {
 /// Structure representing base wallet operations.
 /// Additional operations can be implemented by trait extensions.
 pub struct WalletDb {
-    pub conn: SqlitePool,
+    pub conn: Mutex<Connection>,
 }
 
 impl WalletDb {
-    pub async fn new(path: &str, password: &str) -> Result<WalletPtr> {
-        if password.trim().is_empty() {
-            error!(target: "wallet::walletdb", "Wallet password is empty. You must set a password to use the wallet.");
-            return Err(Error::WalletEmptyPassword)
-        }
-
-        if path != "sqlite::memory:" {
-            let p = Path::new(path.strip_prefix("sqlite://").unwrap());
-            if let Some(dirname) = p.parent() {
-                info!(target: "wallet::walletdb", "Creating path to wallet database: {}", dirname.display());
-                create_dir_all(&dirname).await?;
-            }
-        }
-
-        let mut connect_opts = SqliteConnectOptions::from_str(path)?
-            //.pragma("key", password.to_string())
-            .pragma("foreign_keys", "ON")
-            .create_if_missing(true)
-            .journal_mode(SqliteJournalMode::Off);
-
-        connect_opts.log_statements(LevelFilter::Trace);
-        connect_opts.log_slow_statements(LevelFilter::Trace, Duration::from_micros(10));
+    /// Create a new wallet. If `path` is `None`, create it in memory.
+    pub async fn new(path: Option<PathBuf>, _password: &str) -> Result<WalletPtr> {
+        let conn = match path.clone() {
+            Some(p) => Connection::open(p)?,
+            None => Connection::open_in_memory()?,
+        };
 
-        let conn = SqlitePool::connect_with(connect_opts).await?;
+        conn.pragma_update(None, "foreign_keys", "ON")?;
 
-        info!(target: "wallet::walletdb", "Opened wallet Sqlite connection at path {}", path);
-        Ok(Arc::new(WalletDb { conn }))
+        info!(target: "wallet::walletdb", "[WalletDb] Opened Sqlite connection at \"{:?}\"", path);
+        Ok(Arc::new(Self { conn: Mutex::new(conn) }))
     }
 
     /// This function executes a given SQL query, but isn't able to return anything.
     /// Therefore it's best to use it for initializing a table or similar things.
     pub async fn exec_sql(&self, query: &str) -> Result<()> {
-        info!(target: "wallet::walletdb", "walletdb: Executing SQL query");
+        info!(target: "wallet::walletdb", "[WalletDb] Executing SQL query");
         debug!(target: "wallet::walletdb", "\n{}", query);
-        let mut conn = self.conn.acquire().await?;
-        sqlx::query(query).execute(&mut conn).await?;
+        self.conn.lock().await.execute(query, ())?;
         Ok(())
     }
 }