Account for upstreamed KVStore trait

Author: tnull

bindings/ldk_node.udl

@@ -132,6 +132,7 @@ enum BuildError {
 	"ReadFailed",
 	"WriteFailed",
 	"StoragePathAccessFailed",
+	"KVStoreSetupFailed",
 	"WalletSetupFailed",
 	"LoggerSetupFailed",
 };

src/builder.rs

@@ -1,9 +1,7 @@
 use crate::event::EventQueue;
 use crate::gossip::GossipSource;
 use crate::io;
-use crate::io::fs_store::FilesystemStore;
 use crate::io::sqlite_store::SqliteStore;
-use crate::io::{KVStore, CHANNEL_MANAGER_PERSISTENCE_KEY, CHANNEL_MANAGER_PERSISTENCE_NAMESPACE};
 use crate::logger::{log_error, FilesystemLogger, Logger};
 use crate::payment_store::PaymentStore;
 use crate::peer_store::PeerStore;
@@ -29,8 +27,14 @@ use lightning::routing::scoring::{
 use lightning::sign::EntropySource;
 
 use lightning::util::config::UserConfig;
+use lightning::util::persist::{
+	read_channel_monitors, KVStore, CHANNEL_MANAGER_PERSISTENCE_KEY,
+	CHANNEL_MANAGER_PERSISTENCE_NAMESPACE, CHANNEL_MANAGER_PERSISTENCE_SUB_NAMESPACE,
+};
 use lightning::util::ser::ReadableArgs;
 
+use lightning_persister::fs_store::FilesystemStore;
+
 use lightning_transaction_sync::EsploraSyncClient;
 
 use bdk::bitcoin::secp256k1::Secp256k1;
@@ -48,6 +52,8 @@ use std::convert::TryInto;
 use std::default::Default;
 use std::fmt;
 use std::fs;
+use std::io::Cursor;
+use std::path::PathBuf;
 use std::sync::{Arc, Mutex, RwLock};
 use std::time::SystemTime;
 
@@ -86,6 +92,8 @@ pub enum BuildError {
 	WriteFailed,
 	/// We failed to access the given `storage_dir_path`.
 	StoragePathAccessFailed,
+	/// We failed to setup our [`KVStore`].
+	KVStoreSetupFailed,
 	/// We failed to setup the onchain wallet.
 	WalletSetupFailed,
 	/// We failed to setup the logger.
@@ -103,6 +111,7 @@ impl fmt::Display for BuildError {
 			Self::ReadFailed => write!(f, "Failed to read from store."),
 			Self::WriteFailed => write!(f, "Failed to write to store."),
 			Self::StoragePathAccessFailed => write!(f, "Failed to access the given storage path."),
+			Self::KVStoreSetupFailed => write!(f, "Failed to setup KVStore."),
 			Self::WalletSetupFailed => write!(f, "Failed to setup onchain wallet."),
 			Self::LoggerSetupFailed => write!(f, "Failed to setup the logger."),
 		}
@@ -234,17 +243,26 @@ impl NodeBuilder {
 		let storage_dir_path = self.config.storage_dir_path.clone();
 		fs::create_dir_all(storage_dir_path.clone())
 			.map_err(|_| BuildError::StoragePathAccessFailed)?;
-		let kv_store = Arc::new(SqliteStore::new(storage_dir_path.into()));
+		let kv_store = Arc::new(
+			SqliteStore::new(
+				storage_dir_path.into(),
+				Some(io::sqlite_store::SQLITE_DB_FILE_NAME.to_string()),
+				Some(io::sqlite_store::KV_TABLE_NAME.to_string()),
+			)
+			.map_err(|_| BuildError::KVStoreSetupFailed)?,
+		);
 		self.build_with_store(kv_store)
 	}
 
 	/// Builds a [`Node`] instance with a [`FilesystemStore`] backend and according to the options
 	/// previously configured.
 	pub fn build_with_fs_store(&self) -> Result<Node<FilesystemStore>, BuildError> {
-		let storage_dir_path = self.config.storage_dir_path.clone();
+		let mut storage_dir_path: PathBuf = self.config.storage_dir_path.clone().into();
+		storage_dir_path.push("fs_store");
+
 		fs::create_dir_all(storage_dir_path.clone())
 			.map_err(|_| BuildError::StoragePathAccessFailed)?;
-		let kv_store = Arc::new(FilesystemStore::new(storage_dir_path.into()));
+		let kv_store = Arc::new(FilesystemStore::new(storage_dir_path));
 		self.build_with_store(kv_store)
 	}
 
@@ -510,7 +528,7 @@ fn build_with_store_internal<K: KVStore + Sync + Send + 'static>(
 	));
 
 	// Read ChannelMonitor state from store
