Add async migratable filesystem stores

diff --git a/lightning-persister/src/fs_store/common.rs b/lightning-persister/src/fs_store/common.rs
index 6eaa0db..885f806 100644
--- a/lightning-persister/src/fs_store/common.rs
+++ b/lightning-persister/src/fs_store/common.rs
@@ -470,6 +470,94 @@ impl FilesystemStoreInner {
 
 		Ok(keys)
 	}
+
+	fn list_all_keys(
+		&self, use_empty_ns_dir: bool,
+	) -> Result<Vec<(String, String, String)>, lightning::io::Error> {
+		let prefixed_dest = &self.data_dir;
+		if !prefixed_dest.exists() {
+			return Ok(Vec::new());
+		}
+
+		let mut keys = Vec::new();
+
+		'primary_loop: for primary_entry in fs::read_dir(prefixed_dest)? {
+			let primary_entry = primary_entry?;
+			let primary_path = primary_entry.path();
+			if dir_entry_is_store_artifact(&primary_path) {
+				continue 'primary_loop;
+			}
+
+			if dir_entry_is_key(&primary_entry)? {
+				let primary_namespace = String::new();
+				let secondary_namespace = String::new();
+				let key = get_key_from_dir_entry_path(&primary_path, prefixed_dest, false)?;
+				keys.push((primary_namespace, secondary_namespace, key));
+				continue 'primary_loop;
+			}
+
+			// The primary_entry is actually also a directory.
+			'secondary_loop: for secondary_entry in fs::read_dir(&primary_path)? {
+				let secondary_entry = secondary_entry?;
+				let secondary_path = secondary_entry.path();
+				if dir_entry_is_store_artifact(&secondary_path) {
+					continue 'secondary_loop;
+				}
+
+				if dir_entry_is_key(&secondary_entry)? {
+					let primary_namespace = get_key_from_dir_entry_path(
+						&primary_path,
+						prefixed_dest,
+						use_empty_ns_dir,
+					)?;
+					let secondary_namespace = String::new();
+					let key = get_key_from_dir_entry_path(&secondary_path, &primary_path, false)?;
+					keys.push((primary_namespace, secondary_namespace, key));
+					continue 'secondary_loop;
+				}
+
+				// The secondary_entry is actually also a directory.
+				for tertiary_entry in fs::read_dir(&secondary_path)? {
+					let tertiary_entry = tertiary_entry?;
+					let tertiary_path = tertiary_entry.path();
+					if dir_entry_is_store_artifact(&tertiary_path) {
+						continue;
+					}
+
+					if dir_entry_is_key(&tertiary_entry)? {
+						let primary_namespace = get_key_from_dir_entry_path(
+							&primary_path,
+							prefixed_dest,
+							use_empty_ns_dir,
+						)?;
+						let secondary_namespace = get_key_from_dir_entry_path(
+							&secondary_path,
+							&primary_path,
+							use_empty_ns_dir,
+						)?;
+						let key =
+							get_key_from_dir_entry_path(&tertiary_path, &secondary_path, false)?;
+						keys.push((primary_namespace, secondary_namespace, key));
+					} else {
+						debug_assert!(
+							false,
+							"Failed to list keys of path {}: only two levels of namespaces are supported",
+							PrintableString(tertiary_path.to_str().unwrap_or_default())
+						);
+						let msg = format!(
+							"Failed to list keys of path {}: only two levels of namespaces are supported",
+							PrintableString(tertiary_path.to_str().unwrap_or_default())
+						);
+						return Err(lightning::io::Error::new(
+							lightning::io::ErrorKind::Other,
+							msg,
+						));
+					}
+				}
+			}
+		}
+		Ok(keys)
+	}
 }
 
 impl FilesystemStoreState {
@@ -640,92 +728,26 @@ impl FilesystemStoreState {
 		}
 	}
 
