Add FilesystemStoreV2 with paginated listing support

Implements PaginatedKVStore traits with timestamp-prefixed filenames for
newest-first pagination and [empty] directory markers for consistent
namespace hierarchy.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: benthecarman

lightning-persister/src/fs_store/common.rs

@@ -1,6 +1,7 @@
-//! Common utilities shared between [`FilesystemStore`].
+//! Common utilities shared between [`FilesystemStore`] and [`FilesystemStoreV2`] implementations.
 //!
 //! [`FilesystemStore`]: crate::fs_store::v1::FilesystemStore
+//! [`FilesystemStoreV2`]: crate::fs_store::v2::FilesystemStoreV2
 
 use crate::utils::{check_namespace_key_validity, is_valid_kvstore_str};
 
@@ -45,6 +46,11 @@ fn path_to_windows_str<T: AsRef<OsStr>>(path: &T) -> Vec<u16> {
 // a consistent view and error out.
 const LIST_DIR_CONSISTENCY_RETRIES: usize = 10;
 
+// The directory name used for empty namespaces in v2.
+// Uses brackets which are not in KVSTORE_NAMESPACE_KEY_ALPHABET, preventing collisions
+// with valid namespace names.
+pub(crate) const EMPTY_NAMESPACE_DIR: &str = "[empty]";
+
 /// Inner state shared between sync and async operations for filesystem stores.
 ///
 /// This struct manages the data directory, temporary file counter, and per-path locks
@@ -103,6 +109,19 @@ impl FilesystemStoreState {
 		let outer_lock = self.inner.locks.lock().unwrap();
 		outer_lock.len()
 	}
+
+	pub(crate) fn get_checked_dest_file_path(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: Option<&str>,
+		operation: &str, use_empty_ns_dir: bool,
+	) -> lightning::io::Result<PathBuf> {
+		self.inner.get_checked_dest_file_path(
+			primary_namespace,
+			secondary_namespace,
+			key,
+			operation,
+			use_empty_ns_dir,
+		)
+	}
 }
 
 impl FilesystemStoreInner {
@@ -112,7 +131,7 @@ impl FilesystemStoreInner {
 	}
 
 	fn get_dest_dir_path(
-		&self, primary_namespace: &str, secondary_namespace: &str,
+		&self, primary_namespace: &str, secondary_namespace: &str, use_empty_ns_dir: bool,
 	) -> std::io::Result<PathBuf> {
 		let mut dest_dir_path = {
 			#[cfg(target_os = "windows")]
@@ -127,21 +146,35 @@ impl FilesystemStoreInner {
 			}
 		};
 
-		dest_dir_path.push(primary_namespace);
-		if !secondary_namespace.is_empty() {
-			dest_dir_path.push(secondary_namespace);
+		if use_empty_ns_dir {
+			dest_dir_path.push(if primary_namespace.is_empty() {
+				EMPTY_NAMESPACE_DIR
+			} else {
+				primary_namespace
+			});
+			dest_dir_path.push(if secondary_namespace.is_empty() {
+				EMPTY_NAMESPACE_DIR
+			} else {
+				secondary_namespace
+			});
+		} else {
+			dest_dir_path.push(primary_namespace);
+			if !secondary_namespace.is_empty() {
+				dest_dir_path.push(secondary_namespace);
+			}
 		}
 
 		Ok(dest_dir_path)
 	}
 
 	fn get_checked_dest_file_path(
 		&self, primary_namespace: &str, secondary_namespace: &str, key: Option<&str>,
-		operation: &str,
+		operation: &str, use_empty_ns_dir: bool,
 	) -> lightning::io::Result<PathBuf> {
 		check_namespace_key_validity(primary_namespace, secondary_namespace, key, operation)?;
 
-		let mut dest_file_path = self.get_dest_dir_path(primary_namespace, secondary_namespace)?;
+		let mut dest_file_path =
+			self.get_dest_dir_path(primary_namespace, secondary_namespace, use_empty_ns_dir)?;
 		if let Some(key) = key {
 			dest_file_path.push(key);
 		}
@@ -217,8 +250,13 @@ impl FilesystemStoreInner {
 	/// returns early without writing.
 	fn write_version(
 		&self, inner_lock_ref: Arc<RwLock<u64>>, dest_file_path: PathBuf, buf: Vec<u8>,
-		version: u64,
+		version: u64, preserve_mtime: bool,
 	) -> lightning::io::Result<()> {
+		let mtime = if preserve_mtime {
+			fs::metadata(&dest_file_path).ok().and_then(|m| m.modified().ok())
+		} else {
+			None
+		};
 		let parent_directory = dest_file_path.parent().ok_or_else(|| {
 			let msg =
 				format!("Could not retrieve parent directory of {}.", dest_file_path.display());
@@ -238,6 +276,13 @@ impl FilesystemStoreInner {
 		{
 			let mut tmp_file = fs::File::create(&tmp_file_path)?;
 			tmp_file.write_all(&buf)?;
+
+			// If we need to preserve the original mtime (for updates), set it before fsync.
+			if let Some(mtime) = mtime {
+				let times = fs::FileTimes::new().set_modified(mtime);
+				tmp_file.set_times(times)?;
+			}
+
 			tmp_file.sync_all()?;
 		}
 
@@ -370,13 +415,13 @@ impl FilesystemStoreInner {
 		})
 	}
 
-	fn list(&self, prefixed_dest: PathBuf) -> lightning::io::Result<Vec<String>> {
+	fn list(&self, prefixed_dest: PathBuf, is_v2: bool) -> lightning::io::Result<Vec<String>> {
 		if !Path::new(&prefixed_dest).exists() {
 			return Ok(Vec::new());
 		}
 
 		let mut keys;
-		let mut retries = LIST_DIR_CONSISTENCY_RETRIES;
+		let mut retries = if is_v2 { 0 } else { LIST_DIR_CONSISTENCY_RETRIES };
 
 		'retry_list: loop {
 			keys = Vec::new();
@@ -387,7 +432,7 @@ impl FilesystemStoreInner {
 				let res = dir_entry_is_key(&entry);
 				match res {
 					Ok(true) => {
-						let key = get_key_from_dir_entry_path(&p, &prefixed_dest)?;
+						let key = get_key_from_dir_entry_path(&p, &prefixed_dest, false)?;
 						keys.push(key);
 					},
 					Ok(false) => {
@@ -396,6 +441,14 @@ impl FilesystemStoreInner {
 						continue 'skip_entry;
 					},
 					Err(e) => {
+						// In version 2 if a file has been deleted between the `read_dir` and our attempt
+						// to access it, we should just add it to the list to give a more consistent view.
+						if is_v2 {
+							let key = get_key_from_dir_entry_path(&p, &prefixed_dest, false)?;
+							keys.push(key);
+							continue 'skip_entry;
+						}
+
 						if e.kind() == lightning::io::ErrorKind::NotFound && retries > 0 {
 							// We had found the entry in `read_dir` above, so some race happend.
 							// Retry the `read_dir` to get a consistent view.
@@ -418,64 +471,73 @@ impl FilesystemStoreInner {
 impl FilesystemStoreState {
 	pub(crate) fn read_impl(
 		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
+		use_empty_ns_dir: bool,
 	) -> Result<Vec<u8>, lightning::io::Error> {
 		let path = self.inner.get_checked_dest_file_path(
 			primary_namespace,
 			secondary_namespace,
 			Some(key),
 			"read",
+			use_empty_ns_dir,
 		)?;
 		self.inner.read(path)
 	}
 
 	pub(crate) fn write_impl(
 		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>,
+		use_empty_ns_dir: bool,
 	) -> Result<(), lightning::io::Error> {
 		let path = self.inner.get_checked_dest_file_path(
 			primary_namespace,
 			secondary_namespace,
 			Some(key),
 			"write",
+			use_empty_ns_dir,
 		)?;
 		let (inner_lock_ref, version) = self.get_new_version_and_lock_ref(path.clone());
-		self.inner.write_version(inner_lock_ref, path, buf, version)
+		self.inner.write_version(inner_lock_ref, path, buf, version, use_empty_ns_dir)
 	}
 
 	pub(crate) fn remove_impl(
 		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
+		use_empty_ns_dir: bool,
 	) -> Result<(), lightning::io::Error> {
 		let path = self.inner.get_checked_dest_file_path(
 			primary_namespace,
 			secondary_namespace,
 			Some(key),
 			"remove",
+			use_empty_ns_dir,
 		)?;
 		let (inner_lock_ref, version) = self.get_new_version_and_lock_ref(path.clone());
 		self.inner.remove_version(inner_lock_ref, path, lazy, version)
 	}
 
 	pub(crate) fn list_impl(
-		&self, primary_namespace: &str, secondary_namespace: &str,
+		&self, primary_namespace: &str, secondary_namespace: &str, use_empty_ns_dir: bool,
 	) -> Result<Vec<String>, lightning::io::Error> {
 		let path = self.inner.get_checked_dest_file_path(
 			primary_namespace,
 			secondary_namespace,
 			None,
 			"list",
+			use_empty_ns_dir,
 		)?;
-		self.inner.list(path)
+		self.inner.list(path, use_empty_ns_dir)
 	}
 
 	#[cfg(feature = "tokio")]
 	pub(crate) fn read_async(
 		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
+		use_empty_ns_dir: bool,
 	) -> impl Future<Output = Result<Vec<u8>, lightning::io::Error>> + 'static + Send {
 		let this = Arc::clone(&self.inner);
 		let path = this.get_checked_dest_file_path(
 			primary_namespace,
 			secondary_namespace,
 			Some(key),
 			"read",
+			use_empty_ns_dir,
 		);
 
 		async move {
@@ -492,10 +554,17 @@ impl FilesystemStoreState {
 	#[cfg(feature = "tokio")]
 	pub(crate) fn write_async(
 		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>,
+		use_empty_ns_dir: bool,
 	) -> impl Future<Output = Result<(), lightning::io::Error>> + 'static + Send {
 		let this = Arc::clone(&self.inner);
 		let path = this
-			.get_checked_dest_file_path(primary_namespace, secondary_namespace, Some(key), "write")
+			.get_checked_dest_file_path(
+				primary_namespace,
+				secondary_namespace,
+				Some(key),
+				"write",
+				use_empty_ns_dir,
+			)
 			.map(|path| (self.get_new_version_and_lock_ref(path.clone()), path));
 
 		async move {
@@ -504,7 +573,7 @@ impl FilesystemStoreState {
 				Err(e) => return Err(e),
 			};
 			tokio::task::spawn_blocking(move || {
-				this.write_version(inner_lock_ref, path, buf, version)
+				this.write_version(inner_lock_ref, path, buf, version, use_empty_ns_dir)
 			})
 			.await
 			.unwrap_or_else(|e| Err(lightning::io::Error::new(lightning::io::ErrorKind::Other, e)))
@@ -514,10 +583,17 @@ impl FilesystemStoreState {
 	#[cfg(feature = "tokio")]
 	pub(crate) fn remove_async(
 		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
+		use_empty_ns_dir: bool,
 	) -> impl Future<Output = Result<(), lightning::io::Error>> + 'static + Send {
 		let this = Arc::clone(&self.inner);
 		let path = this
-			.get_checked_dest_file_path(primary_namespace, secondary_namespace, Some(key), "remove")
+			.get_checked_dest_file_path(
+				primary_namespace,
+				secondary_namespace,
+				Some(key),
+				"remove",
+				use_empty_ns_dir,
+			)
 			.map(|path| (self.get_new_version_and_lock_ref(path.clone()), path));
 
 		async move {
@@ -535,26 +611,33 @@ impl FilesystemStoreState {
 
 	#[cfg(feature = "tokio")]
 	pub(crate) fn list_async(
-		&self, primary_namespace: &str, secondary_namespace: &str,
+		&self, primary_namespace: &str, secondary_namespace: &str, use_empty_ns_dir: bool,
 	) -> impl Future<Output = Result<Vec<String>, lightning::io::Error>> + 'static + Send {
 		let this = Arc::clone(&self.inner);
 
-		let path =
-			this.get_checked_dest_file_path(primary_namespace, secondary_namespace, None, "list");
+		let path = this.get_checked_dest_file_path(
+			primary_namespace,
+			secondary_namespace,
+			None,
+			"list",
+			use_empty_ns_dir,
+		);
 
 		async move {
 			let path = match path {
 				Ok(path) => path,
 				Err(e) => return Err(e),
 			};
-			tokio::task::spawn_blocking(move || this.list(path)).await.unwrap_or_else(|e| {
-				Err(lightning::io::Error::new(lightning::io::ErrorKind::Other, e))
-			})
+			tokio::task::spawn_blocking(move || this.list(path, use_empty_ns_dir))
+				.await
+				.unwrap_or_else(|e| {
+					Err(lightning::io::Error::new(lightning::io::ErrorKind::Other, e))
+				})
 		}
 	}
 
 	pub(crate) fn list_all_keys_impl(
-		&self,
+		&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() {
@@ -570,7 +653,7 @@ impl FilesystemStoreState {
 			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)?;
+				let key = get_key_from_dir_entry_path(&primary_path, prefixed_dest, false)?;
 				keys.push((primary_namespace, secondary_namespace, key));
 				continue 'primary_loop;
 			}
@@ -581,10 +664,13 @@ impl FilesystemStoreState {
 				let secondary_path = secondary_entry.path();
 
 				if dir_entry_is_key(&secondary_entry)? {
-					let primary_namespace =
-						get_key_from_dir_entry_path(&primary_path, prefixed_dest)?;
+					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)?;
+					let key = get_key_from_dir_entry_path(&secondary_path, &primary_path, false)?;
 					keys.push((primary_namespace, secondary_namespace, key));
 					continue 'secondary_loop;
 				}
@@ -595,11 +681,18 @@ impl FilesystemStoreState {
 					let tertiary_path = tertiary_entry.path();
 
 					if dir_entry_is_key(&tertiary_entry)? {
-						let primary_namespace =
-							get_key_from_dir_entry_path(&primary_path, prefixed_dest)?;
-						let secondary_namespace =
-							get_key_from_dir_entry_path(&secondary_path, &primary_path)?;
-						let key = get_key_from_dir_entry_path(&tertiary_path, &secondary_path)?;
+						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!(
@@ -663,10 +756,18 @@ fn dir_entry_is_key(dir_entry: &fs::DirEntry) -> Result<bool, lightning::io::Err
 	Ok(true)
 }
 
-fn get_key_from_dir_entry_path(p: &Path, base_path: &Path) -> Result<String, lightning::io::Error> {
+/// Gets the key from a directory entry path by stripping the base path and validating the result.
+/// If `map_empty_ns_dir` is true, treats entries with the name of `EMPTY_NAMESPACE_DIR` as an empty string.
+/// `map_empty_ns_dir` should always be false when reading keys and only be true when listing namespaces.
+pub(crate) fn get_key_from_dir_entry_path(
+	p: &Path, base_path: &Path, map_empty_ns_dir: bool,
+) -> Result<String, lightning::io::Error> {
 	match p.strip_prefix(&base_path) {
 		Ok(stripped_path) => {
 			if let Some(relative_path) = stripped_path.to_str() {
+				if map_empty_ns_dir && relative_path == EMPTY_NAMESPACE_DIR {
+					return Ok(String::new());
+				}
 				if is_valid_kvstore_str(relative_path) {
 					return Ok(relative_path.to_string());
 				} else {