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

@@ -32,7 +32,7 @@ default = []
 #lightning-types = { version = "0.2.0" }
 #lightning-invoice = { version = "0.33.0", features = ["std"] }
 #lightning-net-tokio = { version = "0.1.0" }
-#lightning-persister = { version = "0.1.0" }
+#lightning-persister = { version = "0.1.0", features = ["tokio"] }
 #lightning-background-processor = { version = "0.1.0" }
 #lightning-rapid-gossip-sync = { version = "0.1.0" }
 #lightning-block-sync = { version = "0.1.0", features = ["rest-client", "rpc-client", "tokio"] }
@@ -44,7 +44,7 @@ default = []
 #lightning-types = { git = "https://github.com/lightningdevkit/rust-lightning", branch = "main" }
 #lightning-invoice = { git = "https://github.com/lightningdevkit/rust-lightning", branch = "main", features = ["std"] }
 #lightning-net-tokio = { git = "https://github.com/lightningdevkit/rust-lightning", branch = "main" }
-#lightning-persister = { git = "https://github.com/lightningdevkit/rust-lightning", branch = "main" }
+#lightning-persister = { git = "https://github.com/lightningdevkit/rust-lightning", branch = "main", features = ["tokio"] }
 #lightning-background-processor = { git = "https://github.com/lightningdevkit/rust-lightning", branch = "main" }
 #lightning-rapid-gossip-sync = { git = "https://github.com/lightningdevkit/rust-lightning", branch = "main" }
 #lightning-block-sync = { git = "https://github.com/lightningdevkit/rust-lightning", branch = "main", features = ["rest-client", "rpc-client", "tokio"] }
@@ -56,7 +56,7 @@ lightning = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "
 lightning-types = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "3e21ba37a133977d4247e86f25df983b39326994" }
 lightning-invoice = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "3e21ba37a133977d4247e86f25df983b39326994", features = ["std"] }
 lightning-net-tokio = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "3e21ba37a133977d4247e86f25df983b39326994" }
-lightning-persister = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "3e21ba37a133977d4247e86f25df983b39326994" }
+lightning-persister = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "3e21ba37a133977d4247e86f25df983b39326994", features = ["tokio"] }
 lightning-background-processor = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "3e21ba37a133977d4247e86f25df983b39326994" }
 lightning-rapid-gossip-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "3e21ba37a133977d4247e86f25df983b39326994" }
 lightning-block-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "3e21ba37a133977d4247e86f25df983b39326994", features = ["rest-client", "rpc-client", "tokio"] }
@@ -68,7 +68,7 @@ lightning-macros = { git = "https://github.com/lightningdevkit/rust-lightning",
 #lightning-types = { path = "../rust-lightning/lightning-types" }
 #lightning-invoice = { path = "../rust-lightning/lightning-invoice", features = ["std"] }
 #lightning-net-tokio = { path = "../rust-lightning/lightning-net-tokio" }
-#lightning-persister = { path = "../rust-lightning/lightning-persister" }
+#lightning-persister = { path = "../rust-lightning/lightning-persister", features = ["tokio"] }
 #lightning-background-processor = { path = "../rust-lightning/lightning-background-processor" }
 #lightning-rapid-gossip-sync = { path = "../rust-lightning/lightning-rapid-gossip-sync" }
 #lightning-block-sync = { path = "../rust-lightning/lightning-block-sync", features = ["rest-client", "rpc-client", "tokio"] }

src/builder.rs

@@ -31,7 +31,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;
@@ -1419,7 +1419,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,
@@ -1657,7 +1658,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 std::collections::{hash_map, HashMap};
 use std::ops::Deref;
 use std::sync::{Arc, Mutex};
 
+use lightning::util::persist::KVStoreSync;
 use lightning::util::ser::{Readable, Writeable};
 
 use crate::logger::{log_error, LdkLogger};
@@ -97,19 +98,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(())
 	}
@@ -141,9 +147,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: {}",
@@ -241,13 +252,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

@@ -26,6 +26,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_liquidity::lsps2::utils::compute_opening_fee;
 use lightning_types::payment::{PaymentHash, PaymentPreimage};
@@ -348,24 +349,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(())
 	}
 }
@@ -1620,13 +1621,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);