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,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()));