Add PaginatedKVStore traits upstreamed from ldk-server

Allows for a paginated KV store for more efficient listing of keys so
you don't need to fetch all at once. Uses monotonic counter or timestamp
to track the order of keys and allow for pagination. The traits are
largely just copy-pasted from ldk-server.

Adds some basic tests that were generated using claude code.

Author: benthecarman

lightning/src/util/persist.rs

@@ -347,6 +347,225 @@ pub trait KVStore {
 	) -> impl Future<Output = Result<Vec<String>, io::Error>> + 'static + MaybeSend;
 }
 
+/// An opaque token used for paginated key listing.
+///
+/// This token is returned by [`PaginatedKVStore::list_paginated`] and can be passed to subsequent
+/// calls to retrieve the next page of results. The internal representation is implementation-defined
+/// and should be treated as opaque by callers.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct PageToken(pub String);
+
+/// Represents the response from a paginated `list` operation.
+///
+/// Contains the list of keys and an optional `next_page_token` that can be used to retrieve the
+/// next set of keys.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct PaginatedListResponse {
+	/// A vector of keys, ordered in descending order of `order`.
+	pub keys: Vec<String>,
+
+	/// A token that can be used to retrieve the next set of keys.
+	///
+	/// Is `None` if there are no more pages to retrieve.
+	pub next_page_token: Option<PageToken>,
+}
+
+/// Provides an interface that allows storage and retrieval of persisted values that are associated
+/// with given keys, with support for pagination.
+///
+/// In order to avoid collisions, the key space is segmented based on the given `primary_namespace`s
+/// and `secondary_namespace`s. Implementations of this trait are free to handle them in different
+/// ways, as long as per-namespace key uniqueness is asserted.
+///
+/// Keys and namespaces are required to be valid ASCII strings in the range of
+/// [`KVSTORE_NAMESPACE_KEY_ALPHABET`] and no longer than [`KVSTORE_NAMESPACE_KEY_MAX_LEN`]. Empty
+/// primary namespaces and secondary namespaces (`""`) are considered valid; however, if
+/// `primary_namespace` is empty, `secondary_namespace` must also be empty. This means that concerns
+/// should always be separated by primary namespace first, before secondary namespaces are used.
+/// While the number of primary namespaces will be relatively small and determined at compile time,
+/// there may be many secondary namespaces per primary namespace. Note that per-namespace uniqueness
+/// needs to also hold for keys *and* namespaces in any given namespace, i.e., conflicts between keys
+/// and equally named primary or secondary namespaces must be avoided.
+///
+/// **Note:** This trait extends the functionality of [`KVStoreSync`] by adding support for
+/// paginated listing of keys based on a monotonic counter or logical timestamp. This is useful
+/// when dealing with a large number of keys that cannot be efficiently retrieved all at once.
+///
+/// For an asynchronous version of this trait, see [`PaginatedKVStore`].
+pub trait PaginatedKVStoreSync: KVStoreSync {
+	/// Persists the given data under the given `key`.
+	///
+	/// If the key does not exist, it will be created with the given `order`. If the key already
+	/// exists, the data will be updated but the original `order` will be preserved. This ensures
+	/// consistent pagination even when entries are updated.
+	///
+	/// The `order` parameter is an `i64` used to track the order of keys for list operations,
+	/// allowing results to be returned in descending order. It is recommended to use a timestamp
+	/// (e.g., UNIX timestamp in seconds or milliseconds) or a monotonic counter. Since `order` is
+	/// immutable after initial creation, pagination remains consistent even when entries are
+	/// updated.
+	///
+	/// Will create the given `primary_namespace` and `secondary_namespace` if not already present
+	/// in the store.
+	fn write_ordered(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, order: i64,
+		buf: Vec<u8>,
+	) -> Result<(), io::Error>;
+
+	/// Returns a paginated list of keys that are stored under the given `secondary_namespace` in
+	/// `primary_namespace`, ordered in descending order of `order`.
+	///
+	/// The `list_paginated` method returns the latest records first, based on the `order`
+	/// associated with each key. Pagination is controlled by the `page_token`, which is used to
+	/// determine the starting point for the next page of results. If `page_token` is `None`, the
+	/// listing starts from the most recent entry. The `next_page_token` in the returned
+	/// [`PaginatedListResponse`] can be used to fetch the next page of results.
+	///
+	/// Returns an empty list if `primary_namespace` or `secondary_namespace` is unknown or if
+	/// there are no more keys to return.
+	fn list_paginated(
+		&self, primary_namespace: &str, secondary_namespace: &str, page_token: Option<PageToken>,
+	) -> Result<PaginatedListResponse, io::Error>;
+}
+
+/// A wrapper around a [`PaginatedKVStoreSync`] that implements the [`PaginatedKVStore`] trait.
+/// It is not necessary to use this type directly.
+#[derive(Clone)]
+pub struct PaginatedKVStoreSyncWrapper<K: Deref>(pub K)
+where
+	K::Target: PaginatedKVStoreSync;
+
+impl<K: Deref> Deref for PaginatedKVStoreSyncWrapper<K>
+where
+	K::Target: PaginatedKVStoreSync,
+{
+	type Target = Self;
+	fn deref(&self) -> &Self::Target {
+		self
+	}
+}
+
+/// This is not exported to bindings users as async is only supported in Rust.
+impl<K: Deref> KVStore for PaginatedKVStoreSyncWrapper<K>
+where
+	K::Target: PaginatedKVStoreSync,
+{
+	fn read(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
+	) -> impl Future<Output = Result<Vec<u8>, io::Error>> + 'static + MaybeSend {
+		let res = self.0.read(primary_namespace, secondary_namespace, key);
+
+		async move { res }
+	}
+
+	fn write(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>,
+	) -> impl Future<Output = Result<(), io::Error>> + 'static + MaybeSend {
+		let res = self.0.write(primary_namespace, secondary_namespace, key, buf);
+
+		async move { res }
+	}
+
+	fn remove(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
+	) -> impl Future<Output = Result<(), io::Error>> + 'static + MaybeSend {
+		let res = self.0.remove(primary_namespace, secondary_namespace, key, lazy);
+
+		async move { res }
+	}
+
+	fn list(
+		&self, primary_namespace: &str, secondary_namespace: &str,
+	) -> impl Future<Output = Result<Vec<String>, io::Error>> + 'static + MaybeSend {
+		let res = self.0.list(primary_namespace, secondary_namespace);
+
+		async move { res }
+	}
+}
+
+/// This is not exported to bindings users as async is only supported in Rust.
+impl<K: Deref> PaginatedKVStore for PaginatedKVStoreSyncWrapper<K>
+where
+	K::Target: PaginatedKVStoreSync,
+{
+	fn write_ordered(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, order: i64,
+		buf: Vec<u8>,
+	) -> impl Future<Output = Result<(), io::Error>> + 'static + MaybeSend {
+		let res = self.0.write_ordered(primary_namespace, secondary_namespace, key, order, buf);
+
+		async move { res }
+	}
+
+	fn list_paginated(
+		&self, primary_namespace: &str, secondary_namespace: &str, page_token: Option<PageToken>,
+	) -> impl Future<Output = Result<PaginatedListResponse, io::Error>> + 'static + MaybeSend {
+		let res = self.0.list_paginated(primary_namespace, secondary_namespace, page_token);
+
+		async move { res }
+	}
+}
+
+/// Provides an interface that allows storage and retrieval of persisted values that are associated
+/// with given keys, with support for pagination.
+///
+/// In order to avoid collisions, the key space is segmented based on the given `primary_namespace`s
+/// and `secondary_namespace`s. Implementations of this trait are free to handle them in different
+/// ways, as long as per-namespace key uniqueness is asserted.
+///
+/// Keys and namespaces are required to be valid ASCII strings in the range of
+/// [`KVSTORE_NAMESPACE_KEY_ALPHABET`] and no longer than [`KVSTORE_NAMESPACE_KEY_MAX_LEN`]. Empty
+/// primary namespaces and secondary namespaces (`""`) are considered valid; however, if
+/// `primary_namespace` is empty, `secondary_namespace` must also be empty. This means that concerns
+/// should always be separated by primary namespace first, before secondary namespaces are used.
+/// While the number of primary namespaces will be relatively small and determined at compile time,
+/// there may be many secondary namespaces per primary namespace. Note that per-namespace uniqueness
+/// needs to also hold for keys *and* namespaces in any given namespace, i.e., conflicts between keys
+/// and equally named primary or secondary namespaces must be avoided.
+///
+/// **Note:** This trait extends the functionality of [`KVStore`] by adding support for
+/// paginated listing of keys based on a monotonic counter or logical timestamp. This is useful
+/// when dealing with a large number of keys that cannot be efficiently retrieved all at once.
+///
+/// For a synchronous version of this trait, see [`PaginatedKVStoreSync`].
+///
+/// This is not exported to bindings users as async is only supported in Rust.
+pub trait PaginatedKVStore: KVStore {
+	/// Persists the given data under the given `key`.
+	///
+	/// If the key does not exist, it will be created with the given `order`. If the key already
+	/// exists, the data will be updated but the original `order` will be preserved. This ensures
+	/// consistent pagination even when entries are updated.
+	///
+	/// The `order` parameter is an `i64` used to track the order of keys for list operations,
+	/// allowing results to be returned in descending order. It is recommended to use a timestamp
+	/// (e.g., UNIX timestamp in seconds or milliseconds) or a monotonic counter. Since `order` is
+	/// immutable after initial creation, pagination remains consistent even when entries are
+	/// updated.
+	///
+	/// Will create the given `primary_namespace` and `secondary_namespace` if not already present
+	/// in the store.
+	fn write_ordered(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, order: i64,
+		buf: Vec<u8>,
+	) -> impl Future<Output = Result<(), io::Error>> + 'static + MaybeSend;
+
+	/// Returns a paginated list of keys that are stored under the given `secondary_namespace` in
+	/// `primary_namespace`, ordered in descending order of `order`.
+	///
+	/// The `list_paginated` method returns the latest records first, based on the `order`
+	/// associated with each key. Pagination is controlled by the `page_token`, which is used to
+	/// determine the starting point for the next page of results. If `page_token` is `None`, the
+	/// listing starts from the most recent entry. The `next_page_token` in the returned
+	/// [`PaginatedListResponse`] can be used to fetch the next page of results.
+	///
+	/// Returns an empty list if `primary_namespace` or `secondary_namespace` is unknown or if
+	/// there are no more keys to return.
+	fn list_paginated(
+		&self, primary_namespace: &str, secondary_namespace: &str, page_token: Option<PageToken>,
+	) -> impl Future<Output = Result<PaginatedListResponse, io::Error>> + 'static + MaybeSend;
+}
+
 /// Provides additional interface methods that are required for [`KVStore`]-to-[`KVStore`]
 /// data migration.
 pub trait MigratableKVStore: KVStoreSync {
@@ -1565,7 +1784,7 @@ mod tests {
 	use crate::ln::msgs::BaseMessageHandler;
 	use crate::sync::Arc;
 	use crate::util::test_channel_signer::TestChannelSigner;
-	use crate::util::test_utils::{self, TestStore};
+	use crate::util::test_utils::{self, TestPaginatedStore, TestStore};
 	use bitcoin::hashes::hex::FromHex;
 	use core::cmp;
 
@@ -1975,4 +2194,78 @@ mod tests {
 		let store: Arc<dyn KVStoreSync + Send + Sync> = Arc::new(TestStore::new(false));
 		assert!(persist_fn::<_, TestChannelSigner>(Arc::clone(&store)));
 	}
+
+	#[test]
+	fn paginated_store_basic_operations() {
+		let store = TestPaginatedStore::new(10);
+
+		// Write some data
+		store.write_ordered("ns1", "ns2", "key1", 100, vec![1, 2, 3]).unwrap();
+		store.write_ordered("ns1", "ns2", "key2", 200, vec![4, 5, 6]).unwrap();
+
+		// Read it back
+		assert_eq!(KVStoreSync::read(&store, "ns1", "ns2", "key1").unwrap(), vec![1, 2, 3]);
+		assert_eq!(KVStoreSync::read(&store, "ns1", "ns2", "key2").unwrap(), vec![4, 5, 6]);
+
+		// List should return keys in descending order
+		let response = store.list_paginated("ns1", "ns2", None).unwrap();
+		assert_eq!(response.keys, vec!["key2", "key1"]);
+		assert!(response.next_page_token.is_none());
+
+		// Remove a key
+		KVStoreSync::remove(&store, "ns1", "ns2", "key1", false).unwrap();
+		assert!(KVStoreSync::read(&store, "ns1", "ns2", "key1").is_err());
+	}
+
+	#[test]
+	fn paginated_store_pagination() {
+		let store = TestPaginatedStore::new(2);
+
+		// Write 5 items with different order values
+		for i in 0..5i64 {
+			store.write_ordered("ns", "", &format!("key{i}"), i, vec![i as u8]).unwrap();
+		}
+
+		// First page should have 2 items (highest order first: key4, key3)
+		let page1 = store.list_paginated("ns", "", None).unwrap();
+		assert_eq!(page1.keys.len(), 2);
+		assert_eq!(page1.keys, vec!["key4", "key3"]);
+		assert!(page1.next_page_token.is_some());
+
+		// Second page
+		let page2 = store.list_paginated("ns", "", page1.next_page_token).unwrap();
+		assert_eq!(page2.keys.len(), 2);
+		assert_eq!(page2.keys, vec!["key2", "key1"]);
+		assert!(page2.next_page_token.is_some());
+
+		// Third page (last item)
+		let page3 = store.list_paginated("ns", "", page2.next_page_token).unwrap();
+		assert_eq!(page3.keys.len(), 1);
+		assert_eq!(page3.keys, vec!["key0"]);
+		assert!(page3.next_page_token.is_none());
+	}
+
+	#[test]
+	fn paginated_store_update_preserves_order() {
+		let store = TestPaginatedStore::new(10);
+
+		// Write items with specific order values
+		store.write_ordered("ns", "", "key1", 100, vec![1]).unwrap();
+		store.write_ordered("ns", "", "key2", 200, vec![2]).unwrap();
+		store.write_ordered("ns", "", "key3", 300, vec![3]).unwrap();
+
+		// Verify initial order (descending by order)
+		let response = store.list_paginated("ns", "", None).unwrap();
+		assert_eq!(response.keys, vec!["key3", "key2", "key1"]);
+
+		// Update key1 with a new order value that would put it first if used
+		store.write_ordered("ns", "", "key1", 999, vec![1, 1]).unwrap();
+
+		// Verify data was updated
+		assert_eq!(KVStoreSync::read(&store, "ns", "", "key1").unwrap(), vec![1, 1]);
+
+		// Verify order is unchanged - original order should have been preserved
+		let response = store.list_paginated("ns", "", None).unwrap();
+		assert_eq!(response.keys, vec!["key3", "key2", "key1"]);
+	}
 }