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 <[email protected]>

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,15 @@ impl FilesystemStoreInner {
 		})
 	}
 
-	fn list(&self, prefixed_dest: PathBuf) -> lightning::io::Result<Vec<String>> {
+	fn list(
+		&self, prefixed_dest: PathBuf, retry_on_race: 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 retry_on_race { LIST_DIR_CONSISTENCY_RETRIES } else { 0 };
 
 		'retry_list: loop {
 			keys = Vec::new();
@@ -418,64 +465,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 +548,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 +567,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 +577,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,32 +605,50 @@ 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() {
 			return Ok(Vec::new());
 		}
 
+		// When use_empty_ns_dir is true (v2), namespace directories may be named
+		// [empty] to represent empty namespaces, so we resolve via namespace_from_dir_path.
+		// When false (v1), directory names are always valid kvstore strings.
+		let resolve_ns = |path: &Path, base: &Path| -> Result<String, lightning::io::Error> {
+			if use_empty_ns_dir {
+				namespace_from_dir_path(path)
+			} else {
+				get_key_from_dir_entry_path(path, base)
+			}
+		};
+
 		let mut keys = Vec::new();
 
 		'primary_loop: for primary_entry in fs::read_dir(prefixed_dest)? {
@@ -581,8 +669,7 @@ 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 = resolve_ns(&primary_path, prefixed_dest)?;
 					let secondary_namespace = String::new();
 					let key = get_key_from_dir_entry_path(&secondary_path, &primary_path)?;
 					keys.push((primary_namespace, secondary_namespace, key));
@@ -595,10 +682,8 @@ 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 primary_namespace = resolve_ns(&primary_path, prefixed_dest)?;
+						let secondary_namespace = resolve_ns(&secondary_path, &primary_path)?;
 						let key = get_key_from_dir_entry_path(&tertiary_path, &secondary_path)?;
 						keys.push((primary_namespace, secondary_namespace, key));
 					} else {
@@ -623,6 +708,25 @@ impl FilesystemStoreState {
 	}
 }
 
+/// Extracts a namespace string from a directory path, converting [`EMPTY_NAMESPACE_DIR`] to an
+/// empty string.
+fn namespace_from_dir_path(path: &Path) -> Result<String, lightning::io::Error> {
+	let name = path.file_name().and_then(|n| n.to_str()).ok_or_else(|| {
+		lightning::io::Error::new(
+			lightning::io::ErrorKind::Other,
+			format!(
+				"Failed to extract namespace from path {}",
+				PrintableString(path.to_str().unwrap_or_default())
+			),
+		)
+	})?;
+	if name == EMPTY_NAMESPACE_DIR {
+		Ok(String::new())
+	} else {
+		Ok(name.to_string())
+	}
+}
+
 fn dir_entry_is_key(dir_entry: &fs::DirEntry) -> Result<bool, lightning::io::Error> {
 	let p = dir_entry.path();
 	if let Some(ext) = p.extension() {
@@ -663,7 +767,9 @@ 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> {
+pub(crate) fn get_key_from_dir_entry_path(
+	p: &Path, base_path: &Path,
+) -> Result<String, lightning::io::Error> {
 	match p.strip_prefix(&base_path) {
 		Ok(stripped_path) => {
 			if let Some(relative_path) = stripped_path.to_str() {

lightning-persister/src/fs_store/mod.rs

@@ -1,5 +1,6 @@
 //! Implementations of filesystem-backed key-value stores.
 
 pub mod v1;
+pub mod v2;
 
 pub(crate) mod common;