Stop enqueueing error messages for disconnected peers

If a channel is failed while a peer is disconnected, we'll still
have a `PeerState` for that peer. Historically, we haven't bothered
to check if a peer is actually connected before we push the `error`
message onto the `PeerState::pending_msg_events` queue, leaving us
sending messages into the void.

This is generally not an issue as
`ChannelManager::get_and_clear_pending_msg_events` should be called
very regularly, removing these messages and then dropping them as
`PeerManager` won't have anything to do with them. Further, when
the the message is an `error`, if a peer happens to connect between
when we push the message and when `get_and_clear_pending_msg_events`
is called the worst that happens is they get the `error` message
we'd end up sending them when they try to reestablish the channel
anyway.

Still, its awkward to leave the `error`s lying around in a message
queue for a disconnected peer, so we remove them here.

Author: TheBlueMatt

lightning/src/ln/chanmon_update_fail_tests.rs

@@ -3897,7 +3897,7 @@ fn do_test_durable_preimages_on_closed_channel(
 		}
 	}
 	if !close_chans_before_reload {
-		check_closed_broadcast(&nodes[1], 1, true);
+		check_closed_broadcast(&nodes[1], 1, false);
 		let reason = ClosureReason::CommitmentTxConfirmed;
 		check_closed_event(&nodes[1], 1, reason, false, &[node_a_id], 100000);
 	}
