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

bindings/ldk_node.udl

@@ -13,7 +13,6 @@ dictionary Config {
 	u64 probing_liquidity_limit_multiplier;
 	AnchorChannelsConfig? anchor_channels_config;
 	RouteParametersConfig? route_parameters;
-	boolean async_payment_services_enabled;
 };
 
 dictionary AnchorChannelsConfig {
@@ -95,6 +94,7 @@ interface Builder {
 	void set_announcement_addresses(sequence<SocketAddress> announcement_addresses);
 	[Throws=BuildError]
 	void set_node_alias(string node_alias);
+	void set_async_payments_role(AsyncPaymentsRole? role);
 	[Throws=BuildError]
 	Node build();
 	[Throws=BuildError]
@@ -720,6 +720,11 @@ enum Currency {
 	"Signet",
 };
 
+enum AsyncPaymentsRole {
+	"Client",
+	"Server",
+};
+
 dictionary RouteHintHop {
 	PublicKey src_node_id;
 	u64 short_channel_id;

src/builder.rs

@@ -7,9 +7,9 @@
 
 use crate::chain::ChainSource;
 use crate::config::{
-	default_user_config, may_announce_channel, AnnounceError, BitcoindRestClientConfig, Config,
-	ElectrumSyncConfig, EsploraSyncConfig, DEFAULT_ESPLORA_SERVER_URL, DEFAULT_LOG_FILENAME,
-	DEFAULT_LOG_LEVEL, WALLET_KEYS_SEED_LEN,
+	default_user_config, may_announce_channel, AnnounceError, AsyncPaymentsRole,
+	BitcoindRestClientConfig, Config, ElectrumSyncConfig, EsploraSyncConfig,
+	DEFAULT_ESPLORA_SERVER_URL, DEFAULT_LOG_FILENAME, DEFAULT_LOG_LEVEL, WALLET_KEYS_SEED_LEN,
 };
 
 use crate::connection::ConnectionManager;
@@ -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;
@@ -241,6 +242,7 @@ pub struct NodeBuilder {
 	liquidity_source_config: Option<LiquiditySourceConfig>,
 	log_writer_config: Option<LogWriterConfig>,
 	runtime_handle: Option<tokio::runtime::Handle>,
+	async_payments_role: Option<AsyncPaymentsRole>,
 }
 
 impl NodeBuilder {
@@ -266,6 +268,7 @@ impl NodeBuilder {
 			liquidity_source_config,
 			log_writer_config,
 			runtime_handle,
+			async_payments_role: None,
 		}
 	}
 
@@ -544,6 +547,12 @@ impl NodeBuilder {
 		Ok(self)
 	}
 
+	/// Sets the role of the node in an asynchronous payments context.
+	pub fn set_async_payments_role(&mut self, role: Option<AsyncPaymentsRole>) -> &mut Self {
+		self.async_payments_role = role;
+		self
+	}
+
 	/// Builds a [`Node`] instance with a [`SqliteStore`] backend and according to the options
 	/// previously configured.
 	pub fn build(&self) -> Result<Node, BuildError> {
@@ -704,6 +713,7 @@ impl NodeBuilder {
 			runtime,
 			logger,
 			Arc::new(vss_store),
+			self.async_payments_role,
 		)
 	}
 
@@ -736,6 +746,7 @@ impl NodeBuilder {
 			runtime,
 			logger,
 			kv_store,
+			self.async_payments_role,
 		)
 	}
 }
@@ -989,6 +1000,11 @@ impl ArcedNodeBuilder {
 		self.inner.write().unwrap().set_node_alias(node_alias).map(|_| ())
 	}
 
+	/// Sets the role of the node in an asynchronous payments context.
+	pub fn set_async_payments_role(&self, role: Option<AsyncPaymentsRole>) {
+		_ = self.inner.write().unwrap().set_async_payments_role(role)
+	}
+
 	/// Builds a [`Node`] instance with a [`SqliteStore`] backend and according to the options
 	/// previously configured.
 	pub fn build(&self) -> Result<Arc<Node>, BuildError> {
@@ -1084,6 +1100,7 @@ fn build_with_store_internal(
 	gossip_source_config: Option<&GossipSourceConfig>,
 	liquidity_source_config: Option<&LiquiditySourceConfig>, seed_bytes: [u8; 64],
 	runtime: Arc<Runtime>, logger: Arc<Logger>, kv_store: Arc<DynStore>,
+	async_payments_role: Option<AsyncPaymentsRole>,
 ) -> Result<Node, BuildError> {
 	optionally_install_rustls_cryptoprovider();
 
@@ -1378,8 +1395,14 @@ fn build_with_store_internal(
 			100;
 	}
 
-	if config.async_payment_services_enabled {
-		user_config.accept_forwards_to_priv_channels = true;
+	if let Some(role) = async_payments_role {
+		match role {
+			AsyncPaymentsRole::Server => {
+				user_config.accept_forwards_to_priv_channels = true;
+				user_config.enable_htlc_hold = true;
+			},
+			AsyncPaymentsRole::Client => user_config.hold_outbound_htlcs_at_next_hop = true,
+		}
 	}
 
 	let message_router =
@@ -1452,17 +1475,32 @@ 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 let Some(AsyncPaymentsRole::Server) = async_payments_role {
+			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 +1687,12 @@ fn build_with_store_internal(
 		},
 	};
 
+	let om_mailbox = if let Some(AsyncPaymentsRole::Server) = async_payments_role {
+		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 +1725,8 @@ fn build_with_store_internal(
 		is_running,
 		is_listening,
 		node_metrics,
+		om_mailbox,
+		async_payments_role,
 	})
 }
 

src/config.rs

@@ -179,8 +179,6 @@ pub struct Config {
 	/// **Note:** If unset, default parameters will be used, and you will be able to override the
 	/// parameters on a per-payment basis in the corresponding method calls.
 	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,
 }
 
 impl Default for Config {
@@ -195,7 +193,6 @@ impl Default for Config {
 			anchor_channels_config: Some(AnchorChannelsConfig::default()),
 			route_parameters: None,
 			node_alias: None,
-			async_payment_services_enabled: false,
 		}
 	}
 }
@@ -537,6 +534,17 @@ impl From<MaxDustHTLCExposure> for LdkMaxDustHTLCExposure {
 	}
 }
 
+#[derive(Debug, Clone, Copy)]
+/// The role of the node in an asynchronous payments context.
+pub enum AsyncPaymentsRole {
+	/// Node acts a client in an async payments context. This means that if possible, it will instruct its peers to hold
+	/// htlcs for it, so that it can go offline.
+	Client,
+	/// Node acts as a server in an async payments context. This means that it will hold async payments htlcs and onion
+	/// messages for its peers.
+	Server,
+}
+
 #[cfg(test)]
 mod tests {
 	use std::str::FromStr;

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,33 @@ 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 {
+						if let Err(e) =
+							self.onion_messenger.forward_onion_message(message, &peer_node_id)
+						{
+							log_trace!(
+								self.logger,
+								"Failed to forward onion message to peer {}: {:?}",
+								peer_node_id,
+								e
+							);
+						}
+					}
+				}
 			},
 
 			LdkEvent::PersistStaticInvoice {

src/lib.rs

@@ -126,6 +126,7 @@ pub use builder::BuildError;
 pub use builder::NodeBuilder as Builder;
 
 use chain::ChainSource;
+use config::AsyncPaymentsRole;
 use config::{
 	default_user_config, may_announce_channel, ChannelConfig, Config, NODE_ANN_BCAST_INTERVAL,
 	PEER_RECONNECTION_INTERVAL, RGS_SYNC_INTERVAL,
@@ -136,6 +137,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 +207,8 @@ pub struct Node {
 	is_running: Arc<RwLock<bool>>,
 	is_listening: Arc<AtomicBool>,
 	node_metrics: Arc<RwLock<NodeMetrics>>,
+	om_mailbox: Option<Arc<OnionMessageMailbox>>,
+	async_payments_role: Option<AsyncPaymentsRole>,
 }
 
 impl Node {
@@ -499,7 +503,8 @@ impl Node {
 			Arc::clone(&self.logger),
 		));
 
-		let static_invoice_store = if self.config.async_payment_services_enabled {
+		let static_invoice_store = if let Some(AsyncPaymentsRole::Server) = self.async_payments_role
+		{
 			Some(StaticInvoiceStore::new(Arc::clone(&self.kv_store)))
 		} else {
 			None
@@ -517,6 +522,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),
@@ -826,9 +833,9 @@ impl Node {
 		Bolt12Payment::new(
 			Arc::clone(&self.channel_manager),
 			Arc::clone(&self.payment_store),
-			Arc::clone(&self.config),
 			Arc::clone(&self.is_running),
 			Arc::clone(&self.logger),
+			self.async_payments_role,
 		)
 	}
 
@@ -840,9 +847,9 @@ impl Node {
 		Arc::new(Bolt12Payment::new(
 			Arc::clone(&self.channel_manager),
 			Arc::clone(&self.payment_store),
-			Arc::clone(&self.config),
 			Arc::clone(&self.is_running),
 			Arc::clone(&self.logger),
+			self.async_payments_role,
 		))
 	}
 

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;