Wire OnionMessageInterceptor into LSPS2 service handler

Define the `OnionMessageInterceptor` trait with
`register_peer_for_interception()` and
`deregister_peer_for_interception()` methods, and implement it for
`OnionMessenger`. This allows external components to register peers
for onion message interception via a trait object, without needing
to know the concrete `OnionMessenger` type.

Wire the trait into `LSPS2ServiceHandler` as an optional
`Arc<dyn OnionMessageInterceptor>`. When provided:
- On init, all peers with active intercept SCIDs are registered
- In `invoice_parameters_generated()`, the counterparty is registered
  when a new intercept SCID is assigned

This ensures that onion messages for LSPS2 clients with active JIT
channel sessions are intercepted when those clients are offline,
enabling the LSP to store and forward messages when the client
reconnects.

Co-Authored-By: HAL 9000

Author: tnull

fuzz/src/lsps_message.rs

@@ -87,6 +87,7 @@ pub fn do_test(data: &[u8]) {
 			Arc::clone(&tx_broadcaster),
 			None,
 			None,
+			None,
 		)
 		.unwrap(),
 	);

lightning-background-processor/src/lib.rs

@@ -2556,6 +2556,7 @@ mod tests {
 					Arc::clone(&tx_broadcaster),
 					None,
 					None,
+					None,
 				)
 				.unwrap(),
 			);

lightning-liquidity/src/lsps2/service.rs

@@ -45,6 +45,7 @@ use lightning::events::HTLCHandlingFailureType;
 use lightning::ln::channelmanager::{AChannelManager, FailureCode, InterceptId};
 use lightning::ln::msgs::{ErrorAction, LightningError};
 use lightning::ln::types::ChannelId;
+use lightning::onion_message::messenger::OnionMessageInterceptor;
 use lightning::util::errors::APIError;
 use lightning::util::logger::Level;
 use lightning::util::ser::Writeable;
@@ -717,6 +718,7 @@ where
 	total_pending_requests: AtomicUsize,
 	config: LSPS2ServiceConfig,
 	persistence_in_flight: AtomicUsize,
+	onion_message_interceptor: Option<Arc<dyn OnionMessageInterceptor + Send + Sync>>,
 }
 
 impl<CM: Deref, K: KVStore + Clone, T: BroadcasterInterface + Clone> LSPS2ServiceHandler<CM, K, T>
@@ -728,6 +730,7 @@ where
 		per_peer_state: HashMap<PublicKey, Mutex<PeerState>>, pending_messages: Arc<MessageQueue>,
 		pending_events: Arc<EventQueue<K>>, channel_manager: CM, kv_store: K, tx_broadcaster: T,
 		config: LSPS2ServiceConfig,
+		onion_message_interceptor: Option<Arc<dyn OnionMessageInterceptor + Send + Sync>>,
 	) -> Result<Self, lightning::io::Error> {
 		let mut peer_by_intercept_scid = new_hash_map();
 		let mut peer_by_channel_id = new_hash_map();
@@ -756,6 +759,17 @@ where
 			}
 		}
 
+		// Register all peers and SCIDs with active intercept SCIDs for onion message
+		// interception, so that messages for offline peers are held rather than dropped.
+		// Both peer-based and SCID-based registration are needed to support clients using
+		// either pubkey or compact SCID encoding in their message blinded paths.
+		if let Some(ref interceptor) = onion_message_interceptor {
+			for (scid, node_id) in &peer_by_intercept_scid {
+				interceptor.register_peer_for_interception(*node_id);
+				interceptor.register_scid_for_interception(*scid, *node_id);
+			}
+		}
+
 		Ok(Self {
 			pending_messages,
 			pending_events,
@@ -768,6 +782,7 @@ where
 			kv_store,
 			tx_broadcaster,
 			config,
+			onion_message_interceptor,
 		})
 	}
 
@@ -776,6 +791,33 @@ where
 		&self.config
 	}
 
+	/// Cleans up `peer_by_intercept_scid` entries for the given SCIDs, and deregisters the peer
+	/// from onion message interception if they have no remaining active intercept SCIDs.
+	fn cleanup_intercept_scids(
+		&self, counterparty_node_id: &PublicKey, pruned_scids: &[u64], has_remaining_channels: bool,
+	) {
+		if pruned_scids.is_empty() {
+			return;
+		}
+
+		{
+			let mut peer_by_intercept_scid = self.peer_by_intercept_scid.write().unwrap();
+			for scid in pruned_scids {
+				peer_by_intercept_scid.remove(scid);
+			}
+		}
+
+		if let Some(ref interceptor) = self.onion_message_interceptor {
+			for scid in pruned_scids {
+				interceptor.deregister_scid_for_interception(*scid);
+			}
+
+			if !has_remaining_channels {
+				interceptor.deregister_peer_for_interception(counterparty_node_id);
+			}
+		}
+	}
+
 	/// Returns whether the peer has any active LSPS2 requests.
 	pub(crate) fn has_active_requests(&self, counterparty_node_id: &PublicKey) -> bool {
 		let outer_state_lock = self.per_peer_state.read().unwrap();
@@ -921,6 +963,14 @@ where
 							peer_by_intercept_scid.insert(intercept_scid, *counterparty_node_id);
 						}
 
