Add onion mailbox for async receivers

This introduces an in-memory mailbox to hold onion messages until the
receiver comes online. This is required for async payment
`held_htlc_available` messages.

The mailbox is bounded by a maximum number of peers and a maximum
number of messages per peer.

Author: joostjager

Cargo.toml

@@ -52,17 +52,17 @@ default = []
 #lightning-liquidity = { git = "https://github.com/lightningdevkit/rust-lightning", branch = "main" }
 #lightning-macros = { git = "https://github.com/lightningdevkit/rust-lightning", branch = "main" }
 
-lightning = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "50391d3a3efa7a8f32d119d126a633e4b1981ee6", features = ["std"] }
-lightning-types = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "50391d3a3efa7a8f32d119d126a633e4b1981ee6" }
-lightning-invoice = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "50391d3a3efa7a8f32d119d126a633e4b1981ee6", features = ["std"] }
-lightning-net-tokio = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "50391d3a3efa7a8f32d119d126a633e4b1981ee6" }
-lightning-persister = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "50391d3a3efa7a8f32d119d126a633e4b1981ee6" }
-lightning-background-processor = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "50391d3a3efa7a8f32d119d126a633e4b1981ee6" }
-lightning-rapid-gossip-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "50391d3a3efa7a8f32d119d126a633e4b1981ee6" }
-lightning-block-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "50391d3a3efa7a8f32d119d126a633e4b1981ee6", features = ["rest-client", "rpc-client", "tokio"] }
-lightning-transaction-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "50391d3a3efa7a8f32d119d126a633e4b1981ee6", features = ["esplora-async-https", "electrum-rustls-ring", "time"] }
-lightning-liquidity = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "50391d3a3efa7a8f32d119d126a633e4b1981ee6" }
-lightning-macros = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "50391d3a3efa7a8f32d119d126a633e4b1981ee6" }
+lightning = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "341e8e436968dd37b82594a693dbf5c535662880", features = ["std"] }
+lightning-types = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "341e8e436968dd37b82594a693dbf5c535662880" }
+lightning-invoice = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "341e8e436968dd37b82594a693dbf5c535662880", features = ["std"] }
+lightning-net-tokio = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "341e8e436968dd37b82594a693dbf5c535662880" }
+lightning-persister = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "341e8e436968dd37b82594a693dbf5c535662880" }
+lightning-background-processor = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "341e8e436968dd37b82594a693dbf5c535662880" }
+lightning-rapid-gossip-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "341e8e436968dd37b82594a693dbf5c535662880" }
+lightning-block-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "341e8e436968dd37b82594a693dbf5c535662880", features = ["rest-client", "rpc-client", "tokio"] }
+lightning-transaction-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "341e8e436968dd37b82594a693dbf5c535662880", features = ["esplora-async-https", "electrum-rustls-ring", "time"] }
+lightning-liquidity = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "341e8e436968dd37b82594a693dbf5c535662880" }
+lightning-macros = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "341e8e436968dd37b82594a693dbf5c535662880" }
 
 #lightning = { path = "../rust-lightning/lightning", features = ["std"] }
 #lightning-types = { path = "../rust-lightning/lightning-types" }
@@ -109,7 +109,7 @@ winapi = { version = "0.3", features = ["winbase"] }
 [dev-dependencies]
 #lightning = { version = "0.1.0", features = ["std", "_test_utils"] }
 #lightning = { git = "https://github.com/lightningdevkit/rust-lightning", branch="main", features = ["std", "_test_utils"] }
-lightning = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "50391d3a3efa7a8f32d119d126a633e4b1981ee6", features = ["std", "_test_utils"] }
+lightning = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "341e8e436968dd37b82594a693dbf5c535662880", features = ["std", "_test_utils"] }
 #lightning = { path = "../rust-lightning/lightning", features = ["std", "_test_utils"] }
 proptest = "1.0.0"
 regex = "1.5.6"

src/builder.rs

@@ -27,6 +27,7 @@ use crate::liquidity::{
 };
 use crate::logger::{log_error, LdkLogger, LogLevel, LogWriter, Logger};
 use crate::message_handler::NodeCustomMessageHandler;
+use crate::payment::asynchronous::om_mailbox::OnionMessageMailbox;
 use crate::peer_store::PeerStore;
 use crate::runtime::Runtime;
 use crate::tx_broadcaster::TransactionBroadcaster;
@@ -1452,17 +1453,31 @@ fn build_with_store_internal(
 	}
 
 	// Initialize the PeerManager
