X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=lightning%2Fsrc%2Fonion_message%2Fmessenger.rs;h=d333eb2103c5ca516217518ab144dab2f84b9807;hb=7d1745e7210975cc0fdc17b31072e33f0dfbf8e3;hp=d956b392a39631a6ce2ca761d0b457e7bb252da0;hpb=1c28cc071300005d9766fca3a402246973a811a5;p=rust-lightning diff --git a/lightning/src/onion_message/messenger.rs b/lightning/src/onion_message/messenger.rs index d956b392..d333eb21 100644 --- a/lightning/src/onion_message/messenger.rs +++ b/lightning/src/onion_message/messenger.rs @@ -15,8 +15,8 @@ use bitcoin::hashes::hmac::{Hmac, HmacEngine}; use bitcoin::hashes::sha256::Hash as Sha256; use bitcoin::secp256k1::{self, PublicKey, Scalar, Secp256k1, SecretKey}; -use crate::blinded_path::{BlindedPath, IntroductionNode, NodeIdLookUp}; -use crate::blinded_path::message::{advance_path_by_one, ForwardTlvs, NextHop, ReceiveTlvs}; +use crate::blinded_path::{BlindedPath, IntroductionNode, NextMessageHop, NodeIdLookUp}; +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}; @@ -135,6 +200,7 @@ pub(super) const MAX_TIMER_TICKS: usize = 2; /// # let your_custom_message_type = 42; /// your_custom_message_type /// } +/// fn msg_type(&self) -> &'static str { "YourCustomMessageType" } /// } /// // Send a custom onion message to a node id. /// let destination = Destination::Node(destination_node_id); @@ -144,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,7 +245,12 @@ where offers_handler: OMH, custom_handler: CMH, intercept_messages_for_offline_peers: bool, - pending_events: Mutex>, + pending_events: Mutex, +} + +struct PendingEvents { + intercepted_msgs: Vec, + peer_connecteds: Vec, } /// [`OnionMessage`]s buffered to be sent. @@ -248,6 +322,72 @@ 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. + pub(super) fn new(reply_path: BlindedPath, path_id: Option<[u8; 32]>) -> Self { + Responder { + reply_path, + path_id, + } + } + + /// 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, + reply_path: self.reply_path, + 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. +pub struct OnionMessageResponse { + message: T, + reply_path: BlindedPath, + path_id: Option<[u8; 32]>, +} + +/// `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), + /// Indicates that there's no response to send back. + NoResponse, +} + /// An [`OnionMessage`] for [`OnionMessenger`] to send. /// /// These are obtained when released from [`OnionMessenger`]'s handlers after which they are @@ -294,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, @@ -315,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: ExactSizeIterator, 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; @@ -372,13 +505,18 @@ where let is_recipient_announced = network_graph.nodes().contains_key(&NodeId::from_pubkey(&recipient)); - let mut peer_info = peers.iter() - // Limit to peers with announced channels - .filter_map(|pubkey| + let has_one_peer = peers.len() == 1; + let mut peer_info = peers + // Limit to peers with announced channels unless the recipient is unannounced. + .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())) + // Allow messages directly with the only peer when unannounced. + .or_else(|| (!is_recipient_announced && has_one_peer) + .then(|| (peer, false, 0)) + ) ) // Exclude Tor-only nodes when the recipient is announced. .filter(|(_, is_tor_only, _)| !(*is_tor_only && is_recipient_announced)) @@ -390,12 +528,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 { @@ -405,7 +544,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) } } @@ -509,7 +714,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, @@ -548,7 +757,7 @@ pub trait CustomOnionMessageHandler { /// Called with the custom message that was received, returning a response to send, if any. /// /// The returned [`Self::CustomMessage`], if any, is enqueued to be sent by [`OnionMessenger`]. - fn handle_custom_message(&self, msg: Self::CustomMessage) -> Option; + fn handle_custom_message(&self, message: Self::CustomMessage, responder: Option) -> ResponseInstruction; /// Read a custom message of type `message_type` from `buffer`, returning `Ok(None)` if the /// message type is unknown. @@ -571,10 +780,10 @@ pub trait CustomOnionMessageHandler { /// A processed incoming onion message, containing either a Forward (another onion message) /// or a Receive payload with decrypted contents. -#[derive(Debug)] +#[derive(Clone, Debug)] pub enum PeeledOnion { /// Forwarded onion, with the next node id and a new onion - Forward(NextHop, OnionMessage), + Forward(NextMessageHop, OnionMessage), /// Received onion message, with decrypted contents, path_id, and reply path Receive(ParsedOnionMessageContents, Option<[u8; 32]>, Option) } @@ -805,7 +1014,27 @@ where ) } + /// 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 @@ -834,7 +1063,10 @@ where offers_handler, custom_handler, intercept_messages_for_offline_peers, - pending_events: Mutex::new(Vec::new()), + pending_events: Mutex::new(PendingEvents { + intercepted_msgs: Vec::new(), + peer_connecteds: Vec::new(), + }), } } @@ -858,13 +1090,12 @@ where &self, contents: T, destination: Destination, reply_path: Option, log_suffix: fmt::Arguments ) -> Result { - let mut logger = WithContext::from(&self.logger, None, None); - let result = self.find_path(destination) - .and_then(|path| { - let first_hop = path.intermediate_nodes.get(0).map(|p| *p); - logger = WithContext::from(&self.logger, first_hop, None); - self.enqueue_onion_message(path, contents, reply_path, log_suffix) - }); + let mut logger = WithContext::from(&self.logger, None, None, None); + let result = self.find_path(destination).and_then(|path| { + let first_hop = path.intermediate_nodes.get(0).map(|p| *p); + logger = WithContext::from(&self.logger, first_hop, None, None); + self.enqueue_onion_message(path, contents, reply_path, log_suffix) + }); match result.as_ref() { Err(SendError::GetNodeIdFailed) => { @@ -907,6 +1138,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 @@ -943,6 +1192,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 @@ -958,21 +1228,48 @@ where ) } - fn handle_onion_message_response( - &self, response: Option, reply_path: Option, log_suffix: fmt::Arguments - ) { - if let Some(response) = response { - match reply_path { - Some(reply_path) => { - let _ = self.find_path_and_enqueue_onion_message( - response, Destination::BlindedPath(reply_path), None, log_suffix + /// 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 + ) -> 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 ); - }, - None => { - log_trace!(self.logger, "Missing reply path {}", log_suffix); - }, + 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)] @@ -986,6 +1283,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 { @@ -1031,7 +1385,20 @@ where } } let mut events = Vec::new(); - core::mem::swap(&mut *self.pending_events.lock().unwrap(), &mut events); + { + 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); } @@ -1050,7 +1417,7 @@ where CMH::Target: CustomOnionMessageHandler, { fn handle_onion_message(&self, peer_node_id: &PublicKey, msg: &OnionMessage) { - let logger = WithContext::from(&self.logger, Some(*peer_node_id), None); + let logger = WithContext::from(&self.logger, Some(*peer_node_id), None, None); match self.peel_onion_message(msg) { Ok(PeeledOnion::Receive(message, path_id, reply_path)) => { log_trace!( @@ -1060,29 +1427,25 @@ where match message { ParsedOnionMessageContents::Offers(msg) => { - let response = self.offers_handler.handle_message(msg); - self.handle_onion_message_response( - response, reply_path, format_args!( - "when responding to Offers onion message with path_id {:02x?}", - path_id - ) + let responder = reply_path.map( + |reply_path| Responder::new(reply_path, path_id) ); + let response_instructions = self.offers_handler.handle_message(msg, responder); + let _ = self.handle_onion_message_response(response_instructions); }, ParsedOnionMessageContents::Custom(msg) => { - let response = self.custom_handler.handle_custom_message(msg); - self.handle_onion_message_response( - response, reply_path, format_args!( - "when responding to Custom onion message with path_id {:02x?}", - path_id - ) + let responder = reply_path.map( + |reply_path| Responder::new(reply_path, path_id) ); + let response_instructions = self.custom_handler.handle_custom_message(msg, responder); + let _ = self.handle_onion_message_response(response_instructions); }, } }, Ok(PeeledOnion::Forward(next_hop, onion_message)) => { let next_node_id = match next_hop { - NextHop::NodeId(pubkey) => pubkey, - NextHop::ShortChannelId(scid) => match self.node_id_lookup.next_node_id(scid) { + NextMessageHop::NodeId(pubkey) => pubkey, + NextMessageHop::ShortChannelId(scid) => match self.node_id_lookup.next_node_id(scid) { Some(pubkey) => pubkey, None => { log_trace!(self.logger, "Dropping forwarded onion messager: unable to resolve next hop using SCID {}", scid); @@ -1113,7 +1476,7 @@ where log_trace!(logger, "Forwarding an onion message to peer {}", next_node_id); }, _ if self.intercept_messages_for_offline_peers => { - self.pending_events.lock().unwrap().push( + self.enqueue_intercepted_event( Event::OnionMessageIntercepted { peer_node_id: next_node_id, message: onion_message } @@ -1141,7 +1504,7 @@ where .or_insert_with(|| OnionMessageRecipient::ConnectedPeer(VecDeque::new())) .mark_connected(); if self.intercept_messages_for_offline_peers { - self.pending_events.lock().unwrap().push( + self.pending_events.lock().unwrap().peer_connecteds.push( Event::OnionMessagePeerConnected { peer_node_id: *their_node_id } ); } @@ -1298,7 +1661,7 @@ fn packet_payloads_and_keys