Add FilesystemStoreV2 with paginated listing support

diff --git a/lightning-persister/src/fs_store/common.rs b/lightning-persister/src/fs_store/common.rs
index c4aa1d0..f2f5eeb 100644
--- a/lightning-persister/src/fs_store/common.rs
+++ b/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,9 +146,22 @@ 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)
@@ -137,11 +169,12 @@ impl FilesystemStoreInner {
 
 	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,57 +471,65 @@ 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(
@@ -476,6 +537,7 @@ impl FilesystemStoreState {
 			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 {
diff --git a/lightning-persister/src/fs_store/mod.rs b/lightning-persister/src/fs_store/mod.rs
index b460bdb..5fe7f65 100644
--- a/lightning-persister/src/fs_store/mod.rs
+++ b/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;
diff --git a/lightning-persister/src/fs_store/v1.rs b/lightning-persister/src/fs_store/v1.rs
index 8aa988e..776aba6 100644
--- a/lightning-persister/src/fs_store/v1.rs
+++ b/lightning-persister/src/fs_store/v1.rs
@@ -39,25 +39,25 @@ impl KVStoreSync for FilesystemStore {
 	fn read(
 		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
 	) -> Result<Vec<u8>, lightning::io::Error> {
-		self.state.read_impl(primary_namespace, secondary_namespace, key)
+		self.state.read_impl(primary_namespace, secondary_namespace, key, false)
 	}
 
 	fn write(
 		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>,
 	) -> Result<(), lightning::io::Error> {
-		self.state.write_impl(primary_namespace, secondary_namespace, key, buf)
+		self.state.write_impl(primary_namespace, secondary_namespace, key, buf, false)
 	}
 
 	fn remove(
 		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
 	) -> Result<(), lightning::io::Error> {
-		self.state.remove_impl(primary_namespace, secondary_namespace, key, lazy)
+		self.state.remove_impl(primary_namespace, secondary_namespace, key, lazy, false)
 	}
 
 	fn list(
 		&self, primary_namespace: &str, secondary_namespace: &str,
 	) -> Result<Vec<String>, lightning::io::Error> {
-		self.state.list_impl(primary_namespace, secondary_namespace)
+		self.state.list_impl(primary_namespace, secondary_namespace, false)
 	}
 }
 
@@ -66,31 +66,31 @@ impl KVStore for FilesystemStore {
 	fn read(
 		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
 	) -> impl Future<Output = Result<Vec<u8>, lightning::io::Error>> + 'static + Send {
-		self.state.read_async(primary_namespace, secondary_namespace, key)
+		self.state.read_async(primary_namespace, secondary_namespace, key, false)
 	}
 
 	fn write(
 		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>,
 	) -> impl Future<Output = Result<(), lightning::io::Error>> + 'static + Send {
-		self.state.write_async(primary_namespace, secondary_namespace, key, buf)
+		self.state.write_async(primary_namespace, secondary_namespace, key, buf, false)
 	}
 
 	fn remove(
 		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
 	) -> impl Future<Output = Result<(), lightning::io::Error>> + 'static + Send {
-		self.state.remove_async(primary_namespace, secondary_namespace, key, lazy)
+		self.state.remove_async(primary_namespace, secondary_namespace, key, lazy, false)
 	}
 
 	fn list(
 		&self, primary_namespace: &str, secondary_namespace: &str,
 	) -> impl Future<Output = Result<Vec<String>, lightning::io::Error>> + 'static + Send {
-		self.state.list_async(primary_namespace, secondary_namespace)
+		self.state.list_async(primary_namespace, secondary_namespace, false)
 	}
 }
 
 impl MigratableKVStore for FilesystemStore {
 	fn list_all_keys(&self) -> Result<Vec<(String, String, String)>, lightning::io::Error> {
-		self.state.list_all_keys_impl()
+		self.state.list_all_keys_impl(false)
 	}
 }
 
