From 11768502d8bfeb97e4488252de33348133ee1508 Mon Sep 17 00:00:00 2001 From: benthecarman Date: Fri, 21 Aug 2026 05:34:51 -0500 Subject: [PATCH 1/4] Bump LDK 0.3 dependencies Use the latest LDK 0.3 revision so forwarding events include stable HTLC identifiers. Update the payment-instructions pin to keep its LDK types compatible. This change was made with assistance from OpenAI Codex. --- Cargo.toml | 30 +++++++++++++++--------------- 1 file changed, 15 insertions(+), 15 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 0110c027b..ebd808c00 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -86,18 +86,18 @@ uniffi-default = [ #lightning-macros = { version = "0.2.0" } #lightning-dns-resolver = { version = "0.3.0" } -lightning = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "9174965af9437196c527a9aa0df36bbcf050c8bb", features = ["std"] } -lightning-types = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "9174965af9437196c527a9aa0df36bbcf050c8bb" } -lightning-invoice = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "9174965af9437196c527a9aa0df36bbcf050c8bb", features = ["std"] } -lightning-net-tokio = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "9174965af9437196c527a9aa0df36bbcf050c8bb" } -lightning-persister = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "9174965af9437196c527a9aa0df36bbcf050c8bb", features = ["tokio"], optional = true } -lightning-background-processor = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "9174965af9437196c527a9aa0df36bbcf050c8bb" } -lightning-rapid-gossip-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "9174965af9437196c527a9aa0df36bbcf050c8bb" } -lightning-block-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "9174965af9437196c527a9aa0df36bbcf050c8bb", features = ["rest-client", "rpc-client", "tokio"], optional = true } -lightning-transaction-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "9174965af9437196c527a9aa0df36bbcf050c8bb", optional = true } -lightning-liquidity = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "9174965af9437196c527a9aa0df36bbcf050c8bb", features = ["std"] } -lightning-macros = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "9174965af9437196c527a9aa0df36bbcf050c8bb" } -lightning-dns-resolver = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "9174965af9437196c527a9aa0df36bbcf050c8bb", optional = true } +lightning = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "22cdec713c9c2f077d4e9f51c8f5cd615300085a", features = ["std"] } +lightning-types = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "22cdec713c9c2f077d4e9f51c8f5cd615300085a" } +lightning-invoice = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "22cdec713c9c2f077d4e9f51c8f5cd615300085a", features = ["std"] } +lightning-net-tokio = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "22cdec713c9c2f077d4e9f51c8f5cd615300085a" } +lightning-persister = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "22cdec713c9c2f077d4e9f51c8f5cd615300085a", features = ["tokio"], optional = true } +lightning-background-processor = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "22cdec713c9c2f077d4e9f51c8f5cd615300085a" } +lightning-rapid-gossip-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "22cdec713c9c2f077d4e9f51c8f5cd615300085a" } +lightning-block-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "22cdec713c9c2f077d4e9f51c8f5cd615300085a", features = ["rest-client", "rpc-client", "tokio"], optional = true } +lightning-transaction-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "22cdec713c9c2f077d4e9f51c8f5cd615300085a", optional = true } +lightning-liquidity = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "22cdec713c9c2f077d4e9f51c8f5cd615300085a", features = ["std"] } +lightning-macros = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "22cdec713c9c2f077d4e9f51c8f5cd615300085a" } +lightning-dns-resolver = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "22cdec713c9c2f077d4e9f51c8f5cd615300085a", optional = true } bdk_chain = { version = "0.23.3", default-features = false, features = ["std"] } bdk_esplora = { version = "0.22.2", default-features = false, features = ["async-https-rustls", "tokio"], optional = true } @@ -132,14 +132,14 @@ postgres-native-tls = { version = "0.5", default-features = false, features = [" vss-client = { package = "vss-client-ng", version = "0.6", optional = true } prost = { version = "0.11.6", default-features = false, optional = true} #bitcoin-payment-instructions = { version = "0.6" } -bitcoin-payment-instructions = { git = "https://github.com/tnull/bitcoin-payment-instructions", rev = "1d6ffaa8962391ddc84aeef98bd7439e55ccae9d", optional = true } +bitcoin-payment-instructions = { git = "https://github.com/benthecarman/bitcoin-payment-instructions", rev = "224b07fcc45a687bd56d8a9bded1e837d3d1bc37", optional = true } [target.'cfg(windows)'.dependencies] winapi = { version = "0.3", features = ["winbase"] } [dev-dependencies] -lightning = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "9174965af9437196c527a9aa0df36bbcf050c8bb", features = ["std", "_test_utils"] } -lightning-persister = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "9174965af9437196c527a9aa0df36bbcf050c8bb", features = ["tokio"] } +lightning = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "22cdec713c9c2f077d4e9f51c8f5cd615300085a", features = ["std", "_test_utils"] } +lightning-persister = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "22cdec713c9c2f077d4e9f51c8f5cd615300085a", features = ["tokio"] } rand = { version = "0.9.2", default-features = false, features = ["std", "thread_rng", "os_rng"] } proptest = "1.0.0" regex = "1.5.6" From 031ec5014ee2ecc5e55de5b5e82692a323505f3e Mon Sep 17 00:00:00 2001 From: benthecarman Date: Wed, 19 Aug 2026 03:37:46 -0500 Subject: [PATCH 2/4] Split merge behavior out of StorableObject StorableObject required an update representation from every type a store holds. Stores that only read, write, and delete whole objects still had to supply update methods that no caller used. Move merge behavior to UpdatableObject. Keep DataStore available to all storable objects, and restrict only merge operations to updatable objects. AI-assisted-by: OpenAI Codex and Anthropic Fable --- src/data_store.rs | 109 ++++++++++++++++----------- src/payment/pending_payment_store.rs | 7 +- src/payment/store.rs | 7 +- src/wallet/mod.rs | 2 +- 4 files changed, 74 insertions(+), 51 deletions(-) diff --git a/src/data_store.rs b/src/data_store.rs index a9fe0d0f5..ada994cf7 100644 --- a/src/data_store.rs +++ b/src/data_store.rs @@ -21,11 +21,20 @@ use crate::logger::{log_debug, log_error, LdkLogger}; use crate::types::DynStore; use crate::Error; +/// An object a store can read, write and delete as a whole, keyed by its own id. pub(crate) trait StorableObject: Clone + Readable + Writeable { type Id: StorableObjectId; - type Update: StorableObjectUpdate; fn id(&self) -> Self::Id; +} + +/// A [`StorableObject`] that a [`DataStore`] can merge an update into in place. +/// +/// Separate from [`StorableObject`] because stores that only ever replace whole objects have no use +/// for this, and requiring it of them would mean supplying an update representation nothing calls. +pub(crate) trait UpdatableObject: StorableObject { + type Update: StorableObjectUpdate; + fn update(&mut self, update: Self::Update) -> bool; fn to_update(&self) -> Self::Update; } @@ -321,33 +330,6 @@ where Ok(()) } - /// Like [`Self::insert`], but when an entry with the object's id already exists, merges the - /// object's full update ([`StorableObject::to_update`]) into it instead of replacing it. - /// - /// Returns whether anything was written. - pub(crate) async fn insert_or_update(&self, object: SO) -> Result { - let _guard = self.mutation_lock.write().await; - - let id = object.id(); - // Note we have to look through to the store here: merging against a cache miss would - // overwrite an evicted object with whatever the caller happens to know about it. - let data_to_persist = match self.lookup(&id).await? { - Some(mut existing_object) => { - existing_object.update(object.to_update()).then_some(existing_object) - }, - None => Some(object), - }; - - match data_to_persist { - Some(updated_object) => { - self.persist(&updated_object).await?; - self.cache.lock().expect("lock").insert(id, updated_object); - Ok(true) - }, - None => Ok(false), - } - } - /// Removes the object stored under `id`, if any. pub(crate) async fn remove(&self, id: &SO::Id) -> Result<(), Error> { let _guard = self.mutation_lock.write().await; @@ -390,23 +372,6 @@ where self.lookup(id).await } - /// Applies `update` to the object stored under its id. - pub(crate) async fn update(&self, update: SO::Update) -> Result { - let _guard = self.mutation_lock.write().await; - - let id = update.id(); - let Some(mut updated_object) = self.lookup(&id).await? else { - return Ok(DataStoreUpdateResult::NotFound); - }; - if !updated_object.update(update) { - return Ok(DataStoreUpdateResult::Unchanged); - } - - self.persist(&updated_object).await?; - self.cache.lock().expect("lock").insert(id, updated_object); - Ok(DataStoreUpdateResult::Updated) - } - /// Atomically transforms the entry for `id` through `f` and persists the result. /// /// `f` receives the current entry (`None` when absent) and returns the new state to write; @@ -730,6 +695,55 @@ where } } +impl DataStore +where + L::Target: LdkLogger, +{ + /// Like [`Self::insert`], but when an entry with the object's id already exists, merges the + /// object's full update ([`UpdatableObject::to_update`]) into it instead of replacing it. + /// + /// Returns whether anything was written. + pub(crate) async fn insert_or_update(&self, object: SO) -> Result { + let _guard = self.mutation_lock.write().await; + + let id = object.id(); + // Note we have to look through to the store here: merging against a cache miss would + // overwrite an evicted object with whatever the caller happens to know about it. + let data_to_persist = match self.lookup(&id).await? { + Some(mut existing_object) => { + existing_object.update(object.to_update()).then_some(existing_object) + }, + None => Some(object), + }; + + match data_to_persist { + Some(updated_object) => { + self.persist(&updated_object).await?; + self.cache.lock().expect("lock").insert(id, updated_object); + Ok(true) + }, + None => Ok(false), + } + } + + /// Applies `update` to the object stored under its id. + pub(crate) async fn update(&self, update: SO::Update) -> Result { + let _guard = self.mutation_lock.write().await; + + let id = update.id(); + let Some(mut updated_object) = self.lookup(&id).await? else { + return Ok(DataStoreUpdateResult::NotFound); + }; + if !updated_object.update(update) { + return Ok(DataStoreUpdateResult::Unchanged); + } + + self.persist(&updated_object).await?; + self.cache.lock().expect("lock").insert(id, updated_object); + Ok(DataStoreUpdateResult::Updated) + } +} + impl DataStore where L::Target: LdkLogger, @@ -832,11 +846,14 @@ mod tests { impl StorableObject for TestObject { type Id = TestObjectId; - type Update = TestObjectUpdate; fn id(&self) -> Self::Id { self.id } + } + + impl UpdatableObject for TestObject { + type Update = TestObjectUpdate; fn update(&mut self, update: Self::Update) -> bool { let mut updated = false; diff --git a/src/payment/pending_payment_store.rs b/src/payment/pending_payment_store.rs index 30a113537..e14f64c38 100644 --- a/src/payment/pending_payment_store.rs +++ b/src/payment/pending_payment_store.rs @@ -9,7 +9,7 @@ use bitcoin::Txid; use lightning::impl_writeable_tlv_based; use lightning::ln::channelmanager::PaymentId; -use crate::data_store::{StorableObject, StorableObjectUpdate}; +use crate::data_store::{StorableObject, StorableObjectUpdate, UpdatableObject}; use crate::payment::store::PaymentDetailsUpdate; use crate::payment::{PaymentDetails, PaymentKind}; @@ -78,11 +78,14 @@ pub(crate) struct PendingPaymentDetailsUpdate { impl StorableObject for PendingPaymentDetails { type Id = PaymentId; - type Update = PendingPaymentDetailsUpdate; fn id(&self) -> Self::Id { self.details.id } +} + +impl UpdatableObject for PendingPaymentDetails { + type Update = PendingPaymentDetailsUpdate; fn update(&mut self, update: Self::Update) -> bool { let mut updated = false; diff --git a/src/payment/store.rs b/src/payment/store.rs index 3163ed15b..41c39045f 100644 --- a/src/payment/store.rs +++ b/src/payment/store.rs @@ -22,7 +22,7 @@ use lightning::{ use lightning_types::payment::{PaymentHash, PaymentPreimage, PaymentSecret}; use lightning_types::string::UntrustedString; -use crate::data_store::{StorableObject, StorableObjectId, StorableObjectUpdate}; +use crate::data_store::{StorableObject, StorableObjectId, StorableObjectUpdate, UpdatableObject}; use crate::hex_utils; /// An opaque token used to continue a paginated listing. @@ -192,11 +192,14 @@ impl StorableObjectId for PaymentId { } impl StorableObject for PaymentDetails { type Id = PaymentId; - type Update = PaymentDetailsUpdate; fn id(&self) -> Self::Id { self.id } +} + +impl UpdatableObject for PaymentDetails { + type Update = PaymentDetailsUpdate; fn update(&mut self, update: Self::Update) -> bool { debug_assert_eq!( diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index b9c12b4a7..48d6b17a6 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -54,7 +54,7 @@ use lightning_invoice::RawBolt11Invoice; use persist::KVStoreWalletPersister; use crate::config::{Config, ADDRESS_POOL_SIZE}; -use crate::data_store::StorableObject; +use crate::data_store::UpdatableObject; #[cfg(test)] use crate::data_store::{KeepAllEntries, KeepLeastRecentlyUsed}; use crate::fee_estimator::{ConfirmationTarget, FeeEstimator, OnchainFeeEstimator}; From 767cbcc7254b6048ede4d47a20b4a0c6e4a9f625 Mon Sep 17 00:00:00 2001 From: benthecarman Date: Fri, 21 Aug 2026 02:58:09 -0500 Subject: [PATCH 3/4] Add uncached data store policy Add a cache policy that reads objects from persistent storage without keeping them in memory. Add a key-only empty check for namespaces that can grow without a bounded working set. AI-assisted-by: OpenAI Codex and Anthropic Fable --- src/data_store.rs | 127 +++++++++++++++++++++++++++++++++++++--------- 1 file changed, 103 insertions(+), 24 deletions(-) diff --git a/src/data_store.rs b/src/data_store.rs index ada994cf7..bdd190621 100644 --- a/src/data_store.rs +++ b/src/data_store.rs @@ -13,7 +13,7 @@ use std::ops::Deref; use std::sync::{Arc, Mutex}; use lightning::io::ErrorKind; -use lightning::util::persist::{KVStore, PageToken, PaginatedKVStore}; +use lightning::util::persist::{KVStore, PageToken, PaginatedKVStore, PaginatedListResponse}; use lightning::util::ser::{Readable, Writeable}; use crate::io::utils::process_kv_store_reads; @@ -64,6 +64,8 @@ pub(crate) enum DataStoreUpdateResult { /// How many of a namespace's objects a [`DataStore`] keeps in memory. #[derive(PartialEq, Eq, Debug, Clone, Copy)] pub(crate) enum CacheLimit { + /// Do not keep objects in memory. + Disabled, /// Keep every object in memory. Unbounded, /// Keep at most this many objects in memory. @@ -80,6 +82,18 @@ pub(crate) trait CachePolicy: Send + Sync + 'static { fn cache_limit(&self) -> CacheLimit; } +/// Keeps no objects in memory. +/// +/// Reads always go to the [`KVStore`]. Suitable for namespaces that can grow without bound and do +/// not need a cached working set. +pub(crate) struct KeepNoEntries; + +impl CachePolicy for KeepNoEntries { + fn cache_limit(&self) -> CacheLimit { + CacheLimit::Disabled + } +} + /// Keeps every object of the namespace in memory. /// /// Reads are served entirely from memory and never hit the [`KVStore`]. Required for stores whose @@ -178,6 +192,7 @@ impl LruCache { /// The in-memory part of a [`DataStore`]. enum ObjectCache { + None, KeepAll(HashMap), BoundedLru(LruCache), } @@ -185,6 +200,7 @@ enum ObjectCache { impl ObjectCache { fn new(cache_limit: CacheLimit, objects: Vec) -> Self { match cache_limit { + CacheLimit::Disabled => Self::None, CacheLimit::Unbounded => Self::KeepAll(HashMap::from_iter( objects.into_iter().map(|object| (object.id(), object)), )), @@ -205,6 +221,7 @@ impl ObjectCache { /// Returns the cached object for `id`, marking it as most recently used. fn get(&mut self, id: &SO::Id) -> Option { match self { + Self::None => None, Self::KeepAll(objects) => objects.get(id).cloned(), Self::BoundedLru(lru) => lru.get(id), } @@ -213,6 +230,7 @@ impl ObjectCache { /// Returns the cached object for `id`, without marking it as most recently used. fn peek(&self, id: &SO::Id) -> Option { match self { + Self::None => None, Self::KeepAll(objects) => objects.get(id).cloned(), Self::BoundedLru(lru) => lru.entries.get(id).map(|(object, _)| object.clone()), } @@ -221,6 +239,7 @@ impl ObjectCache { /// Returns whether `id` is cached, without marking it as most recently used. fn contains(&self, id: &SO::Id) -> bool { match self { + Self::None => false, Self::KeepAll(objects) => objects.contains_key(id), Self::BoundedLru(lru) => lru.entries.contains_key(id), } @@ -228,6 +247,7 @@ impl ObjectCache { fn insert(&mut self, id: SO::Id, object: SO) { match self { + Self::None => {}, Self::KeepAll(objects) => { objects.insert(id, object); }, @@ -237,6 +257,7 @@ impl ObjectCache { fn remove(&mut self, id: &SO::Id) { match self { + Self::None => {}, Self::KeepAll(objects) => { objects.remove(id); }, @@ -248,6 +269,7 @@ impl ObjectCache { /// namespace if [`Self::is_keep_all`]. fn filter bool>(&self, f: F) -> Vec { match self { + Self::None => Vec::new(), Self::KeepAll(objects) => objects.values().filter(f).cloned().collect(), Self::BoundedLru(lru) => { lru.entries.values().map(|(object, _)| object).filter(f).cloned().collect() @@ -258,6 +280,7 @@ impl ObjectCache { #[cfg(test)] fn len(&self) -> usize { match self { + Self::None => 0, Self::KeepAll(objects) => objects.len(), Self::BoundedLru(lru) => lru.entries.len(), } @@ -428,6 +451,11 @@ where self.contains(id).await } + /// Returns whether this store contains no objects. + pub(crate) async fn is_empty(&self) -> Result { + Ok(self.list_keys_page(None).await?.keys.is_empty()) + } + /// Returns a page of objects, ordered from most recently created to least recently created. /// /// Pass `None` to start at the most recently created object, and the returned @@ -451,29 +479,7 @@ where pub(crate) async fn list_page( &self, page_token: Option, ) -> Result, Error> { - 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 objects under {}/{} failed due to: {}", - &self.primary_namespace, - &self.secondary_namespace, - e - ); - // The backend rejects a token it didn't issue, which is the caller's problem rather - // than a persistence failure. - if e.kind() == ErrorKind::InvalidInput { - Error::InvalidPageToken - } else { - Error::PersistenceFailed - } - })?; + let response = self.list_keys_page(page_token).await?; // Serve whatever we already hold, and note the rest to read below. We take the mutation // lock only for this, so that we observe a consistent view of the cache without holding up @@ -501,6 +507,34 @@ where }) } + async fn list_keys_page( + &self, page_token: Option, + ) -> Result { + PaginatedKVStore::list_paginated( + &*self.kv_store, + &self.primary_namespace, + &self.secondary_namespace, + page_token, + ) + .await + .map_err(|e| { + log_error!( + self.logger, + "Listing objects under {}/{} failed due to: {}", + &self.primary_namespace, + &self.secondary_namespace, + e + ); + // The backend rejects a token it didn't issue, which is the caller's problem rather + // than a persistence failure. + if e.kind() == ErrorKind::InvalidInput { + Error::InvalidPageToken + } else { + Error::PersistenceFailed + } + }) + } + /// Reads the objects we couldn't serve from the cache into their slots in `objects`. /// /// Reads run concurrently but are tracked by slot, as the order in which they finish says @@ -1478,6 +1512,51 @@ mod tests { assert_eq!(0, lists.load(Ordering::Relaxed)); } + #[tokio::test] + async fn keep_no_entries_reads_through_without_caching() { + let reads = Arc::new(AtomicUsize::new(0)); + let kv_store: Arc = Arc::new(DynStoreWrapper(CountingStore { + inner: InMemoryStore::new(), + reads: Arc::clone(&reads), + writes: Arc::new(AtomicUsize::new(0)), + removes: Arc::new(AtomicUsize::new(0)), + lists: Arc::new(AtomicUsize::new(0)), + })); + let data_store = new_data_store(kv_store, KeepNoEntries, Vec::new()); + let id = test_id(1); + let object = TestObject::new(id, [23u8; 3]); + + data_store.insert(object).await.unwrap(); + assert_eq!(0, data_store.cached_len()); + + assert_eq!(Some(object), data_store.get(&id).await.unwrap()); + assert_eq!(Some(object), data_store.get(&id).await.unwrap()); + assert!(data_store.contains_key(&id).await.unwrap()); + assert_eq!(3, reads.load(Ordering::Relaxed)); + assert_eq!(0, data_store.cached_len()); + } + + #[tokio::test] + async fn is_empty_lists_keys_without_reading_objects() { + let reads = Arc::new(AtomicUsize::new(0)); + let lists = Arc::new(AtomicUsize::new(0)); + let kv_store: Arc = Arc::new(DynStoreWrapper(CountingStore { + inner: InMemoryStore::new(), + reads: Arc::clone(&reads), + writes: Arc::new(AtomicUsize::new(0)), + removes: Arc::new(AtomicUsize::new(0)), + lists: Arc::clone(&lists), + })); + let data_store = new_data_store(kv_store, KeepNoEntries, Vec::new()); + + assert!(data_store.is_empty().await.unwrap()); + data_store.insert(TestObject::new(test_id(1), [23u8; 3])).await.unwrap(); + assert!(!data_store.is_empty().await.unwrap()); + + assert_eq!(0, reads.load(Ordering::Relaxed)); + assert_eq!(2, lists.load(Ordering::Relaxed)); + } + #[tokio::test] async fn lru_evicts_least_recently_used() { let (data_store, _kv_store, ids) = new_lru_store_with_objects(2, 3).await; From 2397afdd5ff7f8d49af9fcb7840364a72bb12831 Mon Sep 17 00:00:00 2001 From: benthecarman Date: Fri, 21 Aug 2026 02:58:17 -0500 Subject: [PATCH 4/4] Add forwarded payment tracking Store unambiguous single-HTLC forwarding events. Aggregate them into per-channel and channel-pair statistics. Use fixed one-hour buckets for detailed records. Keep details out of the payment LRU cache. Use one persistence namespace for forwarding data. Expose analytics through Rust and UniFFI. Keep forwarding persistence and event-recording logic behind one internal store. AI-assisted-by: OpenAI Codex and Anthropic Fable --- bindings/ldk_node.udl | 25 + src/builder.rs | 82 +- src/config.rs | 32 +- src/error.rs | 10 + src/event.rs | 79 +- src/ffi/types.rs | 21 + src/io/mod.rs | 9 + src/lib.rs | 37 +- src/payment/forwarding.rs | 388 +++++++ src/payment/forwarding_store.rs | 1758 +++++++++++++++++++++++++++++++ src/payment/mod.rs | 7 + src/types.rs | 12 +- tests/integration_tests_rust.rs | 153 ++- 13 files changed, 2579 insertions(+), 34 deletions(-) create mode 100644 src/payment/forwarding.rs create mode 100644 src/payment/forwarding_store.rs diff --git a/bindings/ldk_node.udl b/bindings/ldk_node.udl index 4c4c1a438..b2f3d74ac 100644 --- a/bindings/ldk_node.udl +++ b/bindings/ldk_node.udl @@ -10,6 +10,8 @@ typedef dictionary ElectrumSyncConfig; typedef dictionary TorConfig; +typedef enum ForwardedPaymentTrackingMode; + typedef interface NodeEntropy; typedef interface ProbingConfig; @@ -68,6 +70,7 @@ interface Node { SpontaneousPayment spontaneous_payment(); OnchainPayment onchain_payment(); Liquidity liquidity(); + ForwardingAnalytics forwarding_analytics(); [Throws=NodeError] void lnurl_auth(string lnurl); [Throws=NodeError] @@ -139,6 +142,8 @@ interface FeeRate { typedef interface Liquidity; +typedef interface ForwardingAnalytics; + [Error] enum NodeError { "AlreadyRunning", @@ -177,6 +182,8 @@ enum NodeError { "InvalidOfferId", "InvalidNodeId", "InvalidPaymentId", + "InvalidForwardedPaymentId", + "InvalidChannelPairForwardingStatsId", "InvalidPaymentHash", "InvalidPaymentPreimage", "InvalidPaymentSecret", @@ -358,6 +365,12 @@ typedef string OfferId; [Custom] typedef string PaymentId; +[Custom] +typedef string ForwardedPaymentId; + +[Custom] +typedef string ChannelPairForwardingStatsId; + [Custom] typedef string PaymentHash; @@ -395,3 +408,15 @@ typedef enum Event; typedef interface HRNResolverConfig; typedef dictionary HumanReadableNamesConfig; + +typedef dictionary ForwardedPaymentDetails; + +typedef dictionary ChannelForwardingStats; + +typedef dictionary ChannelPairForwardingStats; + +typedef dictionary ForwardedPaymentDetailsPage; + +typedef dictionary ChannelForwardingStatsPage; + +typedef dictionary ChannelPairForwardingStatsPage; diff --git a/src/builder.rs b/src/builder.rs index ab641cf86..a85afb24c 100644 --- a/src/builder.rs +++ b/src/builder.rs @@ -80,7 +80,9 @@ use crate::io::utils::{ #[cfg(feature = "storage-vss")] use crate::io::vss_store::VssStoreBuilder; use crate::io::{ - self, PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, + self, CHANNEL_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE, + FORWARDED_PAYMENT_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, + PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, }; @@ -89,6 +91,7 @@ use crate::lnurl_auth::LnurlAuth; use crate::logger::{log_error, LdkLogger, LogLevel, LogWriter, Logger}; use crate::message_handler::NodeCustomMessageHandler; use crate::payment::asynchronous::om_mailbox::OnionMessageMailbox; +use crate::payment::forwarding_store::ForwardingStore; #[cfg(feature = "unified-payments")] use crate::payment::HRNResolver; use crate::peer_store::PeerStore; @@ -1524,26 +1527,37 @@ fn build_with_store_internal( let kv_store_ref = Arc::clone(&kv_store); let logger_ref = Arc::clone(&logger); - let (payment_store_res, node_metris_res, pending_payment_store_res, address_pool_res) = runtime - .block_on(async move { - tokio::join!( - read_n_objects( - &*kv_store_ref, - PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, - PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, - PAYMENT_CACHE_WARMUP_COUNT, - Arc::clone(&logger_ref), - ), - read_node_metrics(&*kv_store_ref, Arc::clone(&logger_ref)), - read_all_objects( - &*kv_store_ref, - PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, - PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, - Arc::clone(&logger_ref), - ), - read_address_pool(&*kv_store_ref, &*logger_ref) - ) - }); + let ( + payment_store_res, + channel_forwarding_stats_res, + node_metris_res, + pending_payment_store_res, + address_pool_res, + ) = runtime.block_on(async move { + tokio::join!( + read_n_objects( + &*kv_store_ref, + PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, + PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, + PAYMENT_CACHE_WARMUP_COUNT, + Arc::clone(&logger_ref), + ), + read_all_objects( + &*kv_store_ref, + FORWARDED_PAYMENT_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE, + Arc::clone(&logger_ref), + ), + read_node_metrics(&*kv_store_ref, Arc::clone(&logger_ref)), + read_all_objects( + &*kv_store_ref, + PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, + PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, + Arc::clone(&logger_ref), + ), + read_address_pool(&*kv_store_ref, &*logger_ref), + ) + }); // Initialize the status fields. let node_metrics = match node_metris_res { @@ -1576,6 +1590,14 @@ fn build_with_store_internal( }, }; + let channel_forwarding_stats = match channel_forwarding_stats_res { + Ok(stats) => stats, + Err(e) => { + log_error!(logger, "Failed to read channel forwarding stats from store: {}", e); + return Err(BuildError::ReadFailed); + }, + }; + let (chain_source, chain_tip_opt) = match chain_data_source_config { #[cfg(feature = "chain-esplora")] Some(ChainDataSourceConfig::Esplora { server_url, headers, sync_config }) => { @@ -1902,6 +1924,12 @@ fn build_with_store_internal( Arc::clone(&wallet), Arc::clone(&logger), )); + let forwarding_store = Arc::new(ForwardingStore::new( + channel_forwarding_stats, + config.forwarded_payment_tracking_mode, + Arc::clone(&kv_store), + Arc::clone(&logger), + )); let peer_storage_key = keys_manager.get_peer_storage_key(); let monitor_reader = Arc::new(AsyncPersister::new( @@ -2457,6 +2485,16 @@ fn build_with_store_internal( _leak_checker.0.push(Arc::downgrade(&wallet) as Weak); } + // How long detail records are kept before being folded into channel-pair buckets. `Stats` keeps + // none of its own, and only drains records a previous `Detailed` configuration left behind. + let forwarded_payment_aggregation_retention_secs = match config.forwarded_payment_tracking_mode + { + crate::config::ForwardedPaymentTrackingMode::Detailed => { + crate::payment::forwarding_store::FORWARDED_PAYMENT_AGGREGATION_BUCKET_SIZE_SECS + }, + crate::config::ForwardedPaymentTrackingMode::Stats => 0, + }; + Ok(Node { runtime, stop_sender, @@ -2484,6 +2522,8 @@ fn build_with_store_internal( scorer, peer_store, payment_store, + forwarding_store, + forwarded_payment_aggregation_retention_secs, lnurl_auth, is_running, node_metrics, diff --git a/src/config.rs b/src/config.rs index 65e256117..ac3b8e6e8 100644 --- a/src/config.rs +++ b/src/config.rs @@ -169,6 +169,30 @@ pub(crate) const LIQUIDITY_DISCOVERY_RETRY_INITIAL_DELAY: Duration = Duration::f // thereafter until every configured LSP has been discovered. pub(crate) const LIQUIDITY_DISCOVERY_RETRY_MAX_DELAY: Duration = Duration::from_secs(60 * 60); +/// The mode used for tracking forwarded payments. +/// +/// In either mode, a forward is tracked only when it has exactly one incoming HTLC and one outgoing +/// HTLC, and LDK reports both the outbound amount and total fee. +#[derive(Copy, Clone, Debug, PartialEq, Eq)] +#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] +pub enum ForwardedPaymentTrackingMode { + /// Track eligible new forwarded payments only as per-channel aggregate statistics. + /// + /// Any detailed records left by a previous configuration are aggregated and removed after their + /// current one-hour bucket closes. + Stats, + /// Store eligible individual forwarded payments for the current and previous one-hour buckets. + /// + /// Payments from older buckets are aggregated into channel-pair statistics and removed. + Detailed, +} + +impl Default for ForwardedPaymentTrackingMode { + fn default() -> Self { + Self::Stats + } +} + #[derive(Debug, Clone)] #[cfg_attr(feature = "uniffi", derive(uniffi::Record))] /// Represents the configuration of an [`Node`] instance. @@ -192,9 +216,10 @@ pub(crate) const LIQUIDITY_DISCOVERY_RETRY_MAX_DELAY: Duration = Duration::from_ doc = "| `hrn_config` | HumanReadableNamesConfig::default() |" )] /// | `manually_handle_unknown_bolt11_payments` | false | +/// | `forwarded_payment_tracking_mode` | Stats | /// -/// See [`AnchorChannelsConfig`] and [`RouteParametersConfig`] for more information regarding their -/// respective default values. +/// See [`AnchorChannelsConfig`], [`RouteParametersConfig`], and +/// [`ForwardedPaymentTrackingMode`] for more information regarding their respective default values. /// /// [`Node`]: crate::Node pub struct Config { @@ -268,6 +293,8 @@ pub struct Config { /// /// [`Event::PaymentClaimable`]: crate::Event::PaymentClaimable pub manually_handle_unknown_bolt11_payments: bool, + /// The mode used for tracking forwarded payments. + pub forwarded_payment_tracking_mode: ForwardedPaymentTrackingMode, } impl Default for Config { @@ -286,6 +313,7 @@ impl Default for Config { #[cfg(feature = "unified-payments")] hrn_config: HumanReadableNamesConfig::default(), manually_handle_unknown_bolt11_payments: false, + forwarded_payment_tracking_mode: ForwardedPaymentTrackingMode::default(), } } } diff --git a/src/error.rs b/src/error.rs index 485f944c2..d187c6820 100644 --- a/src/error.rs +++ b/src/error.rs @@ -89,6 +89,10 @@ pub enum Error { InvalidNodeId, /// The given payment id is invalid. InvalidPaymentId, + /// The given forwarded payment id is invalid. + InvalidForwardedPaymentId, + /// The given channel-pair forwarding statistics id is invalid. + InvalidChannelPairForwardingStatsId, /// The given payment hash is invalid. InvalidPaymentHash, /// The given payment pre-image is invalid. @@ -194,6 +198,12 @@ impl fmt::Display for Error { Self::InvalidOfferId => write!(f, "The given offer id is invalid."), Self::InvalidNodeId => write!(f, "The given node id is invalid."), Self::InvalidPaymentId => write!(f, "The given payment id is invalid."), + Self::InvalidForwardedPaymentId => { + write!(f, "The given forwarded payment id is invalid.") + }, + Self::InvalidChannelPairForwardingStatsId => { + write!(f, "The given channel-pair forwarding statistics id is invalid.") + }, Self::InvalidPaymentHash => write!(f, "The given payment hash is invalid."), Self::InvalidPaymentPreimage => write!(f, "The given payment preimage is invalid."), Self::InvalidPaymentSecret => write!(f, "The given payment secret is invalid."), diff --git a/src/event.rs b/src/event.rs index 0a3569755..4b689fbde 100644 --- a/src/event.rs +++ b/src/event.rs @@ -48,6 +48,7 @@ use crate::liquidity::LiquiditySource; use crate::logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger}; use crate::payment::asynchronous::om_mailbox::OnionMessageMailbox; use crate::payment::asynchronous::static_invoice_store::StaticInvoiceStore; +use crate::payment::forwarding_store::{ForwardRecord, ForwardingStore}; use crate::payment::store::{ PaymentDetails, PaymentDetailsUpdate, PaymentDirection, PaymentKind, PaymentStatus, }; @@ -551,6 +552,7 @@ where network_graph: Arc, liquidity_source: Arc>>, payment_store: Arc, + forwarding_store: Arc, peer_store: Arc>, keys_manager: Arc, static_invoice_store: Option, @@ -572,10 +574,10 @@ where channel_manager: Arc, connection_manager: Arc>, output_sweeper: Arc, network_graph: Arc, liquidity_source: Arc>>, payment_store: Arc, - peer_store: Arc>, keys_manager: Arc, - static_invoice_store: Option, onion_messenger: Arc, - om_mailbox: Option>, prober: Option>, - runtime: Arc, logger: L, config: Arc, + forwarding_store: Arc, peer_store: Arc>, + keys_manager: Arc, static_invoice_store: Option, + onion_messenger: Arc, om_mailbox: Option>, + prober: Option>, runtime: Arc, logger: L, config: Arc, ) -> Self { Self { event_queue, @@ -587,6 +589,7 @@ where network_graph, liquidity_source, payment_store, + forwarding_store, peer_store, keys_manager, static_invoice_store, @@ -1751,6 +1754,18 @@ where .await; } + self.forwarding_store + .record_forward(ForwardRecord { + prev_htlcs: &prev_htlcs, + next_htlcs: &next_htlcs, + total_fee_earned_msat, + skimmed_fee_msat, + claim_from_onchain_tx, + outbound_amount_forwarded_msat, + }) + .await + .map_err(|_| ReplayEvent())?; + let event = Event::PaymentForwarded { prev_htlcs: prev_htlcs.into_iter().map(HTLCLocator::from).collect(), next_htlcs: next_htlcs.into_iter().map(HTLCLocator::from).collect(), @@ -2237,6 +2252,16 @@ mod tests { use crate::payment::store::LSPS2Parameters; use crate::types::DynStoreWrapper; + fn ldk_htlc_locator(channel_byte: u8) -> LdkHtlcLocator { + LdkHtlcLocator { + channel_id: ChannelId([channel_byte; 32]), + htlc_id: None, + amount_msat: Some(channel_byte as u64), + user_channel_id: Some(channel_byte as u128), + node_id: None, + } + } + #[test] fn lsps2_payment_metadata_decodes_total_fee_limit() { let metadata = PaymentMetadata { @@ -2461,6 +2486,52 @@ mod tests { ); } + #[test] + fn event_queue_reads_legacy_multi_htlc_forward() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(TestLogger::new()); + let prev_htlcs = + vec![HTLCLocator::from(ldk_htlc_locator(1)), HTLCLocator::from(ldk_htlc_locator(2))]; + let next_htlcs = vec![HTLCLocator::from(ldk_htlc_locator(3))]; + let legacy_event = LegacyEvent::PaymentForwarded { + prev_htlcs: prev_htlcs.clone(), + next_htlcs: next_htlcs.clone(), + total_fee_earned_msat: Some(200), + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: Some(800), + }; + let persisted_bytes = encode_legacy_event_queue(legacy_event); + + let event_queue = + EventQueue::read(&mut &persisted_bytes[..], (Arc::clone(&store), logger)).unwrap(); + assert_eq!( + event_queue.next_event(), + Some(Event::PaymentForwarded { + prev_htlcs, + next_htlcs, + total_fee_earned_msat: Some(200), + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 800, + }) + ); + } + + #[test] + fn payment_forwarded_event_roundtrips() { + let event = Event::PaymentForwarded { + prev_htlcs: vec![HTLCLocator::from(ldk_htlc_locator(1))], + next_htlcs: vec![HTLCLocator::from(ldk_htlc_locator(2))], + total_fee_earned_msat: Some(200), + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 800, + }; + + assert_eq!(Event::read(&mut &event.encode()[..]).unwrap(), event); + } + #[tokio::test] async fn event_queue_concurrency() { let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); diff --git a/src/ffi/types.rs b/src/ffi/types.rs index d213f9d51..a153a77ef 100644 --- a/src/ffi/types.rs +++ b/src/ffi/types.rs @@ -159,6 +159,7 @@ pub use crate::config::default_config; use crate::error::Error; pub use crate::liquidity::LSPS1OrderStatus; pub use crate::logger::{LogLevel, LogRecord, LogWriter}; +use crate::payment::{ChannelPairForwardingStatsId, ForwardedPaymentId}; pub use crate::probing::ProbingConfig; use crate::{hex_utils, SocketAddress, UserChannelId}; @@ -1060,6 +1061,26 @@ uniffi::custom_type!(PaymentId, String, { }, }); +uniffi::custom_type!(ForwardedPaymentId, String, { + remote, + try_lift: |val| { + Ok(ForwardedPaymentId::from_str(&val)?) + }, + lower: |obj| { + obj.to_string() + }, +}); + +uniffi::custom_type!(ChannelPairForwardingStatsId, String, { + remote, + try_lift: |val| { + Ok(ChannelPairForwardingStatsId::from_str(&val)?) + }, + lower: |obj| { + obj.to_string() + }, +}); + uniffi::custom_type!(PaymentHash, String, { remote, try_lift: |val| { diff --git a/src/io/mod.rs b/src/io/mod.rs index c11475c43..b7e4d2131 100644 --- a/src/io/mod.rs +++ b/src/io/mod.rs @@ -37,6 +37,15 @@ pub(crate) const PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE: &str = ""; pub(crate) const PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE: &str = "pending_payments"; pub(crate) const PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE: &str = ""; +/// Forwarded payment information is persisted under this primary namespace. +pub(crate) const FORWARDED_PAYMENT_PERSISTENCE_PRIMARY_NAMESPACE: &str = "forwarded_payments"; +pub(crate) const FORWARDED_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE: &str = "details"; +pub(crate) const FORWARDED_PAYMENT_REPLAY_MARKER_PERSISTENCE_SECONDARY_NAMESPACE: &str = + "replay_markers"; +pub(crate) const CHANNEL_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE: &str = "channel_stats"; +pub(crate) const CHANNEL_PAIR_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE: &str = + "channel_pair_stats"; + /// The node metrics will be persisted under this key. pub(crate) const NODE_METRICS_PRIMARY_NAMESPACE: &str = ""; pub(crate) const NODE_METRICS_SECONDARY_NAMESPACE: &str = ""; diff --git a/src/lib.rs b/src/lib.rs index b6f790767..59fd362e2 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -175,9 +175,11 @@ use lnurl_auth::LnurlAuth; use logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger}; use payment::asynchronous::om_mailbox::OnionMessageMailbox; use payment::asynchronous::static_invoice_store::StaticInvoiceStore; +pub use payment::forwarding_store::aggregate_channel_pair_stats; +use payment::forwarding_store::{run_forwarded_payment_aggregation, ForwardingStore}; use payment::{ - Bolt11Payment, Bolt12Payment, OnchainPayment, PaymentDetails, PaymentDetailsPage, - SpontaneousPayment, + Bolt11Payment, Bolt12Payment, ForwardingAnalytics, OnchainPayment, PaymentDetails, + PaymentDetailsPage, SpontaneousPayment, }; #[cfg(feature = "unified-payments")] use payment::{HRNResolver, UnifiedPayment}; @@ -271,6 +273,8 @@ pub struct Node { scorer: Arc>, peer_store: Arc>>, payment_store: Arc, + forwarding_store: Arc, + forwarded_payment_aggregation_retention_secs: u64, lnurl_auth: Arc, is_running: Arc>, node_metrics: Arc, @@ -654,6 +658,14 @@ impl Node { chain_source.continuously_process_broadcast_queue(stop_tx_bcast).await }); + let retention_secs = self.forwarded_payment_aggregation_retention_secs; + let stop_aggregation = self.stop_sender.subscribe(); + let forwarding_store = Arc::clone(&self.forwarding_store); + self.runtime.spawn_cancellable_background_task(async move { + run_forwarded_payment_aggregation(stop_aggregation, forwarding_store, retention_secs) + .await; + }); + let bump_tx_event_handler = Arc::new(BumpTransactionEventHandler::new( Arc::clone(&self.tx_broadcaster), Arc::new(LdkWallet::new(Arc::clone(&self.wallet), Arc::clone(&self.logger))), @@ -678,6 +690,7 @@ impl Node { Arc::clone(&self.network_graph), Arc::clone(&self.liquidity_source), Arc::clone(&self.payment_store), + Arc::clone(&self.forwarding_store), Arc::clone(&self.peer_store), Arc::clone(&self.keys_manager), static_invoice_store, @@ -1200,6 +1213,26 @@ impl Node { } impl Node { + /// Returns a handler allowing to query forwarded payments and forwarding statistics. + #[cfg(not(feature = "uniffi"))] + pub fn forwarding_analytics(&self) -> ForwardingAnalytics { + ForwardingAnalytics::new( + Arc::clone(&self.runtime), + Arc::clone(&self.forwarding_store), + Arc::clone(&self.config), + ) + } + + /// Returns a handler allowing to query forwarded payments and forwarding statistics. + #[cfg(feature = "uniffi")] + pub fn forwarding_analytics(&self) -> Arc { + Arc::new(ForwardingAnalytics::new( + Arc::clone(&self.runtime), + Arc::clone(&self.forwarding_store), + Arc::clone(&self.config), + )) + } + /// Authenticates the user via [LNURL-auth] for the given LNURL string. /// /// [LNURL-auth]: https://github.com/lnurl/luds/blob/luds/04.md diff --git a/src/payment/forwarding.rs b/src/payment/forwarding.rs new file mode 100644 index 000000000..2a50f24c8 --- /dev/null +++ b/src/payment/forwarding.rs @@ -0,0 +1,388 @@ +// This file is Copyright its original authors, visible in version control history. +// +// This file is licensed under the Apache License, Version 2.0 or the MIT license , at your option. You may not use this file except in +// accordance with one or both of these licenses. + +//! Holds a handler allowing to query forwarded payments and forwarding statistics. + +use std::fmt; +use std::str::FromStr; +use std::sync::Arc; + +use bitcoin::secp256k1::PublicKey; +use lightning::impl_writeable_tlv_based; +use lightning::ln::types::ChannelId; +use lightning::util::ser::{Readable, Writeable, Writer}; + +use crate::config::{Config, ForwardedPaymentTrackingMode}; +use crate::error::Error; +use crate::ffi::{maybe_deref, maybe_wrap}; +use crate::hex_utils; +use crate::runtime::Runtime; +use crate::UserChannelId; + +use super::forwarding_store::ForwardingStore; +use super::PageToken; + +fn parse_forwarding_id(id: &str) -> Option<[u8; 32]> { + hex_utils::to_vec(id)?.try_into().ok() +} + +/// An identifier for a forwarded payment. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub struct ForwardedPaymentId(pub [u8; 32]); + +impl FromStr for ForwardedPaymentId { + type Err = Error; + + fn from_str(id: &str) -> Result { + parse_forwarding_id(id).map(Self).ok_or(Error::InvalidForwardedPaymentId) + } +} + +impl fmt::Display for ForwardedPaymentId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(&hex_utils::to_string(&self.0)) + } +} + +impl Writeable for ForwardedPaymentId { + fn write(&self, writer: &mut W) -> Result<(), lightning::io::Error> { + Ok(self.0.write(writer)?) + } +} + +impl Readable for ForwardedPaymentId { + fn read( + reader: &mut R, + ) -> Result { + Ok(Self(Readable::read(reader)?)) + } +} + +/// An identifier for an aggregated channel-pair forwarding statistics bucket. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub struct ChannelPairForwardingStatsId(pub [u8; 32]); + +impl FromStr for ChannelPairForwardingStatsId { + type Err = Error; + + fn from_str(id: &str) -> Result { + parse_forwarding_id(id).map(Self).ok_or(Error::InvalidChannelPairForwardingStatsId) + } +} + +impl fmt::Display for ChannelPairForwardingStatsId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(&hex_utils::to_string(&self.0)) + } +} + +impl Writeable for ChannelPairForwardingStatsId { + fn write(&self, writer: &mut W) -> Result<(), lightning::io::Error> { + Ok(self.0.write(writer)?) + } +} + +impl Readable for ChannelPairForwardingStatsId { + fn read( + reader: &mut R, + ) -> Result { + Ok(Self(Readable::read(reader)?)) + } +} + +/// Details of a payment that has been forwarded through this node. +/// +/// A forward is recorded only when it consisted of exactly one incoming and one outgoing HTLC and +/// LDK reported a total fee. LDK reports no fee when the incoming channel was force-closed and the +/// funds are claimed on chain, because the on-chain fees are not yet known at that point, so those +/// forwards are not recorded at all. +#[derive(Clone, Debug, PartialEq, Eq)] +#[cfg_attr(feature = "uniffi", derive(uniffi::Record))] +pub struct ForwardedPaymentDetails { + /// An opaque identifier for this forwarded payment. + pub id: ForwardedPaymentId, + /// The incoming channel id. + pub prev_channel_id: ChannelId, + /// The outgoing channel id. + pub next_channel_id: ChannelId, + /// The incoming user channel id, if available. + pub prev_user_channel_id: Option, + /// The outgoing user channel id, if available. + pub next_user_channel_id: Option, + /// The previous node id, if available. + pub prev_node_id: Option, + /// The next node id, if available. + pub next_node_id: Option, + /// The inbound amount attributed to this channel pair, in millisatoshis. + pub inbound_amount_forwarded_msat: Option, + /// The fee attributed to this channel pair, in millisatoshis. + pub total_fee_earned_msat: Option, + /// The skimmed fee attributed to this channel pair, in millisatoshis. + /// + /// This is the share of [`Self::total_fee_earned_msat`] that was withheld in addition to the + /// forwarding fee, not an amount earned on top of it. Adding the two would double-count. + pub skimmed_fee_msat: Option, + /// Whether the forwarded HTLC was claimed from an on-chain transaction. + pub claim_from_onchain_tx: bool, + /// The outbound amount attributed to this channel pair, in millisatoshis. + pub outbound_amount_forwarded_msat: Option, + /// The timestamp when this payment was forwarded. + pub forwarded_at_timestamp: u64, +} + +impl_writeable_tlv_based!(ForwardedPaymentDetails, { + (0, id, required), + (2, prev_channel_id, required), + (4, next_channel_id, required), + (6, prev_user_channel_id, option), + (8, next_user_channel_id, option), + (10, prev_node_id, option), + (12, next_node_id, option), + (14, total_fee_earned_msat, option), + (16, skimmed_fee_msat, option), + (18, claim_from_onchain_tx, required), + (20, outbound_amount_forwarded_msat, option), + (22, forwarded_at_timestamp, required), + (24, inbound_amount_forwarded_msat, option), +}); + +/// Aggregate statistics for forwarded payments through a single channel. +#[derive(Clone, Debug, PartialEq, Eq)] +#[cfg_attr(feature = "uniffi", derive(uniffi::Record))] +pub struct ChannelForwardingStats { + /// The channel id these stats apply to. + pub channel_id: ChannelId, + /// The channel counterparty node id, if known. + pub counterparty_node_id: Option, + /// Number of forwarded payments where this was the incoming channel. + pub inbound_payments_forwarded: u64, + /// Number of forwarded payments where this was the outgoing channel. + pub outbound_payments_forwarded: u64, + /// Total inbound amount forwarded through this channel, in millisatoshis. + pub total_inbound_amount_msat: u64, + /// Total outbound amount forwarded through this channel, in millisatoshis. + pub total_outbound_amount_msat: u64, + /// Total forwarding fees earned through this channel, in millisatoshis, if known for every + /// recorded forward. + /// + /// A single record covers this channel in both roles. A forward contributes its fee here when + /// this was the incoming channel, and contributes nothing when this was the outgoing channel, + /// because fees are attributed to the incoming side. + pub total_fee_earned_msat: Option, + /// Total skimmed fees earned through this channel, in millisatoshis. + /// + /// This is the share of [`Self::total_fee_earned_msat`] that was withheld in addition to the + /// forwarding fee, not an amount earned on top of it. Adding the two would double-count. + pub total_skimmed_fee_msat: u64, + /// Number of forwarded HTLCs that the next hop claimed from an on-chain transaction. + /// + /// A forward contributes here when this was the outgoing channel. + pub onchain_claims_count: u64, + /// Timestamp of the first forward recorded for this channel. + pub first_forwarded_at_timestamp: u64, + /// Timestamp of the latest forward recorded for this channel. + pub last_forwarded_at_timestamp: u64, +} + +impl_writeable_tlv_based!(ChannelForwardingStats, { + (0, channel_id, required), + (2, counterparty_node_id, option), + (4, inbound_payments_forwarded, required), + (6, outbound_payments_forwarded, required), + (8, total_inbound_amount_msat, required), + (10, total_outbound_amount_msat, required), + (12, total_fee_earned_msat, option), + (14, total_skimmed_fee_msat, required), + (16, onchain_claims_count, required), + (18, first_forwarded_at_timestamp, required), + (20, last_forwarded_at_timestamp, required), +}); + +/// Aggregated statistics for a specific channel pair. +#[derive(Clone, Debug, PartialEq, Eq)] +#[cfg_attr(feature = "uniffi", derive(uniffi::Record))] +pub struct ChannelPairForwardingStats { + /// An opaque identifier for this channel-pair bucket. + pub id: ChannelPairForwardingStatsId, + /// The incoming channel id. + pub prev_channel_id: ChannelId, + /// The outgoing channel id. + pub next_channel_id: ChannelId, + /// Start timestamp of this aggregation bucket. + pub bucket_start_timestamp: u64, + /// Width of this aggregation bucket, in seconds. + pub bucket_size_secs: u64, + /// The previous node id, if available. + pub prev_node_id: Option, + /// The next node id, if available. + pub next_node_id: Option, + /// Number of payments aggregated in this bucket. + pub payment_count: u64, + /// Total inbound amount in this bucket, in millisatoshis. + pub total_inbound_amount_msat: u64, + /// Total outbound amount in this bucket, in millisatoshis. + pub total_outbound_amount_msat: u64, + /// Total forwarding fees earned in this bucket, in millisatoshis, if known for every payment. + pub total_fee_earned_msat: Option, + /// Total skimmed fees in this bucket, in millisatoshis. + /// + /// This is the share of [`Self::total_fee_earned_msat`] that was withheld in addition to the + /// forwarding fee, not an amount earned on top of it. Adding the two would double-count. + pub total_skimmed_fee_msat: u64, + /// Number of forwarded HTLCs that the next hop claimed from an on-chain transaction. + pub onchain_claims_count: u64, + /// Average forwarding fee per payment, in millisatoshis, if known for every payment. + pub avg_fee_msat: Option, + /// Average inbound amount per payment, in millisatoshis. + pub avg_inbound_amount_msat: u64, + /// Timestamp of the first forward in this bucket. + pub first_forwarded_at_timestamp: u64, + /// Timestamp of the latest forward in this bucket. + pub last_forwarded_at_timestamp: u64, + /// Timestamp when this bucket was aggregated. + pub aggregated_at_timestamp: u64, +} + +impl_writeable_tlv_based!(ChannelPairForwardingStats, { + (0, id, required), + (2, prev_channel_id, required), + (4, next_channel_id, required), + (6, prev_node_id, option), + (8, next_node_id, option), + (10, payment_count, required), + (12, total_inbound_amount_msat, required), + (14, total_outbound_amount_msat, required), + (16, total_fee_earned_msat, option), + (18, total_skimmed_fee_msat, required), + (20, onchain_claims_count, required), + (22, avg_fee_msat, option), + (24, avg_inbound_amount_msat, required), + (26, first_forwarded_at_timestamp, required), + (28, last_forwarded_at_timestamp, required), + (30, aggregated_at_timestamp, required), + (32, bucket_start_timestamp, required), + (34, bucket_size_secs, required), +}); + +/// A page of forwarded payments returned from a paginated listing. +#[derive(Clone, Debug, PartialEq, Eq)] +#[cfg_attr(feature = "uniffi", derive(uniffi::Record))] +pub struct ForwardedPaymentDetailsPage { + /// Forwarded payments in this page. + pub payments: Vec, + /// Token to pass to the next call to continue listing, if another page exists. + pub next_page_token: Option, +} + +/// A page of channel forwarding statistics returned from a paginated listing. +#[derive(Clone, Debug, PartialEq, Eq)] +#[cfg_attr(feature = "uniffi", derive(uniffi::Record))] +pub struct ChannelForwardingStatsPage { + /// Channel forwarding statistics in this page. + pub stats: Vec, + /// Token to pass to the next call to continue listing, if another page exists. + pub next_page_token: Option, +} + +/// A page of channel-pair forwarding statistics returned from a paginated listing. +#[derive(Clone, Debug, PartialEq, Eq)] +#[cfg_attr(feature = "uniffi", derive(uniffi::Record))] +pub struct ChannelPairForwardingStatsPage { + /// Channel-pair forwarding statistics in this page. + pub stats: Vec, + /// Token to pass to the next call to continue listing, if another page exists. + pub next_page_token: Option, +} + +/// A handler allowing to query forwarded payments and forwarding statistics. +/// +/// Should be retrieved by calling [`Node::forwarding_analytics`]. +/// +/// [`Node::forwarding_analytics`]: crate::Node::forwarding_analytics +#[cfg_attr(feature = "uniffi", derive(uniffi::Object))] +pub struct ForwardingAnalytics { + runtime: Arc, + forwarding_store: Arc, + config: Arc, +} + +impl ForwardingAnalytics { + pub(crate) fn new( + runtime: Arc, forwarding_store: Arc, config: Arc, + ) -> Self { + Self { runtime, forwarding_store, config } + } + + /// Retrieves all channel forwarding statistics that match the given predicate. + pub fn list_channel_stats_with_filter bool>( + &self, f: F, + ) -> Vec { + self.runtime.block_on(self.forwarding_store.list_channel_stats_with_filter(f)) + } +} + +#[cfg_attr(feature = "uniffi", uniffi::export)] +impl ForwardingAnalytics { + /// Returns the configured forwarded payment tracking mode. + pub fn tracking_mode(&self) -> ForwardedPaymentTrackingMode { + self.config.forwarded_payment_tracking_mode + } + + /// Retrieve the details of a specific forwarded payment using its opaque identifier. + /// + /// The identifier is returned in [`ForwardedPaymentDetails::id`]. + pub fn payment( + &self, forwarded_payment_id: &ForwardedPaymentId, + ) -> Result, Error> { + self.runtime.block_on(self.forwarding_store.payment(forwarded_payment_id)) + } + + /// Retrieves a page of forwarded payments from the underlying paginated store. + pub fn list_payments( + &self, page_token: Option, + ) -> Result { + let ldk_page_token = page_token.as_ref().map(|token| maybe_deref(token).clone()); + let page = self.runtime.block_on(self.forwarding_store.list_payments(ldk_page_token))?; + Ok(ForwardedPaymentDetailsPage { + payments: page.objects, + next_page_token: page.next_page_token.map(maybe_wrap), + }) + } + + /// Retrieve the forwarding statistics for a specific channel. + pub fn channel_stats( + &self, channel_id: &ChannelId, + ) -> Result, Error> { + self.runtime.block_on(self.forwarding_store.channel_stats(channel_id)) + } + + /// Retrieves a page of channel forwarding statistics from the underlying paginated store. + pub fn list_channel_stats( + &self, page_token: Option, + ) -> Result { + let ldk_page_token = page_token.as_ref().map(|token| maybe_deref(token).clone()); + let page = + self.runtime.block_on(self.forwarding_store.list_channel_stats(ldk_page_token))?; + Ok(ChannelForwardingStatsPage { + stats: page.objects, + next_page_token: page.next_page_token.map(maybe_wrap), + }) + } + + /// Retrieves a page of channel pair forwarding statistics from the underlying paginated store. + pub fn list_channel_pair_stats( + &self, page_token: Option, + ) -> Result { + let ldk_page_token = page_token.as_ref().map(|token| maybe_deref(token).clone()); + let page = + self.runtime.block_on(self.forwarding_store.list_channel_pair_stats(ldk_page_token))?; + Ok(ChannelPairForwardingStatsPage { + stats: page.objects, + next_page_token: page.next_page_token.map(maybe_wrap), + }) + } +} diff --git a/src/payment/forwarding_store.rs b/src/payment/forwarding_store.rs new file mode 100644 index 000000000..722a9138d --- /dev/null +++ b/src/payment/forwarding_store.rs @@ -0,0 +1,1758 @@ +// This file is Copyright its original authors, visible in version control history. +// +// This file is licensed under the Apache License, Version 2.0 or the MIT license , at your option. You may not use this file except in +// accordance with one or both of these licenses. + +use std::collections::{HashMap, HashSet}; +use std::sync::Arc; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use bitcoin::hashes::{sha256, Hash}; +use bitcoin::secp256k1::PublicKey; +use lightning::events::HTLCLocator; +use lightning::impl_writeable_tlv_based; +use lightning::ln::types::ChannelId; +use lightning::util::logger::Logger as _; +use lightning::util::persist::PageToken; + +use super::forwarding::{ + ChannelForwardingStats, ChannelPairForwardingStats, ChannelPairForwardingStatsId, + ForwardedPaymentDetails, ForwardedPaymentId, +}; +use crate::config::ForwardedPaymentTrackingMode; +use crate::data_store::{ + DataStore, DataStorePage, KeepAllEntries, KeepNoEntries, StorableObject, StorableObjectId, + StorableObjectUpdate, UpdatableObject, +}; +use crate::hex_utils; +use crate::io::{ + CHANNEL_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE, + CHANNEL_PAIR_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE, + FORWARDED_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, + FORWARDED_PAYMENT_PERSISTENCE_PRIMARY_NAMESPACE, + FORWARDED_PAYMENT_REPLAY_MARKER_PERSISTENCE_SECONDARY_NAMESPACE, +}; +use crate::logger::{log_debug, log_error, Logger}; +use crate::types::{ + ChannelForwardingStatsStore, ChannelPairForwardingStatsStore, DynStore, ForwardedPaymentStore, +}; +use crate::{Error, UserChannelId}; + +pub(crate) const FORWARDED_PAYMENT_AGGREGATION_BUCKET_SIZE_SECS: u64 = 60 * 60; + +impl StorableObjectId for ForwardedPaymentId { + fn encode_to_hex_str(&self) -> String { + hex_utils::to_string(&self.0) + } + + fn decode_from_hex_str(s: &str) -> Option { + s.parse().ok() + } +} + +impl StorableObjectId for ChannelPairForwardingStatsId { + fn encode_to_hex_str(&self) -> String { + hex_utils::to_string(&self.0) + } + + fn decode_from_hex_str(s: &str) -> Option { + s.parse().ok() + } +} + +pub(crate) struct ForwardRecord<'a> { + pub prev_htlcs: &'a [HTLCLocator], + pub next_htlcs: &'a [HTLCLocator], + pub total_fee_earned_msat: Option, + pub skimmed_fee_msat: Option, + pub claim_from_onchain_tx: bool, + pub outbound_amount_forwarded_msat: u64, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +struct ForwardedPaymentReplayMarker { + id: ForwardedPaymentId, +} + +impl_writeable_tlv_based!(ForwardedPaymentReplayMarker, { (0, id, required) }); + +impl StorableObject for ForwardedPaymentReplayMarker { + type Id = ForwardedPaymentId; + + fn id(&self) -> Self::Id { + self.id + } +} + +type ForwardedPaymentReplayMarkerStore = + DataStore, KeepNoEntries>; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct StoredChannelForwardingStats { + channel_id: ChannelId, + counterparty_node_id: Option, + inbound_payments_forwarded: u64, + outbound_payments_forwarded: u64, + total_inbound_amount_msat: u64, + total_outbound_amount_msat: u64, + total_fee_earned_msat: Option, + total_skimmed_fee_msat: u64, + onchain_claims_count: u64, + first_forwarded_at_timestamp: u64, + last_forwarded_at_timestamp: u64, + last_inbound_forward_id: Option, + last_outbound_forward_id: Option, +} + +impl_writeable_tlv_based!(StoredChannelForwardingStats, { + (0, channel_id, required), + (2, counterparty_node_id, option), + (4, inbound_payments_forwarded, required), + (6, outbound_payments_forwarded, required), + (8, total_inbound_amount_msat, required), + (10, total_outbound_amount_msat, required), + (12, total_fee_earned_msat, option), + (14, total_skimmed_fee_msat, required), + (16, onchain_claims_count, required), + (18, first_forwarded_at_timestamp, required), + (20, last_forwarded_at_timestamp, required), + (22, last_inbound_forward_id, option), + (24, last_outbound_forward_id, option), +}); + +impl From for ChannelForwardingStats { + fn from(stats: StoredChannelForwardingStats) -> Self { + Self { + channel_id: stats.channel_id, + counterparty_node_id: stats.counterparty_node_id, + inbound_payments_forwarded: stats.inbound_payments_forwarded, + outbound_payments_forwarded: stats.outbound_payments_forwarded, + total_inbound_amount_msat: stats.total_inbound_amount_msat, + total_outbound_amount_msat: stats.total_outbound_amount_msat, + total_fee_earned_msat: stats.total_fee_earned_msat, + total_skimmed_fee_msat: stats.total_skimmed_fee_msat, + onchain_claims_count: stats.onchain_claims_count, + first_forwarded_at_timestamp: stats.first_forwarded_at_timestamp, + last_forwarded_at_timestamp: stats.last_forwarded_at_timestamp, + } + } +} + +impl From for StoredChannelForwardingStats { + fn from(stats: ChannelForwardingStats) -> Self { + Self { + channel_id: stats.channel_id, + counterparty_node_id: stats.counterparty_node_id, + inbound_payments_forwarded: stats.inbound_payments_forwarded, + outbound_payments_forwarded: stats.outbound_payments_forwarded, + total_inbound_amount_msat: stats.total_inbound_amount_msat, + total_outbound_amount_msat: stats.total_outbound_amount_msat, + total_fee_earned_msat: stats.total_fee_earned_msat, + total_skimmed_fee_msat: stats.total_skimmed_fee_msat, + onchain_claims_count: stats.onchain_claims_count, + first_forwarded_at_timestamp: stats.first_forwarded_at_timestamp, + last_forwarded_at_timestamp: stats.last_forwarded_at_timestamp, + last_inbound_forward_id: None, + last_outbound_forward_id: None, + } + } +} + +/// Owns all forwarding data stores and the operations that must keep them consistent. +pub(crate) struct ForwardingStore { + details: ForwardedPaymentStore, + replay_markers: ForwardedPaymentReplayMarkerStore, + channel_stats: ChannelForwardingStatsStore, + channel_pair_stats: ChannelPairForwardingStatsStore, + tracking_mode: ForwardedPaymentTrackingMode, + logger: Arc, +} + +impl ForwardingStore { + pub(crate) fn new( + channel_stats: Vec, + tracking_mode: ForwardedPaymentTrackingMode, kv_store: Arc, logger: Arc, + ) -> Self { + Self { + details: ForwardedPaymentStore::new( + Vec::new(), + KeepNoEntries, + FORWARDED_PAYMENT_PERSISTENCE_PRIMARY_NAMESPACE.to_string(), + FORWARDED_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE.to_string(), + Arc::clone(&kv_store), + Arc::clone(&logger), + ), + replay_markers: DataStore::new( + Vec::new(), + KeepNoEntries, + FORWARDED_PAYMENT_PERSISTENCE_PRIMARY_NAMESPACE.to_string(), + FORWARDED_PAYMENT_REPLAY_MARKER_PERSISTENCE_SECONDARY_NAMESPACE.to_string(), + Arc::clone(&kv_store), + Arc::clone(&logger), + ), + channel_stats: ChannelForwardingStatsStore::new( + channel_stats, + KeepAllEntries, + FORWARDED_PAYMENT_PERSISTENCE_PRIMARY_NAMESPACE.to_string(), + CHANNEL_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE.to_string(), + Arc::clone(&kv_store), + Arc::clone(&logger), + ), + channel_pair_stats: ChannelPairForwardingStatsStore::new( + Vec::new(), + KeepNoEntries, + FORWARDED_PAYMENT_PERSISTENCE_PRIMARY_NAMESPACE.to_string(), + CHANNEL_PAIR_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE.to_string(), + kv_store, + Arc::clone(&logger), + ), + tracking_mode, + logger, + } + } + + pub(crate) async fn record_forward(&self, forward: ForwardRecord<'_>) -> Result<(), Error> { + let ([prev_htlc], [next_htlc], Some(fee_earned_msat), Some(prev_htlc_id)) = ( + forward.prev_htlcs, + forward.next_htlcs, + forward.total_fee_earned_msat, + forward.prev_htlcs.first().and_then(|htlc| htlc.htlc_id), + ) else { + log_debug!( + self.logger, + "Skipping forwarding payment tracking for forward with {} inbound and {} outbound HTLC(s), because tracking requires a single HTLC in each direction, a known fee, and an incoming HTLC ID", + forward.prev_htlcs.len(), + forward.next_htlcs.len() + ); + return Ok(()); + }; + + let forward_id = forwarded_payment_id(&prev_htlc.channel_id, prev_htlc_id); + if self.replay_markers.contains_key(&forward_id).await.map_err(|e| { + log_error!(self.logger, "Failed to check forwarded payment replay marker: {e}"); + e + })? { + return Ok(()); + } + let details_id = matches!(self.tracking_mode, ForwardedPaymentTrackingMode::Detailed) + .then_some(forward_id); + let forwarded_at_timestamp = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("current time should not be earlier than the Unix epoch") + .as_secs(); + let inbound_amount_msat = + forward.outbound_amount_forwarded_msat.saturating_add(fee_earned_msat); + + // Store the stable forward ID with each directional update. If a later write fails, a replay + // can finish the remaining updates without applying a successful update twice. + let mut outbound_stats = StoredChannelForwardingStats::from(ChannelForwardingStats { + channel_id: next_htlc.channel_id, + counterparty_node_id: next_htlc.node_id, + inbound_payments_forwarded: 0, + outbound_payments_forwarded: 1, + total_inbound_amount_msat: 0, + total_outbound_amount_msat: forward.outbound_amount_forwarded_msat, + total_fee_earned_msat: Some(0), + total_skimmed_fee_msat: 0, + onchain_claims_count: u64::from(forward.claim_from_onchain_tx), + first_forwarded_at_timestamp: forwarded_at_timestamp, + last_forwarded_at_timestamp: forwarded_at_timestamp, + }); + outbound_stats.last_outbound_forward_id = Some(forward_id); + self.channel_stats.insert_or_update(outbound_stats).await.map_err(|e| { + log_error!(self.logger, "Failed to update outbound channel forwarding stats: {e}"); + e + })?; + + let mut inbound_stats = StoredChannelForwardingStats::from(ChannelForwardingStats { + channel_id: prev_htlc.channel_id, + counterparty_node_id: prev_htlc.node_id, + inbound_payments_forwarded: 1, + outbound_payments_forwarded: 0, + total_inbound_amount_msat: inbound_amount_msat, + total_outbound_amount_msat: 0, + total_fee_earned_msat: Some(fee_earned_msat), + total_skimmed_fee_msat: forward.skimmed_fee_msat.unwrap_or(0), + onchain_claims_count: 0, + first_forwarded_at_timestamp: forwarded_at_timestamp, + last_forwarded_at_timestamp: forwarded_at_timestamp, + }); + inbound_stats.last_inbound_forward_id = Some(forward_id); + self.channel_stats.insert_or_update(inbound_stats).await.map_err(|e| { + log_error!(self.logger, "Failed to update inbound channel forwarding stats: {e}"); + e + })?; + + if let Some(id) = details_id { + self.details + .insert(ForwardedPaymentDetails { + id, + prev_channel_id: prev_htlc.channel_id, + next_channel_id: next_htlc.channel_id, + prev_user_channel_id: prev_htlc.user_channel_id.map(UserChannelId), + next_user_channel_id: next_htlc.user_channel_id.map(UserChannelId), + prev_node_id: prev_htlc.node_id, + next_node_id: next_htlc.node_id, + inbound_amount_forwarded_msat: Some(inbound_amount_msat), + total_fee_earned_msat: Some(fee_earned_msat), + skimmed_fee_msat: forward.skimmed_fee_msat, + claim_from_onchain_tx: forward.claim_from_onchain_tx, + outbound_amount_forwarded_msat: Some(forward.outbound_amount_forwarded_msat), + forwarded_at_timestamp, + }) + .await + .map_err(|e| { + log_error!(self.logger, "Failed to store forwarded payment: {e}"); + e + })?; + } + + // Keep this marker after the event is handled. LDK can replay an older event after later + // events have replaced the directional retry tokens, and it provides no callback after its + // handled-event state is durable. + self.replay_markers.insert(ForwardedPaymentReplayMarker { id: forward_id }).await.map_err( + |e| { + log_error!(self.logger, "Failed to store forwarded payment replay marker: {e}"); + e + }, + )?; + + Ok(()) + } + + pub(crate) async fn payment( + &self, id: &ForwardedPaymentId, + ) -> Result, Error> { + self.details.get(id).await + } + + pub(crate) async fn list_payments( + &self, page_token: Option, + ) -> Result, Error> { + self.details.list_page(page_token).await + } + + pub(crate) async fn channel_stats( + &self, id: &ChannelId, + ) -> Result, Error> { + Ok(self.channel_stats.get(id).await?.map(Into::into)) + } + + pub(crate) async fn list_channel_stats( + &self, page_token: Option, + ) -> Result, Error> { + let page = self.channel_stats.list_page(page_token).await?; + Ok(DataStorePage { + objects: page.objects.into_iter().map(Into::into).collect(), + next_page_token: page.next_page_token, + }) + } + + pub(crate) async fn list_channel_stats_with_filter< + F: FnMut(&&ChannelForwardingStats) -> bool, + >( + &self, f: F, + ) -> Vec { + let stats: Vec<_> = + self.channel_stats.list_filter(|_| true).await.into_iter().map(Into::into).collect(); + let mut f = f; + stats.into_iter().filter(|stats| f(&stats)).collect() + } + + pub(crate) async fn list_channel_pair_stats( + &self, page_token: Option, + ) -> Result, Error> { + self.channel_pair_stats.list_page(page_token).await + } + + pub(crate) async fn aggregate_expired(&self, retention_secs: u64) -> Result<(u64, u64), Error> { + aggregate_expired_forwarded_payments( + &self.details, + &self.replay_markers, + &self.channel_pair_stats, + retention_secs, + &self.logger, + ) + .await + } +} + +fn forwarded_payment_id(channel_id: &ChannelId, htlc_id: u64) -> ForwardedPaymentId { + let mut bytes = [0u8; 40]; + bytes[..32].copy_from_slice(&channel_id.0); + bytes[32..].copy_from_slice(&htlc_id.to_be_bytes()); + ForwardedPaymentId(sha256::Hash::hash(&bytes).to_byte_array()) +} + +impl StorableObject for ForwardedPaymentDetails { + type Id = ForwardedPaymentId; + + fn id(&self) -> Self::Id { + self.id + } +} + +fn channel_pair_stats_id( + prev: &ChannelId, next: &ChannelId, bucket_start_timestamp: u64, +) -> ChannelPairForwardingStatsId { + // Hash the ordered channel pair and bucket start. The bucket size is fixed and therefore does + // not need to be part of the ID. Keeping the channels ordered preserves forwarding direction, + // unlike combining them with XOR. + let mut bytes = [0u8; 72]; + bytes[0..32].copy_from_slice(&prev.0); + bytes[32..64].copy_from_slice(&next.0); + bytes[64..72].copy_from_slice(&bucket_start_timestamp.to_be_bytes()); + ChannelPairForwardingStatsId(sha256::Hash::hash(&bytes).to_byte_array()) +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct ChannelForwardingStatsUpdate { + pub channel_id: ChannelId, + pub counterparty_node_id: Option, + pub inbound_payments_increment: u64, + pub outbound_payments_increment: u64, + pub inbound_amount_increment_msat: u64, + pub outbound_amount_increment_msat: u64, + pub fee_earned_increment_msat: Option, + pub skimmed_fee_increment_msat: u64, + pub onchain_claims_increment: u64, + pub timestamp: u64, + pub last_inbound_forward_id: Option, + pub last_outbound_forward_id: Option, +} + +impl StorableObjectUpdate for ChannelForwardingStatsUpdate { + fn id(&self) -> ChannelId { + self.channel_id + } +} + +impl StorableObjectId for ChannelId { + fn encode_to_hex_str(&self) -> String { + hex_utils::to_string(&self.0) + } + + fn decode_from_hex_str(s: &str) -> Option { + let bytes = hex_utils::to_vec(s)?; + Some(ChannelId(bytes.try_into().ok()?)) + } +} + +impl StorableObject for StoredChannelForwardingStats { + type Id = ChannelId; + + fn id(&self) -> Self::Id { + self.channel_id + } +} + +impl UpdatableObject for StoredChannelForwardingStats { + type Update = ChannelForwardingStatsUpdate; + + fn update(&mut self, update: Self::Update) -> bool { + debug_assert_eq!(self.channel_id, update.channel_id); + let mut updated = false; + if self.counterparty_node_id.is_none() && update.counterparty_node_id.is_some() { + self.counterparty_node_id = update.counterparty_node_id; + updated = true; + } + let apply_inbound = update.inbound_payments_increment > 0 + && (update.last_inbound_forward_id.is_none() + || update.last_inbound_forward_id != self.last_inbound_forward_id); + let apply_outbound = update.outbound_payments_increment > 0 + && (update.last_outbound_forward_id.is_none() + || update.last_outbound_forward_id != self.last_outbound_forward_id); + if apply_inbound { + self.inbound_payments_forwarded += update.inbound_payments_increment; + self.total_inbound_amount_msat += update.inbound_amount_increment_msat; + match (self.total_fee_earned_msat.as_mut(), update.fee_earned_increment_msat) { + (Some(total), Some(increment)) => *total += increment, + (Some(_), None) => self.total_fee_earned_msat = None, + _ => {}, + } + self.total_skimmed_fee_msat += update.skimmed_fee_increment_msat; + self.last_inbound_forward_id = update.last_inbound_forward_id; + updated = true; + } + if apply_outbound { + self.outbound_payments_forwarded += update.outbound_payments_increment; + self.total_outbound_amount_msat += update.outbound_amount_increment_msat; + self.onchain_claims_count += update.onchain_claims_increment; + self.last_outbound_forward_id = update.last_outbound_forward_id; + updated = true; + } + if updated { + self.first_forwarded_at_timestamp = + self.first_forwarded_at_timestamp.min(update.timestamp); + self.last_forwarded_at_timestamp = + self.last_forwarded_at_timestamp.max(update.timestamp); + } + updated + } + + fn to_update(&self) -> Self::Update { + ChannelForwardingStatsUpdate { + channel_id: self.channel_id, + counterparty_node_id: self.counterparty_node_id, + inbound_payments_increment: self.inbound_payments_forwarded, + outbound_payments_increment: self.outbound_payments_forwarded, + inbound_amount_increment_msat: self.total_inbound_amount_msat, + outbound_amount_increment_msat: self.total_outbound_amount_msat, + fee_earned_increment_msat: self.total_fee_earned_msat, + skimmed_fee_increment_msat: self.total_skimmed_fee_msat, + onchain_claims_increment: self.onchain_claims_count, + timestamp: self.last_forwarded_at_timestamp, + last_inbound_forward_id: self.last_inbound_forward_id, + last_outbound_forward_id: self.last_outbound_forward_id, + } + } +} + +impl StorableObject for ChannelPairForwardingStats { + type Id = ChannelPairForwardingStatsId; + + fn id(&self) -> Self::Id { + self.id + } +} + +fn seconds_until_next_forwarding_aggregation(now_timestamp: u64, bucket_size_secs: u64) -> u64 { + debug_assert!(bucket_size_secs > 0); + bucket_size_secs - (now_timestamp % bucket_size_secs) +} + +async fn aggregate_forwarded_payments_and_log( + forwarding_store: &ForwardingStore, retention_secs: u64, +) { + match forwarding_store.aggregate_expired(retention_secs).await { + Ok((pair_count, payment_count)) if pair_count > 0 => { + log_debug!( + forwarding_store.logger, + "Aggregated {} forwarded payments into {} channel pair buckets", + payment_count, + pair_count + ); + }, + Ok((0, payment_count)) if payment_count > 0 => { + log_debug!( + forwarding_store.logger, + "Removed {} forwarded payment details from previously aggregated buckets", + payment_count + ); + }, + Err(e) => { + log_error!(forwarding_store.logger, "Forwarded payment aggregation failed: {}", e) + }, + _ => {}, + } +} + +pub(crate) async fn run_forwarded_payment_aggregation( + mut stop_receiver: tokio::sync::watch::Receiver<()>, forwarding_store: Arc, + retention_secs: u64, +) { + if retention_secs == 0 { + match forwarding_store.details.is_empty().await { + Ok(true) => return, + Ok(false) => {}, + Err(e) => log_error!( + forwarding_store.logger, + "Failed to check forwarded payment store: {}", + e + ), + } + } + + aggregate_forwarded_payments_and_log(&forwarding_store, retention_secs).await; + + if retention_secs == 0 { + match forwarding_store.details.is_empty().await { + Ok(true) => return, + Ok(false) => {}, + Err(e) => log_error!( + forwarding_store.logger, + "Failed to check forwarded payment store: {}", + e + ), + } + } + + let period = Duration::from_secs(FORWARDED_PAYMENT_AGGREGATION_BUCKET_SIZE_SECS); + let now = + SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or(Duration::from_secs(0)).as_secs(); + let secs_until_next_bucket = seconds_until_next_forwarding_aggregation( + now, + FORWARDED_PAYMENT_AGGREGATION_BUCKET_SIZE_SECS, + ); + let first_tick = tokio::time::Instant::now() + Duration::from_secs(secs_until_next_bucket); + let mut interval = tokio::time::interval_at(first_tick, period); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + tokio::select! { + _ = stop_receiver.changed() => break, + _ = interval.tick() => { + aggregate_forwarded_payments_and_log(&forwarding_store, retention_secs).await; + if retention_secs == 0 { + match forwarding_store.details.is_empty().await { + Ok(true) => break, + Ok(false) => {}, + Err(e) => log_error!(forwarding_store.logger, "Failed to check forwarded payment store: {}", e), + } + } + } + } + } +} + +/// Aggregate forwarded payments older than the configured retention period into fixed-width +/// channel-pair statistics buckets. +async fn aggregate_expired_forwarded_payments( + forwarded_payment_store: &ForwardedPaymentStore, + replay_marker_store: &ForwardedPaymentReplayMarkerStore, + channel_pair_stats_store: &ChannelPairForwardingStatsStore, retention_secs: u64, + logger: &Arc, +) -> Result<(u64, u64), Error> { + let now = + SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or(Duration::from_secs(0)).as_secs(); + aggregate_expired_forwarded_payments_at( + forwarded_payment_store, + replay_marker_store, + channel_pair_stats_store, + FORWARDED_PAYMENT_AGGREGATION_BUCKET_SIZE_SECS, + retention_secs, + now, + logger, + ) + .await +} + +async fn aggregate_expired_forwarded_payments_at( + forwarded_payment_store: &ForwardedPaymentStore, + replay_marker_store: &ForwardedPaymentReplayMarkerStore, + channel_pair_stats_store: &ChannelPairForwardingStatsStore, bucket_size_secs: u64, + retention_secs: u64, now: u64, logger: &Arc, +) -> Result<(u64, u64), Error> { + if bucket_size_secs == 0 { + return Ok((0, 0)); + } + let retention_cutoff = now.saturating_sub(retention_secs); + let oldest_retained_bucket_start = + (retention_cutoff / bucket_size_secs).saturating_mul(bucket_size_secs); + + // Retain details for at least the configured period. We only aggregate complete buckets, so + // details may remain for up to one additional bucket width. Once an older bucket is persisted, + // it is a durable commit marker: retries can skip updating its totals and finish deleting detail + // records left behind by an interrupted cleanup. + // + // The scan below runs unlocked, so that a pass over a large store never stalls event handling. + // What keeps a record out of a bucket we have already closed is that width of slack, not any + // locking: a record would have to be stamped and then take longer than a whole bucket to be + // written. Event handling is serialized, so long before that the node has stopped forwarding + // anything at all. Insertions interleaving between pages are therefore fine, and a straggler we + // miss is found by the next pass, whose bucket marker exists by then, so it is deleted rather + // than counted twice. + let mut bucket_groups: HashMap<(ChannelId, ChannelId, u64), Vec> = + HashMap::new(); + let mut incomplete_buckets = HashSet::new(); + let mut page_token = None; + loop { + let page = forwarded_payment_store.list_page(page_token).await?; + for payment in page.objects { + if payment.forwarded_at_timestamp >= oldest_retained_bucket_start { + continue; + } + let bucket_start = + (payment.forwarded_at_timestamp / bucket_size_secs) * bucket_size_secs; + let bucket_key = (payment.prev_channel_id, payment.next_channel_id, bucket_start); + // The marker confirms that every analytics write for this forward succeeded. Wait for + // the whole bucket if any marker is absent, so a later replay cannot add a detail after + // the bucket totals are committed. + if !replay_marker_store.contains_key(&payment.id).await? { + incomplete_buckets.insert(bucket_key); + bucket_groups.remove(&bucket_key); + continue; + } + if incomplete_buckets.contains(&bucket_key) { + continue; + } + bucket_groups.entry(bucket_key).or_default().push(payment); + } + let Some(next_page_token) = page.next_page_token else { + break; + }; + page_token = Some(next_page_token); + } + if bucket_groups.is_empty() { + log_debug!(logger, "No forwarded payments in closed aggregation buckets found"); + return Ok((0, 0)); + } + + let mut aggregated_bucket_count = 0u64; + let mut payment_ids_to_remove = Vec::new(); + for ((prev_channel_id, next_channel_id, bucket_start), payments) in bucket_groups { + let pair_id = channel_pair_stats_id(&prev_channel_id, &next_channel_id, bucket_start); + + if channel_pair_stats_store.get(&pair_id).await?.is_none() { + let mut total_inbound_amount_msat = 0u64; + let mut total_outbound_amount_msat = 0u64; + let mut total_fee_earned_msat = Some(0u64); + let mut total_skimmed_fee_msat = 0u64; + let mut onchain_claims_count = 0u64; + let mut first_timestamp = u64::MAX; + let mut last_timestamp = 0u64; + + for payment in &payments { + let outbound = payment.outbound_amount_forwarded_msat.unwrap_or(0); + let fee = payment.total_fee_earned_msat; + let skimmed = payment.skimmed_fee_msat.unwrap_or(0); + let inbound = payment + .inbound_amount_forwarded_msat + .unwrap_or_else(|| outbound.saturating_add(fee.unwrap_or(0))); + total_inbound_amount_msat = total_inbound_amount_msat.saturating_add(inbound); + total_outbound_amount_msat = total_outbound_amount_msat.saturating_add(outbound); + total_fee_earned_msat = match (total_fee_earned_msat, fee) { + (Some(total), Some(fee)) => Some(total.saturating_add(fee)), + _ => None, + }; + total_skimmed_fee_msat = total_skimmed_fee_msat.saturating_add(skimmed); + if payment.claim_from_onchain_tx { + onchain_claims_count += 1; + } + first_timestamp = first_timestamp.min(payment.forwarded_at_timestamp); + last_timestamp = last_timestamp.max(payment.forwarded_at_timestamp); + } + + let payment_count = payments.len() as u64; + let prev_node_id = payments.iter().find_map(|payment| payment.prev_node_id); + let next_node_id = payments.iter().find_map(|payment| payment.next_node_id); + let stats = ChannelPairForwardingStats { + id: pair_id, + prev_channel_id, + next_channel_id, + bucket_start_timestamp: bucket_start, + bucket_size_secs, + prev_node_id, + next_node_id, + payment_count, + total_inbound_amount_msat, + total_outbound_amount_msat, + total_fee_earned_msat, + total_skimmed_fee_msat, + onchain_claims_count, + avg_fee_msat: total_fee_earned_msat.map(|total| total / payment_count), + avg_inbound_amount_msat: total_inbound_amount_msat / payment_count, + first_forwarded_at_timestamp: first_timestamp, + last_forwarded_at_timestamp: last_timestamp, + aggregated_at_timestamp: now, + }; + + channel_pair_stats_store.insert(stats).await.map_err(|e| { + log_error!( + logger, + "Failed to insert channel pair stats bucket for {pair_id:?}: {e}" + ); + e + })?; + aggregated_bucket_count += 1; + } + + payment_ids_to_remove.extend(payments.into_iter().map(|payment| payment.id())); + } + + // Removals acquire the mutation lock themselves, one at a time. + let mut removed_payment_count = 0u64; + for payment_id in payment_ids_to_remove { + forwarded_payment_store.remove(&payment_id).await.map_err(|e| { + log_error!(logger, "Failed to remove forwarded payment {:?}: {}", payment_id, e); + e + })?; + removed_payment_count += 1; + } + + Ok((aggregated_bucket_count, removed_payment_count)) +} + +/// Aggregates multiple channel-pair statistics buckets into cumulative totals. +/// +/// The result is computed here rather than read from the node, so the four fields that describe a +/// stored bucket do not carry their usual meaning: +/// +/// - [`bucket_start_timestamp`] and [`bucket_size_secs`] span the earliest input bucket start +/// through the latest input bucket end, gaps included, instead of describing one fixed-width +/// bucket. A gap is a period in which this channel pair forwarded nothing, so no bucket was +/// stored for it. +/// - `id` is the key of the earliest input bucket, because that is the bucket the span starts at. +/// It does not identify this result, and writing the result back under it would overwrite that +/// real bucket with a differently-shaped one. +/// - [`aggregated_at_timestamp`] is when this call ran, not when any input was aggregated. +/// +/// Returns `None` if `buckets` is empty or contains statistics for different channel pairs. +/// +/// [`bucket_start_timestamp`]: crate::payment::ChannelPairForwardingStats::bucket_start_timestamp +/// [`bucket_size_secs`]: crate::payment::ChannelPairForwardingStats::bucket_size_secs +/// [`aggregated_at_timestamp`]: crate::payment::ChannelPairForwardingStats::aggregated_at_timestamp +pub fn aggregate_channel_pair_stats( + buckets: &[ChannelPairForwardingStats], +) -> Option { + let first = buckets.first()?; + for bucket in &buckets[1..] { + if bucket.prev_channel_id != first.prev_channel_id + || bucket.next_channel_id != first.next_channel_id + { + return None; + } + } + + let mut payment_count = 0u64; + let mut total_inbound_amount_msat = 0u64; + let mut total_outbound_amount_msat = 0u64; + let mut total_fee_earned_msat = Some(0u64); + let mut total_skimmed_fee_msat = 0u64; + let mut onchain_claims_count = 0u64; + let mut first_forwarded_at_timestamp = u64::MAX; + let mut last_forwarded_at_timestamp = 0u64; + let mut earliest_bucket_start = u64::MAX; + let mut latest_bucket_end = 0u64; + let mut prev_node_id = None; + let mut next_node_id = None; + for bucket in buckets { + payment_count = payment_count.saturating_add(bucket.payment_count); + total_inbound_amount_msat = + total_inbound_amount_msat.saturating_add(bucket.total_inbound_amount_msat); + total_outbound_amount_msat = + total_outbound_amount_msat.saturating_add(bucket.total_outbound_amount_msat); + total_fee_earned_msat = match (total_fee_earned_msat, bucket.total_fee_earned_msat) { + (Some(total), Some(fee)) => Some(total.saturating_add(fee)), + _ => None, + }; + total_skimmed_fee_msat = + total_skimmed_fee_msat.saturating_add(bucket.total_skimmed_fee_msat); + onchain_claims_count = onchain_claims_count.saturating_add(bucket.onchain_claims_count); + first_forwarded_at_timestamp = + first_forwarded_at_timestamp.min(bucket.first_forwarded_at_timestamp); + last_forwarded_at_timestamp = + last_forwarded_at_timestamp.max(bucket.last_forwarded_at_timestamp); + earliest_bucket_start = earliest_bucket_start.min(bucket.bucket_start_timestamp); + latest_bucket_end = latest_bucket_end + .max(bucket.bucket_start_timestamp.saturating_add(bucket.bucket_size_secs)); + if prev_node_id.is_none() { + prev_node_id = bucket.prev_node_id; + } + if next_node_id.is_none() { + next_node_id = bucket.next_node_id; + } + } + let now = + SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or(Duration::from_secs(0)).as_secs(); + Some(ChannelPairForwardingStats { + id: channel_pair_stats_id( + &first.prev_channel_id, + &first.next_channel_id, + earliest_bucket_start, + ), + prev_channel_id: first.prev_channel_id, + next_channel_id: first.next_channel_id, + bucket_start_timestamp: earliest_bucket_start, + bucket_size_secs: latest_bucket_end.saturating_sub(earliest_bucket_start), + prev_node_id, + next_node_id, + payment_count, + total_inbound_amount_msat, + total_outbound_amount_msat, + total_fee_earned_msat, + total_skimmed_fee_msat, + onchain_claims_count, + avg_fee_msat: if payment_count > 0 { + total_fee_earned_msat.map(|total| total / payment_count) + } else { + None + }, + avg_inbound_amount_msat: if payment_count > 0 { + total_inbound_amount_msat / payment_count + } else { + 0 + }, + first_forwarded_at_timestamp, + last_forwarded_at_timestamp, + aggregated_at_timestamp: now, + }) +} + +#[cfg(test)] +mod forwarding_stats_tests { + use std::str::FromStr; + + use lightning::util::persist::{ + KVStore, KVSTORE_NAMESPACE_KEY_ALPHABET, KVSTORE_NAMESPACE_KEY_MAX_LEN, + }; + use lightning::util::ser::{Readable, Writeable}; + + use super::*; + use crate::data_store::KeepNoEntries; + #[cfg(feature = "storage-sqlite")] + use crate::io::sqlite_store::SqliteStore; + #[cfg(feature = "storage-sqlite")] + use crate::io::test_utils::random_storage_path; + use crate::io::test_utils::InMemoryStore; + use crate::types::{DynStore, DynStoreWrapper}; + + type TestForwardedPaymentStore = ForwardedPaymentStore; + type TestReplayMarkerStore = ForwardedPaymentReplayMarkerStore; + type TestChannelPairStatsStore = ChannelPairForwardingStatsStore; + + fn test_stores( + ) -> (TestForwardedPaymentStore, TestReplayMarkerStore, TestChannelPairStatsStore, Arc) + { + let (forwarded_payment_store, replay_marker_store, channel_pair_stats_store, logger, _) = + test_stores_with_kv(); + (forwarded_payment_store, replay_marker_store, channel_pair_stats_store, logger) + } + + fn test_stores_with_kv() -> ( + TestForwardedPaymentStore, + TestReplayMarkerStore, + TestChannelPairStatsStore, + Arc, + Arc, + ) { + let kv_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(Logger::new_log_facade()); + let primary_namespace = "test_forwarded_payments"; + let forwarded_payment_store = ForwardedPaymentStore::new( + Vec::new(), + KeepNoEntries, + primary_namespace.to_string(), + "details".to_string(), + Arc::clone(&kv_store), + Arc::clone(&logger), + ); + let replay_marker_store = ForwardedPaymentReplayMarkerStore::new( + Vec::new(), + KeepNoEntries, + primary_namespace.to_string(), + "replay_markers".to_string(), + Arc::clone(&kv_store), + Arc::clone(&logger), + ); + let channel_pair_stats_store = ChannelPairForwardingStatsStore::new( + Vec::new(), + KeepNoEntries, + primary_namespace.to_string(), + "pair_stats".to_string(), + Arc::clone(&kv_store), + Arc::clone(&logger), + ); + (forwarded_payment_store, replay_marker_store, channel_pair_stats_store, logger, kv_store) + } + + async fn insert_completed_payment( + forwarded_payment_store: &TestForwardedPaymentStore, + replay_marker_store: &TestReplayMarkerStore, payment: ForwardedPaymentDetails, + ) { + let id = payment.id(); + forwarded_payment_store.insert(payment).await.unwrap(); + replay_marker_store.insert(ForwardedPaymentReplayMarker { id }).await.unwrap(); + } + + fn forwarded_payment( + id: u8, forwarded_at_timestamp: u64, inbound_amount_msat: u64, outbound_amount_msat: u64, + fee_msat: u64, + ) -> ForwardedPaymentDetails { + ForwardedPaymentDetails { + id: ForwardedPaymentId([id; 32]), + prev_channel_id: ChannelId([1; 32]), + next_channel_id: ChannelId([2; 32]), + prev_user_channel_id: None, + next_user_channel_id: None, + prev_node_id: None, + next_node_id: None, + inbound_amount_forwarded_msat: Some(inbound_amount_msat), + total_fee_earned_msat: Some(fee_msat), + skimmed_fee_msat: Some(0), + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: Some(outbound_amount_msat), + forwarded_at_timestamp, + } + } + + fn channel_pair_stats( + bucket_start_timestamp: u64, bucket_size_secs: u64, aggregated_at_timestamp: u64, + payment_count: u64, total_inbound_amount_msat: u64, total_outbound_amount_msat: u64, + total_fee_earned_msat: u64, first_forwarded_at_timestamp: u64, + last_forwarded_at_timestamp: u64, + ) -> ChannelPairForwardingStats { + let prev_channel_id = ChannelId([1; 32]); + let next_channel_id = ChannelId([2; 32]); + ChannelPairForwardingStats { + id: channel_pair_stats_id(&prev_channel_id, &next_channel_id, bucket_start_timestamp), + prev_channel_id, + next_channel_id, + bucket_start_timestamp, + bucket_size_secs, + prev_node_id: None, + next_node_id: None, + payment_count, + total_inbound_amount_msat, + total_outbound_amount_msat, + total_fee_earned_msat: Some(total_fee_earned_msat), + total_skimmed_fee_msat: 0, + onchain_claims_count: 0, + avg_fee_msat: Some(total_fee_earned_msat / payment_count), + avg_inbound_amount_msat: total_inbound_amount_msat / payment_count, + first_forwarded_at_timestamp, + last_forwarded_at_timestamp, + aggregated_at_timestamp, + } + } + + #[test] + fn channel_pair_persistence_key_fits_kvstore_limit() { + let id = channel_pair_stats_id(&ChannelId([1; 32]), &ChannelId([2; 32]), 42); + let other_id = channel_pair_stats_id(&ChannelId([1; 32]), &ChannelId([2; 32]), 43); + let reversed_id = channel_pair_stats_id(&ChannelId([2; 32]), &ChannelId([1; 32]), 42); + + let key = id.encode_to_hex_str(); + assert_eq!(key.len(), 64); + assert!(key.len() <= KVSTORE_NAMESPACE_KEY_MAX_LEN); + assert!(key.chars().all(|c| KVSTORE_NAMESPACE_KEY_ALPHABET.contains(c))); + assert_ne!(key, other_id.encode_to_hex_str()); + assert_ne!(key, reversed_id.encode_to_hex_str()); + assert_eq!(ChannelPairForwardingStatsId::from_str(&key), Ok(id)); + assert!(ChannelPairForwardingStatsId::decode_from_hex_str("00").is_none()); + assert_eq!(ChannelPairForwardingStatsId::from_str(&key.to_uppercase()), Ok(id)); + } + + #[tokio::test] + async fn aggregation_retains_current_and_previous_buckets() { + let (forwarded_payment_store, replay_marker_store, channel_pair_stats_store, logger) = + test_stores(); + let closed_bucket_payment = forwarded_payment(1, 899, 110, 100, 10); + let partial_bucket_payment = forwarded_payment(2, 939, 220, 200, 20); + insert_completed_payment( + &forwarded_payment_store, + &replay_marker_store, + closed_bucket_payment.clone(), + ) + .await; + insert_completed_payment( + &forwarded_payment_store, + &replay_marker_store, + partial_bucket_payment.clone(), + ) + .await; + + // At timestamp 1,000, the current bucket starts at 960 and the previous bucket starts at + // 900. Only payments older than the previous bucket are aggregated. + assert_eq!( + aggregate_expired_forwarded_payments_at( + &forwarded_payment_store, + &replay_marker_store, + &channel_pair_stats_store, + 60, + 60, + 1_000, + &logger, + ) + .await, + Ok((1, 1)) + ); + + assert!(forwarded_payment_store.get(&closed_bucket_payment.id()).await.unwrap().is_none()); + assert_eq!( + forwarded_payment_store.get(&partial_bucket_payment.id()).await.unwrap(), + Some(partial_bucket_payment) + ); + let bucket_id = channel_pair_stats_id(&ChannelId([1; 32]), &ChannelId([2; 32]), 840); + let stats = channel_pair_stats_store.get(&bucket_id).await.unwrap().unwrap(); + assert_eq!(stats.bucket_size_secs, 60); + assert_eq!(stats.payment_count, 1); + assert_eq!(stats.total_inbound_amount_msat, 110); + assert_eq!(stats.total_outbound_amount_msat, 100); + assert_eq!(stats.total_fee_earned_msat, Some(10)); + assert_eq!(stats.aggregated_at_timestamp, 1_000); + } + + #[test] + fn aggregation_schedule_aligns_to_bucket_closure() { + assert_eq!(seconds_until_next_forwarding_aggregation(120, 60), 60); + assert_eq!(seconds_until_next_forwarding_aggregation(121, 60), 59); + assert_eq!(seconds_until_next_forwarding_aggregation(179, 60), 1); + } + + #[tokio::test] + async fn aggregation_waits_for_forward_replay_marker() { + let (forwarded_payment_store, replay_marker_store, channel_pair_stats_store, logger) = + test_stores(); + let payment = forwarded_payment(1, 850, 110, 100, 10); + let completed_payment = forwarded_payment(2, 851, 220, 200, 20); + forwarded_payment_store.insert(payment.clone()).await.unwrap(); + insert_completed_payment( + &forwarded_payment_store, + &replay_marker_store, + completed_payment.clone(), + ) + .await; + + assert_eq!( + aggregate_expired_forwarded_payments_at( + &forwarded_payment_store, + &replay_marker_store, + &channel_pair_stats_store, + 60, + 60, + 1_000, + &logger, + ) + .await, + Ok((0, 0)) + ); + assert_eq!( + forwarded_payment_store.get(&payment.id()).await.unwrap(), + Some(payment.clone()) + ); + assert_eq!( + forwarded_payment_store.get(&completed_payment.id()).await.unwrap(), + Some(completed_payment) + ); + + replay_marker_store + .insert(ForwardedPaymentReplayMarker { id: payment.id() }) + .await + .unwrap(); + assert_eq!( + aggregate_expired_forwarded_payments_at( + &forwarded_payment_store, + &replay_marker_store, + &channel_pair_stats_store, + 60, + 60, + 1_000, + &logger, + ) + .await, + Ok((1, 2)) + ); + assert!(forwarded_payment_store.get(&payment.id()).await.unwrap().is_none()); + let bucket_id = + channel_pair_stats_id(&payment.prev_channel_id, &payment.next_channel_id, 840); + assert_eq!( + channel_pair_stats_store.get(&bucket_id).await.unwrap().unwrap().payment_count, + 2 + ); + } + + #[tokio::test] + async fn background_aggregation_runs_immediately() { + let kv_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(Logger::new_log_facade()); + let forwarding_store = Arc::new(ForwardingStore::new( + Vec::new(), + ForwardedPaymentTrackingMode::Stats, + kv_store, + Arc::clone(&logger), + )); + let payment = forwarded_payment(1, 1, 110, 100, 10); + forwarding_store.details.insert(payment.clone()).await.unwrap(); + forwarding_store + .replay_markers + .insert(ForwardedPaymentReplayMarker { id: payment.id() }) + .await + .unwrap(); + let (_stop_sender, stop_receiver) = tokio::sync::watch::channel(()); + + tokio::time::timeout( + Duration::from_secs(1), + run_forwarded_payment_aggregation(stop_receiver, Arc::clone(&forwarding_store), 0), + ) + .await + .unwrap(); + + assert!(forwarding_store.details.is_empty().await.unwrap()); + let bucket_id = + channel_pair_stats_id(&payment.prev_channel_id, &payment.next_channel_id, 0); + assert_eq!( + forwarding_store + .channel_pair_stats + .get(&bucket_id) + .await + .unwrap() + .unwrap() + .payment_count, + 1 + ); + } + + #[tokio::test] + async fn forwarding_store_records_details_and_channel_stats() { + let kv_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(Logger::new_log_facade()); + let forwarding_store = ForwardingStore::new( + Vec::new(), + ForwardedPaymentTrackingMode::Detailed, + kv_store, + logger, + ); + let prev_channel_id = ChannelId([1; 32]); + let next_channel_id = ChannelId([2; 32]); + let prev_htlcs = [HTLCLocator { + channel_id: prev_channel_id, + htlc_id: Some(7), + amount_msat: Some(110), + user_channel_id: Some(3), + node_id: None, + }]; + let next_htlcs = [HTLCLocator { + channel_id: next_channel_id, + htlc_id: Some(8), + amount_msat: Some(100), + user_channel_id: Some(4), + node_id: None, + }]; + let details_id = forwarded_payment_id(&prev_channel_id, 7); + + let forward = || ForwardRecord { + prev_htlcs: &prev_htlcs, + next_htlcs: &next_htlcs, + total_fee_earned_msat: Some(10), + skimmed_fee_msat: Some(2), + claim_from_onchain_tx: true, + outbound_amount_forwarded_msat: 100, + }; + forwarding_store.record_forward(forward()).await.unwrap(); + forwarding_store.record_forward(forward()).await.unwrap(); + + let details = forwarding_store.payment(&details_id).await.unwrap().unwrap(); + assert_eq!(details.inbound_amount_forwarded_msat, Some(110)); + assert_eq!(details.outbound_amount_forwarded_msat, Some(100)); + assert_eq!(details.total_fee_earned_msat, Some(10)); + assert_eq!(details.skimmed_fee_msat, Some(2)); + assert_eq!(ForwardedPaymentId::from_str(&details_id.to_string()), Ok(details_id)); + assert_eq!( + ForwardedPaymentId::from_str("not-an-id"), + Err(Error::InvalidForwardedPaymentId) + ); + assert_eq!(ForwardedPaymentId::from_str("00"), Err(Error::InvalidForwardedPaymentId)); + assert_eq!( + ForwardedPaymentId::from_str(&details_id.to_string().to_uppercase()), + Ok(details_id) + ); + + let inbound = forwarding_store.channel_stats(&prev_channel_id).await.unwrap().unwrap(); + assert_eq!(inbound.inbound_payments_forwarded, 1); + assert_eq!(inbound.total_inbound_amount_msat, 110); + assert_eq!(inbound.total_fee_earned_msat, Some(10)); + assert_eq!(inbound.total_skimmed_fee_msat, 2); + + let outbound = forwarding_store.channel_stats(&next_channel_id).await.unwrap().unwrap(); + assert_eq!(outbound.outbound_payments_forwarded, 1); + assert_eq!(outbound.total_outbound_amount_msat, 100); + assert_eq!(outbound.total_fee_earned_msat, Some(0)); + assert_eq!(outbound.onchain_claims_count, 1); + assert_eq!(forwarding_store.list_payments(None).await.unwrap().objects.len(), 1); + } + + #[tokio::test] + async fn forwarding_store_deduplicates_stats_mode_replays() { + let kv_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(Logger::new_log_facade()); + let forwarding_store = + ForwardingStore::new(Vec::new(), ForwardedPaymentTrackingMode::Stats, kv_store, logger); + let prev_channel_id = ChannelId([1; 32]); + let next_channel_id = ChannelId([2; 32]); + let prev_htlcs = [HTLCLocator { + channel_id: prev_channel_id, + htlc_id: Some(7), + amount_msat: Some(110), + user_channel_id: Some(3), + node_id: None, + }]; + let next_htlcs = [HTLCLocator { + channel_id: next_channel_id, + htlc_id: Some(8), + amount_msat: Some(100), + user_channel_id: Some(4), + node_id: None, + }]; + let forward = || ForwardRecord { + prev_htlcs: &prev_htlcs, + next_htlcs: &next_htlcs, + total_fee_earned_msat: Some(10), + skimmed_fee_msat: Some(2), + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 100, + }; + + forwarding_store.record_forward(forward()).await.unwrap(); + forwarding_store.record_forward(forward()).await.unwrap(); + + let inbound = forwarding_store.channel_stats(&prev_channel_id).await.unwrap().unwrap(); + assert_eq!(inbound.inbound_payments_forwarded, 1); + assert_eq!(inbound.total_inbound_amount_msat, 110); + assert_eq!(inbound.total_fee_earned_msat, Some(10)); + let outbound = forwarding_store.channel_stats(&next_channel_id).await.unwrap().unwrap(); + assert_eq!(outbound.outbound_payments_forwarded, 1); + assert_eq!(outbound.total_outbound_amount_msat, 100); + assert!(forwarding_store.list_payments(None).await.unwrap().objects.is_empty()); + } + + #[tokio::test] + async fn forwarding_store_skips_ineligible_forwards() { + let kv_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(Logger::new_log_facade()); + let forwarding_store = ForwardingStore::new( + Vec::new(), + ForwardedPaymentTrackingMode::Detailed, + kv_store, + logger, + ); + let prev_htlcs = [ + HTLCLocator { + channel_id: ChannelId([1; 32]), + htlc_id: None, + amount_msat: Some(110), + user_channel_id: None, + node_id: None, + }, + HTLCLocator { + channel_id: ChannelId([2; 32]), + htlc_id: None, + amount_msat: Some(110), + user_channel_id: None, + node_id: None, + }, + ]; + let next_htlcs = [HTLCLocator { + channel_id: ChannelId([3; 32]), + htlc_id: None, + amount_msat: Some(100), + user_channel_id: None, + node_id: None, + }]; + + forwarding_store + .record_forward(ForwardRecord { + prev_htlcs: &prev_htlcs, + next_htlcs: &next_htlcs, + total_fee_earned_msat: Some(10), + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 100, + }) + .await + .unwrap(); + forwarding_store + .record_forward(ForwardRecord { + prev_htlcs: &prev_htlcs[..1], + next_htlcs: &next_htlcs, + total_fee_earned_msat: None, + skimmed_fee_msat: None, + claim_from_onchain_tx: true, + outbound_amount_forwarded_msat: 100, + }) + .await + .unwrap(); + + assert!(forwarding_store.details.is_empty().await.unwrap()); + assert!(forwarding_store.channel_stats.is_empty().await.unwrap()); + } + + #[tokio::test] + async fn aggregation_preserves_unknown_fees() { + let (forwarded_payment_store, replay_marker_store, channel_pair_stats_store, logger) = + test_stores(); + let known_fee_payment = forwarded_payment(1, 850, 110, 100, 10); + let mut unknown_fee_payment = forwarded_payment(2, 851, 200, 200, 0); + unknown_fee_payment.total_fee_earned_msat = None; + insert_completed_payment(&forwarded_payment_store, &replay_marker_store, known_fee_payment) + .await; + insert_completed_payment( + &forwarded_payment_store, + &replay_marker_store, + unknown_fee_payment, + ) + .await; + + assert_eq!( + aggregate_expired_forwarded_payments_at( + &forwarded_payment_store, + &replay_marker_store, + &channel_pair_stats_store, + 60, + 60, + 1_000, + &logger, + ) + .await, + Ok((1, 2)) + ); + + let bucket_id = channel_pair_stats_id(&ChannelId([1; 32]), &ChannelId([2; 32]), 840); + let stats = channel_pair_stats_store.get(&bucket_id).await.unwrap().unwrap(); + assert_eq!(stats.payment_count, 2); + assert_eq!(stats.total_fee_earned_msat, None); + assert_eq!(stats.avg_fee_msat, None); + } + + #[test] + fn channel_stats_update_preserves_unknown_fees() { + let mut stats = StoredChannelForwardingStats::from(ChannelForwardingStats { + channel_id: ChannelId([1; 32]), + counterparty_node_id: None, + inbound_payments_forwarded: 1, + outbound_payments_forwarded: 0, + total_inbound_amount_msat: 110, + total_outbound_amount_msat: 0, + total_fee_earned_msat: Some(10), + total_skimmed_fee_msat: 0, + onchain_claims_count: 0, + first_forwarded_at_timestamp: 850, + last_forwarded_at_timestamp: 850, + }); + + assert!(stats.update(ChannelForwardingStatsUpdate { + channel_id: stats.channel_id, + counterparty_node_id: None, + inbound_payments_increment: 1, + outbound_payments_increment: 0, + inbound_amount_increment_msat: 200, + outbound_amount_increment_msat: 0, + fee_earned_increment_msat: None, + skimmed_fee_increment_msat: 0, + onchain_claims_increment: 0, + timestamp: 851, + last_inbound_forward_id: None, + last_outbound_forward_id: None, + })); + assert_eq!(stats.total_fee_earned_msat, None); + } + + #[test] + fn stored_channel_stats_reads_legacy_stats() { + let legacy = ChannelForwardingStats { + channel_id: ChannelId([1; 32]), + counterparty_node_id: None, + inbound_payments_forwarded: 1, + outbound_payments_forwarded: 2, + total_inbound_amount_msat: 110, + total_outbound_amount_msat: 200, + total_fee_earned_msat: Some(10), + total_skimmed_fee_msat: 2, + onchain_claims_count: 1, + first_forwarded_at_timestamp: 850, + last_forwarded_at_timestamp: 851, + }; + + let encoded = legacy.encode(); + let stored = StoredChannelForwardingStats::read(&mut &encoded[..]).unwrap(); + assert_eq!(stored.last_inbound_forward_id, None); + assert_eq!(stored.last_outbound_forward_id, None); + assert_eq!(ChannelForwardingStats::from(stored), legacy); + } + + #[tokio::test] + async fn aggregation_keeps_both_retained_bucket_boundaries() { + let (forwarded_payment_store, replay_marker_store, channel_pair_stats_store, logger) = + test_stores(); + let older_bucket_payment = forwarded_payment(1, 839, 110, 100, 10); + let previous_bucket_payment = forwarded_payment(2, 840, 220, 200, 20); + let current_bucket_payment = forwarded_payment(3, 900, 330, 300, 30); + insert_completed_payment( + &forwarded_payment_store, + &replay_marker_store, + older_bucket_payment.clone(), + ) + .await; + insert_completed_payment( + &forwarded_payment_store, + &replay_marker_store, + previous_bucket_payment.clone(), + ) + .await; + insert_completed_payment( + &forwarded_payment_store, + &replay_marker_store, + current_bucket_payment.clone(), + ) + .await; + + assert_eq!( + aggregate_expired_forwarded_payments_at( + &forwarded_payment_store, + &replay_marker_store, + &channel_pair_stats_store, + 60, + 60, + 900, + &logger, + ) + .await, + Ok((1, 1)) + ); + + assert!(forwarded_payment_store.get(&older_bucket_payment.id()).await.unwrap().is_none()); + assert_eq!( + forwarded_payment_store.get(&previous_bucket_payment.id()).await.unwrap(), + Some(previous_bucket_payment) + ); + assert_eq!( + forwarded_payment_store.get(¤t_bucket_payment.id()).await.unwrap(), + Some(current_bucket_payment) + ); + } + + #[tokio::test] + async fn zero_retention_cleans_up_after_the_current_bucket_closes() { + let (forwarded_payment_store, replay_marker_store, channel_pair_stats_store, logger) = + test_stores(); + let payment = forwarded_payment(1, 899, 110, 100, 10); + insert_completed_payment(&forwarded_payment_store, &replay_marker_store, payment.clone()) + .await; + + assert_eq!( + aggregate_expired_forwarded_payments_at( + &forwarded_payment_store, + &replay_marker_store, + &channel_pair_stats_store, + 60, + 0, + 899, + &logger, + ) + .await, + Ok((0, 0)) + ); + assert_eq!( + forwarded_payment_store.get(&payment.id()).await.unwrap(), + Some(payment.clone()) + ); + + assert_eq!( + aggregate_expired_forwarded_payments_at( + &forwarded_payment_store, + &replay_marker_store, + &channel_pair_stats_store, + 60, + 0, + 900, + &logger, + ) + .await, + Ok((1, 1)) + ); + assert!(forwarded_payment_store.get(&payment.id()).await.unwrap().is_none()); + } + + #[tokio::test] + async fn aggregation_retry_only_cleans_up_committed_bucket() { + let (forwarded_payment_store, replay_marker_store, channel_pair_stats_store, logger) = + test_stores(); + let remaining_payment = forwarded_payment(2, 851, 220, 200, 20); + let committed_stats = channel_pair_stats(840, 60, 950, 2, 330, 300, 30, 850, 851); + channel_pair_stats_store.insert(committed_stats.clone()).await.unwrap(); + insert_completed_payment( + &forwarded_payment_store, + &replay_marker_store, + remaining_payment.clone(), + ) + .await; + + // This represents a retry after the bucket write and one of two detail deletions + // succeeded. The existing bucket is the commit marker, so its totals must not change. + assert_eq!( + aggregate_expired_forwarded_payments_at( + &forwarded_payment_store, + &replay_marker_store, + &channel_pair_stats_store, + 60, + 60, + 1_000, + &logger, + ) + .await, + Ok((0, 1)) + ); + + assert!(forwarded_payment_store.get(&remaining_payment.id()).await.unwrap().is_none()); + assert_eq!( + channel_pair_stats_store.get(&committed_stats.id()).await.unwrap(), + Some(committed_stats) + ); + } + + #[tokio::test] + async fn aggregation_preserves_details_if_commit_marker_is_corrupt() { + let ( + forwarded_payment_store, + replay_marker_store, + channel_pair_stats_store, + logger, + kv_store, + ) = test_stores_with_kv(); + let payment = forwarded_payment(2, 851, 220, 200, 20); + let pair_id = + channel_pair_stats_id(&payment.prev_channel_id, &payment.next_channel_id, 840); + insert_completed_payment(&forwarded_payment_store, &replay_marker_store, payment.clone()) + .await; + KVStore::write( + &*kv_store, + "test_forwarded_payments", + "pair_stats", + &pair_id.encode_to_hex_str(), + vec![0xff], + ) + .await + .unwrap(); + + assert_eq!( + aggregate_expired_forwarded_payments_at( + &forwarded_payment_store, + &replay_marker_store, + &channel_pair_stats_store, + 60, + 60, + 1_000, + &logger, + ) + .await, + Err(Error::PersistenceFailed) + ); + assert_eq!(forwarded_payment_store.get(&payment.id()).await.unwrap(), Some(payment)); + } + + #[test] + fn cumulative_stats_prefer_known_node_ids_and_cover_bucket_span() { + let unknown_nodes = channel_pair_stats(840, 60, 950, 2, 330, 300, 30, 850, 851); + let mut known_nodes = channel_pair_stats(900, 120, 1_000, 1, 110, 100, 10, 902, 902); + let node_id = PublicKey::from_str( + "0279be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798", + ) + .unwrap(); + known_nodes.prev_node_id = Some(node_id); + known_nodes.next_node_id = Some(node_id); + + let cumulative = aggregate_channel_pair_stats(&[unknown_nodes, known_nodes]).unwrap(); + assert_eq!(cumulative.prev_node_id, Some(node_id)); + assert_eq!(cumulative.next_node_id, Some(node_id)); + assert_eq!(cumulative.payment_count, 3); + assert_eq!(cumulative.total_fee_earned_msat, Some(40)); + // The span runs from the earliest bucket start to the latest bucket end, gaps included. + assert_eq!(cumulative.bucket_start_timestamp, 840); + assert_eq!(cumulative.bucket_size_secs, 180); + } + + #[test] + fn cumulative_stats_reject_mismatched_channel_pairs() { + let mut other_pair = channel_pair_stats(900, 60, 1_000, 1, 110, 100, 10, 902, 902); + other_pair.next_channel_id = ChannelId([3; 32]); + + assert!(aggregate_channel_pair_stats(&[]).is_none()); + assert!(aggregate_channel_pair_stats(&[ + channel_pair_stats(840, 60, 950, 2, 330, 300, 30, 850, 851), + other_pair, + ]) + .is_none()); + } + + #[tokio::test] + async fn retention_does_not_change_bucket_geometry() { + let (forwarded_payment_store, replay_marker_store, channel_pair_stats_store, logger) = + test_stores(); + let old_stats = channel_pair_stats(0, 3_600, 7_200, 1, 110, 100, 10, 100, 100); + let payment = forwarded_payment(2, 4_000, 220, 200, 20); + channel_pair_stats_store.insert(old_stats.clone()).await.unwrap(); + insert_completed_payment(&forwarded_payment_store, &replay_marker_store, payment.clone()) + .await; + + assert_eq!( + aggregate_expired_forwarded_payments_at( + &forwarded_payment_store, + &replay_marker_store, + &channel_pair_stats_store, + 3_600, + 7_200, + 15_000, + &logger, + ) + .await, + Ok((1, 1)) + ); + + let new_id = + channel_pair_stats_id(&payment.prev_channel_id, &payment.next_channel_id, 3_600); + assert_eq!(channel_pair_stats_store.get(&old_stats.id()).await.unwrap(), Some(old_stats)); + assert_eq!(channel_pair_stats_store.get(&new_id).await.unwrap().unwrap().payment_count, 1); + assert!(forwarded_payment_store.get(&payment.id()).await.unwrap().is_none()); + } + + #[tokio::test] + #[cfg(feature = "storage-sqlite")] + async fn channel_pair_stats_persist_to_sqlite() { + let mut storage_path = random_storage_path(); + storage_path.push("channel_pair_stats_persist_to_sqlite"); + let sqlite_store = + SqliteStore::new(storage_path, Some("stats.sqlite".to_string()), None).unwrap(); + let kv_store: Arc = Arc::new(DynStoreWrapper(sqlite_store)); + let logger = Arc::new(Logger::new_log_facade()); + let namespace = "sqlite_channel_pair_stats"; + let secondary_namespace = "pair_stats"; + let stats_store = ChannelPairForwardingStatsStore::new( + Vec::new(), + KeepNoEntries, + namespace.to_string(), + secondary_namespace.to_string(), + Arc::clone(&kv_store), + logger, + ); + let stats = channel_pair_stats(840, 60, 1_000, 2, 330, 300, 30, 850, 851); + + stats_store.insert(stats.clone()).await.unwrap(); + let keys = KVStore::list(&*kv_store, namespace, secondary_namespace).await.unwrap(); + assert_eq!(keys, vec![stats.id.to_string()]); + let bytes = + KVStore::read(&*kv_store, namespace, secondary_namespace, &keys[0]).await.unwrap(); + assert_eq!(ChannelPairForwardingStats::read(&mut &bytes[..]).unwrap(), stats); + } + + #[tokio::test] + async fn uncached_data_store_reads_existing_objects_across_pages() { + let kv_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(Logger::new_log_facade()); + let namespace = "paged_forwarded_payments"; + let store = ForwardedPaymentStore::new( + Vec::new(), + KeepNoEntries, + namespace.to_string(), + "details".to_string(), + Arc::clone(&kv_store), + Arc::clone(&logger), + ); + for id in 0..=50 { + store.insert(forwarded_payment(id, id as u64, 110, 100, 10)).await.unwrap(); + } + assert_eq!(store.cached_len(), 0); + + // Recreate the store to verify it has no in-memory state to preload. + let reopened_store = ForwardedPaymentStore::new( + Vec::new(), + KeepNoEntries, + namespace.to_string(), + "details".to_string(), + kv_store, + logger, + ); + let oldest_payment = forwarded_payment(0, 0, 110, 100, 10); + assert_eq!( + reopened_store.get(&oldest_payment.id()).await.unwrap(), + Some(oldest_payment.clone()) + ); + assert_eq!(reopened_store.cached_len(), 0); + + let first_page = reopened_store.list_page(None).await.unwrap(); + assert_eq!(first_page.objects.len(), 50); + let second_page = reopened_store.list_page(first_page.next_page_token).await.unwrap(); + assert_eq!(second_page.objects, vec![oldest_payment.clone()]); + assert!(second_page.next_page_token.is_none()); + } +} diff --git a/src/payment/mod.rs b/src/payment/mod.rs index 13dbe5106..e1c60da79 100644 --- a/src/payment/mod.rs +++ b/src/payment/mod.rs @@ -10,6 +10,8 @@ pub(crate) mod asynchronous; mod bolt11; mod bolt12; +mod forwarding; +pub(crate) mod forwarding_store; #[cfg(feature = "unified-payments")] mod hrn; mod onchain; @@ -22,6 +24,11 @@ mod unified; pub use bolt11::Bolt11Payment; pub(crate) use bolt11::PaymentMetadata; pub use bolt12::{Bolt12Payment, PayerProofOptions}; +pub use forwarding::{ + ChannelForwardingStats, ChannelForwardingStatsPage, ChannelPairForwardingStats, + ChannelPairForwardingStatsId, ChannelPairForwardingStatsPage, ForwardedPaymentDetails, + ForwardedPaymentDetailsPage, ForwardedPaymentId, ForwardingAnalytics, +}; #[cfg(feature = "unified-payments")] pub(crate) use hrn::HRNResolver; pub use onchain::OnchainPayment; diff --git a/src/types.rs b/src/types.rs index 1a61daa10..fd86d1bcd 100644 --- a/src/types.rs +++ b/src/types.rs @@ -37,12 +37,14 @@ use lightning_types::features::ChannelTypeFeatures; use crate::chain::ChainSource; use crate::config::{AnchorChannelsConfig, ChannelConfig}; -use crate::data_store::{DataStore, KeepAllEntries, KeepLeastRecentlyUsed}; +use crate::data_store::{DataStore, KeepAllEntries, KeepLeastRecentlyUsed, KeepNoEntries}; use crate::fee_estimator::OnchainFeeEstimator; use crate::ffi::maybe_wrap; use crate::logger::Logger; use crate::message_handler::NodeCustomMessageHandler; -use crate::payment::{PaymentDetails, PendingPaymentDetails}; +use crate::payment::{ + ChannelPairForwardingStats, ForwardedPaymentDetails, PaymentDetails, PendingPaymentDetails, +}; use crate::runtime::RuntimeSpawner; #[cfg(feature = "uniffi")] @@ -333,6 +335,12 @@ pub(crate) type BumpTransactionEventHandler = >; pub(crate) type PaymentStore = DataStore, KeepLeastRecentlyUsed>; +pub(crate) type ForwardedPaymentStore = + DataStore, KeepNoEntries>; +pub(crate) type ChannelForwardingStatsStore = + DataStore>; +pub(crate) type ChannelPairForwardingStatsStore = + DataStore, KeepNoEntries>; /// A local, potentially user-provided, identifier of a channel. /// diff --git a/tests/integration_tests_rust.rs b/tests/integration_tests_rust.rs index 0dad32d6a..457ed2309 100644 --- a/tests/integration_tests_rust.rs +++ b/tests/integration_tests_rust.rs @@ -35,13 +35,14 @@ use common::{ use electrsd::corepc_node::{self, Node as BitcoinD}; use electrsd::ElectrsD; use ldk_node::config::{ - AsyncPaymentsRole, EsploraSyncConfig, ADDRESS_POOL_SIZE, DEFAULT_FULL_SCAN_STOP_GAP, + AsyncPaymentsRole, EsploraSyncConfig, ForwardedPaymentTrackingMode, ADDRESS_POOL_SIZE, + DEFAULT_FULL_SCAN_STOP_GAP, }; use ldk_node::entropy::NodeEntropy; use ldk_node::liquidity::LSPS2ServiceConfig; use ldk_node::payment::{ - ConfirmationStatus, PayerProofOptions, PaymentDetails, PaymentDirection, PaymentKind, - PaymentStatus, TransactionType, UnifiedPaymentResult, + ConfirmationStatus, ForwardedPaymentId, PayerProofOptions, PaymentDetails, PaymentDirection, + PaymentKind, PaymentStatus, TransactionType, UnifiedPaymentResult, }; use ldk_node::{BuildError, Builder, Event, Node, NodeError, ReserveType}; use lightning::ln::channelmanager::PaymentId; @@ -731,6 +732,152 @@ async fn multi_hop_sending() { expect_payment_received_event!(&nodes[4], 2_500_000); let fee_paid_msat = Some(2000); expect_payment_successful_event!(nodes[0], outbound_payment_id, Some(fee_paid_msat)); + + // N1 forwarded the payment, so it records the forward against both of its channels: the one it + // received on and the one it sent on. N0 only sent, so it records nothing. + let forwarding = nodes[1].forwarding_analytics(); + assert_eq!(forwarding.tracking_mode(), ForwardedPaymentTrackingMode::Stats); + + let inbound_channel_id = nodes[1] + .list_channels() + .iter() + .find(|c| c.counterparty.node_id == nodes[0].node_id()) + .unwrap() + .channel_id; + let inbound_stats = forwarding.channel_stats(&inbound_channel_id).unwrap().unwrap(); + assert_eq!(inbound_stats.inbound_payments_forwarded, 1); + assert_eq!(inbound_stats.outbound_payments_forwarded, 0); + assert_eq!(inbound_stats.counterparty_node_id, Some(nodes[0].node_id())); + // Fees are attributed to the incoming channel, so all of N1's fee lands here. + let fee_earned_msat = inbound_stats.total_fee_earned_msat.unwrap(); + assert!(fee_earned_msat > 0); + assert_eq!(inbound_stats.total_outbound_amount_msat, 0); + + // Exactly one of N1's two outgoing channels carried the forward, since the payment took either + // the N2 or the N3 route. + let outbound_stats = forwarding + .list_channel_stats(None) + .unwrap() + .stats + .into_iter() + .filter(|s| s.outbound_payments_forwarded > 0) + .collect::>(); + assert_eq!(outbound_stats.len(), 1); + assert_eq!(outbound_stats[0].outbound_payments_forwarded, 1); + assert_eq!(outbound_stats[0].inbound_payments_forwarded, 0); + assert_ne!(outbound_stats[0].channel_id, inbound_channel_id); + + // N1 sends on more than the recipient gets, because the next hop takes a fee of its own too. + // What N1 received is what it sent on plus the fee it kept. + assert!(outbound_stats[0].total_outbound_amount_msat > 2_500_000); + assert_eq!( + inbound_stats.total_inbound_amount_msat, + outbound_stats[0].total_outbound_amount_msat + fee_earned_msat + ); + + // `Stats` is the default mode, so no per-payment detail is kept. + assert!(forwarding.list_payments(None).unwrap().payments.is_empty()); + + assert!(nodes[0].forwarding_analytics().list_channel_stats(None).unwrap().stats.is_empty()); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn detailed_forwarded_payment_tracking() { + let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); + let chain_source = random_chain_source(&bitcoind, &electrsd); + let node_a = setup_node(&chain_source, random_config()); + let mut router_config = random_config(); + router_config.node_config.forwarded_payment_tracking_mode = + ForwardedPaymentTrackingMode::Detailed; + let node_b = setup_node(&chain_source, router_config); + let node_c = setup_node(&chain_source, random_config()); + + let addr_a = node_a.onchain_payment().new_address().unwrap(); + let addr_b = node_b.onchain_payment().new_address().unwrap(); + let addr_c = node_c.onchain_payment().new_address().unwrap(); + let premine_amount_sat = 5_000_000; + premine_and_distribute_funds( + &bitcoind.client, + &electrsd.client, + vec![addr_a, addr_b, addr_c], + Amount::from_sat(premine_amount_sat), + ) + .await; + + for node in [&node_a, &node_b, &node_c] { + node.sync_wallets().unwrap(); + } + + // A -> B -> C, announced so that A can find the route through B. + open_channel(&node_a, &node_b, 1_000_000, true, &electrsd).await; + open_channel(&node_b, &node_c, 1_000_000, true, &electrsd).await; + + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + + for node in [&node_a, &node_b, &node_c] { + node.sync_wallets().unwrap(); + } + + expect_channel_ready_event!(node_a, node_b.node_id()); + expect_channel_ready_events!(node_b, node_a.node_id(), node_c.node_id()); + expect_channel_ready_event!(node_c, node_b.node_id()); + + // Sleep a bit for gossip to propagate. + tokio::time::sleep(Duration::from_secs(1)).await; + + let forwarding = node_b.forwarding_analytics(); + assert_eq!(forwarding.tracking_mode(), ForwardedPaymentTrackingMode::Detailed); + + let amount_msat = 2_500_000; + let invoice_description = + Bolt11InvoiceDescription::Direct(Description::new(String::from("detailed")).unwrap()); + let invoice = + node_c.bolt11_payment().receive(amount_msat, &invoice_description.into(), 3600).unwrap(); + let payment_id = node_a.bolt11_payment().send(&invoice, None).unwrap(); + + expect_event!(node_b, PaymentForwarded); + expect_payment_received_event!(node_c, amount_msat); + expect_payment_successful_event!(node_a, payment_id, None); + + // In `Detailed` mode the forward is kept as an individual record until its bucket closes. + let page = forwarding.list_payments(None).unwrap(); + assert_eq!(page.payments.len(), 1); + assert!(page.next_page_token.is_none()); + let details = &page.payments[0]; + + let inbound_channel_id = node_b + .list_channels() + .iter() + .find(|c| c.counterparty.node_id == node_a.node_id()) + .unwrap() + .channel_id; + let outbound_channel_id = node_b + .list_channels() + .iter() + .find(|c| c.counterparty.node_id == node_c.node_id()) + .unwrap() + .channel_id; + assert_eq!(details.prev_channel_id, inbound_channel_id); + assert_eq!(details.next_channel_id, outbound_channel_id); + assert_eq!(details.prev_node_id, Some(node_a.node_id())); + assert_eq!(details.next_node_id, Some(node_c.node_id())); + assert_eq!(details.outbound_amount_forwarded_msat, Some(amount_msat)); + assert!(!details.claim_from_onchain_tx); + + let fee_earned_msat = details.total_fee_earned_msat.unwrap(); + assert_eq!(details.inbound_amount_forwarded_msat, Some(amount_msat + fee_earned_msat)); + + // The opaque id round-trips through the single-payment lookup. + assert_eq!(forwarding.payment(&details.id).unwrap().as_ref(), Some(details)); + assert!(forwarding.payment(&ForwardedPaymentId([0; 32])).unwrap().is_none()); + + // Channel statistics are recorded in both modes, so they agree with the detail record. + let inbound_stats = forwarding.channel_stats(&inbound_channel_id).unwrap().unwrap(); + assert_eq!(inbound_stats.inbound_payments_forwarded, 1); + assert_eq!(inbound_stats.total_fee_earned_msat, Some(fee_earned_msat)); + let outbound_stats = forwarding.channel_stats(&outbound_channel_id).unwrap().unwrap(); + assert_eq!(outbound_stats.outbound_payments_forwarded, 1); + assert_eq!(outbound_stats.total_outbound_amount_msat, amount_msat); } #[tokio::test(flavor = "multi_thread", worker_threads = 1)]