Test that do_attempt_write_data does not infinitely loop

when all the channel messages can't fit into the buffer.

Adds a test for PR lightningdevkit#550.

Author: valentinewallace

lightning/src/ln/peer_handler.rs

@@ -343,6 +343,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref> PeerManager<Descriptor, CM> where
 			($msg: expr) => {
 				{
 					log_trace!(self, "Encoding and sending sync update message of type {} to {}", $msg.type_id(), log_pubkey!(peer.their_node_id.unwrap()));
+					println!("VMW: len of peer.pending_outbound_buffer: {}", peer.pending_outbound_buffer.len());
 					peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!($msg)[..]));
 				}
 			}
@@ -353,6 +354,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref> PeerManager<Descriptor, CM> where
 				match peer.sync_status {
 					InitSyncTracker::NoSyncRequested => {},
 					InitSyncTracker::ChannelsSyncing(c) if c < 0xffff_ffff_ffff_ffff => {
+						println!("VMW: peer.pending_outbound_buffer.len() before calculating step: {}", peer.pending_outbound_buffer.len());
 						let steps = ((MSG_BUFF_SIZE - peer.pending_outbound_buffer.len() + 2) / 3) as u8;
 						let all_messages = self.message_handler.route_handler.get_next_channel_announcements(c, steps);
 						for &(ref announce, ref update_a, ref update_b) in all_messages.iter() {
@@ -482,6 +484,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref> PeerManager<Descriptor, CM> where
 									{
 										log_trace!(self, "Encoding and sending message of type {} to {}", $msg.type_id(), log_pubkey!(peer.their_node_id.unwrap()));
 										peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(&$msg)[..]));
+										println!("VMW in encode_and_send_msg: len of peer.pending_outbound_buffer: {}", peer.pending_outbound_buffer.len());
 										peers.peers_needing_send.insert(peer_descriptor.clone());
 									}
 								}
@@ -953,6 +956,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref> PeerManager<Descriptor, CM> where
 								//TODO: Do whatever we're gonna do for handling dropped messages
 							});
 						peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
+						println!("VMW: len of peer.pending_outbound_buffer: {}", peer.pending_outbound_buffer.len());
 						self.do_attempt_write_data(&mut descriptor, peer);
 					},
 					MessageSendEvent::BroadcastChannelAnnouncement { ref msg, ref update_msg } => {
@@ -976,6 +980,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref> PeerManager<Descriptor, CM> where
 								}
 								peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_msg[..]));
 								peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_update_msg[..]));
+								println!("VMW: len of peer.pending_outbound_buffer: {}", peer.pending_outbound_buffer.len());
 								self.do_attempt_write_data(&mut (*descriptor).clone(), peer);
 							}
 						}
@@ -1006,6 +1011,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref> PeerManager<Descriptor, CM> where
 									continue
 								}
 								peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_msg[..]));
+								println!("VMW: len of peer.pending_outbound_buffer post BroadcastChannelUpdate: {}", peer.pending_outbound_buffer.len());
 								self.do_attempt_write_data(&mut (*descriptor).clone(), peer);
 							}
 						}
