Implement SendErrorMessage event with disconnection flag + test disconnection

Author: Antoine Riard

src/ln/channelmanager.rs

@@ -677,6 +677,15 @@ impl ChannelManager {
 	/// Call this upon creation of a funding transaction for the given channel.
 	/// Panics if a funding transaction has already been provided for this channel.
 	pub fn funding_transaction_generated(&self, temporary_channel_id: &Uint256, funding_txo: OutPoint) {
+
+		macro_rules! add_pending_event {
+			($event: expr) => {
+				let mut pending_events = self.pending_events.lock().unwrap();
+				pending_events.push($event);
+			}
+		}
+
+		let mut temporary_pending_events = Vec::new();
 		let (chan, msg, chan_monitor) = {
 			let mut channel_state = self.channel_state.lock().unwrap();
 			match channel_state.by_id.remove(&temporary_channel_id) {
@@ -685,25 +694,40 @@ impl ChannelManager {
 						Ok(funding_msg) => {
 							(chan, funding_msg.0, funding_msg.1)
 						},
-						Err(_e) => {
-							//TODO: Push e to pendingevents
+						Err(e) => {
+							println!("Got error handling message: {}!", e.err);
+							if let Some(action) = e.msg {
+								match action {
+									msgs::ErrorAction::DisconnectPeer {} => {
+										temporary_pending_events.push(events::Event::SendErrorMessage {
+											node_id: chan.get_their_node_id(),
+											msg: msgs::ErrorMessage { err: "" },
+											disconnect: true,
+											hard: true,
+										});
+									},
+									_ => panic!("Kind of error not implemented yet for funding_transaction_generated"),
+								}
+							} else {
+								panic!("ErrorAction needed for this HandleError into funding_transaction_generated");
+							}
 							return;
-						}
+						},
 					}
 				},
 				None => return
 			}
 		}; // Release channel lock for install_watch_outpoint call,
+		if let Some(event) = temporary_pending_events.pop() {
+			add_pending_event!(event);
+		}
 		if let Err(_e) = self.monitor.add_update_monitor(chan_monitor.get_funding_txo().unwrap(), chan_monitor) {
 			unimplemented!(); // maybe remove from claimable_htlcs?
 		}
-		{
-			let mut pending_events = self.pending_events.lock().unwrap();
-			pending_events.push(events::Event::SendFundingCreated {
-				node_id: chan.get_their_node_id(),
-				msg: msg,
-			});
-		}
+		add_pending_event!(events::Event::SendFundingCreated {
+			node_id: chan.get_their_node_id(),
+			msg: msg,
+		});
 
 		let mut channel_state = self.channel_state.lock().unwrap();
 		channel_state.by_id.insert(chan.channel_id(), chan);
@@ -1811,6 +1835,13 @@ impl ChannelMessageHandler for ChannelManager {
 			}
 		}
 	}
+
+	//For testing purpose in peer_handler, to manually push target event
+	#[cfg(test)]
+	fn push_event(&self, event: events::Event) {
+		let mut pending_events = self.pending_events.lock().unwrap();
+		pending_events.push(event);
+	}
 }
 
 #[cfg(test)]

src/ln/msgs.rs

@@ -139,6 +139,10 @@ pub struct Init {
 	pub local_features: LocalFeatures,
 }
 
+pub struct ErrorMessage {
+	pub err: &'static str,
+}
+
 pub struct Ping {
 	pub ponglen: u16,
 	pub byteslen: u16,
@@ -435,6 +439,10 @@ pub trait ChannelMessageHandler : events::EventsProvider + Send + Sync {
 	/// understand or indicate they require unknown feature bits), no_connection_possible is set
 	/// and any outstanding channels should be failed.
 	fn peer_disconnected(&self, their_node_id: &PublicKey, no_connection_possible: bool);
+
+	//Testing
+	#[cfg(test)]
+	fn push_event(&self, event: events::Event);
 }
 
 pub trait RoutingMessageHandler : Send + Sync {
@@ -1534,3 +1542,11 @@ impl MsgEncodable for OnionErrorPacket {
 		res
 	}
 }
+
+impl MsgEncodable for ErrorMessage {
+	fn encode(&self) -> Vec<u8> {
+		let mut res = Vec::with_capacity(self.err.len());
+		res.extend_from_slice(&self.err.as_bytes());
+		res
+	}
+}

src/ln/peer_handler.rs

@@ -37,6 +37,9 @@ pub trait SocketDescriptor : cmp::Eq + hash::Hash + Clone {
 	/// indicating that read events on this descriptor should resume. A resume_read of false does
 	/// *not* imply that further read events should be paused.
 	fn send_data(&mut self, data: &Vec<u8>, write_offset: usize, resume_read: bool) -> usize;
+	///Indicate to the network provider that we want to disconnect this peer, no more allowing any call
+	///to disconnect_event or read_event
+	fn notify_disconnection(&mut self);
 }
 
 /// Error for PeerManager errors. If you get one of these, you must disconnect the socket and
@@ -576,6 +579,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 	/// calls to ChannelManager::process_pending_htlc_forward.
 	pub fn process_events(&self) {
 		let mut upstream_events = Vec::new();
+		let mut disconnect_peers = Vec::new();
 		{
 			// TODO: There are some DoS attacks here where you can flood someone's outbound send
 			// buffer by doing things like announcing channels on another node. We should be willing to
@@ -710,12 +714,30 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 						}
 						continue;
 					},
+					Event::SendErrorMessage { ref node_id, ref msg, ref disconnect, ref hard } => {
+						let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
+								//TODO: Do whatever we're gonna do for handling dropped messages
+							});
+						if *disconnect == true {
+							disconnect_peers.push(descriptor.clone());
+						}
+						if *hard == false {
+							peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 17)));
+							Self::do_attempt_write_data(&mut descriptor, peer);
+						}
+						continue;
+					},
 				}
 
 				upstream_events.push(event);
 			}
 		}
 
