Switch to use NetAddress for peer addresses

While we're still blocked on upstream changes, we now switch our peer info
to use a newtype around `NetAddress` so that we won't have to break
serialization compatibility when the upstream changes becom available
post-0.1.

Author: tnull

README.md

@@ -8,7 +8,7 @@ LDK Node is a non-custodial Lightning node in library form. Its central goal is
 The primary abstraction of the library is the `Node`, which can be retrieved by setting up and configuring a `Builder` to your liking and calling `build()`. `Node` can then be controlled via commands such as `start`, `stop`, `connect_open_channel`, `send_payment`, etc.:
 
 ```rust
-use ldk_node::Builder;
+use ldk_node::{Builder, NetAddress};
 use ldk_node::lightning_invoice::Invoice;
 use ldk_node::bitcoin::secp256k1::PublicKey;
 use std::str::FromStr;
@@ -28,7 +28,7 @@ fn main() {
 	node.sync_wallets().unwrap();
 
 	let node_id = PublicKey::from_str("NODE_ID").unwrap();
-	let node_addr = "IP_ADDR:PORT".parse().unwrap();
+	let node_addr = NetAddress::from_str("IP_ADDR:PORT").unwrap();
 	node.connect_open_channel(node_id, node_addr, 10000, None, false).unwrap();
 
 	let invoice = Invoice::from_str("INVOICE_STR").unwrap();

src/error.rs

@@ -25,6 +25,8 @@ pub enum Error {
 	PaymentFailed,
 	/// A given peer info could not be parsed.
 	PeerInfoParseFailed,
+	/// A given peer info could not be found.
+	PeerInfoNotFound,
 	/// A channel could not be opened.
 	ChannelCreationFailed,
 	/// A channel could not be closed.
@@ -57,6 +59,7 @@ impl fmt::Display for Error {
 			}
 			Self::PaymentFailed => write!(f, "Failed to send the given payment."),
 			Self::PeerInfoParseFailed => write!(f, "Failed to parse the given peer information."),
+			Self::PeerInfoNotFound => write!(f, "Failed to resolve the given peer information."),
 			Self::ChannelCreationFailed => write!(f, "Failed to create channel."),
 			Self::ChannelClosingFailed => write!(f, "Failed to close channel."),
 			Self::PersistenceFailed => write!(f, "Failed to persist data."),

src/lib.rs

@@ -88,6 +88,8 @@ pub use lightning_invoice;
 
 pub use error::Error;
 pub use event::Event;
+pub use types::NetAddress;
+
 use event::{EventHandler, EventQueue};
 use io::fs_store::FilesystemStore;
 use io::{KVStore, CHANNEL_MANAGER_PERSISTENCE_KEY, CHANNEL_MANAGER_PERSISTENCE_NAMESPACE};
@@ -140,7 +142,7 @@ use rand::Rng;
 use std::convert::TryInto;
 use std::default::Default;
 use std::fs;
-use std::net::SocketAddr;
+use std::net::{SocketAddr, ToSocketAddrs};
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex, RwLock};
 use std::time::{Duration, Instant, SystemTime};
@@ -754,7 +756,7 @@ impl Node {
 				{
 					if let Some(peer_info) = connect_peer_store.get_peer(&node_id) {
 						let _ = do_connect_peer(
-							peer_info.pubkey,
+							peer_info.node_id,
 							peer_info.address,
 							Arc::clone(&connect_pm),
 							Arc::clone(&connect_logger),
@@ -896,18 +898,18 @@ impl Node {
 	///
 	/// If `permanently` is set to `true`, we'll remember the peer and reconnect to it on restart.
 	pub fn connect(
-		&self, node_id: PublicKey, address: SocketAddr, permanently: bool,
+		&self, node_id: PublicKey, address: NetAddress, permanently: bool,
 	) -> Result<(), Error> {
 		let rt_lock = self.runtime.read().unwrap();
 		if rt_lock.is_none() {
 			return Err(Error::NotRunning);
 		}
 		let runtime = rt_lock.as_ref().unwrap();
 
-		let peer_info = PeerInfo { pubkey: node_id, address };
+		let peer_info = PeerInfo { node_id, address };
 
-		let con_peer_pubkey = peer_info.pubkey;
-		let con_peer_addr = peer_info.address;
+		let con_node_id = peer_info.node_id;
+		let con_addr = peer_info.address.clone();
 		let con_success = Arc::new(AtomicBool::new(false));
 		let con_success_cloned = Arc::clone(&con_success);
 		let con_logger = Arc::clone(&self.logger);
@@ -916,8 +918,7 @@ impl Node {
 		tokio::task::block_in_place(move || {
 			runtime.block_on(async move {
 				let res =
-					connect_peer_if_necessary(con_peer_pubkey, con_peer_addr, con_pm, con_logger)
-						.await;
+					connect_peer_if_necessary(con_node_id, con_addr, con_pm, con_logger).await;
 				con_success_cloned.store(res.is_ok(), Ordering::Release);
 			})
 		});
@@ -926,7 +927,7 @@ impl Node {
 			return Err(Error::ConnectionFailed);
 		}
 
-		log_info!(self.logger, "Connected to peer {}@{}. ", peer_info.pubkey, peer_info.address,);
+		log_info!(self.logger, "Connected to peer {}@{}. ", peer_info.node_id, peer_info.address);
 
 		if permanently {
 			self.peer_store.add_peer(peer_info)?;
@@ -968,7 +969,7 @@ impl Node {
 	///
 	/// Returns a temporary channel id.
 	pub fn connect_open_channel(
-		&self, node_id: PublicKey, address: SocketAddr, channel_amount_sats: u64,
+		&self, node_id: PublicKey, address: NetAddress, channel_amount_sats: u64,
 		push_to_counterparty_msat: Option<u64>, announce_channel: bool,
 	) -> Result<(), Error> {
 		let rt_lock = self.runtime.read().unwrap();
@@ -983,10 +984,10 @@ impl Node {
 			return Err(Error::InsufficientFunds);
 		}
 
-		let peer_info = PeerInfo { pubkey: node_id, address };
+		let peer_info = PeerInfo { node_id, address };
 
-		let con_peer_pubkey = peer_info.pubkey;
-		let con_peer_addr = peer_info.address;
+		let con_node_id = peer_info.node_id;
+		let con_addr = peer_info.address.clone();
 		let con_success = Arc::new(AtomicBool::new(false));
 		let con_success_cloned = Arc::clone(&con_success);
 		let con_logger = Arc::clone(&self.logger);
@@ -995,8 +996,7 @@ impl Node {
 		tokio::task::block_in_place(move || {
 			runtime.block_on(async move {
 				let res =
-					connect_peer_if_necessary(con_peer_pubkey, con_peer_addr, con_pm, con_logger)
-						.await;
+					connect_peer_if_necessary(con_node_id, con_addr, con_pm, con_logger).await;
 				con_success_cloned.store(res.is_ok(), Ordering::Release);
 			})
 		});
@@ -1022,7 +1022,7 @@ impl Node {
 		let user_channel_id: u128 = rand::thread_rng().gen::<u128>();
 
 		match self.channel_manager.create_channel(
-			peer_info.pubkey,
+			peer_info.node_id,
 			channel_amount_sats,
 			push_msat,
 			user_channel_id,
@@ -1032,7 +1032,7 @@ impl Node {
 				log_info!(
 					self.logger,
 					"Initiated channel creation with peer {}. ",
-					peer_info.pubkey
+					peer_info.node_id
 				);
 				self.peer_store.add_peer(peer_info)?;
 				Ok(())
@@ -1438,44 +1438,52 @@ impl Drop for Node {
 }
 
 async fn connect_peer_if_necessary(
-	pubkey: PublicKey, peer_addr: SocketAddr, peer_manager: Arc<PeerManager>,
+	node_id: PublicKey, addr: NetAddress, peer_manager: Arc<PeerManager>,
 	logger: Arc<FilesystemLogger>,
 ) -> Result<(), Error> {
-	for (node_pubkey, _addr) in peer_manager.get_peer_node_ids() {
-		if node_pubkey == pubkey {
+	for (pman_node_id, _pman_addr) in peer_manager.get_peer_node_ids() {
+		if node_id == pman_node_id {
 			return Ok(());
 		}
 	}
 
-	do_connect_peer(pubkey, peer_addr, peer_manager, logger).await
+	do_connect_peer(node_id, addr, peer_manager, logger).await
 }
 
 async fn do_connect_peer(
-	pubkey: PublicKey, peer_addr: SocketAddr, peer_manager: Arc<PeerManager>,
+	node_id: PublicKey, addr: NetAddress, peer_manager: Arc<PeerManager>,
 	logger: Arc<FilesystemLogger>,
 ) -> Result<(), Error> {
-	log_info!(logger, "Connecting to peer: {}@{}", pubkey, peer_addr);
-	match lightning_net_tokio::connect_outbound(Arc::clone(&peer_manager), pubkey, peer_addr).await
+	log_info!(logger, "Connecting to peer: {}@{}", node_id, addr);
+
+	let socket_addr = addr
+		.to_socket_addrs()
+		.map_err(|_| Error::PeerInfoNotFound)?
+		.next()
+		.ok_or(Error::ConnectionFailed)?;
+
+	match lightning_net_tokio::connect_outbound(Arc::clone(&peer_manager), node_id, socket_addr)
+		.await
 	{
 		Some(connection_closed_future) => {
 			let mut connection_closed_future = Box::pin(connection_closed_future);
 			loop {
 				match futures::poll!(&mut connection_closed_future) {
 					std::task::Poll::Ready(_) => {
-						log_info!(logger, "Peer connection closed: {}@{}", pubkey, peer_addr);
+						log_info!(logger, "Peer connection closed: {}@{}", node_id, addr);
 						return Err(Error::ConnectionFailed);
 					}
 					std::task::Poll::Pending => {}
 				}
 				// Avoid blocking the tokio context by sleeping a bit
-				match peer_manager.get_peer_node_ids().iter().find(|(id, _addr)| *id == pubkey) {
+				match peer_manager.get_peer_node_ids().iter().find(|(id, _addr)| *id == node_id) {
 					Some(_) => return Ok(()),
 					None => tokio::time::sleep(Duration::from_millis(10)).await,
 				}
 			}
 		}
 		None => {
-			log_error!(logger, "Failed to connect to peer: {}@{}", pubkey, peer_addr);
+			log_error!(logger, "Failed to connect to peer: {}@{}", node_id, addr);
 			Err(Error::ConnectionFailed)
 		}
 	}

src/peer_store.rs

@@ -2,14 +2,14 @@ use crate::io::{
 	KVStore, TransactionalWrite, PEER_INFO_PERSISTENCE_KEY, PEER_INFO_PERSISTENCE_NAMESPACE,
 };
 use crate::logger::{log_error, Logger};
-use crate::Error;
+use crate::{Error, NetAddress};
 
+use lightning::impl_writeable_tlv_based;
 use lightning::util::ser::{Readable, ReadableArgs, Writeable, Writer};
 
 use bitcoin::secp256k1::PublicKey;
 
 use std::collections::HashMap;
-use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
 use std::ops::Deref;
 use std::sync::RwLock;
 
@@ -36,23 +36,23 @@ where
 	pub(crate) fn add_peer(&self, peer_info: PeerInfo) -> Result<(), Error> {
 		let mut locked_peers = self.peers.write().unwrap();
 
-		locked_peers.insert(peer_info.pubkey, peer_info);
+		locked_peers.insert(peer_info.node_id, peer_info);
 		self.write_peers_and_commit(&*locked_peers)
 	}
 
-	pub(crate) fn remove_peer(&self, peer_pubkey: &PublicKey) -> Result<(), Error> {
+	pub(crate) fn remove_peer(&self, node_id: &PublicKey) -> Result<(), Error> {
 		let mut locked_peers = self.peers.write().unwrap();
 
-		locked_peers.remove(peer_pubkey);
+		locked_peers.remove(node_id);
 		self.write_peers_and_commit(&*locked_peers)
 	}
 
 	pub(crate) fn list_peers(&self) -> Vec<PeerInfo> {
 		self.peers.read().unwrap().values().cloned().collect()
 	}
 
-	pub(crate) fn get_peer(&self, peer_pubkey: &PublicKey) -> Option<PeerInfo> {
-		self.peers.read().unwrap().get(peer_pubkey).cloned()
+	pub(crate) fn get_peer(&self, node_id: &PublicKey) -> Option<PeerInfo> {
+		self.peers.read().unwrap().get(node_id).cloned()
 	}
 
 	fn write_peers_and_commit(
@@ -143,56 +143,14 @@ impl Writeable for PeerStoreSerWrapper<'_> {
 
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub(crate) struct PeerInfo {
-	pub pubkey: PublicKey,
-	pub address: SocketAddr,
+	pub node_id: PublicKey,
+	pub address: NetAddress,
 }
 
-impl Readable for PeerInfo {
-	fn read<R: lightning::io::Read>(
-		reader: &mut R,
-	) -> Result<Self, lightning::ln::msgs::DecodeError> {
-		let pubkey = Readable::read(reader)?;
-
-		let ip_type: u8 = Readable::read(reader)?;
-
-		let ip_addr = if ip_type == 0 {
-			let v4bytes: u32 = Readable::read(reader)?;
-			let v4addr = Ipv4Addr::from(v4bytes);
-			IpAddr::from(v4addr)
-		} else {
-			let v6bytes: u128 = Readable::read(reader)?;
-			let v6addr = Ipv6Addr::from(v6bytes);
-			IpAddr::from(v6addr)
-		};
-
-		let port: u16 = Readable::read(reader)?;
-
-		let address = SocketAddr::new(ip_addr, port);
-
-		Ok(PeerInfo { pubkey, address })
-	}
-}
-
-impl Writeable for PeerInfo {
-	fn write<W: Writer>(&self, writer: &mut W) -> Result<(), lightning::io::Error> {
-		self.pubkey.write(writer)?;
-
-		match self.address.ip() {
-			IpAddr::V4(v4addr) => {
-				0u8.write(writer)?;
-				u32::from(v4addr).write(writer)?;
-			}
-			IpAddr::V6(v6addr) => {
-				1u8.write(writer)?;
-				u128::from(v6addr).write(writer)?;
-			}
-		}
-
-		self.address.port().write(writer)?;
-
-		Ok(())
-	}
-}
+impl_writeable_tlv_based!(PeerInfo, {
+	(0, node_id, required),
+	(2, address, required),
+});
 
 #[cfg(test)]
 mod tests {
@@ -207,12 +165,12 @@ mod tests {
 		let logger = Arc::new(TestLogger::new());
 		let peer_store = PeerStore::new(Arc::clone(&store), Arc::clone(&logger));
 
-		let pubkey = PublicKey::from_str(
+		let node_id = PublicKey::from_str(
 			"0276607124ebe6a6c9338517b6f485825b27c2dcc0b9fc2aa6a4c0df91194e5993",
 		)
 		.unwrap();
-		let address: SocketAddr = "127.0.0.1:9738".parse().unwrap();
-		let expected_peer_info = PeerInfo { pubkey, address };
+		let address = NetAddress::from_str("127.0.0.1:9738").unwrap();
+		let expected_peer_info = PeerInfo { node_id, address };
 		peer_store.add_peer(expected_peer_info.clone()).unwrap();
 		assert!(store.get_and_clear_did_persist());
 
@@ -226,7 +184,7 @@ mod tests {
 		let peers = deser_peer_store.list_peers();
 		assert_eq!(peers.len(), 1);
 		assert_eq!(peers[0], expected_peer_info);
-		assert_eq!(deser_peer_store.get_peer(&pubkey), Some(expected_peer_info));
+		assert_eq!(deser_peer_store.get_peer(&node_id), Some(expected_peer_info));
 		assert!(!store.get_and_clear_did_persist());
 	}
 }

src/test/functional_tests.rs

@@ -39,7 +39,7 @@ fn channel_full_cycle() {
 	node_a
 		.connect_open_channel(
 			node_b.node_id(),
-			node_b.listening_address().unwrap(),
+			node_b.listening_address().unwrap().into(),
 			funding_amount_sat,
 			Some(push_msat),
 			true,
@@ -240,7 +240,7 @@ fn channel_open_fails_when_funds_insufficient() {
 		Err(Error::InsufficientFunds),
 		node_a.connect_open_channel(
 			node_b.node_id(),
-			node_b.listening_address().unwrap(),
+			node_b.listening_address().unwrap().into(),
 			120000,
 			None,
 			true