X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=lightning%2Fsrc%2Fonion_message%2Fmessenger.rs;h=8a44eb2a5beebf3128582bdbce23cc8edfbda381;hb=2b04f193b9dc8eb6d0d0e5b513a47e668324debd;hp=c2e2bc0292a9a00495d57eb397cfa6c72a13f924;hpb=06b05df75533bbbe1400bb3efca7e97cff78146f;p=rust-lightning diff --git a/lightning/src/onion_message/messenger.rs b/lightning/src/onion_message/messenger.rs index c2e2bc02..8a44eb2a 100644 --- a/lightning/src/onion_message/messenger.rs +++ b/lightning/src/onion_message/messenger.rs @@ -40,6 +40,8 @@ use crate::io; use crate::sync::{Arc, Mutex}; use crate::prelude::*; +pub(super) const MAX_TIMER_TICKS: usize = 2; + /// A sender, receiver and forwarder of [`OnionMessage`]s. /// /// # Handling Messages @@ -85,7 +87,7 @@ use crate::prelude::*; /// # Ok(OnionMessagePath { /// # intermediate_nodes: vec![hop_node_id1, hop_node_id2], /// # destination, -/// # addresses: None, +/// # first_node_addresses: None, /// # }) /// # } /// # } @@ -149,12 +151,12 @@ where L::Target: Logger, MR::Target: MessageRouter, OMH::Target: OffersMessageHandler, - CMH:: Target: CustomOnionMessageHandler, + CMH::Target: CustomOnionMessageHandler, { entropy_source: ES, node_signer: NS, logger: L, - message_buffers: Mutex>, + message_recipients: Mutex>, secp_ctx: Secp256k1, message_router: MR, offers_handler: OMH, @@ -162,26 +164,31 @@ where } /// [`OnionMessage`]s buffered to be sent. -enum OnionMessageBuffer { +enum OnionMessageRecipient { /// Messages for a node connected as a peer. ConnectedPeer(VecDeque), - /// Messages for a node that is not yet connected. - PendingConnection(VecDeque, Option>), + /// Messages for a node that is not yet connected, which are dropped after [`MAX_TIMER_TICKS`] + /// and tracked here. + PendingConnection(VecDeque, Option>, usize), } -impl OnionMessageBuffer { +impl OnionMessageRecipient { + fn pending_connection(addresses: Vec) -> Self { + Self::PendingConnection(VecDeque::new(), Some(addresses), 0) + } + fn pending_messages(&self) -> &VecDeque { match self { - OnionMessageBuffer::ConnectedPeer(pending_messages) => pending_messages, - OnionMessageBuffer::PendingConnection(pending_messages, _) => pending_messages, + OnionMessageRecipient::ConnectedPeer(pending_messages) => pending_messages, + OnionMessageRecipient::PendingConnection(pending_messages, _, _) => pending_messages, } } fn enqueue_message(&mut self, message: OnionMessage) { let pending_messages = match self { - OnionMessageBuffer::ConnectedPeer(pending_messages) => pending_messages, - OnionMessageBuffer::PendingConnection(pending_messages, _) => pending_messages, + OnionMessageRecipient::ConnectedPeer(pending_messages) => pending_messages, + OnionMessageRecipient::PendingConnection(pending_messages, _, _) => pending_messages, }; pending_messages.push_back(message); @@ -189,8 +196,8 @@ impl OnionMessageBuffer { fn dequeue_message(&mut self) -> Option { let pending_messages = match self { - OnionMessageBuffer::ConnectedPeer(pending_messages) => pending_messages, - OnionMessageBuffer::PendingConnection(pending_messages, _) => { + OnionMessageRecipient::ConnectedPeer(pending_messages) => pending_messages, + OnionMessageRecipient::PendingConnection(pending_messages, _, _) => { debug_assert!(false); pending_messages }, @@ -202,18 +209,25 @@ impl OnionMessageBuffer { #[cfg(test)] fn release_pending_messages(&mut self) -> VecDeque { let pending_messages = match self { - OnionMessageBuffer::ConnectedPeer(pending_messages) => pending_messages, - OnionMessageBuffer::PendingConnection(pending_messages, _) => pending_messages, + OnionMessageRecipient::ConnectedPeer(pending_messages) => pending_messages, + OnionMessageRecipient::PendingConnection(pending_messages, _, _) => pending_messages, }; core::mem::take(pending_messages) } fn mark_connected(&mut self) { - if let OnionMessageBuffer::PendingConnection(pending_messages, _) = self { + if let OnionMessageRecipient::PendingConnection(pending_messages, _, _) = self { let mut new_pending_messages = VecDeque::new(); core::mem::swap(pending_messages, &mut new_pending_messages); - *self = OnionMessageBuffer::ConnectedPeer(new_pending_messages); + *self = OnionMessageRecipient::ConnectedPeer(new_pending_messages); + } + } + + fn is_connected(&self) -> bool { + match self { + OnionMessageRecipient::ConnectedPeer(..) => true, + OnionMessageRecipient::PendingConnection(..) => false, } } } @@ -285,7 +299,9 @@ where ) -> Result { let first_node = destination.first_node(); if peers.contains(&first_node) { - Ok(OnionMessagePath { intermediate_nodes: vec![], destination, addresses: None }) + Ok(OnionMessagePath { + intermediate_nodes: vec![], destination, first_node_addresses: None + }) } else { let network_graph = self.network_graph.deref().read_only(); let node_announcement = network_graph @@ -296,8 +312,10 @@ where match node_announcement { Some(node_announcement) if node_announcement.features.supports_onion_messages() => { - let addresses = Some(node_announcement.addresses.clone()); - Ok(OnionMessagePath { intermediate_nodes: vec![], destination, addresses }) + let first_node_addresses = Some(node_announcement.addresses.clone()); + Ok(OnionMessagePath { + intermediate_nodes: vec![], destination, first_node_addresses + }) }, _ => Err(()), } @@ -318,7 +336,7 @@ pub struct OnionMessagePath { /// /// Only needs to be set if a connection to the node is required. [`OnionMessenger`] may use /// this to initiate such a connection. - pub addresses: Option>, + pub first_node_addresses: Option>, } impl OnionMessagePath { @@ -452,7 +470,8 @@ pub enum PeeledOnion { /// Creates an [`OnionMessage`] with the given `contents` for sending to the destination of /// `path`. /// -/// Returns both the node id of the peer to send the message to and the message itself. +/// Returns the node id of the peer to send the message to, the message itself, and any addresses +/// need to connect to the first node. pub fn create_onion_message( entropy_source: &ES, node_signer: &NS, secp_ctx: &Secp256k1, path: OnionMessagePath, contents: T, reply_path: Option, @@ -461,7 +480,7 @@ where ES::Target: EntropySource, NS::Target: NodeSigner, { - let OnionMessagePath { intermediate_nodes, mut destination, addresses } = path; + let OnionMessagePath { intermediate_nodes, mut destination, first_node_addresses } = path; if let Destination::BlindedPath(BlindedPath { ref blinded_hops, .. }) = destination { if blinded_hops.is_empty() { return Err(SendError::TooFewBlindedHops); @@ -503,7 +522,7 @@ where packet_payloads, packet_keys, prng_seed).map_err(|()| SendError::TooBigPacket)?; let message = OnionMessage { blinding_point, onion_routing_packet }; - Ok((first_node_id, message, addresses)) + Ok((first_node_id, message, first_node_addresses)) } /// Decode one layer of an incoming [`OnionMessage`]. @@ -624,7 +643,7 @@ where OnionMessenger { entropy_source, node_signer, - message_buffers: Mutex::new(HashMap::new()), + message_recipients: Mutex::new(HashMap::new()), secp_ctx, logger, message_router, @@ -680,9 +699,9 @@ where .get_node_id(Recipient::Node) .map_err(|_| SendError::GetNodeIdFailed)?; - let peers = self.message_buffers.lock().unwrap() + let peers = self.message_recipients.lock().unwrap() .iter() - .filter(|(_, buffer)| matches!(buffer, OnionMessageBuffer::ConnectedPeer(_))) + .filter(|(_, recipient)| matches!(recipient, OnionMessageRecipient::ConnectedPeer(_))) .map(|(node_id, _)| *node_id) .collect(); @@ -701,24 +720,27 @@ where &self.entropy_source, &self.node_signer, &self.secp_ctx, path, contents, reply_path )?; - let mut message_buffers = self.message_buffers.lock().unwrap(); - if outbound_buffer_full(&first_node_id, &message_buffers) { + let mut message_recipients = self.message_recipients.lock().unwrap(); + if outbound_buffer_full(&first_node_id, &message_recipients) { return Err(SendError::BufferFull); } - match message_buffers.entry(first_node_id) { + match message_recipients.entry(first_node_id) { hash_map::Entry::Vacant(e) => match addresses { None => Err(SendError::InvalidFirstHop(first_node_id)), Some(addresses) => { - e.insert( - OnionMessageBuffer::PendingConnection(VecDeque::new(), Some(addresses)) - ).enqueue_message(onion_message); + e.insert(OnionMessageRecipient::pending_connection(addresses)) + .enqueue_message(onion_message); Ok(SendSuccess::BufferedAwaitingConnection(first_node_id)) }, }, hash_map::Entry::Occupied(mut e) => { e.get_mut().enqueue_message(onion_message); - Ok(SendSuccess::Buffered) + if e.get().is_connected() { + Ok(SendSuccess::Buffered) + } else { + Ok(SendSuccess::BufferedAwaitingConnection(first_node_id)) + } }, } } @@ -749,18 +771,18 @@ where #[cfg(test)] pub(super) fn release_pending_msgs(&self) -> HashMap> { - let mut message_buffers = self.message_buffers.lock().unwrap(); + let mut message_recipients = self.message_recipients.lock().unwrap(); let mut msgs = HashMap::new(); // We don't want to disconnect the peers by removing them entirely from the original map, so we // release the pending message buffers individually. - for (peer_node_id, buffer) in &mut *message_buffers { - msgs.insert(*peer_node_id, buffer.release_pending_messages()); + for (node_id, recipient) in &mut *message_recipients { + msgs.insert(*node_id, recipient.release_pending_messages()); } msgs } } -fn outbound_buffer_full(peer_node_id: &PublicKey, buffer: &HashMap) -> bool { +fn outbound_buffer_full(peer_node_id: &PublicKey, buffer: &HashMap) -> bool { const MAX_TOTAL_BUFFER_SIZE: usize = (1 << 20) * 128; const MAX_PER_PEER_BUFFER_SIZE: usize = (1 << 10) * 256; let mut total_buffered_bytes = 0; @@ -794,8 +816,8 @@ where CMH::Target: CustomOnionMessageHandler, { fn process_pending_events(&self, handler: H) where H::Target: EventHandler { - for (node_id, recipient) in self.message_buffers.lock().unwrap().iter_mut() { - if let OnionMessageBuffer::PendingConnection(_, addresses) = recipient { + for (node_id, recipient) in self.message_recipients.lock().unwrap().iter_mut() { + if let OnionMessageRecipient::PendingConnection(_, addresses, _) = recipient { if let Some(addresses) = addresses.take() { handler.handle_event(Event::ConnectionNeeded { node_id: *node_id, addresses }); } @@ -846,20 +868,20 @@ where } }, Ok(PeeledOnion::Forward(next_node_id, onion_message)) => { - let mut message_buffers = self.message_buffers.lock().unwrap(); - if outbound_buffer_full(&next_node_id, &message_buffers) { + let mut message_recipients = self.message_recipients.lock().unwrap(); + if outbound_buffer_full(&next_node_id, &message_recipients) { log_trace!(self.logger, "Dropping forwarded onion message to peer {:?}: outbound buffer full", next_node_id); return } #[cfg(fuzzing)] - message_buffers + message_recipients .entry(next_node_id) - .or_insert_with(|| OnionMessageBuffer::ConnectedPeer(VecDeque::new())); + .or_insert_with(|| OnionMessageRecipient::ConnectedPeer(VecDeque::new())); - match message_buffers.entry(next_node_id) { + match message_recipients.entry(next_node_id) { hash_map::Entry::Occupied(mut e) if matches!( - e.get(), OnionMessageBuffer::ConnectedPeer(..) + e.get(), OnionMessageRecipient::ConnectedPeer(..) ) => { e.get_mut().enqueue_message(onion_message); log_trace!(self.logger, "Forwarding an onion message to peer {}", next_node_id); @@ -878,21 +900,42 @@ where fn peer_connected(&self, their_node_id: &PublicKey, init: &msgs::Init, _inbound: bool) -> Result<(), ()> { if init.features.supports_onion_messages() { - self.message_buffers.lock().unwrap() + self.message_recipients.lock().unwrap() .entry(*their_node_id) - .or_insert_with(|| OnionMessageBuffer::ConnectedPeer(VecDeque::new())) + .or_insert_with(|| OnionMessageRecipient::ConnectedPeer(VecDeque::new())) .mark_connected(); } else { - self.message_buffers.lock().unwrap().remove(their_node_id); + self.message_recipients.lock().unwrap().remove(their_node_id); } Ok(()) } fn peer_disconnected(&self, their_node_id: &PublicKey) { - match self.message_buffers.lock().unwrap().remove(their_node_id) { - Some(OnionMessageBuffer::ConnectedPeer(..)) => {}, - _ => debug_assert!(false), + match self.message_recipients.lock().unwrap().remove(their_node_id) { + Some(OnionMessageRecipient::ConnectedPeer(..)) => {}, + Some(_) => debug_assert!(false), + None => {}, + } + } + + fn timer_tick_occurred(&self) { + let mut message_recipients = self.message_recipients.lock().unwrap(); + + // Drop any pending recipients since the last call to avoid retaining buffered messages for + // too long. + message_recipients.retain(|_, recipient| match recipient { + OnionMessageRecipient::PendingConnection(_, None, ticks) => *ticks < MAX_TIMER_TICKS, + OnionMessageRecipient::PendingConnection(_, Some(_), _) => true, + _ => true, + }); + + // Increment a timer tick for pending recipients so that their buffered messages are dropped + // at MAX_TIMER_TICKS. + for recipient in message_recipients.values_mut() { + if let OnionMessageRecipient::PendingConnection(_, None, ticks) = recipient { + *ticks += 1; + } } } @@ -934,7 +977,7 @@ where ); } - self.message_buffers.lock().unwrap() + self.message_recipients.lock().unwrap() .get_mut(&peer_node_id) .and_then(|buffer| buffer.dequeue_message()) }