Add SqliteStore backend

We here add an `KVStore` implementation on SQLite, which becomes the new
default for `Node::build`. The `FilesystemStore` is still configurable
via `Node::build_with_fs_store`

Author: tnull

src/io/fs_store.rs

@@ -45,7 +45,8 @@ pub struct FilesystemStore {
 }
 
 impl FilesystemStore {
-	pub(crate) fn new(dest_dir: PathBuf) -> Self {
+	pub(crate) fn new(mut dest_dir: PathBuf) -> Self {
+		dest_dir.push("fs_store");
 		let locks = Mutex::new(HashMap::new());
 		Self { dest_dir, locks }
 	}

src/io/mod.rs

@@ -1,9 +1,11 @@
 //! Objects and traits for data persistence.
 
 pub(crate) mod fs_store;
+pub(crate) mod sqlite_store;
 pub(crate) mod utils;
 
 pub use fs_store::FilesystemStore;
+pub use sqlite_store::SqliteStore;
 
 use lightning::util::persist::KVStorePersister;
 

src/io/sqlite_store.rs

@@ -0,0 +1,228 @@
+use super::*;
+
+use lightning::util::persist::KVStorePersister;
+use lightning::util::ser::Writeable;
+
+use rusqlite::{named_params, Connection};
+
+use std::fs;
+use std::io::Cursor;
+use std::path::PathBuf;
+use std::str::FromStr;
+use std::sync::{Arc, Mutex};
+
+// The database file name.
+const SQLITE_DB_FILE: &str = "ldk_node.sqlite";
+
+// The table in which we store all data.
+const KV_TABLE_NAME: &str = "ldk_node_data";
+
+// The current SQLite `user_version`, which we can use if we'd ever need to do a schema migration.
+const SCHEMA_USER_VERSION: u16 = 1;
+
+/// A [`KVStore`] implementation that writes to and reads from an [SQLite] database.
+///
+/// [SQLIte]: https://sqlite.org
+pub struct SqliteStore {
+	connection: Arc<Mutex<Connection>>,
+}
+
+impl SqliteStore {
+	pub(crate) fn new(dest_dir: PathBuf) -> Self {
+		let msg =
+			format!("Failed to create database destination directory: {}", dest_dir.display());
+		fs::create_dir_all(dest_dir.clone()).expect(&msg);
+		let mut db_file_path = dest_dir.clone();
+		db_file_path.push(SQLITE_DB_FILE);
+
+		let msg = format!("Failed to open/create database file: {}", db_file_path.display());
+		let connection = Connection::open(db_file_path).expect(&msg);
+
+		let msg = format!("Failed to set PRAGMA user_version");
+		connection
+			.pragma(Some(rusqlite::DatabaseName::Main), "user_version", SCHEMA_USER_VERSION, |_| {
+				Ok(())
+			})
+			.expect(&msg);
+
+		let sql = format!(
+			"CREATE TABLE IF NOT EXISTS {} (
+			namespace TEXT NOT NULL,
+			key TEXT NOT NULL,
+			value BLOB, PRIMARY KEY ( namespace, key )
+		);",
+			KV_TABLE_NAME
+		);
+		let msg = format!("Failed to create table: {}", KV_TABLE_NAME);
+		connection.execute(&sql, []).expect(&msg);
+
+		let connection = Arc::new(Mutex::new(connection));
+		Self { connection }
+	}
+}
+
+impl KVStore for SqliteStore {
+	type Reader = Cursor<Vec<u8>>;
+
+	fn read(&self, namespace: &str, key: &str) -> std::io::Result<Self::Reader> {
+		let sql =
+			format!("SELECT value FROM {} WHERE namespace=:namespace AND key=:key;", KV_TABLE_NAME);
+
+		let msg = format!("Failed to read from key: {}/{}", namespace, key);
+		let res = self
+			.connection
+			.lock()
+			.unwrap()
+			.query_row(
+				&sql,
+				named_params! {
+					":namespace": namespace,
+					":key": key,
+				},
+				|row| row.get(0),
+			)
+			.map_err(|_| std::io::Error::new(std::io::ErrorKind::NotFound, msg))?;
+		Ok(Cursor::new(res))
+	}
+
+	fn write(&self, namespace: &str, key: &str, buf: &[u8]) -> std::io::Result<()> {
+		let mut locked_conn = self.connection.lock().unwrap();
+
+		let msg = format!("Failed to start transaction");
+		let sql_tx = locked_conn
+			.transaction()
+			.map_err(|_| std::io::Error::new(std::io::ErrorKind::Other, msg))?;
+
+		let sql = format!(
+			"INSERT OR REPLACE INTO {} (namespace, key, value) VALUES (:namespace, :key, :data);",
+			KV_TABLE_NAME
+		);
+		let msg = format!("Failed to write to key: {}/{}", namespace, key);
+		sql_tx
+			.execute(
+				&sql,
+				named_params! {
+					":namespace": namespace,
+					":key": key,
+					":data": buf,
+				},
+			)
+			.map_err(|_| std::io::Error::new(std::io::ErrorKind::Other, msg))?;
+
+		let msg = format!("Failed to commit transaction");
+		sql_tx.commit().map_err(|_| std::io::Error::new(std::io::ErrorKind::Other, msg))?;
+
+		Ok(())
+	}
+
+	fn remove(&self, namespace: &str, key: &str) -> std::io::Result<bool> {
+		let sql = format!("DELETE FROM {} WHERE namespace=:namespace AND key=:key;", KV_TABLE_NAME);
+		let msg = format!("Failed to delete key: {}/{}", namespace, key);
+		let changes = self
+			.connection
+			.lock()
+			.unwrap()
+			.execute(
+				&sql,
+				named_params! {
+					":namespace": namespace,
+					":key": key,
+				},
+			)
+			.map_err(|_| std::io::Error::new(std::io::ErrorKind::Other, msg))?;
+
+		let was_present = changes != 0;
+
+		Ok(was_present)
+	}
+
+	fn list(&self, namespace: &str) -> std::io::Result<Vec<String>> {
+		let locked_conn = self.connection.lock().unwrap();
+
+		let sql = format!("SELECT key FROM {} WHERE namespace=:namespace", KV_TABLE_NAME);
+		let msg = format!("Failed to prepare statement");
+		let mut stmt = locked_conn
+			.prepare(&sql)
+			.map_err(|_| std::io::Error::new(std::io::ErrorKind::Other, msg))?;
+
+		let mut keys = Vec::new();
+
+		let msg = format!("Failed to retrieve queried rows");
+		let rows_iter = stmt
+			.query_map(named_params! {":namespace": namespace, }, |row| row.get(0))
+			.map_err(|_| std::io::Error::new(std::io::ErrorKind::Other, msg))?;
+
+		for k in rows_iter {
+			let msg = format!("Failed to retrieve queried rows");
+			keys.push(k.map_err(|_| std::io::Error::new(std::io::ErrorKind::Other, msg))?);
+		}
+
+		Ok(keys)
+	}
+}
+
+impl KVStorePersister for SqliteStore {
+	fn persist<W: Writeable>(&self, prefixed_key: &str, object: &W) -> lightning::io::Result<()> {
+		let msg = format!("Could not persist file for key {}.", prefixed_key);
+		let dest_file_path = PathBuf::from_str(prefixed_key).map_err(|_| {
+			lightning::io::Error::new(lightning::io::ErrorKind::InvalidInput, msg.clone())
+		})?;
+
+		let parent_directory = dest_file_path.parent().ok_or(lightning::io::Error::new(
+			lightning::io::ErrorKind::InvalidInput,
+			msg.clone(),
+		))?;
+		let namespace = parent_directory.display().to_string();
+
+		let dest_without_namespace = dest_file_path
+			.strip_prefix(&namespace)
+			.map_err(|_| lightning::io::Error::new(lightning::io::ErrorKind::InvalidInput, msg))?;
+		let key = dest_without_namespace.display().to_string();
+
+		self.write(&namespace, &key, &object.encode())?;
+		Ok(())
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use crate::test::utils::random_storage_path;
+	use lightning::util::persist::KVStorePersister;
+	use lightning::util::ser::Readable;
+
+	use proptest::prelude::*;
+	proptest! {
+		#[test]
+		fn read_write_remove_list_persist(data in any::<[u8; 32]>()) {
+			let rand_dir = random_storage_path();
+
+			let sqlite_store = SqliteStore::new(rand_dir.into());
+			let namespace = "testspace";
+			let key = "testkey";
+
+			// Test the basic KVStore operations.
+			sqlite_store.write(namespace, key, &data).unwrap();
+
+			let listed_keys = sqlite_store.list(namespace).unwrap();
+			assert_eq!(listed_keys.len(), 1);
+			assert_eq!(listed_keys[0], "testkey");
+
+			let mut reader = sqlite_store.read(namespace, key).unwrap();
+			let read_data: [u8; 32] = Readable::read(&mut reader).unwrap();
+			assert_eq!(data, read_data);
+
+			sqlite_store.remove(namespace, key).unwrap();
+
+			let listed_keys = sqlite_store.list(namespace).unwrap();
+			assert_eq!(listed_keys.len(), 0);
+
+			// Test KVStorePersister
+			let prefixed_key = format!("{}/{}", namespace, key);
+			sqlite_store.persist(&prefixed_key, &data).unwrap();
+			let mut reader = sqlite_store.read(namespace, key).unwrap();
+			let read_data: [u8; 32] = Readable::read(&mut reader).unwrap();
+			assert_eq!(data, read_data);
+		}
+	}
+}

src/lib.rs

@@ -100,6 +100,7 @@ pub use event::Event;
 use event::{EventHandler, EventQueue};
 use gossip::GossipSource;
 use io::fs_store::FilesystemStore;
+use io::sqlite_store::SqliteStore;
 use io::{KVStore, CHANNEL_MANAGER_PERSISTENCE_KEY, CHANNEL_MANAGER_PERSISTENCE_NAMESPACE};
 use payment_store::PaymentStore;
 pub use payment_store::{PaymentDetails, PaymentDirection, PaymentStatus};
@@ -312,11 +313,21 @@ impl Builder {
 		self
 	}
 
+	/// Builds a [`Node`] instance with a [`SqliteStore`] backend and according to the options
+	/// previously configured.
+	pub fn build(&self) -> Arc<Node<SqliteStore>> {
+		let storage_dir = self.config.storage_dir_path.clone();
+		fs::create_dir_all(storage_dir.clone()).expect("Failed to create LDK data directory");
+		let kv_store = Arc::new(SqliteStore::new(storage_dir.into()).into());
+		self.build_with_store(kv_store)
+	}
+
 	/// Builds a [`Node`] instance with a [`FilesystemStore`] backend and according to the options
 	/// previously configured.
-	pub fn build(&self) -> Arc<Node<FilesystemStore>> {
-		let ldk_data_dir = format!("{}/ldk", self.config.storage_dir_path);
-		let kv_store = Arc::new(FilesystemStore::new(ldk_data_dir.clone().into()));
+	pub fn build_with_fs_store(&self) -> Arc<Node<FilesystemStore>> {
+		let storage_dir = self.config.storage_dir_path.clone();
+		fs::create_dir_all(storage_dir.clone()).expect("Failed to create LDK data directory");
+		let kv_store = Arc::new(FilesystemStore::new(storage_dir.into()));
 		self.build_with_store(kv_store)
 	}
 
@@ -326,12 +337,6 @@ impl Builder {
 	) -> Arc<Node<K>> {
 		let config = Arc::new(self.config.clone());
 
-		let ldk_data_dir = format!("{}/ldk", config.storage_dir_path);
-		fs::create_dir_all(ldk_data_dir.clone()).expect("Failed to create LDK data directory");
-
-		let bdk_data_dir = format!("{}/bdk", config.storage_dir_path);
-		fs::create_dir_all(bdk_data_dir.clone()).expect("Failed to create BDK data directory");
-
 		// Initialize the Logger
 		let log_file_path = format!("{}/ldk_node.log", config.storage_dir_path);
 		let logger = Arc::new(FilesystemLogger::new(log_file_path));
@@ -366,7 +371,8 @@ impl Builder {
 		)
 		.expect("Failed to derive on-chain wallet name");
 
-		let database_path = format!("{}/{}.sqlite", bdk_data_dir, wallet_name);
+		let database_path =
+			format!("{}/bdk_wallet_{}.sqlite", config.storage_dir_path, wallet_name);
 		let database = SqliteDatabase::new(database_path);
 
 		let bdk_wallet = bdk::Wallet::new(
@@ -465,8 +471,12 @@ impl Builder {
 		) {
 			Ok(monitors) => monitors,
 			Err(e) => {
-				log_error!(logger, "Failed to read channel monitors: {}", e.to_string());
-				panic!("Failed to read channel monitors: {}", e.to_string());
+				if e.kind() == std::io::ErrorKind::NotFound {
+					Vec::new()
+				} else {
+					log_error!(logger, "Failed to read channel monitors: {}", e.to_string());
+					panic!("Failed to read channel monitors: {}", e.to_string());
+				}
 			}
 		};
 
@@ -663,7 +673,7 @@ impl Builder {
 
 /// This type alias is required as Uniffi doesn't support generics, i.e., we can only expose only
 /// concretized types via this aliasing hack.
-type LDKNode = Node<FilesystemStore>;
+type LDKNode = Node<SqliteStore>;
 
 /// The main interface object of LDK Node, wrapping the necessary LDK and BDK functionalities.
 ///

src/test/functional_tests.rs

@@ -312,6 +312,55 @@ fn start_stop_reinit() {
 	reinitialized_node.stop().unwrap();
 }
 
+#[test]
+fn start_stop_reinit_fs_store() {
+	let (bitcoind, electrsd) = setup_bitcoind_and_electrsd();
+	let esplora_url = electrsd.esplora_url.as_ref().unwrap();
+	let config = random_config(&esplora_url);
+	let node = Builder::from_config(config.clone()).build_with_fs_store();
+	let expected_node_id = node.node_id();
+
+	let funding_address = node.new_funding_address().unwrap();
+	let expected_amount = Amount::from_sat(100000);
+
+	premine_and_distribute_funds(&bitcoind, &electrsd, vec![funding_address], expected_amount);
+	assert_eq!(node.onchain_balance().unwrap().get_total(), 0);
+
+	node.start().unwrap();
+	assert_eq!(node.start(), Err(Error::AlreadyRunning));
+
+	node.sync_wallets().unwrap();
+	assert_eq!(node.onchain_balance().unwrap().get_spendable(), expected_amount.to_sat());
+
+	node.stop().unwrap();
+	assert_eq!(node.stop(), Err(Error::NotRunning));
+
+	node.start().unwrap();
+	assert_eq!(node.start(), Err(Error::AlreadyRunning));
+
+	node.stop().unwrap();
+	assert_eq!(node.stop(), Err(Error::NotRunning));
+	drop(node);
+
+	let reinitialized_node = Builder::from_config(config).build_with_fs_store();
+	assert_eq!(reinitialized_node.node_id(), expected_node_id);
+
+	reinitialized_node.start().unwrap();
+
+	assert_eq!(
+		reinitialized_node.onchain_balance().unwrap().get_spendable(),
+		expected_amount.to_sat()
+	);
+
+	reinitialized_node.sync_wallets().unwrap();
+	assert_eq!(
+		reinitialized_node.onchain_balance().unwrap().get_spendable(),
+		expected_amount.to_sat()
+	);
+
+	reinitialized_node.stop().unwrap();
+}
+
 #[test]
 fn onchain_spend_receive() {
 	let (bitcoind, electrsd) = setup_bitcoind_and_electrsd();