Add paginated payment listing API

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))]