Debug-Assert that there are no queued messages when a peer connects

In the previous two commits we stopped enqueueing messages for
disconnected peers. Here we add some basic test coverage of these
changes by asserting that there are no queued messages for a peer
when they (re-)connect.

We also add a few assertions that a peer is connected when we push
a message onto the queu, in cases where we expect a peer to be
connected.

Author: TheBlueMatt

lightning/src/ln/channelmanager.rs

@@ -9559,6 +9559,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 					msg: msgs::ErrorMessage { channel_id: *temporary_channel_id, data: "No zero confirmation channels accepted".to_owned(), }
 				}
 			};
+			debug_assert!(peer_state.is_connected);
 			peer_state.pending_msg_events.push(send_msg_err_event);
 			let err_str = "Please use accept_inbound_channel_from_trusted_peer_0conf to accept channels with zero confirmations.".to_owned();
 			log_error!(logger, "{}", err_str);
@@ -9575,6 +9576,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 						msg: msgs::ErrorMessage { channel_id: *temporary_channel_id, data: "Have too many peers with unfunded channels, not accepting new ones".to_owned(), }
 					}
 				};
+				debug_assert!(peer_state.is_connected);
 				peer_state.pending_msg_events.push(send_msg_err_event);
 				let err_str = "Too many peers with unfunded channels, refusing to accept new ones".to_owned();
 				log_error!(logger, "{}", err_str);
@@ -9588,6 +9590,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 		channel.context_mut().set_outbound_scid_alias(outbound_scid_alias);
 
 		if let Some(message_send_event) = message_send_event {
+			debug_assert!(peer_state.is_connected);
 			peer_state.pending_msg_events.push(message_send_event);
 		}
 		peer_state.channel_by_id.insert(channel_id, channel);
@@ -13339,7 +13342,7 @@ where
 
 			{
 				let mut peer_state_lock = self.per_peer_state.write().unwrap();
-				match peer_state_lock.entry(counterparty_node_id.clone()) {
+				match peer_state_lock.entry(counterparty_node_id) {
 					hash_map::Entry::Vacant(e) => {
 						if inbound_peer_limited {
 							res = Err(());
@@ -13371,6 +13374,9 @@ where
 							return NotifyOption::SkipPersistNoEvents;
 						}
 
+						debug_assert!(peer_state.pending_msg_events.is_empty());
+						peer_state.pending_msg_events.clear();
+
 						debug_assert!(!peer_state.is_connected, "A peer shouldn't be connected twice");
 						peer_state.is_connected = true;
 					},