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.

AI-assisted-by: OpenAI Codex

Author: benthecarman

CHANGELOG.md

@@ -20,6 +20,8 @@
   succeeds, allowing restored wallets to rediscover funds sent to previously-unknown addresses.
 - `Config::anchor_channels_config` is no longer optional, hence anchor channels can no longer be
   disabled. We still negotiate legacy channels if the peer does not support anchor channels.
+- `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/kotlin/ldk-node-jvm/lib/src/test/kotlin/org/lightningdevkit/ldknode/LibraryTest.kt

@@ -301,8 +301,8 @@ class LibraryTest {
         assert(paymentReceivedEvent is Event.PaymentReceived)
         node2.eventHandled()
 
-        assert(node1.listPayments().size == 3)
-        assert(node2.listPayments().size == 2)
+        assert(node1.listPayments(null).payments.size == 3)
+        assert(node2.listPayments(null).payments.size == 2)
 
         node2.closeChannel(userChannelId, nodeId1)
 

bindings/ldk_node.udl

@@ -151,7 +151,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();
@@ -277,6 +278,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};
@@ -44,7 +44,7 @@ pub(crate) struct DataStore<SO: StorableObject, L: Deref>
 where
 	L::Target: LdkLogger,
 {
-	objects: Mutex<HashMap<SO::Id, SO>>,
+	objects: Mutex<HashMap<String, SO>>,
 	mutation_lock: tokio::sync::Mutex<()>,
 	primary_namespace: String,
 	secondary_namespace: String,
@@ -60,8 +60,9 @@ where
 		objects: Vec<SO>, primary_namespace: String, secondary_namespace: String,
 		kv_store: Arc<DynStore>, logger: L,
 	) -> Self {
-		let objects =
-			Mutex::new(HashMap::from_iter(objects.into_iter().map(|obj| (obj.id(), obj))));
+		let objects = Mutex::new(HashMap::from_iter(
+			objects.into_iter().map(|obj| (obj.id().encode_to_hex_str(), obj)),
+		));
 		Self {
 			objects,
 			mutation_lock: tokio::sync::Mutex::new(()),
@@ -74,20 +75,22 @@ where
 
 	pub(crate) async fn insert(&self, object: SO) -> Result<bool, Error> {
 		let _guard = self.mutation_lock.lock().await;
+		let store_key = object.id().encode_to_hex_str();
 
 		self.persist(&object).await?;
 		let mut locked_objects = self.objects.lock().expect("lock");
-		let updated = locked_objects.insert(object.id(), object).is_some();
+		let updated = locked_objects.insert(store_key, object).is_some();
 		Ok(updated)
 	}
 
 	pub(crate) async fn insert_or_update(&self, object: SO) -> Result<bool, Error> {
 		let _guard = self.mutation_lock.lock().await;
 
 		let id = object.id();
+		let store_key = id.encode_to_hex_str();
 		let data_to_persist = {
 			let locked_objects = self.objects.lock().expect("lock");
-			if let Some(existing_object) = locked_objects.get(&id) {
+			if let Some(existing_object) = locked_objects.get(&store_key) {
 				let mut updated_object = existing_object.clone();
 				let updated = updated_object.update(object.to_update());
 				if updated {
@@ -104,7 +107,7 @@ where
 			Some(updated_object) => {
 				self.persist(&updated_object).await?;
 				let mut locked_objects = self.objects.lock().expect("lock");
-				locked_objects.insert(id, updated_object);
+				locked_objects.insert(store_key, updated_object);
 				Ok(true)
 			},
 			None => Ok(false),
@@ -113,9 +116,9 @@ where
 
 	pub(crate) async fn remove(&self, id: &SO::Id) -> Result<(), Error> {
 		let _guard = self.mutation_lock.lock().await;
-		let should_remove = { self.objects.lock().expect("lock").contains_key(id) };
+		let store_key = id.encode_to_hex_str();
+		let should_remove = { self.objects.lock().expect("lock").contains_key(&store_key) };
 		if should_remove {
-			let store_key = id.encode_to_hex_str();
 			KVStore::remove(
 				&*self.kv_store,
 				&self.primary_namespace,
@@ -135,7 +138,7 @@ where
 				);
 				Error::PersistenceFailed
 			})?;
-			self.objects.lock().expect("lock").remove(id);
+			self.objects.lock().expect("lock").remove(&store_key);
 		}
 		Ok(())
 	}
@@ -146,15 +149,16 @@ where
 	/// Until store reads are async, callers may temporarily see in-memory state that has not yet
 	/// caught up to a write in progress.
 	pub(crate) fn get(&self, id: &SO::Id) -> Option<SO> {
-		self.objects.lock().expect("lock").get(id).cloned()
+		self.objects.lock().expect("lock").get(&id.encode_to_hex_str()).cloned()
 	}
 
 	pub(crate) async fn update(&self, update: SO::Update) -> Result<DataStoreUpdateResult, Error> {
 		let _guard = self.mutation_lock.lock().await;
 		let id = update.id();
+		let store_key = id.encode_to_hex_str();
 		let updated_object = {
 			let locked_objects = self.objects.lock().expect("lock");
-			let Some(object) = locked_objects.get(&id) else {
+			let Some(object) = locked_objects.get(&store_key) else {
 				return Ok(DataStoreUpdateResult::NotFound);
 			};
 			let mut updated_object = object.clone();
@@ -166,7 +170,7 @@ where
 
 		self.persist(&updated_object).await?;
 		let mut locked_objects = self.objects.lock().expect("lock");
-		locked_objects.insert(id, updated_object);
+		locked_objects.insert(store_key, updated_object);
 		Ok(DataStoreUpdateResult::Updated)
 	}
 
@@ -179,6 +183,40 @@ 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 _guard = self.mutation_lock.lock().await;
+		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 =
+			response.keys.iter().filter_map(|key| locked_objects.get(key).cloned()).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
@@ -217,7 +255,7 @@ where
 	/// Until store reads are async, callers may temporarily see in-memory state that has not yet
 	/// caught up to a write in progress.
 	pub(crate) fn contains_key(&self, id: &SO::Id) -> bool {
-		self.objects.lock().expect("lock").contains_key(id)
+		self.objects.lock().expect("lock").contains_key(&id.encode_to_hex_str())
 	}
 }
 
@@ -337,6 +375,74 @@ 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 list_page_waits_for_mutations() {
+		let store: Arc<DynStore> = Arc::new(DynStoreWrapper(InMemoryStore::new()));
+		let logger = Arc::new(TestLogger::new());
+		let data_store: Arc<DataStore<TestObject, Arc<TestLogger>>> = Arc::new(DataStore::new(
+			Vec::new(),
+			"datastore_test_primary".to_string(),
+			"datastore_test_secondary".to_string(),
+			store,
+			logger,
+		));
+
+		let mutation_guard = data_store.mutation_lock.lock().await;
+		let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+		let task_store = Arc::clone(&data_store);
+		let list_task = tokio::spawn(async move {
+			started_tx.send(()).unwrap();
+			task_store.list_page(None).await
+		});
+
+		started_rx.await.unwrap();
+		tokio::task::yield_now().await;
+		assert!(!list_task.is_finished());
+
+		drop(mutation_guard);
+		let (page, next_page_token) = list_task.await.unwrap().unwrap();
+		assert!(page.is_empty());
+		assert!(next_page_token.is_none());
+	}
+
 	#[tokio::test]
 	async fn data_is_persisted() {
 		let store: Arc<DynStore> = Arc::new(DynStoreWrapper(InMemoryStore::new()));

src/lib.rs

@@ -158,6 +158,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;
@@ -2207,15 +2208,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.
@@ -2333,6 +2363,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))]