+						if let Some(ref interceptor) = self.onion_message_interceptor {
+							interceptor.register_peer_for_interception(*counterparty_node_id);
+							interceptor.register_scid_for_interception(
+								intercept_scid,
+								*counterparty_node_id,
+							);
+						}
+
 						let outbound_jit_channel = OutboundJITChannel::new(
 							buy_request.payment_size_msat,
 							buy_request.opening_fee_params,
@@ -990,17 +1040,17 @@ where
 		let event_queue_notifier = self.pending_events.notifier();
 		let mut should_persist = None;
 
-		if let Some(counterparty_node_id) =
-			self.peer_by_intercept_scid.read().unwrap().get(&intercept_scid)
-		{
+		let counterparty_node_id =
+			self.peer_by_intercept_scid.read().unwrap().get(&intercept_scid).copied();
+		if let Some(counterparty_node_id) = counterparty_node_id {
 			let outer_state_lock = self.per_peer_state.read().unwrap();
-			match outer_state_lock.get(counterparty_node_id) {
+			match outer_state_lock.get(&counterparty_node_id) {
 				Some(inner_state_lock) => {
 					let mut peer_state = inner_state_lock.lock().unwrap();
 					if let Some(jit_channel) =
 						peer_state.outbound_channels_by_intercept_scid.get_mut(&intercept_scid)
 					{
-						should_persist = Some(*counterparty_node_id);
+						should_persist = Some(counterparty_node_id);
 						let htlc = InterceptedHTLC {
 							intercept_id,
 							expected_outbound_amount_msat,
@@ -1009,7 +1059,7 @@ where
 						match jit_channel.htlc_intercepted(htlc) {
 							Ok(Some(HTLCInterceptedAction::OpenChannel(open_channel_params))) => {
 								let event = LSPS2ServiceEvent::OpenChannel {
-									their_network_key: counterparty_node_id.clone(),
+									their_network_key: counterparty_node_id,
 									amt_to_forward_msat: open_channel_params.amt_to_forward_msat,
 									opening_fee_msat: open_channel_params.opening_fee_msat,
 									user_channel_id: jit_channel.user_channel_id,
@@ -1021,7 +1071,7 @@ where
 								self.channel_manager.get_cm().forward_intercepted_htlc(
 									intercept_id,
 									&channel_id,
-									*counterparty_node_id,
+									counterparty_node_id,
 									expected_outbound_amount_msat,
 								)?;
 							},
@@ -1038,7 +1088,7 @@ where
 									self.channel_manager.get_cm().forward_intercepted_htlc(
 										intercept_id,
 										&channel_id,
-										*counterparty_node_id,
+										counterparty_node_id,
 										amount_to_forward_msat,
 									)?;
 								}
@@ -1051,7 +1101,13 @@ where
 								peer_state
 									.outbound_channels_by_intercept_scid
 									.remove(&intercept_scid);
-								// TODO: cleanup peer_by_intercept_scid
+								let has_remaining =
+									!peer_state.outbound_channels_by_intercept_scid.is_empty();
+								self.cleanup_intercept_scids(
+									&counterparty_node_id,
+									&[intercept_scid],
+									has_remaining,
+								);
 								return Err(APIError::APIMisuseError { err: e.err });
 							},
 						}
@@ -1858,6 +1914,22 @@ where
 						debug_assert!(false);
 					}
 				}
+				if future_opt.is_some() {
+					// Clean up handler-level maps for the removed peer.
+					let removed_scids: Vec<u64> = self
+						.peer_by_intercept_scid
+						.read()
+						.unwrap()
+						.iter()
+						.filter(|(_, nid)| **nid == counterparty_node_id)
+						.map(|(scid, _)| *scid)
+						.collect();
+					self.cleanup_intercept_scids(&counterparty_node_id, &removed_scids, false);
+					self.peer_by_channel_id
+						.write()
+						.unwrap()
+						.retain(|_, node_id| *node_id != counterparty_node_id);
+				}
 				if let Some(future) = future_opt {
 					future.await?;
 					did_persist = true;

lightning-liquidity/src/manager.rs

@@ -47,6 +47,7 @@ use lightning::ln::channelmanager::AChannelManager;
 use lightning::ln::msgs::{ErrorAction, LightningError};
 use lightning::ln::peer_handler::CustomMessageHandler;
 use lightning::ln::wire::CustomMessageReader;
+use lightning::onion_message::messenger::OnionMessageInterceptor;
 use lightning::sign::{EntropySource, NodeSigner};
 use lightning::util::logger::Level;
 use lightning::util::persist::{KVStore, KVStoreSync, KVStoreSyncWrapper};
@@ -310,6 +311,7 @@ where
 		entropy_source: ES, node_signer: NS, channel_manager: CM, kv_store: K,
 		transaction_broadcaster: T, service_config: Option<LiquidityServiceConfig>,
 		client_config: Option<LiquidityClientConfig>,
+		onion_message_interceptor: Option<Arc<dyn OnionMessageInterceptor + Send + Sync>>,
 	) -> Result<Self, lightning::io::Error> {
 		Self::new_with_custom_time_provider(
 			entropy_source,
@@ -320,6 +322,7 @@ where
 			service_config,
 			client_config,
 			DefaultTimeProvider,
+			onion_message_interceptor,
 		)
 		.await
 	}
@@ -349,6 +352,7 @@ where
 		entropy_source: ES, node_signer: NS, channel_manager: CM, transaction_broadcaster: T,
 		kv_store: K, service_config: Option<LiquidityServiceConfig>,
 		client_config: Option<LiquidityClientConfig>, time_provider: TP,
+		onion_message_interceptor: Option<Arc<dyn OnionMessageInterceptor + Send + Sync>>,
 	) -> Result<Self, lightning::io::Error> {
 		let pending_msgs_or_needs_persist_notifier = Arc::new(Notifier::new());
 		let pending_messages =
@@ -391,6 +395,7 @@ where
 					kv_store.clone(),
 					transaction_broadcaster.clone(),
 					lsps2_service_config.clone(),
+					onion_message_interceptor.clone(),
 				)?)
 			} else {
 				None
@@ -940,6 +945,7 @@ where
 		entropy_source: ES, node_signer: NS, channel_manager: CM, kv_store_sync: KS,
 		transaction_broadcaster: T, service_config: Option<LiquidityServiceConfig>,
 		client_config: Option<LiquidityClientConfig>,
+		onion_message_interceptor: Option<Arc<dyn OnionMessageInterceptor + Send + Sync>>,
 	) -> Result<Self, lightning::io::Error> {
 		let kv_store = KVStoreSyncWrapper(kv_store_sync);
 
@@ -951,6 +957,7 @@ where
 			transaction_broadcaster,
 			service_config,
 			client_config,
+			onion_message_interceptor,
 		));
 
 		let mut waker = dummy_waker();
@@ -986,6 +993,7 @@ where
 		entropy_source: ES, node_signer: NS, channel_manager: CM, kv_store_sync: KS,
 		transaction_broadcaster: T, service_config: Option<LiquidityServiceConfig>,
 		client_config: Option<LiquidityClientConfig>, time_provider: TP,
+		onion_message_interceptor: Option<Arc<dyn OnionMessageInterceptor + Send + Sync>>,
 	) -> Result<Self, lightning::io::Error> {
 		let kv_store = KVStoreSyncWrapper(kv_store_sync);
 		let mut fut = pin!(LiquidityManager::new_with_custom_time_provider(
@@ -997,6 +1005,7 @@ where
 			service_config,
 			client_config,
 			time_provider,
+			onion_message_interceptor,
 		));
 
 		let mut waker = dummy_waker();

lightning-liquidity/tests/common/mod.rs

@@ -36,6 +36,7 @@ fn build_service_and_client_nodes<'a, 'b, 'c>(
 		Some(service_config),
 		None,
 		Arc::clone(&time_provider),
+		None,
 	)
 	.unwrap();
 
