From 264852e14a22069a0a33d4b8cb1f0b980d779c55 Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Wed, 16 Sep 2026 12:21:07 +0200 Subject: [PATCH] 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 728f625ab4..7348da2770 100644 --- a/src/event.rs +++ b/src/event.rs @@ -48,7 +48,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::{ PaymentDetails, PaymentDetailsUpdate, PaymentDirection, PaymentKind, PaymentStatus, @@ -2115,7 +2117,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()); }, @@ -2151,7 +2157,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() {