Introduce InMemoryStore for testing

Recently, `rust-lightning` broke the (async) API of the `TestStore`,
making it ~impossible to use in regular tests.

Here, we un-DRY our `TestStore` implementation and simply copy over the
previous `TestStore` version, now named `InMemoryStore` to discern the
objects. We also switch all feasible instances over to use
`InMemoryStore` rather than LDK's `test_utils::TestStore`.

Author: tnull

src/data_store.rs

@@ -172,10 +172,11 @@ where
 #[cfg(test)]
 mod tests {
 	use lightning::impl_writeable_tlv_based;
-	use lightning::util::test_utils::{TestLogger, TestStore};
+	use lightning::util::test_utils::TestLogger;
 
 	use super::*;
 	use crate::hex_utils;
+	use crate::io::test_utils::InMemoryStore;
 
 	#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
 	struct TestObjectId {
@@ -234,7 +235,7 @@ mod tests {
 
 	#[test]
 	fn data_is_persisted() {
-		let store: Arc<DynStore> = Arc::new(TestStore::new(false));
+		let store: Arc<DynStore> = Arc::new(InMemoryStore::new());
 		let logger = Arc::new(TestLogger::new());
 		let primary_namespace = "datastore_test_primary".to_string();
 		let secondary_namespace = "datastore_test_secondary".to_string();

src/event.rs

@@ -1605,13 +1605,14 @@ mod tests {
 	use std::sync::atomic::{AtomicU16, Ordering};
 	use std::time::Duration;
 
-	use lightning::util::test_utils::{TestLogger, TestStore};
+	use lightning::util::test_utils::TestLogger;
 
 	use super::*;
+	use crate::io::test_utils::InMemoryStore;
 
 	#[tokio::test]
 	async fn event_queue_persistence() {
-		let store: Arc<DynStore> = Arc::new(TestStore::new(false));
+		let store: Arc<DynStore> = Arc::new(InMemoryStore::new());
 		let logger = Arc::new(TestLogger::new());
 		let event_queue = Arc::new(EventQueue::new(Arc::clone(&store), Arc::clone(&logger)));
 		assert_eq!(event_queue.next_event(), None);
@@ -1647,7 +1648,7 @@ mod tests {
 
 	#[tokio::test]
 	async fn event_queue_concurrency() {
-		let store: Arc<DynStore> = Arc::new(TestStore::new(false));
+		let store: Arc<DynStore> = Arc::new(InMemoryStore::new());
 		let logger = Arc::new(TestLogger::new());
 		let event_queue = Arc::new(EventQueue::new(Arc::clone(&store), Arc::clone(&logger)));
 		assert_eq!(event_queue.next_event(), None);

src/io/test_utils.rs

@@ -5,19 +5,24 @@
 // http://opensource.org/licenses/MIT>, at your option. You may not use this file except in
 // accordance with one or both of these licenses.
 
+use std::boxed::Box;
+use std::collections::{hash_map, HashMap};
+use std::future::Future;
 use std::panic::RefUnwindSafe;
 use std::path::PathBuf;
+use std::pin::Pin;
+use std::sync::Mutex;
 
 use lightning::events::ClosureReason;
 use lightning::ln::functional_test_utils::{
 	connect_block, create_announced_chan_between_nodes, create_chanmon_cfgs, create_dummy_block,
 	create_network, create_node_cfgs, create_node_chanmgrs, send_payment, TestChanMonCfg,
 };
 use lightning::util::persist::{
-	KVStoreSync, MonitorUpdatingPersister, KVSTORE_NAMESPACE_KEY_MAX_LEN,
+	KVStore, KVStoreSync, MonitorUpdatingPersister, KVSTORE_NAMESPACE_KEY_MAX_LEN,
 };
 use lightning::util::test_utils;
-use lightning::{check_added_monitors, check_closed_broadcast, check_closed_event};
+use lightning::{check_added_monitors, check_closed_broadcast, check_closed_event, io};
 use rand::distr::Alphanumeric;
 use rand::{rng, Rng};
 
@@ -32,6 +37,125 @@ type TestMonitorUpdatePersister<'a, K> = MonitorUpdatingPersister<
 
 const EXPECTED_UPDATES_PER_PAYMENT: u64 = 5;
 
+pub struct InMemoryStore {
+	persisted_bytes: Mutex<HashMap<String, HashMap<String, Vec<u8>>>>,
+}
+
+impl InMemoryStore {
+	pub fn new() -> Self {
+		let persisted_bytes = Mutex::new(HashMap::new());
+		Self { persisted_bytes }
+	}
+
+	fn read_internal(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
+	) -> io::Result<Vec<u8>> {
+		let persisted_lock = self.persisted_bytes.lock().unwrap();
+		let prefixed = format!("{primary_namespace}/{secondary_namespace}");
+
+		if let Some(outer_ref) = persisted_lock.get(&prefixed) {
+			if let Some(inner_ref) = outer_ref.get(key) {
+				let bytes = inner_ref.clone();
+				Ok(bytes)
+			} else {
+				Err(io::Error::new(io::ErrorKind::NotFound, "Key not found"))
+			}
+		} else {
+			Err(io::Error::new(io::ErrorKind::NotFound, "Namespace not found"))
+		}
+	}
+
+	fn write_internal(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>,
+	) -> io::Result<()> {
+		let mut persisted_lock = self.persisted_bytes.lock().unwrap();
+
+		let prefixed = format!("{primary_namespace}/{secondary_namespace}");
+		let outer_e = persisted_lock.entry(prefixed).or_insert(HashMap::new());
+		outer_e.insert(key.to_string(), buf);
+		Ok(())
+	}
+
+	fn remove_internal(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, _lazy: bool,
+	) -> io::Result<()> {
+		let mut persisted_lock = self.persisted_bytes.lock().unwrap();
+
+		let prefixed = format!("{primary_namespace}/{secondary_namespace}");
+		if let Some(outer_ref) = persisted_lock.get_mut(&prefixed) {
+			outer_ref.remove(&key.to_string());
+		}
+
+		Ok(())
+	}
+
+	fn list_internal(
+		&self, primary_namespace: &str, secondary_namespace: &str,
+	) -> io::Result<Vec<String>> {
+		let mut persisted_lock = self.persisted_bytes.lock().unwrap();
+
+		let prefixed = format!("{primary_namespace}/{secondary_namespace}");
+		match persisted_lock.entry(prefixed) {
+			hash_map::Entry::Occupied(e) => Ok(e.get().keys().cloned().collect()),
+			hash_map::Entry::Vacant(_) => Ok(Vec::new()),
+		}
+	}
+}
+
+impl KVStore for InMemoryStore {
+	fn read(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
+	) -> Pin<Box<dyn Future<Output = Result<Vec<u8>, io::Error>> + 'static + Send>> {
+		let res = self.read_internal(&primary_namespace, &secondary_namespace, &key);
+		Box::pin(async move { res })
+	}
+	fn write(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>,
+	) -> Pin<Box<dyn Future<Output = Result<(), io::Error>> + 'static + Send>> {
+		let res = self.write_internal(&primary_namespace, &secondary_namespace, &key, buf);
+		Box::pin(async move { res })
+	}
+	fn remove(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
+	) -> Pin<Box<dyn Future<Output = Result<(), io::Error>> + 'static + Send>> {
+		let res = self.remove_internal(&primary_namespace, &secondary_namespace, &key, lazy);
+		Box::pin(async move { res })
+	}
+	fn list(
+		&self, primary_namespace: &str, secondary_namespace: &str,
+	) -> Pin<Box<dyn Future<Output = Result<Vec<String>, io::Error>> + 'static + Send>> {
+		let res = self.list_internal(primary_namespace, secondary_namespace);
+		Box::pin(async move { res })
+	}
+}
+
+impl KVStoreSync for InMemoryStore {
+	fn read(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
+	) -> io::Result<Vec<u8>> {
+		self.read_internal(primary_namespace, secondary_namespace, key)
+	}
+
+	fn write(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>,
+	) -> io::Result<()> {
+		self.write_internal(primary_namespace, secondary_namespace, key, buf)
+	}
+
+	fn remove(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
+	) -> io::Result<()> {
+		self.remove_internal(primary_namespace, secondary_namespace, key, lazy)
+	}
+
+	fn list(&self, primary_namespace: &str, secondary_namespace: &str) -> io::Result<Vec<String>> {
+		self.list_internal(primary_namespace, secondary_namespace)
+	}
+}
+
+unsafe impl Sync for InMemoryStore {}
+unsafe impl Send for InMemoryStore {}
+
 pub(crate) fn random_storage_path() -> PathBuf {
 	let mut temp_path = std::env::temp_dir();
 	let mut rng = rng();

src/payment/asynchronous/static_invoice_store.rs

@@ -157,15 +157,15 @@ mod tests {
 	use lightning::offers::offer::OfferBuilder;
 	use lightning::offers::static_invoice::{StaticInvoice, StaticInvoiceBuilder};
 	use lightning::sign::EntropySource;
-	use lightning::util::test_utils::TestStore;
 	use lightning_types::features::BlindedHopFeatures;
 
+	use crate::io::test_utils::InMemoryStore;
 	use crate::payment::asynchronous::static_invoice_store::StaticInvoiceStore;
 	use crate::types::DynStore;
 
 	#[tokio::test]
 	async fn static_invoice_store_test() {
-		let store: Arc<DynStore> = Arc::new(TestStore::new(false));
+		let store: Arc<DynStore> = Arc::new(InMemoryStore::new());
 		let static_invoice_store = StaticInvoiceStore::new(Arc::clone(&store));
 
 		let static_invoice = invoice();

src/peer_store.rs

@@ -152,13 +152,14 @@ mod tests {
 	use std::str::FromStr;
 	use std::sync::Arc;
 
-	use lightning::util::test_utils::{TestLogger, TestStore};
+	use lightning::util::test_utils::TestLogger;
 
 	use super::*;
+	use crate::io::test_utils::InMemoryStore;
 
 	#[test]
 	fn peer_info_persistence() {
-		let store: Arc<DynStore> = Arc::new(TestStore::new(false));
+		let store: Arc<DynStore> = Arc::new(InMemoryStore::new());
 		let logger = Arc::new(TestLogger::new());
 		let peer_store = PeerStore::new(Arc::clone(&store), Arc::clone(&logger));