use crate::blinded_path::BlindedPath;
use crate::blinded_path::message::{advance_path_by_one, ForwardTlvs, ReceiveTlvs};
use crate::blinded_path::utils;
+use crate::events::{Event, EventHandler, EventsProvider};
use crate::sign::{EntropySource, KeysManager, NodeSigner, Recipient};
#[cfg(not(c_bindings))]
use crate::ln::channelmanager::{SimpleArcChannelManager, SimpleRefChannelManager};
use crate::ln::msgs::{self, OnionMessage, OnionMessageHandler, SocketAddress};
use crate::ln::onion_utils;
use crate::ln::peer_handler::IgnoringMessageHandler;
-use crate::routing::gossip::NetworkGraph;
+use crate::routing::gossip::{NetworkGraph, NodeId};
pub use super::packet::OnionMessageContents;
use super::packet::ParsedOnionMessageContents;
use super::offers::OffersMessageHandler;
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
L::Target: Logger,
MR::Target: MessageRouter,
OMH::Target: OffersMessageHandler,
- CMH:: Target: CustomOnionMessageHandler,
+ CMH::Target: CustomOnionMessageHandler,
{
entropy_source: ES,
node_signer: NS,
/// Messages for a node connected as a peer.
ConnectedPeer(VecDeque<OnionMessage>),
- /// Messages for a node that is not yet connected.
- PendingConnection(VecDeque<OnionMessage>),
+ /// Messages for a node that is not yet connected, which are dropped after [`MAX_TIMER_TICKS`]
+ /// and tracked here.
+ PendingConnection(VecDeque<OnionMessage>, Option<Vec<SocketAddress>>, usize),
}
impl OnionMessageBuffer {
+ fn pending_connection(addresses: Vec<SocketAddress>) -> Self {
+ Self::PendingConnection(VecDeque::new(), Some(addresses), 0)
+ }
+
fn pending_messages(&self) -> &VecDeque<OnionMessage> {
match self {
OnionMessageBuffer::ConnectedPeer(pending_messages) => pending_messages,
- OnionMessageBuffer::PendingConnection(pending_messages) => pending_messages,
+ OnionMessageBuffer::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,
+ OnionMessageBuffer::PendingConnection(pending_messages, _, _) => pending_messages,
};
pending_messages.push_back(message);
fn dequeue_message(&mut self) -> Option<OnionMessage> {
let pending_messages = match self {
OnionMessageBuffer::ConnectedPeer(pending_messages) => pending_messages,
- OnionMessageBuffer::PendingConnection(pending_messages) => {
+ OnionMessageBuffer::PendingConnection(pending_messages, _, _) => {
debug_assert!(false);
pending_messages
},
fn release_pending_messages(&mut self) -> VecDeque<OnionMessage> {
let pending_messages = match self {
OnionMessageBuffer::ConnectedPeer(pending_messages) => pending_messages,
- OnionMessageBuffer::PendingConnection(pending_messages) => pending_messages,
+ OnionMessageBuffer::PendingConnection(pending_messages, _, _) => pending_messages,
};
core::mem::take(pending_messages)
}
fn mark_connected(&mut self) {
- if let OnionMessageBuffer::PendingConnection(pending_messages) = self {
+ if let OnionMessageBuffer::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);
fn find_path(
&self, _sender: PublicKey, peers: Vec<PublicKey>, destination: Destination
) -> Result<OnionMessagePath, ()> {
- if peers.contains(&destination.first_node()) {
+ let first_node = destination.first_node();
+ if peers.contains(&first_node) {
Ok(OnionMessagePath { intermediate_nodes: vec![], destination, addresses: None })
} else {
- Err(())
+ let network_graph = self.network_graph.deref().read_only();
+ 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 addresses = Some(node_announcement.addresses.clone());
+ Ok(OnionMessagePath { intermediate_nodes: vec![], destination, addresses })
+ },
+ _ => Err(()),
+ }
}
}
}
/// The provided [`Destination`] was an invalid [`BlindedPath`] due to not having any blinded
/// hops.
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`].
PathNotFound,
/// Onion message contents must have a TLV type >= 64.
pub fn create_onion_message<ES: Deref, NS: Deref, T: OnionMessageContents>(
entropy_source: &ES, node_signer: &NS, secp_ctx: &Secp256k1<secp256k1::All>,
path: OnionMessagePath, contents: T, reply_path: Option<BlindedPath>,
-) -> Result<(PublicKey, OnionMessage), SendError>
+) -> Result<(PublicKey, OnionMessage, Option<Vec<SocketAddress>>), SendError>
where
ES::Target: EntropySource,
NS::Target: NodeSigner,
{
- let OnionMessagePath { intermediate_nodes, mut destination, .. } = path;
+ let OnionMessagePath { intermediate_nodes, mut destination, addresses } = path;
if let Destination::BlindedPath(BlindedPath { ref blinded_hops, .. }) = destination {
if blinded_hops.is_empty() {
return Err(SendError::TooFewBlindedHops);
let onion_routing_packet = construct_onion_message_packet(
packet_payloads, packet_keys, prng_seed).map_err(|()| SendError::TooBigPacket)?;
- Ok((first_node_id, OnionMessage {
- blinding_point,
- onion_routing_packet
- }))
+ let message = OnionMessage { blinding_point, onion_routing_packet };
+ Ok((first_node_id, message, addresses))
}
/// Decode one layer of an incoming [`OnionMessage`].
) -> Result<SendSuccess, SendError> {
log_trace!(self.logger, "Constructing onion message {}: {:?}", log_suffix, contents);
- let (first_node_id, onion_message) = create_onion_message(
+ let (first_node_id, onion_message, addresses) = create_onion_message(
&self.entropy_source, &self.node_signer, &self.secp_ctx, path, contents, reply_path
)?;
}
match message_buffers.entry(first_node_id) {
- hash_map::Entry::Vacant(e) => {
- e.insert(OnionMessageBuffer::PendingConnection(VecDeque::new()))
- .enqueue_message(onion_message);
- Ok(SendSuccess::BufferedAwaitingConnection(first_node_id))
+ hash_map::Entry::Vacant(e) => match addresses {
+ None => Err(SendError::InvalidFirstHop(first_node_id)),
+ Some(addresses) => {
+ e.insert(OnionMessageBuffer::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);
false
}
+impl<ES: Deref, NS: Deref, L: Deref, MR: Deref, OMH: Deref, CMH: Deref> EventsProvider
+for OnionMessenger<ES, NS, L, MR, OMH, CMH>
+where
+ ES::Target: EntropySource,
+ NS::Target: NodeSigner,
+ L::Target: Logger,
+ MR::Target: MessageRouter,
+ OMH::Target: OffersMessageHandler,
+ CMH::Target: CustomOnionMessageHandler,
+{
+ fn process_pending_events<H: Deref>(&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 {
+ if let Some(addresses) = addresses.take() {
+ handler.handle_event(Event::ConnectionNeeded { node_id: *node_id, addresses });
+ }
+ }
+ }
+ }
+}
+
impl<ES: Deref, NS: Deref, L: Deref, MR: Deref, OMH: Deref, CMH: Deref> OnionMessageHandler
for OnionMessenger<ES, NS, L, MR, OMH, CMH>
where
}
}
+ fn timer_tick_occurred(&self) {
+ let mut message_buffers = self.message_buffers.lock().unwrap();
+
+ // Drop any pending recipients since the last call to avoid retaining buffered messages for
+ // too long.
+ message_buffers.retain(|_, recipient| match recipient {
+ OnionMessageBuffer::PendingConnection(_, None, ticks) => *ticks < MAX_TIMER_TICKS,
+ OnionMessageBuffer::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_buffers.values_mut() {
+ if let OnionMessageBuffer::PendingConnection(_, None, ticks) = recipient {
+ *ticks += 1;
+ }
+ }
+ }
+
fn provided_node_features(&self) -> NodeFeatures {
let mut features = NodeFeatures::empty();
features.set_onion_messages_optional();