@@ -48,6 +49,7 @@ fn build_service_and_client_nodes<'a, 'b, 'c>(
 		None,
 		Some(client_config),
 		time_provider,
+		None,
 	)
 	.unwrap();
 

lightning-liquidity/tests/lsps1_integration_tests.rs

@@ -434,6 +434,7 @@ fn lsps1_service_handler_persistence_across_restarts() {
 			Some(service_config),
 			None,
 			Arc::clone(&time_provider),
+			None,
 		)
 		.unwrap();
 
@@ -454,6 +455,7 @@ fn lsps1_service_handler_persistence_across_restarts() {
 			None,
 			Some(client_config),
 			time_provider,
+			None,
 		)
 		.unwrap();
 
@@ -1087,6 +1089,7 @@ fn lsps1_expired_orders_are_pruned_and_not_persisted() {
 			Some(service_config),
 			None,
 			Arc::clone(&time_provider),
+			None,
 		)
 		.unwrap();
 
@@ -1106,6 +1109,7 @@ fn lsps1_expired_orders_are_pruned_and_not_persisted() {
 			None,
 			Some(client_config),
 			time_provider,
+			None,
 		)
 		.unwrap();
 

lightning-liquidity/tests/lsps2_integration_tests.rs

@@ -1079,6 +1079,7 @@ fn lsps2_service_handler_persistence_across_restarts() {
 			Some(service_config),
 			None,
 			time_provider,
+			None,
 		)
 		.unwrap();
 

lightning-liquidity/tests/lsps5_integration_tests.rs

@@ -1604,6 +1604,7 @@ fn lsps5_service_handler_persistence_across_restarts() {
 			Some(service_config),
 			None,
 			Arc::clone(&time_provider),
+			None,
 		)
 		.unwrap();