@@ -1149,8 +1155,13 @@ impl<Descriptor: SocketDescriptor, CM: Deref> PeerManager<Descriptor, CM> where
 
 #[cfg(test)]
 mod tests {
+	use secp256k1::Signature;
+	use bitcoin::BitcoinHash;
+	use bitcoin::network::constants::Network;
+	use bitcoin::blockdata::constants::genesis_block;
 	use ln::peer_handler::{PeerManager, MessageHandler, SocketDescriptor};
 	use ln::msgs;
+	use ln::features::ChannelFeatures;
 	use util::events;
 	use util::test_utils;
 	use util::logger::Logger;
@@ -1162,6 +1173,7 @@ mod tests {
 
 	use std;
 	use std::sync::{Arc, Mutex};
+	use std::sync::atomic::{AtomicUsize, Ordering};
 
 	#[derive(Clone)]
 	struct FileDescriptor {
@@ -1199,7 +1211,7 @@ mod tests {
 		chan_handlers
 	}
 
-	fn create_network<'a>(peer_count: usize, chan_handlers: &'a Vec<test_utils::TestChannelMessageHandler>) -> Vec<PeerManager<FileDescriptor, &'a test_utils::TestChannelMessageHandler>> {
+	fn create_network<'a>(peer_count: usize, chan_handlers: &'a Vec<test_utils::TestChannelMessageHandler>, routing_handlers: Option<&'a Vec<Arc<msgs::RoutingMessageHandler>>>) -> Vec<PeerManager<FileDescriptor, &'a test_utils::TestChannelMessageHandler>> {
 		let mut peers = Vec::new();
 		let mut rng = thread_rng();
 		let logger : Arc<Logger> = Arc::new(test_utils::TestLogger::new());
@@ -1208,29 +1220,37 @@ mod tests {
 
 		for i in 0..peer_count {
 			let router = test_utils::TestRoutingMessageHandler::new();
+			let router = if let Some(routers) = routing_handlers { routers[i].clone() } else {
+				Arc::new(router)
+			};
 			let node_id = {
 				let mut key_slice = [0;32];
 				rng.fill_bytes(&mut key_slice);
 				SecretKey::from_slice(&key_slice).unwrap()
 			};
-			let msg_handler = MessageHandler { chan_handler: &chan_handlers[i], route_handler: Arc::new(router) };
+			let msg_handler = MessageHandler { chan_handler: &chan_handlers[i], route_handler: router };
 			let peer = PeerManager::new(msg_handler, node_id, &ephemeral_bytes, Arc::clone(&logger));
 			peers.push(peer);
 		}
 
 		peers
 	}
 
-	fn establish_connection<'a>(peer_a: &PeerManager<FileDescriptor, &'a test_utils::TestChannelMessageHandler>, peer_b: &PeerManager<FileDescriptor, &'a test_utils::TestChannelMessageHandler>) {
+	fn establish_connection<'a>(peer_a: &PeerManager<FileDescriptor, &'a test_utils::TestChannelMessageHandler>, peer_b: &PeerManager<FileDescriptor, &'a test_utils::TestChannelMessageHandler>) -> (FileDescriptor, FileDescriptor) {
 		let secp_ctx = Secp256k1::new();
 		let a_id = PublicKey::from_secret_key(&secp_ctx, &peer_a.our_node_secret);
 		let mut fd_a = FileDescriptor { fd: 1, outbound_data: Arc::new(Mutex::new(Vec::new())) };
 		let mut fd_b = FileDescriptor { fd: 1, outbound_data: Arc::new(Mutex::new(Vec::new())) };
 		let initial_data = peer_b.new_outbound_connection(a_id, fd_b.clone()).unwrap();
 		peer_a.new_inbound_connection(fd_a.clone()).unwrap();
+		println!("doing peer_a read event");
 		assert_eq!(peer_a.read_event(&mut fd_a, &initial_data).unwrap(), false);
+		println!("just did peer_a read event");
 		assert_eq!(peer_b.read_event(&mut fd_b, &fd_a.outbound_data.lock().unwrap().split_off(0)).unwrap(), false);
+		println!("just did peer_b read event");
 		assert_eq!(peer_a.read_event(&mut fd_a, &fd_b.outbound_data.lock().unwrap().split_off(0)).unwrap(), false);
+		println!("just did peer_a read event");
+		(fd_a.clone(), fd_b.clone())
 	}
 
 	#[test]
@@ -1239,7 +1259,7 @@ mod tests {
 		// push a DisconnectPeer event to remove the node flagged by id
 		let chan_handlers = create_chan_handlers(2);
 		let chan_handler = test_utils::TestChannelMessageHandler::new();
-		let mut peers = create_network(2, &chan_handlers);
+		let mut peers = create_network(2, &chan_handlers, None);
 		establish_connection(&peers[0], &peers[1]);
 		assert_eq!(peers[0].peers.lock().unwrap().peers.len(), 1);
 
@@ -1256,11 +1276,12 @@ mod tests {
 		peers[0].process_events();
 		assert_eq!(peers[0].peers.lock().unwrap().peers.len(), 0);
 	}
+
 	#[test]
-	fn test_timer_tick_occured(){
+	fn test_timer_tick_occurred() {
 		// Create peers, a vector of two peer managers, perform initial set up and check that peers[0] has one Peer.
 		let chan_handlers = create_chan_handlers(2);
-		let peers = create_network(2, &chan_handlers);
+		let peers = create_network(2, &chan_handlers, None);
 		establish_connection(&peers[0], &peers[1]);
 		assert_eq!(peers[0].peers.lock().unwrap().peers.len(), 1);
 
@@ -1272,4 +1293,142 @@ mod tests {
 		peers[0].timer_tick_occured();
 		assert_eq!(peers[0].peers.lock().unwrap().peers.len(), 0);
 	}
+
+	pub struct TestRoutingMessageHandler {
+		pub chan_upds_recvd: AtomicUsize,
+		pub chan_anns_recvd: AtomicUsize,
+	}
+
+	impl TestRoutingMessageHandler {
+		pub fn new() -> Self {
+			TestRoutingMessageHandler {
+				chan_upds_recvd: AtomicUsize::new(0),
+				chan_anns_recvd: AtomicUsize::new(0),
+			}
+		}
+
+	}
+	impl msgs::RoutingMessageHandler for TestRoutingMessageHandler {
+		fn handle_node_announcement(&self, _msg: &msgs::NodeAnnouncement) -> Result<bool, msgs::LightningError> {
+			Err(msgs::LightningError { err: "", action: msgs::ErrorAction::IgnoreError })
+		}
+		fn handle_channel_announcement(&self, _msg: &msgs::ChannelAnnouncement) -> Result<bool, msgs::LightningError> {
+			self.chan_anns_recvd.fetch_add(1, Ordering::AcqRel);
+			Err(msgs::LightningError { err: "", action: msgs::ErrorAction::IgnoreError })
+		}
+		fn handle_channel_update(&self, _msg: &msgs::ChannelUpdate) -> Result<bool, msgs::LightningError> {
+			self.chan_upds_recvd.fetch_add(1, Ordering::AcqRel);
+			Err(msgs::LightningError { err: "", action: msgs::ErrorAction::IgnoreError })
+		}
+		fn handle_htlc_fail_channel_update(&self, _update: &msgs::HTLCFailChannelUpdate) {}
+		fn get_next_channel_announcements(&self, starting_point: u64, batch_amount: u8) -> Vec<(msgs::ChannelAnnouncement, msgs::ChannelUpdate,msgs::ChannelUpdate)> {
+			println!("VMW: starting_point: {}, batch_amt: {}", starting_point, batch_amount);
+			let mut chan_anns = Vec::new();
+			let batch_amt = batch_amount as u64;
+			let end: u64 = if starting_point > 0 { starting_point + batch_amt - 1 } else { batch_amount.into() };
+			for j in starting_point..end {
+				let chan_upd_1 = get_dummy_channel_update(j);
+				let chan_upd_2 = get_dummy_channel_update(j);
+				let chan_ann = get_dummy_channel_announcement(j);
+
+				chan_anns.push((chan_ann, chan_upd_1, chan_upd_2));
+			}
+
+			chan_anns
+		}
+
+		fn get_next_node_announcements(&self, _starting_point: Option<&PublicKey>, _batch_amount: u8) -> Vec<msgs::NodeAnnouncement> {
+			Vec::new()
+		}
+
+		fn should_request_full_sync(&self, _node_id: &PublicKey) -> bool {
+			true
+		}
+	}
+
+	fn get_dummy_channel_announcement(short_chan_id: u64) -> msgs::ChannelAnnouncement {
+		use secp256k1::ffi::Signature as FFISignature;
+		let secp_ctx = Secp256k1::new();
+		let network = Network::Testnet;
+		let node_1_privkey = SecretKey::from_slice(&[42; 32]).unwrap();
+		let node_2_privkey = SecretKey::from_slice(&[41; 32]).unwrap();
+		let node_1_btckey = SecretKey::from_slice(&[40; 32]).unwrap();
+		let node_2_btckey = SecretKey::from_slice(&[39; 32]).unwrap();
+		let unsigned_ann = msgs::UnsignedChannelAnnouncement {
+			features: ChannelFeatures::supported(),
+			chain_hash: genesis_block(network).header.bitcoin_hash(),
+			short_channel_id: short_chan_id,
+			node_id_1: PublicKey::from_secret_key(&secp_ctx, &node_1_privkey),
+			node_id_2: PublicKey::from_secret_key(&secp_ctx, &node_2_privkey),
+			bitcoin_key_1: PublicKey::from_secret_key(&secp_ctx, &node_1_btckey),
+			bitcoin_key_2: PublicKey::from_secret_key(&secp_ctx, &node_2_btckey),
+			excess_data: Vec::new(),
+		};
+
+		msgs::ChannelAnnouncement {
+			node_signature_1: Signature::from(FFISignature::new()),
+			node_signature_2: Signature::from(FFISignature::new()),
+			bitcoin_signature_1: Signature::from(FFISignature::new()),
+			bitcoin_signature_2: Signature::from(FFISignature::new()),
+			contents: unsigned_ann,
+		}
+	}
+
+	fn get_dummy_channel_update(short_chan_id: u64) -> msgs::ChannelUpdate {
+		use secp256k1::ffi::Signature as FFISignature;
+		let network = Network::Testnet;
+		msgs::ChannelUpdate {
+			signature: Signature::from(FFISignature::new()),
+			contents: msgs::UnsignedChannelUpdate {
+				chain_hash: genesis_block(network).header.bitcoin_hash(),
+				short_channel_id: short_chan_id,
+				timestamp: 0,
+				flags: 0,
+				cltv_expiry_delta: 0,
+				htlc_minimum_msat: 0,
+				fee_base_msat: 0,
+				fee_proportional_millionths: 0,
+				excess_data: vec![],
+			}
+		}
+	}
+
+	#[test]
+	fn test_do_attempt_write_data() {
+		// Create 2 peers with custom TestRoutingMessageHandlers and connect them.
+		let chan_handlers = create_chan_handlers(2);
+		let mut routing_handlers: Vec<Arc<msgs::RoutingMessageHandler>> = Vec::new();
+		let mut routing_handlers_concrete: Vec<Arc<TestRoutingMessageHandler>> = Vec::new();
+		for _ in 0..2 {
+			let routing_handler = Arc::new(TestRoutingMessageHandler::new());
+			routing_handlers.push(routing_handler.clone());
+			routing_handlers_concrete.push(routing_handler.clone());
+		}
+		let peers = create_network(2, &chan_handlers, Some(&routing_handlers));
+
+		// By calling establish_connect, we trigger do_attempt_write_data between
+		// the peers. Previously this function would mistakenly enter an infinite loop
+		// when there were more channel messages available than could fit into a peer's
+		// buffer. This issue would now be detected by this test (because we use custom
+		// RoutingMessageHandlers that intentionally return more channel messages
+		// than can fit into a peer's buffer).
+		let (mut fd_a, mut fd_b) = establish_connection(&peers[0], &peers[1]);
+
+		// Make each peer to read the messages that the other peer just wrote to them.
+		println!("VMW IN TEST: doing peer_b read_event");
+		peers[1].read_event(&mut fd_b, &fd_a.outbound_data.lock().unwrap().split_off(0)).unwrap();
+		println!("VMW IN TEST: just did peer_b read event, doing peer_a read_event");
+		peers[0].read_event(&mut fd_a, &fd_b.outbound_data.lock().unwrap().split_off(0)).unwrap();
+		println!("VMW IN TEST: just did peer_a read_event");
+
+		// Check that each peer has received the expected number of channel updates and channel
+		// announcements. peer_a sends one fewer channel announcement to peer_b because, at the
+		// time of graph sync, peer_a already has an Init message for peer_b in its buffer of
+		// messages to send to peer_b (and the number of messages sent is dependent on the size
+		// of said buffer; see how `steps` is calculated in do_attempt_write_data).
+		assert_eq!(routing_handlers_concrete[0].clone().chan_upds_recvd.load(Ordering::Acquire), 8);
+		assert_eq!(routing_handlers_concrete[0].clone().chan_anns_recvd.load(Ordering::Acquire), 4);
+		assert_eq!(routing_handlers_concrete[1].clone().chan_upds_recvd.load(Ordering::Acquire), 6);
+		assert_eq!(routing_handlers_concrete[1].clone().chan_anns_recvd.load(Ordering::Acquire), 3);
+	}
 }