@@ -3914,7 +3914,7 @@ fn do_test_durable_preimages_on_closed_channel(
 	check_spends!(bs_preimage_tx, as_closing_tx[0]);
 
 	mine_transactions(&nodes[0], &[&as_closing_tx[0], bs_preimage_tx]);
-	check_closed_broadcast(&nodes[0], 1, true);
+	check_closed_broadcast(&nodes[0], 1, false);
 	expect_payment_sent(&nodes[0], payment_preimage, None, true, true);
 
 	if !close_chans_before_reload || close_only_a {
@@ -4063,7 +4063,7 @@ fn do_test_reload_mon_update_completion_actions(close_during_reload: bool) {
 			Event::ChannelClosed { .. } => {},
 			_ => panic!(),
 		}
-		check_closed_broadcast!(nodes[1], true);
+		check_closed_broadcast(&nodes[1], 1, false);
 	}
 
 	// Once we run event processing the monitor should free, check that it was indeed the B<->C

lightning/src/ln/channelmanager.rs

@@ -3228,7 +3228,9 @@ macro_rules! handle_error {
 					let per_peer_state = $self.per_peer_state.read().unwrap();
 					if let Some(peer_state_mutex) = per_peer_state.get(&$counterparty_node_id) {
 						let mut peer_state = peer_state_mutex.lock().unwrap();
-						peer_state.pending_msg_events.push(msg_event);
+						if peer_state.is_connected {
+							peer_state.pending_msg_events.push(msg_event);
+						}
 					}
 				}
 
@@ -18808,7 +18810,7 @@ mod tests {
 			.node
 			.force_close_broadcasting_latest_txn(&chan_id, &nodes[1].node.get_our_node_id(), message.clone())
 			.unwrap();
-		check_closed_broadcast(&nodes[0], 1, true);
+		check_closed_broadcast(&nodes[0], 1, false);
 		check_added_monitors(&nodes[0], 1);
 		let reason = ClosureReason::HolderForceClosed { broadcasted_latest_txn: Some(true), message };
 		check_closed_event!(nodes[0], 1, reason, [nodes[1].node.get_our_node_id()], 100000);

lightning/src/ln/functional_tests.rs

@@ -8023,7 +8023,7 @@ fn do_test_tx_confirmed_skipping_blocks_immediate_broadcast(test_height_before_t
 		.force_close_broadcasting_latest_txn(&channel_id, &node_c_id, message.clone())
 		.unwrap();
 
-	check_closed_broadcast!(nodes[1], true);
+	check_closed_broadcast(&nodes[1], 1, false);
 	let reason = ClosureReason::HolderForceClosed { broadcasted_latest_txn: Some(true), message };
 	check_closed_event!(nodes[1], 1, reason, [node_c_id], 100000);
 	check_added_monitors(&nodes[1], 1);

lightning/src/ln/monitor_tests.rs

@@ -3044,7 +3044,7 @@ fn do_test_anchors_monitor_fixes_counterparty_payment_script_on_reload(confirm_c
 		let serialized_monitor = get_monitor!(nodes[1], chan_id).encode();
 		reload_node!(nodes[1], user_config, &nodes[1].node.encode(), &[&serialized_monitor], persister, chain_monitor, node_deserialized);
 		let commitment_tx_conf_height = block_from_scid(mine_transaction(&nodes[1], &commitment_tx));
-		check_closed_broadcast(&nodes[1], 1, true);
+		check_closed_broadcast(&nodes[1], 1, false);
 		check_added_monitors(&nodes[1], 1);
 		commitment_tx_conf_height
 	};
@@ -3407,7 +3407,7 @@ fn do_test_lost_preimage_monitor_events(on_counterparty_tx: bool) {
 	check_added_monitors(&nodes[2], 1);
 	let c_reason = ClosureReason::HolderForceClosed { broadcasted_latest_txn: Some(true), message };
 	check_closed_event!(nodes[2], 1, c_reason, [node_b_id], 1_000_000);
-	check_closed_broadcast!(nodes[2], true);
+	check_closed_broadcast(&nodes[2], 1, false);
 
 	handle_bump_events(&nodes[2], true, 0);
 	let cs_commit_tx = nodes[2].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0);
@@ -3421,7 +3421,7 @@ fn do_test_lost_preimage_monitor_events(on_counterparty_tx: bool) {
 	check_added_monitors(&nodes[1], 1);
 	let b_reason = ClosureReason::HolderForceClosed { broadcasted_latest_txn: Some(true), message };
 	check_closed_event!(nodes[1], 1, b_reason, [node_c_id], 1_000_000);
-	check_closed_broadcast!(nodes[1], true);
+	check_closed_broadcast(&nodes[1], 1, false);
 
 	handle_bump_events(&nodes[1], true, 0);
 	let bs_commit_tx = nodes[1].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0);
@@ -3618,7 +3618,7 @@ fn do_test_lost_timeout_monitor_events(confirm_tx: CommitmentType, dust_htlcs: b
 	check_added_monitors(&nodes[2], 1);
 	let c_reason = ClosureReason::HolderForceClosed { broadcasted_latest_txn: Some(true), message };
 	check_closed_event!(nodes[2], 1, c_reason, [node_b_id], 1_000_000);
-	check_closed_broadcast!(nodes[2], true);
+	check_closed_broadcast(&nodes[2], 1, false);
 
 	handle_bump_events(&nodes[2], true, 0);
 	let cs_commit_tx = nodes[2].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0);
@@ -3632,7 +3632,7 @@ fn do_test_lost_timeout_monitor_events(confirm_tx: CommitmentType, dust_htlcs: b
 	check_added_monitors(&nodes[1], 1);
 	let b_reason = ClosureReason::HolderForceClosed { broadcasted_latest_txn: Some(true), message };
 	check_closed_event!(nodes[1], 1, b_reason, [node_c_id], 1_000_000);
-	check_closed_broadcast!(nodes[1], true);
+	check_closed_broadcast(&nodes[1], 1, false);
 
 	handle_bump_events(&nodes[1], true, 0);
 	let bs_commit_tx = nodes[1].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0);

lightning/src/ln/payment_tests.rs

@@ -1287,7 +1287,7 @@ fn do_test_dup_htlc_onchain_doesnt_fail_on_reload(
 	expect_payment_claimed!(nodes[1], payment_hash, 10_000_000);
 
 	mine_transaction(&nodes[1], &commitment_tx);
-	check_closed_broadcast!(nodes[1], true);
+	check_closed_broadcast(&nodes[1], 1, false);
 	check_added_monitors!(nodes[1], 1);
 	check_closed_event!(nodes[1], 1, ClosureReason::CommitmentTxConfirmed, [node_a_id], 100000);
 	let htlc_success_tx = {

lightning/src/ln/reorg_tests.rs

@@ -325,6 +325,16 @@ fn do_test_unconf_chan(reload_node: bool, reorg_after_reload: bool, use_funding_
 		let chan_0_monitor_serialized = get_monitor!(nodes[0], chan.2).encode();
 
 		reload_node!(nodes[0], nodes[0].node.get_current_config(), &nodes_0_serialized, &[&chan_0_monitor_serialized], persister, new_chain_monitor, nodes_0_deserialized);
+
+		nodes[1].node.peer_disconnected(nodes[0].node.get_our_node_id());
+
+		if reorg_after_reload {
+			// If we haven't yet closed the channel, reconnect the peers so that nodes[0] will
+			// generate an error message we can handle below.
+			let mut reconnect_args = ReconnectArgs::new(&nodes[0], &nodes[1]);
+			reconnect_args.send_channel_ready = (true, true);
+			reconnect_nodes(reconnect_args);
+		}
 	}
 
 	if reorg_after_reload {
@@ -387,7 +397,7 @@ fn do_test_unconf_chan(reload_node: bool, reorg_after_reload: bool, use_funding_
 		[nodes[1].node.get_our_node_id()], 100000);
 
 	// Now check that we can create a new channel
-	if reload_node {
+	if reload_node && !reorg_after_reload {
 		// If we dropped the channel before reloading the node, nodes[1] was also dropped from
 		// nodes[0] storage, and hence not connected again on startup. We therefore need to
 		// reconnect to the node before attempting to create a new channel.
@@ -879,7 +889,7 @@ fn do_test_retries_own_commitment_broadcast_after_reorg(anchors: bool, revoked_c
 		.node
 		.force_close_broadcasting_latest_txn(&chan_id, &nodes[0].node.get_our_node_id(), message.clone())
 		.unwrap();
-	check_closed_broadcast(&nodes[1], 1, true);
+	check_closed_broadcast(&nodes[1], 1, !revoked_counterparty_commitment);
 	check_added_monitors(&nodes[1], 1);
 	let reason = ClosureReason::HolderForceClosed { broadcasted_latest_txn: Some(true), message };
 	check_closed_event(&nodes[1], 1, reason, false, &[nodes[0].node.get_our_node_id()], 100_000);
@@ -1002,7 +1012,7 @@ fn do_test_split_htlc_expiry_tracking(use_third_htlc: bool, reorg_out: bool) {
 	// Force-close and fetch node B's commitment transaction and the transaction claiming the first
 	// two HTLCs.
 	nodes[1].node.force_close_broadcasting_latest_txn(&chan_id, &node_a_id, err).unwrap();
-	check_closed_broadcast(&nodes[1], 1, true);
+	check_closed_broadcast(&nodes[1], 1, false);
 	check_added_monitors(&nodes[1], 1);
 	let message = "Channel force-closed".to_owned();
 	let reason = ClosureReason::HolderForceClosed { broadcasted_latest_txn: Some(true), message };
@@ -1015,7 +1025,7 @@ fn do_test_split_htlc_expiry_tracking(use_third_htlc: bool, reorg_out: bool) {
 	check_spends!(commitment_tx, funding_tx);
 
 	mine_transaction(&nodes[0], &commitment_tx);
-	check_closed_broadcast(&nodes[0], 1, true);
+	check_closed_broadcast(&nodes[0], 1, false);
 	let reason = ClosureReason::CommitmentTxConfirmed;
 	check_closed_event(&nodes[0], 1, reason, false, &[node_b_id], 10_000_000);
 	check_added_monitors(&nodes[0], 1);