diff --git a/lightning-persister/src/fs_store/v2.rs b/lightning-persister/src/fs_store/v2.rs
new file mode 100644
index 0000000..7cff1d3
--- /dev/null
+++ b/lightning-persister/src/fs_store/v2.rs
@@ -0,0 +1,651 @@
+//! Objects related to [`FilesystemStoreV2`] live here.
+use crate::fs_store::common::{get_key_from_dir_entry_path, FilesystemStoreState};
+
+use lightning::util::persist::{
+	KVStoreSync, MigratableKVStore, PageToken, PaginatedKVStoreSync, PaginatedListResponse,
+};
+
+use std::fs;
+use std::path::PathBuf;
+use std::time::UNIX_EPOCH;
+
+#[cfg(feature = "tokio")]
+use core::future::Future;
+#[cfg(feature = "tokio")]
+use lightning::util::persist::{KVStore, PaginatedKVStore};
+use std::sync::Arc;
+
+/// A [`KVStore`] and [`KVStoreSync`] implementation that writes to and reads from the file system.
+///
+/// This is version 2 of the filesystem store which provides:
+/// - Consistent directory structure using `[empty]` for empty namespaces
+/// - File modification times for creation-order pagination
+/// - Support for [`PaginatedKVStoreSync`] with newest-first ordering
+///
+/// ## Directory Structure
+///
+/// Files are stored with a consistent two-level namespace hierarchy:
+/// ```text
+/// data_dir/
+///   [empty]/                      # empty primary namespace
+///     [empty]/                    # empty secondary namespace
+///       {key}
+///   primary_ns/
+///     [empty]/                    # empty secondary namespace
+///       {key}
+///     secondary_ns/
+///       {key}
+/// ```
+///
+/// ## File Ordering
+///
+/// Files are ordered by their modification time (mtime). When a file is created, it gets
+/// the current time. When updated, the original creation time is preserved by setting
+/// the mtime of the new file to match the original before the atomic rename.
+///
+/// [`KVStore`]: lightning::util::persist::KVStore
+pub struct FilesystemStoreV2 {
+	inner: Arc<FilesystemStoreState>,
+}
+
+impl FilesystemStoreV2 {
+	/// Constructs a new [`FilesystemStoreV2`].
+	///
+	/// Returns an error if the data directory already exists and contains files at the top level,
+	/// which would indicate it was previously used by a [`FilesystemStore`] (v1). The v2 store
+	/// expects only directories (namespaces) at the top level.
+	///
+	/// [`FilesystemStore`]: crate::fs_store::v1::FilesystemStore
+	pub fn new(data_dir: PathBuf) -> std::io::Result<Self> {
+		if data_dir.exists() {
+			for entry in fs::read_dir(&data_dir)? {
+				let entry = entry?;
+				if entry.file_type()?.is_file() {
+					return Err(std::io::Error::new(
+						std::io::ErrorKind::InvalidData,
+						format!(
+							"Found file `{}` in the top-level data directory. \
+							This indicates the directory was previously used by FilesystemStore (v1). \
+							Please migrate your data or use a different directory.",
+							entry.path().display()
+						),
+					));
+				}
+			}
+		}
+
+		Ok(Self { inner: Arc::new(FilesystemStoreState::new(data_dir)) })
+	}
+
+	/// Returns the data directory.
+	pub fn get_data_dir(&self) -> PathBuf {
+		self.inner.get_data_dir()
+	}
+
+	#[cfg(any(all(feature = "tokio", test), fuzzing))]
+	/// Returns the size of the async state.
+	pub fn state_size(&self) -> usize {
+		self.inner.state_size()
+	}
+}
+
+/// The fixed page size for paginated listing operations.
+pub(crate) const PAGE_SIZE: usize = 50;
+
+/// The length of the timestamp in a page token (milliseconds since epoch as 16-digit decimal).
+const PAGE_TOKEN_TIMESTAMP_LEN: usize = 16;
+
+impl FilesystemStoreState {
+	fn list_paginated_impl(
+		&self, prefixed_dest: PathBuf, page_token: Option<PageToken>,
+	) -> Result<PaginatedListResponse, lightning::io::Error> {
+		if !prefixed_dest.exists() {
+			return Ok(PaginatedListResponse { keys: Vec::new(), next_page_token: None });
+		}
+
+		// Collect all entries with their modification times
+		let mut entries: Vec<(u64, String)> = Vec::new();
+		for dir_entry in fs::read_dir(&prefixed_dest)? {
+			let dir_entry = dir_entry?;
+
+			let key =
+				get_key_from_dir_entry_path(&dir_entry.path(), prefixed_dest.as_path(), false)?;
+			// Get modification time as millis since epoch
+			let mtime_millis = dir_entry
+				.metadata()
+				.ok()
+				.and_then(|m| m.modified().ok())
+				.and_then(|t| t.duration_since(UNIX_EPOCH).ok())
+				.map(|d| d.as_millis() as u64)
+				.unwrap_or(0);
+
+			entries.push((mtime_millis, key));
+		}
+
+		// Sort by mtime descending (newest first), then by key descending for same mtime
+		entries.sort_by(|a, b| b.0.cmp(&a.0).then_with(|| b.1.cmp(&a.1)));
+
+		// Find starting position based on page token
+		let start_idx = if let Some(token) = page_token {
+			let (token_mtime, token_key) = parse_page_token(token.as_str())?;
+
+			// Find entries that come after the token (older entries = lower mtime)
+			// or same mtime but lexicographically smaller key (since we sort descending)
+			entries
+				.iter()
+				.position(|(mtime, key)| {
+					*mtime < token_mtime
+						|| (*mtime == token_mtime && key.as_str() < token_key.as_str())
+				})
+				.unwrap_or(entries.len())
+		} else {
+			0
+		};
+
+		// Take PAGE_SIZE entries starting from start_idx
+		let page_entries: Vec<_> =
+			entries.iter().skip(start_idx).take(PAGE_SIZE).cloned().collect();
+
+		let keys: Vec<String> = page_entries.iter().map(|(_, key)| key.clone()).collect();
+
+		// Determine next page token
+		let next_page_token = if start_idx + PAGE_SIZE < entries.len() {
+			page_entries.last().map(|(mtime, key)| PageToken::new(format_page_token(*mtime, key)))
+		} else {
+			None
+		};
+
+		Ok(PaginatedListResponse { keys, next_page_token })
+	}
+}
+
+impl KVStoreSync for FilesystemStoreV2 {
+	fn read(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
+	) -> Result<Vec<u8>, lightning::io::Error> {
+		self.inner.read_impl(primary_namespace, secondary_namespace, key, true)
+	}
+
+	fn write(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>,
+	) -> Result<(), lightning::io::Error> {
+		self.inner.write_impl(primary_namespace, secondary_namespace, key, buf, true)
+	}
+
+	fn remove(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
+	) -> Result<(), lightning::io::Error> {
+		self.inner.remove_impl(primary_namespace, secondary_namespace, key, lazy, true)
+	}
+
+	fn list(
+		&self, primary_namespace: &str, secondary_namespace: &str,
+	) -> Result<Vec<String>, lightning::io::Error> {
+		self.inner.list_impl(primary_namespace, secondary_namespace, true)
+	}
+}
+
+impl PaginatedKVStoreSync for FilesystemStoreV2 {
+	fn list_paginated(
+		&self, primary_namespace: &str, secondary_namespace: &str, page_token: Option<PageToken>,
+	) -> Result<PaginatedListResponse, lightning::io::Error> {
+		let prefixed_dest = self.inner.get_checked_dest_file_path(
+			primary_namespace,
+			secondary_namespace,
+			None,
+			"list_paginated",
+			true,
+		)?;
+		self.inner.list_paginated_impl(prefixed_dest, page_token)
+	}
+}
+
+#[cfg(feature = "tokio")]
+impl KVStore for FilesystemStoreV2 {
+	fn read(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
+	) -> impl Future<Output = Result<Vec<u8>, lightning::io::Error>> + 'static + Send {
+		self.inner.read_async(primary_namespace, secondary_namespace, key, true)
+	}
+
+	fn write(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>,
+	) -> impl Future<Output = Result<(), lightning::io::Error>> + 'static + Send {
+		self.inner.write_async(primary_namespace, secondary_namespace, key, buf, true)
+	}
+
+	fn remove(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
+	) -> impl Future<Output = Result<(), lightning::io::Error>> + 'static + Send {
+		self.inner.remove_async(primary_namespace, secondary_namespace, key, lazy, true)
+	}
+
+	fn list(
+		&self, primary_namespace: &str, secondary_namespace: &str,
+	) -> impl Future<Output = Result<Vec<String>, lightning::io::Error>> + 'static + Send {
+		self.inner.list_async(primary_namespace, secondary_namespace, true)
+	}
+}
+
+#[cfg(feature = "tokio")]
+impl PaginatedKVStore for FilesystemStoreV2 {
+	fn list_paginated(
+		&self, primary_namespace: &str, secondary_namespace: &str, page_token: Option<PageToken>,
+	) -> impl Future<Output = Result<PaginatedListResponse, 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_paginated",
+			true,
+		);
+
+		async move {
+			let path = match path {
+				Ok(path) => path,
+				Err(e) => return Err(e),
+			};
+			tokio::task::spawn_blocking(move || this.list_paginated_impl(path, page_token))
+				.await
+				.unwrap_or_else(|e| {
+					Err(lightning::io::Error::new(lightning::io::ErrorKind::Other, e))
+				})
+		}
+	}
+}
+
+impl MigratableKVStore for FilesystemStoreV2 {
+	fn list_all_keys(&self) -> Result<Vec<(String, String, String)>, lightning::io::Error> {
+		self.inner.list_all_keys_impl(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}")
+}
+
+/// Parses a page token into mtime (millis since epoch) and key.
+pub(crate) fn parse_page_token(token: &str) -> lightning::io::Result<(u64, String)> {
+	if token.as_bytes().get(PAGE_TOKEN_TIMESTAMP_LEN) != Some(&b':') {
+		return Err(lightning::io::Error::new(
+			lightning::io::ErrorKind::InvalidInput,
+			"Invalid page token format",
+		));
+	}
+
+	let mtime = token[..PAGE_TOKEN_TIMESTAMP_LEN].parse::<u64>().map_err(|_| {
+		lightning::io::Error::new(
+			lightning::io::ErrorKind::InvalidInput,
+			"Invalid page token timestamp",
+		)
+	})?;
+
+	let key = token[PAGE_TOKEN_TIMESTAMP_LEN + 1..].to_string();
+
+	Ok((mtime, key))
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use crate::fs_store::common::EMPTY_NAMESPACE_DIR;
+	use crate::test_utils::{
+		do_read_write_remove_list_persist, do_test_data_migration, do_test_store,
+	};
+	use std::fs::FileTimes;
+	use std::time::UNIX_EPOCH;
+
+	impl Drop for FilesystemStoreV2 {
+		fn drop(&mut self) {
+			// We test for invalid directory names, so it's OK if directory removal
+			// fails.
+			match fs::remove_dir_all(&self.inner.get_data_dir()) {
+				Err(e) => println!("Failed to remove test persister directory: {}", e),
+				_ => {},
+			}
+		}
+	}
+
+	#[test]
+	fn read_write_remove_list_persist() {
+		let mut temp_path = std::env::temp_dir();
+		temp_path.push("test_read_write_remove_list_persist_v2");
+		let fs_store = FilesystemStoreV2::new(temp_path).unwrap();
+		do_read_write_remove_list_persist(&fs_store);
+	}
+
+	#[cfg(feature = "tokio")]
+	#[tokio::test]
+	async fn read_write_remove_list_persist_async() {
+		use lightning::util::persist::KVStore;
+		use std::sync::Arc;
+
+		let mut temp_path = std::env::temp_dir();
+		temp_path.push("test_read_write_remove_list_persist_async_v2");
+		let fs_store = Arc::new(FilesystemStoreV2::new(temp_path).unwrap());
+		assert_eq!(fs_store.state_size(), 0);
+
+		let async_fs_store = Arc::clone(&fs_store);
+
+		let data1 = vec![42u8; 32];
+		let data2 = vec![43u8; 32];
+
+		let primary = "testspace";
+		let secondary = "testsubspace";
+		let key = "testkey";
+
+		// Test writing the same key twice with different data. Execute the asynchronous part out of order to ensure
+		// that eventual consistency works.
+		let fut1 = KVStore::write(&*async_fs_store, primary, secondary, key, data1);
+		assert_eq!(fs_store.state_size(), 1);
+
+		let fut2 = KVStore::remove(&*async_fs_store, primary, secondary, key, false);
+		assert_eq!(fs_store.state_size(), 1);
+
+		let fut3 = KVStore::write(&*async_fs_store, primary, secondary, key, data2.clone());
+		assert_eq!(fs_store.state_size(), 1);
+
+		fut3.await.unwrap();
+		assert_eq!(fs_store.state_size(), 1);
+
+		fut2.await.unwrap();
+		assert_eq!(fs_store.state_size(), 1);
+
+		fut1.await.unwrap();
+		assert_eq!(fs_store.state_size(), 0);
+
+		// Test list.
+		let listed_keys = KVStore::list(&*async_fs_store, primary, secondary).await.unwrap();
+		assert_eq!(listed_keys.len(), 1);
+		assert_eq!(listed_keys[0], key);
+
+		// Test read. We expect to read data2, as the write call was initiated later.
+		let read_data = KVStore::read(&*async_fs_store, primary, secondary, key).await.unwrap();
+		assert_eq!(data2, &*read_data);
+
+		// Test remove.
+		KVStore::remove(&*async_fs_store, primary, secondary, key, false).await.unwrap();
+
+		let listed_keys = KVStore::list(&*async_fs_store, primary, secondary).await.unwrap();
+		assert_eq!(listed_keys.len(), 0);
+	}
+
+	#[test]
+	fn test_data_migration() {
+		let mut source_temp_path = std::env::temp_dir();
+		source_temp_path.push("test_data_migration_source_v2");
+		let mut 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_v2");
+		let mut target_store = FilesystemStoreV2::new(target_temp_path).unwrap();
+
+		do_test_data_migration(&mut source_store, &mut target_store);
+	}
+
+	#[test]
+	fn test_filesystem_store_v2() {
+		// Create the nodes, giving them FilesystemStoreV2s for data stores.
+		let store_0 = FilesystemStoreV2::new("test_filesystem_store_v2_0".into()).unwrap();
+		let store_1 = FilesystemStoreV2::new("test_filesystem_store_v2_1".into()).unwrap();
+		do_test_store(&store_0, &store_1)
+	}
+
+	#[test]
+	fn test_page_token_format() {
+		let mtime: u64 = 1706500000000;
+		let key = "test_key";
+		let token = format_page_token(mtime, key);
+		assert_eq!(token, "0001706500000000:test_key");
+
+		let parsed = parse_page_token(&token).unwrap();
+		assert_eq!(parsed, (mtime, key.to_string()));
+
+		// Test invalid tokens
+		assert!(parse_page_token("invalid").is_err());
+		assert!(parse_page_token("0001706500000000_key").is_err()); // wrong separator
+		assert!(parse_page_token("0001706500000000").is_err()); // no separator and key
+		assert!(parse_page_token("1706500000000:key").is_err()); // too short timestamp
+	}
+
+	#[test]
+	fn test_directory_structure() {
+		use lightning::util::persist::KVStoreSync;
+
+		let mut temp_path = std::env::temp_dir();
+		temp_path.push("test_directory_structure_v2");
+		let fs_store = FilesystemStoreV2::new(temp_path.clone()).unwrap();
+
+		let data = vec![42u8; 32];
+
+		// Write with empty namespaces
+		KVStoreSync::write(&fs_store, "", "", "key1", data.clone()).unwrap();
+		assert!(temp_path.join(EMPTY_NAMESPACE_DIR).join(EMPTY_NAMESPACE_DIR).exists());
+
+		// Write with non-empty primary, empty secondary
+		KVStoreSync::write(&fs_store, "primary", "", "key2", data.clone()).unwrap();
+		assert!(temp_path.join("primary").join(EMPTY_NAMESPACE_DIR).exists());
+
+		// Write with both non-empty
+		KVStoreSync::write(&fs_store, "primary", "secondary", "key3", data.clone()).unwrap();
+		assert!(temp_path.join("primary").join("secondary").exists());
+
+		// Verify we can read them back
+		assert_eq!(KVStoreSync::read(&fs_store, "", "", "key1").unwrap(), data);
+		assert_eq!(KVStoreSync::read(&fs_store, "primary", "", "key2").unwrap(), data);
+		assert_eq!(KVStoreSync::read(&fs_store, "primary", "secondary", "key3").unwrap(), data);
+
+		// Verify files are named just by key (no timestamp prefix)
+		assert!(temp_path
+			.join(EMPTY_NAMESPACE_DIR)
+			.join(EMPTY_NAMESPACE_DIR)
+			.join("key1")
+			.exists());
+		assert!(temp_path.join("primary").join(EMPTY_NAMESPACE_DIR).join("key2").exists());
+		assert!(temp_path.join("primary").join("secondary").join("key3").exists());
+	}
+
+	#[test]
+	fn test_update_preserves_mtime() {
+		use lightning::util::persist::KVStoreSync;
+
+		let mut temp_path = std::env::temp_dir();
+		temp_path.push("test_update_preserves_mtime_v2");
+		let fs_store = FilesystemStoreV2::new(temp_path.clone()).unwrap();
+
+		let data1 = vec![42u8; 32];
+		let data2 = vec![43u8; 32];
+
+		// Write initial data
+		KVStoreSync::write(&fs_store, "ns", "sub", "key", data1).unwrap();
+
+		// Get the original mtime
+		let file_path = temp_path.join("ns").join("sub").join("key");
+		let original_mtime = fs::metadata(&file_path).unwrap().modified().unwrap();
+
+		// Sleep briefly to ensure different timestamp if not preserved
+		std::thread::sleep(std::time::Duration::from_millis(50));
+
+		// Update with new data
+		KVStoreSync::write(&fs_store, "ns", "sub", "key", data2.clone()).unwrap();
+
+		// Verify mtime is preserved
+		let updated_mtime = fs::metadata(&file_path).unwrap().modified().unwrap();
+		assert_eq!(original_mtime, updated_mtime);
+
+		// Verify data was updated
+		assert_eq!(KVStoreSync::read(&fs_store, "ns", "sub", "key").unwrap(), data2);
+	}
+
+	#[test]
+	fn test_paginated_listing() {
+		use lightning::util::persist::{KVStoreSync, PaginatedKVStoreSync};
+
+		let mut temp_path = std::env::temp_dir();
+		temp_path.push("test_paginated_listing_v2");
+		let fs_store = FilesystemStoreV2::new(temp_path).unwrap();
+
+		let data = vec![42u8; 32];
+
+		// Write several keys with small delays to ensure different mtimes
+		let keys: Vec<String> = (0..5).map(|i| format!("key{}", i)).collect();
+		for key in &keys {
+			KVStoreSync::write(&fs_store, "ns", "sub", key, data.clone()).unwrap();
+			std::thread::sleep(std::time::Duration::from_millis(10));
+		}
+
+		// List paginated - should return newest first
+		let response = PaginatedKVStoreSync::list_paginated(&fs_store, "ns", "sub", None).unwrap();
+		assert_eq!(response.keys.len(), 5);
+		// Newest key (key4) should be first
+		assert_eq!(response.keys[0], "key4");
+		assert_eq!(response.keys[4], "key0");
+		assert!(response.next_page_token.is_none()); // Less than PAGE_SIZE items
+	}
+
+	#[test]
+	fn test_paginated_listing_with_pagination() {
+		use lightning::util::persist::{KVStoreSync, PaginatedKVStoreSync};
+
+		let mut temp_path = std::env::temp_dir();
+		temp_path.push("test_paginated_listing_with_pagination_v2");
+		let fs_store = FilesystemStoreV2::new(temp_path).unwrap();
+
+		let data = vec![42u8; 32];
+
+		// Write more than PAGE_SIZE keys
+		let num_keys = PAGE_SIZE + 50;
+		for i in 0..num_keys {
+			let key = format!("key{:04}", i);
+			KVStoreSync::write(&fs_store, "ns", "sub", &key, data.clone()).unwrap();
+			// Small delay to ensure ordering
+			if i % 10 == 0 {
+				std::thread::sleep(std::time::Duration::from_millis(1));
+			}
+		}
+
+		// First page
+		let response1 = PaginatedKVStoreSync::list_paginated(&fs_store, "ns", "sub", None).unwrap();
+		assert_eq!(response1.keys.len(), PAGE_SIZE);
+		assert!(response1.next_page_token.is_some());
+
+		// Second page
+		let response2 =
+			PaginatedKVStoreSync::list_paginated(&fs_store, "ns", "sub", response1.next_page_token)
+				.unwrap();
+		assert_eq!(response2.keys.len(), 50);
+		assert!(response2.next_page_token.is_none());
+
+		// Verify no duplicates between pages
+		let all_keys: std::collections::HashSet<_> =
+			response1.keys.iter().chain(response2.keys.iter()).collect();
+		assert_eq!(all_keys.len(), num_keys);
+	}
+
+	#[test]
+	fn test_page_token_after_deletion() {
+		use lightning::util::persist::{KVStoreSync, PaginatedKVStoreSync};
+
+		let mut temp_path = std::env::temp_dir();
+		temp_path.push("test_page_token_after_deletion_v2");
+		let fs_store = FilesystemStoreV2::new(temp_path).unwrap();
+
+		let data = vec![42u8; 32];
+
+		// Write keys
+		for i in 0..10 {
+			let key = format!("key{}", i);
+			KVStoreSync::write(&fs_store, "ns", "sub", &key, data.clone()).unwrap();
+			std::thread::sleep(std::time::Duration::from_millis(10));
+		}
+
+		// Verify initial listing
+		let response1 = PaginatedKVStoreSync::list_paginated(&fs_store, "ns", "sub", None).unwrap();
+		assert_eq!(response1.keys.len(), 10);
+
+		// Delete some keys
+		KVStoreSync::remove(&fs_store, "ns", "sub", "key5", false).unwrap();
+		KVStoreSync::remove(&fs_store, "ns", "sub", "key3", false).unwrap();
+
+		// List again - should work fine with deleted keys
+		let response2 = PaginatedKVStoreSync::list_paginated(&fs_store, "ns", "sub", None).unwrap();
+		assert_eq!(response2.keys.len(), 8); // 10 - 2 deleted
+	}
+
+	#[test]
+	fn test_same_mtime_sorted_by_key() {
+		use lightning::util::persist::PaginatedKVStoreSync;
+		use std::time::Duration;
+
+		// Create files directly on disk first with the same mtime
+		let mut temp_path = std::env::temp_dir();
+		temp_path.push("test_same_mtime_sorted_by_key_v2");
+		let _ = fs::remove_dir_all(&temp_path);
+
+		let data = vec![42u8; 32];
+		let dir = temp_path.join("ns").join("sub");
+		fs::create_dir_all(&dir).unwrap();
+
+		// Write files with the same mtime but different keys
+		let keys = vec!["zebra", "apple", "mango", "banana"];
+		let fixed_time = UNIX_EPOCH + Duration::from_secs(1706500000);
+
+		for key in &keys {
+			let file_path = dir.join(key);
+			let file = fs::File::create(&file_path).unwrap();
+			std::io::Write::write_all(&mut &file, &data).unwrap();
+			file.set_times(FileTimes::new().set_modified(fixed_time)).unwrap();
+		}
+
+		// Open the store
+		let fs_store = FilesystemStoreV2::new(temp_path.clone()).unwrap();
+
+		// List paginated - should return keys sorted by key in reverse order
+		// (for same mtime, keys are sorted reverse alphabetically)
+		let response = PaginatedKVStoreSync::list_paginated(&fs_store, "ns", "sub", None).unwrap();
+		assert_eq!(response.keys.len(), 4);
+
+		// Same mtime means sorted by key in reverse order (z > m > b > a)
+		assert_eq!(response.keys[0], "zebra");
+		assert_eq!(response.keys[1], "mango");
+		assert_eq!(response.keys[2], "banana");
+		assert_eq!(response.keys[3], "apple");
+	}
+
+	#[test]
+	fn test_rejects_v1_data_directory() {
+		let mut temp_path = std::env::temp_dir();
+		temp_path.push("test_rejects_v1_data_directory");
+		let _ = fs::remove_dir_all(&temp_path);
+		fs::create_dir_all(&temp_path).unwrap();
+
+		// Create a file at the top level, as v1 would for an empty primary namespace
+		fs::write(temp_path.join("some_key"), b"data").unwrap();
+
+		// V2 construction should fail
+		match FilesystemStoreV2::new(temp_path.clone()) {
+			Err(err) => {
+				assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
+				assert!(err.to_string().contains("FilesystemStore (v1)"));
+			},
+			Ok(_) => panic!("Expected error for directory with top-level files"),
+		}
+
+		// Clean up
+		let _ = fs::remove_dir_all(&temp_path);
+
+		// An empty directory should succeed
+		fs::create_dir_all(&temp_path).unwrap();
+		let result = FilesystemStoreV2::new(temp_path.clone());
+		assert!(result.is_ok());
+
+		// A directory with only subdirectories should succeed
+		fs::create_dir_all(temp_path.join("some_namespace")).unwrap();
+		let result = FilesystemStoreV2::new(temp_path);
+		assert!(result.is_ok());
+	}
+}