Add paginated payment listing API

CHANGELOG.md

@@ -20,6 +20,10 @@
   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. This is a breaking API change,
+  and `Node::list_payments_with_filter` is now deprecated. Generic KV store migrations do not
+  preserve creation-order metadata and may change the order of existing payments (#959).
 
 ## 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,9 @@ 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)
+        assert(PageToken("1").toString() == "1")
 
         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,14 @@ enum PaymentFailureReason {
 
 typedef dictionary PaymentDetails;
 
+typedef dictionary PaymentDetailsPage;
+
+[Remote]
+interface PageToken {
+	constructor(string token);
+	string to_string();
+};
+
 [Remote]
 dictionary RouteParametersConfig {
 	u64? max_total_routing_fee_msat;

src/data_store.rs

@@ -5,11 +5,11 @@
 // http://opensource.org/licenses/MIT>, at your option. You may not use this file except in
 // accordance with one or both of these licenses.
 
-use std::collections::HashMap;
+use std::collections::{HashMap, VecDeque};
 use std::ops::Deref;
 use std::sync::{Arc, Mutex};
 
-use lightning::util::persist::KVStore;
+use lightning::util::persist::{KVStore, PageToken};
 use lightning::util::ser::{Readable, Writeable};
 
 use crate::logger::{log_error, LdkLogger};
@@ -25,7 +25,7 @@ pub(crate) trait StorableObject: Clone + Readable + Writeable {
 	fn to_update(&self) -> Self::Update;
 }
 
-pub(crate) trait StorableObjectId: std::hash::Hash + PartialEq + Eq {
+pub(crate) trait StorableObjectId: Clone + std::hash::Hash + Eq {
 	fn encode_to_hex_str(&self) -> String;
 }
 
@@ -40,11 +40,17 @@ pub(crate) enum DataStoreUpdateResult {
 	NotFound,
 }
 
+struct InMemoryObjects<SO: StorableObject> {
+	objects: HashMap<SO::Id, SO>,
+	creation_order: VecDeque<(u64, SO::Id)>,
+	next_creation_order: u64,
+}
+
 pub(crate) struct DataStore<SO: StorableObject, L: Deref>
 where
 	L::Target: LdkLogger,
 {
-	objects: Mutex<HashMap<SO::Id, SO>>,
+	objects: Mutex<InMemoryObjects<SO>>,
 	mutation_lock: tokio::sync::Mutex<()>,
 	primary_namespace: String,
 	secondary_namespace: String,
@@ -60,8 +66,19 @@ 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 next_creation_order = objects.len() as u64;
+		let mut creation_order = VecDeque::with_capacity(objects.len());
+		let mut objects_by_id = HashMap::with_capacity(objects.len());
+		for (index, object) in objects.into_iter().enumerate() {
+			let id = object.id();
+			creation_order.push_back((next_creation_order - index as u64, id.clone()));
+			objects_by_id.insert(id, object);
+		}
+		let objects = Mutex::new(InMemoryObjects {
+			objects: objects_by_id,
+			creation_order,
+			next_creation_order,
+		});
 		Self {
 			objects,
 			mutation_lock: tokio::sync::Mutex::new(()),
@@ -77,7 +94,13 @@ where
 
 		self.persist(&object).await?;
 		let mut locked_objects = self.objects.lock().expect("lock");
-		let updated = locked_objects.insert(object.id(), object).is_some();
+		let id = object.id();
+		let updated = locked_objects.objects.insert(id.clone(), object).is_some();
+		if !updated {
+			locked_objects.next_creation_order += 1;
+			let creation_order = locked_objects.next_creation_order;
+			locked_objects.creation_order.push_front((creation_order, id));
+		}
 		Ok(updated)
 	}
 
@@ -87,7 +110,7 @@ where
 		let id = object.id();
 		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.objects.get(&id) {
 				let mut updated_object = existing_object.clone();
 				let updated = updated_object.update(object.to_update());
 				if updated {
@@ -104,7 +127,12 @@ where
 			Some(updated_object) => {
 				self.persist(&updated_object).await?;
 				let mut locked_objects = self.objects.lock().expect("lock");
-				locked_objects.insert(id, updated_object);
+				let is_new = locked_objects.objects.insert(id.clone(), updated_object).is_none();
+				if is_new {
+					locked_objects.next_creation_order += 1;
+					let creation_order = locked_objects.next_creation_order;
+					locked_objects.creation_order.push_front((creation_order, id));
+				}
 				Ok(true)
 			},
 			None => Ok(false),
@@ -113,7 +141,7 @@ 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 should_remove = { self.objects.lock().expect("lock").objects.contains_key(id) };
 		if should_remove {
 			let store_key = id.encode_to_hex_str();
 			KVStore::remove(
@@ -135,7 +163,9 @@ where
 				);
 				Error::PersistenceFailed
 			})?;
-			self.objects.lock().expect("lock").remove(id);
+			let mut locked_objects = self.objects.lock().expect("lock");
+			locked_objects.objects.remove(id);
+			locked_objects.creation_order.retain(|(_, object_id)| object_id != id);
 		}
 		Ok(())
 	}
@@ -146,15 +176,15 @@ 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").objects.get(id).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 updated_object = {
 			let locked_objects = self.objects.lock().expect("lock");
-			let Some(object) = locked_objects.get(&id) else {
+			let Some(object) = locked_objects.objects.get(&id) else {
 				return Ok(DataStoreUpdateResult::NotFound);
 			};
 			let mut updated_object = object.clone();
@@ -166,7 +196,7 @@ where
 
 		self.persist(&updated_object).await?;
 		let mut locked_objects = self.objects.lock().expect("lock");
-		locked_objects.insert(id, updated_object);
+		locked_objects.objects.insert(id, updated_object);
 		Ok(DataStoreUpdateResult::Updated)
 	}
 
@@ -176,7 +206,55 @@ 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 list_filter<F: FnMut(&&SO) -> bool>(&self, f: F) -> Vec<SO> {
-		self.objects.lock().expect("lock").values().filter(f).cloned().collect::<Vec<SO>>()
+		self.objects.lock().expect("lock").objects.values().filter(f).cloned().collect()
+	}
+
+	/// 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.
+	pub(crate) fn list_page(
+		&self, page_token: Option<PageToken>,
+	) -> Result<(Vec<SO>, Option<PageToken>), Error> {
+		const PAGE_SIZE: usize = 50;
+
+		let locked_objects = self.objects.lock().expect("lock");
+		let token_creation_order = page_token
+			.map(|token| {
+				token.as_str().parse::<u64>().map_err(|_| {
+					log_error!(self.logger, "Invalid object page token: {}", token);
+					Error::PersistenceFailed
+				})
+			})
+			.transpose()?;
+		if let Some(order) = token_creation_order {
+			if order > locked_objects.next_creation_order {
+				log_error!(
+					self.logger,
+					"Invalid object page token exceeding latest creation order: {}",
+					order
+				);
+				return Err(Error::PersistenceFailed);
+			}
+		}
+
+		let mut entries = locked_objects
+			.creation_order
+			.iter()
+			.filter(|(order, _)| token_creation_order.is_none_or(|token| *order < token))
+			.filter_map(|(order, id)| {
+				locked_objects.objects.get(id).cloned().map(|object| (*order, object))
+			})
+			.take(PAGE_SIZE + 1)
+			.collect::<Vec<_>>();
+		let has_more = entries.len() > PAGE_SIZE;
+		entries.truncate(PAGE_SIZE);
+
+		let next_page_token = if has_more {
+			entries.last().map(|(order, _)| PageToken::new(order.to_string()))
+		} else {
+			None
+		};
+		let objects = entries.into_iter().map(|(_, object)| object).collect();
+		Ok((objects, next_page_token))
 	}
 
 	async fn persist(&self, object: &SO) -> Result<(), Error> {
@@ -217,7 +295,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").objects.contains_key(id)
 	}
 }
 
@@ -337,6 +415,55 @@ 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).unwrap();
+			assert!(!page.is_empty());
+			listed.extend(page);
+			page_token = next_page_token.map(|token| PageToken::new(token.to_string()));
+			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);
+	}
+
+	#[test]
+	fn list_page_only_reads_in_memory() {
+		let newest = TestObject { id: TestObjectId { id: 2u32.to_be_bytes() }, data: [2u8; 3] };
+		let oldest = TestObject { id: TestObjectId { id: 1u32.to_be_bytes() }, data: [1u8; 3] };
+		let data_store = new_failing_data_store(vec![newest, oldest]);
+
+		let (page, next_page_token) = data_store.list_page(None).unwrap();
+		assert_eq!(page, vec![newest, oldest]);
+		assert!(next_page_token.is_none());
+	}
+
 	#[tokio::test]
 	async fn data_is_persisted() {
 		let store: Arc<DynStore> = Arc::new(DynStoreWrapper(InMemoryStore::new()));