X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=lightning%2Fsrc%2Fonion_message%2Fmessenger.rs;h=0fb72c52d2784a03973bfb3ac2bbf846fc3daba6;hb=a9dcfaf952584ed835d733cb4688d5f96e86349d;hp=2c4e86bd6a92753ca92dcae778b12c0c7f29992b;hpb=3fb329b004aa5107750ef87dfd2d1766badc705e;p=rust-lightning diff --git a/lightning/src/onion_message/messenger.rs b/lightning/src/onion_message/messenger.rs index 2c4e86bd..0fb72c52 100644 --- a/lightning/src/onion_message/messenger.rs +++ b/lightning/src/onion_message/messenger.rs @@ -16,7 +16,7 @@ use bitcoin::hashes::sha256::Hash as Sha256; use bitcoin::secp256k1::{self, PublicKey, Scalar, Secp256k1, SecretKey}; use crate::blinded_path::{BlindedPath, IntroductionNode, NextMessageHop, NodeIdLookUp}; -use crate::blinded_path::message::{advance_path_by_one, ForwardTlvs, ReceiveTlvs}; +use crate::blinded_path::message::{advance_path_by_one, ForwardNode, ForwardTlvs, ReceiveTlvs}; use crate::blinded_path::utils; use crate::events::{Event, EventHandler, EventsProvider}; use crate::sign::{EntropySource, NodeSigner, Recipient}; @@ -47,6 +47,70 @@ use { pub(super) const MAX_TIMER_TICKS: usize = 2; +/// A trivial trait which describes any [`OnionMessenger`]. +/// +/// This is not exported to bindings users as general cover traits aren't useful in other +/// languages. +pub trait AOnionMessenger { + /// A type implementing [`EntropySource`] + type EntropySource: EntropySource + ?Sized; + /// A type that may be dereferenced to [`Self::EntropySource`] + type ES: Deref; + /// A type implementing [`NodeSigner`] + type NodeSigner: NodeSigner + ?Sized; + /// A type that may be dereferenced to [`Self::NodeSigner`] + type NS: Deref; + /// A type implementing [`Logger`] + type Logger: Logger + ?Sized; + /// A type that may be dereferenced to [`Self::Logger`] + type L: Deref; + /// A type implementing [`NodeIdLookUp`] + type NodeIdLookUp: NodeIdLookUp + ?Sized; + /// A type that may be dereferenced to [`Self::NodeIdLookUp`] + type NL: Deref; + /// A type implementing [`MessageRouter`] + type MessageRouter: MessageRouter + ?Sized; + /// A type that may be dereferenced to [`Self::MessageRouter`] + type MR: Deref; + /// A type implementing [`OffersMessageHandler`] + type OffersMessageHandler: OffersMessageHandler + ?Sized; + /// A type that may be dereferenced to [`Self::OffersMessageHandler`] + type OMH: Deref; + /// A type implementing [`CustomOnionMessageHandler`] + type CustomOnionMessageHandler: CustomOnionMessageHandler + ?Sized; + /// A type that may be dereferenced to [`Self::CustomOnionMessageHandler`] + type CMH: Deref; + /// Returns a reference to the actual [`OnionMessenger`] object. + fn get_om(&self) -> &OnionMessenger; +} + +impl AOnionMessenger +for OnionMessenger where + ES::Target: EntropySource, + NS::Target: NodeSigner, + L::Target: Logger, + NL::Target: NodeIdLookUp, + MR::Target: MessageRouter, + OMH::Target: OffersMessageHandler, + CMH::Target: CustomOnionMessageHandler, +{ + type EntropySource = ES::Target; + type ES = ES; + type NodeSigner = NS::Target; + type NS = NS; + type Logger = L::Target; + type L = L; + type NodeIdLookUp = NL::Target; + type NL = NL; + type MessageRouter = MR::Target; + type MR = MR; + type OffersMessageHandler = OMH::Target; + type OMH = OMH; + type CustomOnionMessageHandler = CMH::Target; + type CMH = CMH; + fn get_om(&self) -> &OnionMessenger { self } +} + /// A sender, receiver and forwarder of [`OnionMessage`]s. /// /// # Handling Messages @@ -71,6 +135,7 @@ pub(super) const MAX_TIMER_TICKS: usize = 2; /// # use bitcoin::hashes::hex::FromHex; /// # use bitcoin::secp256k1::{PublicKey, Secp256k1, SecretKey, self}; /// # use lightning::blinded_path::{BlindedPath, EmptyNodeIdLookUp}; +/// # use lightning::blinded_path::message::ForwardNode; /// # use lightning::sign::{EntropySource, KeysManager}; /// # use lightning::ln::peer_handler::IgnoringMessageHandler; /// # use lightning::onion_message::messenger::{Destination, MessageRouter, OnionMessagePath, OnionMessenger}; @@ -145,8 +210,11 @@ pub(super) const MAX_TIMER_TICKS: usize = 2; /// /// // Create a blinded path to yourself, for someone to send an onion message to. /// # let your_node_id = hop_node_id1; -/// let hops = [hop_node_id3, hop_node_id4, your_node_id]; -/// let blinded_path = BlindedPath::new_for_message(&hops, &keys_manager, &secp_ctx).unwrap(); +/// let hops = [ +/// ForwardNode { node_id: hop_node_id3, short_channel_id: None }, +/// ForwardNode { node_id: hop_node_id4, short_channel_id: None }, +/// ]; +/// let blinded_path = BlindedPath::new_for_message(&hops, your_node_id, &keys_manager, &secp_ctx).unwrap(); /// /// // Send a custom onion message to a blinded path. /// let destination = Destination::BlindedPath(blinded_path); @@ -176,6 +244,13 @@ where message_router: MR, offers_handler: OMH, custom_handler: CMH, + intercept_messages_for_offline_peers: bool, + pending_events: Mutex, +} + +struct PendingEvents { + intercepted_msgs: Vec, + peer_connecteds: Vec, } /// [`OnionMessage`]s buffered to be sent. @@ -250,22 +325,30 @@ impl OnionMessageRecipient { /// The `Responder` struct creates an appropriate [`ResponseInstruction`] /// for responding to a message. +#[derive(Clone, Debug, Eq, PartialEq)] pub struct Responder { /// The path along which a response can be sent. reply_path: BlindedPath, path_id: Option<[u8; 32]> } +impl_writeable_tlv_based!(Responder, { + (0, reply_path, required), + (2, path_id, option), +}); + impl Responder { /// Creates a new [`Responder`] instance with the provided reply path. - fn new(reply_path: BlindedPath, path_id: Option<[u8; 32]>) -> Self { + pub(super) fn new(reply_path: BlindedPath, path_id: Option<[u8; 32]>) -> Self { Responder { reply_path, path_id, } } - /// Creates the appropriate [`ResponseInstruction`] for a given response. + /// Creates a [`ResponseInstruction::WithoutReplyPath`] for a given response. + /// + /// Use when the recipient doesn't need to send back a reply to us. pub fn respond(self, response: T) -> ResponseInstruction { ResponseInstruction::WithoutReplyPath(OnionMessageResponse { message: response, @@ -273,6 +356,17 @@ impl Responder { path_id: self.path_id, }) } + + /// Creates a [`ResponseInstruction::WithReplyPath`] for a given response. + /// + /// Use when the recipient needs to send back a reply to us. + pub fn respond_with_reply_path(self, response: T) -> ResponseInstruction { + ResponseInstruction::WithReplyPath(OnionMessageResponse { + message: response, + reply_path: self.reply_path, + path_id: self.path_id, + }) + } } /// This struct contains the information needed to reply to a received message. @@ -284,6 +378,9 @@ pub struct OnionMessageResponse { /// `ResponseInstruction` represents instructions for responding to received messages. pub enum ResponseInstruction { + /// Indicates that a response should be sent including a reply path for + /// the recipient to respond back. + WithReplyPath(OnionMessageResponse), /// Indicates that a response should be sent without including a reply path /// for the recipient to respond back. WithoutReplyPath(OnionMessageResponse), @@ -337,9 +434,41 @@ pub trait MessageRouter { >( &self, recipient: PublicKey, peers: Vec, secp_ctx: &Secp256k1, ) -> Result, ()>; + + /// Creates compact [`BlindedPath`]s to the `recipient` node. The nodes in `peers` are assumed + /// to be direct peers with the `recipient`. + /// + /// Compact blinded paths use short channel ids instead of pubkeys for a smaller serialization, + /// which is beneficial when a QR code is used to transport the data. The SCID is passed using a + /// [`ForwardNode`] but may be `None` for graceful degradation. + /// + /// Implementations using additional intermediate nodes are responsible for using a + /// [`ForwardNode`] with `Some` short channel id, if possible. Similarly, implementations should + /// call [`BlindedPath::use_compact_introduction_node`]. + /// + /// The provided implementation simply delegates to [`MessageRouter::create_blinded_paths`], + /// ignoring the short channel ids. + fn create_compact_blinded_paths< + T: secp256k1::Signing + secp256k1::Verification + >( + &self, recipient: PublicKey, peers: Vec, secp_ctx: &Secp256k1, + ) -> Result, ()> { + let peers = peers + .into_iter() + .map(|ForwardNode { node_id, short_channel_id: _ }| node_id) + .collect(); + self.create_blinded_paths(recipient, peers, secp_ctx) + } } /// A [`MessageRouter`] that can only route to a directly connected [`Destination`]. +/// +/// # Privacy +/// +/// Creating [`BlindedPath`]s may affect privacy since, if a suitable path cannot be found, it will +/// create a one-hop path using the recipient as the introduction node if it is a announced node. +/// Otherwise, there is no way to find a path to the introduction node in order to send a message, +/// and thus an `Err` is returned. pub struct DefaultMessageRouter>, L: Deref, ES: Deref> where L::Target: Logger, @@ -358,51 +487,12 @@ where pub fn new(network_graph: G, entropy_source: ES) -> Self { Self { network_graph, entropy_source } } -} - -impl>, L: Deref, ES: Deref> MessageRouter for DefaultMessageRouter -where - L::Target: Logger, - ES::Target: EntropySource, -{ - fn find_path( - &self, sender: PublicKey, peers: Vec, mut destination: Destination - ) -> Result { - let network_graph = self.network_graph.deref().read_only(); - destination.resolve(&network_graph); - - let first_node = match destination.first_node() { - Some(first_node) => first_node, - None => return Err(()), - }; - - if peers.contains(&first_node) || sender == first_node { - Ok(OnionMessagePath { - intermediate_nodes: vec![], destination, first_node_addresses: None - }) - } else { - let node_announcement = network_graph - .node(&NodeId::from_pubkey(&first_node)) - .and_then(|node_info| node_info.announcement_info.as_ref()) - .and_then(|announcement_info| announcement_info.announcement_message.as_ref()) - .map(|node_announcement| &node_announcement.contents); - - match node_announcement { - Some(node_announcement) if node_announcement.features.supports_onion_messages() => { - let first_node_addresses = Some(node_announcement.addresses.clone()); - Ok(OnionMessagePath { - intermediate_nodes: vec![], destination, first_node_addresses - }) - }, - _ => Err(()), - } - } - } - fn create_blinded_paths< + fn create_blinded_paths_from_iter< + I: Iterator, T: secp256k1::Signing + secp256k1::Verification >( - &self, recipient: PublicKey, peers: Vec, secp_ctx: &Secp256k1, + &self, recipient: PublicKey, peers: I, secp_ctx: &Secp256k1, compact_paths: bool ) -> Result, ()> { // Limit the number of blinded paths that are computed. const MAX_PATHS: usize = 3; @@ -415,13 +505,13 @@ where let is_recipient_announced = network_graph.nodes().contains_key(&NodeId::from_pubkey(&recipient)); - let mut peer_info = peers.iter() + let mut peer_info = peers // Limit to peers with announced channels - .filter_map(|pubkey| + .filter_map(|peer| network_graph - .node(&NodeId::from_pubkey(pubkey)) + .node(&NodeId::from_pubkey(&peer.node_id)) .filter(|info| info.channels.len() >= MIN_PEER_CHANNELS) - .map(|info| (*pubkey, info.is_tor_only(), info.channels.len())) + .map(|info| (peer, info.is_tor_only(), info.channels.len())) ) // Exclude Tor-only nodes when the recipient is announced. .filter(|(_, is_tor_only, _)| !(*is_tor_only && is_recipient_announced)) @@ -433,12 +523,13 @@ where }); let paths = peer_info.into_iter() - .map(|(pubkey, _, _)| vec![pubkey, recipient]) - .map(|node_pks| BlindedPath::new_for_message(&node_pks, &*self.entropy_source, secp_ctx)) + .map(|(peer, _, _)| { + BlindedPath::new_for_message(&[peer], recipient, &*self.entropy_source, secp_ctx) + }) .take(MAX_PATHS) .collect::, _>>(); - match paths { + let mut paths = match paths { Ok(paths) if !paths.is_empty() => Ok(paths), _ => { if is_recipient_announced { @@ -448,7 +539,73 @@ where Err(()) } }, + }?; + + if compact_paths { + for path in &mut paths { + path.use_compact_introduction_node(&network_graph); + } } + + Ok(paths) + } +} + +impl>, L: Deref, ES: Deref> MessageRouter for DefaultMessageRouter +where + L::Target: Logger, + ES::Target: EntropySource, +{ + fn find_path( + &self, sender: PublicKey, peers: Vec, mut destination: Destination + ) -> Result { + let network_graph = self.network_graph.deref().read_only(); + destination.resolve(&network_graph); + + let first_node = match destination.first_node() { + Some(first_node) => first_node, + None => return Err(()), + }; + + if peers.contains(&first_node) || sender == first_node { + Ok(OnionMessagePath { + intermediate_nodes: vec![], destination, first_node_addresses: None + }) + } else { + let node_details = network_graph + .node(&NodeId::from_pubkey(&first_node)) + .and_then(|node_info| node_info.announcement_info.as_ref()) + .map(|announcement_info| (announcement_info.features(), announcement_info.addresses())); + + match node_details { + Some((features, addresses)) if features.supports_onion_messages() && addresses.len() > 0 => { + let first_node_addresses = Some(addresses.clone()); + Ok(OnionMessagePath { + intermediate_nodes: vec![], destination, first_node_addresses + }) + }, + _ => Err(()), + } + } + } + + fn create_blinded_paths< + T: secp256k1::Signing + secp256k1::Verification + >( + &self, recipient: PublicKey, peers: Vec, secp_ctx: &Secp256k1, + ) -> Result, ()> { + let peers = peers + .into_iter() + .map(|node_id| ForwardNode { node_id, short_channel_id: None }); + self.create_blinded_paths_from_iter(recipient, peers, secp_ctx, false) + } + + fn create_compact_blinded_paths< + T: secp256k1::Signing + secp256k1::Verification + >( + &self, recipient: PublicKey, peers: Vec, secp_ctx: &Secp256k1, + ) -> Result, ()> { + self.create_blinded_paths_from_iter(recipient, peers.into_iter(), secp_ctx, true) } } @@ -552,7 +709,11 @@ pub enum SendError { TooFewBlindedHops, /// The first hop is not a peer and doesn't have a known [`SocketAddress`]. InvalidFirstHop(PublicKey), - /// A path from the sender to the destination could not be found by the [`MessageRouter`]. + /// Indicates that a path could not be found by the [`MessageRouter`]. + /// + /// This occurs when either: + /// - No path from the sender to the destination was found to send the onion message + /// - No reply path to the sender could be created when responding to an onion message PathNotFound, /// Onion message contents must have a TLV type >= 64. InvalidMessage, @@ -841,6 +1002,48 @@ where pub fn new( entropy_source: ES, node_signer: NS, logger: L, node_id_lookup: NL, message_router: MR, offers_handler: OMH, custom_handler: CMH + ) -> Self { + Self::new_inner( + entropy_source, node_signer, logger, node_id_lookup, message_router, + offers_handler, custom_handler, false + ) + } + + /// Similar to [`Self::new`], but rather than dropping onion messages that are + /// intended to be forwarded to offline peers, we will intercept them for + /// later forwarding. + /// + /// Interception flow: + /// 1. If an onion message for an offline peer is received, `OnionMessenger` will + /// generate an [`Event::OnionMessageIntercepted`]. Event handlers can + /// then choose to persist this onion message for later forwarding, or drop + /// it. + /// 2. When the offline peer later comes back online, `OnionMessenger` will + /// generate an [`Event::OnionMessagePeerConnected`]. Event handlers will + /// then fetch all previously intercepted onion messages for this peer. + /// 3. Once the stored onion messages are fetched, they can finally be + /// forwarded to the now-online peer via [`Self::forward_onion_message`]. + /// + /// # Note + /// + /// LDK will not rate limit how many [`Event::OnionMessageIntercepted`]s + /// are generated, so it is the caller's responsibility to limit how many + /// onion messages are persisted and only persist onion messages for relevant + /// peers. + pub fn new_with_offline_peer_interception( + entropy_source: ES, node_signer: NS, logger: L, node_id_lookup: NL, + message_router: MR, offers_handler: OMH, custom_handler: CMH + ) -> Self { + Self::new_inner( + entropy_source, node_signer, logger, node_id_lookup, message_router, + offers_handler, custom_handler, true + ) + } + + fn new_inner( + entropy_source: ES, node_signer: NS, logger: L, node_id_lookup: NL, + message_router: MR, offers_handler: OMH, custom_handler: CMH, + intercept_messages_for_offline_peers: bool ) -> Self { let mut secp_ctx = Secp256k1::new(); secp_ctx.seeded_randomize(&entropy_source.get_secure_random_bytes()); @@ -854,6 +1057,11 @@ where message_router, offers_handler, custom_handler, + intercept_messages_for_offline_peers, + pending_events: Mutex::new(PendingEvents { + intercepted_msgs: Vec::new(), + peer_connecteds: Vec::new(), + }), } } @@ -925,6 +1133,24 @@ where .map_err(|_| SendError::PathNotFound) } + fn create_blinded_path(&self) -> Result { + let recipient = self.node_signer + .get_node_id(Recipient::Node) + .map_err(|_| SendError::GetNodeIdFailed)?; + let secp_ctx = &self.secp_ctx; + + let peers = self.message_recipients.lock().unwrap() + .iter() + .filter(|(_, peer)| matches!(peer, OnionMessageRecipient::ConnectedPeer(_))) + .map(|(node_id, _ )| *node_id) + .collect::>(); + + self.message_router + .create_blinded_paths(recipient, peers, secp_ctx) + .and_then(|paths| paths.into_iter().next().ok_or(())) + .map_err(|_| SendError::PathNotFound) + } + fn enqueue_onion_message( &self, path: OnionMessagePath, contents: T, reply_path: Option, log_suffix: fmt::Arguments @@ -961,6 +1187,27 @@ where } } + /// Forwards an [`OnionMessage`] to `peer_node_id`. Useful if we initialized + /// the [`OnionMessenger`] with [`Self::new_with_offline_peer_interception`] + /// and want to forward a previously intercepted onion message to a peer that + /// has just come online. + pub fn forward_onion_message( + &self, message: OnionMessage, peer_node_id: &PublicKey + ) -> Result<(), SendError> { + let mut message_recipients = self.message_recipients.lock().unwrap(); + if outbound_buffer_full(&peer_node_id, &message_recipients) { + return Err(SendError::BufferFull); + } + + match message_recipients.entry(*peer_node_id) { + hash_map::Entry::Occupied(mut e) if e.get().is_connected() => { + e.get_mut().enqueue_message(message); + Ok(()) + }, + _ => Err(SendError::InvalidFirstHop(*peer_node_id)) + } + } + #[cfg(any(test, feature = "_test_utils"))] pub fn send_onion_message_using_path( &self, path: OnionMessagePath, contents: T, reply_path: Option @@ -976,20 +1223,48 @@ where ) } - fn handle_onion_message_response( + /// Handles the response to an [`OnionMessage`] based on its [`ResponseInstruction`], + /// enqueueing any response for sending. + /// + /// This function is useful for asynchronous handling of [`OnionMessage`]s. + /// Handlers have the option to return [`ResponseInstruction::NoResponse`], indicating that + /// no immediate response should be sent. Then, they can transfer the associated [`Responder`] + /// to another task responsible for generating the response asynchronously. Subsequently, when + /// the response is prepared and ready for sending, that task can invoke this method to enqueue + /// the response for delivery. + pub fn handle_onion_message_response( &self, response: ResponseInstruction - ) { - if let ResponseInstruction::WithoutReplyPath(response) = response { - let message_type = response.message.msg_type(); - let _ = self.find_path_and_enqueue_onion_message( - response.message, Destination::BlindedPath(response.reply_path), None, - format_args!( - "when responding with {} to an onion message with path_id {:02x?}", - message_type, - response.path_id - ) - ); - } + ) -> Result, SendError> { + let (response, create_reply_path) = match response { + ResponseInstruction::WithReplyPath(response) => (response, true), + ResponseInstruction::WithoutReplyPath(response) => (response, false), + ResponseInstruction::NoResponse => return Ok(None), + }; + + let message_type = response.message.msg_type(); + let reply_path = if create_reply_path { + match self.create_blinded_path() { + Ok(reply_path) => Some(reply_path), + Err(err) => { + log_trace!( + self.logger, + "Failed to create reply path when responding with {} to an onion message \ + with path_id {:02x?}: {:?}", + message_type, response.path_id, err + ); + return Err(err); + } + } + } else { None }; + + self.find_path_and_enqueue_onion_message( + response.message, Destination::BlindedPath(response.reply_path), reply_path, + format_args!( + "when responding with {} to an onion message with path_id {:02x?}", + message_type, + response.path_id + ) + ).map(|result| Some(result)) } #[cfg(test)] @@ -1003,6 +1278,63 @@ where } msgs } + + fn enqueue_intercepted_event(&self, event: Event) { + const MAX_EVENTS_BUFFER_SIZE: usize = (1 << 10) * 256; + let mut pending_events = self.pending_events.lock().unwrap(); + let total_buffered_bytes: usize = + pending_events.intercepted_msgs.iter().map(|ev| ev.serialized_length()).sum(); + if total_buffered_bytes >= MAX_EVENTS_BUFFER_SIZE { + log_trace!(self.logger, "Dropping event {:?}: buffer full", event); + return + } + pending_events.intercepted_msgs.push(event); + } + + /// Processes any events asynchronously using the given handler. + /// + /// Note that the event handler is called in the order each event was generated, however + /// futures are polled in parallel for some events to allow for parallelism where events do not + /// have an ordering requirement. + /// + /// See the trait-level documentation of [`EventsProvider`] for requirements. + pub async fn process_pending_events_async + core::marker::Unpin, H: Fn(Event) -> Future>( + &self, handler: H + ) { + let mut intercepted_msgs = Vec::new(); + let mut peer_connecteds = Vec::new(); + { + let mut pending_events = self.pending_events.lock().unwrap(); + core::mem::swap(&mut pending_events.intercepted_msgs, &mut intercepted_msgs); + core::mem::swap(&mut pending_events.peer_connecteds, &mut peer_connecteds); + } + + let mut futures = Vec::with_capacity(intercepted_msgs.len()); + for (node_id, recipient) in self.message_recipients.lock().unwrap().iter_mut() { + if let OnionMessageRecipient::PendingConnection(_, addresses, _) = recipient { + if let Some(addresses) = addresses.take() { + futures.push(Some(handler(Event::ConnectionNeeded { node_id: *node_id, addresses }))); + } + } + } + + for ev in intercepted_msgs { + if let Event::OnionMessageIntercepted { .. } = ev {} else { debug_assert!(false); } + futures.push(Some(handler(ev))); + } + // Let the `OnionMessageIntercepted` events finish before moving on to peer_connecteds + crate::util::async_poll::MultiFuturePoller(futures).await; + + if peer_connecteds.len() <= 1 { + for event in peer_connecteds { handler(event).await; } + } else { + let mut futures = Vec::new(); + for event in peer_connecteds { + futures.push(Some(handler(event))); + } + crate::util::async_poll::MultiFuturePoller(futures).await; + } + } } fn outbound_buffer_full(peer_node_id: &PublicKey, buffer: &HashMap) -> bool { @@ -1047,6 +1379,24 @@ where } } } + let mut events = Vec::new(); + { + let mut pending_events = self.pending_events.lock().unwrap(); + #[cfg(debug_assertions)] { + for ev in pending_events.intercepted_msgs.iter() { + if let Event::OnionMessageIntercepted { .. } = ev {} else { panic!(); } + } + for ev in pending_events.peer_connecteds.iter() { + if let Event::OnionMessagePeerConnected { .. } = ev {} else { panic!(); } + } + } + core::mem::swap(&mut pending_events.intercepted_msgs, &mut events); + events.append(&mut pending_events.peer_connecteds); + pending_events.peer_connecteds.shrink_to(10); // Limit total heap usage + } + for ev in events { + handler.handle_event(ev); + } } } @@ -1076,14 +1426,14 @@ where |reply_path| Responder::new(reply_path, path_id) ); let response_instructions = self.offers_handler.handle_message(msg, responder); - self.handle_onion_message_response(response_instructions); + let _ = self.handle_onion_message_response(response_instructions); }, ParsedOnionMessageContents::Custom(msg) => { let responder = reply_path.map( |reply_path| Responder::new(reply_path, path_id) ); let response_instructions = self.custom_handler.handle_custom_message(msg, responder); - self.handle_onion_message_response(response_instructions); + let _ = self.handle_onion_message_response(response_instructions); }, } }, @@ -1120,6 +1470,13 @@ where e.get_mut().enqueue_message(onion_message); log_trace!(logger, "Forwarding an onion message to peer {}", next_node_id); }, + _ if self.intercept_messages_for_offline_peers => { + self.enqueue_intercepted_event( + Event::OnionMessageIntercepted { + peer_node_id: next_node_id, message: onion_message + } + ); + }, _ => { log_trace!( logger, @@ -1141,6 +1498,11 @@ where .entry(*their_node_id) .or_insert_with(|| OnionMessageRecipient::ConnectedPeer(VecDeque::new())) .mark_connected(); + if self.intercept_messages_for_offline_peers { + self.pending_events.lock().unwrap().peer_connecteds.push( + Event::OnionMessagePeerConnected { peer_node_id: *their_node_id } + ); + } } else { self.message_recipients.lock().unwrap().remove(their_node_id); }