Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,8 @@
- `EsploraSyncConfig` and `ElectrumSyncConfig` now support `force_wallet_full_scan`. When set,
the on-chain wallet keeps using BDK `full_scan` instead of incremental sync until a full scan
succeeds, allowing restored wallets to rediscover funds sent to previously-unknown addresses.
- `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
Expand Down
8 changes: 7 additions & 1 deletion bindings/ldk_node.udl
Original file line number Diff line number Diff line change
Expand Up @@ -136,7 +136,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();
Expand Down Expand Up @@ -261,6 +262,11 @@ enum PaymentFailureReason {

typedef dictionary PaymentDetails;

typedef dictionary PaymentDetailsPage;

[Remote]
interface PageToken {};

[Remote]
dictionary RouteParametersConfig {
u64? max_total_routing_fee_msat;
Expand Down
81 changes: 80 additions & 1 deletion src/data_store.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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};
Expand Down Expand Up @@ -179,6 +179,47 @@ 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 response = PaginatedKVStore::list_paginated(

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Currently, we're still leaning on all DataStore entries being kept in memory everywhere else (which we should change soon as mentioned above). So right now it's a bit odd to have the paginated version being the only one calling through to KVStore, and not even using any cache (so it's likely pretty slow).

So, should this just use the in-memory entries? If not, we could consider already making the jump to not keep all entries in memory, but then we'd need some (presumably LRU?) caching logic for DataStore first, before we add pagination support/switch it to use pagination?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Changed to just use in memory entries. Changing to not keep everything in memory would be a separate PR and maybe should wait for 0.9 as we have a lot of in flight storage changes right now.

&*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_by_store_key: HashMap<String, &SO> =
locked_objects.values().map(|obj| (obj.id().encode_to_hex_str(), obj)).collect();

// Objects removed between listing the page's keys and this lookup are skipped rather
// than failing the page.
let objects = response
.keys
.iter()
.filter_map(|key| objects_by_store_key.get(key).map(|obj| (*obj).clone()))
.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
Expand Down Expand Up @@ -338,6 +379,44 @@ 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 data_is_persisted() {
let store: Arc<DynStore> = Arc::new(DynStoreWrapper(InMemoryStore::new()));
Expand Down
50 changes: 47 additions & 3 deletions src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -156,6 +156,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;
Expand Down Expand Up @@ -2124,15 +2125,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.
Expand Down Expand Up @@ -2250,6 +2280,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))]
Expand Down
Loading
Loading