-	pub(crate) fn list_all_keys_impl(
+	#[cfg(feature = "tokio")]
+	pub(crate) fn list_all_keys_async(
 		&self, use_empty_ns_dir: bool,
-	) -> Result<Vec<(String, String, String)>, lightning::io::Error> {
-		let prefixed_dest = &self.inner.data_dir;
-		if !prefixed_dest.exists() {
-			return Ok(Vec::new());
-		}
-
-		let mut keys = Vec::new();
-
-		'primary_loop: for primary_entry in fs::read_dir(prefixed_dest)? {
-			let primary_entry = primary_entry?;
-			let primary_path = primary_entry.path();
-			if dir_entry_is_store_artifact(&primary_path) {
-				continue 'primary_loop;
-			}
-
-			if dir_entry_is_key(&primary_entry)? {
-				let primary_namespace = String::new();
-				let secondary_namespace = String::new();
-				let key = get_key_from_dir_entry_path(&primary_path, prefixed_dest, false)?;
-				keys.push((primary_namespace, secondary_namespace, key));
-				continue 'primary_loop;
-			}
-
-			// The primary_entry is actually also a directory.
-			'secondary_loop: for secondary_entry in fs::read_dir(&primary_path)? {
-				let secondary_entry = secondary_entry?;
-				let secondary_path = secondary_entry.path();
-				if dir_entry_is_store_artifact(&secondary_path) {
-					continue 'secondary_loop;
-				}
-
-				if dir_entry_is_key(&secondary_entry)? {
-					let primary_namespace = get_key_from_dir_entry_path(
-						&primary_path,
-						prefixed_dest,
-						use_empty_ns_dir,
-					)?;
-					let secondary_namespace = String::new();
-					let key = get_key_from_dir_entry_path(&secondary_path, &primary_path, false)?;
-					keys.push((primary_namespace, secondary_namespace, key));
-					continue 'secondary_loop;
-				}
-
-				// The secondary_entry is actually also a directory.
-				for tertiary_entry in fs::read_dir(&secondary_path)? {
-					let tertiary_entry = tertiary_entry?;
-					let tertiary_path = tertiary_entry.path();
-					if dir_entry_is_store_artifact(&tertiary_path) {
-						continue;
-					}
+	) -> impl Future<Output = Result<Vec<(String, String, String)>, lightning::io::Error>> + 'static + Send
+	{
+		let this = Arc::clone(&self.inner);
 
-					if dir_entry_is_key(&tertiary_entry)? {
-						let primary_namespace = get_key_from_dir_entry_path(
-							&primary_path,
-							prefixed_dest,
-							use_empty_ns_dir,
-						)?;
-						let secondary_namespace = get_key_from_dir_entry_path(
-							&secondary_path,
-							&primary_path,
-							use_empty_ns_dir,
-						)?;
-						let key =
-							get_key_from_dir_entry_path(&tertiary_path, &secondary_path, false)?;
-						keys.push((primary_namespace, secondary_namespace, key));
-					} else {
-						debug_assert!(
-							false,
-							"Failed to list keys of path {}: only two levels of namespaces are supported",
-							PrintableString(tertiary_path.to_str().unwrap_or_default())
-						);
-						let msg = format!(
-							"Failed to list keys of path {}: only two levels of namespaces are supported",
-							PrintableString(tertiary_path.to_str().unwrap_or_default())
-						);
-						return Err(lightning::io::Error::new(
-							lightning::io::ErrorKind::Other,
-							msg,
-						));
-					}
-				}
-			}
+		async move {
+			tokio::task::spawn_blocking(move || this.list_all_keys(use_empty_ns_dir))
+				.await
+				.unwrap_or_else(|e| {
+					Err(lightning::io::Error::new(lightning::io::ErrorKind::Other, e))
+				})
 		}
-		Ok(keys)
+	}
+
+	pub(crate) fn list_all_keys_impl(
+		&self, use_empty_ns_dir: bool,
+	) -> Result<Vec<(String, String, String)>, lightning::io::Error> {
+		self.inner.list_all_keys(use_empty_ns_dir)
 	}
 }
 
diff --git a/lightning-persister/src/fs_store/v1.rs b/lightning-persister/src/fs_store/v1.rs
index 4768b81..4f24d8d 100644
--- a/lightning-persister/src/fs_store/v1.rs
+++ b/lightning-persister/src/fs_store/v1.rs
@@ -94,9 +94,21 @@ impl MigratableKVStoreSync for FilesystemStore {
 	}
 }
 
