f: review comments

Author: joostjager

src/builder.rs

@@ -27,7 +27,7 @@ use crate::liquidity::{
 };
 use crate::logger::{log_error, LdkLogger, LogLevel, LogWriter, Logger};
 use crate::message_handler::NodeCustomMessageHandler;
-use crate::om_mailbox::OnionMessageMailbox;
+use crate::payment::asynchronous::om_mailbox::OnionMessageMailbox;
 use crate::peer_store::PeerStore;
 use crate::runtime::Runtime;
 use crate::tx_broadcaster::TransactionBroadcaster;
@@ -1664,7 +1664,11 @@ fn build_with_store_internal(
 		},
 	};
 
-	let om_mailbox = Arc::new(OnionMessageMailbox::new());
+	let om_mailbox = if config.async_payment_services_enabled {
+		Some(Arc::new(OnionMessageMailbox::new()))
+	} else {
+		None
+	};
 
 	let (stop_sender, _) = tokio::sync::watch::channel(());
 	let (background_processor_stop_sender, _) = tokio::sync::watch::channel(());

src/config.rs

@@ -330,7 +330,9 @@ pub(crate) fn default_user_config(config: &Config) -> UserConfig {
 		user_config.channel_handshake_limits.force_announced_channel_preference = true;
 	}
 
-	user_config.enable_htlc_hold = true;
+	if config.async_payment_services_enabled {
+		user_config.enable_htlc_hold = true;
+	}
 
 	user_config
 }

src/event.rs

@@ -5,7 +5,7 @@
 // http://opensource.org/licenses/MIT>, at your option. You may not use this file except in
 // accordance with one or both of these licenses.
 
-use crate::om_mailbox::OnionMessageMailbox;
+use crate::payment::asynchronous::om_mailbox::OnionMessageMailbox;
 use crate::types::{CustomTlvRecord, DynStore, OnionMessenger, PaymentStore, Sweeper, Wallet};
 use crate::{
 	hex_utils, BumpTransactionEventHandler, ChannelManager, Error, Graph, PeerInfo, PeerStore,
@@ -461,7 +461,7 @@ where
 	config: Arc<Config>,
 	static_invoice_store: Option<StaticInvoiceStore>,
 	onion_messenger: Arc<OnionMessenger>,
-	om_mailbox: Arc<OnionMessageMailbox>,
+	om_mailbox: Option<Arc<OnionMessageMailbox>>,
 }
 
 impl<L: Deref + Clone + Sync + Send + 'static> EventHandler<L>
@@ -476,7 +476,7 @@ where
 		liquidity_source: Option<Arc<LiquiditySource<Arc<Logger>>>>,
 		payment_store: Arc<PaymentStore>, peer_store: Arc<PeerStore<L>>,
 		static_invoice_store: Option<StaticInvoiceStore>, onion_messenger: Arc<OnionMessenger>,
