Require both types of KVStore

As an intermediary step, we require any store to implement both
`KVStore` and `KVStoreSync`, allowing us to switch over step-by-step.

We already switch to the fully-async background processor variant here.

Author: tnull

Cargo.toml

@@ -56,7 +56,7 @@ lightning = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "
 lightning-types = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "4e32d85249359d8ef8ece97d89848e40154363ab" }
 lightning-invoice = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "4e32d85249359d8ef8ece97d89848e40154363ab", features = ["std"] }
 lightning-net-tokio = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "4e32d85249359d8ef8ece97d89848e40154363ab" }
-lightning-persister = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "4e32d85249359d8ef8ece97d89848e40154363ab" }
+lightning-persister = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "4e32d85249359d8ef8ece97d89848e40154363ab", features = ["tokio"] }
 lightning-background-processor = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "4e32d85249359d8ef8ece97d89848e40154363ab" }
 lightning-rapid-gossip-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "4e32d85249359d8ef8ece97d89848e40154363ab" }
 lightning-block-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "4e32d85249359d8ef8ece97d89848e40154363ab", features = ["rest-client", "rpc-client", "tokio"] }

src/builder.rs

@@ -51,7 +51,7 @@ use lightning::routing::scoring::{
 use lightning::sign::{EntropySource, NodeSigner};
 
 use lightning::util::persist::{
-	read_channel_monitors, CHANNEL_MANAGER_PERSISTENCE_KEY,
+	read_channel_monitors, KVStoreSync, CHANNEL_MANAGER_PERSISTENCE_KEY,
 	CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
 };
 use lightning::util::ser::ReadableArgs;
@@ -1383,7 +1383,8 @@ fn build_with_store_internal(
 
 	// Initialize the ChannelManager
 	let channel_manager = {
-		if let Ok(res) = kv_store.read(
+		if let Ok(res) = KVStoreSync::read(
+			&*kv_store,
 			CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
 			CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
 			CHANNEL_MANAGER_PERSISTENCE_KEY,
@@ -1606,7 +1607,7 @@ fn build_with_store_internal(
 		Ok(output_sweeper) => Arc::new(output_sweeper),
 		Err(e) => {
 			if e.kind() == std::io::ErrorKind::NotFound {
-				Arc::new(OutputSweeper::new_with_kv_store_sync(
+				Arc::new(OutputSweeper::new(
 					channel_manager.current_best_block(),
 					Arc::clone(&tx_broadcaster),
 					Arc::clone(&fee_estimator),

src/data_store.rs

@@ -9,6 +9,7 @@ use crate::logger::{log_error, LdkLogger};
 use crate::types::DynStore;
 use crate::Error;
 
+use lightning::util::persist::KVStoreSync;
 use lightning::util::ser::{Readable, Writeable};
 
 use std::collections::hash_map;
@@ -98,19 +99,24 @@ where
 		let removed = self.objects.lock().unwrap().remove(id).is_some();
 		if removed {
 			let store_key = id.encode_to_hex_str();
-			self.kv_store
-				.remove(&self.primary_namespace, &self.secondary_namespace, &store_key, false)
-				.map_err(|e| {
-					log_error!(
-						self.logger,
-						"Removing object data for key {}/{}/{} failed due to: {}",
-						&self.primary_namespace,
-						&self.secondary_namespace,
-						store_key,
-						e
-					);
-					Error::PersistenceFailed
-				})?;
+			KVStoreSync::remove(
+				&*self.kv_store,
+				&self.primary_namespace,
+				&self.secondary_namespace,
+				&store_key,
+				false,
+			)
+			.map_err(|e| {
+				log_error!(
+					self.logger,
+					"Removing object data for key {}/{}/{} failed due to: {}",
+					&self.primary_namespace,
+					&self.secondary_namespace,
+					store_key,
+					e
+				);
+				Error::PersistenceFailed
+			})?;
 		}
 		Ok(())
 	}
@@ -142,9 +148,14 @@ where
 	fn persist(&self, object: &SO) -> Result<(), Error> {
 		let store_key = object.id().encode_to_hex_str();
 		let data = object.encode();
-		self.kv_store
-			.write(&self.primary_namespace, &self.secondary_namespace, &store_key, data)
-			.map_err(|e| {
+		KVStoreSync::write(
+			&*self.kv_store,
+			&self.primary_namespace,
+			&self.secondary_namespace,
+			&store_key,
+			data,
+		)
+		.map_err(|e| {
 			log_error!(
 				self.logger,
 				"Write for key {}/{}/{} failed due to: {}",
@@ -243,13 +254,15 @@ mod tests {
 		let store_key = id.encode_to_hex_str();
 
 		// Check we start empty.
-		assert!(store.read(&primary_namespace, &secondary_namespace, &store_key).is_err());
+		assert!(KVStoreSync::read(&*store, &primary_namespace, &secondary_namespace, &store_key)
+			.is_err());
 
 		// Check we successfully store an object and return `false`
 		let object = TestObject { id, data: [23u8; 3] };
 		assert_eq!(Ok(false), data_store.insert(object.clone()));
 		assert_eq!(Some(object), data_store.get(&id));
-		assert!(store.read(&primary_namespace, &secondary_namespace, &store_key).is_ok());
+		assert!(KVStoreSync::read(&*store, &primary_namespace, &secondary_namespace, &store_key)
+			.is_ok());
 
 		// Test re-insertion returns `true`
 		let mut override_object = object.clone();

src/event.rs

@@ -42,6 +42,7 @@ use lightning::util::config::{
 	ChannelConfigOverrides, ChannelConfigUpdate, ChannelHandshakeConfigUpdate,
 };
 use lightning::util::errors::APIError;
+use lightning::util::persist::KVStoreSync;
 use lightning::util::ser::{Readable, ReadableArgs, Writeable, Writer};
 
 use lightning_types::payment::{PaymentHash, PaymentPreimage};
@@ -355,24 +356,24 @@ where
 
 	fn persist_queue(&self, locked_queue: &VecDeque<Event>) -> Result<(), Error> {
 		let data = EventQueueSerWrapper(locked_queue).encode();
-		self.kv_store
-			.write(
+		KVStoreSync::write(
+			&*self.kv_store,
+			EVENT_QUEUE_PERSISTENCE_PRIMARY_NAMESPACE,
+			EVENT_QUEUE_PERSISTENCE_SECONDARY_NAMESPACE,
+			EVENT_QUEUE_PERSISTENCE_KEY,
+			data,
+		)
+		.map_err(|e| {
+			log_error!(
+				self.logger,
+				"Write for key {}/{}/{} failed due to: {}",
 				EVENT_QUEUE_PERSISTENCE_PRIMARY_NAMESPACE,
 				EVENT_QUEUE_PERSISTENCE_SECONDARY_NAMESPACE,
 				EVENT_QUEUE_PERSISTENCE_KEY,
-				data,
-			)
-			.map_err(|e| {
-				log_error!(
-					self.logger,
-					"Write for key {}/{}/{} failed due to: {}",
-					EVENT_QUEUE_PERSISTENCE_PRIMARY_NAMESPACE,
-					EVENT_QUEUE_PERSISTENCE_SECONDARY_NAMESPACE,
-					EVENT_QUEUE_PERSISTENCE_KEY,
-					e
-				);
-				Error::PersistenceFailed
-			})?;
+				e
+			);
+			Error::PersistenceFailed
+		})?;
 		Ok(())
 	}
 }
@@ -1537,13 +1538,13 @@ mod tests {
 		}
 
 		// Check we can read back what we persisted.
-		let persisted_bytes = store
-			.read(
-				EVENT_QUEUE_PERSISTENCE_PRIMARY_NAMESPACE,
-				EVENT_QUEUE_PERSISTENCE_SECONDARY_NAMESPACE,
-				EVENT_QUEUE_PERSISTENCE_KEY,
-			)
-			.unwrap();
+		let persisted_bytes = KVStoreSync::read(
+			&*store,
+			EVENT_QUEUE_PERSISTENCE_PRIMARY_NAMESPACE,
+			EVENT_QUEUE_PERSISTENCE_SECONDARY_NAMESPACE,
+			EVENT_QUEUE_PERSISTENCE_KEY,
+		)
+		.unwrap();
 		let deser_event_queue =
 			EventQueue::read(&mut &persisted_bytes[..], (Arc::clone(&store), logger)).unwrap();
 		assert_eq!(deser_event_queue.wait_next_event(), expected_event);