From 3445447c67e385ad74d2069b5f2b65b142e105be Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Wed, 16 Sep 2026 12:21:07 +0200 Subject: [PATCH 1/2] Drop rate-limited static invoice requests Drop static invoice requests when the rate limit is reached instead of retrying them and delaying later events. Keep retrying storage errors and only acknowledge invoices after saving them. Fixes an issue reported by Ajay Sehwal. Co-Authored-By: HAL 9000 --- src/event.rs | 16 +- .../asynchronous/static_invoice_store.rs | 227 ++++++++++++++++-- 2 files changed, 226 insertions(+), 17 deletions(-) diff --git a/src/event.rs b/src/event.rs index 657c938ed6..88520b3e53 100644 --- a/src/event.rs +++ b/src/event.rs @@ -50,7 +50,9 @@ use crate::io::{ 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::asynchronous::static_invoice_store::{ + StaticInvoiceStore, StaticInvoiceStoreError, +}; use crate::payment::forwarding_store::{ForwardRecord, ForwardingStore}; use crate::payment::store::{ Channel, PaymentDetails, PaymentDetailsUpdate, PaymentDirection, PaymentKind, PaymentStatus, @@ -2518,7 +2520,11 @@ where Ok(_) => { self.channel_manager.static_invoice_persisted(invoice_persisted_path); }, - Err(e) => { + Err(StaticInvoiceStoreError::RateLimited) => { + // Drop silently: logging each rejected request, even at trace level, + // can cause excessive log output and I/O under sustained load. + }, + Err(StaticInvoiceStoreError::Io(e)) => { log_error!(self.logger, "Failed to persist static invoice: {}", e); return Err(ReplayEvent()); }, @@ -2554,7 +2560,11 @@ where invoice_slot ); }, - Err(e) => { + Err(StaticInvoiceStoreError::RateLimited) => { + // Drop silently: logging each rejected request, even at trace level, + // can cause excessive log output and I/O under sustained load. + }, + Err(StaticInvoiceStoreError::Io(e)) => { log_error!(self.logger, "Failed to retrieve static invoice: {}", e); return Err(ReplayEvent()); }, diff --git a/src/payment/asynchronous/static_invoice_store.rs b/src/payment/asynchronous/static_invoice_store.rs index f1e2378c23..c6b3648a18 100644 --- a/src/payment/asynchronous/static_invoice_store.rs +++ b/src/payment/asynchronous/static_invoice_store.rs @@ -33,6 +33,12 @@ impl_writeable_tlv_based!(PersistedStaticInvoice, { (2, request_path, required) }); +#[derive(Debug)] +pub(crate) enum StaticInvoiceStoreError { + RateLimited, + Io(lightning::io::Error), +} + pub(crate) struct StaticInvoiceStore { kv_store: Arc, request_rate_limiter: Mutex, @@ -62,10 +68,10 @@ impl StaticInvoiceStore { fn check_rate_limit( limiter: &Mutex, recipient_id: &[u8], - ) -> Result<(), lightning::io::Error> { + ) -> Result<(), StaticInvoiceStoreError> { let mut limiter = limiter.lock().expect("lock"); if !limiter.allow(recipient_id) { - Err(lightning::io::Error::new(lightning::io::ErrorKind::Other, "Rate limit exceeded")) + Err(StaticInvoiceStoreError::RateLimited) } else { Ok(()) } @@ -73,7 +79,7 @@ impl StaticInvoiceStore { pub(crate) async fn handle_static_invoice_requested( &self, recipient_id: &[u8], invoice_slot: u16, - ) -> Result, lightning::io::Error> { + ) -> Result, StaticInvoiceStoreError> { Self::check_rate_limit(&self.request_rate_limiter, &recipient_id)?; let (secondary_namespace, key) = Self::get_storage_location(invoice_slot, recipient_id); @@ -97,21 +103,14 @@ impl StaticInvoiceStore { ) }) }) - .or_else( - |e| { - if e.kind() == lightning::io::ErrorKind::NotFound { - Ok(None) - } else { - Err(e) - } - }, - ) + .or_else(|e| if e.kind() == lightning::io::ErrorKind::NotFound { Ok(None) } else { Err(e) }) + .map_err(StaticInvoiceStoreError::Io) } pub(crate) async fn handle_persist_static_invoice( &self, invoice: StaticInvoice, invoice_request_path: BlindedMessagePath, invoice_slot: u16, recipient_id: Vec, - ) -> Result<(), lightning::io::Error> { + ) -> Result<(), StaticInvoiceStoreError> { Self::check_rate_limit(&self.persist_rate_limiter, &recipient_id)?; let (secondary_namespace, key) = Self::get_storage_location(invoice_slot, &recipient_id); @@ -120,7 +119,7 @@ impl StaticInvoiceStore { PersistedStaticInvoice { invoice, request_path: invoice_request_path }; let mut buf = Vec::new(); - persisted_invoice.write(&mut buf)?; + persisted_invoice.write(&mut buf).map_err(StaticInvoiceStoreError::Io)?; // Static invoices will be persisted at "static_invoices//". // @@ -133,6 +132,7 @@ impl StaticInvoiceStore { buf, ) .await + .map_err(StaticInvoiceStoreError::Io) } fn get_storage_location(invoice_slot: u16, recipient_id: &[u8]) -> (String, String) { @@ -146,6 +146,7 @@ impl StaticInvoiceStore { #[cfg(test)] mod tests { + use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::time::Duration; @@ -154,16 +155,214 @@ mod tests { use lightning::blinded_path::message::BlindedMessagePath; use lightning::blinded_path::payment::{BlindedPayInfo, BlindedPaymentPath}; use lightning::blinded_path::BlindedHop; + use lightning::events::Event; + use lightning::ln::channelmanager::PaymentId; use lightning::ln::inbound_payment::ExpandedKey; use lightning::offers::nonce::Nonce; use lightning::offers::offer::OfferBuilder; use lightning::offers::static_invoice::{StaticInvoice, StaticInvoiceBuilder}; + use lightning::onion_message::async_payments::{ + AsyncPaymentsMessage, AsyncPaymentsMessageHandler, + }; + use lightning::onion_message::messenger::Responder; + use lightning::onion_message::offers::OffersMessageHandler; use lightning::sign::EntropySource; + use lightning::util::persist::KVStore; + use lightning::util::ser::{Readable, Writeable}; + use lightning::util::wallet_utils::Wallet as LdkWallet; use lightning_types::features::BlindedHopFeatures; + use crate::builder::NodeBuilder; + use crate::entropy::NodeEntropy; + use crate::event::EventHandler; use crate::io::test_utils::InMemoryStore; + use crate::io::STATIC_INVOICE_STORE_PRIMARY_NAMESPACE; + use crate::logger::{LogRecord, LogWriter, Logger}; + use crate::payment::asynchronous::rate_limiter::RateLimiter; use crate::payment::asynchronous::static_invoice_store::StaticInvoiceStore; use crate::types::{DynStore, DynStoreWrapper}; + use crate::{BumpTransactionEventHandler, Node}; + + #[derive(Default)] + struct TestLogWriter { + logged: AtomicBool, + } + + impl LogWriter for TestLogWriter { + fn log(&self, _record: LogRecord) { + self.logged.store(true, Ordering::Relaxed); + } + } + + fn event_handler( + store: StaticInvoiceStore, + ) -> (Node, EventHandler>, Arc) { + let mut builder = NodeBuilder::new(); + builder.set_log_facade_logger(); + #[cfg(not(feature = "uniffi"))] + let entropy = NodeEntropy::from_seed_bytes([42; 64]); + #[cfg(feature = "uniffi")] + let entropy = NodeEntropy::from_seed_bytes(vec![42; 64]).unwrap(); + let node = builder.build_with_store(entropy, InMemoryStore::new()).unwrap(); + let bump_handler = Arc::new(BumpTransactionEventHandler::new( + Arc::clone(&node.tx_broadcaster), + Arc::new(LdkWallet::new(Arc::clone(&node.wallet), Arc::clone(&node.logger))), + Arc::clone(&node.keys_manager), + Arc::clone(&node.logger), + )); + let log_writer = Arc::new(TestLogWriter::default()); + let handler = EventHandler::new( + Arc::clone(&node.event_queue), + Arc::clone(&node.wallet), + bump_handler, + Arc::clone(&node.channel_manager), + Arc::clone(&node.connection_manager), + Arc::clone(&node.output_sweeper), + Arc::clone(&node.network_graph), + Arc::clone(&node.liquidity_source), + Arc::clone(&node.payment_store), + Arc::clone(&node.forwarding_store), + Arc::clone(&node.peer_store), + Arc::clone(&node.keys_manager), + Some(store), + Arc::clone(&node.onion_messenger), + None, + None, + Arc::clone(&node.runtime), + Arc::new(Logger::new_custom_writer(log_writer.clone())), + Arc::clone(&node.config), + ); + (node, handler, log_writer) + } + + fn responder() -> Responder { + // Responder has no public constructor, so use its serialized representation. + struct ReplyPath { + path: BlindedMessagePath, + } + lightning::impl_writeable_tlv_based!(ReplyPath, { (0, path, required) }); + let bytes = ReplyPath { path: blinded_path() }.encode(); + Responder::read(&mut &bytes[..]).unwrap() + } + + fn static_invoice_event(persist: bool) -> Event { + if persist { + Event::PersistStaticInvoice { + invoice: invoice(), + invoice_request_path: blinded_path(), + invoice_slot: 1, + recipient_id: vec![1, 1, 1], + invoice_persisted_path: responder(), + } + } else { + let invoice_request = OfferBuilder::new(recipient_pubkey()) + .amount_msats(1_000) + .build() + .unwrap() + .request_invoice( + &ExpandedKey::new([42; 32]), + Nonce::from_entropy_source(&FixedEntropy {}), + &Secp256k1::new(), + PaymentId([42; 32]), + ) + .unwrap() + .build_and_sign() + .unwrap(); + Event::StaticInvoiceRequested { + recipient_id: vec![1, 1, 1], + invoice_slot: 0, + reply_path: responder(), + invoice_request, + } + } + } + + async fn check_rate_limited_static_invoice_event(persist: bool) { + let kv_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let mut store = StaticInvoiceStore::new(Arc::clone(&kv_store)); + store + .handle_persist_static_invoice(invoice(), blinded_path(), 0, vec![1, 1, 1]) + .await + .unwrap(); + // Reject requests deterministically, without depending on refill timing. + *store.request_rate_limiter.get_mut().unwrap() = + RateLimiter::new(0, Duration::from_secs(1), Duration::from_secs(600)); + *store.persist_rate_limiter.get_mut().unwrap() = + RateLimiter::new(0, Duration::from_secs(1), Duration::from_secs(600)); + let (node, handler, log_writer) = event_handler(store); + node.channel_manager.push_pending_event(static_invoice_event(persist)); + node.channel_manager.push_pending_event(Event::PaymentFailed { + payment_id: PaymentId([42; 32]), + payment_hash: None, + reason: None, + }); + let handled_next_event = AtomicBool::new(false); + node.channel_manager + .process_pending_events_async(|event| async { + if matches!(event, Event::PaymentFailed { .. }) { + handled_next_event.store(true, Ordering::Relaxed); + Ok(()) + } else { + handler.handle_event(event).await + } + }) + .await; + assert!( + handled_next_event.load(Ordering::Relaxed), + "a rate-limited static invoice event must not delay the next event" + ); + assert!( + !log_writer.logged.load(Ordering::Relaxed), + "a rate-limited static invoice event must not produce a log message" + ); + assert!(AsyncPaymentsMessageHandler::release_pending_messages(&*node.channel_manager) + .is_empty()); + assert!(OffersMessageHandler::release_pending_messages(&*node.channel_manager).is_empty()); + let store = StaticInvoiceStore::new(kv_store); + assert!(store.handle_static_invoice_requested(&[1, 1, 1], 0).await.unwrap().is_some()); + assert!(store.handle_static_invoice_requested(&[1, 1, 1], 1).await.unwrap().is_none()); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn rate_limited_static_invoice_persistence_does_not_replay() { + check_rate_limited_static_invoice_event(true).await; + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn rate_limited_static_invoice_request_does_not_replay() { + check_rate_limited_static_invoice_event(false).await; + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn static_invoice_persistence_is_acknowledged() { + let kv_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let (node, handler, _) = event_handler(StaticInvoiceStore::new(Arc::clone(&kv_store))); + handler.handle_event(static_invoice_event(true)).await.unwrap(); + let messages = + AsyncPaymentsMessageHandler::release_pending_messages(&*node.channel_manager); + assert_eq!(messages.len(), 1); + assert!(matches!(messages[0].0, AsyncPaymentsMessage::StaticInvoicePersisted(_))); + let store = StaticInvoiceStore::new(kv_store); + assert!(store.handle_static_invoice_requested(&[1, 1, 1], 1).await.unwrap().is_some()); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn static_invoice_read_error_is_replayed() { + let kv_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let (namespace, key) = StaticInvoiceStore::get_storage_location(0, &[1, 1, 1]); + KVStore::write( + &*kv_store, + STATIC_INVOICE_STORE_PRIMARY_NAMESPACE, + &namespace, + &key, + vec![0xff], + ) + .await + .unwrap(); + let (_node, handler, log_writer) = event_handler(StaticInvoiceStore::new(kv_store)); + assert!(handler.handle_event(static_invoice_event(false)).await.is_err()); + assert!(log_writer.logged.load(Ordering::Relaxed)); + } #[tokio::test] async fn static_invoice_store_test() { From 0033e6b1757a9e16f784ded5bf34eff20fdd5930 Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Fri, 16 Oct 2026 15:08:13 +0200 Subject: [PATCH 2/2] f Preserve admission across static invoice retries Storage retries consumed another rate-limit token on every attempt. Repeated failures could exhaust the bucket and silently drop an event that had already been admitted. Preserve admission across storage failures so requests can recover without losing their place in the event queue. Successful operations still consume quota, and regression tests cover retry recovery as well as silent rejection and later admission after refill. Co-Authored-By: HAL 9000 --- src/payment/asynchronous/rate_limiter.rs | 14 + .../asynchronous/static_invoice_store.rs | 281 ++++++++++++++---- 2 files changed, 229 insertions(+), 66 deletions(-) diff --git a/src/payment/asynchronous/rate_limiter.rs b/src/payment/asynchronous/rate_limiter.rs index bf12508927..47438cd1a3 100644 --- a/src/payment/asynchronous/rate_limiter.rs +++ b/src/payment/asynchronous/rate_limiter.rs @@ -70,10 +70,24 @@ impl RateLimiter { allow } + /// Returns a token consumed by an admitted request that could not be completed. + pub(crate) fn refund(&mut self, user_id: &[u8]) { + if let Some(bucket) = self.users.get_mut(user_id) { + bucket.tokens = bucket.tokens.saturating_add(1).min(self.capacity); + } + } + fn garbage_collect(&mut self, max_idle: Duration) { let now = Instant::now(); self.users.retain(|_, bucket| now.duration_since(bucket.last_refill) < max_idle); } + + #[cfg(test)] + pub(super) fn advance_refill_interval(&mut self) { + for bucket in self.users.values_mut() { + bucket.last_refill -= self.refill_interval; + } + } } #[cfg(test)] diff --git a/src/payment/asynchronous/static_invoice_store.rs b/src/payment/asynchronous/static_invoice_store.rs index c6b3648a18..74643bcda1 100644 --- a/src/payment/asynchronous/static_invoice_store.rs +++ b/src/payment/asynchronous/static_invoice_store.rs @@ -77,6 +77,16 @@ impl StaticInvoiceStore { } } + fn refund_on_io_error( + limiter: &Mutex, recipient_id: &[u8], error: lightning::io::Error, + ) -> StaticInvoiceStoreError { + // ChannelManager processes these events serially and replays a failed event before handling + // later events. Returning its token keeps retries admitted until storage recovers. If these + // operations are ever processed concurrently, admission must be tracked per request instead. + limiter.lock().expect("lock").refund(recipient_id); + StaticInvoiceStoreError::Io(error) + } + pub(crate) async fn handle_static_invoice_requested( &self, recipient_id: &[u8], invoice_slot: u16, ) -> Result, StaticInvoiceStoreError> { @@ -104,7 +114,7 @@ impl StaticInvoiceStore { }) }) .or_else(|e| if e.kind() == lightning::io::ErrorKind::NotFound { Ok(None) } else { Err(e) }) - .map_err(StaticInvoiceStoreError::Io) + .map_err(|e| Self::refund_on_io_error(&self.request_rate_limiter, recipient_id, e)) } pub(crate) async fn handle_persist_static_invoice( @@ -119,7 +129,9 @@ impl StaticInvoiceStore { PersistedStaticInvoice { invoice, request_path: invoice_request_path }; let mut buf = Vec::new(); - persisted_invoice.write(&mut buf).map_err(StaticInvoiceStoreError::Io)?; + persisted_invoice + .write(&mut buf) + .map_err(|e| Self::refund_on_io_error(&self.persist_rate_limiter, &recipient_id, e))?; // Static invoices will be persisted at "static_invoices//". // @@ -132,7 +144,7 @@ impl StaticInvoiceStore { buf, ) .await - .map_err(StaticInvoiceStoreError::Io) + .map_err(|e| Self::refund_on_io_error(&self.persist_rate_limiter, &recipient_id, e)) } fn get_storage_location(invoice_slot: u16, recipient_id: &[u8]) -> (String, String) { @@ -146,7 +158,8 @@ impl StaticInvoiceStore { #[cfg(test)] mod tests { - use std::sync::atomic::{AtomicBool, Ordering}; + use std::future::Future; + use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::Arc; use std::time::Duration; @@ -156,6 +169,7 @@ mod tests { use lightning::blinded_path::payment::{BlindedPayInfo, BlindedPaymentPath}; use lightning::blinded_path::BlindedHop; use lightning::events::Event; + use lightning::io; use lightning::ln::channelmanager::PaymentId; use lightning::ln::inbound_payment::ExpandedKey; use lightning::offers::nonce::Nonce; @@ -167,7 +181,7 @@ mod tests { use lightning::onion_message::messenger::Responder; use lightning::onion_message::offers::OffersMessageHandler; use lightning::sign::EntropySource; - use lightning::util::persist::KVStore; + use lightning::util::persist::{KVStore, PageToken, PaginatedKVStore, PaginatedListResponse}; use lightning::util::ser::{Readable, Writeable}; use lightning::util::wallet_utils::Wallet as LdkWallet; use lightning_types::features::BlindedHopFeatures; @@ -179,7 +193,9 @@ mod tests { use crate::io::STATIC_INVOICE_STORE_PRIMARY_NAMESPACE; use crate::logger::{LogRecord, LogWriter, Logger}; use crate::payment::asynchronous::rate_limiter::RateLimiter; - use crate::payment::asynchronous::static_invoice_store::StaticInvoiceStore; + use crate::payment::asynchronous::static_invoice_store::{ + StaticInvoiceStore, StaticInvoiceStoreError, + }; use crate::types::{DynStore, DynStoreWrapper}; use crate::{BumpTransactionEventHandler, Node}; @@ -194,6 +210,60 @@ mod tests { } } + struct FailingStore { + inner: InMemoryStore, + fail: Arc, + attempts: Arc, + } + + impl FailingStore { + fn check_failure(&self) -> Result<(), io::Error> { + self.attempts.fetch_add(1, Ordering::Relaxed); + if self.fail.load(Ordering::Relaxed) { + Err(io::Error::new(io::ErrorKind::Other, "injected storage failure")) + } else { + Ok(()) + } + } + } + + impl KVStore for FailingStore { + fn read( + &self, primary: &str, secondary: &str, key: &str, + ) -> impl Future, io::Error>> + Send + 'static { + let result = self.check_failure().map(|()| self.inner.read(primary, secondary, key)); + async move { result?.await } + } + + fn write( + &self, primary: &str, secondary: &str, key: &str, buf: Vec, + ) -> impl Future> + Send + 'static { + let result = + self.check_failure().map(|()| self.inner.write(primary, secondary, key, buf)); + async move { result?.await } + } + + fn remove( + &self, primary: &str, secondary: &str, key: &str, lazy: bool, + ) -> impl Future> + Send + 'static { + self.inner.remove(primary, secondary, key, lazy) + } + + fn list( + &self, primary: &str, secondary: &str, + ) -> impl Future, io::Error>> + Send + 'static { + self.inner.list(primary, secondary) + } + } + + impl PaginatedKVStore for FailingStore { + fn list_paginated( + &self, primary: &str, secondary: &str, page_token: Option, + ) -> impl Future> + Send + 'static { + self.inner.list_paginated(primary, secondary, page_token) + } + } + fn event_handler( store: StaticInvoiceStore, ) -> (Node, EventHandler>, Arc) { @@ -216,6 +286,7 @@ mod tests { Arc::clone(&node.wallet), bump_handler, Arc::clone(&node.channel_manager), + Arc::clone(&node.chain_monitor), Arc::clone(&node.connection_manager), Arc::clone(&node.output_sweeper), Arc::clone(&node.network_graph), @@ -231,10 +302,35 @@ mod tests { Arc::clone(&node.runtime), Arc::new(Logger::new_custom_writer(log_writer.clone())), Arc::clone(&node.config), + Arc::clone(&node.pending_funding_fee_rates), ); (node, handler, log_writer) } + async fn process_events(node: &Node, handler: &EventHandler>) -> bool { + let handled_next_event = AtomicBool::new(false); + node.channel_manager + .process_pending_events_async(|event| async { + if matches!(event, Event::PaymentFailed { .. }) { + handled_next_event.store(true, Ordering::Relaxed); + Ok(()) + } else { + handler.handle_event(event).await + } + }) + .await; + handled_next_event.load(Ordering::Relaxed) + } + + fn queue_events(node: &Node, event: Event) { + node.channel_manager.push_pending_event(event); + node.channel_manager.push_pending_event(Event::PaymentFailed { + payment_id: PaymentId([42; 32]), + payment_hash: None, + reason: None, + }); + } + fn responder() -> Responder { // Responder has no public constructor, so use its serialized representation. struct ReplyPath { @@ -277,97 +373,129 @@ mod tests { } } - async fn check_rate_limited_static_invoice_event(persist: bool) { - let kv_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + async fn check_static_invoice_storage_retry(persist: bool) { + let fail = Arc::new(AtomicBool::new(false)); + let attempts = Arc::new(AtomicUsize::new(0)); + let kv_store: Arc = Arc::new(DynStoreWrapper(FailingStore { + inner: InMemoryStore::new(), + fail: Arc::clone(&fail), + attempts: Arc::clone(&attempts), + })); let mut store = StaticInvoiceStore::new(Arc::clone(&kv_store)); + let stored_invoice = invoice(); store - .handle_persist_static_invoice(invoice(), blinded_path(), 0, vec![1, 1, 1]) + .handle_persist_static_invoice(stored_invoice.clone(), blinded_path(), 0, vec![1, 1, 1]) .await .unwrap(); - // Reject requests deterministically, without depending on refill timing. - *store.request_rate_limiter.get_mut().unwrap() = - RateLimiter::new(0, Duration::from_secs(1), Duration::from_secs(600)); - *store.persist_rate_limiter.get_mut().unwrap() = - RateLimiter::new(0, Duration::from_secs(1), Duration::from_secs(600)); + // One admission and no refills during the test: retries must retain their quota. + for limiter in [&mut store.request_rate_limiter, &mut store.persist_rate_limiter] { + *limiter.get_mut().unwrap() = + RateLimiter::new(1, Duration::from_secs(3600), Duration::from_secs(7200)); + } let (node, handler, log_writer) = event_handler(store); - node.channel_manager.push_pending_event(static_invoice_event(persist)); - node.channel_manager.push_pending_event(Event::PaymentFailed { - payment_id: PaymentId([42; 32]), - payment_hash: None, - reason: None, - }); - let handled_next_event = AtomicBool::new(false); - node.channel_manager - .process_pending_events_async(|event| async { - if matches!(event, Event::PaymentFailed { .. }) { - handled_next_event.store(true, Ordering::Relaxed); - Ok(()) - } else { - handler.handle_event(event).await - } - }) - .await; + let event = static_invoice_event(persist); + let expected_invoice = match &event { + Event::PersistStaticInvoice { invoice, .. } => invoice.clone(), + _ => stored_invoice, + }; + queue_events(&node, event); + fail.store(true, Ordering::Relaxed); + attempts.store(0, Ordering::Relaxed); + for attempt in 1..=3 { + assert!( + !process_events(&node, &handler).await, + "a storage failure must keep replaying the event, including attempt {attempt}" + ); + assert_eq!(attempts.load(Ordering::Relaxed), attempt); + assert!(AsyncPaymentsMessageHandler::release_pending_messages(&*node.channel_manager) + .is_empty()); + assert!( + OffersMessageHandler::release_pending_messages(&*node.channel_manager).is_empty() + ); + } + assert!(log_writer.logged.load(Ordering::Relaxed)); + + fail.store(false, Ordering::Relaxed); + assert!(process_events(&node, &handler).await); + assert_eq!(attempts.load(Ordering::Relaxed), 4); + let messages = + AsyncPaymentsMessageHandler::release_pending_messages(&*node.channel_manager); + if persist { + assert_eq!(messages.len(), 1); + assert!(matches!(messages[0].0, AsyncPaymentsMessage::StaticInvoicePersisted(_))); + } else { + assert!(messages.is_empty()); + } + OffersMessageHandler::release_pending_messages(&*node.channel_manager); + + // Once the event succeeds, a new request must consume a new admission. + log_writer.logged.store(false, Ordering::Relaxed); + queue_events(&node, static_invoice_event(persist)); assert!( - handled_next_event.load(Ordering::Relaxed), + process_events(&node, &handler).await, "a rate-limited static invoice event must not delay the next event" ); assert!( !log_writer.logged.load(Ordering::Relaxed), "a rate-limited static invoice event must not produce a log message" ); + assert_eq!( + attempts.load(Ordering::Relaxed), + 4, + "a rate-limited request must not access storage" + ); assert!(AsyncPaymentsMessageHandler::release_pending_messages(&*node.channel_manager) .is_empty()); assert!(OffersMessageHandler::release_pending_messages(&*node.channel_manager).is_empty()); let store = StaticInvoiceStore::new(kv_store); - assert!(store.handle_static_invoice_requested(&[1, 1, 1], 0).await.unwrap().is_some()); - assert!(store.handle_static_invoice_requested(&[1, 1, 1], 1).await.unwrap().is_none()); + let stored = + store.handle_static_invoice_requested(&[1, 1, 1], u16::from(persist)).await.unwrap(); + assert_eq!(stored, Some((expected_invoice, blinded_path()))); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - async fn rate_limited_static_invoice_persistence_does_not_replay() { - check_rate_limited_static_invoice_event(true).await; + async fn static_invoice_persistence_retries_until_recovery() { + check_static_invoice_storage_retry(true).await; } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - async fn rate_limited_static_invoice_request_does_not_replay() { - check_rate_limited_static_invoice_event(false).await; + async fn static_invoice_read_retries_until_recovery() { + check_static_invoice_storage_retry(false).await; } - #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - async fn static_invoice_persistence_is_acknowledged() { - let kv_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); - let (node, handler, _) = event_handler(StaticInvoiceStore::new(Arc::clone(&kv_store))); - handler.handle_event(static_invoice_event(true)).await.unwrap(); - let messages = - AsyncPaymentsMessageHandler::release_pending_messages(&*node.channel_manager); - assert_eq!(messages.len(), 1); - assert!(matches!(messages[0].0, AsyncPaymentsMessage::StaticInvoicePersisted(_))); - let store = StaticInvoiceStore::new(kv_store); - assert!(store.handle_static_invoice_requested(&[1, 1, 1], 1).await.unwrap().is_some()); - } - - #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - async fn static_invoice_read_error_is_replayed() { + #[tokio::test] + async fn static_invoice_decode_error_retains_admission() { let kv_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let mut store = StaticInvoiceStore::new(Arc::clone(&kv_store)); + *store.request_rate_limiter.get_mut().unwrap() = + RateLimiter::new(1, Duration::from_secs(3600), Duration::from_secs(7200)); let (namespace, key) = StaticInvoiceStore::get_storage_location(0, &[1, 1, 1]); - KVStore::write( - &*kv_store, - STATIC_INVOICE_STORE_PRIMARY_NAMESPACE, - &namespace, - &key, - vec![0xff], - ) - .await - .unwrap(); - let (_node, handler, log_writer) = event_handler(StaticInvoiceStore::new(kv_store)); - assert!(handler.handle_event(static_invoice_event(false)).await.is_err()); - assert!(log_writer.logged.load(Ordering::Relaxed)); + kv_store + .write(STATIC_INVOICE_STORE_PRIMARY_NAMESPACE, &namespace, &key, vec![0xff]) + .await + .unwrap(); + for _ in 0..3 { + assert!( + matches!(store.handle_static_invoice_requested(&[1, 1, 1], 0).await, + Err(StaticInvoiceStoreError::Io(e)) if e.kind() == io::ErrorKind::InvalidData), + "a decode error must not exhaust the request's admission" + ); + } + let invoice = invoice(); + store + .handle_persist_static_invoice(invoice.clone(), blinded_path(), 0, vec![1, 1, 1]) + .await + .unwrap(); + assert_eq!( + store.handle_static_invoice_requested(&[1, 1, 1], 0).await.unwrap(), + Some((invoice, blinded_path())) + ); } #[tokio::test] async fn static_invoice_store_test() { let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); - let static_invoice_store = StaticInvoiceStore::new(Arc::clone(&store)); + let mut static_invoice_store = StaticInvoiceStore::new(Arc::clone(&store)); let static_invoice = invoice(); let recipient_id = vec![1, 1, 1]; @@ -398,6 +526,27 @@ mod tests { .await .unwrap() .is_none()); + + *static_invoice_store.request_rate_limiter.get_mut().unwrap() = + RateLimiter::new(1, Duration::from_secs(3600), Duration::from_secs(7200)); + // Missing invoices consume quota too, and rejected requests must not refund it. + assert!(static_invoice_store + .handle_static_invoice_requested(&recipient_id, 1) + .await + .unwrap() + .is_none()); + for _ in 0..2 { + assert!(matches!( + static_invoice_store.handle_static_invoice_requested(&recipient_id, 0).await, + Err(StaticInvoiceStoreError::RateLimited) + )); + } + static_invoice_store.request_rate_limiter.get_mut().unwrap().advance_refill_interval(); + assert!(static_invoice_store + .handle_static_invoice_requested(&recipient_id, 0) + .await + .unwrap() + .is_some()); } fn invoice() -> StaticInvoice {