-		om_mailbox: Arc<OnionMessageMailbox>, runtime: Arc<Runtime>, logger: L,
+		om_mailbox: Option<Arc<OnionMessageMailbox>>, runtime: Arc<Runtime>, logger: L,
 		config: Arc<Config>,
 	) -> Self {
 		Self {
@@ -1498,13 +1498,22 @@ where
 				self.bump_tx_event_handler.handle_event(&bte).await;
 			},
 			LdkEvent::OnionMessageIntercepted { peer_node_id, message } => {
-				self.om_mailbox.onion_message_intercepted(peer_node_id, message);
+				if let Some(om_mailbox) = self.om_mailbox.as_ref() {
+					om_mailbox.onion_message_intercepted(peer_node_id, message);
+				} else {
+					log_trace!(
+						self.logger,
+						"Onion message intercepted, but no onion message mailbox available"
+					);
+				}
 			},
 			LdkEvent::OnionMessagePeerConnected { peer_node_id } => {
-				let messages = self.om_mailbox.onion_message_peer_connected(peer_node_id);
+				if let Some(om_mailbox) = self.om_mailbox.as_ref() {
+					let messages = om_mailbox.onion_message_peer_connected(peer_node_id);
 
-				for message in messages {
-					let _ = self.onion_messenger.forward_onion_message(message, &peer_node_id);
+					for message in messages {
+						let _ = self.onion_messenger.forward_onion_message(message, &peer_node_id);
+					}
 				}
 			},
 

src/lib.rs

@@ -92,7 +92,6 @@ pub mod io;
 pub mod liquidity;
 pub mod logger;
 mod message_handler;
-mod om_mailbox;
 pub mod payment;
 mod peer_store;
 mod runtime;
@@ -137,7 +136,7 @@ use gossip::GossipSource;
 use graph::NetworkGraph;
 use io::utils::write_node_metrics;
 use liquidity::{LSPS1Liquidity, LiquiditySource};
-use om_mailbox::OnionMessageMailbox;
+use payment::asynchronous::om_mailbox::OnionMessageMailbox;
 use payment::asynchronous::static_invoice_store::StaticInvoiceStore;
 use payment::{
 	Bolt11Payment, Bolt12Payment, OnchainPayment, PaymentDetails, SpontaneousPayment,
@@ -207,7 +206,7 @@ pub struct Node {
 	is_running: Arc<RwLock<bool>>,
 	is_listening: Arc<AtomicBool>,
 	node_metrics: Arc<RwLock<NodeMetrics>>,
-	om_mailbox: Arc<OnionMessageMailbox>,
+	om_mailbox: Option<Arc<OnionMessageMailbox>>,
 }
 
 impl Node {
@@ -521,7 +520,7 @@ impl Node {
 			Arc::clone(&self.peer_store),
 			static_invoice_store,
 			Arc::clone(&self.onion_messenger),
-			Arc::clone(&self.om_mailbox),
+			self.om_mailbox.clone(),
 			Arc::clone(&self.runtime),
 			Arc::clone(&self.logger),
 			Arc::clone(&self.config),
@@ -1496,11 +1495,6 @@ impl Node {
 				Error::PersistenceFailed
 			})
 	}
-
-	#[allow(missing_docs)]
-	pub fn om_mailbox_is_empty(&self) -> bool {
-		self.om_mailbox.is_empty()
-	}
 }
 
 impl Drop for Node {

src/payment/asynchronous/mod.rs

@@ -5,5 +5,6 @@
 // http://opensource.org/licenses/MIT>, at your option. You may not use this file except in
 // accordance with one or both of these licenses.
 
+pub(crate) mod om_mailbox;
 mod rate_limiter;
 pub(crate) mod static_invoice_store;

src/om_mailbox.rs renamed to src/payment/asynchronous/om_mailbox.rs

@@ -1,26 +1,22 @@
-use std::{
-	collections::{HashMap, VecDeque},
-	sync::Mutex,
-};
+use std::collections::{HashMap, VecDeque};
+use std::sync::Mutex;
 
+use bitcoin::secp256k1::PublicKey;
 use lightning::ln::msgs::OnionMessage;
 
 pub(crate) struct OnionMessageMailbox {
 	map: Mutex<HashMap<bitcoin::secp256k1::PublicKey, VecDeque<lightning::ln::msgs::OnionMessage>>>,
 }
 
 impl OnionMessageMailbox {
-	const MAX_MESSAGES_PER_PEER: usize = 100;
-	const MAX_PEERS: usize = 100;
+	const MAX_MESSAGES_PER_PEER: usize = 30;
+	const MAX_PEERS: usize = 300;
 
 	pub fn new() -> Self {
-		Self { map: Mutex::new(HashMap::new()) }
+		Self { map: Mutex::new(HashMap::with_capacity(Self::MAX_PEERS)) }
 	}
 
-	pub(crate) fn onion_message_intercepted(
-		&self, peer_node_id: bitcoin::secp256k1::PublicKey,
-		message: lightning::ln::msgs::OnionMessage,
-	) {
+	pub(crate) fn onion_message_intercepted(&self, peer_node_id: PublicKey, message: OnionMessage) {
 		let mut map = self.map.lock().unwrap();
 
 		let queue = map.entry(peer_node_id).or_insert_with(VecDeque::new);
@@ -31,11 +27,8 @@ impl OnionMessageMailbox {
 
 		// Enforce a peers limit. If exceeded, evict the peer with the longest queue.
 		if map.len() > Self::MAX_PEERS {
-			let peer_to_remove = map
-				.iter()
-				.max_by_key(|(_, queue)| queue.len())
-				.map(|(peer, _)| peer.clone())
-				.unwrap();
+			let peer_to_remove =
+				map.iter().max_by_key(|(_, queue)| queue.len()).map(|(peer, _)| *peer).unwrap();
 
 			map.remove(&peer_to_remove);
 		}
@@ -53,6 +46,7 @@ impl OnionMessageMailbox {
 		}
 	}
 
+	#[cfg(test)]
 	pub(crate) fn is_empty(&self) -> bool {
 		let map = self.map.lock().unwrap();
 		map.is_empty()
@@ -61,13 +55,11 @@ impl OnionMessageMailbox {
 
 #[cfg(test)]
 mod tests {
-	use bitcoin::{
-		key::Secp256k1,
-		secp256k1::{PublicKey, SecretKey},
-	};
+	use bitcoin::key::Secp256k1;
+	use bitcoin::secp256k1::{PublicKey, SecretKey};
 	use lightning::onion_message;
 
-	use crate::om_mailbox::OnionMessageMailbox;
+	use crate::payment::asynchronous::om_mailbox::OnionMessageMailbox;
 
 	#[test]
 	fn onion_message_mailbox() {

tests/integration_tests_rust.rs

@@ -1247,9 +1247,7 @@ fn async_payment() {
 		node_sender.bolt12_payment().send_using_amount(&offer, 5_000, None, None).unwrap();
 
 	// Sleep to allow the payment reach a state where the htlc is held and waiting for the receiver to come online.
-	while node_receiver_lsp.om_mailbox_is_empty() {
-		std::thread::sleep(std::time::Duration::from_millis(100));
-	}
+	std::thread::sleep(std::time::Duration::from_millis(3000));
 
 	node_receiver.start().unwrap();