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() {