+#[cfg(feature = "tokio")]
+impl lightning::util::persist::MigratableKVStore for FilesystemStore {
+	fn list_all_keys(
+		&self,
+	) -> impl Future<Output = Result<Vec<(String, String, String)>, lightning::io::Error>> + 'static + Send
+	{
+		self.state.list_all_keys_async(false)
+	}
+}
+
 #[cfg(test)]
 mod tests {
 	use super::*;
+	#[cfg(feature = "tokio")]
+	use crate::test_utils::do_test_data_migration_async;
 	use crate::test_utils::{
 		do_read_write_remove_list_persist, do_test_data_migration, do_test_store,
 	};
@@ -221,6 +233,20 @@ mod tests {
 		do_test_data_migration(&mut source_store, &mut target_store);
 	}
 
+	#[cfg(feature = "tokio")]
+	#[tokio::test]
+	async fn test_data_migration_async() {
+		let mut source_temp_path = std::env::temp_dir();
+		source_temp_path.push("test_data_migration_source_async");
+		let source_store = FilesystemStore::new(source_temp_path);
+
+		let mut target_temp_path = std::env::temp_dir();
+		target_temp_path.push("test_data_migration_target_async");
+		let target_store = FilesystemStore::new(target_temp_path);
+
+		do_test_data_migration_async(&source_store, &target_store).await;
+	}
+
 	#[test]
 	fn test_if_monitors_is_not_dir() {
 		let store = FilesystemStore::new("test_monitors_is_not_dir".into());
diff --git a/lightning-persister/src/fs_store/v2.rs b/lightning-persister/src/fs_store/v2.rs
index fd18e20..fe1fdf6 100644
--- a/lightning-persister/src/fs_store/v2.rs
+++ b/lightning-persister/src/fs_store/v2.rs
@@ -321,6 +321,16 @@ impl MigratableKVStoreSync for FilesystemStoreV2 {
 	}
 }
 
+#[cfg(feature = "tokio")]
+impl lightning::util::persist::MigratableKVStore for FilesystemStoreV2 {
+	fn list_all_keys(
+		&self,
+	) -> impl Future<Output = Result<Vec<(String, String, String)>, lightning::io::Error>> + 'static + Send
+	{
+		self.inner.list_all_keys_async(true)
+	}
+}
+
 /// Formats a page token from mtime (millis since epoch) and key.
 pub(crate) fn format_page_token(mtime_millis: u64, key: &str) -> String {
 	format!("{mtime_millis:016}:{key}")
@@ -351,6 +361,8 @@ pub(crate) fn parse_page_token(token: &str) -> lightning::io::Result<(u64, Strin
 mod tests {
 	use super::*;
 	use crate::fs_store::common::EMPTY_NAMESPACE_DIR;
+	#[cfg(feature = "tokio")]
+	use crate::test_utils::do_test_data_migration_async;
 	use crate::test_utils::{
 		do_read_write_remove_list_persist, do_test_data_migration, do_test_store,
 	};
@@ -445,6 +457,20 @@ mod tests {
 		do_test_data_migration(&mut source_store, &mut target_store);
 	}
 
+	#[cfg(feature = "tokio")]
+	#[tokio::test]
+	async fn test_data_migration_async() {
+		let mut source_temp_path = std::env::temp_dir();
+		source_temp_path.push("test_data_migration_source_async_v2");
+		let source_store = FilesystemStoreV2::new(source_temp_path).unwrap();
+
+		let mut target_temp_path = std::env::temp_dir();
+		target_temp_path.push("test_data_migration_target_async_v2");
+		let target_store = FilesystemStoreV2::new(target_temp_path).unwrap();
+
+		do_test_data_migration_async(&source_store, &target_store).await;
+	}
+
 	#[test]
 	fn test_filesystem_store_v2() {
 		// Create the nodes, giving them FilesystemStoreV2s for data stores.
diff --git a/lightning-persister/src/test_utils.rs b/lightning-persister/src/test_utils.rs
index 115f251..34e0619 100644
--- a/lightning-persister/src/test_utils.rs
+++ b/lightning-persister/src/test_utils.rs
@@ -59,15 +59,11 @@ pub(crate) fn do_read_write_remove_list_persist<K: KVStoreSync + RefUnwindSafe>(
 	assert_eq!(listed_keys.len(), 0);
 }
 
-pub(crate) fn do_test_data_migration<S: MigratableKVStoreSync, T: MigratableKVStoreSync>(
-	source_store: &mut S, target_store: &mut T,
-) {
-	// We fill the source with some bogus keys.
-	let dummy_data = vec![42u8; 32];
+fn data_migration_test_keys() -> Vec<(String, String, String)> {
 	let num_primary_namespaces = 3;
 	let num_secondary_namespaces = 3;
 	let num_keys = 3;
-	let mut expected_keys = Vec::new();
+	let mut keys = Vec::new();
 	for i in 0..num_primary_namespaces {
 		let primary_namespace = if i == 0 {
 			String::new()
@@ -83,13 +79,25 @@ pub(crate) fn do_test_data_migration<S: MigratableKVStoreSync, T: MigratableKVSt
 			for k in 0..num_keys {
 				let key =
 					format!("testkey{}", KVSTORE_NAMESPACE_KEY_ALPHABET.chars().nth(k).unwrap());
-				source_store
-					.write(&primary_namespace, &secondary_namespace, &key, dummy_data.clone())
-					.unwrap();
-				expected_keys.push((primary_namespace.clone(), secondary_namespace.clone(), key));
+				keys.push((primary_namespace.clone(), secondary_namespace.clone(), key));
 			}
 		}
 	}
+
+	keys
+}
+
+pub(crate) fn do_test_data_migration<S: MigratableKVStoreSync, T: MigratableKVStoreSync>(
+	source_store: &mut S, target_store: &mut T,
+) {
+	// We fill the source with some bogus keys.
+	let dummy_data = vec![42u8; 32];
+	let mut expected_keys = data_migration_test_keys();
+	for (primary_namespace, secondary_namespace, key) in &expected_keys {
+		source_store
+			.write(primary_namespace, secondary_namespace, key, dummy_data.clone())
+			.unwrap();
+	}
 	expected_keys.sort();
 	expected_keys.dedup();
 
@@ -108,6 +116,47 @@ pub(crate) fn do_test_data_migration<S: MigratableKVStoreSync, T: MigratableKVSt
 	}
 }
 
+#[cfg(feature = "tokio")]
+pub(crate) async fn do_test_data_migration_async<
+	S: lightning::util::persist::MigratableKVStore,
+	T: lightning::util::persist::MigratableKVStore,
+>(
+	source_store: &S, target_store: &T,
+) {
+	use lightning::util::persist::{migrate_kv_store_data_async, KVStore, MigratableKVStore};
+
+	// We fill the source with some bogus keys.
+	let dummy_data = vec![42u8; 32];
+	let mut expected_keys = data_migration_test_keys();
+	for (primary_namespace, secondary_namespace, key) in &expected_keys {
+		KVStore::write(
+			source_store,
+			primary_namespace,
+			secondary_namespace,
+			key,
+			dummy_data.clone(),
+		)
+		.await
+		.unwrap();
+	}
+	expected_keys.sort();
+	expected_keys.dedup();
+
+	let mut source_list = MigratableKVStore::list_all_keys(source_store).await.unwrap();
+	source_list.sort();
+	assert_eq!(source_list, expected_keys);
+
+	migrate_kv_store_data_async(source_store, target_store).await.unwrap();
+
+	let mut target_list = MigratableKVStore::list_all_keys(target_store).await.unwrap();
+	target_list.sort();
+	assert_eq!(target_list, expected_keys);
+
+	for (p, s, k) in expected_keys.iter() {
+		assert_eq!(KVStore::read(target_store, p, s, k).await.unwrap(), dummy_data.clone());
+	}
+}
+
 // Integration-test the given KVStore implementation. Test relaying a few payments and check that
 // the persisted data is updated the appropriate number of times.
 pub(crate) fn do_test_store<K: KVStoreSync + Sync>(store_0: &K, store_1: &K) {