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.

Co-Authored-By: HAL 9000

Author: tnull

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