Merge pull request #4658 from tnull/2026-06-migratable-kv-store-async

Add async variant of `MigratableKVStore`

Author: tnull

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)
 	}
 }
 

lightning-persister/src/fs_store/v1.rs

@@ -1,7 +1,7 @@
 //! Objects related to [`FilesystemStore`] live here.
 use crate::fs_store::common::FilesystemStoreState;
 
-use lightning::util::persist::{KVStoreSync, MigratableKVStore};
+use lightning::util::persist::{KVStoreSync, MigratableKVStoreSync};
 
 use std::path::PathBuf;
 
@@ -88,15 +88,27 @@ impl KVStore for FilesystemStore {
 	}
 }
 
-impl MigratableKVStore for FilesystemStore {
+impl MigratableKVStoreSync for FilesystemStore {
 	fn list_all_keys(&self) -> Result<Vec<(String, String, String)>, lightning::io::Error> {
 		self.state.list_all_keys_impl(false)
 	}
 }
 
+#[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());

lightning-persister/src/fs_store/v2.rs

@@ -4,7 +4,7 @@ use crate::fs_store::common::{
 };
 
 use lightning::util::persist::{
-	KVStoreSync, MigratableKVStore, PageToken, PaginatedKVStoreSync, PaginatedListResponse,
+	KVStoreSync, MigratableKVStoreSync, PageToken, PaginatedKVStoreSync, PaginatedListResponse,
 };
 
 use std::fs;
@@ -315,12 +315,22 @@ impl PaginatedKVStore for FilesystemStoreV2 {
 	}
 }
 
-impl MigratableKVStore for FilesystemStoreV2 {
+impl MigratableKVStoreSync for FilesystemStoreV2 {
 	fn list_all_keys(&self) -> Result<Vec<(String, String, String)>, lightning::io::Error> {
 		self.inner.list_all_keys_impl(true)
 	}
 }
 
+#[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.