Add paginated payment listing API

Replace the full payment listing API with paginated listing so callers
can migrate away from fetching every payment at once. Deprecate the
filtering helper because it still implies scanning the full in-memory
store.

Author: benthecarman

CHANGELOG.md

@@ -16,6 +16,8 @@
 - `EsploraSyncConfig` and `ElectrumSyncConfig` now support `force_wallet_full_scan`. When set,
   the on-chain wallet keeps using BDK `full_scan` instead of incremental sync until a full scan
   succeeds, allowing restored wallets to rediscover funds sent to previously-unknown addresses.
+- `Node::list_payments` now retrieves payments page-by-page, ordered from most recently created to
+  least recently created, instead of returning all payments at once.
 
 ## Bug Fixes and Improvements
 - Building a fresh node against a Bitcoin Core RPC or REST chain source that fails to return the

bindings/ldk_node.udl

@@ -136,7 +136,8 @@ interface Node {
 	[Throws=NodeError]
 	void remove_payment([ByRef]PaymentId payment_id);
 	BalanceDetails list_balances();
-	sequence<PaymentDetails> list_payments();
+	[Throws=NodeError]
+	PaymentDetailsPage list_payments(PageToken? page_token);
 	sequence<PeerDetails> list_peers();
 	sequence<ChannelDetails> list_channels();
 	NetworkGraph network_graph();
@@ -261,6 +262,11 @@ enum PaymentFailureReason {
 
 typedef dictionary PaymentDetails;
 
+typedef dictionary PaymentDetailsPage;
+
+[Remote]
+interface PageToken {};
+
 [Remote]
 dictionary RouteParametersConfig {
 	u64? max_total_routing_fee_msat;

src/data_store.rs

@@ -9,7 +9,7 @@ use std::collections::HashMap;
 use std::ops::Deref;
 use std::sync::{Arc, Mutex};
 
-use lightning::util::persist::KVStore;
+use lightning::util::persist::{KVStore, PageToken, PaginatedKVStore};
 use lightning::util::ser::{Readable, Writeable};
 
 use crate::logger::{log_error, LdkLogger};
@@ -179,6 +179,47 @@ where
 		self.objects.lock().expect("lock").values().filter(f).cloned().collect::<Vec<SO>>()
 	}
 
+	/// Returns a page of objects, ordered from most recently created to least recently created,
+	/// together with a token that can be passed to a subsequent call to retrieve the next page.
+	///
+	/// The underlying store is only queried for the page's key order, which the in-memory map
+	/// doesn't track; the objects themselves are served from memory.
+	pub(crate) async fn list_page(
+		&self, page_token: Option<PageToken>,
+	) -> Result<(Vec<SO>, Option<PageToken>), Error> {
+		let response = PaginatedKVStore::list_paginated(
+			&*self.kv_store,
+			&self.primary_namespace,
+			&self.secondary_namespace,
+			page_token,
+		)
+		.await
+		.map_err(|e| {
+			log_error!(
+				self.logger,
+				"Listing object data under {}/{} failed due to: {}",
+				&self.primary_namespace,
+				&self.secondary_namespace,
+				e
+			);
+			Error::PersistenceFailed
+		})?;
+
+		let locked_objects = self.objects.lock().expect("lock");
+		let objects_by_store_key: HashMap<String, &SO> =
+			locked_objects.values().map(|obj| (obj.id().encode_to_hex_str(), obj)).collect();
+
+		// Objects removed between listing the page's keys and this lookup are skipped rather
+		// than failing the page.
+		let objects = response
+			.keys
+			.iter()
+			.filter_map(|key| objects_by_store_key.get(key).map(|obj| (*obj).clone()))
+			.collect();
+
+		Ok((objects, response.next_page_token))
+	}
+
 	async fn persist(&self, object: &SO) -> Result<(), Error> {
 		let (store_key, data) = Self::encode_object(object);
 		self.persist_encoded(store_key, data).await
@@ -338,6 +379,44 @@ mod tests {
 		)
 	}
 
+	#[tokio::test]
+	async fn list_page_paginates_in_reverse_creation_order() {
+		let store: Arc<DynStore> = Arc::new(DynStoreWrapper(InMemoryStore::new()));
+		let logger = Arc::new(TestLogger::new());
+		let data_store: DataStore<TestObject, Arc<TestLogger>> = DataStore::new(
+			Vec::new(),
+			"datastore_test_primary".to_string(),
+			"datastore_test_secondary".to_string(),
+			Arc::clone(&store),
+			logger,
+		);
+
+		// Insert more objects than fit in a single page to exercise the pagination loop.
+		let num_objects = 120u32;
+		for i in 0..num_objects {
+			let id = TestObjectId { id: i.to_be_bytes() };
+			data_store.insert(TestObject { id, data: [7u8; 3] }).await.unwrap();
+		}
+
+		let mut listed = Vec::with_capacity(num_objects as usize);
+		let mut page_token = None;
+		loop {
+			let (page, next_page_token) = data_store.list_page(page_token).await.unwrap();
+			assert!(!page.is_empty());
+			listed.extend(page);
+			page_token = next_page_token;
+			if page_token.is_none() {
+				break;
+			}
+		}
+
+		let expected: Vec<TestObject> = (0..num_objects)
+			.rev()
+			.map(|i| TestObject { id: TestObjectId { id: i.to_be_bytes() }, data: [7u8; 3] })
+			.collect();
+		assert_eq!(listed, expected);
+	}
+
 	#[tokio::test]
 	async fn data_is_persisted() {
 		let store: Arc<DynStore> = Arc::new(DynStoreWrapper(InMemoryStore::new()));

src/lib.rs

@@ -156,6 +156,7 @@ use lightning::ln::peer_handler::CustomMessageHandler;
 use lightning::routing::gossip::NodeAlias;
 use lightning::sign::EntropySource;
 use lightning::util::persist::KVStore;
+pub use lightning::util::persist::PageToken;
 use lightning::util::wallet_utils::{Input, Wallet as LdkWallet};
 use lightning_background_processor::process_events_async;
 pub use lightning_invoice;
@@ -2124,15 +2125,44 @@ impl Node {
 	/// # let node = builder.build(node_entropy.into()).unwrap();
 	/// node.list_payments_with_filter(|p| p.direction == PaymentDirection::Outbound);
 	/// ```
+	#[deprecated(
+		note = "Use the paginated list_payments API and filter the returned pages instead."
+	)]
 	pub fn list_payments_with_filter<F: FnMut(&&PaymentDetails) -> bool>(
 		&self, f: F,
 	) -> Vec<PaymentDetails> {
 		self.payment_store.list_filter(f)
 	}
 
-	/// Retrieves all payments.
-	pub fn list_payments(&self) -> Vec<PaymentDetails> {
-		self.payment_store.list_filter(|_| true)
+	/// Retrieves a page of payments from the underlying paginated store, ordered from most
+	/// recently created to least recently created.
+	///
+	/// Pass `None` to start listing from the most recently created payment. If the returned
+	/// [`PaymentDetailsPage::next_page_token`] is `Some`, pass it to a subsequent call to
+	/// retrieve the next page.
+	#[cfg(not(feature = "uniffi"))]
+	pub fn list_payments(
+		&self, page_token: Option<PageToken>,
+	) -> Result<PaymentDetailsPage, Error> {
+		let (payments, next_page_token) =
+			self.runtime.block_on(self.payment_store.list_page(page_token))?;
+		Ok(PaymentDetailsPage { payments, next_page_token })
+	}
+
+	/// Retrieves a page of payments from the underlying paginated store, ordered from most
+	/// recently created to least recently created.
+	///
+	/// Pass `None` to start listing from the most recently created payment. If the returned
+	/// [`PaymentDetailsPage::next_page_token`] is `Some`, pass it to a subsequent call to
+	/// retrieve the next page.
+	#[cfg(feature = "uniffi")]
+	pub fn list_payments(
+		&self, page_token: Option<Arc<PageToken>>,
+	) -> Result<PaymentDetailsPage, Error> {
+		let page_token = page_token.map(|t| (*t).clone());
+		let (payments, next_page_token) =
+			self.runtime.block_on(self.payment_store.list_page(page_token))?;
+		Ok(PaymentDetailsPage { payments, next_page_token: next_page_token.map(Arc::new) })
 	}
 
 	/// Retrieves a list of known peers.
@@ -2250,6 +2280,20 @@ impl Drop for Node {
 	}
 }
 
+/// A page of payments returned from a paginated listing.
+#[derive(Clone, Debug, PartialEq, Eq)]
+#[cfg_attr(feature = "uniffi", derive(uniffi::Record))]
+pub struct PaymentDetailsPage {
+	/// Payments in this page, ordered from most recently created to least recently created.
+	pub payments: Vec<PaymentDetails>,
+	/// Token to pass to the next call to continue listing, if another page exists.
+	#[cfg(not(feature = "uniffi"))]
+	pub next_page_token: Option<PageToken>,
+	/// Token to pass to the next call to continue listing, if another page exists.
+	#[cfg(feature = "uniffi")]
+	pub next_page_token: Option<Arc<PageToken>>,
+}
+
 /// The best known block as identified by its hash and height.
 #[derive(Clone, Copy, Debug, Hash, PartialEq, Eq)]
 #[cfg_attr(feature = "uniffi", derive(uniffi::Record))]