+		for mut descriptor in disconnect_peers {
+			descriptor.notify_disconnection();
+			self.disconnect_event(&descriptor);
+		}
+
 		let mut pending_events = self.pending_events.lock().unwrap();
 		for event in upstream_events.drain(..) {
 			pending_events.push(event);
@@ -756,3 +778,92 @@ impl<Descriptor: SocketDescriptor> EventsProvider for PeerManager<Descriptor> {
 		ret
 	}
 }
+
+#[cfg(test)]
+mod tests {
+	use chain::chaininterface;
+	use ln::peer_handler::{PeerManager, MessageHandler, SocketDescriptor};
+	use ln::channelmanager::{ChannelManager};
+	use ln::router::{Router};
+	use ln::msgs;
+	use ln::msgs::{HandleError};
+	use util::test_utils;
+	use util::events;
+
+	use bitcoin::network::constants::Network;
+	use bitcoin::util::misc::hex_bytes;
+
+	use secp256k1::Secp256k1;
+	use secp256k1::key::{SecretKey, PublicKey};
+
+	use rand::{thread_rng,Rng};
+	
+	use std::sync::{Arc, Mutex};
+
+	#[derive(PartialEq, Eq, Clone, Hash)]
+	struct FileDescriptor {
+		fd: u16,
+	}
+
+	impl SocketDescriptor for FileDescriptor {
+		fn send_data(&mut self, data: &Vec<u8>, _write_offset: usize, _resume_read: bool) -> usize {
+			data.len()
+		}
+
+		fn notify_disconnection(&mut self) {}
+	}
+
+	fn create_network(peer_count: usize) -> Vec<PeerManager<FileDescriptor>> {
+		let secp_ctx = Secp256k1::new();
+		let mut peers = Vec::new();
+		let mut rng = thread_rng();
+
+		for _ in 0..peer_count {
+			let feeest = Arc::new(test_utils::TestFeeEstimator { sat_per_vbyte: 1 });
+			let chain_monitor = Arc::new(chaininterface::ChainWatchInterfaceUtil::new());
+			let tx_broadcaster = Arc::new(test_utils::TestBroadcaster{txn_broadcasted: Mutex::new(Vec::new())});
+			let chan_monitor = Arc::new(test_utils::TestChannelMonitor::new(chain_monitor.clone(), tx_broadcaster.clone()));
+			let node_id = {
+				let mut key_slice = [0;32];
+				rng.fill_bytes(&mut key_slice);
+				SecretKey::from_slice(&secp_ctx, &key_slice).unwrap()
+			};
+			let node = ChannelManager::new(node_id.clone(), 0, true, Network::Testnet, feeest.clone(), chan_monitor.clone(), chain_monitor.clone(), tx_broadcaster.clone()).unwrap();
+			let router = Router::new(PublicKey::from_secret_key(&secp_ctx, &node_id).unwrap());
+			let msg_handler = MessageHandler { chan_handler: node, route_handler: Arc::new(router) };
+			let peer = PeerManager::new(msg_handler, node_id);
+			peers.push(peer);
+		}
+
+		peers
+	}
+
+	fn establish_connection(peer_a: &PeerManager<FileDescriptor>, peer_b: &PeerManager<FileDescriptor>) {
+		let secp_ctx = Secp256k1::new();
+		let their_id = PublicKey::from_secret_key(&secp_ctx, &peer_b.our_node_secret).unwrap();
+		let fd = FileDescriptor { fd: 1};
+		peer_a.new_inbound_connection(fd.clone());
+		peer_a.peers.lock().unwrap().node_id_to_descriptor.insert(their_id, fd.clone());
+	}
+
+	#[test]
+	fn test_disconnect_peer() {
+		// Simple test which builds a network of PeerManager, connects and brings them to NoiseState::Finished and
+		// push an DisconnectPeer event to remove the node flagged by id
+		let peers = create_network(2);
+		establish_connection(&peers[0], &peers[1]);
+		assert_eq!(peers[0].peers.lock().unwrap().peers.len(), 1);
+
+		let secp_ctx = Secp256k1::new();
+		let their_id = PublicKey::from_secret_key(&secp_ctx, &peers[1].our_node_secret).unwrap();
+		peers[0].message_handler.chan_handler.push_event(events::Event::SendErrorMessage {
+			node_id: their_id,
+			msg: msgs::ErrorMessage { err: "" },
+			disconnect: true,
+			hard: true,
+		});
+
+		peers[0].process_events();
+		assert_eq!(peers[0].peers.lock().unwrap().peers.len(), 0);
+	}
+}

src/util/events.rs

@@ -100,6 +100,13 @@ pub enum Event {
 	BroadcastChannelUpdate {
 		msg: msgs::ChannelUpdate,
 	},
+	/// Used to tell a peer that something is incorrect
+	SendErrorMessage {
+		node_id: PublicKey,
+		msg: msgs::ErrorMessage,
+		disconnect: bool,
+		hard: bool, //disconnect without message
+	},
 }
 
 pub trait EventsProvider {