-	let onion_messenger: Arc<OnionMessenger> = Arc::new(OnionMessenger::new(
-		Arc::clone(&keys_manager),
-		Arc::clone(&keys_manager),
-		Arc::clone(&logger),
-		Arc::clone(&channel_manager),
-		message_router,
-		Arc::clone(&channel_manager),
-		Arc::clone(&channel_manager),
-		IgnoringMessageHandler {},
-		IgnoringMessageHandler {},
-	));
+	let onion_messenger: Arc<OnionMessenger> = if config.async_payment_services_enabled {
+		Arc::new(OnionMessenger::new_with_offline_peer_interception(
+			Arc::clone(&keys_manager),
+			Arc::clone(&keys_manager),
+			Arc::clone(&logger),
+			Arc::clone(&channel_manager),
+			message_router,
+			Arc::clone(&channel_manager),
+			Arc::clone(&channel_manager),
+			IgnoringMessageHandler {},
+			IgnoringMessageHandler {},
+		))
+	} else {
+		Arc::new(OnionMessenger::new(
+			Arc::clone(&keys_manager),
+			Arc::clone(&keys_manager),
+			Arc::clone(&logger),
+			Arc::clone(&channel_manager),
+			message_router,
+			Arc::clone(&channel_manager),
+			Arc::clone(&channel_manager),
+			IgnoringMessageHandler {},
+			IgnoringMessageHandler {},
+		))
+	};
 	let ephemeral_bytes: [u8; 32] = keys_manager.get_secure_random_bytes();
 
 	// Initialize the GossipSource
@@ -1649,6 +1664,12 @@ fn build_with_store_internal(
 		},
 	};
 
+	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(());
 	let is_running = Arc::new(RwLock::new(false));
@@ -1681,6 +1702,7 @@ fn build_with_store_internal(
 		is_running,
 		is_listening,
 		node_metrics,
+		om_mailbox,
 	})
 }
 

src/config.rs

@@ -181,6 +181,13 @@ pub struct Config {
 	pub route_parameters: Option<RouteParametersConfig>,
 	/// Whether to enable the static invoice service to support async payment reception for clients.
 	pub async_payment_services_enabled: bool,
+	/// If this is set to true, then if we as an often-offline payer receive a [`StaticInvoice`] to
+	/// pay, we will attempt to hold the corresponding outbound HTLCs with our next-hop channel
+	/// counterparty(s) that support the `htlc_hold` feature. This allows our node to go offline once
+	/// the HTLCs are locked in even though the recipient may not yet be online to receive them.
+	///
+	/// [`StaticInvoice`]: lightning::offers::static_invoice::StaticInvoice
+	pub hold_outbound_htlcs_at_next_hop: bool,
 }
 
 impl Default for Config {
@@ -196,6 +203,7 @@ impl Default for Config {
 			route_parameters: None,
 			node_alias: None,
 			async_payment_services_enabled: false,
+			hold_outbound_htlcs_at_next_hop: false,
 		}
 	}
 }
@@ -330,6 +338,13 @@ pub(crate) fn default_user_config(config: &Config) -> UserConfig {
 		user_config.channel_handshake_limits.force_announced_channel_preference = true;
 	}
 
+	if config.async_payment_services_enabled {
+		user_config.enable_htlc_hold = true;
+	}
+	if config.hold_outbound_htlcs_at_next_hop {
+		user_config.hold_outbound_htlcs_at_next_hop = true;
+	}
+
 	user_config
 }
 

src/event.rs

@@ -5,7 +5,8 @@
 // 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::types::{CustomTlvRecord, DynStore, PaymentStore, Sweeper, Wallet};
+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,
 	UserChannelId,
@@ -459,6 +460,8 @@ where
 	logger: L,
 	config: Arc<Config>,
 	static_invoice_store: Option<StaticInvoiceStore>,
+	onion_messenger: Arc<OnionMessenger>,
+	om_mailbox: Option<Arc<OnionMessageMailbox>>,
 }
 
 impl<L: Deref + Clone + Sync + Send + 'static> EventHandler<L>
@@ -472,7 +475,8 @@ where
 		output_sweeper: Arc<Sweeper>, network_graph: Arc<Graph>,
 		liquidity_source: Option<Arc<LiquiditySource<Arc<Logger>>>>,
 		payment_store: Arc<PaymentStore>, peer_store: Arc<PeerStore<L>>,
-		static_invoice_store: Option<StaticInvoiceStore>, runtime: Arc<Runtime>, logger: L,
+		static_invoice_store: Option<StaticInvoiceStore>, onion_messenger: Arc<OnionMessenger>,
+		om_mailbox: Option<Arc<OnionMessageMailbox>>, runtime: Arc<Runtime>, logger: L,
 		config: Arc<Config>,
 	) -> Self {
 		Self {
@@ -490,6 +494,8 @@ where
 			runtime,
 			config,
 			static_invoice_store,
+			onion_messenger,
+			om_mailbox,
 		}
 	}
 
@@ -1491,11 +1497,24 @@ where
 
 				self.bump_tx_event_handler.handle_event(&bte).await;
 			},
