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,11 @@
   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. Invalid pagination tokens return the new
+  `Error::InvalidPageToken` variant. 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();
@@ -235,6 +236,7 @@ enum NodeError {
 	"InvalidDateTime",
 	"InvalidFeeRate",
 	"InvalidScriptPubKey",
+	"InvalidPageToken",
 	"DuplicatePayment",
 	"UnsupportedCurrency",
 	"InsufficientFunds",
@@ -277,6 +279,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,16 @@ pub(crate) enum DataStoreUpdateResult {
 	NotFound,
 }
 
+struct InMemoryObjects<SO: StorableObject> {
+	objects: HashMap<SO::Id, SO>,
+	creation_order: VecDeque<SO::Id>,
+}
+
 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 +65,14 @@ 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 mut creation_order = VecDeque::with_capacity(objects.len());
+		let mut objects_by_id = HashMap::with_capacity(objects.len());
+		for object in objects {
+			let id = object.id();
+			creation_order.push_back(id.clone());
+			objects_by_id.insert(id, object);
+		}
+		let objects = Mutex::new(InMemoryObjects { objects: objects_by_id, creation_order });
 		Self {
 			objects,
 			mutation_lock: tokio::sync::Mutex::new(()),
@@ -77,7 +88,11 @@ 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.creation_order.push_front(id);
+		}
 		Ok(updated)
 	}
 
@@ -87,7 +102,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 +119,10 @@ 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.creation_order.push_front(id);
+				}
 				Ok(true)
 			},
 			None => Ok(false),
@@ -113,7 +131,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 +153,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 +166,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 +186,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 +196,48 @@ 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 start_index = if let Some(token) = page_token {
+			locked_objects
+				.creation_order
+				.iter()
+				.position(|id| id.encode_to_hex_str() == token.as_str())
+				.map(|index| index + 1)
+				.ok_or_else(|| {
+					log_error!(self.logger, "Object page token not found: {}", token);
+					Error::InvalidPageToken
+				})?
+		} else {
+			0
+		};
+
+		let mut entries = locked_objects
+			.creation_order
+			.iter()
+			.skip(start_index)
+			.filter_map(|id| locked_objects.objects.get(id).cloned().map(|object| (id, 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(|(id, _)| PageToken::new(id.encode_to_hex_str()))
+		} 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 +278,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)
 	}
 }
 
@@ -230,6 +291,7 @@ mod tests {
 	use super::*;
 	use crate::hex_utils;
 	use crate::io::test_utils::InMemoryStore;
+	use crate::io::utils::read_all_objects;
 	use crate::types::DynStoreWrapper;
 
 	#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
@@ -337,6 +399,116 @@ 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);
+	}
+
+	#[tokio::test]
+	async fn list_page_token_survives_reload_after_unseen_object_is_removed() {
+		let store: Arc<DynStore> = Arc::new(DynStoreWrapper(InMemoryStore::new()));
+		let logger = Arc::new(TestLogger::new());
+		let primary_namespace = "datastore_test_primary".to_string();
+		let secondary_namespace = "datastore_test_secondary".to_string();
+		let data_store: DataStore<TestObject, Arc<TestLogger>> = DataStore::new(
+			Vec::new(),
+			primary_namespace.clone(),
+			secondary_namespace.clone(),
+			Arc::clone(&store),
+			Arc::clone(&logger),
+		);
+
+		for i in 0..101u32 {
+			let id = TestObjectId { id: i.to_be_bytes() };
+			data_store.insert(TestObject { id, data: [7u8; 3] }).await.unwrap();
+		}
+
+		let (first_page, page_token) = data_store.list_page(None).unwrap();
+		assert_eq!(first_page.first().unwrap().id.id, 100u32.to_be_bytes());
+		assert_eq!(first_page.last().unwrap().id.id, 51u32.to_be_bytes());
+		let page_token = page_token.unwrap();
+		assert_eq!(page_token.as_str(), "00000033");
+
+		let oldest_id = TestObjectId { id: 0u32.to_be_bytes() };
+		data_store.remove(&oldest_id).await.unwrap();
+		let reloaded_objects = read_all_objects(
+			&*store,
+			&primary_namespace,
+			&secondary_namespace,
+			Arc::clone(&logger),
+		)
+		.await
+		.unwrap();
+		let reloaded_data_store: DataStore<TestObject, Arc<TestLogger>> =
+			DataStore::new(reloaded_objects, primary_namespace, secondary_namespace, store, logger);
+
+		let (second_page, next_page_token) =
+			reloaded_data_store.list_page(Some(page_token)).unwrap();
+		assert_eq!(second_page.first().unwrap().id.id, 50u32.to_be_bytes());
+		assert_eq!(second_page.last().unwrap().id.id, 1u32.to_be_bytes());
+		assert_eq!(second_page.len(), 50);
+		assert!(next_page_token.is_none());
+	}
+
+	#[test]
+	fn list_page_rejects_invalid_tokens() {
+		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 malformed_error =
+			data_store.list_page(Some(PageToken::new("3".to_string()))).unwrap_err();
+		assert_eq!(malformed_error, Error::InvalidPageToken);
+
+		let unknown_error =
+			data_store.list_page(Some(PageToken::new("ffffffff".to_string()))).unwrap_err();
+		assert_eq!(unknown_error, Error::InvalidPageToken);
+	}
+
+	#[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()));

src/error.rs

@@ -115,6 +115,8 @@ pub enum Error {
 	InvalidFeeRate,
 	/// The given script public key is invalid.
 	InvalidScriptPubKey,
+	/// The given page token is invalid.
+	InvalidPageToken,
 	/// A payment with the given hash has already been initiated.
 	DuplicatePayment,
 	/// The provided offer was denonminated in an unsupported currency.
@@ -199,6 +201,7 @@ impl fmt::Display for Error {
 			Self::InvalidDateTime => write!(f, "The given date time is invalid."),
 			Self::InvalidFeeRate => write!(f, "The given fee rate is invalid."),
 			Self::InvalidScriptPubKey => write!(f, "The given script pubkey is invalid."),
+			Self::InvalidPageToken => write!(f, "The given page token is invalid."),
 			Self::DuplicatePayment => {
 				write!(f, "A payment with the given hash has already been initiated.")
 			},