-	let mut channel_monitors = match io::utils::read_channel_monitors(
+	let mut channel_monitors = match read_channel_monitors(
 		Arc::clone(&kv_store),
 		Arc::clone(&keys_manager),
 		Arc::clone(&keys_manager),
@@ -536,9 +554,12 @@ fn build_with_store_internal<K: KVStore + Sync + Send + 'static>(
 		user_config.manually_accept_inbound_channels = true;
 	}
 	let channel_manager = {
-		if let Ok(mut reader) =
-			kv_store.read(CHANNEL_MANAGER_PERSISTENCE_NAMESPACE, CHANNEL_MANAGER_PERSISTENCE_KEY)
-		{
+		if let Ok(res) = kv_store.read(
+			CHANNEL_MANAGER_PERSISTENCE_NAMESPACE,
+			CHANNEL_MANAGER_PERSISTENCE_SUB_NAMESPACE,
+			CHANNEL_MANAGER_PERSISTENCE_KEY,
+		) {
+			let mut reader = Cursor::new(res);
 			let channel_monitor_references =
 				channel_monitors.iter_mut().map(|(_, chanmon)| chanmon).collect();
 			let read_args = ChannelManagerReadArgs::new(

src/event.rs

@@ -7,7 +7,10 @@ use crate::payment_store::{
 	PaymentDetails, PaymentDetailsUpdate, PaymentDirection, PaymentStatus, PaymentStore,
 };
 
-use crate::io::{KVStore, EVENT_QUEUE_PERSISTENCE_KEY, EVENT_QUEUE_PERSISTENCE_NAMESPACE};
+use crate::io::{
+	EVENT_QUEUE_PERSISTENCE_KEY, EVENT_QUEUE_PERSISTENCE_NAMESPACE,
+	EVENT_QUEUE_PERSISTENCE_SUB_NAMESPACE,
+};
 use crate::logger::{log_debug, log_error, log_info, Logger};
 
 use lightning::chain::chaininterface::{BroadcasterInterface, ConfirmationTarget, FeeEstimator};
@@ -17,6 +20,7 @@ use lightning::impl_writeable_tlv_based_enum;
 use lightning::ln::PaymentHash;
 use lightning::routing::gossip::NodeId;
 use lightning::util::errors::APIError;
+use lightning::util::persist::KVStore;
 use lightning::util::ser::{Readable, ReadableArgs, Writeable, Writer};
 
 use bitcoin::secp256k1::{PublicKey, Secp256k1};
@@ -171,7 +175,12 @@ where
 	fn persist_queue(&self, locked_queue: &VecDeque<Event>) -> Result<(), Error> {
 		let data = EventQueueSerWrapper(locked_queue).encode();
 		self.kv_store
-			.write(EVENT_QUEUE_PERSISTENCE_NAMESPACE, EVENT_QUEUE_PERSISTENCE_KEY, &data)
+			.write(
+				EVENT_QUEUE_PERSISTENCE_NAMESPACE,
+				EVENT_QUEUE_PERSISTENCE_SUB_NAMESPACE,
+				EVENT_QUEUE_PERSISTENCE_KEY,
+				&data,
+			)
 			.map_err(|e| {
 				log_error!(
 					self.logger,
@@ -814,7 +823,11 @@ mod tests {
 
 		// Check we can read back what we persisted.
 		let persisted_bytes = store
-			.get_persisted_bytes(EVENT_QUEUE_PERSISTENCE_NAMESPACE, EVENT_QUEUE_PERSISTENCE_KEY)
+			.read(
+				EVENT_QUEUE_PERSISTENCE_NAMESPACE,
+				EVENT_QUEUE_PERSISTENCE_SUB_NAMESPACE,
+				EVENT_QUEUE_PERSISTENCE_KEY,
+			)
 			.unwrap();
 		let deser_event_queue =
 			EventQueue::read(&mut &persisted_bytes[..], (Arc::clone(&store), logger)).unwrap();