-			LdkEvent::OnionMessageIntercepted { .. } => {
-				debug_assert!(false, "We currently don't support onion message interception, so this event should never be emitted.");
+			LdkEvent::OnionMessageIntercepted { 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 { .. } => {
-				debug_assert!(false, "We currently don't support onion message interception, so this event should never be emitted.");
+			LdkEvent::OnionMessagePeerConnected { 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);
+					}
+				}
 			},
 
 			LdkEvent::PersistStaticInvoice {

src/lib.rs

@@ -136,6 +136,7 @@ use gossip::GossipSource;
 use graph::NetworkGraph;
 use io::utils::write_node_metrics;
 use liquidity::{LSPS1Liquidity, LiquiditySource};
+use payment::asynchronous::om_mailbox::OnionMessageMailbox;
 use payment::asynchronous::static_invoice_store::StaticInvoiceStore;
 use payment::{
 	Bolt11Payment, Bolt12Payment, OnchainPayment, PaymentDetails, SpontaneousPayment,
@@ -205,6 +206,7 @@ pub struct Node {
 	is_running: Arc<RwLock<bool>>,
 	is_listening: Arc<AtomicBool>,
 	node_metrics: Arc<RwLock<NodeMetrics>>,
+	om_mailbox: Option<Arc<OnionMessageMailbox>>,
 }
 
 impl Node {
@@ -517,6 +519,8 @@ impl Node {
 			Arc::clone(&self.payment_store),
 			Arc::clone(&self.peer_store),
 			static_invoice_store,
+			Arc::clone(&self.onion_messenger),
+			self.om_mailbox.clone(),
 			Arc::clone(&self.runtime),
 			Arc::clone(&self.logger),
 			Arc::clone(&self.config),

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/payment/asynchronous/om_mailbox.rs

@@ -0,0 +1,99 @@
+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 = 30;
+	const MAX_PEERS: usize = 300;
+
+	pub fn new() -> Self {
+		Self { map: Mutex::new(HashMap::with_capacity(Self::MAX_PEERS)) }
+	}
+
+	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);
+		if queue.len() >= Self::MAX_MESSAGES_PER_PEER {
+			queue.pop_front();
+		}
+		queue.push_back(message);
+
+		// 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).unwrap();
+
+			map.remove(&peer_to_remove);
+		}
+	}
+
+	pub(crate) fn onion_message_peer_connected(
+		&self, peer_node_id: bitcoin::secp256k1::PublicKey,
+	) -> Vec<OnionMessage> {
+		let mut map = self.map.lock().unwrap();
+
+		if let Some(queue) = map.remove(&peer_node_id) {
+			queue.into()
+		} else {
+			Vec::new()
+		}
+	}
+
+	#[cfg(test)]
+	pub(crate) fn is_empty(&self) -> bool {
+		let map = self.map.lock().unwrap();
+		map.is_empty()
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use bitcoin::key::Secp256k1;
+	use bitcoin::secp256k1::{PublicKey, SecretKey};
+	use lightning::onion_message;
+
+	use crate::payment::asynchronous::om_mailbox::OnionMessageMailbox;
+
+	#[test]
+	fn onion_message_mailbox() {
+		let mailbox = OnionMessageMailbox::new();
+
+		let secp = Secp256k1::new();
+		let sk_bytes = [12; 32];
+		let sk = SecretKey::from_slice(&sk_bytes).unwrap();
+		let peer_node_id = PublicKey::from_secret_key(&secp, &sk);
+
+		let blinding_sk = SecretKey::from_slice(&[13; 32]).unwrap();
+		let blinding_point = PublicKey::from_secret_key(&secp, &blinding_sk);
+
+		let message_sk = SecretKey::from_slice(&[13; 32]).unwrap();
+		let message_point = PublicKey::from_secret_key(&secp, &message_sk);
+
+		let message = lightning::ln::msgs::OnionMessage {
+			blinding_point,
+			onion_routing_packet: onion_message::packet::Packet {
+				version: 0,
+				public_key: message_point,
+				hop_data: vec![1, 2, 3],
+				hmac: [0; 32],
+			},
+		};
+		mailbox.onion_message_intercepted(peer_node_id, message.clone());
+
+		let messages = mailbox.onion_message_peer_connected(peer_node_id);
+		assert_eq!(messages.len(), 1);
+		assert_eq!(messages[0], message);
+
+		assert!(mailbox.is_empty());
+
+		let messages = mailbox.onion_message_peer_connected(peer_node_id);
+		assert_eq!(messages.len(), 0);
+	}
+}

tests/integration_tests_rust.rs

@@ -1132,13 +1132,14 @@ fn simple_bolt12_send_receive() {
 }
 
 #[test]
-fn static_invoice_server() {
+fn async_payment() {
 	let (bitcoind, electrsd) = setup_bitcoind_and_electrsd();
 	let chain_source = TestChainSource::Esplora(&electrsd);
 
 	let mut config_sender = random_config(true);
 	config_sender.node_config.listening_addresses = None;
 	config_sender.node_config.node_alias = None;
+	config_sender.node_config.hold_outbound_htlcs_at_next_hop = true;
 	config_sender.log_writer =
 		TestLogWriter::Custom(Arc::new(MultiNodeLogger::new("sender      ".to_string())));
 	let node_sender = setup_node(&chain_source, config_sender, None);
@@ -1241,9 +1242,16 @@ fn static_invoice_server() {
 		std::thread::sleep(std::time::Duration::from_millis(100));
 	};
 
+	node_receiver.stop().unwrap();
+
 	let payment_id =
 		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.
+	std::thread::sleep(std::time::Duration::from_millis(3000));
+
+	node_receiver.start().unwrap();
+
 	expect_payment_successful_event!(node_sender, Some(payment_id), None);
 }