Drop rate-limited static invoice requests

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());
 						},

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<DynStore>,
 	request_rate_limiter: Mutex<RateLimiter>,
@@ -62,18 +68,18 @@ impl StaticInvoiceStore {
 
 	fn check_rate_limit(
 		limiter: &Mutex<RateLimiter>, 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(())
 		}
 	}
 
 	pub(crate) async fn handle_static_invoice_requested(
 		&self, recipient_id: &[u8], invoice_slot: u16,
-	) -> Result<Option<(StaticInvoice, BlindedMessagePath)>, lightning::io::Error> {
+	) -> Result<Option<(StaticInvoice, BlindedMessagePath)>, 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<u8>,
-	) -> 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/<sha256(recipient_id)>/<invoice_slot>".
 		//
@@ -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<Logger>>, Arc<TestLogWriter>) {
+		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<DynStore> = 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<DynStore> = 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<DynStore> = 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() {